CSY/reowolf Changeset - 18c70be7107f · Centrum Wiskunde & Informatica (CWI)

Changeset - 18c70be7107f

Parent rev.

Child rev.

[Not reviewed]

0 8 0

mh - 4 years ago 2021-11-25 12:38:15
contact@maxhenger.nl

WIP on fixing delayed message bug

8 files changed with 397 insertions and 130 deletions:

src/collections/raw_vec.rs

src/runtime2/connector.rs

src/runtime2/consensus.rs

src/runtime2/inbox.rs

src/runtime2/native.rs

src/runtime2/scheduler.rs

240

src/runtime2/tests/mod.rs

src/runtime2/tests/network_shapes.rs

0 comments (0 inline, 0 general)

src/collections/raw_vec.rs

➞

Show inline comments

 use std::{mem, ptr, cmp};
 use std::alloc::{Layout, alloc, dealloc};
 #[derive(Debug)]
 enum AllocError {
     CapacityOverflow,
+}
 /// Generic raw vector. It has a base pointer, a capacity and a length. Basic
 /// operations are supported, but the user of the structure is responsible for
 /// ensuring that no illegal mutable access occurs.
 /// A lot of the logic is simply stolen from the std lib. The destructor will
 /// free the backing memory, but will not run any destructors.
 /// Try to use functions to modify the length. But feel free if you know what
 /// you're doing
 pub struct RawVec<T: Sized> {
     base: *mut T,
     cap: usize,
     len: usize,
+    pub len: usize,
+}
 impl<T: Sized> RawVec<T> {
     const T_ALIGNMENT: usize = mem::align_of::<T>();
     const T_SIZE: usize = mem::size_of::<T>();
     const GROWTH_RATE: usize = 2;
     pub fn new() -> Self {
         Self{
             base: ptr::null_mut(),
             cap: 0,
             len: 0,
+        }
+    }
     pub fn with_capacity(capacity: usize) -> Self {
         // Could be done a bit more efficiently
         let mut result = Self::new();
         result.ensure_space(capacity).unwrap();
         return result;
+    }
     #[inline]
     pub unsafe fn get(&self, idx: usize) -> *const T {
         debug_assert!(idx < self.len);
         return self.base.add(idx);
+    }
     #[inline]
     pub unsafe fn get_mut(&self, idx: usize) -> *mut T {
         debug_assert!(idx < self.len);
         return self.base.add(idx);
+    }
     /// Pushes a new element to the end of the list.
     pub fn push(&mut self, item: T) {
         self.ensure_space(1).unwrap();
         unsafe {
             let target = self.base.add(self.len);
             std::ptr::write(target, item);
             self.len += 1;
+        }
+    }
     /// Moves the elements in the range [from_idx, from_idx + num_to_move) to
     /// the range [to_idx, to_idx + num_to_move). Caller must make sure that all
     /// non-overlapping elements of the second range had their destructor called
     /// in case those elements were used.
     pub fn move_range(&mut self, from_idx: usize, to_idx: usize, num_to_move: usize) {
         debug_assert!(from_idx + num_to_move <= self.len);
         debug_assert!(to_idx + num_to_move <= self.len); // maybe not in future, for now this is fine
         unsafe {
             let source = self.base.add(from_idx);
             let target = self.base.add(to_idx);
             std::ptr::copy(source, target, num_to_move);
+        }
+    }
     pub fn len(&self) -> usize {
         return self.len;
+    }
     pub fn as_slice(&self) -> &[T] {
         return unsafe{
             std::slice::from_raw_parts(self.base, self.len)
         };
+    }
     fn ensure_space(&mut self, additional: usize) -> Result<(), AllocError>{
         debug_assert!(Self::T_SIZE != 0);
         debug_assert!(self.cap >= self.len);
         if self.cap - self.len < additional {
             // Need to resize. Note that due to all checked conditions we have
             // that new_cap >= 1.
             debug_assert!(additional > 0);
             let new_cap = self.len.checked_add(additional).unwrap();
             let new_cap = cmp::max(new_cap, self.cap * Self::GROWTH_RATE);
             let layout = Layout::array::<T>(new_cap)
                 .map_err(|_| AllocError::CapacityOverflow)?;
             debug_assert_eq!(new_cap * Self::T_SIZE, layout.size());
             unsafe {
                 // Allocate new storage, transfer bits, deallocate old store
                 let new_base = alloc(layout);
                 if self.cap > 0 {
                     let old_base = self.base as *mut u8;
                     let (old_size, old_layout) = self.current_layout();
                     ptr::copy_nonoverlapping(old_base, new_base, old_size);
                     dealloc(old_base, old_layout);
+                }
                 self.base = new_base as *mut T;
                 self.cap = new_cap;
+            }
         } // else: still enough space
         return Ok(());
+    }
     #[inline]
     fn current_layout(&self) -> (usize, Layout) {
         debug_assert!(Self::T_SIZE > 0);
         let old_size = self.cap * Self::T_SIZE;
         unsafe {
             return (
                 old_size,
                 Layout::from_size_align_unchecked(old_size, Self::T_ALIGNMENT)
             );
+        }
+    }
+}
 impl<T: Sized> Drop for RawVec<T> {
     fn drop(&mut self) {
         if self.cap > 0 {
             debug_assert!(!self.base.is_null());
             let (_, layout) = self.current_layout();
             unsafe {
                 dealloc(self.base as *mut u8, layout);
                 if cfg!(debug_assertions) {
                     self.base = ptr::null_mut();
+                }
+            }
+        }
+    }
+}
@@ \ No newline at end of file @@

src/runtime2/connector.rs

➞

Show inline comments

 // connector.rs
 //
 // Represents a component. A component (and the scheduler that is running it)
 // has many properties that are not easy to subdivide into aspects that are
 // conceptually handled by particular data structures. That is to say: the code
 // that we run governs: running PDL code, keeping track of ports, instantiating
 // new components and transports (i.e. interacting with the runtime), running
 // a consensus algorithm, etc. But on the other hand, our data is rather
 // simple: we have a speculative execution tree, a set of ports that we own,
 // and a bit of code that we should run.
 //
 // So currently the code is organized as following:
 // - The scheduler that is running the component is the authoritative source on
 //     ports during *non-sync* mode. The consensus algorithm is the
 //     authoritative source during *sync* mode. They retrieve each other's
 //     state during the transitions. Hence port data exists duplicated between
 //     these two datastructures.
 // - The execution tree is where executed branches reside. But the execution
 //     tree is only aware of the tree shape itself (and keeps track of some
 //     queues of branches that are in a particular state), and tends to store
 //     the PDL program state. The consensus algorithm is also somewhat aware
 //     of the execution tree, but only in terms of what is needed to complete
 //     a sync round (for now, that means the port mapping in each branch).
 //     Hence once more we have properties conceptually associated with branches
 //     in two places.
 // - TODO: Write about handling messages, consensus wrapping data
 // - TODO: Write about way information is exchanged between PDL/component and scheduler through ctx
 use std::collections::HashMap;
 use std::sync::atomic::AtomicBool;
 use crate::{PortId, ProtocolDescription};
 use crate::common::ComponentState;
 use crate::protocol::eval::{EvalContinuation, EvalError, Prompt, Value, ValueGroup};
 use crate::protocol::{RunContext, RunResult};
 use super::branch::{BranchId, ExecTree, QueueKind, SpeculativeState, PreparedStatement};
 use super::consensus::{Consensus, Consistency, RoundConclusion, find_ports_in_value_group};
 use super::inbox::{DataMessage, Message, SyncCompMessage, SyncPortMessage, SyncControlMessage, PublicInbox};
 use super::native::Connector;
 use super::port::{PortKind, PortIdLocal};
 use super::scheduler::{ComponentCtx, SchedulerCtx};
+use super::scheduler::{ComponentCtx, SchedulerCtx, MessageTicket};
 pub(crate) struct ConnectorPublic {
     pub inbox: PublicInbox,
     pub sleeping: AtomicBool,
+}
 impl ConnectorPublic {
     pub fn new(initialize_as_sleeping: bool) -> Self {
         ConnectorPublic{
             inbox: PublicInbox::new(),
             sleeping: AtomicBool::new(initialize_as_sleeping),
+        }
+    }
+}
 #[derive(Debug, PartialEq, Eq, Clone, Copy)]
 enum Mode {
     NonSync,    // running non-sync code
     Sync,       // running sync code (in potentially multiple branches)
     SyncError,  // encountered an unrecoverable error in sync mode
     Error,      // encountered an error in non-sync mode (or finished handling the sync mode error).
+}
 #[derive(Debug)]
 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 {
     mode: Mode,
     eval_error: Option<EvalError>,
     tree: ExecTree,
     consensus: Consensus,
     last_finished_handled: Option<BranchId>,
+}
 // TODO: Remove remaining fields once 'fires()' is removed from language.
 struct ConnectorRunContext<'a> {
     branch_id: BranchId,
     consensus: &'a Consensus,
     prepared: PreparedStatement,
+}
 impl<'a> RunContext for ConnectorRunContext<'a>{
     fn performed_put(&mut self, _port: PortId) -> bool {
         return match self.prepared.take() {
             PreparedStatement::None => false,
             PreparedStatement::PerformedPut => true,
             taken => unreachable!("prepared statement is '{:?}' during 'performed_put()'", taken)
         };
+    }
     fn performed_get(&mut self, _port: PortId) -> Option<ValueGroup> {
         return match self.prepared.take() {
             PreparedStatement::None => None,
             PreparedStatement::PerformedGet(value) => Some(value),
             taken => unreachable!("prepared statement is '{:?}' during 'performed_get()'", taken),
         };
+    }
     fn fires(&mut self, port: PortId) -> Option<Value> {
         todo!("Remove fires() now");
         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 created_channel(&mut self) -> Option<(Value, Value)> {
         return match self.prepared.take() {
             PreparedStatement::None => None,
             PreparedStatement::CreatedChannel(ports) => Some(ports),
             taken => unreachable!("prepared statement is '{:?}' during 'created_channel()'", taken),
         };
+    }
     fn performed_fork(&mut self) -> Option<bool> {
         return match self.prepared.take() {
             PreparedStatement::None => None,
             PreparedStatement::ForkedExecution(path) => Some(path),
             taken => unreachable!("prepared statement is '{:?}' during 'performed_fork()'", taken),
         };
+    }
+}
 impl Connector for ConnectorPDL {
     fn run(&mut self, sched_ctx: SchedulerCtx, comp_ctx: &mut ComponentCtx) -> ConnectorScheduling {
         if let Some(scheduling) = self.handle_new_messages(comp_ctx) {
             return scheduling;
+        }
         match self.mode {
             Mode::Sync => {
                 // Run in sync mode
                 let scheduling = self.run_in_sync_mode(sched_ctx, comp_ctx);
                 // Handle any new finished branches
                 let mut iter_id = self.last_finished_handled.or(self.tree.get_queue_first(QueueKind::FinishedSync));
                 while let Some(branch_id) = iter_id {
                     iter_id = self.tree.get_queue_next(branch_id);
                     self.last_finished_handled = Some(branch_id);
                     if let Some(round_conclusion) = self.consensus.handle_new_finished_sync_branch(branch_id, comp_ctx) {
                         // Actually found a solution
                         return self.enter_non_sync_mode(round_conclusion, comp_ctx);
+                    }
                     self.last_finished_handled = Some(branch_id);
+                }
                 return scheduling;
             },
             Mode::NonSync => {
                 let scheduling = self.run_in_deterministic_mode(sched_ctx, comp_ctx);
                 return scheduling;
             },
             Mode::SyncError => {
                 let scheduling = self.run_in_sync_mode(sched_ctx, comp_ctx);
                 return scheduling;
             },
             Mode::Error => {
                 // This shouldn't really be called. Because when we reach exit
                 // mode the scheduler should not run the component anymore
                 unreachable!("called component run() during error-mode");
             },
+        }
+    }
+}
 impl ConnectorPDL {
     pub fn new(initial: Prompt) -> Self {
         Self{
             mode: Mode::NonSync,
             eval_error: None,
             tree: ExecTree::new(initial),
             consensus: Consensus::new(),
             last_finished_handled: None,
+        }
+    }
     // --- Handling messages
     pub fn handle_new_messages(&mut self, ctx: &mut ComponentCtx) -> Option<ConnectorScheduling> {
         while let Some(message) = ctx.read_next_message() {
             match message {
                 Message::Data(message) => self.handle_new_data_message(message, ctx),
         while let Some(ticket) = ctx.get_next_message_ticket() {
             let message = ctx.read_message_using_ticket(ticket);
             let immediate_result = if let Message::Data(_) = message {
                 self.handle_new_data_message(ticket, ctx);
                 None
             } else {
                 match ctx.take_message_using_ticket(ticket) {
                     Message::Data(_) => unreachable!(),
                     Message::SyncComp(message) => {
                     if let Some(result) = self.handle_new_sync_comp_message(message, ctx) {
                         return Some(result);
+                    }
                         self.handle_new_sync_comp_message(message, ctx)
                     },
                     Message::SyncPort(message) => {
                         self.handle_new_sync_port_message(message, ctx);
                         None
                     },
                 Message::SyncPort(message) => self.handle_new_sync_port_message(message, ctx),
                     Message::SyncControl(message) => {
                     if let Some(result) = self.handle_new_sync_control_message(message, ctx) {
                         return Some(result);
+                    }
                         self.handle_new_sync_control_message(message, ctx)
                     },
                     Message::Control(_) => unreachable!("control message in component"),
+                }
             };
             if let Some(result) = immediate_result {
                 return Some(result);
+            }
+        }
         return None;
+    }
-    pub fn handle_new_data_message(&mut self, message: DataMessage, ctx: &mut ComponentCtx) {
+    pub fn handle_new_data_message(&mut self, ticket: MessageTicket, ctx: &mut ComponentCtx) {
         // Go through all branches that are awaiting new messages and see if
         // there is one that can receive this message.
         if !self.consensus.handle_new_data_message(&message, ctx) {
             // Old message, so drop it
         if self.consensus.handle_new_data_message(ticket, ctx) {
             // Message should not be handled now
             return;
+        }
         let message = ctx.read_message_using_ticket(ticket).as_data();
         let mut iter_id = self.tree.get_queue_first(QueueKind::AwaitingMessage);
         while let Some(branch_id) = iter_id {
             iter_id = self.tree.get_queue_next(branch_id);
             let branch = &self.tree[branch_id];
             if branch.awaiting_port != message.data_header.target_port { continue; }
             if !self.consensus.branch_can_receive(branch_id, &message) { continue; }
             // 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];
             debug_assert!(receiving_branch.awaiting_port == message.data_header.target_port);
             receiving_branch.awaiting_port = PortIdLocal::new_invalid();
             receiving_branch.prepared = PreparedStatement::PerformedGet(message.content.clone());
             self.consensus.notify_of_received_message(receiving_branch_id, &message, ctx);
             // And prepare the branch for running
             self.tree.push_into_queue(QueueKind::Runnable, receiving_branch_id);
+        }
+    }
     pub fn handle_new_sync_comp_message(&mut self, message: SyncCompMessage, ctx: &mut ComponentCtx) -> Option<ConnectorScheduling> {
         if let Some(round_conclusion) = self.consensus.handle_new_sync_comp_message(message, ctx) {
             return Some(self.enter_non_sync_mode(round_conclusion, ctx));
+        }
         return None;
+    }
     pub fn handle_new_sync_port_message(&mut self, message: SyncPortMessage, ctx: &mut ComponentCtx) {
         self.consensus.handle_new_sync_port_message(message, ctx);
+    }
     pub fn handle_new_sync_control_message(&mut self, message: SyncControlMessage, ctx: &mut ComponentCtx) -> Option<ConnectorScheduling> {
         if let Some(round_conclusion) = self.consensus.handle_new_sync_control_message(message, ctx) {
             return Some(self.enter_non_sync_mode(round_conclusion, ctx));
+        }
         return None;
+    }
     // --- 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,
             prepared: branch.prepared.take(),
         };
         let run_result = Self::run_prompt(&mut branch.code_state, &sched_ctx.runtime.protocol_description, &mut run_context);
         if let Err(eval_error) = run_result {
             self.eval_error = Some(eval_error);
             self.mode = Mode::SyncError;
             if let Some(conclusion) = self.consensus.notify_of_fatal_branch(branch_id, comp_ctx) {
                 // We can exit immediately
                 return self.enter_non_sync_mode(conclusion, comp_ctx);
             } else {
                 // Current branch failed. But we may have other things that are
                 // running.
                 return ConnectorScheduling::Immediate;
+            }
+        }
         let run_result = run_result.unwrap();
         // 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 {
             EvalContinuation::BranchInconsistent => {
                 // Branch became inconsistent
                 branch.sync_state = SpeculativeState::Inconsistent;
             },
             EvalContinuation::BlockFires(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, comp_ctx);
                 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, comp_ctx);
                 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;
             },
             EvalContinuation::BlockGet(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);
                 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_message_received = false;
                 for message in comp_ctx.get_read_data_messages(port_id) {
                     if self.consensus.branch_can_receive(branch_id, &message) {
                         // 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.awaiting_port = PortIdLocal::new_invalid();
                         branch.prepared = PreparedStatement::PerformedGet(message.content.clone());
                         self.consensus.notify_of_new_branch(branch_id, receiving_branch_id);
                         self.consensus.notify_of_received_message(receiving_branch_id, &message, comp_ctx);
                         self.tree.push_into_queue(QueueKind::Runnable, receiving_branch_id);
                         any_message_received = true;
+                    }
+                }
                 if any_message_received {
                     return ConnectorScheduling::Immediate;
+                }
+            }
             EvalContinuation::SyncBlockEnd => {
                 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 {
                     branch.sync_state = SpeculativeState::Inconsistent;
+                }
             },
             EvalContinuation::NewFork => {
                 // Like the `NewChannel` result. This means we're setting up
                 // a branch and putting a marker inside the RunContext for the
                 // next time we run the PDL code
                 let left_id = branch_id;
                 let right_id = self.tree.fork_branch(left_id);
                 self.consensus.notify_of_new_branch(left_id, right_id);
                 self.tree.push_into_queue(QueueKind::Runnable, left_id);
                 self.tree.push_into_queue(QueueKind::Runnable, right_id);
                 let left_branch = &mut self.tree[left_id];
                 left_branch.prepared = PreparedStatement::ForkedExecution(true);
                 let right_branch = &mut self.tree[right_id];
                 right_branch.prepared = PreparedStatement::ForkedExecution(false);
+            }
             EvalContinuation::Put(port_id, content) => {
                 // Branch is attempting to send data
                 let port_id = PortIdLocal::new(port_id.0.u32_suffix);
                 let (sync_header, data_header) = self.consensus.handle_message_to_send(branch_id, port_id, &content, comp_ctx);
                 let message = DataMessage{ sync_header, data_header, content };
                 match comp_ctx.submit_message(Message::Data(message)) {
                     Ok(_) => {
                         // Message is underway
                         branch.prepared = PreparedStatement::PerformedPut;
                         self.tree.push_into_queue(QueueKind::Runnable, branch_id);
                         return ConnectorScheduling::Immediate;
                     },
                     Err(_) => {
                         // We don't own the port
                         let pd = &sched_ctx.runtime.protocol_description;
                         let eval_error = branch.code_state.new_error_at_expr(
                             &pd.modules, &pd.heap,
                             String::from("attempted to 'put' on port that is no longer owned")
                         );
                         self.eval_error = Some(eval_error);
                         self.mode = Mode::SyncError;
                         println!("DEBUGERINO: Notify of fatal branch");
                         if let Some(conclusion) = self.consensus.notify_of_fatal_branch(branch_id, comp_ctx) {
                             println!("DEBUGERINO: Actually got {:?}", conclusion);
                             return self.enter_non_sync_mode(conclusion, comp_ctx);
+                        }
+                    }
+                }
             },
             _ => 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,
             prepared: branch.prepared.take(),
         };
         let run_result = Self::run_prompt(&mut branch.code_state, &sched_ctx.runtime.protocol_description, &mut run_context);
         if let Err(eval_error) = run_result {
             comp_ctx.push_error(eval_error);
             return ConnectorScheduling::Exit
+        }
         let run_result = run_result.unwrap();
         match run_result {
             EvalContinuation::ComponentTerminated => {
                 branch.sync_state = SpeculativeState::Finished;
                 return ConnectorScheduling::Exit;
             },
             EvalContinuation::SyncBlockStart => {
                 comp_ctx.notify_sync_start();
                 let sync_branch_id = self.tree.start_sync();
                 debug_assert!(self.last_finished_handled.is_none());
                 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);
                 self.mode = Mode::Sync;
                 return ConnectorScheduling::Immediate;
             },
             EvalContinuation::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_prompt = Prompt::new(
                     &sched_ctx.runtime.protocol_description.types,
                     &sched_ctx.runtime.protocol_description.heap,
                     definition_id, monomorph_idx, arguments
                 );
                 let new_component = ConnectorPDL::new(new_prompt);
                 comp_ctx.push_component(new_component, comp_ctx.workspace_ports.clone());
                 comp_ctx.workspace_ports.clear();
                 return ConnectorScheduling::Later;
             },
             EvalContinuation::NewChannel => {
                 let (getter, putter) = sched_ctx.runtime.create_channel(comp_ctx.id);
                 debug_assert!(getter.kind == PortKind::Getter && putter.kind == PortKind::Putter);
                 branch.prepared = PreparedStatement::CreatedChannel((
                     Value::Output(PortId::new(putter.self_id.index)),
                     Value::Input(PortId::new(getter.self_id.index)),
                 ));
                 comp_ctx.push_port(putter);
                 comp_ctx.push_port(getter);
                 return ConnectorScheduling::Immediate;
             },
             _ => unreachable!("unexpected run result '{:?}' while running in non-sync mode", run_result),
+        }
+    }
     /// Helper that moves the component's state back into non-sync mode, using
     /// the provided solution branch ID as the branch that should be comitted to
     /// memory. If this function returns false, then the component is supposed
     /// to exit.
     fn enter_non_sync_mode(&mut self, conclusion: RoundConclusion, ctx: &mut ComponentCtx) -> ConnectorScheduling {
         debug_assert!(self.mode == Mode::Sync || self.mode == Mode::SyncError);
         // Depending on local state decide what to do
         let final_branch_id = match conclusion {
             RoundConclusion::Success(branch_id) => Some(branch_id),
             RoundConclusion::Failure => None,
         };
         if let Some(solution_branch_id) = final_branch_id {
             let mut fake_vec = Vec::new();
             self.tree.end_sync(solution_branch_id);
             self.consensus.end_sync(solution_branch_id, &mut fake_vec);
             debug_assert!(fake_vec.is_empty());
             ctx.notify_sync_end(&[]);
             self.last_finished_handled = None;
             self.eval_error = None; // in case we came from the SyncError mode
             self.mode = Mode::NonSync;
             return ConnectorScheduling::Immediate;
         } else {
             // No final branch, because we're supposed to exit!
             self.last_finished_handled = None;
             self.mode = Mode::Error;
             if let Some(eval_error) = self.eval_error.take() {
                 ctx.push_error(eval_error);
+            }
             return ConnectorScheduling::Exit;
+        }
+    }
     /// Runs the prompt repeatedly until some kind of execution-blocking
     /// condition appears.
     #[inline]
     fn run_prompt(prompt: &mut Prompt, pd: &ProtocolDescription, ctx: &mut ConnectorRunContext) -> Result<EvalContinuation, EvalError> {
         loop {
             let result = prompt.step(&pd.types, &pd.heap, &pd.modules, ctx);
             if let Ok(EvalContinuation::Stepping) = result {
                 continue;
+            }
             return result;
+        }
+    }
+}
@@ \ No newline at end of file @@

src/runtime2/consensus.rs

➞

Show inline comments

 use crate::collections::VecSet;
 use crate::protocol::eval::ValueGroup;
 use super::ConnectorId;
 use super::branch::BranchId;
 use super::port::{ChannelId, PortIdLocal, PortState};
 use super::inbox::{
     Message, DataHeader, SyncHeader, ChannelAnnotation, BranchMarker,
     DataMessage,
     SyncCompMessage, SyncCompContent,
     SyncPortMessage, SyncPortContent,
     SyncControlMessage, SyncControlContent
 };
 use super::scheduler::{ComponentCtx, ComponentPortChange};
+use super::scheduler::{ComponentCtx, ComponentPortChange, MessageTicket};
 struct BranchAnnotation {
     channel_mapping: Vec<ChannelAnnotation>,
     cur_marker: BranchMarker,
+}
 #[derive(Debug)]
 pub(crate) struct LocalSolution {
     component: ConnectorId,
     final_branch_id: BranchId,
     port_mapping: Vec<(ChannelId, BranchMarker)>,
+}
 #[derive(Debug, Clone)]
 pub(crate) struct GlobalSolution {
     component_branches: Vec<(ConnectorId, BranchId)>,
     channel_mapping: Vec<(ChannelId, BranchMarker)>, // TODO: This can go, is debugging info
+}
 #[derive(Debug, PartialEq, Eq)]
 pub enum RoundConclusion {
     Failure,
     Success(BranchId),
+}
 // -----------------------------------------------------------------------------
 // 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. Should reduce locking by quite a
 //  bit.
 // TODO: Needs a refactor. Firstly we have cases where we don't have a branch ID
 //  but we do want to enumerate all current ports. So put that somewhere in a
 //  central place. Secondly. Error handling and regular message handling is
 //  becoming a mess.
 pub(crate) struct Consensus {
     // --- State that is cleared after each round
     // Local component's state
     highest_connector_id: ConnectorId,
     branch_annotations: Vec<BranchAnnotation>, // index is branch ID
     branch_markers: Vec<BranchId>, // index is branch marker, maps to branch
     // Gathered state from communication
     encountered_ports: VecSet<PortIdLocal>, // to determine if we should send "port remains silent" messages.
     solution_combiner: SolutionCombiner,
     handled_wave: bool, // encountered notification wave in this round
     conclusion: Option<RoundConclusion>,
     ack_remaining: u32,
     // --- Persistent state
     peers: Vec<Peer>,
     sync_round: u32,
     // --- Workspaces
     workspace_ports: Vec<PortIdLocal>,
+}
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub(crate) enum Consistency {
     Valid,
     Inconsistent,
+}
 #[derive(PartialEq, Eq)]
 pub(crate) enum MessageOrigin {
     Past,
     Present,
     Future
+}
 impl Consensus {
     pub fn new() -> Self {
         return Self {
             highest_connector_id: ConnectorId::new_invalid(),
             branch_annotations: Vec::new(),
             branch_markers: Vec::new(),
             encountered_ports: VecSet::new(),
             solution_combiner: SolutionCombiner::new(),
             handled_wave: false,
             conclusion: None,
             ack_remaining: 0,
             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
     #[deprecated]
     pub fn get_annotation(&self, branch_id: BranchId, channel_id: PortIdLocal) -> &ChannelAnnotation {
         let branch = &self.branch_annotations[branch_id.index as usize];
         let port = branch.channel_mapping.iter().find(|v| v.channel_id.index == channel_id.index).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.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{
             channel_mapping: ctx.get_ports().iter()
                 .map(|v| ChannelAnnotation {
                     channel_id: v.channel_id,
                     registered_id: None,
                     expected_firing: None,
                 })
                 .collect(),
             cur_marker: BranchMarker::new_invalid(),
         });
         self.branch_markers.push(BranchId::new_invalid());
         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_marker = BranchMarker::new(self.branch_markers.len() as u32);
         let new_branch_annotations = BranchAnnotation{
             channel_mapping: parent_branch_annotations.channel_mapping.clone(),
             cur_marker: new_marker,
         };
         self.branch_annotations.push(new_branch_annotations);
         self.branch_markers.push(new_branch_id);
+    }
     /// Notifies the consensus algorithm that a particular branch has
     /// encountered an unrecoverable error.
     pub fn notify_of_fatal_branch(&mut self, failed_branch_id: BranchId, ctx: &mut ComponentCtx) -> Option<RoundConclusion> {
         debug_assert!(self.is_in_sync());
         // Check for trivial case, where branch has not yet communicated within
         // the consensus algorithm
         let branch = &self.branch_annotations[failed_branch_id.index as usize];
         if branch.channel_mapping.iter().all(|v| v.registered_id.is_none()) {
             println!("DEBUG: Failure everything silent");
             return Some(RoundConclusion::Failure);
+        }
         // We're not in the trivial case: since we've communicated we need to
         // let everyone know that this round is probably not going to end well.
         return self.initiate_sync_failure(ctx);
+    }
     /// 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.channel_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, ctx: &ComponentCtx) -> Consistency {
         debug_assert!(self.is_in_sync());
         let port_desc = ctx.get_port_by_id(port_id).unwrap();
         let channel_id = port_desc.channel_id;
         let branch = &mut self.branch_annotations[branch_id.index as usize];
         for mapping in &mut branch.channel_mapping {
             if mapping.channel_id == channel_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 a new local solution from a finished branch. If the component
     /// is not the leader of the sync region then it will be sent to the
     /// appropriate component. If it is the leader then there is a chance that
     /// this solution completes a global solution. In that case the solution
     /// branch ID will be returned.
     pub(crate) fn handle_new_finished_sync_branch(&mut self, branch_id: BranchId, ctx: &mut ComponentCtx) -> Option<RoundConclusion> {
         // Turn the port mapping into a local solution
         let source_mapping = &self.branch_annotations[branch_id.index as usize].channel_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_channel_id(port.channel_id).unwrap();
             let self_port_id = port_desc.self_id;
             let peer_port_id = port_desc.peer_id;
             let channel_id = port_desc.channel_id;
             if !self.encountered_ports.contains(&self_port_id) {
                 let message = SyncPortMessage {
                     sync_header: SyncHeader{
                         sending_component_id: ctx.id,
                         highest_component_id: self.highest_connector_id,
                         sync_round: self.sync_round
                     },
                     source_port: self_port_id,
                     target_port: peer_port_id,
                     content: SyncPortContent::SilentPortNotification,
                 };
                 match ctx.submit_message(Message::SyncPort(message)) {
                     Ok(_) => {
                         self.encountered_ports.push(self_port_id);
                     },
                     Err(_) => {
                         // Seems like we were done with this branch, but one of
                         // the silent ports (in scope) is actually closed
                         return self.notify_of_fatal_branch(branch_id, ctx);
+                    }
+                }
+            }
             target_mapping.push((
                 channel_id,
                 port.registered_id.unwrap_or(BranchMarker::new_invalid())
             ));
+        }
         let local_solution = LocalSolution{
             component: ctx.id,
             final_branch_id: branch_id,
             port_mapping: target_mapping,
         };
         let maybe_conclusion = self.send_to_leader_or_handle_as_leader(SyncCompContent::LocalSolution(local_solution), ctx);
         return maybe_conclusion;
+    }
     /// Notifies the consensus algorithm about the chosen branch to commit to
     /// memory (may be the invalid "start" branch)
     pub fn end_sync(&mut self, branch_id: BranchId, final_ports: &mut Vec<ComponentPortChange>) {
         debug_assert!(self.is_in_sync());
         // TODO: Handle sending and receiving ports
         // Set final ports
         let branch = &self.branch_annotations[branch_id.index as usize];
         // Clear out internal storage to defaults
         self.highest_connector_id = ConnectorId::new_invalid();
         self.branch_annotations.clear();
         self.branch_markers.clear();
         self.encountered_ports.clear();
         self.solution_combiner.clear();
         self.handled_wave = false;
         self.conclusion = None;
         self.ack_remaining = 0;
         // And modify persistent storage
         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];
         let port_info = ctx.get_port_by_id(source_port_id).unwrap();
         if cfg!(debug_assertions) {
             // Check for consistent mapping
             let port = branch.channel_mapping.iter()
                 .find(|v| v.channel_id == port_info.channel_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
         let data_header = DataHeader{
             expected_mapping: branch.channel_mapping.iter()
                 .filter(|v| v.registered_id.is_some() || v.channel_id == port_info.channel_id)
                 .copied()
                 .collect(),
             sending_port: port_info.self_id,
             target_port: port_info.peer_id,
             new_mapping: branch.cur_marker,
         };
         // Update port mapping
         for mapping in &mut branch.channel_mapping {
             if mapping.channel_id == port_info.channel_id {
                 mapping.expected_firing = Some(true);
                 mapping.registered_id = Some(branch.cur_marker);
+            }
+        }
         // Update branch marker
         let new_marker = BranchMarker::new(self.branch_markers.len() as u32);
         branch.cur_marker = new_marker;
         self.branch_markers.push(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 sync header. The caller is
     /// responsible for checking for branches that might be able to receive
     /// the message.
     pub fn handle_new_data_message(&mut self, message: &DataMessage, ctx: &mut ComponentCtx) -> bool {
         let handled = self.handle_received_sync_header(&message.sync_header, ctx);
         if handled {
             self.encountered_ports.push(message.data_header.target_port);
     pub fn handle_new_data_message(&mut self, ticket: MessageTicket, ctx: &mut ComponentCtx) -> bool {
         let message = ctx.read_message_using_ticket(ticket).as_data();
         let target_port = message.data_header.target_port;
         match self.handle_received_sync_header(message.sync_header, ctx) {
             MessageOrigin::Past => return false,
             MessageOrigin::Present => {
                 self.encountered_ports.push(target_port);
                 return true;
             },
             MessageOrigin::Future => {
                 let message = ctx.take_message_using_ticket(ticket);
                 ctx.put_back_message(message);
                 return false;
+            }
+        }
         return handled;
+    }
     /// 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_comp_message(&mut self, message: SyncCompMessage, ctx: &mut ComponentCtx) -> Option<RoundConclusion> {
         if !self.handle_received_sync_header(&message.sync_header, ctx) {
             return None;
         match self.handle_received_sync_header(message.sync_header, ctx) {
             MessageOrigin::Past => return None,
             MessageOrigin::Present => {},
             MessageOrigin::Future => {
                 ctx.put_back_message(Message::SyncComp(message));
                 return None
+            }
+        }
         // And handle the contents
         debug_assert_eq!(message.target_component_id, ctx.id);
         match &message.content {
             SyncCompContent::LocalFailure |
             SyncCompContent::LocalSolution(_) |
             SyncCompContent::PartialSolution(_) |
             SyncCompContent::AckFailure |
             SyncCompContent::Presence(_) => {
                 // Needs to be handled by the leader
                 return self.send_to_leader_or_handle_as_leader(message.content, ctx);
             },
             SyncCompContent::GlobalSolution(solution) => {
                 // Found a global solution
                 debug_assert_ne!(self.highest_connector_id, ctx.id); // not the leader
                 let (_, branch_id) = solution.component_branches.iter()
                     .find(|(component_id, _)| *component_id == ctx.id)
                     .unwrap();
                 return Some(RoundConclusion::Success(*branch_id));
             },
             SyncCompContent::GlobalFailure => {
                 // Global failure of round, send Ack to leader
                 println!("DEBUGERINO: Got GlobalFailure, sending Ack in response");
                 debug_assert_ne!(self.highest_connector_id, ctx.id); // not the leader
                 let _result = self.send_to_leader_or_handle_as_leader(SyncCompContent::AckFailure, ctx);
                 debug_assert!(_result.is_none());
                 return Some(RoundConclusion::Failure);
             },
             SyncCompContent::Notification => {
                 // We were just interested in the sync header we handled above
                 return None;
+            }
+        }
+    }
     pub fn handle_new_sync_port_message(&mut self, message: SyncPortMessage, ctx: &mut ComponentCtx) -> Option<RoundConclusion> {
         if !self.handle_received_sync_header(&message.sync_header, ctx) {
         match self.handle_received_sync_header(message.sync_header, ctx) {
             MessageOrigin::Past => return None,
             MessageOrigin::Present => {},
             MessageOrigin::Future => {
                 ctx.put_back_message(Message::SyncPort(message));
                 return None;
+            }
+        }
         debug_assert!(self.is_in_sync());
         debug_assert!(ctx.get_port_by_id(message.target_port).is_some());
         match message.content {
             SyncPortContent::SilentPortNotification => {
                 // The point here is to let us become part of the sync round and
                 // take note of the leader in case all of our ports are silent.
                 self.encountered_ports.push(message.target_port);
                 return None
+            }
             SyncPortContent::NotificationWave => {
                 // Wave to discover everyone in the network, handling sync
                 // header takes care of leader discovery, here we need to make
                 // sure we propagate the wave
                 if self.handled_wave {
                     return None;
+                }
                 self.handled_wave = true;
                 // Propagate wave to all peers except the one that has sent us
                 // the wave.
                 for mapping in &self.branch_annotations[0].channel_mapping {
                     let channel_id = mapping.channel_id;
                     let port_desc = ctx.get_port_by_channel_id(channel_id).unwrap();
                     if port_desc.self_id == message.target_port {
                         // Wave came from this port, no need to send one back
                         continue;
+                    }
                     let message = SyncPortMessage{
                         sync_header: self.create_sync_header(ctx),
                         source_port: port_desc.self_id,
                         target_port: port_desc.peer_id,
                         content: SyncPortContent::NotificationWave,
                     };
                     // As with the other SyncPort where we throw away the
                     // result: we're dealing with an error here anyway
                     let _unused = ctx.submit_message(Message::SyncPort(message));
+                }
                 // And let the leader know about our port state
                 let annotations = &self.branch_annotations[0];
                 let mut channels = Vec::with_capacity(annotations.channel_mapping.len());
                 for mapping in &annotations.channel_mapping {
                     let port_info = ctx.get_port_by_channel_id(mapping.channel_id).unwrap();
                     channels.push(LocalChannelPresence{
                         channel_id: mapping.channel_id,
                         is_closed: port_info.state == PortState::Closed,
                     });
+                }
-                let maybe_conlusion = self.send_to_leader_or_handle_as_leader(SyncCompContent::Presence(ComponentPresence{
+                let maybe_conclusion = self.send_to_leader_or_handle_as_leader(SyncCompContent::Presence(ComponentPresence{
                     component_id: ctx.id,
                     channels,
                 }), ctx);
-                return maybe_conlusion;
+                return maybe_conclusion;
+            }
+        }
+    }
     pub fn handle_new_sync_control_message(&mut self, message: SyncControlMessage, ctx: &mut ComponentCtx) -> Option<RoundConclusion> {
         if message.in_response_to_sync_round < self.sync_round {
             // Old message
             return None
+        }
         // Because the message is always sent in response to a message
         // originating here, the sync round number can never be larger than the
         // currently stored one.
         debug_assert_eq!(message.in_response_to_sync_round, self.sync_round);
         match message.content {
             SyncControlContent::ChannelIsClosed(_) => {
                 return self.initiate_sync_failure(ctx);
+            }
+        }
+    }
     pub fn notify_of_received_message(&mut self, branch_id: BranchId, message: &DataMessage, ctx: &ComponentCtx) {
         debug_assert!(self.branch_can_receive(branch_id, message));
         let target_port = ctx.get_port_by_id(message.data_header.target_port).unwrap();
         let branch = &mut self.branch_annotations[branch_id.index as usize];
         for mapping in &mut branch.channel_mapping {
             if mapping.channel_id == target_port.channel_id {
                 // Found the port in which the message should be inserted
                 mapping.registered_id = Some(message.data_header.new_mapping);
                 // Check for sent ports
                 debug_assert!(self.workspace_ports.is_empty());
                 find_ports_in_value_group(&message.content, &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, message: &DataMessage) -> bool {
         if let Some(peer) = self.peers.iter().find(|v| v.id == message.sync_header.sending_component_id) {
             if message.sync_header.sync_round < peer.expected_sync_round {
                 return false;
+            }
+        }
         let annotation = &self.branch_annotations[branch_id.index as usize];
         for expected in &message.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.channel_mapping {
                 if expected.channel_id == current.channel_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
-    fn handle_received_sync_header(&mut self, sync_header: &SyncHeader, ctx: &mut ComponentCtx) -> bool {
+    fn handle_received_sync_header(&mut self, sync_header: SyncHeader, ctx: &mut ComponentCtx) -> MessageOrigin {
         debug_assert!(sync_header.sending_component_id != ctx.id); // not sending to ourselves
         if !self.handle_peer(sync_header) {
             // We can drop this package
             return false;
         let origin = self.handle_peer(&sync_header);
         if origin != MessageOrigin::Present {
             // We do not have to handle it now
             return origin;
+        }
         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 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 = SyncCompMessage {
                     sync_header: self.create_sync_header(ctx),
                     target_component_id: peer.id,
                     content: SyncCompContent::Notification,
                 };
                 ctx.submit_message(Message::SyncComp(message)).unwrap(); // unwrap: sending to component instead of through channel
+            }
             // But also send our locally combined solution
             self.forward_local_data_to_new_leader(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 = SyncCompMessage {
                 sync_header: self.create_sync_header(ctx),
                 target_component_id: sync_header.sending_component_id,
                 content: SyncCompContent::Notification
             };
             ctx.submit_message(Message::SyncComp(message)).unwrap(); // unwrap: sending to component instead of through channel
         } // else: exactly equal, so do nothing
-        return true;
+        return MessageOrigin::Present;
+    }
     /// 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 {
+    fn handle_peer(&mut self, sync_header: &SyncHeader) -> MessageOrigin {
         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;
                 if entry.encountered_this_round {
                     // Already encountered this round
                     if sync_header.sync_round < entry.expected_sync_round {
                         return MessageOrigin::Past;
                     } else if sync_header.sync_round == entry.expected_sync_round {
                         return MessageOrigin::Present;
                     } else {
                         return MessageOrigin::Future;
+                    }
                 } else {
                     // TODO: Proper handling of potential overflow
                     entry.encountered_this_round = true;
                     if sync_header.sync_round >= entry.expected_sync_round {
                         entry.expected_sync_round = sync_header.sync_round;
-                    return true;
+                        return MessageOrigin::Present;
                     } else {
                     return false;
                         return MessageOrigin::Past;
+                    }
+                }
             },
             None => {
                 self.peers.push(Peer{
                     id: sync_header.sending_component_id,
                     encountered_this_round: true,
                     expected_sync_round: sync_header.sync_round,
                 });
-                return true;
+                return MessageOrigin::Present;
+            }
+        }
+    }
     /// Sends a message towards the leader, if already the leader then the
     /// message will be handled immediately.
     fn send_to_leader_or_handle_as_leader(&mut self, content: SyncCompContent, ctx: &mut ComponentCtx) -> Option<RoundConclusion> {
         if self.highest_connector_id == ctx.id {
             // We are the leader
             match content {
                 SyncCompContent::LocalFailure => {
                     if self.solution_combiner.mark_failure_and_check_for_global_failure() {
                         return self.handle_global_failure_as_leader(ctx);
+                    }
                 },
                 SyncCompContent::LocalSolution(local_solution) => {
                     if let Some(global_solution) = self.solution_combiner.add_solution_and_check_for_global_solution(local_solution) {
                         return self.handle_global_solution_as_leader(global_solution, ctx);
+                    }
                 },
                 SyncCompContent::PartialSolution(partial_solution) => {
                     if let Some(conclusion) = self.solution_combiner.combine(partial_solution) {
                         match conclusion {
                             LeaderConclusion::Solution(global_solution) => {
                                 return self.handle_global_solution_as_leader(global_solution, ctx);
                             },
                             LeaderConclusion::Failure => {
                                 return self.handle_global_failure_as_leader(ctx);
+                            }
+                        }
+                    }
                 },
                 SyncCompContent::Presence(component_presence) => {
                     if self.solution_combiner.add_presence_and_check_for_global_failure(component_presence.component_id, &component_presence.channels) {
                         return self.handle_global_failure_as_leader(ctx);
+                    }
                 },
                 SyncCompContent::AckFailure => {
                     debug_assert_eq!(Some(RoundConclusion::Failure), self.conclusion);
                     debug_assert!(self.ack_remaining > 0);
                     self.ack_remaining -= 1;
                     if self.ack_remaining == 0 {
                         return Some(RoundConclusion::Failure);
+                    }
+                }
                 SyncCompContent::Notification | SyncCompContent::GlobalSolution(_) |
                 SyncCompContent::GlobalFailure => {
                     unreachable!("unexpected message content for leader");
                 },
+            }
         } else {
             // Someone else is the leader
             let message = SyncCompMessage {
                 sync_header: self.create_sync_header(ctx),
                 target_component_id: self.highest_connector_id,
                 content,
             };
             ctx.submit_message(Message::SyncComp(message)).unwrap(); // unwrap: sending to component instead of through channel
+        }
         return None;
+    }
     fn handle_global_solution_as_leader(&mut self, global_solution: GlobalSolution, ctx: &mut ComponentCtx) -> Option<RoundConclusion> {
         if self.conclusion.is_some() {
             return None;
+        }
         // Handle the global 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 = SyncCompMessage {
                 sync_header: self.create_sync_header(ctx),
                 target_component_id: connector_id,
                 content: SyncCompContent::GlobalSolution(global_solution.clone()),
             };
             ctx.submit_message(Message::SyncComp(message)).unwrap(); // unwrap: sending to component instead of through channel
+        }
         debug_assert!(my_final_branch_id.is_valid());
         self.conclusion = Some(RoundConclusion::Success(my_final_branch_id));
         return Some(RoundConclusion::Success(my_final_branch_id));
+    }
     fn handle_global_failure_as_leader(&mut self, ctx: &mut ComponentCtx) -> Option<RoundConclusion> {
         debug_assert!(self.solution_combiner.failure_reported && self.solution_combiner.check_for_global_failure());
         if self.conclusion.is_some() {
             // Already sent out a failure
             return None;
+        }
         // TODO: Performance
         let mut encountered = VecSet::new();
         for presence in &self.solution_combiner.presence {
             if presence.owner_a != ctx.id {
                 // Did not add it ourselves
                 if encountered.push(presence.owner_a) {
                     // Not yet sent a message
                     let message = SyncCompMessage{
                         sync_header: self.create_sync_header(ctx),
                         target_component_id: presence.owner_a,
                         content: SyncCompContent::GlobalFailure,
                     };
                     ctx.submit_message(Message::SyncComp(message)).unwrap(); // unwrap: sending to component instead of through channel
+                }
             } else if let Some(owner_b) = presence.owner_b {
                 if owner_b != ctx.id {
                     if encountered.push(owner_b) {
                         let message = SyncCompMessage{
                             sync_header: self.create_sync_header(ctx),
                             target_component_id: owner_b,
                             content: SyncCompContent::GlobalFailure,
                         };
                         ctx.submit_message(Message::SyncComp(message)).unwrap();
+                    }
+                }
+            }
+        }
         println!("DEBUERINO: Leader entering error state, we need to wait on {:?}", encountered.iter().map(|v| v.0).collect::<Vec<_>>());
         self.conclusion = Some(RoundConclusion::Failure);
         if encountered.is_empty() {
             // We don't have to wait on Acks
             return Some(RoundConclusion::Failure);
         } else {
             self.ack_remaining = encountered.len() as u32;
             return None;
+        }
+    }
     fn initiate_sync_failure(&mut self, ctx: &mut ComponentCtx) -> Option<RoundConclusion> {
         debug_assert!(self.is_in_sync());
         // Notify leader of our channels and the fact that we just failed
         let channel_mapping = &self.branch_annotations[0].channel_mapping;
         let mut channel_presence = Vec::with_capacity(channel_mapping.len());
         for mapping in channel_mapping {
             let port = ctx.get_port_by_channel_id(mapping.channel_id).unwrap();
             channel_presence.push(LocalChannelPresence{
                 channel_id: mapping.channel_id,
                 is_closed: port.state == PortState::Closed,
             });
+        }
         let maybe_already = self.send_to_leader_or_handle_as_leader(SyncCompContent::Presence(ComponentPresence{
             component_id: ctx.id,
             channels: channel_presence,
         }), ctx);
         if self.handled_wave {
             // Someone (or us) has already initiated a sync failure.
             return maybe_already;
+        }
         let maybe_conclusion = self.send_to_leader_or_handle_as_leader(SyncCompContent::LocalFailure, ctx);
         debug_assert!(if maybe_already.is_some() { maybe_conclusion.is_some() } else { true });
         println!("DEBUG: Maybe conclusion is {:?}", maybe_conclusion);
         // Initiate a discovery wave so peers can do the same
         self.handled_wave = true;
         for mapping in &self.branch_annotations[0].channel_mapping {
             let channel_id = mapping.channel_id;
             let port_info = ctx.get_port_by_channel_id(channel_id).unwrap();
             let message = SyncPortMessage{
                 sync_header: self.create_sync_header(ctx),
                 source_port: port_info.self_id,
                 target_port: port_info.peer_id,
                 content: SyncPortContent::NotificationWave,
             };
             // Note: submitting the message might fail. But we're attempting to
             // handle the error anyway.
             // TODO: Think about this a second time: how do we make sure the
             //  entire network will fail if we reach this condition
             let _unused = ctx.submit_message(Message::SyncPort(message));
+        }
         return maybe_conclusion;
+    }
     #[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_data_to_new_leader(&mut self, ctx: &mut ComponentCtx) {
         debug_assert_ne!(self.highest_connector_id, ctx.id);
         if let Some(partial_solution) = self.solution_combiner.drain() {
             let message = SyncCompMessage {
                 sync_header: self.create_sync_header(ctx),
                 target_component_id: self.highest_connector_id,
                 content: SyncCompContent::PartialSolution(partial_solution),
             };
             ctx.submit_message(Message::SyncComp(message)).unwrap(); // unwrap: sending to component instead of through channel
+        }
+    }
+}
 // -----------------------------------------------------------------------------
 // Solution storage and algorithms
 // -----------------------------------------------------------------------------
 // TODO: Remove all debug derives
 #[derive(Debug, Clone)]
 struct MatchedLocalSolution {
     final_branch_id: BranchId,
     channel_mapping: Vec<(ChannelId, BranchMarker)>,
     matches: Vec<ComponentMatches>,
+}
 #[derive(Debug, Clone)]
 struct ComponentMatches {
     target_id: ConnectorId,
     target_index: usize,
     match_indices: Vec<usize>, // of local solution in connector
+}
 #[derive(Debug, Clone)]
 struct ComponentPeer {
     target_id: ConnectorId,
     target_index: usize, // in array of global solution components
     involved_channels: Vec<ChannelId>,
+}
 #[derive(Debug, Clone)]
 struct ComponentLocalSolutions {
     component: ConnectorId,
     peers: Vec<ComponentPeer>,
     solutions: Vec<MatchedLocalSolution>,
     all_peers_present: bool,
+}
 #[derive(Debug, Clone)]
 pub(crate) struct ComponentPresence {
     component_id: ConnectorId,
     channels: Vec<LocalChannelPresence>,
+}
 #[derive(Debug, Clone)]
 pub(crate) struct LocalChannelPresence {
     channel_id: ChannelId,
     is_closed: bool,
+}
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 enum PresenceState {
     OnePresent, // one component reported the channel being open
     BothPresent, // two components reported the channel being open
     Closed, // one component reported the channel being closed
+}
 /// Record to hold channel state during the error-resolving mode of the leader.
 /// This is used to determine when the sync region has grown to its largest
 /// size. The structure is eventually consistent in the sense that a component
 /// might initially presume a channel is open, only to figure out later it is
 /// actually closed.
 #[derive(Debug, Clone)]
 struct ChannelPresence {
     owner_a: ConnectorId,
     owner_b: Option<ConnectorId>,
     id: ChannelId,
     state: PresenceState,
+}
 // TODO: Flatten? Flatten. Flatten everything.
 #[derive(Debug)]
 pub(crate) struct SolutionCombiner {
     local: Vec<ComponentLocalSolutions>, // used for finding solution
     presence: Vec<ChannelPresence>, // used to detect all channels present in case of failure
     failure_reported: bool,
+}
 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
+}
 enum LeaderConclusion {
     Solution(GlobalSolution),
     Failure,
+}
 impl SolutionCombiner {
     fn new() -> Self {
         return Self{
             local: Vec::new(),
             presence: Vec::new(),
             failure_reported: false,
         };
+    }
     /// 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();
+                }
                 if num_ports_in_peers == other_component.solutions[0].channel_mapping.len() {
                     other_component.all_peers_present = true;
+                }
                 other_component.peers.push(ComponentPeer{
                     target_id: component_id,
                     target_index: component_index,
                     involved_channels: component_match.involved_channels.clone(),
                 });
                 let new_component = &mut self.local[component_index];
                 new_component.peers.push(component_match);
+            }
+        }
         // 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_for_global_solution(component_index, solution_index);
+    }
     fn add_presence_and_check_for_global_failure(&mut self, component_id: ConnectorId, channels: &[LocalChannelPresence]) -> bool {
         'new_report_loop: for entry in channels {
             let mut found = false;
             for existing in &mut self.presence {
                 if existing.id == entry.channel_id {
                     // Same entry. We only update if we have the second
                     // component coming in it owns one end of the channel, or if
                     // a component is telling us that the channel is (now)
                     // closed.
                     if entry.is_closed {
                         existing.state = PresenceState::Closed;
                     } else if component_id != existing.owner_a && existing.state != PresenceState::Closed {
                         existing.state = PresenceState::BothPresent;
+                    }
                     if existing.owner_a != component_id {
                         existing.owner_b = Some(component_id);
+                    }
                     found = true;
                     break;
+                }
+            }
             if !found {
                 self.presence.push(ChannelPresence{
                     owner_a: component_id,
                     owner_b: None,
                     id: entry.channel_id,
                     state: if entry.is_closed { PresenceState::Closed } else { PresenceState::OnePresent },
                 });
+            }
+        }
         println!("DEBUGGERINO Presence is now:\n{:#?}", self.presence);
         return self.check_for_global_failure();
+    }
     fn mark_failure_and_check_for_global_failure(&mut self) -> bool {
         self.failure_reported = true;
         return self.check_for_global_failure();
+    }
     /// Checks if, starting at the provided local solution, a global solution
     /// can be formed.
     // TODO: At some point, check if divide and conquer is faster?
     fn check_for_global_solution(&self, initial_component_index: usize, initial_solution_index: usize) -> Option<GlobalSolution> {
         // Small trivial test necessary (but not sufficient) for a global
         // solution
         for component in &self.local {
             if !component.all_peers_present {
                 return None;
+            }
+        }
         // 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 all_match {
                 // All components matched until now.
                 if stack.len() == self.local.len() {
                     // We have found a global solution
                     break 'check_last_stack;
+                }
                 // 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;
+                        }
                         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();
+                }
+            }
             // Stack length is 1, hence we're back at our initial solution.
             // Since that doesn't yield a global solution, we simply:
             return None;
+        }
         // 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 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 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,
         });
+    }
     /// Checks if all preconditions for global sync failure have been met
     fn check_for_global_failure(&self) -> bool {
         if !self.failure_reported {
             return false;
+        }
         // Failure is reported, if all components are present then we may emit
         // the global failure broadcast
         // Check if all are present and we're preparing to fail this round
         let mut all_present = true;
         for presence in &self.presence {
             if presence.state == PresenceState::OnePresent {
                 all_present = false;
                 break;
+            }
+        }
         return all_present; // && failure_reported, which is checked above
+    }
     /// Turns the entire (partially resolved) global solution into a structure
     /// that can be forwarded to a new parent. The new parent may then merge
     /// already obtained information.
     fn drain(&mut self) -> Option<SolutionCombiner> {
         if self.local.is_empty() && self.presence.is_empty() && !self.failure_reported {
             return None;
+        }
         let result = SolutionCombiner{
             local: self.local.clone(),
             presence: self.presence.clone(),
             failure_reported: self.failure_reported,
         };
         self.local.clear();
         self.presence.clear();
         self.failure_reported = false;
         return Some(result);
+    }
     // TODO: Entire routine is quite wasteful. Combine instead of doing all work
     //  again.
     fn combine(&mut self, combiner: SolutionCombiner) -> Option<LeaderConclusion> {
         self.failure_reported = self.failure_reported || combiner.failure_reported;
         // Handle local solutions
         if self.local.is_empty() {
             // Trivial case
             self.local = combiner.local;
         } else {
             for local in combiner.local {
                 for matched in local.solutions {
                     let local_solution = LocalSolution{
                         component: local.component,
                         final_branch_id: matched.final_branch_id,
                         port_mapping: matched.channel_mapping,
                     };
                     let maybe_solution = self.add_solution_and_check_for_global_solution(local_solution);
                     if let Some(global_solution) = maybe_solution {
                         return Some(LeaderConclusion::Solution(global_solution));
+                    }
+                }
+            }
+        }
         // Handle channel presence
         println!("DEBUGERINO: Presence before joining is {:#?}", &self.presence);
         if self.presence.is_empty() {
             // Trivial case
             self.presence = combiner.presence;
             println!("DEBUGERINO: Trivial merging")
         } else {
             for presence in combiner.presence {
                 match self.presence.iter_mut().find(|v| v.id == presence.id) {
                     Some(entry) => {
                         // Combine entries. Take first that has Closed, then
                         // check first that has both, then check if they are
                         // combinable
                         if entry.state == PresenceState::Closed {
                             // Do nothing
                         } else if presence.state == PresenceState::Closed {
                             entry.owner_a = presence.owner_a;
                             entry.owner_b = presence.owner_b;
                             entry.state = PresenceState::Closed;
                         } else if entry.state == PresenceState::BothPresent {
                             // Again: do nothing
                         } else if presence.state == PresenceState::BothPresent {
                             entry.owner_a = presence.owner_a;
                             entry.owner_b = presence.owner_b;
                             entry.state = PresenceState::BothPresent;
                         } else {
                             // Both have one presence, combine into both present
                             debug_assert!(entry.state == PresenceState::OnePresent && presence.state == PresenceState::OnePresent);
                             entry.owner_b = Some(presence.owner_a);
                             entry.state = PresenceState::BothPresent;
+                        }
                     },
                     None => {
                         self.presence.push(presence);
+                    }
+                }
+            }
             println!("DEBUGERINO: Presence after joining is {:#?}", &self.presence);
             // After adding everything we might have immediately found a solution
             if self.check_for_global_failure() {
                 println!("DEBUG: Returning immediate failure?");
                 return Some(LeaderConclusion::Failure);
+            }
+        }
         return None;
+    }
     fn clear(&mut self) {
         self.local.clear();
         self.presence.clear();
         self.failure_reported = false;
+    }
+}
 // -----------------------------------------------------------------------------
 // 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 {
                     find_port_in_value(group, embedded_value, ports);
+                }
             },
             _ => {}, // values we don't care about
+        }
+    }
     // Clear the ports, then scan all the available values
     ports.clear();
     for value in &value_group.values {
         find_port_in_value(value_group, value, ports);
+    }
+}
@@ \ No newline at end of file @@

src/runtime2/inbox.rs

➞

Show inline comments

 use std::sync::Mutex;
 use std::collections::VecDeque;
 use crate::protocol::eval::ValueGroup;
 use crate::runtime2::consensus::{ComponentPresence, SolutionCombiner};
 use crate::runtime2::port::ChannelId;
 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 ChannelAnnotation {
     pub channel_id: ChannelId,
     pub registered_id: Option<BranchMarker>,
     pub expected_firing: Option<bool>,
+}
 /// Marker for a branch in a port mapping. A marker is, like a branch ID, a
 /// unique identifier for a branch, but differs in that a branch only has one
 /// branch ID, but might have multiple associated markers (i.e. one branch
 /// performing a `put` three times will generate three markers.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub(crate) struct BranchMarker{
     marker: u32,
+}
 impl BranchMarker {
     #[inline]
     pub(crate) fn new(marker: u32) -> Self {
         debug_assert!(marker != 0);
         return Self{ marker };
+    }
     #[inline]
     pub(crate) fn new_invalid() -> Self {
         return Self{ marker: 0 }
+    }
+}
 /// The header added by the synchronization algorithm to all.
 #[derive(Debug, Clone)]
+#[derive(Debug, Clone, Copy)]
 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<ChannelAnnotation>,
     pub sending_port: PortIdLocal,
     pub target_port: PortIdLocal,
     pub new_mapping: BranchMarker,
+}
 /// 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: ValueGroup,
+}
 #[derive(Debug)]
 pub(crate) enum SyncCompContent {
     LocalFailure, // notifying leader that component has failed (e.g. timeout, whatever)
     LocalSolution(LocalSolution), // sending a local solution to the leader
     PartialSolution(SolutionCombiner), // when new leader is detected, forward all local results
     GlobalSolution(GlobalSolution), // broadcasting to everyone
     GlobalFailure, // broadcasting to everyone
     AckFailure, // acknowledgement of failure to leader
     Notification, // just a notification (so purpose of message is to send the SyncHeader)
     Presence(ComponentPresence), // notifying leader of component presence (needed to ensure failing a round involves all components in a sync round)
+}
 /// A sync message is a message that is intended only for the consensus
 /// algorithm. The message goes directly to a component.
 #[derive(Debug)]
 pub(crate) struct SyncCompMessage {
     pub sync_header: SyncHeader,
     pub target_component_id: ConnectorId,
     pub content: SyncCompContent,
+}
 #[derive(Debug)]
 pub(crate) enum SyncPortContent {
     SilentPortNotification,
     NotificationWave,
+}
 /// A sync message intended for the consensus algorithm. This message does not
 /// go to a component, but through a channel (and results in potential
 /// rerouting) because we're not sure about the ID of the component that holds
 /// the other end of the channel.
 #[derive(Debug)]
 pub(crate) struct SyncPortMessage {
     pub sync_header: SyncHeader,
     pub source_port: PortIdLocal,
     pub target_port: PortIdLocal,
     pub content: SyncPortContent,
+}
 #[derive(Debug)]
 pub(crate) enum SyncControlContent {
     ChannelIsClosed(PortIdLocal), // contains port that is owned by the recipient of the message
+}
 /// A sync control message: originating from the scheduler, but intended for the
 /// current sync round of the recipient. Every kind of consensus algorithm must
 /// be able to handle such a message.
 #[derive(Debug)]
 pub(crate) struct SyncControlMessage {
     // For now these control messages are only aimed at components. Might change
     // in the future. But for now we respond to messages from components that
     // have, because of that message, published their ID.
     pub in_response_to_sync_round: u32,
     pub target_component_id: ConnectorId,
     pub content: SyncControlContent,
+}
 /// 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),
     SyncComp(SyncCompMessage),
     SyncPort(SyncPortMessage),
     SyncControl(SyncControlMessage),
     Control(ControlMessage),
+}
 impl Message {
     /// If the message is sent through a particular channel, then this function
     /// returns the port through which the message was sent.
     pub(crate) fn source_port(&self) -> Option<PortIdLocal> {
         // Currently only data messages have a source port
         match self {
             Message::Data(message) => return Some(message.data_header.sending_port),
             Message::SyncPort(message) => return Some(message.source_port),
             Message::SyncComp(_) => return None,
             Message::SyncControl(_) => return None,
             Message::Control(_) => return None,
+        }
+    }
     /// If the message is sent through a particular channel, then this function
     /// returns the target port through which the message was sent.
     pub(crate) fn target_port(&self) -> Option<PortIdLocal> {
         match self {
             Message::Data(message) => return Some(message.data_header.target_port),
             Message::SyncPort(message) => return Some(message.target_port),
             Message::SyncComp(_) => return None,
             Message::SyncControl(_) => return None,
             Message::Control(message) => {
                 match &message.content {
                     ControlContent::PortPeerChanged(port_id, _) => return Some(*port_id),
                     ControlContent::CloseChannel(port_id) => return Some(*port_id),
                     ControlContent::Ping => return None,
                     ControlContent::Ack => return None,
+                }
+            }
+        }
+    }
     pub(crate) fn as_data(&self) -> &DataMessage {
         match self {
             Message::Data(v) => v,
             _ => unreachable!(),
+        }
+    }
+}
 /// 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()),
+        }
+    }
     pub(crate) fn insert_message(&self, message: Message) {
         let mut lock = self.messages.lock().unwrap();
         lock.push_back(message);
+    }
     pub(crate) fn take_message(&self) -> Option<Message> {
         let mut lock = self.messages.lock().unwrap();
         return lock.pop_front();
+    }
     pub fn is_empty(&self) -> bool {
         let lock = self.messages.lock().unwrap();
         return lock.is_empty();
+    }
+}
@@ \ No newline at end of file @@

src/runtime2/native.rs

➞

Show inline comments

 use std::collections::VecDeque;
 use std::sync::{Arc, Mutex, Condvar};
 use std::sync::atomic::Ordering;
 use std::collections::HashMap;
 use crate::protocol::ComponentCreationError;
 use crate::protocol::eval::ValueGroup;
 use crate::runtime2::consensus::RoundConclusion;
 use super::{ConnectorKey, ConnectorId, RuntimeInner};
 use super::branch::{BranchId, FakeTree, QueueKind, SpeculativeState};
 use super::scheduler::{SchedulerCtx, ComponentCtx};
+use super::scheduler::{SchedulerCtx, ComponentCtx, MessageTicket};
 use super::port::{Port, PortIdLocal, Channel, PortKind};
 use super::consensus::{Consensus, Consistency, find_ports_in_value_group};
 use super::connector::{ConnectorScheduling, ConnectorPDL};
 use super::inbox::{
     Message, DataMessage,
     SyncCompMessage, SyncPortMessage,
     ControlContent, ControlMessage
 };
 /// Generic connector interface from the scheduler's point of view.
 pub(crate) trait Connector {
     /// Should run the connector's behaviour up until the next blocking point.
     /// One should generally request and handle new messages from the component
     /// context. Then perform any logic the component has to do, and in the
     /// process perhaps queue up some state changes using the same context.
     fn run(&mut self, sched_ctx: SchedulerCtx, comp_ctx: &mut ComponentCtx) -> ConnectorScheduling;
+}
 pub(crate) struct FinishedSync {
     // In the order of the `get` calls
     success: bool,
     inbox: Vec<ValueGroup>,
+}
 type SyncDone = Arc<(Mutex<Option<FinishedSync>>, Condvar)>;
 type JobQueue = Arc<Mutex<VecDeque<ApplicationJob>>>;
 enum ApplicationJob {
     NewChannel((Port, Port)),
     NewConnector(ConnectorPDL, Vec<PortIdLocal>),
     SyncRound(Vec<ApplicationSyncAction>),
     Shutdown,
+}
 // -----------------------------------------------------------------------------
 // ConnectorApplication
 // -----------------------------------------------------------------------------
 /// The connector which an application can directly interface with. Once may set
 /// up the next synchronous round, and retrieve the data afterwards.
 // TODO: Strong candidate for logic reduction in handling put/get. A lot of code
 //  is an approximate copy-pasta from the regular component logic. I'm going to
 //  wait until I'm implementing more native components to see which logic is
 //  truly common.
 pub struct ConnectorApplication {
     // Communicating about new jobs and setting up sync rounds
     sync_done: SyncDone,
     job_queue: JobQueue,
     is_in_sync: bool,
     // Handling current sync round
     sync_desc: Vec<ApplicationSyncAction>,
     tree: FakeTree,
     consensus: Consensus,
     last_finished_handled: Option<BranchId>,
     branch_extra: Vec<usize>, // instruction counter per branch
+}
 impl Connector for ConnectorApplication {
     fn run(&mut self, sched_ctx: SchedulerCtx, comp_ctx: &mut ComponentCtx) -> ConnectorScheduling {
         if self.is_in_sync {
             let scheduling = self.run_in_sync_mode(sched_ctx, comp_ctx);
             let mut iter_id = self.last_finished_handled.or(self.tree.get_queue_first(QueueKind::FinishedSync));
             while let Some(branch_id) = iter_id {
                 iter_id = self.tree.get_queue_next(branch_id);
                 self.last_finished_handled = Some(branch_id);
                 if let Some(conclusion) = self.consensus.handle_new_finished_sync_branch(branch_id, comp_ctx) {
                     // Can finish sync round immediately
                     self.collapse_sync_to_conclusion(conclusion, comp_ctx);
                     return ConnectorScheduling::Immediate;
+                }
+            }
             return scheduling;
         } else {
             return self.run_in_deterministic_mode(sched_ctx, comp_ctx);
+        }
+    }
+}
 impl ConnectorApplication {
     pub(crate) fn new(runtime: Arc<RuntimeInner>) -> (Self, ApplicationInterface) {
         let sync_done = Arc::new(( Mutex::new(None), Condvar::new() ));
         let job_queue = Arc::new(Mutex::new(VecDeque::with_capacity(32)));
         let connector = ConnectorApplication {
             sync_done: sync_done.clone(),
             job_queue: job_queue.clone(),
             is_in_sync: false,
             sync_desc: Vec::new(),
             tree: FakeTree::new(),
             consensus: Consensus::new(),
             last_finished_handled: None,
             branch_extra: vec![0],
         };
         let interface = ApplicationInterface::new(sync_done, job_queue, runtime);
         return (connector, interface);
+    }
     fn handle_new_messages(&mut self, comp_ctx: &mut ComponentCtx) {
         while let Some(message) = comp_ctx.read_next_message() {
             match message {
                 Message::Data(message) => self.handle_new_data_message(message, comp_ctx),
         while let Some(ticket) = comp_ctx.get_next_message_ticket() {
             let message = comp_ctx.read_message_using_ticket(ticket);
             if let Message::Data(_) = message {
                 self.handle_new_data_message(ticket, comp_ctx)
             } else {
                 match comp_ctx.take_message_using_ticket(ticket) {
                     Message::Data(message) => unreachable!(),
                     Message::SyncComp(message) => self.handle_new_sync_comp_message(message, comp_ctx),
                     Message::SyncPort(message) => self.handle_new_sync_port_message(message, comp_ctx),
                     Message::SyncControl(message) => todo!("implement"),
                     Message::Control(_) => unreachable!("control message in native API component"),
+                }
+            }
+        }
+    }
-    pub(crate) fn handle_new_data_message(&mut self, message: DataMessage, ctx: &mut ComponentCtx) {
+    pub(crate) fn handle_new_data_message(&mut self, ticket: MessageTicket, ctx: &mut ComponentCtx) {
         // Go through all branches that are awaiting new messages and see if
         // there is one that can receive this message.
-        if !self.consensus.handle_new_data_message(&message, ctx) {
+        if !self.consensus.handle_new_data_message(ticket, ctx) {
             // Old message, so drop it
             return;
+        }
         let mut iter_id = self.tree.get_queue_first(QueueKind::AwaitingMessage);
         while let Some(branch_id) = iter_id {
             let message = ctx.read_message_using_ticket(ticket).as_data();
             iter_id = self.tree.get_queue_next(branch_id);
             let branch = &self.tree[branch_id];
             if branch.awaiting_port != message.data_header.target_port { continue; }
             if !self.consensus.branch_can_receive(branch_id, &message) { continue; }
             // This branch can receive, so fork and given it the message
             let receiving_branch_id = self.tree.fork_branch(branch_id);
             debug_assert!(receiving_branch_id.index as usize == self.branch_extra.len());
             self.branch_extra.push(self.branch_extra[branch_id.index as usize]); // copy instruction index
             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.clone());
             self.consensus.notify_of_received_message(receiving_branch_id, &message, ctx);
             // And prepare the branch for running
             self.tree.push_into_queue(QueueKind::Runnable, receiving_branch_id);
+        }
+    }
     pub(crate) fn handle_new_sync_comp_message(&mut self, message: SyncCompMessage, ctx: &mut ComponentCtx) {
         if let Some(conclusion) = self.consensus.handle_new_sync_comp_message(message, ctx) {
             self.collapse_sync_to_conclusion(conclusion, ctx);
+        }
+    }
     pub(crate) fn handle_new_sync_port_message(&mut self, message: SyncPortMessage, ctx: &mut ComponentCtx) {
         self.consensus.handle_new_sync_port_message(message, ctx);
+    }
     fn run_in_sync_mode(&mut self, _sched_ctx: SchedulerCtx, comp_ctx: &mut ComponentCtx) -> ConnectorScheduling {
         debug_assert!(self.is_in_sync);
         self.handle_new_messages(comp_ctx);
         let branch_id = self.tree.pop_from_queue(QueueKind::Runnable);
         if branch_id.is_none() {
             return ConnectorScheduling::NotNow;
+        }
         let branch_id = branch_id.unwrap();
         let branch = &mut self.tree[branch_id];
         let mut instruction_idx = self.branch_extra[branch_id.index as usize];
         if instruction_idx >= self.sync_desc.len() {
             // Performed last instruction, so this branch is officially at the
             // end of the synchronous interaction.
             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 {
                 branch.sync_state = SpeculativeState::Inconsistent;
+            }
         } else {
             // We still have instructions to perform
             let cur_instruction = &self.sync_desc[instruction_idx];
             self.branch_extra[branch_id.index as usize] += 1;
             match &cur_instruction {
                 ApplicationSyncAction::Put(port_id, content) => {
                     let port_id = *port_id;
                     let (sync_header, data_header) = self.consensus.handle_message_to_send(branch_id, port_id, &content, comp_ctx);
                     let message = Message::Data(DataMessage {
                         sync_header,
                         data_header,
                         content: content.clone(),
                     });
                     comp_ctx.submit_message(message);
                     self.tree.push_into_queue(QueueKind::Runnable, branch_id);
                     return ConnectorScheduling::Immediate;
                 },
                 ApplicationSyncAction::Get(port_id) => {
                     let port_id = *port_id;
                     branch.sync_state = SpeculativeState::HaltedAtBranchPoint;
                     branch.awaiting_port = port_id;
                     self.tree.push_into_queue(QueueKind::AwaitingMessage, branch_id);
                     let mut any_message_received = false;
                     for message in comp_ctx.get_read_data_messages(port_id) {
                         if self.consensus.branch_can_receive(branch_id, &message) {
                             // 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];
                             debug_assert!(receiving_branch_id.index as usize == self.branch_extra.len());
                             self.branch_extra.push(instruction_idx + 1);
                             branch.insert_message(port_id, message.content.clone());
                             self.consensus.notify_of_new_branch(branch_id, receiving_branch_id);
                             self.consensus.notify_of_received_message(receiving_branch_id, &message, comp_ctx);
                             self.tree.push_into_queue(QueueKind::Runnable, receiving_branch_id);
                             any_message_received = true;
+                        }
+                    }
                     if any_message_received {
                         return ConnectorScheduling::Immediate;
+                    }
+                }
+            }
+        }
         if self.tree.queue_is_empty(QueueKind::Runnable) {
             return ConnectorScheduling::NotNow;
         } else {
             return ConnectorScheduling::Later;
+        }
+    }
     fn run_in_deterministic_mode(&mut self, _sched_ctx: SchedulerCtx, comp_ctx: &mut ComponentCtx) -> ConnectorScheduling {
         debug_assert!(!self.is_in_sync);
         // In non-sync mode the application component doesn't really do anything
         // except performing jobs submitted from the API. This is the only
         // case where we expect to be woken up.
         // Note that we have to communicate to the scheduler when we've received
         // ports or created components (hence: given away ports) *before* we
         // enter a sync round.
         let mut queue = self.job_queue.lock().unwrap();
         while let Some(job) = queue.pop_front() {
             match job {
                 ApplicationJob::NewChannel((endpoint_a, endpoint_b)) => {
                     comp_ctx.push_port(endpoint_a);
                     comp_ctx.push_port(endpoint_b);
                     return ConnectorScheduling::Immediate;
+                }
                 ApplicationJob::NewConnector(connector, initial_ports) => {
                     comp_ctx.push_component(connector, initial_ports);
                     return ConnectorScheduling::Later;
                 },
                 ApplicationJob::SyncRound(mut description) => {
                     // Entering sync mode
                     comp_ctx.notify_sync_start();
                     self.sync_desc = description;
                     self.is_in_sync = true;
                     debug_assert!(self.last_finished_handled.is_none());
                     debug_assert!(self.branch_extra.len() == 1);
                     let first_branch_id = self.tree.start_sync();
                     self.tree.push_into_queue(QueueKind::Runnable, first_branch_id);
                     debug_assert!(first_branch_id.index == 1);
                     self.consensus.start_sync(comp_ctx);
                     self.consensus.notify_of_new_branch(BranchId::new_invalid(), first_branch_id);
                     self.branch_extra.push(0); // set first branch to first instruction
                     return ConnectorScheduling::Immediate;
                 },
                 ApplicationJob::Shutdown => {
                     debug_assert!(queue.is_empty());
                     return ConnectorScheduling::Exit;
+                }
+            }
+        }
         // Queue was empty
         return ConnectorScheduling::NotNow;
+    }
     fn collapse_sync_to_conclusion(&mut self, conclusion: RoundConclusion, comp_ctx: &mut ComponentCtx) {
         // Notifying tree, consensus algorithm and context of ending sync
         let mut fake_vec = Vec::new();
         let (branch_id, success) = match conclusion {
             RoundConclusion::Success(branch_id) => {
                 debug_assert!(self.branch_extra[branch_id.index as usize] >= self.sync_desc.len()); // finished program provided by API
                 (branch_id, true)
             },
             RoundConclusion::Failure => (BranchId::new_invalid(), false),
         };
         let mut solution_branch = self.tree.end_sync(branch_id);
         self.consensus.end_sync(branch_id, &mut fake_vec);
         debug_assert!(fake_vec.is_empty());
         comp_ctx.notify_sync_end(&[]);
         // Turning hashmapped inbox into vector of values
         let mut inbox = Vec::with_capacity(solution_branch.inbox.len());
         for action in &self.sync_desc {
             match action {
                 ApplicationSyncAction::Put(_, _) => {},
                 ApplicationSyncAction::Get(port_id) => {
                     debug_assert!(solution_branch.inbox.contains_key(port_id));
                     inbox.push(solution_branch.inbox.remove(port_id).unwrap());
                 },
+            }
+        }
         // Notifying interface of ending sync
         self.is_in_sync = false;
         self.sync_desc.clear();
         self.branch_extra.truncate(1);
         self.last_finished_handled = None;
         let (results, notification) = &*self.sync_done;
         let mut results = results.lock().unwrap();
         *results = Some(FinishedSync{ success, inbox });
         notification.notify_one();
+    }
+}
 // -----------------------------------------------------------------------------
 // ApplicationInterface
 // -----------------------------------------------------------------------------
 #[derive(Debug)]
 pub enum ChannelCreationError {
     InSync,
+}
 #[derive(Debug)]
 pub enum ApplicationStartSyncError {
     AlreadyInSync,
     NoSyncActions,
     IncorrectPortKind,
     UnownedPort,
+}
 #[derive(Debug)]
 pub enum ApplicationEndSyncError {
     NotInSync,
     Failure,
+}
 pub enum ApplicationSyncAction {
     Put(PortIdLocal, ValueGroup),
     Get(PortIdLocal),
+}
 /// The interface to a `ApplicationConnector`. This allows setting up the
 /// interactions the `ApplicationConnector` performs within a synchronous round.
 pub struct ApplicationInterface {
     sync_done: SyncDone,
     job_queue: JobQueue,
     runtime: Arc<RuntimeInner>,
     is_in_sync: bool,
     connector_id: ConnectorId,
     owned_ports: Vec<(PortKind, PortIdLocal)>,
+}
 impl ApplicationInterface {
     fn new(sync_done: SyncDone, job_queue: JobQueue, runtime: Arc<RuntimeInner>) -> Self {
         return Self{
             sync_done, job_queue, runtime,
             is_in_sync: false,
             connector_id: ConnectorId::new_invalid(),
             owned_ports: Vec::new(),
+        }
+    }
     /// Creates a new channel. Can only fail if the application interface is
     /// currently in sync mode.
     pub fn create_channel(&mut self) -> Result<Channel, ChannelCreationError> {
         if self.is_in_sync {
             return Err(ChannelCreationError::InSync);
+        }
         let (getter_port, putter_port) = self.runtime.create_channel(self.connector_id);
         debug_assert_eq!(getter_port.kind, PortKind::Getter);
         let getter_id = getter_port.self_id;
         let putter_id = putter_port.self_id;
+        {
             let mut lock = self.job_queue.lock().unwrap();
             lock.push_back(ApplicationJob::NewChannel((getter_port, putter_port)));
+        }
         // Add to owned ports for error checking while creating a connector
         self.owned_ports.reserve(2);
         self.owned_ports.push((PortKind::Putter, putter_id));
         self.owned_ports.push((PortKind::Getter, getter_id));
         return Ok(Channel{ putter_id, getter_id });
+    }
     /// Creates a new connector. Note that it is not scheduled immediately, but
     /// depends on the `ApplicationConnector` to run, followed by the created
     /// connector being scheduled.
     pub fn create_connector(&mut self, module: &str, routine: &str, arguments: ValueGroup) -> Result<(), ComponentCreationError> {
         if self.is_in_sync {
             return Err(ComponentCreationError::InSync);
+        }
         // Retrieve ports and make sure that we own the ones that are currently
         // specified. This is also checked by the scheduler, but that is done
         // asynchronously.
         let mut initial_ports = Vec::new();
         find_ports_in_value_group(&arguments, &mut initial_ports);
         for initial_port in &initial_ports {
             if !self.owned_ports.iter().any(|(_, v)| v == initial_port) {
                 return Err(ComponentCreationError::UnownedPort);
+            }
+        }
         // We own all ports, so remove them on this side
         for initial_port in &initial_ports {
             let position = self.owned_ports.iter().position(|(_, v)| v == initial_port).unwrap();
             self.owned_ports.remove(position);
+        }
         let prompt = self.runtime.protocol_description.new_component_v2(module.as_bytes(), routine.as_bytes(), arguments)?;
         let connector = ConnectorPDL::new(prompt);
         // Put on job queue
+        {
             let mut queue = self.job_queue.lock().unwrap();
             queue.push_back(ApplicationJob::NewConnector(connector, initial_ports));
+        }
         self.wake_up_connector_with_ping();
         return Ok(());
+    }
     /// Queues up a description of a synchronous round to run. Will not actually
     /// run the synchronous behaviour in blocking fashion. The results *must* be
     /// retrieved using `try_wait` or `wait` for the interface to be considered
     /// in non-sync mode.
     // TODO: Maybe change API in the future. For now it does the job
     pub fn perform_sync_round(&mut self, actions: Vec<ApplicationSyncAction>) -> Result<(), ApplicationStartSyncError> {
         if self.is_in_sync {
             return Err(ApplicationStartSyncError::AlreadyInSync);
+        }
         // Check the action ports for consistency
         for action in &actions {
             let (port_id, expected_kind) = match action {
                 ApplicationSyncAction::Put(port_id, _) => (*port_id, PortKind::Putter),
                 ApplicationSyncAction::Get(port_id) => (*port_id, PortKind::Getter),
             };
             match self.find_port_by_id(port_id) {
                 Some(port_kind) => {
                     if port_kind != expected_kind {
                         return Err(ApplicationStartSyncError::IncorrectPortKind)
+                    }
                 },
                 None => {
                     return Err(ApplicationStartSyncError::UnownedPort);
+                }
+            }
+        }
         // Everything is consistent, go into sync mode and send the actions off
         // to the component that will actually perform the sync round
         self.is_in_sync = true;
+        {
             let (is_done, _) = &*self.sync_done;
             let mut lock = is_done.lock().unwrap();
             *lock = None;
+        }
+        {
             let mut lock = self.job_queue.lock().unwrap();
             lock.push_back(ApplicationJob::SyncRound(actions));
+        }
         self.wake_up_connector_with_ping();
         return Ok(())
+    }
     /// Wait until the next sync-round is finished, returning the received
     /// messages in order of `get` calls.
     pub fn wait(&mut self) -> Result<Vec<ValueGroup>, ApplicationEndSyncError> {
         if !self.is_in_sync {
             return Err(ApplicationEndSyncError::NotInSync);
+        }
         let (is_done, condition) = &*self.sync_done;
         let mut lock = is_done.lock().unwrap();
         lock = condition.wait_while(lock, |v| v.is_none()).unwrap(); // wait while not done
         self.is_in_sync = false;
         let result = lock.take().unwrap();
         if result.success {
             return Ok(result.inbox);
         } else {
             return Err(ApplicationEndSyncError::Failure);
+        }
+    }
     /// Called by runtime to set associated connector's ID.
     pub(crate) fn set_connector_id(&mut self, id: ConnectorId) {
         self.connector_id = id;
+    }
     fn wake_up_connector_with_ping(&self) {
         let connector = self.runtime.get_component_public(self.connector_id);
         connector.inbox.insert_message(Message::Control(ControlMessage {
             id: 0,
             sending_component_id: self.connector_id,
             content: ControlContent::Ping,
         }));
         let should_wake_up = connector.sleeping
             .compare_exchange(true, false, Ordering::SeqCst, Ordering::Acquire)
             .is_ok();
         if should_wake_up {
             let key = unsafe{ ConnectorKey::from_id(self.connector_id) };
             self.runtime.push_work(key);
+        }
+    }
     fn find_port_by_id(&self, port_id: PortIdLocal) -> Option<PortKind> {
         return self.owned_ports.iter()
             .find(|(_, owned_id)| *owned_id == port_id)
             .map(|(port_kind, _)| *port_kind);
+    }
+}
 impl Drop for ApplicationInterface {
     fn drop(&mut self) {
+        {
             let mut lock = self.job_queue.lock().unwrap();
             lock.push_back(ApplicationJob::Shutdown);
+        }
         self.wake_up_connector_with_ping();
         self.runtime.decrement_active_interfaces();
+    }
+}
@@ \ No newline at end of file @@

src/runtime2/scheduler.rs

➞

Show inline comments

 use std::collections::VecDeque;
 use std::mem::MaybeUninit;
 use std::sync::Arc;
 use std::sync::atomic::Ordering;
 use crate::collections::RawVec;
 use crate::protocol::eval::EvalError;
 use crate::runtime2::port::ChannelId;
 use super::{ScheduledConnector, RuntimeInner, ConnectorId, ConnectorKey};
 use super::port::{Port, PortState, PortIdLocal};
 use super::native::Connector;
 use super::branch::{BranchId};
 use super::connector::{ConnectorPDL, ConnectorScheduling};
 use super::inbox::{
     Message, DataMessage, SyncHeader,
     ControlMessage, ControlContent,
     SyncControlMessage, SyncControlContent,
 };
 // Because it contains pointers we're going to do a copy by value on this one
 #[derive(Clone, Copy)]
 pub(crate) struct SchedulerCtx<'a> {
     pub(crate) runtime: &'a RuntimeInner
+}
 pub(crate) struct Scheduler {
     runtime: Arc<RuntimeInner>,
     scheduler_id: u32,
+}
 impl Scheduler {
     pub fn new(runtime: Arc<RuntimeInner>, scheduler_id: u32) -> Self {
         return Self{ runtime, scheduler_id };
+    }
     pub fn run(&mut self) {
         // Setup global storage and workspaces that are reused for every
         // connector that we run
         'thread_loop: loop {
             // Retrieve a unit of work
             self.debug("Waiting for work");
             let connector_key = self.runtime.wait_for_work();
             if connector_key.is_none() {
                 // We should exit
                 self.debug(" ... No more work, quitting");
                 break 'thread_loop;
+            }
             // We have something to do
             let connector_key = connector_key.unwrap();
             let connector_id = connector_key.downcast();
             self.debug_conn(connector_id, &format!(" ... Got work, running {}", connector_key.index));
             let scheduled = self.runtime.get_component_private(&connector_key);
             // Keep running until we should no longer immediately schedule the
             // connector.
             let mut cur_schedule = ConnectorScheduling::Immediate;
             while let ConnectorScheduling::Immediate = cur_schedule {
                 self.handle_inbox_messages(scheduled);
                 // Run the main behaviour of the connector, depending on its
                 // current state.
                 if scheduled.shutting_down {
                     // Nothing to do. But we're stil waiting for all our pending
                     // control messages to be answered.
                     self.debug_conn(connector_id, &format!("Shutting down, {} Acks remaining", scheduled.router.num_pending_acks()));
                     self.handle_inbox_while_shutting_down(scheduled);
                     if scheduled.router.num_pending_acks() == 0 {
                         // We're actually done, we can safely destroy the
                         // currently running connector
                         self.runtime.destroy_component(connector_key);
                         continue 'thread_loop;
                     } else {
                         cur_schedule = ConnectorScheduling::NotNow;
+                    }
                 } else {
                     self.debug_conn(connector_id, "Running ...");
                     let scheduler_ctx = SchedulerCtx{ runtime: &*self.runtime };
                     let new_schedule = scheduled.connector.run(scheduler_ctx, &mut scheduled.ctx);
                     self.debug_conn(connector_id, &format!("Finished running (new scheduling is {:?})", new_schedule));
                     // Handle all of the output from the current run: messages to
                     // send and connectors to instantiate.
                     self.handle_changes_in_context(scheduled);
                     cur_schedule = new_schedule;
+                }
+            }
             // If here then the connector does not require immediate execution.
             // So enqueue it if requested, and otherwise put it in a sleeping
             // state.
             match cur_schedule {
                 ConnectorScheduling::Immediate => unreachable!(),
                 ConnectorScheduling::Later => {
                     // Simply queue it again later
                     self.runtime.push_work(connector_key);
                 },
                 ConnectorScheduling::NotNow => {
                     // Need to sleep, note that we are the only ones which are
                     // allows to set the sleeping state to `true`, and since
                     // we're running it must currently be `false`.
                     self.try_go_to_sleep(connector_key, scheduled);
                 },
                 ConnectorScheduling::Exit => {
                     // Prepare for exit. Set the shutdown flag and broadcast
                     // messages to notify peers of closing channels
                     scheduled.shutting_down = true;
                     for port in &scheduled.ctx.ports {
                         if port.state != PortState::Closed {
                             let message = scheduled.router.prepare_closing_channel(
                                 port.self_id, port.peer_id,
                                 connector_id
                             );
                             self.debug_conn(connector_id, &format!("Sending message [ exit ] \n --- {:?}", message));
                             self.runtime.send_message(port.peer_connector, Message::Control(message));
+                        }
+                    }
                     if scheduled.router.num_pending_acks() == 0 {
                         // All ports (if any) already closed
                         self.runtime.destroy_component(connector_key);
                         continue 'thread_loop;
+                    }
                     self.try_go_to_sleep(connector_key, scheduled);
                 },
+            }
+        }
+    }
     /// Receiving messages from the public inbox and handling them or storing
     /// them in the component's private inbox
     fn handle_inbox_messages(&mut self, scheduled: &mut ScheduledConnector) {
         let connector_id = scheduled.ctx.id;
         while let Some(message) = scheduled.public.inbox.take_message() {
             // Check if the message has to be rerouted because we have moved the
             // target port to another component.
             self.debug_conn(connector_id, &format!("Handling message\n --- {:#?}", message));
-            if let Some(target_port) = Self::get_message_target_port(&message) {
+            if let Some(target_port) = message.target_port() {
                 if let Some(other_component_id) = scheduled.router.should_reroute(target_port) {
                     self.debug_conn(connector_id, " ... Rerouting the message");
                     self.runtime.send_message(other_component_id, message);
                     continue;
+                }
+            }
             // If here, then we should handle the message
             self.debug_conn(connector_id, " ... Handling the message");
             match message {
                 Message::Control(message) => {
                     match message.content {
                         ControlContent::PortPeerChanged(port_id, new_target_connector_id) => {
                             // Need to change port target
                             let port = scheduled.ctx.get_port_mut_by_id(port_id).unwrap();
                             port.peer_connector = new_target_connector_id;
                             // Note: for simplicity we program the scheduler to always finish
                             // running a connector with an empty outbox. If this ever changes
                             // then accepting the "port peer changed" message implies we need
                             // to change the recipient of the message in the outbox.
                             debug_assert!(scheduled.ctx.outbox.is_empty());
                             // And respond with an Ack
                             let ack_message = Message::Control(ControlMessage {
                                 id: message.id,
                                 sending_component_id: connector_id,
                                 content: ControlContent::Ack,
                             });
                             self.debug_conn(connector_id, &format!("Sending message [pp ack]\n --- {:?}", ack_message));
                             self.runtime.send_message(message.sending_component_id, ack_message);
                         },
                         ControlContent::CloseChannel(port_id) => {
                             // Mark the port as being closed
                             let port = scheduled.ctx.get_port_mut_by_id(port_id).unwrap();
                             port.state = PortState::Closed;
                             // Send an Ack
                             let ack_message = Message::Control(ControlMessage {
                                 id: message.id,
                                 sending_component_id: connector_id,
                                 content: ControlContent::Ack,
                             });
                             self.debug_conn(connector_id, &format!("Sending message [cc ack] \n --- {:?}", ack_message));
                             self.runtime.send_message(message.sending_component_id, ack_message);
                         },
                         ControlContent::Ack => {
                             scheduled.router.handle_ack(message.id);
                             if let Some((target_component, new_control_message)) = scheduled.router.handle_ack(connector_id, message.id) {
                                 self.debug_conn(connector_id, &format!("Sending message [ack ack] \n --- {:?}", new_control_message));
                                 self.runtime.send_message(target_component, new_control_message);
                             };
                         },
                         ControlContent::Ping => {},
+                    }
                 },
                 _ => {
                     // All other cases have to be handled by the component
-                    scheduled.ctx.inbox_messages.push(message);
+                    scheduled.ctx.inbox.insert_new(message);
+                }
+            }
+        }
+    }
     /// Handles inbox messages while shutting down. This intends to handle the
     /// case where a component cleanly exited outside of a sync region, but a
     /// peer, before receiving the `CloseChannel` message, sent a message inside
     /// a sync region. This peer should be notified that its message is not
     /// received by a component in a sync region.
     fn handle_inbox_while_shutting_down(&mut self, scheduled: &mut ScheduledConnector) {
         // Note: we're not handling the public inbox, we're dealing with the
         // private one!
         debug_assert!(scheduled.shutting_down);
         while let Some(message) = scheduled.ctx.read_next_message_even_if_not_in_sync() {
             let target_port_and_round_number = match &message {
         while let Some(ticket) = scheduled.ctx.get_next_message_ticket_even_if_not_in_sync() {
             let message = scheduled.ctx.read_message_using_ticket(ticket);
             let target_port_and_round_number = match message {
                 Message::Data(msg) => Some((msg.data_header.target_port, msg.sync_header.sync_round)),
                 Message::SyncComp(_) => None,
                 Message::SyncPort(msg) => Some((msg.target_port, msg.sync_header.sync_round)),
                 Message::SyncControl(_) => None,
                 Message::Control(_) => None,
             };
             if let Some((target_port, sync_round)) = target_port_and_round_number {
                 // This message is aimed at a port, but we're shutting down, so
                 // notify the peer that its was not received properly.
                 // (also: since we're shutting down, we're not in sync mode and
                 // the context contains the definitive set of owned ports)
                 let port = scheduled.ctx.get_port_by_id(target_port).unwrap();
                 let message = SyncControlMessage{
                     in_response_to_sync_round: sync_round,
                     target_component_id: port.peer_connector,
                     content: SyncControlContent::ChannelIsClosed(port.peer_id),
                 };
                 self.debug_conn(scheduled.ctx.id, &format!("Sending message [shutdown]\n --- {:?}", message));
                 self.runtime.send_message(port.peer_connector, Message::SyncControl(message));
+            }
+        }
+    }
     /// Handles changes to the context that were made by the component. This is
     /// the way (due to Rust's borrowing rules) that we bubble up changes in the
     /// component's state that the scheduler needs to know about (e.g. a message
     /// that the component wants to send, a port that has been added).
     fn handle_changes_in_context(&mut self, scheduled: &mut ScheduledConnector) {
         let connector_id = scheduled.ctx.id;
         // Handling any messages that were sent
         while let Some(message) = scheduled.ctx.outbox.pop_front() {
             self.debug_conn(connector_id, &format!("Sending message [outbox] \n --- {:#?}", message));
             let target_component_id = match &message {
                 Message::Data(content) => {
                     // Data messages are always sent to a particular port, and
                     // may end up being rerouted.
                     let port_desc = scheduled.ctx.get_port_by_id(content.data_header.sending_port).unwrap();
                     debug_assert_eq!(port_desc.peer_id, content.data_header.target_port);
                     if port_desc.state == PortState::Closed {
                         todo!("handle sending over a closed port")
+                    }
                     port_desc.peer_connector
                 },
                 Message::SyncComp(content) => {
                     // Sync messages are always sent to a particular component,
                     // the sender must make sure it actually wants to send to
                     // the specified component (and is not using an inconsistent
                     // component ID associated with a port).
                     content.target_component_id
                 },
                 Message::SyncPort(content) => {
                     let port_desc = scheduled.ctx.get_port_by_id(content.source_port).unwrap();
                     debug_assert_eq!(port_desc.peer_id, content.target_port);
                     if port_desc.state == PortState::Closed {
                         todo!("handle sending over a closed port")
+                    }
                     port_desc.peer_connector
                 },
                 Message::SyncControl(_) => unreachable!("component sending 'SyncControl' messages directly"),
                 Message::Control(_) => unreachable!("component sending 'Control' messages directly"),
             };
             self.runtime.send_message(target_component_id, message);
+        }
         while let Some(state_change) = scheduled.ctx.state_changes.pop_front() {
             match state_change {
                 ComponentStateChange::CreatedComponent(component, initial_ports) => {
                     // Creating a new component. The creator needs to relinquish
                     // ownership of the ports that are given to the new
                     // component. All data messages that were intended for that
                     // port also needs to be transferred.
                     let new_key = self.runtime.create_pdl_component(component, false);
                     let new_connector = self.runtime.get_component_private(&new_key);
                     for port_id in initial_ports {
                         // Transfer messages associated with the transferred port
                         let mut message_idx = 0;
                         while message_idx < scheduled.ctx.inbox_messages.len() {
                             let message = &scheduled.ctx.inbox_messages[message_idx];
                             if Self::get_message_target_port(message) == Some(port_id) {
                                 // Need to transfer this message
                                 // TODO: Revise messages, this is becoming messy and error-prone
                                 let message = scheduled.ctx.inbox_messages.remove(message_idx);
                                 if message_idx < scheduled.ctx.inbox_len_read {
                                     scheduled.ctx.inbox_len_read -= 1;
+                                }
                                 new_connector.ctx.inbox_messages.push(message);
                             } else {
                                 message_idx += 1;
+                            }
+                        }
                         scheduled.ctx.inbox.transfer_messages_for_port(port_id, &mut new_connector.ctx.inbox);
                         // Transfer the port itself
                         let port_index = scheduled.ctx.ports.iter()
                             .position(|v| v.self_id == port_id)
                             .unwrap();
                         let port = scheduled.ctx.ports.remove(port_index);
                         new_connector.ctx.ports.push(port.clone());
                         // Notify the peer that the port has changed, but only
                         // if the port wasn't already closed (otherwise the peer
                         // is gone).
                         if port.state == PortState::Open {
                             let reroute_message = scheduled.router.prepare_reroute(
                                 port.self_id, port.peer_id, scheduled.ctx.id,
                                 port.peer_connector, new_connector.ctx.id
                                 port.peer_connector, new_connector.ctx.id,
                                 &mut new_connector.router
                             );
                             self.debug_conn(connector_id, &format!("Sending message [newcon]\n --- {:?}", reroute_message));
                             self.runtime.send_message(port.peer_connector, Message::Control(reroute_message));
+                        }
+                    }
                     // Schedule new connector to run
                     self.runtime.push_work(new_key);
                 },
                 ComponentStateChange::CreatedPort(port) => {
                     scheduled.ctx.ports.push(port);
                 },
                 ComponentStateChange::ChangedPort(port_change) => {
                     if port_change.is_acquired {
                         scheduled.ctx.ports.push(port_change.port);
                     } else {
                         let index = scheduled.ctx.ports
                             .iter()
                             .position(|v| v.self_id == port_change.port.self_id)
                             .unwrap();
                         scheduled.ctx.ports.remove(index);
+                    }
+                }
+            }
+        }
         // Finally, check if we just entered or just left a sync region
         if scheduled.ctx.changed_in_sync {
             if scheduled.ctx.is_in_sync {
                 // Just entered sync region
             } else {
                 // Just left sync region. So clear inbox up until the last
                 // message that was read.
                 scheduled.ctx.inbox_messages.drain(0..scheduled.ctx.inbox_len_read);
                 scheduled.ctx.inbox_len_read = 0;
                 // Just left sync region. So prepare inbox for the next sync
                 // round
                 scheduled.ctx.inbox.prepare_for_next_round();
+            }
             scheduled.ctx.changed_in_sync = false; // reset flag
+        }
+    }
     fn try_go_to_sleep(&self, connector_key: ConnectorKey, connector: &mut ScheduledConnector) {
         debug_assert_eq!(connector_key.index, connector.ctx.id.0);
         debug_assert_eq!(connector.public.sleeping.load(Ordering::Acquire), false);
         // This is the running connector, and only the running connector may
         // decide it wants to sleep again.
         connector.public.sleeping.store(true, Ordering::Release);
         // But due to reordering we might have received messages from peers who
         // did not consider us sleeping. If so, then we wake ourselves again.
         if !connector.public.inbox.is_empty() {
             // Try to wake ourselves up (needed because someone might be trying
             // the exact same atomic compare-and-swap at this point in time)
             let should_wake_up_again = connector.public.sleeping
                 .compare_exchange(true, false, Ordering::SeqCst, Ordering::Acquire)
                 .is_ok();
             if should_wake_up_again {
                 self.runtime.push_work(connector_key)
+            }
+        }
+    }
     #[inline]
     fn get_message_target_port(message: &Message) -> Option<PortIdLocal> {
         match message {
             Message::Data(data) => return Some(data.data_header.target_port),
             Message::SyncComp(_) => {},
             Message::SyncPort(content) => return Some(content.target_port),
             Message::SyncControl(_) => return None,
             Message::Control(control) => {
                 match &control.content {
                     ControlContent::PortPeerChanged(port_id, _) => return Some(*port_id),
                     ControlContent::CloseChannel(port_id) => return Some(*port_id),
                     ControlContent::Ping | ControlContent::Ack => {},
+                }
             },
+        }
         return None
+    }
     // TODO: Remove, this is debugging stuff
     fn debug(&self, message: &str) {
         println!("DEBUG [thrd:{:02} conn:  ]: {}", self.scheduler_id, message);
+    }
     fn debug_conn(&self, conn: ConnectorId, message: &str) {
         println!("DEBUG [thrd:{:02} conn:{:02}]: {}", self.scheduler_id, conn.0, message);
+    }
+}
 // -----------------------------------------------------------------------------
 // ComponentCtx
 // -----------------------------------------------------------------------------
 enum ComponentStateChange {
     CreatedComponent(ConnectorPDL, Vec<PortIdLocal>),
     CreatedPort(Port),
     ChangedPort(ComponentPortChange),
+}
 #[derive(Clone)]
 pub(crate) struct ComponentPortChange {
     pub is_acquired: bool, // otherwise: released
     pub port: Port,
+}
 /// The component context (better name may be invented). This was created
 /// because part of the component's state is managed by the scheduler, and part
 /// of it by the component itself. When the component starts a sync block or
 /// exits a sync block the partially managed state by both component and
 /// scheduler need to be exchanged.
 pub(crate) struct ComponentCtx {
     // Mostly managed by the scheduler
     pub(crate) id: ConnectorId,
     ports: Vec<Port>,
     inbox_messages: Vec<Message>,
     inbox_len_read: usize,
     inbox: Inbox,
     // Submitted by the component
     is_in_sync: bool,
     changed_in_sync: bool,
     outbox: VecDeque<Message>,
     state_changes: VecDeque<ComponentStateChange>,
     // Workspaces that may be used by components to (generally) prevent
     // allocations. Be a good scout and leave it empty after you've used it.
     // TODO: Move to scheduler ctx, this is the wrong place
     pub workspace_ports: Vec<PortIdLocal>,
     pub workspace_branches: Vec<BranchId>,
+}
 impl ComponentCtx {
     pub(crate) fn new_empty() -> Self {
         return Self{
             id: ConnectorId::new_invalid(),
             ports: Vec::new(),
             inbox_messages: Vec::new(),
             inbox_len_read: 0,
             inbox: Inbox::new(),
             is_in_sync: false,
             changed_in_sync: false,
             outbox: VecDeque::new(),
             state_changes: VecDeque::new(),
             workspace_ports: Vec::new(),
             workspace_branches: Vec::new(),
         };
+    }
     /// Notify the runtime that the component has created a new component. May
     /// only be called outside of a sync block.
     pub(crate) fn push_component(&mut self, component: ConnectorPDL, initial_ports: Vec<PortIdLocal>) {
         debug_assert!(!self.is_in_sync);
         self.state_changes.push_back(ComponentStateChange::CreatedComponent(component, initial_ports));
+    }
     /// Notify the runtime that the component has created a new port. May only
     /// be called outside of a sync block (for ports received during a sync
     /// block, pass them when calling `notify_sync_end`).
     pub(crate) fn push_port(&mut self, port: Port) {
         debug_assert!(!self.is_in_sync);
         self.state_changes.push_back(ComponentStateChange::CreatedPort(port))
+    }
     /// Notify the runtime of an error. Note that this will not perform any
     /// special action beyond printing the error. The component is responsible
     /// for waiting until it is appropriate to shut down (i.e. being outside
     /// of a sync region) and returning the `Exit` scheduling code.
     pub(crate) fn push_error(&mut self, error: EvalError) {
         println!("ERROR: Component ({}) encountered a critical error:\n{}", self.id.0, error);
+    }
     #[inline]
     pub(crate) fn get_ports(&self) -> &[Port] {
         return self.ports.as_slice();
+    }
     pub(crate) fn get_port_by_id(&self, id: PortIdLocal) -> Option<&Port> {
         return self.ports.iter().find(|v| v.self_id == id);
+    }
     pub(crate) fn get_port_by_channel_id(&self, id: ChannelId) -> Option<&Port> {
         return self.ports.iter().find(|v| v.channel_id == id);
+    }
     fn get_port_mut_by_id(&mut self, id: PortIdLocal) -> Option<&mut Port> {
         return self.ports.iter_mut().find(|v| v.self_id == id);
+    }
     /// Notify that component will enter a sync block. Note that after calling
     /// this function you must allow the scheduler to pick up the changes in the
     /// context by exiting your code-executing loop, and to continue executing
     /// code the next time the scheduler picks up the component.
     pub(crate) fn notify_sync_start(&mut self) {
         debug_assert!(!self.is_in_sync);
         self.is_in_sync = true;
         self.changed_in_sync = true;
+    }
     #[inline]
     pub(crate) fn is_in_sync(&self) -> bool {
         return self.is_in_sync;
+    }
     /// Submit a message for the scheduler to send to the appropriate receiver.
     /// May only be called inside of a sync block.
     pub(crate) fn submit_message(&mut self, contents: Message) -> Result<(), ()> {
         debug_assert!(self.is_in_sync);
         if let Some(port_id) = contents.source_port() {
             let port_info = self.get_port_by_id(port_id);
             let is_valid = match port_info {
                 Some(port_info) => {
                     port_info.state == PortState::Open
                 },
                 None => false,
             };
             if !is_valid {
                 // We don't own the port
                 println!(" ****** DEBUG ****** : Sending through closed port!!! {}", port_id.index);
                 return Err(());
+            }
+        }
         self.outbox.push_back(contents);
         return Ok(());
+    }
     /// Notify that component just finished a sync block. Like
     /// `notify_sync_start`: drop out of the `Component::Run` function.
     pub(crate) fn notify_sync_end(&mut self, changed_ports: &[ComponentPortChange]) {
         debug_assert!(self.is_in_sync);
         self.is_in_sync = false;
         self.changed_in_sync = true;
         self.state_changes.reserve(changed_ports.len());
         for changed_port in changed_ports {
             self.state_changes.push_back(ComponentStateChange::ChangedPort(changed_port.clone()));
+        }
+    }
     /// Retrieves messages matching a particular port and branch id. But only
     /// those messages that have been previously received with
     /// `read_next_message`.
     pub(crate) fn get_read_data_messages(&self, match_port_id: PortIdLocal) -> MessagesIter {
         return MessagesIter {
             messages: &self.inbox_messages,
             next_index: 0,
             max_index: self.inbox_len_read,
             match_port_id
         };
         return self.inbox.get_read_data_messages(match_port_id);
+    }
     /// Retrieves the next unread message from the inbox `None` if there are no
     /// (new) messages to read.
     // TODO: Fix the clone of the data message, entirely unnecessary
     pub(crate) fn read_next_message(&mut self) -> Option<Message> {
     pub(crate) fn get_next_message_ticket(&mut self) -> Option<MessageTicket> {
         if !self.is_in_sync { return None; }
-        return self.read_next_message_even_if_not_in_sync();
+        return self.inbox.get_next_message_ticket();
+    }
     pub(crate) fn read_next_message_even_if_not_in_sync(&mut self) -> Option<Message> {
         if self.inbox_len_read == self.inbox_messages.len() { return None; }
     #[inline]
     pub(crate) fn get_next_message_ticket_even_if_not_in_sync(&mut self) -> Option<MessageTicket> {
         return self.inbox.get_next_message_ticket();
+    }
         // We want to keep data messages in the inbox, because we need to check
         // them in the future. We don't want to keep sync messages around, we
         // should only handle them once. Control messages should never be in
         // here.
         let message = &self.inbox_messages[self.inbox_len_read];
         match &message {
             Message::Data(content) => {
                 // Keep message in inbox for later reading
                 self.inbox_len_read += 1;
                 return Some(Message::Data(content.clone()));
             },
             Message::SyncComp(_) | Message::SyncPort(_) | Message::SyncControl(_) => {
                 // Remove message from inbox
                 let message = self.inbox_messages.remove(self.inbox_len_read);
                 return Some(message);
             },
             Message::Control(_) => unreachable!("control message ended up in component inbox"),
     #[inline]
     pub(crate) fn read_message_using_ticket(&self, ticket: MessageTicket) -> &Message {
         return self.inbox.read_message_using_ticket(ticket);
+    }
     #[inline]
     pub(crate) fn take_message_using_ticket(&mut self, ticket: MessageTicket) -> Message {
         return self.inbox.take_message_using_ticket(ticket)
+    }
     /// Puts back a message back into the inbox. The reason being that the
     /// message is actually part of the next sync round. This will
     pub(crate) fn put_back_message(&mut self, message: Message) {
         self.inbox.put_back_message(message);
+    }
+}
 pub(crate) struct MessagesIter<'a> {
     messages: &'a [Message],
     next_index: usize,
     max_index: usize,
     match_port_id: PortIdLocal,
+}
 impl<'a> Iterator for MessagesIter<'a> {
     type Item = &'a DataMessage;
     fn next(&mut self) -> Option<Self::Item> {
         // Loop until match is found or at end of messages
         while self.next_index < self.max_index {
             let message = &self.messages[self.next_index];
             if let Message::Data(message) = &message {
                 if message.data_header.target_port == self.match_port_id {
                     // Found a match
                     self.next_index += 1;
                     return Some(message);
+                }
             } else {
                 // Unreachable because:
                 //  1. We only iterate over messages that were previously retrieved by `read_next_message`.
                 //  2. Inbox does not contain control/ping messages.
                 //  3. If `read_next_message` encounters anything else than a data message, it is removed from the inbox.
                 unreachable!();
+            }
             self.next_index += 1;
+        }
         // No more messages
         return None;
+    }
+}
 // -----------------------------------------------------------------------------
 // Private Inbox
 // -----------------------------------------------------------------------------
 /// A structure that contains inbox messages. Some messages are left inside and
 /// continuously re-read. Others are taken out, but may potentially be put back
 /// for later reading. Later reading in this case implies that they are put back
 /// for reading in the next sync round.
 struct Inbox {
     messages: RawVec<Message>,
     next_delay_idx: u32,
     start_read_idx: u32,
     next_read_idx: u32,
     generation: u32,
+}
 #[derive(Clone, Copy)]
 pub(crate) struct MessageTicket {
     index: u32,
     generation: u32,
+}
 impl Inbox {
     fn new() -> Self {
         return Inbox {
             messages: RawVec::new(),
             next_delay_idx: 0,
             start_read_idx: 0,
             next_read_idx: 0,
             generation: 0,
+        }
+    }
     fn insert_new(&mut self, message: Message) {
         assert!(self.messages.len() < u32::MAX as usize); // TODO: @Size
         self.messages.push(message);
+    }
     fn get_next_message_ticket(&mut self) -> Option<MessageTicket> {
         let cur_read_idx = self.next_read_idx as usize;
         if cur_read_idx >= self.messages.len() {
             return None;
+        }
         self.generation += 1;
         self.next_read_idx += 1;
         return Some(MessageTicket{
             index: cur_read_idx as u32,
             generation: self.generation
         });
+    }
     fn read_message_using_ticket(&self, ticket: MessageTicket) -> &Message {
         debug_assert_eq!(self.generation, ticket.generation);
         return unsafe{ &*self.messages.get(ticket.index as usize) }
+    }
     fn take_message_using_ticket(&mut self, ticket: MessageTicket) -> Message {
         debug_assert_eq!(self.generation, ticket.generation);
         unsafe {
             let take_idx = ticket.index as usize;
             let val = std::ptr::read(self.messages.get(take_idx));
             // Move messages to the right, clearing up space in the
             // front.
             let num_move_right = take_idx - self.start_read_idx as usize;
             self.messages.move_range(
                 self.start_read_idx as usize,
                 self.start_read_idx as usize + 1,
                 num_move_right
             );
             self.start_read_idx += 1;
             return val;
+        }
+    }
     fn put_back_message(&mut self, message: Message) {
         // We have space in front of the array because we've taken out a message
         // before.
         debug_assert!(self.next_delay_idx < self.start_read_idx);
         unsafe {
             // Write to front of the array
             std::ptr::write(self.messages.get_mut(self.next_delay_idx as usize), message);
             self.next_delay_idx += 1;
+        }
+    }
     fn get_read_data_messages(&self, match_port_id: PortIdLocal) -> MessagesIter {
         return MessagesIter{
             messages: self.messages.as_slice(),
             next_index: self.start_read_idx as usize,
             max_index: self.next_read_idx as usize,
             match_port_id
         };
+    }
     fn prepare_for_next_round(&mut self) {
         // Deallocate everything that was read
         self.destroy_range(self.start_read_idx, self.next_read_idx);
         self.generation += 1;
         // Join up all remaining values with the delayed ones in the front
         let num_to_move = self.messages.len() - self.next_read_idx as usize;
         self.messages.move_range(
             self.next_read_idx as usize,
             self.next_delay_idx as usize,
             num_to_move
         );
         // Set all indices (and the RawVec len) to make sense in this new state
         let new_len = self.next_delay_idx as usize + num_to_move;
         self.next_delay_idx = 0;
         self.start_read_idx = 0;
         self.next_read_idx = 0;
         self.messages.len = new_len;
+    }
     fn transfer_messages_for_port(&mut self, port: PortIdLocal, new_inbox: &mut Inbox) {
         // Convoluted assert to make sure we're in non-sync mode, as that is
         // when this is called, and that makes our lives easier
         let mut idx = 0;
         while idx < self.messages.len() {
             let message = unsafe{ &*self.messages.get(idx) };
             if let Some(target_port) = message.target_port() {
                 if target_port == port {
                     // Transfer port
                     unsafe {
                         let message = std::ptr::read(message as *const _);
                         let remaining = self.messages.len() - idx;
                         if remaining > 1 {
                             self.messages.move_range(idx + 1, idx, remaining - 1);
+                        }
                         self.messages.len -= 1;
                         new_inbox.insert_new(message);
+                    }
                 } else {
                     // Do not transfer port
                     idx += 1;
+                }
+            }
+        }
+    }
     #[inline]
     fn destroy_range(&mut self, start_idx: u32, end_idx: u32) {
         for idx in (start_idx as usize)..(end_idx as usize) {
             unsafe {
                 let msg = self.messages.get_mut(idx);
                 std::ptr::drop_in_place(msg);
+            }
+        }
+    }
+}
 impl Drop for Inbox {
     fn drop(&mut self) {
         // Whether in sync or not in sync. We have two ranges of allocated
         // messages:
         // - delayed messages: from 0 to `next_delay_idx` (which is 0 if in non-sync)
         // - readable messages: from `start_read_idx` to `messages.len`
         self.destroy_range(0, self.next_delay_idx);
         self.destroy_range(self.start_read_idx, self.messages.len as u32);
+    }
+}
 // -----------------------------------------------------------------------------
 // Control messages
 // -----------------------------------------------------------------------------
 struct ControlEntry {
     id: u32,
     variant: ControlVariant,
+}
 enum ControlVariant {
     ChangedPort(ControlChangedPort),
     ClosedChannel(ControlClosedChannel),
     ReroutePending,
+}
 struct ControlChangedPort {
     target_port: PortIdLocal,       // if send to this port, then reroute
     source_connector: ConnectorId,  // connector we expect messages from
     target_connector: ConnectorId,  // connector we need to reroute to
     id_of_ack_after_confirmation: u32, // control message ID we need to send to the target upon receiving an ack
+}
 struct ControlClosedChannel {
     source_port: PortIdLocal,
     target_port: PortIdLocal,
+}
 pub(crate) struct ControlMessageHandler {
     id_counter: u32,
     active: Vec<ControlEntry>,
+}
 impl ControlMessageHandler {
     pub fn new() -> Self {
         ControlMessageHandler {
             id_counter: 0,
             active: Vec::new(),
+        }
+    }
     /// Prepares a message indicating that a channel has closed, we keep a local
     /// entry to match against the (hopefully) returned `Ack` message.
     pub fn prepare_closing_channel(
         &mut self, self_port_id: PortIdLocal, peer_port_id: PortIdLocal,
         self_connector_id: ConnectorId
     ) -> ControlMessage {
         let id = self.take_id();
         self.active.push(ControlEntry{
             id,
             variant: ControlVariant::ClosedChannel(ControlClosedChannel{
                 source_port: self_port_id,
                 target_port: peer_port_id,
             }),
         });
         return ControlMessage {
             id,
             sending_component_id: self_connector_id,
             content: ControlContent::CloseChannel(peer_port_id),
         };
+    }
     /// Prepares rerouting messages due to changed ownership of a port. The
     /// control message returned by this function must be sent to the
     /// transferred port's peer connector.
     pub fn prepare_reroute(
         &mut self,
         port_id: PortIdLocal, peer_port_id: PortIdLocal,
         self_connector_id: ConnectorId, peer_connector_id: ConnectorId,
         new_owner_connector_id: ConnectorId
+        new_owner_connector_id: ConnectorId, new_owner_ctrl_handler: &mut ControlMessageHandler,
     ) -> ControlMessage {
         let id = self.take_id();
         let new_owner_id = new_owner_ctrl_handler.take_id();
         self.active.push(ControlEntry{
             id,
             variant: ControlVariant::ChangedPort(ControlChangedPort{
                 target_port: port_id,
                 source_connector: peer_connector_id,
                 target_connector: new_owner_connector_id,
                 id_of_ack_after_confirmation: new_owner_id,
             }),
         });
         new_owner_ctrl_handler.active.push(ControlEntry{
             id: new_owner_id,
             variant: ControlVariant::ReroutePending,
         });
         return ControlMessage {
             id,
             sending_component_id: self_connector_id,
             content: ControlContent::PortPeerChanged(peer_port_id, new_owner_connector_id),
         };
+    }
     /// Returns true if the supplied message should be rerouted. If so then this
     /// function returns the connector that should retrieve this message.
     pub fn should_reroute(&self, target_port: PortIdLocal) -> Option<ConnectorId> {
         for entry in &self.active {
             if let ControlVariant::ChangedPort(entry) = &entry.variant {
                 if entry.target_port == target_port {
                     // Need to reroute this message
                     return Some(entry.target_connector);
+                }
+            }
+        }
         return None;
+    }
     /// Handles an Ack as an answer to a previously sent control message
     pub fn handle_ack(&mut self, id: u32) {
     /// Handles an Ack as an answer to a previously sent control message.
     /// Handling an Ack might spawn a new message that needs to be sent.
     pub fn handle_ack(&mut self, handler_component_id: ConnectorId, id: u32) -> Option<(ConnectorId, Message)> {
         let index = self.active.iter()
             .position(|v| v.id == id);
         match index {
             Some(index) => { self.active.remove(index); },
             None => { todo!("handling of nefarious ACKs"); },
             Some(index) => {
                 let removed = self.active.remove(index);
                 match removed.variant {
                     ControlVariant::ChangedPort(message) => {
                         return Some((
                             message.target_connector,
                             Message::Control(ControlMessage{
                                 id: message.id_of_ack_after_confirmation,
                                 sending_component_id: handler_component_id,
                                 content: ControlContent::Ack
                             })
                         ));
                     },
                     _ => return None,
+                }
             },
             None => {
                 todo!("handling of nefarious ACKs");
                 return None;
             },
+        }
+    }
     /// Retrieves the number of responses we still expect to receive from our
     /// peers
     #[inline]
     pub fn num_pending_acks(&self) -> usize {
         return self.active.len();
+    }
     fn take_id(&mut self) -> u32 {
         let generated_id = self.id_counter;
         let (new_id, _) = self.id_counter.overflowing_add(1);
         self.id_counter = new_id;
         return generated_id;
+    }
+}
@@ \ No newline at end of file @@

src/runtime2/tests/mod.rs

➞

Show inline comments

 mod network_shapes;
 mod api_component;
 mod speculation;
 mod data_transmission;
 mod sync_failure;
 use super::*;
 use crate::{PortId, ProtocolDescription};
 use crate::common::Id;
 use crate::protocol::eval::*;
 use crate::runtime2::native::{ApplicationSyncAction};
 // Generic testing constants, use when appropriate to simplify stress-testing
 // pub(crate) const NUM_THREADS: u32 = 3;     // number of threads in runtime
 // pub(crate) const NUM_INSTANCES: u32 = 7;   // number of test instances constructed
 // pub(crate) const NUM_LOOPS: u32 = 8;       // number of loops within a single test (not used by all tests)
 pub(crate) const NUM_THREADS: u32 = 6;
 pub(crate) const NUM_INSTANCES: u32 = 2;
 pub(crate) const NUM_LOOPS: u32 = 5;
 pub(crate) const NUM_THREADS: u32 = 4;
 pub(crate) const NUM_INSTANCES: u32 = 1;
 pub(crate) const NUM_LOOPS: u32 = 3;
 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);
+    }
+}
 pub(crate) struct TestTimer {
     name: &'static str,
     started: std::time::Instant
+}
 impl TestTimer {
     pub(crate) 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);
+    }
+}

src/runtime2/tests/network_shapes.rs

➞

Show inline comments

 // Testing particular graph shapes
 use super::*;
 #[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) {
             sync {
                 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\");
             sync {
                 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\");
             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),
         ])).expect("create connector");
     });
+}
 #[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) {
             sync {
                 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) {
             sync {
                 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) {
             sync {
                 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(1),
             Value::UInt32(NUM_LOOPS)
         ])).expect("create connector");
     });
+}
@@ \ No newline at end of file @@

0 comments (0 inline, 0 general)