CSY/reowolf Changeset - 0bc3b84bc033 · Centrum Wiskunde & Informatica (CWI)

Changeset - 0bc3b84bc033

Parent rev.

Child rev.

[Not reviewed]

0 1 0

MH - 4 years ago 2021-11-08 10:18:53
contact@maxhenger.nl

WIP on new consensus approach

1 file changed with 78 insertions and 4 deletions:

src/runtime2/consensus.rs

0 comments (0 inline, 0 general)

src/runtime2/consensus.rs

➞

Show inline comments

 use std::path::Component;
 use crate::collections::VecSet;
 use crate::protocol::eval::ValueGroup;
 use crate::runtime2::branch::{BranchId, ExecTree, QueueKind};
 use crate::runtime2::ConnectorId;
 use crate::runtime2::inbox2::{DataHeader, MessageFancy, SyncContent, SyncHeader, SyncMessageFancy};
 use crate::runtime2::inbox::SyncMessage;
 use crate::runtime2::port::{Port, PortIdLocal};
 use crate::runtime2::scheduler::ComponentCtxFancy;
 use super::inbox2::PortAnnotation;
 struct BranchAnnotation {
     port_mapping: Vec<PortAnnotation>,
+}
 pub(crate) struct LocalSolution {
     component: ConnectorId,
     final_branch_id: BranchId,
     port_mapping: Vec<(PortIdLocal, BranchId)>,
+}
 pub(crate) struct GlobalSolution {
+}
 // -----------------------------------------------------------------------------
 // Consensus
 // -----------------------------------------------------------------------------
 /// The consensus algorithm. Currently only implemented to find the component
 /// with the highest ID within the sync region and letting it handle all the
 /// local solutions.
 ///
 /// The type itself serves as an experiment to see how code should be organized.
 // TODO: Flatten all datastructures
 // TODO: Have a "branch+port position hint" in case multiple operations are
 //  performed on the same port to prevent repeated lookups
 // TODO: A lot of stuff should be batched. Like checking all the sync headers
 //  and sending "I have a higher ID" messages.
 pub(crate) struct Consensus {
     // Local component's state
     highest_connector_id: ConnectorId,
     branch_annotations: Vec<BranchAnnotation>,
     last_finished_handled: Option<BranchId>,
     // Gathered state (in case we are currently the leader of the distributed
     // consensus protocol)
     encountered_peers: VecSet<ConnectorId>,
     local_solutions: Vec<LocalSolution>,
     // Workspaces
     workspace_ports: Vec<PortIdLocal>,
+}
@@ @@ -253,49 +253,49 @@ impl Consensus { @@
     pub fn handle_received_sync_header(&mut self, sync_header: &SyncHeader, ctx: &mut ComponentCtxFancy) {
         debug_assert!(sync_header.sending_component_id != ctx.id); // not sending to ourselves
         self.encountered_peers.push(sync_header.sending_component_id);
         if sync_header.highest_component_id > self.highest_connector_id {
             // Sender has higher component ID. So should be the target of our
             // messages. We should also let all of our peers know
             self.highest_connector_id = sync_header.highest_component_id;
             for encountered_id in self.encountered_peers.iter() {
                 if encountered_id == sync_header.sending_component_id {
                     // Don't need to send it to this one
                     continue
+                }
                 let message = SyncMessageFancy{
                     sync_header: self.create_sync_header(ctx),
                     target_component_id: encountered_id,
                     content: SyncContent::Notification,
                 };
                 ctx.submit_message(MessageFancy::Sync(message));
+            }
             // But also send our locally combined solution
             self.forward_local_solutions();
+            self.forward_local_solutions(ctx);
         } else if sync_header.highest_component_id < self.highest_connector_id {
             // Sender has lower leader ID, so it should know about our higher
             // one.
             let message = SyncMessageFancy{
                 sync_header: self.create_sync_header(ctx),
                 target_component_id: sync_header.sending_component_id,
                 content: SyncContent::Notification
             };
             ctx.submit_message(MessageFancy::Sync(message));
         } // else: exactly equal, so do nothing
+    }
     /// Checks data header and consults the stored port mapping and the
     /// execution tree to see which branches may receive the data message's
     /// contents.
     ///
     /// This function is generally called for freshly received messages that
     /// should be matched against previously halted branches.
     /// TODO: Rename, name confused me after a day
     pub fn handle_received_data_header(&mut self, exec_tree: &ExecTree, data_header: &DataHeader, target_ids: &mut Vec<BranchId>) {
         for branch in exec_tree.iter_queue(QueueKind::AwaitingMessage, None) {
             if branch.awaiting_port == data_header.target_port {
                 // Found a branch awaiting the message, but we need to make sure
                 // the mapping is correct
@@ @@ -379,48 +379,122 @@ impl Consensus { @@
     #[inline]
     fn create_sync_header(&self, ctx: &ComponentCtxFancy) -> SyncHeader {
         return SyncHeader{
             sending_component_id: ctx.id,
             highest_component_id: self.highest_connector_id,
+        }
+    }
     fn forward_local_solutions(&mut self, ctx: &mut ComponentCtxFancy) {
         debug_assert_ne!(self.highest_connector_id, ctx.id);
         if !self.local_solutions.is_empty() {
             for local_solution in self.local_solutions.drain() {
                 let message = SyncMessageFancy{
                     sync_header: self.create_sync_header(ctx),
                     target_component_id: self.highest_connector_id,
                     content: SyncContent::LocalSolution(local_solution),
                 };
                 ctx.submit_message(MessageFancy::Sync(message));
+            }
+        }
+    }
+}
 // -----------------------------------------------------------------------------
 // Solution storage and algorithms
 // -----------------------------------------------------------------------------
 struct MatchedLocalSolution {
     final_branch_id: BranchId,
     port_mapping: Vec<(PortIdLocal, BranchId)>,
     matches: Vec<ComponentMatches>,
+}
 struct ComponentMatches {
     target_id: ConnectorId,
     target_index: usize,
     match_indices: Vec<usize>,
     involved_ports: Vec<PortIdLocal>,
+}
 struct ComponentLocalSolutions {
     component: ConnectorId,
     solutions: Vec<MatchedLocalSolution>,
+}
 // TODO: Flatten? Flatten. Flatten everything.
 pub(crate) struct GlobalSolution {
     local: Vec<ComponentLocalSolutions>
+}
 impl GlobalSolution {
     fn new() -> Self {
         return Self{
             local: Vec::new(),
         };
+    }
     fn add_solution(&mut self, solution: LocalSolution) {
         let component_id = solution.component;
         let solution = MatchedLocalSolution{
             final_branch_id: solution.final_branch_id,
             port_mapping: solution.port_mapping,
             matches: Vec::new(),
         };
         // Create an entry for the solution for the particular component
         let component_exists = self.local.iter_mut()
             .enumerate()
             .find(|(_, v)| v.component == component_id);
         let (component_index, solution_index) = match component_exists {
             Some((component_index, storage)) => {
                 // Entry for component exists, so add to solutions
                 let solution_index = storage.solutions.len();
                 storage.solutions.push(solution);
                 (component_index, solution_index)
+            }
             None => {
                 // Entry for component does not exist yet
                 let component_index = self.local.len();
                 self.local.push(ComponentLocalSolutions{
                     component: component_id,
                     solutions: vec![solution],
                 });
                 (component_index, 0)
+            }
         };
         // Compare this new solution to other solutions of different components
         // to see if we get a closed global solution.
+    }
+}
 // -----------------------------------------------------------------------------
 // Generic Helpers
 // -----------------------------------------------------------------------------
 /// Recursively goes through the value group, attempting to find ports.
 /// Duplicates will only be added once.
 pub(crate) fn find_ports_in_value_group(value_group: &ValueGroup, ports: &mut Vec<PortIdLocal>) {
     // Helper to check a value for a port and recurse if needed.
     use crate::protocol::eval::Value;
     fn find_port_in_value(group: &ValueGroup, value: &Value, ports: &mut Vec<PortIdLocal>) {
         match value {
             Value::Input(port_id) | Value::Output(port_id) => {
                 // This is an actual port
                 let cur_port = PortIdLocal::new(port_id.0.u32_suffix);
                 for prev_port in ports.iter() {
                     if *prev_port == cur_port {
                         // Already added
                         return;
+                    }
+                }
                 ports.push(cur_port);
             },
             Value::Array(heap_pos) |
             Value::Message(heap_pos) |
             Value::String(heap_pos) |
             Value::Struct(heap_pos) |

0 comments (0 inline, 0 general)