CSY/reowolf Changeset - 7662b8fb871d · Centrum Wiskunde & Informatica (CWI)

Changeset - 7662b8fb871d

Parent rev.

Child rev.

[Not reviewed]

0 4 0

mh - 4 years ago 2021-11-10 19:22:54
contact@maxhenger.nl

Rewrite solution composition, add some tests

4 files changed with 290 insertions and 102 deletions:

src/runtime2/connector.rs

src/runtime2/consensus.rs

200

src/runtime2/inbox.rs

src/runtime2/tests/mod.rs

0 comments (0 inline, 0 general)

src/runtime2/connector.rs

➞

Show inline comments

@@ @@ -52,295 +52,298 @@ impl ConnectorPublic { @@
             inbox: PublicInbox::new(),
             sleeping: AtomicBool::new(initialize_as_sleeping),
+        }
+    }
+}
 #[derive(Eq, PartialEq)]
 pub(crate) enum ConnectorScheduling {
     Immediate,      // Run again, immediately
     Later,          // Schedule for running, at some later point in time
     NotNow,         // Do not reschedule for running
     Exit,           // Connector has exited
+}
 pub(crate) struct ConnectorPDL {
     tree: ExecTree,
     consensus: Consensus,
+}
 struct ConnectorRunContext<'a> {
     branch_id: BranchId,
     consensus: &'a Consensus,
     received: &'a HashMap<PortIdLocal, ValueGroup>,
     scheduler: SchedulerCtx<'a>,
     prepared_channel: Option<(Value, Value)>,
+}
 impl<'a> RunContext for ConnectorRunContext<'a>{
     fn did_put(&mut self, port: PortId) -> bool {
         let port_id = PortIdLocal::new(port.0.u32_suffix);
         let annotation = self.consensus.get_annotation(self.branch_id, port_id);
         return annotation.registered_id.is_some();
+    }
     fn get(&mut self, port: PortId) -> Option<ValueGroup> {
         let port_id = PortIdLocal::new(port.0.u32_suffix);
         match self.received.get(&port_id) {
             Some(data) => Some(data.clone()),
             None => None,
+        }
+    }
     fn fires(&mut self, port: PortId) -> Option<Value> {
         let port_id = PortIdLocal::new(port.0.u32_suffix);
         let annotation = self.consensus.get_annotation(self.branch_id, port_id);
         return annotation.expected_firing.map(|v| Value::Bool(v));
+    }
     fn get_channel(&mut self) -> Option<(Value, Value)> {
         return self.prepared_channel.take();
+    }
+}
 impl Connector for ConnectorPDL {
     fn run(&mut self, sched_ctx: SchedulerCtx, comp_ctx: &mut ComponentCtx) -> ConnectorScheduling {
         self.handle_new_messages(comp_ctx);
         if self.tree.is_in_sync() {
             let scheduling = self.run_in_sync_mode(sched_ctx, comp_ctx);
             if let Some(solution_branch_id) = self.consensus.handle_new_finished_sync_branches(&self.tree, comp_ctx) {
                 self.collapse_sync_to_solution_branch(solution_branch_id, comp_ctx);
                 return ConnectorScheduling::Immediate;
             } else {
                 return scheduling
+            }
         } else {
             let scheduling = self.run_in_deterministic_mode(sched_ctx, comp_ctx);
             return scheduling;
+        }
+    }
+}
 impl ConnectorPDL {
     pub fn new(initial: ComponentState) -> Self {
         Self{
             tree: ExecTree::new(initial),
             consensus: Consensus::new(),
+        }
+    }
     // --- Handling messages
     pub fn handle_new_messages(&mut self, ctx: &mut ComponentCtx) {
         while let Some(message) = ctx.read_next_message() {
             match message {
                 Message::Data(message) => self.handle_new_data_message(message, ctx),
                 Message::Sync(message) => self.handle_new_sync_message(message, ctx),
                 Message::Control(_) => unreachable!("control message in component"),
+            }
+        }
+    }
     pub fn handle_new_data_message(&mut self, message: DataMessage, ctx: &mut ComponentCtx) {
         // Go through all branches that are awaiting new messages and see if
         // there is one that can receive this message.
         debug_assert!(ctx.workspace_branches.is_empty());
         let mut branches = Vec::new(); // TODO: @Remove
         self.consensus.handle_new_data_message(&self.tree, &message, ctx, &mut branches);
         if !self.consensus.handle_new_data_message(&self.tree, &message, ctx, &mut branches) {
             // Old message, so drop it
             return;
+        }
         for branch_id in branches.drain(..) {
             // This branch can receive, so fork and given it the message
             let receiving_branch_id = self.tree.fork_branch(branch_id);
             self.consensus.notify_of_new_branch(branch_id, receiving_branch_id);
             let receiving_branch = &mut self.tree[receiving_branch_id];
             receiving_branch.insert_message(message.data_header.target_port, message.content.as_message().unwrap().clone());
             self.consensus.notify_of_received_message(receiving_branch_id, &message.data_header, &message.content);
+            self.consensus.notify_of_received_message(receiving_branch_id, &message.sync_header, &message.data_header, &message.content);
             // And prepare the branch for running
             self.tree.push_into_queue(QueueKind::Runnable, receiving_branch_id);
+        }
+    }
     pub fn handle_new_sync_message(&mut self, message: SyncMessage, ctx: &mut ComponentCtx) {
         if let Some(solution_branch_id) = self.consensus.handle_new_sync_message(message, ctx) {
             self.collapse_sync_to_solution_branch(solution_branch_id, ctx);
+        }
+    }
     // --- Running code
     pub fn run_in_sync_mode(&mut self, sched_ctx: SchedulerCtx, comp_ctx: &mut ComponentCtx) -> ConnectorScheduling {
         // Check if we have any branch that needs running
         debug_assert!(self.tree.is_in_sync() && self.consensus.is_in_sync());
         let branch_id = self.tree.pop_from_queue(QueueKind::Runnable);
         if branch_id.is_none() {
             return ConnectorScheduling::NotNow;
+        }
         // Retrieve the branch and run it
         let branch_id = branch_id.unwrap();
         let branch = &mut self.tree[branch_id];
         let mut run_context = ConnectorRunContext{
             branch_id,
             consensus: &self.consensus,
             received: &branch.inbox,
             scheduler: sched_ctx,
             prepared_channel: branch.prepared_channel.take(),
         };
         let run_result = branch.code_state.run(&mut run_context, &sched_ctx.runtime.protocol_description);
         // Handle the returned result. Note that this match statement contains
         // explicit returns in case the run result requires that the component's
         // code is ran again immediately
         match run_result {
             RunResult::BranchInconsistent => {
                 // Branch became inconsistent
                 branch.sync_state = SpeculativeState::Inconsistent;
             },
             RunResult::BranchMissingPortState(port_id) => {
                 // Branch called `fires()` on a port that has not been used yet.
                 let port_id = PortIdLocal::new(port_id.0.u32_suffix);
                 // Create two forks, one that assumes the port will fire, and
                 // one that assumes the port remains silent
                 branch.sync_state = SpeculativeState::HaltedAtBranchPoint;
                 let firing_branch_id = self.tree.fork_branch(branch_id);
                 let silent_branch_id = self.tree.fork_branch(branch_id);
                 self.consensus.notify_of_new_branch(branch_id, firing_branch_id);
                 let _result = self.consensus.notify_of_speculative_mapping(firing_branch_id, port_id, true);
                 debug_assert_eq!(_result, Consistency::Valid);
                 self.consensus.notify_of_new_branch(branch_id, silent_branch_id);
                 let _result = self.consensus.notify_of_speculative_mapping(silent_branch_id, port_id, false);
                 debug_assert_eq!(_result, Consistency::Valid);
                 // Somewhat important: we push the firing one first, such that
                 // that branch is ran again immediately.
                 self.tree.push_into_queue(QueueKind::Runnable, firing_branch_id);
                 self.tree.push_into_queue(QueueKind::Runnable, silent_branch_id);
                 return ConnectorScheduling::Immediate;
             },
             RunResult::BranchMissingPortValue(port_id) => {
                 // Branch performed a `get()` on a port that does not have a
                 // received message on that port.
                 let port_id = PortIdLocal::new(port_id.0.u32_suffix);
                 let consistency = self.consensus.notify_of_speculative_mapping(branch_id, port_id, true);
                 if consistency == Consistency::Valid {
                     // `get()` is valid, so mark the branch as awaiting a message
                     branch.sync_state = SpeculativeState::HaltedAtBranchPoint;
                     branch.awaiting_port = port_id;
                     self.tree.push_into_queue(QueueKind::AwaitingMessage, branch_id);
                     // Note: we only know that a branch is waiting on a message when
                     // it reaches the `get` call. But we might have already received
                     // a message that targets this branch, so check now.
                     let mut any_branch_received = false;
                     for message in comp_ctx.get_read_data_messages(port_id) {
                         if self.consensus.branch_can_receive(branch_id, &message.data_header, &message.content) {
+                        if self.consensus.branch_can_receive(branch_id, &message.sync_header, &message.data_header, &message.content) {
                             // This branch can receive the message, so we do the
                             // fork-and-receive dance
                             let receiving_branch_id = self.tree.fork_branch(branch_id);
                             let branch = &mut self.tree[receiving_branch_id];
                             branch.insert_message(port_id, message.content.as_message().unwrap().clone());
                             self.consensus.notify_of_new_branch(branch_id, receiving_branch_id);
                             self.consensus.notify_of_received_message(receiving_branch_id, &message.data_header, &message.content);
+                            self.consensus.notify_of_received_message(receiving_branch_id, &message.sync_header, &message.data_header, &message.content);
                             self.tree.push_into_queue(QueueKind::Runnable, receiving_branch_id);
                             any_branch_received = true;
+                        }
+                    }
                     if any_branch_received {
                         return ConnectorScheduling::Immediate;
+                    }
                 } else {
                     branch.sync_state = SpeculativeState::Inconsistent;
+                }
+            }
             RunResult::BranchAtSyncEnd => {
                 let consistency = self.consensus.notify_of_finished_branch(branch_id);
                 if consistency == Consistency::Valid {
                     branch.sync_state = SpeculativeState::ReachedSyncEnd;
                     self.tree.push_into_queue(QueueKind::FinishedSync, branch_id);
                 } else if consistency == Consistency::Inconsistent {
                     branch.sync_state = SpeculativeState::Inconsistent;
+                }
             },
             RunResult::BranchPut(port_id, content) => {
                 // Branch is attempting to send data
                 let port_id = PortIdLocal::new(port_id.0.u32_suffix);
                 let consistency = self.consensus.notify_of_speculative_mapping(branch_id, port_id, true);
                 if consistency == Consistency::Valid {
                     // `put()` is valid.
                     let (sync_header, data_header) = self.consensus.handle_message_to_send(branch_id, port_id, &content, comp_ctx);
                     comp_ctx.submit_message(Message::Data(DataMessage {
                         sync_header, data_header,
                         content: DataContent::Message(content),
                     }));
                     self.tree.push_into_queue(QueueKind::Runnable, branch_id);
                     return ConnectorScheduling::Immediate;
                 } else {
                     branch.sync_state = SpeculativeState::Inconsistent;
+                }
             },
             _ => unreachable!("unexpected run result {:?} in sync mode", run_result),
+        }
         // If here then the run result did not require a particular action. We
         // return whether we have more active branches to run or not.
         if self.tree.queue_is_empty(QueueKind::Runnable) {
             return ConnectorScheduling::NotNow;
         } else {
             return ConnectorScheduling::Later;
+        }
+    }
     pub fn run_in_deterministic_mode(&mut self, sched_ctx: SchedulerCtx, comp_ctx: &mut ComponentCtx) -> ConnectorScheduling {
         debug_assert!(!self.tree.is_in_sync() && !self.consensus.is_in_sync());
         let branch = self.tree.base_branch_mut();
         debug_assert!(branch.sync_state == SpeculativeState::RunningNonSync);
         let mut run_context = ConnectorRunContext{
             branch_id: branch.id,
             consensus: &self.consensus,
             received: &branch.inbox,
             scheduler: sched_ctx,
             prepared_channel: branch.prepared_channel.take(),
         };
         let run_result = branch.code_state.run(&mut run_context, &sched_ctx.runtime.protocol_description);
         match run_result {
             RunResult::ComponentTerminated => {
                 branch.sync_state = SpeculativeState::Finished;
                 return ConnectorScheduling::Exit;
             },
             RunResult::ComponentAtSyncStart => {
                 comp_ctx.notify_sync_start();
                 let sync_branch_id = self.tree.start_sync();
                 self.consensus.start_sync(comp_ctx);
                 self.consensus.notify_of_new_branch(BranchId::new_invalid(), sync_branch_id);
                 self.tree.push_into_queue(QueueKind::Runnable, sync_branch_id);
                 return ConnectorScheduling::Immediate;
             },
             RunResult::NewComponent(definition_id, monomorph_idx, arguments) => {
                 // Note: we're relinquishing ownership of ports. But because
                 // we are in non-sync mode the scheduler will handle and check
                 // port ownership transfer.
                 debug_assert!(comp_ctx.workspace_ports.is_empty());
                 find_ports_in_value_group(&arguments, &mut comp_ctx.workspace_ports);
                 let new_state = ComponentState {
                     prompt: Prompt::new(
                         &sched_ctx.runtime.protocol_description.types,
                         &sched_ctx.runtime.protocol_description.heap,
                         definition_id, monomorph_idx, arguments
                     ),
                 };

src/runtime2/consensus.rs

➞

Show inline comments

 use crate::collections::VecSet;
 use crate::protocol::eval::ValueGroup;
 use super::branch::{BranchId, ExecTree, QueueKind};
 use super::ConnectorId;
 use super::port::{ChannelId, PortIdLocal};
 use super::inbox::{
     Message, PortAnnotation,
     DataMessage, DataContent, DataHeader,
     SyncMessage, SyncContent, SyncHeader,
 };
 use super::scheduler::ComponentCtx;
 struct BranchAnnotation {
     port_mapping: Vec<PortAnnotation>,
+}
 #[derive(Debug)]
 pub(crate) struct LocalSolution {
     component: ConnectorId,
     final_branch_id: BranchId,
     port_mapping: Vec<(ChannelId, BranchId)>,
+}
 #[derive(Debug, Clone)]
 pub(crate) struct GlobalSolution {
     component_branches: Vec<(ConnectorId, BranchId)>,
     channel_mapping: Vec<(ChannelId, BranchId)>, // TODO: This can go, is debugging info
+}
 // -----------------------------------------------------------------------------
 // Consensus
 // -----------------------------------------------------------------------------
 struct Peer {
     id: ConnectorId,
     encountered_this_round: bool,
     expected_sync_round: u32,
+}
 /// 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 {
     // --- State that is cleared after each round
     // Local component's state
     highest_connector_id: ConnectorId,
     branch_annotations: Vec<BranchAnnotation>,
     last_finished_handled: Option<BranchId>,
     // Gathered state from communication
     encountered_peers: VecSet<ConnectorId>, // to determine when we should send "found a higher ID" messages.
     encountered_ports: VecSet<PortIdLocal>, // to determine if we should send "port remains silent" messages.
     solution_combiner: SolutionCombiner,
     // --- Persistent state
     // TODO: Tracking sync round numbers
     peers: Vec<Peer>,
     sync_round: u32,
     // --- Workspaces
     workspace_ports: Vec<PortIdLocal>,
+}
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub(crate) enum Consistency {
     Valid,
     Inconsistent,
+}
 impl Consensus {
     pub fn new() -> Self {
         return Self {
             highest_connector_id: ConnectorId::new_invalid(),
             branch_annotations: Vec::new(),
             last_finished_handled: None,
             encountered_peers: VecSet::new(),
             encountered_ports: VecSet::new(),
             solution_combiner: SolutionCombiner::new(),
             peers: Vec::new(),
             sync_round: 0,
             workspace_ports: Vec::new(),
+        }
+    }
     // --- Controlling sync round and branches
     /// Returns whether the consensus algorithm is running in sync mode
     pub fn is_in_sync(&self) -> bool {
         return !self.branch_annotations.is_empty();
+    }
     /// TODO: Remove this once multi-fire is in place
     pub fn get_annotation(&self, branch_id: BranchId, port_id: PortIdLocal) -> &PortAnnotation {
         let branch = &self.branch_annotations[branch_id.index as usize];
         let port = branch.port_mapping.iter().find(|v| v.port_id == port_id).unwrap();
         return port;
+    }
     /// Sets up the consensus algorithm for a new synchronous round. The
     /// provided ports should be the ports the component owns at the start of
     /// the sync round.
     pub fn start_sync(&mut self, ctx: &ComponentCtx) {
         debug_assert!(!self.highest_connector_id.is_valid());
         debug_assert!(self.branch_annotations.is_empty());
         debug_assert!(self.last_finished_handled.is_none());
         debug_assert!(self.encountered_peers.is_empty());
         debug_assert!(self.solution_combiner.local.is_empty());
         // We'll use the first "branch" (the non-sync one) to store our ports,
         // this allows cloning if we created a new branch.
         self.branch_annotations.push(BranchAnnotation{
             port_mapping: ctx.get_ports().iter()
                 .map(|v| PortAnnotation{
                     port_id: v.self_id,
                     registered_id: None,
                     expected_firing: None,
                 })
                 .collect(),
         });
         self.highest_connector_id = ctx.id;
+    }
     /// Notifies the consensus algorithm that a new branch has appeared. Must be
     /// called for each forked branch in the execution tree.
     pub fn notify_of_new_branch(&mut self, parent_branch_id: BranchId, new_branch_id: BranchId) {
         // If called correctly. Then each time we are notified the new branch's
         // index is the length in `branch_annotations`.
         debug_assert!(self.branch_annotations.len() == new_branch_id.index as usize);
         let parent_branch_annotations = &self.branch_annotations[parent_branch_id.index as usize];
         let new_branch_annotations = BranchAnnotation{
             port_mapping: parent_branch_annotations.port_mapping.clone(),
         };
         self.branch_annotations.push(new_branch_annotations);
+    }
     /// Notifies the consensus algorithm that a branch has reached the end of
     /// the sync block. A final check for consistency will be performed that the
     /// caller has to handle. Note that
     pub fn notify_of_finished_branch(&self, branch_id: BranchId) -> Consistency {
         debug_assert!(self.is_in_sync());
         let branch = &self.branch_annotations[branch_id.index as usize];
         for mapping in &branch.port_mapping {
             match mapping.expected_firing {
                 Some(expected) => {
                     if expected != mapping.registered_id.is_some() {
                         // Inconsistent speculative state and actual state
                         debug_assert!(mapping.registered_id.is_none()); // because if we did fire on a silent port, we should've caught that earlier
                         return Consistency::Inconsistent;
+                    }
                 },
                 None => {},
+            }
+        }
         return Consistency::Valid;
+    }
     /// Notifies the consensus algorithm that a particular branch has assumed
     /// a speculative value for its port mapping.
     pub fn notify_of_speculative_mapping(&mut self, branch_id: BranchId, port_id: PortIdLocal, does_fire: bool) -> Consistency {
         debug_assert!(self.is_in_sync());
         let branch = &mut self.branch_annotations[branch_id.index as usize];
         for mapping in &mut branch.port_mapping {
             if mapping.port_id == port_id {
                 match mapping.expected_firing {
                     None => {
                         // Not yet mapped, perform speculative mapping
                         mapping.expected_firing = Some(does_fire);
                         return Consistency::Valid;
                     },
                     Some(current) => {
                         // Already mapped
                         if current == does_fire {
                             return Consistency::Valid;
                         } else {
                             return Consistency::Inconsistent;
+                        }
+                    }
+                }
+            }
+        }
         unreachable!("notify_of_speculative_mapping called with unowned port");
+    }
     /// Generates sync messages for any branches that are at the end of the
     /// sync block. To find these branches, they should've been put in the
     /// "finished" queue in the execution tree.
     pub fn handle_new_finished_sync_branches(&mut self, tree: &ExecTree, ctx: &mut ComponentCtx) -> Option<BranchId> {
         debug_assert!(self.is_in_sync());
         let mut last_branch_id = self.last_finished_handled;
         for branch in tree.iter_queue(QueueKind::FinishedSync, last_branch_id) {
             // Turn the port mapping into a local solution
             let source_mapping = &self.branch_annotations[branch.id.index as usize].port_mapping;
             let mut target_mapping = Vec::with_capacity(source_mapping.len());
             for port in source_mapping {
                 // Note: if the port is silent, and we've never communicated
                 // over the port, then we need to do so now, to let the peer
                 // component know about our sync leader state.
                 let port_desc = ctx.get_port_by_id(port.port_id).unwrap();
                 let peer_port_id = port_desc.peer_id;
                 let channel_id = port_desc.channel_id;
                 if !self.encountered_ports.contains(&port.port_id) {
                     ctx.submit_message(Message::Data(DataMessage {
                         sync_header: SyncHeader{
                             sending_component_id: ctx.id,
                             highest_component_id: self.highest_connector_id,
                             sync_round: self.sync_round
                         },
                         data_header: DataHeader{
                             expected_mapping: source_mapping.clone(),
                             sending_port: port.port_id,
                             target_port: peer_port_id,
                             new_mapping: BranchId::new_invalid(),
                         },
                         content: DataContent::SilentPortNotification,
                     }));
                     self.encountered_ports.push(port.port_id);
+                }
                 target_mapping.push((
                     channel_id,
                     port.registered_id.unwrap_or(BranchId::new_invalid())
                 ));
+            }
             let local_solution = LocalSolution{
                 component: ctx.id,
                 final_branch_id: branch.id,
                 port_mapping: target_mapping,
             };
             let solution_branch = self.send_or_store_local_solution(local_solution, ctx);
             if solution_branch.is_some() {
                 // No need to continue iterating, we've found the solution
                 return solution_branch;
+            }
             last_branch_id = Some(branch.id);
+        }
         self.last_finished_handled = last_branch_id;
         return None;
+    }
     pub fn end_sync(&mut self, branch_id: BranchId, final_ports: &mut Vec<PortIdLocal>) {
         debug_assert!(self.is_in_sync());
         // TODO: Handle sending and receiving ports
         // Set final ports
         final_ports.clear();
         let branch = &self.branch_annotations[branch_id.index as usize];
         for port in &branch.port_mapping {
             final_ports.push(port.port_id);
+        }
         // Clear out internal storage to defaults
         self.highest_connector_id = ConnectorId::new_invalid();
         self.branch_annotations.clear();
         self.last_finished_handled = None;
         self.encountered_peers.clear();
         self.encountered_ports.clear();
         self.solution_combiner.clear();
         self.sync_round += 1;
         for peer in self.peers.iter_mut() {
             peer.encountered_this_round = false;
             peer.expected_sync_round += 1;
+        }
+    }
     // --- Handling messages
     /// Prepares a message for sending. Caller should have made sure that
     /// sending the message is consistent with the speculative state.
     pub fn handle_message_to_send(&mut self, branch_id: BranchId, source_port_id: PortIdLocal, content: &ValueGroup, ctx: &mut ComponentCtx) -> (SyncHeader, DataHeader) {
         debug_assert!(self.is_in_sync());
         let branch = &mut self.branch_annotations[branch_id.index as usize];
         if cfg!(debug_assertions) {
             // Check for consistent mapping
             let port = branch.port_mapping.iter()
                 .find(|v| v.port_id == source_port_id)
                 .unwrap();
             debug_assert!(port.expected_firing == None || port.expected_firing == Some(true));
+        }
         // Check for ports that are being sent
         debug_assert!(self.workspace_ports.is_empty());
         find_ports_in_value_group(content, &mut self.workspace_ports);
         if !self.workspace_ports.is_empty() {
             todo!("handle sending ports");
             self.workspace_ports.clear();
+        }
         // Construct data header
         // TODO: Handle multiple firings. Right now we just assign the current
         //  branch to the `None` value because we know we can only send once.
         debug_assert!(branch.port_mapping.iter().find(|v| v.port_id == source_port_id).unwrap().registered_id.is_none());
         let port_info = ctx.get_port_by_id(source_port_id).unwrap();
         let data_header = DataHeader{
             expected_mapping: branch.port_mapping.clone(),
             sending_port: port_info.self_id,
             target_port: port_info.peer_id,
             new_mapping: branch_id
         };
         // Update port mapping
         for mapping in &mut branch.port_mapping {
             if mapping.port_id == source_port_id {
                 mapping.expected_firing = Some(true);
                 mapping.registered_id = Some(branch_id);
+            }
+        }
         self.encountered_ports.push(source_port_id);
         return (self.create_sync_header(ctx), data_header);
+    }
     /// Handles a new data message by handling the data and sync header, and
     /// checking which *existing* branches *can* receive the message. So two
     /// cautionary notes:
     /// 1. A future branch might also be able to receive this message, see the
     ///     `branch_can_receive` function.
     /// 2. We return the branches that *can* receive the message, you still
     ///     have to explicitly call `notify_of_received_message`.
     pub fn handle_new_data_message(&mut self, exec_tree: &ExecTree, message: &DataMessage, ctx: &mut ComponentCtx, target_ids: &mut Vec<BranchId>) {
         self.handle_received_data_header(exec_tree, &message.data_header, &message.content, target_ids);
         self.handle_received_sync_header(&message.sync_header, ctx);
     pub fn handle_new_data_message(&mut self, exec_tree: &ExecTree, message: &DataMessage, ctx: &mut ComponentCtx, target_ids: &mut Vec<BranchId>) -> bool {
         self.handle_received_data_header(exec_tree, &message.sync_header, &message.data_header, &message.content, target_ids);
         return self.handle_received_sync_header(&message.sync_header, ctx)
+    }
     /// Handles a new sync message by handling the sync header and the contents
     /// of the message. Returns `Some` with the branch ID of the global solution
     /// if the sync solution has been found.
     pub fn handle_new_sync_message(&mut self, message: SyncMessage, ctx: &mut ComponentCtx) -> Option<BranchId> {
         self.handle_received_sync_header(&message.sync_header, ctx);
         if !self.handle_received_sync_header(&message.sync_header, ctx) {
             return None;
+        }
         // And handle the contents
         debug_assert_eq!(message.target_component_id, ctx.id);
         match message.content {
             SyncContent::Notification => {
                 // We were just interested in the header
                 return None;
             },
             SyncContent::LocalSolution(solution) => {
                 // We might be the leader, or earlier messages caused us to not
                 // be the leader anymore.
                 return self.send_or_store_local_solution(solution, ctx);
             },
             SyncContent::GlobalSolution(solution) => {
                 // Take branch of interest and return it.
                 let (_, branch_id) = solution.component_branches.iter()
                     .find(|(connector_id, _)| *connector_id == ctx.id)
                     .unwrap();
                 return Some(*branch_id);
+            }
+        }
+    }
     pub fn notify_of_received_message(&mut self, branch_id: BranchId, data_header: &DataHeader, content: &DataContent) {
         debug_assert!(self.branch_can_receive(branch_id, data_header, content));
     pub fn notify_of_received_message(&mut self, branch_id: BranchId, sync_header: &SyncHeader, data_header: &DataHeader, content: &DataContent) {
         debug_assert!(self.branch_can_receive(branch_id, sync_header, data_header, content));
         let branch = &mut self.branch_annotations[branch_id.index as usize];
         for mapping in &mut branch.port_mapping {
             if mapping.port_id == data_header.target_port {
                 // Found the port in which the message should be inserted
                 mapping.registered_id = Some(data_header.new_mapping);
                 // Check for sent ports
                 debug_assert!(self.workspace_ports.is_empty());
                 find_ports_in_value_group(content.as_message().unwrap(), &mut self.workspace_ports);
                 if !self.workspace_ports.is_empty() {
                     todo!("handle received ports");
                     self.workspace_ports.clear();
+                }
                 return;
+            }
+        }
         // If here, then the branch didn't actually own the port? Means the
         // caller made a mistake
         unreachable!("incorrect notify_of_received_message");
+    }
     /// Matches the mapping between the branch and the data message. If they
     /// match then the branch can receive the message.
     pub fn branch_can_receive(&self, branch_id: BranchId, data_header: &DataHeader, content: &DataContent) -> bool {
     pub fn branch_can_receive(&self, branch_id: BranchId, sync_header: &SyncHeader, data_header: &DataHeader, content: &DataContent) -> bool {
         if let Some(peer) = self.peers.iter().find(|v| v.id == sync_header.sending_component_id) {
             if sync_header.sync_round < peer.expected_sync_round {
                 return false;
+            }
+        }
         if let DataContent::SilentPortNotification = content {
             // No port can receive a "silent" notification.
             return false;
+        }
         let annotation = &self.branch_annotations[branch_id.index as usize];
         for expected in &data_header.expected_mapping {
             // If we own the port, then we have an entry in the
             // annotation, check if the current mapping matches
             for current in &annotation.port_mapping {
                 if expected.port_id == current.port_id {
                     if expected.registered_id != current.registered_id {
                         // IDs do not match, we cannot receive the
                         // message in this branch
                         return false;
+                    }
+                }
+            }
+        }
         return true;
+    }
     // --- Internal helpers
     /// Checks data header and consults the stored port mapping and the
     /// execution tree to see which branches may receive the data message's
     /// contents.
-    fn handle_received_data_header(&mut self, exec_tree: &ExecTree, data_header: &DataHeader, content: &DataContent, target_ids: &mut Vec<BranchId>) {
+    fn handle_received_data_header(&self, exec_tree: &ExecTree, sync_header: &SyncHeader, data_header: &DataHeader, content: &DataContent, 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
                 if self.branch_can_receive(branch.id, data_header, content) {
+                if self.branch_can_receive(branch.id, sync_header, data_header, content) {
                     target_ids.push(branch.id);
+                }
+            }
+        }
+    }
     fn handle_received_sync_header(&mut self, sync_header: &SyncHeader, ctx: &mut ComponentCtx) {
+    fn handle_received_sync_header(&mut self, sync_header: &SyncHeader, ctx: &mut ComponentCtx) -> bool {
         debug_assert!(sync_header.sending_component_id != ctx.id); // not sending to ourselves
         self.encountered_peers.push(sync_header.sending_component_id);
         if !self.handle_peer(sync_header) {
             // We can drop this package
             return false;
+        }
         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 {
             for peer in self.peers.iter() {
                 if peer.id == sync_header.sending_component_id || !peer.encountered_this_round {
                     // Don't need to send it to this one
                     continue
+                }
                 let message = SyncMessage {
                     sync_header: self.create_sync_header(ctx),
-                    target_component_id: *encountered_id,
+                    target_component_id: peer.id,
                     content: SyncContent::Notification,
                 };
                 ctx.submit_message(Message::Sync(message));
+            }
             // But also send our locally combined solution
             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 = SyncMessage {
                 sync_header: self.create_sync_header(ctx),
                 target_component_id: sync_header.sending_component_id,
                 content: SyncContent::Notification
             };
             ctx.submit_message(Message::Sync(message));
         } // else: exactly equal, so do nothing
         return true;
+    }
     /// Handles a (potentially new) peer. Returns `false` if the provided sync
     /// number is different then the expected one.
     fn handle_peer(&mut self, sync_header: &SyncHeader) -> bool {
         let position = self.peers.iter().position(|v| v.id == sync_header.sending_component_id);
         match position {
             Some(index) => {
                 let entry = &mut self.peers[index];
                 entry.encountered_this_round = true;
                 // TODO: Proper handling of potential overflow
                 if sync_header.sync_round >= entry.expected_sync_round {
                     entry.expected_sync_round = sync_header.sync_round;
                     return true;
                 } else {
                     return false;
+                }
             },
             None => {
                 self.peers.push(Peer{
                     id: sync_header.sending_component_id,
                     encountered_this_round: true,
                     expected_sync_round: sync_header.sync_round,
                 });
                 return true;
+            }
+        }
+    }
     fn send_or_store_local_solution(&mut self, solution: LocalSolution, ctx: &mut ComponentCtx) -> Option<BranchId> {
         if self.highest_connector_id == ctx.id {
             // We are the leader
             if let Some(global_solution) = self.solution_combiner.add_solution_and_check_for_global_solution(solution) {
                 let mut my_final_branch_id = BranchId::new_invalid();
                 for (connector_id, branch_id) in global_solution.component_branches.iter().copied() {
                     if connector_id == ctx.id {
                         // This is our solution branch
                         my_final_branch_id = branch_id;
                         continue;
+                    }
                     let message = SyncMessage {
                         sync_header: self.create_sync_header(ctx),
                         target_component_id: connector_id,
                         content: SyncContent::GlobalSolution(global_solution.clone()),
                     };
                     ctx.submit_message(Message::Sync(message));
+                }
                 debug_assert!(my_final_branch_id.is_valid());
                 return Some(my_final_branch_id);
             } else {
                 return None;
+            }
         } else {
             // Someone else is the leader
             let message = SyncMessage {
                 sync_header: self.create_sync_header(ctx),
                 target_component_id: self.highest_connector_id,
                 content: SyncContent::LocalSolution(solution),
             };
             ctx.submit_message(Message::Sync(message));
             return None;
+        }
+    }
     #[inline]
     fn create_sync_header(&self, ctx: &ComponentCtx) -> SyncHeader {
         return SyncHeader{
             sending_component_id: ctx.id,
             highest_component_id: self.highest_connector_id,
             sync_round: self.sync_round,
+        }
+    }
     fn forward_local_solutions(&mut self, ctx: &mut ComponentCtx) {
         debug_assert_ne!(self.highest_connector_id, ctx.id);
         for local_solution in self.solution_combiner.drain() {
             let message = SyncMessage {
                 sync_header: self.create_sync_header(ctx),
                 target_component_id: self.highest_connector_id,
                 content: SyncContent::LocalSolution(local_solution),
             };
             ctx.submit_message(Message::Sync(message));
+        }
+    }
+}
 // -----------------------------------------------------------------------------
 // Solution storage and algorithms
 // -----------------------------------------------------------------------------
 // TODO: Remove all debug derives
 #[derive(Debug)]
 struct MatchedLocalSolution {
     final_branch_id: BranchId,
     channel_mapping: Vec<(ChannelId, BranchId)>,
     matches: Vec<ComponentMatches>,
+}
 #[derive(Debug)]
 struct ComponentMatches {
     target_id: ConnectorId,
     target_index: usize,
     match_indices: Vec<usize>, // of local solution in connector
+}
 #[derive(Debug)]
 struct ComponentPeer {
     target_id: ConnectorId,
     target_index: usize, // in array of global solution components
     involved_channels: Vec<ChannelId>,
+}
 #[derive(Debug)]
 struct ComponentLocalSolutions {
     component: ConnectorId,
     peers: Vec<ComponentPeer>,
     solutions: Vec<MatchedLocalSolution>,
     all_peers_present: bool,
+}
 // TODO: Flatten? Flatten. Flatten everything.
 pub(crate) struct SolutionCombiner {
     local: Vec<ComponentLocalSolutions>
+}
 struct CheckEntry {
     component_index: usize,         // component index in combiner's vector
     solution_index: usize,          // solution entry in the above component entry
     parent_entry_index: usize,      // parent that caused the creation of this checking entry
     match_index_in_parent: usize,   // index in the matches array of the parent
     solution_index_in_parent: usize,// index in the solution array of the match entry in the parent
+}
 impl SolutionCombiner {
     fn new() -> Self {
         return Self{
             local: Vec::new(),
         };
+    }
     /// Adds a new local solution to the global solution storage. Will check the
     /// new local solutions for matching against already stored local solutions
     /// of peer connectors.
     fn add_solution_and_check_for_global_solution(&mut self, solution: LocalSolution) -> Option<GlobalSolution> {
         let component_id = solution.component;
         let solution = MatchedLocalSolution{
             final_branch_id: solution.final_branch_id,
             channel_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, new_component) = 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, false)
+            }
             None => {
                 // Entry for component does not exist yet
                 let component_index = self.local.len();
                 self.local.push(ComponentLocalSolutions{
                     component: component_id,
                     peers: Vec::new(),
                     solutions: vec![solution],
                     all_peers_present: false,
                 });
                 (component_index, 0, true)
+            }
         };
         // If this is a solution of a component that is new to us, then we check
         // in the stored solutions which other components are peers of the new
         // one.
         if new_component {
             let cur_ports = &self.local[component_index].solutions[0].channel_mapping;
             let mut component_peers = Vec::new();
             // Find the matching components
             for (other_index, other_component) in self.local.iter().enumerate() {
                 if other_index == component_index {
                     // Don't match against ourselves
                     continue;
+                }
                 let mut matching_channels = Vec::new();
                 for (cur_channel_id, _) in cur_ports {
                     for (other_channel_id, _) in &other_component.solutions[0].channel_mapping {
                         if cur_channel_id == other_channel_id {
                             // We have a shared port
                             matching_channels.push(*cur_channel_id);
+                        }
+                    }
+                }
                 if !matching_channels.is_empty() {
                     // We share some ports
                     component_peers.push(ComponentPeer{
                         target_id: other_component.component,
                         target_index: other_index,
                         involved_channels: matching_channels,
                     });
+                }
+            }
             let mut num_ports_in_peers = 0;
             for peer in &component_peers {
                 num_ports_in_peers += peer.involved_channels.len();
+            }
             if num_ports_in_peers == cur_ports.len() {
                 // Newly added component has all required peers present
                 self.local[component_index].all_peers_present = true;
+            }
             // Add the found component pairing entries to the solution entries
             // for the two involved components
             for component_match in component_peers {
                 // Check the other component for having all peers present
                 let mut num_ports_in_peers = component_match.involved_channels.len();
                 let other_component = &mut self.local[component_match.target_index];
                 for existing_peer in &other_component.peers {
                     num_ports_in_peers += existing_peer.involved_channels.len();
+                }
@@ @@ -664,295 +731,335 @@ impl SolutionCombiner { @@
         // We're now sure that we know which other components the currently
         // considered component is linked up to. Now we need to check those
         // entries (if any) to see if any pair of local solutions match
         let mut new_component_matches = Vec::new();
         let cur_component = &self.local[component_index];
         let cur_solution = &cur_component.solutions[solution_index];
         for peer in &cur_component.peers {
             let mut new_solution_matches = Vec::new();
             let other_component = &self.local[peer.target_index];
             for (other_solution_index, other_solution) in other_component.solutions.iter().enumerate() {
                 // Check the port mappings between the pair of solutions.
                 let mut all_matched = true;
                 'mapping_check_loop: for (cur_port, cur_branch) in &cur_solution.channel_mapping {
                     for (other_port, other_branch) in &other_solution.channel_mapping {
                         if cur_port == other_port {
                             if cur_branch == other_branch {
                                 // Same port mapping, go to next port
                                 break;
                             } else {
                                 // Different port mapping, not a match
                                 all_matched = false;
                                 break 'mapping_check_loop;
+                            }
+                        }
+                    }
+                }
                 if !all_matched {
                     continue;
+                }
                 // Port mapping between the component pair is the same, so they
                 // have agreeable local solutions
                 new_solution_matches.push(other_solution_index);
+            }
             new_component_matches.push(ComponentMatches{
                 target_id: peer.target_id,
                 target_index: peer.target_index,
                 match_indices: new_solution_matches,
             });
+        }
         // And now that we have the new solution-to-solution matches, we need to
         // add those in the appropriate storage.
         for new_component_match in new_component_matches {
             let other_component = &mut self.local[new_component_match.target_index];
             for other_solution_index in new_component_match.match_indices.iter().copied() {
                 let other_solution = &mut other_component.solutions[other_solution_index];
                 // Add a completely new entry for the component, or add it to
                 // the existing component entry's matches
                 match other_solution.matches.iter_mut()
                     .find(|v| v.target_id == component_id)
+                {
                     Some(other_match) => {
                         other_match.match_indices.push(solution_index);
                     },
                     None => {
                         other_solution.matches.push(ComponentMatches{
                             target_id: component_id,
                             target_index: component_index,
                             match_indices: vec![solution_index],
                         })
+                    }
+                }
+            }
             let cur_component = &mut self.local[component_index];
             let cur_solution = &mut cur_component.solutions[solution_index];
             match cur_solution.matches.iter_mut()
                 .find(|v| v.target_id == new_component_match.target_id)
+            {
                 Some(other_match) => {
                     // Already have an entry
                     debug_assert_eq!(other_match.target_index, new_component_match.target_index);
                     other_match.match_indices.extend(&new_component_match.match_indices);
                 },
                 None => {
                     // Create a new entry
                     cur_solution.matches.push(new_component_match);
+                }
+            }
+        }
         return self.check_new_solution(component_index, solution_index);
+    }
     /// Checks if, starting at the provided local solution, a global solution
     /// can be formed.
     fn check_new_solution(&self, component_index: usize, solution_index: usize) -> Option<GlobalSolution> {
     // TODO: At some point, check if divide and conquer is faster?
     fn check_new_solution(&self, initial_component_index: usize, initial_solution_index: usize) -> Option<GlobalSolution> {
         if !self.can_have_solution() {
             return None;
+        }
         // By now we're certain that all peers are present. So once our
         // backtracking solution stack is as long as the number of components,
         // then we have found a global solution.
         let mut check_stack = Vec::new();
         let mut check_from = 0;
         check_stack.push((component_index, solution_index));
         'checking_loop: while check_from < check_stack.len() {
             // Prepare for next iteration
             let new_check_from = check_stack.len();
             // Go through all entries on the checking stack. Each entry
             // corresponds to a component's solution. We check that one against
             // previously added ones on the stack, and if they're not already
             // added we push them onto the check stack.
             for check_idx in check_from..new_check_from {
                 // Take the current solution
                 let (component_index, solution_index) = check_stack[check_idx];
                 debug_assert!(!self.local[component_index].solutions.is_empty());
                 let cur_solution = &self.local[component_index].solutions[solution_index];
                 // Go through the matches and check if they're on the stack or
                 // should be added to the stack.
                 for cur_match in &cur_solution.matches {
                     let mut is_already_on_stack = false;
                     let mut has_same_solution = false;
                     for existing_check_idx in 0..check_from {
                         let (existing_component_index, existing_solution_index) = check_stack[existing_check_idx];
                         if existing_component_index == cur_match.target_index {
                             // Already lives on the stack, so the match MUST
                             // contain the same solution index if the checked
                             // local solution is agreeable with the (partially
                             // determined) global solution.
                             is_already_on_stack = true;
                             if cur_match.match_indices.contains(&existing_solution_index) {
                                 has_same_solution = true;
                                 break;
         // Construct initial entry on stack
         let mut stack = Vec::with_capacity(self.local.len());
         stack.push(CheckEntry{
             component_index: initial_component_index,
             solution_index: initial_solution_index,
             parent_entry_index: 0,
             match_index_in_parent: 0,
             solution_index_in_parent: 0,
         });
         'check_last_stack: loop {
             let cur_index = stack.len() - 1;
             let cur_entry = &stack[cur_index];
             // Check if the current component is matching with all other entries
             let mut all_match = true;
             'check_against_existing: for prev_index in 0..cur_index {
                 let prev_entry = &stack[prev_index];
                 let prev_component = &self.local[prev_entry.component_index];
                 let prev_solution = &prev_component.solutions[prev_entry.solution_index];
                 for prev_matching_component in &prev_solution.matches {
                     if prev_matching_component.target_index == cur_entry.component_index {
                         // Previous entry has shared ports with the current
                         // entry, so see if we have a composable pair of
                         // solutions.
                         if !prev_matching_component.match_indices.contains(&cur_entry.solution_index) {
                             all_match = false;
                             break 'check_against_existing;
+                        }
+                    }
+                }
+            }
                     if is_already_on_stack {
                         if !has_same_solution {
                             // We have an inconsistency, so we need to go back
                             // in our stack, and try the next solution
                             let (last_component_index, last_solution_index) = check_stack[check_from];
                             check_stack.truncate(check_from);
                             if check_stack.is_empty() {
                                 // The starting point does not yield a valid
                                 // solution
                                 return None;
             if all_match {
                 // All components matched until now.
                 if stack.len() == self.local.len() {
                     // We have found a global solution
                     break 'check_last_stack;
+                }
                             // Try the next one
                             let last_component = &self.local[last_component_index];
                             let new_solution_index = last_solution_index + 1;
                             if new_solution_index >= last_component.solutions.len() {
                                 // No more things to try, again: no valid
                                 // solution
                                 return None;
                 // Not all components found yet, look for a new one that has not
                 // yet been added yet.
                 for (parent_index, parent_entry) in stack.iter().enumerate() {
                     let parent_component = &self.local[parent_entry.component_index];
                     let parent_solution = &parent_component.solutions[parent_entry.solution_index];
                     for (peer_index, peer_component) in parent_solution.matches.iter().enumerate() {
                         if peer_component.match_indices.is_empty() {
                             continue;
+                        }
                             check_stack.push((last_component_index, new_solution_index));
                             continue 'checking_loop;
                         } // else: we're fine, the solution is agreeable
                     } else {
                         check_stack.push((cur_match.target_index, 0))
                         let already_added = stack.iter().any(|v| v.component_index == peer_component.target_index);
                         if !already_added {
                             // New component to try
                             stack.push(CheckEntry{
                                 component_index: peer_component.target_index,
                                 solution_index: peer_component.match_indices[0],
                                 parent_entry_index: parent_index,
                                 match_index_in_parent: peer_index,
                                 solution_index_in_parent: 0,
                             });
                             continue 'check_last_stack;
+                        }
+                    }
+                }
                 // Cannot find a peer to add. This is possible if, for example,
                 // we have a component A which has the only connection to
                 // component B. And B has sent a local solution saying it is
                 // finished, but the last data message has not yet arrived at A.
                 // In any case, we just exit the if statement and handle not
                 // being able to find a new connector as being forced to try a
                 // new permutation of possible local solutions.
+            }
             // Either the currently considered local solution is inconsistent
             // with other local solutions, or we cannot find a new component to
             // add. This is where we perform backtracking as long as needed to
             // try a new solution.
             while stack.len() > 1 {
                 // Check if our parent has another solution we can try
                 let cur_index = stack.len() - 1;
                 let cur_entry = &stack[cur_index];
                 let parent_entry = &stack[cur_entry.parent_entry_index];
                 let parent_component = &self.local[parent_entry.component_index];
                 let parent_solution = &parent_component.solutions[parent_entry.solution_index];
                 let match_component = &parent_solution.matches[cur_entry.match_index_in_parent];
                 debug_assert!(match_component.target_index == cur_entry.component_index);
                 let new_solution_index_in_parent = cur_entry.solution_index_in_parent + 1;
                 if new_solution_index_in_parent < match_component.match_indices.len() {
                     // We can still try a new one
                     let new_solution_index = match_component.match_indices[new_solution_index_in_parent];
                     let cur_entry = &mut stack[cur_index];
                     cur_entry.solution_index_in_parent = new_solution_index_in_parent;
                     cur_entry.solution_index = new_solution_index;
                     continue 'check_last_stack;
                 } else {
                     // We're out of options here. So pop an entry, then in
                     // the next iteration of this backtracking loop we try
                     // to increment that solution
                     stack.pop();
+                }
+            }
             check_from = new_check_from;
             // Stack length is 1, hence we're back at our initial solution.
             // Since that doesn't yield a global solution, we simply:
             return None;
+        }
         // Because of our earlier checking if we can have a solution at
         // all (all components have their peers), and the exit condition of the
         // while loop: if we're here, then we have a global solution
         debug_assert_eq!(check_stack.len(), self.local.len());
         let mut final_branches = Vec::with_capacity(check_stack.len());
         for (component_index, solution_index) in check_stack.iter().copied() {
             let component = &self.local[component_index];
             let solution = &component.solutions[solution_index];
         // Constructing the representation of the global solution
         debug_assert_eq!(stack.len(), self.local.len());
         let mut final_branches = Vec::with_capacity(stack.len());
         for entry in &stack {
             let component = &self.local[entry.component_index];
             let solution = &component.solutions[entry.solution_index];
             final_branches.push((component.component, solution.final_branch_id));
+        }
         // Just debugging here, TODO: @remove
         let mut total_num_channels = 0;
         for (component_index, _) in check_stack.iter().copied() {
             let component = &self.local[component_index];
         for entry in &stack {
             let component = &self.local[entry.component_index];
             total_num_channels += component.solutions[0].channel_mapping.len();
+        }
         total_num_channels /= 2;
         let mut final_mapping = Vec::with_capacity(total_num_channels);
         let mut total_num_checked = 0;
         for (component_index, solution_index) in check_stack.iter().copied() {
             let component = &self.local[component_index];
             let solution = &component.solutions[solution_index];
         for entry in &stack {
             let component = &self.local[entry.component_index];
             let solution = &component.solutions[entry.solution_index];
             for (channel_id, branch_id) in solution.channel_mapping.iter().copied() {
                 match final_mapping.iter().find(|(v, _)| *v == channel_id) {
                     Some((_, encountered_branch_id)) => {
                         debug_assert_eq!(*encountered_branch_id, branch_id);
                         total_num_checked += 1;
                     },
                     None => {
                         final_mapping.push((channel_id, branch_id));
+                    }
+                }
+            }
+        }
         debug_assert_eq!(total_num_checked, total_num_channels);
         return Some(GlobalSolution{
             component_branches: final_branches,
             channel_mapping: final_mapping,
         });
+    }
     /// Simple test if a solution is at all possible. If this returns true it
     /// does not mean there actually is a solution.
     fn can_have_solution(&self) -> bool {
         for component in &self.local {
             if !component.all_peers_present {
                 return false;
+            }
+        }
         return true;
+    }
     /// Turns the entire (partially resolved) global solution back into local
     /// solutions to ship to another component.
     // TODO: Don't do this, kind of wasteful since a lot of processing has
     //  already been performed.
     fn drain(&mut self) -> Vec<LocalSolution> {
         let mut reserve_len = 0;
         for component in &self.local {
             reserve_len += component.solutions.len();
+        }
         let mut solutions = Vec::with_capacity(reserve_len);
         for component in self.local.drain(..) {
             for solution in component.solutions {
                 solutions.push(LocalSolution{
                     component: component.component,
                     final_branch_id: solution.final_branch_id,
                     port_mapping: solution.channel_mapping,
                 });
+            }
+        }
         return solutions;
+    }
     fn clear(&mut self) {
         self.local.clear();
+    }
+}
 // -----------------------------------------------------------------------------
 // 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) |
             Value::Union(_, heap_pos) => {
                 // Reference to some dynamic thing which might contain ports,
                 // so recurse
                 let heap_region = &group.regions[*heap_pos as usize];
                 for embedded_value in heap_region {

src/runtime2/inbox.rs

➞

Show inline comments

 use std::sync::Mutex;
 use std::collections::VecDeque;
 use crate::protocol::eval::ValueGroup;
 use super::ConnectorId;
 use super::branch::BranchId;
 use super::consensus::{GlobalSolution, LocalSolution};
 use super::port::PortIdLocal;
 // TODO: Remove Debug derive from all types
 #[derive(Debug, Copy, Clone)]
 pub(crate) struct PortAnnotation {
     pub port_id: PortIdLocal,
     pub registered_id: Option<BranchId>,
     pub expected_firing: Option<bool>,
+}
 /// The header added by the synchronization algorithm to all.
 #[derive(Debug, Clone)]
 pub(crate) struct SyncHeader {
     pub sending_component_id: ConnectorId,
     pub highest_component_id: ConnectorId,
     pub sync_round: u32,
+}
 /// The header added to data messages
 #[derive(Debug, Clone)]
 pub(crate) struct DataHeader {
     pub expected_mapping: Vec<PortAnnotation>,
     pub sending_port: PortIdLocal,
     pub target_port: PortIdLocal,
     pub new_mapping: BranchId,
+}
 // TODO: Very much on the fence about this. On one hand I thought making it a
 //  data message was neat because "silent port notification" should be rerouted
 //  like any other data message to determine the component ID of the receiver
 //  and to make it part of the leader election algorithm for the sync leader.
 //  However: it complicates logic quite a bit. Really it might be easier to
 //  create `Message::SyncAtComponent` and `Message::SyncAtPort` messages...
 #[derive(Debug, Clone)]
 pub(crate) enum DataContent {
     SilentPortNotification,
     Message(ValueGroup),
+}
 impl DataContent {
     pub(crate) fn as_message(&self) -> Option<&ValueGroup> {
         match self {
             DataContent::SilentPortNotification => None,
             DataContent::Message(message) => Some(message),
+        }
+    }
+}
 /// A data message is a message that is intended for the receiver's PDL code,
 /// but will also be handled by the consensus algorithm
 #[derive(Debug, Clone)]
 pub(crate) struct DataMessage {
     pub sync_header: SyncHeader,
     pub data_header: DataHeader,
     pub content: DataContent,
+}
 #[derive(Debug)]
 pub(crate) enum SyncContent {
     LocalSolution(LocalSolution), // sending a local solution to the leader
     GlobalSolution(GlobalSolution), // broadcasting to everyone
     Notification, // just a notification (so purpose of message is to send the SyncHeader)
+}
 /// A sync message is a message that is intended only for the consensus
 /// algorithm.
 #[derive(Debug)]
 pub(crate) struct SyncMessage {
     pub sync_header: SyncHeader,
     pub target_component_id: ConnectorId,
     pub content: SyncContent,
+}
 /// A control message is a message intended for the scheduler that is executing
 /// a component.
 #[derive(Debug)]
 pub(crate) struct ControlMessage {
     pub id: u32, // generic identifier, used to match request to response
     pub sending_component_id: ConnectorId,
     pub content: ControlContent,
+}
 #[derive(Debug)]
 pub(crate) enum ControlContent {
     PortPeerChanged(PortIdLocal, ConnectorId),
     CloseChannel(PortIdLocal),
     Ack,
     Ping,
+}
 /// Combination of data message and control messages.
 #[derive(Debug)]
 pub(crate) enum Message {
     Data(DataMessage),
     Sync(SyncMessage),
     Control(ControlMessage),
+}
 /// The public inbox of a connector. The thread running the connector that owns
 /// this inbox may retrieved from it. Non-owning threads may only put new
 /// messages inside of it.
 // TODO: @Optimize, lazy concurrency. Probably ringbuffer with read/write heads.
 //  Should behave as a MPSC queue.
 pub struct PublicInbox {
     messages: Mutex<VecDeque<Message>>,
+}
 impl PublicInbox {
     pub fn new() -> Self {
         Self{
             messages: Mutex::new(VecDeque::new()),
+        }

src/runtime2/tests/mod.rs

➞

Show inline comments

 use super::*;
 use crate::{PortId, ProtocolDescription};
 use crate::common::Id;
 use crate::protocol::eval::*;
 const NUM_THREADS: u32 = 1;  // number of threads in runtime
 const NUM_INSTANCES: u32 = 1;  // number of test instances constructed
 const NUM_LOOPS: u32 = 1500;      // number of loops within a single test (not used by all tests)
 const NUM_THREADS: u32 = 3;     // number of threads in runtime
 const NUM_INSTANCES: u32 = 5;   // number of test instances constructed
 const NUM_LOOPS: u32 = 5;       // number of loops within a single test (not used by all tests)
 fn create_runtime(pdl: &str) -> Runtime {
     let protocol = ProtocolDescription::parse(pdl.as_bytes()).expect("parse pdl");
     let runtime = Runtime::new(NUM_THREADS, protocol);
     return runtime;
+}
 fn run_test_in_runtime<F: Fn(&mut ApplicationInterface)>(pdl: &str, constructor: F) {
     let protocol = ProtocolDescription::parse(pdl.as_bytes())
         .expect("parse PDL");
     let runtime = Runtime::new(NUM_THREADS, protocol);
     let mut api = runtime.create_interface();
     for _ in 0..NUM_INSTANCES {
         constructor(&mut api);
+    }
     // Wait until done :)
+}
 struct TestTimer {
     name: &'static str,
     started: std::time::Instant
+}
 impl TestTimer {
     fn new(name: &'static str) -> Self {
         Self{ name, started: std::time::Instant::now() }
+    }
+}
 impl Drop for TestTimer {
     fn drop(&mut self) {
         let delta = std::time::Instant::now() - self.started;
         let nanos = (delta.as_secs_f64() * 1_000_000.0) as u64;
         let millis = nanos / 1000;
         let nanos = nanos % 1000;
         println!("[{}] Took {:>4}.{:03} ms", self.name, millis, nanos);
+    }
+}
 #[test]
 fn test_put_and_get() {
     const CODE: &'static str = "
     primitive putter(out<bool> sender, u32 loops) {
         u32 index = 0;
         while (index < loops) {
             synchronous {
                 put(sender, true);
+            }
             index += 1;
+        }
+    }
     primitive getter(in<bool> receiver, u32 loops) {
         u32 index = 0;
         while (index < loops) {
             synchronous {
                 auto result = get(receiver);
                 assert(result);
+            }
             index += 1;
+        }
+    }
     ";
     let thing = TestTimer::new("put_and_get");
     run_test_in_runtime(CODE, |api| {
         let channel = api.create_channel();
         api.create_connector("", "putter", ValueGroup::new_stack(vec![
             Value::Output(PortId(Id{ connector_id: 0, u32_suffix: channel.putter_id.index })),
             Value::UInt32(NUM_LOOPS)
         ])).expect("create putter");
         api.create_connector("", "getter", ValueGroup::new_stack(vec![
             Value::Input(PortId(Id{ connector_id: 0, u32_suffix: channel.getter_id.index })),
             Value::UInt32(NUM_LOOPS)
         ])).expect("create getter");
     });
+}
 #[test]
 fn test_star_shaped_request() {
     const CODE: &'static str = "
     primitive edge(in<u32> input, out<u32> output, u32 loops) {
         u32 index = 0;
         while (index < loops) {
             synchronous {
                 auto req = get(input);
                 put(output, req * 2);
+            }
             index += 1;
+        }
+    }
     primitive center(out<u32>[] requests, in<u32>[] responses, u32 loops) {
         u32 loop_index = 0;
         auto num_edges = length(requests);
         while (loop_index < loops) {
             print(\"starting loop\");
+            // print(\"starting loop\");
             synchronous {
                 u32 edge_index = 0;
                 u32 sum = 0;
                 while (edge_index < num_edges) {
                     put(requests[edge_index], edge_index);
                     auto response = get(responses[edge_index]);
                     sum += response;
                     edge_index += 1;
+                }
                 assert(sum == num_edges * (num_edges - 1));
+            }
             print(\"ending loop\");
+            // print(\"ending loop\");
             loop_index += 1;
+        }
+    }
     composite constructor(u32 num_edges, u32 num_loops) {
         auto requests = {};
         auto responses = {};
         u32 edge_index = 0;
         while (edge_index < num_edges) {
             channel req_put -> req_get;
             channel resp_put -> resp_get;
             new edge(req_get, resp_put, num_loops);
             requests @= { req_put };
             responses @= { resp_get };
             edge_index += 1;
+        }
         new center(requests, responses, num_loops);
+    }
     ";
     let thing = TestTimer::new("star_shaped_request");
     run_test_in_runtime(CODE, |api| {
         api.create_connector("", "constructor", ValueGroup::new_stack(vec![
             Value::UInt32(5),
             Value::UInt32(NUM_LOOPS),
         ]));
     });
+}
 #[test]
 fn test_conga_line_request() {
     const CODE: &'static str = "
     primitive start(out<u32> req, in<u32> resp, u32 num_nodes, u32 num_loops) {
         u32 loop_index = 0;
         u32 initial_value = 1337;
         while (loop_index < num_loops) {
             synchronous {
                 put(req, initial_value);
                 auto result = get(resp);
                 assert(result == initial_value + num_nodes * 2);
+            }
             loop_index += 1;
+        }
+    }
     primitive middle(
         in<u32> req_in, out<u32> req_forward,
         in<u32> resp_in, out<u32> resp_forward,
         u32 num_loops
     ) {
         u32 loop_index = 0;
         while (loop_index < num_loops) {
             synchronous {
                 auto req = get(req_in);
                 put(req_forward, req + 1);
                 auto resp = get(resp_in);
                 put(resp_forward, resp + 1);
+            }
             loop_index += 1;
+        }
+    }
     primitive end(in<u32> req_in, out<u32> resp_out, u32 num_loops) {
         u32 loop_index = 0;
         while (loop_index < num_loops) {
             synchronous {
                 auto req = get(req_in);
                 put(resp_out, req);
+            }
             loop_index += 1;
+        }
+    }
     composite constructor(u32 num_nodes, u32 num_loops) {
         channel initial_req -> req_in;
         channel resp_out -> final_resp;
         new start(initial_req, final_resp, num_nodes, num_loops);
         in<u32> last_req_in = req_in;
         out<u32> last_resp_out = resp_out;
         u32 node = 0;
         while (node < num_nodes) {
             channel new_req_fw -> new_req_in;
             channel new_resp_out -> new_resp_in;
             new middle(last_req_in, new_req_fw, new_resp_in, last_resp_out, num_loops);
             last_req_in = new_req_in;
             last_resp_out = new_resp_out;
             node += 1;
+        }
         new end(last_req_in, last_resp_out, num_loops);
+    }
     ";
     let thing = TestTimer::new("conga_line_request");
     run_test_in_runtime(CODE, |api| {
         api.create_connector("", "constructor", ValueGroup::new_stack(vec![
             Value::UInt32(5),
             Value::UInt32(NUM_LOOPS)
         ]));
     });
+}
@@ \ No newline at end of file @@

0 comments (0 inline, 0 general)