merge: #2987

2987: Enable parallelism in dependent values update r=paulocsanz a=paulocsanz Co-authored-by: Paulo Cabral <[email protected]>
systeminit · Nov 29, 2023 · cc0a3aa · cc0a3aa
2 parents eea2e8d + 7dee112
commit cc0a3aa
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 13 deletions.
diff --git a/lib/council-server/src/server.rs b/lib/council-server/src/server.rs
@@ -70,15 +70,12 @@ impl Server {
 
         let mut complete_graph = ChangeSetGraph::default();
         loop {
-            for (reply_channel, node_id) in complete_graph.fetch_all_available() {
-                info!(%reply_channel, %node_id, "Ok to process AttributeValue");
+            for (reply_channel, node_ids) in complete_graph.fetch_all_available() {
+                info!(%reply_channel, ?node_ids, "Ok to process AttributeValue");
                 self.nats
                     .publish(
                         reply_channel,
-                        serde_json::to_vec(&Response::OkToProcess {
-                            node_ids: vec![node_id],
-                        })
-                        .unwrap(),
+                        serde_json::to_vec(&Response::OkToProcess { node_ids }).unwrap(),
                     )
                     .await
                     .unwrap();
@@ -230,9 +227,9 @@ pub async fn job_processed_a_value(
     change_set_id: Id,
     node_id: Id,
 ) -> Result<(), Error> {
-    debug!(%reply_channel, %change_set_id, %node_id, "Job finished processing graph node");
+    info!(%reply_channel, %change_set_id, %node_id, "Job finished processing graph node");
     for reply_channel in
-        complete_graph.mark_node_as_processed(reply_channel, change_set_id, node_id)?
+        complete_graph.mark_node_as_processed(&reply_channel, change_set_id, node_id)?
     {
         info!(%reply_channel, ?node_id, "AttributeValue has been processed by a job");
         nats.publish(

diff --git a/lib/council-server/src/server/graph.rs b/lib/council-server/src/server/graph.rs
@@ -15,12 +15,12 @@ impl ChangeSetGraph {
         self.dependency_data.is_empty()
     }
 
-    pub fn fetch_all_available(&mut self) -> Vec<(String, Id)> {
-        let mut result = Vec::new();
+    pub fn fetch_all_available(&mut self) -> HashMap<String, Vec<Id>> {
+        let mut result: HashMap<String, Vec<Id>> = HashMap::new();
         for graph in self.dependency_data.values_mut() {
             for (id, metadata) in graph.iter_mut() {
                 if let Some(reply_channel) = metadata.next_to_process() {
-                    result.push((reply_channel, *id));
+                    result.entry(reply_channel.clone()).or_default().push(*id);
                 }
             }
         }
@@ -68,15 +68,15 @@ impl ChangeSetGraph {
 
     pub fn mark_node_as_processed(
         &mut self,
-        reply_channel: String,
+        reply_channel: &str,
         change_set_id: Id,
         node_id: Id,
     ) -> Result<HashSet<String>, Error> {
         let change_set_graph_data = self.dependency_data.get_mut(&change_set_id).unwrap();
 
         let (ok_to_remove_node, wanted_by_reply_channels) =
             if let Some(node_metadata) = change_set_graph_data.get_mut(&node_id) {
-                node_metadata.mark_as_processed(&reply_channel)?
+                node_metadata.mark_as_processed(reply_channel)?
             } else {
                 return Err(Error::UnknownNodeId);
             };