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

Changeset - ce98be9707a6

Parent rev.

Child rev.

[Not reviewed]

0 6 1

MH - 4 years ago 2021-11-06 18:59:24
contact@maxhenger.nl

wip on refactoring component

7 files changed with 571 insertions and 18 deletions:

docs/runtime/design.md

src/runtime2/branch.rs

src/runtime2/connector2.rs

223

src/runtime2/consensus.rs

199

src/runtime2/inbox2.rs

src/runtime2/mod.rs

src/runtime2/scheduler.rs

0 comments (0 inline, 0 general)

docs/runtime/design.md

➞

Show inline comments

 # Runtime Design
 ## Preliminary preliminaries
 There will be some confusion when we're using the word "synchronization". When we talk about OS-syncing we mean using synchronization primitives such as atomics, mutexes, semaphores etc. When we talk about sync-blocks, sync-regions, etc. we mean the Reowolf language's distributed consensus feature.
 ## Preliminary Notes
 The runtime was designed in several iterations. For the purpose of documentation, we have had:
 - Reowolf 1.0: A single-threaded, globally locking runtime.
 - Initial 1.2: A single-threaded runtime, no longer globally locking. The newly designed consensus algorithm worked quite well and reasonably efficiently (not measured by comparison to another runtime, rather, the idea of the consensus algorithm was simple and efficient to perform)
 - Multithreaded 1.2, v1: Here is where we moved towards a more multithreaded design. From the start, the idea was to "maximize concurrency", that is to say: we should only use OS-syncing when absolutely appropriate. Furthermore the initial implementation should be somewhat efficient: we should not employ locks when it is not absolutely necessary. The following remarks can be made with respect to this initial multithreaded implementation:
   - Because there will generally be far more components than there are hardware threads, an efficient implementation requires some kind of scheduler that is able to execute the code of components. To track which components are supposed to run, there will be a work queue. The initial implementation features just a single global work queue. Each thread that is executing components is called a scheduler in this document.
   - At the most basic level, a component has properties that allow access by only one writer at a time, and properties that are conceptually (ofcourse we need some kind of OS synchronization) accessible by multiple writers at a time. At the very least, executing the code of a component should only be performed by one writer at a time (the scheduler), while sending messages to a component should be allowed by multiple schedulers. Hence the runtime splits component properties into two: those that should only be accessed by the scheduler that is executing the code, and those that should be accessible by all schedulers at any time.
   - Components communicate with eachother through transport links. Since messages need to arrive at the correct target. TODO: FINISH
@@ \ No newline at end of file @@

src/runtime2/branch.rs

➞

Show inline comments

 use std::collections::HashMap;
 use std::ops::{Index, IndexMut};
 use crate::protocol::ComponentState;
 use crate::protocol::eval::ValueGroup;
 use crate::runtime2::port::PortIdLocal;
 /// Generic branch ID. A component will always have one branch: the
 /// non-speculative branch. This branch has ID 0. Hence in a speculative context
 /// we use this fact to let branch ID 0 denote the ID being invalid.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub struct BranchId {
     pub index: u32
+}
 impl BranchId {
     #[inline]
     fn new_invalid() -> Self {
         return Self{ index: 0 };
+    }
     #[inline]
     fn new(index: u32) -> Self {
         debug_assert!(index != 0);
         return Self{ index };
+    }
     #[inline]
     pub(crate) fn is_valid(&self) -> bool {
         return self.index != 0;
+    }
+}
 #[derive(Debug, PartialEq, Eq)]
 pub(crate) enum SpeculativeState {
     // Non-synchronous variants
     RunningNonSync,         // regular execution of code
     Error,                  // encountered a runtime error
     Finished,               // finished executing connector's code
     // Synchronous variants
     RunningInSync,          // running within a sync block
     HaltedAtBranchPoint,    // at a branching point (at a `get` call)
     ReachedSyncEnd,         // reached end of sync block, branch represents a local solution
     Inconsistent,           // branch can never represent a local solution, so halted
+}
 /// The execution state of a branch. This envelops the PDL code and the
 /// execution state. And derived from that: if we're ready to keep running the
 /// code, or if we're halted for some reason (e.g. waiting for a message).
 pub(crate) struct Branch {
     pub id: BranchId,
     pub parent_id: BranchId,
     // Execution state
     pub code_state: ComponentState,
     pub sync_state: SpeculativeState,
     pub awaiting_port: PortIdLocal, // only valid if in "awaiting message" queue. TODO: Maybe put in enum
     pub next_in_queue: BranchId, // used by `ExecTree`/`BranchQueue`
     pub inbox: HashMap<PortIdLocal, ValueGroup>; // TODO: Remove, currently only valid in single-get/put mode
+}
 impl Branch {
     /// Creates a new non-speculative branch
     pub(crate) fn new_non_sync(component_state: ComponentState) -> Self {
         Branch {
             id: BranchId::new_invalid(),
             parent_id: BranchId::new_invalid(),
             code_state: component_state,
             sync_state: SpeculativeState::RunningNonSync,
             awaiting_port: PortIdLocal::new_invalid(),
             next_in_queue: BranchId::new_invalid(),
             inbox: HashMap::new(),
+        }
+    }
     /// Constructs a sync branch. The provided branch is assumed to be the
     /// parent of the new branch within the execution tree.
     fn new_sync(new_index: u32, parent_branch: &Branch) -> Self {
         debug_assert!(
             (parent_branch.sync_state == SpeculativeState::RunningNonSync && !parent_branch.parent_index.is_valid()) ||
                 (parent_branch.sync_state == SpeculativeState::HaltedAtBranchPoint)
         );
         debug_assert!(parent_branch.prepared_channel.is_none());
         Branch {
             id: BranchId::new(new_index),
             parent_id: parent_branch.index,
             code_state: parent_branch.code_state.clone(),
             sync_state: SpeculativeState::RunningInSync,
             awaiting_port: parent_branch.awaiting_port,
             next_in_queue: BranchId::new_invalid(),
             inbox: parent_branch.inbox.clone(),
+        }
+    }
     /// Inserts a message into the branch for retrieval by a corresponding
     /// `get(port)` call.
     pub(crate) fn insert_message(&mut self, target_port: PortIdLocal, contents: ValueGroup) {
         debug_assert!(target_port.is_valid());
         debug_assert!(self.awaiting_port == target_port);
         self.awaiting_port = PortIdLocal::new_invalid();
         self.inbox.insert(target_port, contents);
+    }
+}
 /// Queue of branches. Just a little helper.
 #[derive(Copy, Clone)]
 struct BranchQueue {
     first: BranchId,
     last: BranchId,
+}
 impl BranchQueue {
     #[inline]
     fn new() -> Self {
         Self{
             first: BranchId::new_invalid(),
             last: BranchId::new_invalid()
+        }
+    }
     #[inline]
     fn is_empty(&self) -> bool {
         debug_assert!(self.first.is_valid() == self.last.is_valid());
         return !self.first.is_valid();
+    }
+}
 const NUM_QUEUES: usize = 3;
 pub(crate) enum QueueKind {
     Runnable,
     AwaitingMessage,
     FinishedSync,
+}
 impl QueueKind {
     fn as_index(&self) -> usize {
         return match self {
             QueueKind::Runnable => 0,
             QueueKind::AwaitingMessage => 1,
             QueueKind::FinishedSync => 2,
+        }
+    }
+}
 /// Execution tree of branches. Tries to keep the extra information stored
 /// herein to a minimum. So the execution tree is aware of the branches, their
 /// execution state and the way they're dependent on each other, but the
 /// execution tree should not be aware of e.g. sync algorithms.
 ///
 /// Note that the tree keeps track of multiple lists of branches. Each list
 /// contains branches that ended up in a particular execution state. The lists
 /// are described by the various `BranchQueue` instances and the `next_in_queue`
 /// field in each branch.
 pub(crate) struct ExecTree {
     // All branches. the `parent_id` field in each branch implies the shape of
     // the tree. Branches are index stable throughout a sync round.
     pub branches: Vec<Branch>,
     pub queues: [BranchQueue; NUM_QUEUES]
+}
 impl ExecTree {
     /// Constructs a new execution tree with a single non-sync branch.
     pub fn new(component: ComponentState) -> Self {
         return Self {
             branches: vec![Branch::new_non_sync(component)],
             queues: [BranchQueue::new(); 3]
+        }
+    }
     // --- Generic branch (queue) management
     /// Returns if tree is in speculative mode
     pub fn is_in_sync(&self) -> bool {
         return self.branches.len() != 1;
+    }
     /// Returns true if the particular queue is empty
     pub fn queue_is_empty(&self, kind: QueueKind) -> bool {
         return self.queues[kind.as_index()].is_empty();
+    }
     /// Pops a branch (ID) from a queue.
     pub fn pop_from_queue(&mut self, kind: QueueKind) -> Option<BranchId> {
         let queue = &mut self.queues[kind.as_index()];
         if queue.is_empty() {
             return None;
         } else {
             let first_branch = &mut self.branches[queue.first.index as usize];
             queue.first = first_branch.next_in_queue;
             first_branch.next_in_queue = BranchId::new_invalid();
             if !queue.first.is_valid() {
                 queue.last = BranchId::new_invalid();
+            }
             return Some(first_branch.id);
+        }
+    }
     /// Pushes a branch (ID) into a queue.
     pub fn push_into_queue(&mut self, kind: QueueKind, id: BranchId) {
         let queue = &mut self.queues[kind.as_index()];
         if queue.is_empty() {
             queue.first = id;
             queue.last = id;
         } else {
             let last_branch = &mut self.branches[queue.last.index as usize];
             last_branch.next_in_queue = id;
             queue.last = id;
+        }
+    }
     pub fn iter_queue(&self, kind: QueueKind) -> BranchIter {
     /// Returns an iterator over all the elements in the queue of the given kind
     pub fn iter_queue(&self, kind: QueueKind) -> BranchQueueIter {
         let queue = &self.queues[kind.as_index()];
         let index = queue.first as usize;
-        return BranchIter{
+        return BranchQueueIter {
             branches: self.branches.as_slice(),
             index,
+        }
+    }
     /// Returns an iterator that starts with the provided branch, and then
     /// continues to visit all of the branch's parents.
     pub fn iter_parents(&self, branch_id: BranchId) -> BranchParentIter {
         return BranchParentIter{
             branches: self.branches.as_slice(),
             index: branch_id.index as usize,
+        }
+    }
     // --- Preparing and finishing a speculative round
     /// Starts a synchronous round by cloning the non-sync branch and marking it
     /// as the root of the speculative tree.
     pub fn start_sync(&mut self) {
         debug_assert!(!self.is_in_sync());
         let sync_branch = Branch::new_sync(1, &self.branches[0]);
         let sync_branch_id = sync_branch.id;
         self.branches.push(sync_branch);
         self.push_into_queue(QueueKind::Runnable, sync_branch_id);
+    }
     /// Creates a new speculative branch based on the provided one. The index to
     /// retrieve this new branch will be returned.
-    pub fn fork_branch(&mut self, parent_branch_id: BranchId, initial_queue: Option<QueueKind>) -> BranchId {
     pub fn fork_branch(&mut self, parent_branch_id: BranchId) -> BranchId {
         debug_assert!(self.is_in_sync());
         let parent_branch = &self[parent_branch_id];
         let new_branch = Branch::new_sync(1, parent_branch);
         let new_branch_id = new_branch.id;
         if let Some(kind) = initial_queue {
             self.push_into_queue(kind, new_branch_id);
+        }
         return new_branch_id;
+    }
     /// Collapses the speculative execution tree back into a deterministic one,
     /// using the provided branch as the final sync result.
     pub fn end_sync(&mut self, branch_id: BranchId) {
         debug_assert!(self.is_in_sync());
         debug_assert!(self.iter_queue(QueueKind::FinishedSync).any(|v| v.id == branch_id));
         // Swap indicated branch into the first position
         self.branches.swap(0, branch_id.index as usize);
         self.branches.truncate(1);
         // Reset all values to non-sync defaults
         let branch = &mut self.branches[0];
         branch.id = BranchId::new_invalid();
         branch.parent_id = BranchId::new_invalid();
         branch.sync_state = SpeculativeState::RunningNonSync;
         branch.next_in_queue = BranchId::new_invalid();
         // Clear out all the queues
         for queue_idx in 0..NUM_QUEUES {
             self.queues[queue_idx] = BranchQueue::new();
+        }
+    }
+}
 impl Index<BranchId> for ExecTree {
     type Output = Branch;
     fn index(&self, index: BranchId) -> &Self::Output {
         debug_assert!(index.is_valid());
         return &self.branches[index.index as usize];
+    }
+}
 impl IndexMut<BranchId> for ExecTree {
     fn index_mut(&mut self, index: BranchId) -> &mut Self::Output {
         debug_assert!(index.is_valid());
         return &mut self.branches[index.index as usize];
+    }
+}
-pub struct BranchIter<'a> {
+pub struct BranchQueueIter<'a> {
     branches: &'a [Branch],
     index: usize,
+}
-impl<'a> Iterator for BranchIter<'a> {
+impl<'a> Iterator for BranchQueueIter<'a> {
     type Item = &'a Branch;
     fn next(&mut self) -> Option<Self::Item> {
         if self.index == 0 {
             // i.e. the invalid branch index
             return None;
+        }
         let branch = &self.branches[self.index];
         self.index = branch.next_in_queue.index as usize;
         return Some(branch);
+    }
+}
 pub struct BranchParentIter<'a> {
     branches: &'a [Branch],
     index: usize,
+}
 impl<'a> Iterator for BranchParentIter<'a> {
     type Item = &'a Branch;
     fn next(&mut self) -> Option<Self::Item> {
         if self.index == 0 {
             return None;
+        }
         let branch = &self.branches[self.index];
         self.index = branch.parent_id.index as usize;
         return Some(branch);
+    }
+}
@@ \ No newline at end of file @@

src/runtime2/connector2.rs

➞

Show inline comments

@@ new file 100644 @@
 use std::sync::atomic::AtomicBool;
 use crate::common::ComponentState;
 use crate::PortId;
 use crate::protocol::eval::{Value, ValueGroup};
 use crate::protocol::{RunContext, RunResult};
 use crate::runtime2::branch::{Branch, BranchId, ExecTree, QueueKind, SpeculativeState};
 use crate::runtime2::connector::ConnectorScheduling;
 use crate::runtime2::consensus::{Consensus, Consistency};
 use crate::runtime2::inbox2::{DataMessageFancy, MessageFancy, SyncMessageFancy};
 use crate::runtime2::inbox::PublicInbox;
 use crate::runtime2::native::Connector;
 use crate::runtime2::port::PortIdLocal;
 use crate::runtime2::scheduler::{ComponentCtxFancy, SchedulerCtx};
 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),
+        }
+    }
+}
 pub(crate) struct ConnectorPDL {
     tree: ExecTree,
     consensus: Consensus,
     branch_workspace: Vec<BranchId>,
+}
 struct ConnectorRunContext {};
 impl RunContext for ConnectorRunContext{
     fn did_put(&mut self, port: PortId) -> bool {
         todo!()
+    }
     fn get(&mut self, port: PortId) -> Option<ValueGroup> {
         todo!()
+    }
     fn fires(&mut self, port: PortId) -> Option<Value> {
         todo!()
+    }
     fn get_channel(&mut self) -> Option<(Value, Value)> {
         todo!()
+    }
+}
 impl Connector for ConnectorPDL {
     fn run(&mut self, sched_ctx: SchedulerCtx, comp_ctx: &mut ComponentCtxFancy) -> ConnectorScheduling {
         todo!()
+    }
+}
 impl ConnectorPDL {
     pub fn new(initial: ComponentState, owned_ports: Vec<PortIdLocal>) -> Self {
         Self{
             tree: ExecTree::new(initial),
             consensus: Consensus::new(),
+        }
+    }
     // --- Handling messages
     pub fn handle_new_messages(&mut self, ctx: &mut ComponentCtxFancy) {
         while let Some(message) = ctx.read_next_message() {
             match message {
                 MessageFancy::Data(message) => handle_new_data_message(message, ctx),
                 MessageFancy::Sync(message) => handle_new_sync_message(message, ctx),
                 MessageFancy::Control(_) => unreachable!("control message in component"),
+            }
+        }
+    }
     pub fn handle_new_data_message(&mut self, message: DataMessageFancy, ctx: &mut ComponentCtxFancy) {
         // Go through all branches that are awaiting new messages and see if
         // there is one that can receive this message.
         debug_assert!(self.branch_workspace.is_empty());
         self.consensus.handle_received_sync_header(&message.sync_header, ctx);
         self.consensus.handle_received_data_header(&self.tree, &message.data_header, &mut self.branch_workspace);
         for branch_id in self.branch_workspace.drain(..) {
             // This branch can receive, so fork and given it the message
             let receiving_branch_id = self.tree.fork_branch(branch_id);
             self.consensus.notify_of_new_branch(branch_id, receiving_branch_id);
             let receiving_branch = &mut self.tree[receiving_branch_id];
             receiving_branch.insert_message(message.data_header.target_port, message.content.clone());
             self.consensus.notify_of_received_message(branch_id, &message.data_header);
             // And prepare the branch for running
             self.tree.push_into_queue(QueueKind::Runnable, receiving_branch_id);
+        }
+    }
     pub fn handle_new_sync_message(&mut self, message: SyncMessageFancy, ctx: &mut ComponentCtxFancy) {
         self.consensus.handle_received_sync_header(&message.sync_header, ctx);
         todo!("handle content of message?");
+    }
     // --- Running code
     pub fn run_in_sync_mode(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut ComponentCtxFancy) -> ConnectorScheduling {
         // Check if we have any branch that needs running
         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{};
         let run_result = branch.code_state.run(&mut run_context, &sched_ctx.runtime.protocol_description);
         // Handle the returned result. Note that this match statement contains
         // explicit returns in case the run result requires that the component's
         // code is ran again immediately
         match run_result {
             RunResult::BranchInconsistent => {
                 // Branch became inconsistent
                 branch.sync_state = SpeculativeState::Inconsistent;
             },
             RunResult::BranchMissingPortState(port_id) => {
                 // Branch called `fires()` on a port that has not been used yet.
                 let port_id = PortIdLocal::new(port_id.0.u32_suffix);
                 // Create two forks, one that assumes the port will fire, and
                 // one that assumes the port remains silent
                 branch.sync_state = SpeculativeState::HaltedAtBranchPoint;
                 let firing_branch_id = self.tree.fork_branch(branch_id);
                 let silent_branch_id = self.tree.fork_branch(branch_id);
                 self.consensus.notify_of_new_branch(branch_id, firing_branch_id);
                 let _result = self.consensus.notify_of_speculative_mapping(firing_branch_id, port_id, true);
                 debug_assert_eq!(_result, Consistency::Valid);
                 self.consensus.notify_of_new_branch(branch_id, silent_branch_id);
                 let _result = self.consensus.notify_of_speculative_mapping(silent_branch_id, port_id, false);
                 debug_assert_eq!(_result, Consistency::Valid);
                 // Somewhat important: we push the firing one first, such that
                 // that branch is ran again immediately.
                 self.tree.push_into_queue(QueueKind::Runnable, firing_branch_id);
                 self.tree.push_into_queue(QueueKind::Runnable, silent_branch_id);
                 return ConnectorScheduling::Immediate;
             },
             RunResult::BranchMissingPortValue(port_id) => {
                 // Branch performed a `get()` on a port that does not have a
                 // received message on that port.
                 let port_id = PortIdLocal::new(port_id.0.u32_suffix);
                 let consistency = self.consensus.notify_of_speculative_mapping(branch_id, port_id, true);
                 if consistency == Consistency::Valid {
                     // `get()` is valid, so mark the branch as awaiting a message
                     branch.sync_state = SpeculativeState::HaltedAtBranchPoint;
                     branch.awaiting_port = port_id;
                     self.tree.push_into_queue(QueueKind::AwaitingMessage, branch_id);
                     // Note: we only know that a branch is waiting on a message when
                     // it reaches the `get` call. But we might have already received
                     // a message that targets this branch, so check now.
                     let mut any_branch_received = false;
                     for message in comp_ctx.get_read_data_messages(port_id) {
                         if self.consensus.branch_can_receive(branch_id, &message.data_header) {
                             // This branch can receive the message, so we do the
                             // fork-and-receive dance
                             let recv_branch_id = self.tree.fork_branch(branch_id);
                             let branch = &mut self.tree[recv_branch_id];
                             branch.insert_message(port_id, message.content.clone());
                             self.consensus.notify_of_new_branch(branch_id, recv_branch_id);
                             self.consensus.notify_of_received_message(recv_branch_id, &message.data_header);
                             self.tree.push_into_queue(QueueKind::Runnable, recv_branch_id);
                             any_branch_received = true;
+                        }
+                    }
                     if any_branch_received {
                         return ConnectorScheduling::Immediate;
+                    }
                 } else {
                     branch.sync_state = SpeculativeState::Inconsistent;
+                }
+            }
             RunResult::BranchAtSyncEnd => {
                 let consistency = self.consensus.notify_of_finished_branch(branch_id);
                 if consistency == Consistency::Valid {
                     branch.sync_state = SpeculativeState::ReachedSyncEnd;
                     self.tree.push_into_queue(QueueKind::FinishedSync, branch_id);
                 } else if consistency == Consistency::Inconsistent {
                     branch.sync_state == SpeculativeState::Inconsistent;
+                }
             },
             RunResult::BranchPut(port_id, contents) => {
                 // Branch is attempting to send data
                 let port_id = PortIdLocal::new(port_id.0.u32_suffix);
                 let consistency = self.consensus.notify_of_speculative_mapping(branch_id, port_id, true);
                 if consistency == Consistency::Valid {
                     // `put()` is valid.
                     self.consensus.
                 } else {
                     branch.sync_state = SpeculativeState::Inconsistent;
+                }
             },
             _ => unreachable!("unexpected run result {:?} in sync mode", run_result),
+        }
         // If here then the run result did not require a particular action. We
         // return whether we have more active branches to run or not.
         if self.tree.queue_is_empty(QueueKind::Runnable) {
             return ConnectorScheduling::NotNow;
         } else {
             return ConnectorScheduling::Later;
+        }
+    }
+}
@@ \ No newline at end of file @@

src/runtime2/consensus.rs

➞

Show inline comments

 use crate::protocol::eval::ValueGroup;
 use crate::runtime2::branch::{BranchId, ExecTree, QueueKind};
 use crate::runtime2::ConnectorId;
 use crate::runtime2::inbox2::{DataHeader, SyncHeader};
 use crate::runtime2::port::PortIdLocal;
 use crate::runtime2::scheduler::ComponentCtxFancy;
 use super::inbox2::PortAnnotation;
 struct BranchAnnotation {
     port_mapping: Vec<PortAnnotation>,
+}
 /// 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
 pub(crate) struct Consensus {
     highest_connector_id: ConnectorId,
     branch_annotations: Vec<BranchAnnotation>,
+}
 #[derive(Clone, Copy, PartialEq, Eq)]
 pub(crate) enum Consistency {
     Valid,
     Inconsistent,
+}
 impl Consensus {
     pub fn new() -> Self {
         return Self {
             highest_connector_id: ConnectorId::new_invalid(),
             branch_annotations: Vec::new(),
+        }
+    }
     // --- Controlling sync round and branches
     /// 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, ports: &[PortIdLocal]) {
         debug_assert!(self.branch_annotations.is_empty());
         debug_assert!(!self.highest_connector_id.is_valid());
         // We'll use the first "branch" (the non-sync one) to store our ports,
         // this allows cloning if we created a new branch.
         self.branch_annotations.push(BranchAnnotation{
             port_mapping: ports.iter()
                 .map(|v| PortAnnotation{
                     port_id: *v,
                     registered_id: None,
                     expected_firing: None,
                 })
                 .collect(),
         });
+    }
     /// Notifies the consensus algorithm that a new branch has appeared. Must be
     /// called for each forked branch in the execution tree.
     pub fn notify_of_new_branch(&mut self, parent_branch_id: BranchId, new_branch_id: BranchId) {
         // If called correctly. Then each time we are notified the new branch's
         // index is the length in `branch_annotations`.
         debug_assert!(self.branch_annotations.len() == new_branch_id.index as usize);
         let parent_branch_annotations = &self.branch_annotations[parent_branch_id.index as usize];
         let new_branch_annotations = BranchAnnotation{
             port_mapping: parent_branch_annotations.port_mapping.clone(),
         };
         self.branch_annotations.push(new_branch_annotations);
+    }
     /// Notifies the consensus algorithm that a branch has reached the end of
     /// the sync block. A final check for consistency will be performed that the
     /// caller has to handle
     pub fn notify_of_finished_branch(&self, branch_id: BranchId) -> Consistency {
         let branch = &self.branch_annotations[branch_id.index as usize];
         for mapping in &branch.port_mapping {
             match mapping.expected_firing {
                 Some(expected) => {
                     if expected != mapping.registered_id.is_some() {
                         // Inconsistent speculative state and actual state
                         debug_assert!(mapping.registered_id.is_none()); // because if we did fire on a silent port, we should've caught that earlier
                         return Consistency::Inconsistent;
+                    }
                 },
                 None => {},
+            }
+        }
         return Consistency::Valid;
+    }
     /// Notifies the consensus algorithm that a particular branch has assumed
     /// a speculative value for its port mapping.
     pub fn notify_of_speculative_mapping(&mut self, branch_id: BranchId, port_id: PortIdLocal, does_fire: bool) -> Consistency {
         let branch = &mut self.branch_annotations[branch_id.index as usize];
         for mapping in &mut branch.port_mapping {
             if mapping.port_id == port_id {
                 match mapping.expected_firing {
                     None => {
                         // Not yet mapped, perform speculative mapping
                         mapping.expected_firing = Some(does_fire);
                         return Consistency::Valid;
                     },
                     Some(current) => {
                         // Already mapped
                         if current == does_fire {
                             return Consistency::Valid;
                         } else {
                             return Consistency::Inconsistent;
+                        }
+                    }
+                }
+            }
+        }
         unreachable!("notify_of_speculative_mapping called with unowned port");
+    }
     pub fn end_sync(&mut self, branch_id: BranchId, final_ports: &mut Vec<PortIdLocal>) {
         todo!("write");
+    }
     // --- Handling messages
     /// Prepares a message for sending. Caller should have made sure that
     /// sending the message is consistent with the speculative state.
     pub fn prepare_message(&mut self, branch_id: BranchId, source_port_id: PortIdLocal, value: &ValueGroup) -> (SyncHeader, DataHeader) {
         if cfg!(debug_assertions) {
             let branch = &self.branch_annotations[branch_id.index as usize];
             let port = branch.port_mapping.iter()
                 .find(|v| v.port_id == source_port_id)
                 .unwrap();
             debug_assert!(port.expected_firing == None || port.expected_firing == Some(true));
+        }
+    }
     pub fn handle_received_sync_header(&mut self, sync_header: &SyncHeader, ctx: &mut ComponentCtxFancy) {
         todo!("should check IDs and maybe send sync messages");
+    }
     /// Checks data header and consults the stored port mapping and the
     /// execution tree to see which branches may receive the data message's
     /// contents.
     ///
     /// This function is generally called for freshly received messages that
     /// should be matched against previously halted branches.
     pub fn handle_received_data_header(&mut self, exec_tree: &ExecTree, data_header: &DataHeader, target_ids: &mut Vec<BranchId>) {
         for branch in exec_tree.iter_queue(QueueKind::AwaitingMessage) {
             if branch.awaiting_port == data_header.target_port {
                 // Found a branch awaiting the message, but we need to make sure
                 // the mapping is correct
                 if self.branch_can_receive(branch.id, data_header) {
                     target_ids.push(branch.id);
+                }
+            }
+        }
+    }
     pub fn notify_of_received_message(&mut self, branch_id: BranchId, data_header: &DataHeader) {
         debug_assert!(self.branch_can_receive(branch_id, data_header));
         let branch = &mut self.branch_annotations[branch_id.index as usize];
         for mapping in &mut branch.port_mapping {
             if mapping.port_id == data_header.target_port {
                 mapping.registered_id = Some(data_header.new_mapping);
                 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(crate) fn branch_can_receive(&self, branch_id: BranchId, data_header: &DataHeader) -> bool {
         let annotation = &self.branch_annotations[branch_id.index as usize];
         for expected in &data_header.expected_mapping {
             // If we own the port, then we have an entry in the
             // annotation, check if the current mapping matches
             for current in &annotation.port_mapping {
                 if expected.port_id == current.port_id {
                     if expected.registered_id != current.registered_id {
                         // IDs do not match, we cannot receive the
                         // message in this branch
                         return false;
+                    }
+                }
+            }
+        }
         return true;
+    }
+}
@@ \ No newline at end of file @@

src/runtime2/inbox2.rs

➞

Show inline comments

 use crate::protocol::eval::ValueGroup;
 use crate::runtime2::branch::BranchId;
 use crate::runtime2::ConnectorId;
 use crate::runtime2::port::PortIdLocal;
 #[derive(Copy, Clone)]
 pub(crate) struct PortAnnotation {
     pub port_id: PortIdLocal,
     pub registered_id: Option<BranchId>,
     pub expected_firing: Option<bool>,
+}
 /// The header added by the synchronization algorithm to all.
 pub(crate) struct SyncHeader {
     pub sending_component_id: ConnectorId,
     pub highest_component_id: ConnectorId,
+}
 /// The header added to data messages
 pub(crate) struct DataHeader {
     pub expected_mapping: Vec<PortAnnotation>,
     pub target_port: PortIdLocal,
     pub new_mapping: BranchId,
+}
 /// A data message is a message that is intended for the receiver's PDL code,
 /// but will also be handled by the consensus algrorithm
 pub(crate) struct DataMessageFancy {
     pub sync_header: SyncHeader,
     pub data_header: DataHeader,
     pub content: ValueGroup,
+}
 pub(crate) enum SyncContent {
+}
 /// A sync message is a message that is intended only for the consensus
 /// algorithm.
 pub(crate) struct SyncMessageFancy {
     pub sync_header: SyncHeader,
     pub content: SyncContent,
+}
 /// A control message is a message intended for the scheduler that is executing
 /// a component.
 pub(crate) struct ControlMessageFancy {
     pub id: u32, // generic identifier, used to match request to response
     pub content: ControlContent,
+}
 pub(crate) enum ControlContent {
     PortPeerChanged(PortIdLocal, ConnectorId),
     CloseChannel(PortIdLocal),
     Ack,
     Ping,
+}
 /// Combination of data message and control messages.
 pub(crate) enum MessageFancy {
     Data(DataMessageFancy),
     Sync(SyncMessageFancy),
     Control(ControlMessageFancy),
+}
@@ \ No newline at end of file @@

src/runtime2/mod.rs

➞

Show inline comments

 // Structure of module
 mod runtime;
 mod messages;
 mod connector;
 mod branch;
 mod native;
 mod port;
 mod scheduler;
 mod inbox;
 mod consensus;
 mod inbox2;
 #[cfg(test)] mod tests;
 mod connector2;
 // Imports
 use std::collections::VecDeque;
 use std::sync::{Arc, Condvar, Mutex, RwLock};
 use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
 use std::thread::{self, JoinHandle};
 use crate::collections::RawVec;
 use crate::ProtocolDescription;
 use inbox::Message;
 use connector::{ConnectorPDL, ConnectorPublic, ConnectorScheduling};
 use scheduler::{Scheduler, ControlMessageHandler};
 use native::{Connector, ConnectorApplication, ApplicationInterface};
 use crate::runtime2::port::{Port, PortState};
 use crate::runtime2::scheduler::{ComponentCtxFancy, SchedulerCtx};
 /// A kind of token that, once obtained, allows mutable access to a connector.
 /// We're trying to use move semantics as much as possible: the owner of this
 /// key is the only one that may execute the connector's code.
 pub(crate) struct ConnectorKey {
     pub index: u32, // of connector
+}
 impl ConnectorKey {
     /// Downcasts the `ConnectorKey` type, which can be used to obtain mutable
     /// access, to a "regular ID" which can be used to obtain immutable access.
     #[inline]
     pub fn downcast(&self) -> ConnectorId {
         return ConnectorId(self.index);
+    }
     /// Turns the `ConnectorId` into a `ConnectorKey`, marked as unsafe as it
     /// bypasses the type-enforced `ConnectorKey`/`ConnectorId` system
     #[inline]
     pub unsafe fn from_id(id: ConnectorId) -> ConnectorKey {
         return ConnectorKey{ index: id.0 };
+    }
+}
 /// A kind of token that allows shared access to a connector. Multiple threads
 /// may hold this
 #[derive(Debug, Copy, Clone, PartialEq, Eq)]
 pub struct ConnectorId(pub u32);
 impl ConnectorId {
     // TODO: Like the other `new_invalid`, maybe remove
     #[inline]
     pub fn new_invalid() -> ConnectorId {
         return ConnectorId(u32::MAX);
+    }
     #[inline]
     pub(crate) fn is_valid(&self) -> bool {
         return self.0 != u32::MAX;
+    }
+}
 // TODO: Change this, I hate this. But I also don't want to put `public` and
 //  `router` of `ScheduledConnector` back into `Connector`. The reason I don't
 //  want `Box<dyn Connector>` everywhere is because of the v-table overhead. But
 //  to truly design this properly I need some benchmarks.
 pub(crate) enum ConnectorVariant {
     UserDefined(ConnectorPDL),
     Native(Box<dyn Connector>),
+}
 impl Connector for ConnectorVariant {
     fn run(&mut self, scheduler_ctx: SchedulerCtx, comp_ctx: &mut ComponentCtxFancy) -> ConnectorScheduling {
         match self {
             ConnectorVariant::UserDefined(c) => c.run(scheduler_ctx, comp_ctx),
             ConnectorVariant::Native(c) => c.run(scheduler_ctx, comp_ctx),
+        }
+    }
+}
 pub(crate) struct ScheduledConnector {
     pub connector: ConnectorVariant, // access by connector
     pub ctx_fancy: ComponentCtxFancy,
     pub public: ConnectorPublic, // accessible by all schedulers and connectors
     pub router: ControlMessageHandler,
     pub shutting_down: bool,
+}
 // -----------------------------------------------------------------------------
 // Runtime
 // -----------------------------------------------------------------------------
 /// Externally facing runtime.
 pub struct Runtime {
     inner: Arc<RuntimeInner>,
+}
 impl Runtime {
     pub fn new(num_threads: u32, protocol_description: ProtocolDescription) -> Runtime {
         // Setup global state
         assert!(num_threads > 0, "need a thread to run connectors");
         let runtime_inner = Arc::new(RuntimeInner{
             protocol_description,
             port_counter: AtomicU32::new(0),
             connectors: RwLock::new(ConnectorStore::with_capacity(32)),
             connector_queue: Mutex::new(VecDeque::with_capacity(32)),
             schedulers: Mutex::new(Vec::new()),
             scheduler_notifier: Condvar::new(),
             active_connectors: AtomicU32::new(0),
             active_interfaces: AtomicU32::new(1), // this `Runtime` instance
             should_exit: AtomicBool::new(false),
         });
         // Launch threads
+        {
             let mut schedulers = Vec::with_capacity(num_threads as usize);
             for thread_index in 0..num_threads {
                 let cloned_runtime_inner = runtime_inner.clone();
                 let thread = thread::Builder::new()
                     .name(format!("thread-{}", thread_index))
                     .spawn(move || {
                         let mut scheduler = Scheduler::new(cloned_runtime_inner, thread_index);
                         scheduler.run();
                     })
                     .unwrap();
                 schedulers.push(thread);
+            }
             let mut lock = runtime_inner.schedulers.lock().unwrap();
             *lock = schedulers;
+        }
         // Return runtime
         return Runtime{ inner: runtime_inner };
+    }
     /// Returns a new interface through which channels and connectors can be
     /// created.
     pub fn create_interface(&self) -> ApplicationInterface {
         self.inner.increment_active_interfaces();
         let (connector, mut interface) = ConnectorApplication::new(self.inner.clone());
         let connector_key = self.inner.create_interface_component(connector);
         interface.set_connector_id(connector_key.downcast());
         // Note that we're not scheduling. That is done by the interface in case
         // it is actually needed.
         return interface;
+    }
+}
 impl Drop for Runtime {
     fn drop(&mut self) {
         self.inner.decrement_active_interfaces();
         let mut lock = self.inner.schedulers.lock().unwrap();
         for handle in lock.drain(..) {
             handle.join().unwrap();
+        }
+    }
+}
 // -----------------------------------------------------------------------------
 // RuntimeInner
 // -----------------------------------------------------------------------------
 pub(crate) struct RuntimeInner {
     // Protocol
     pub(crate) protocol_description: ProtocolDescription,
     // Regular counter for port IDs
     port_counter: AtomicU32,
     // Storage of connectors and the work queue
     connectors: RwLock<ConnectorStore>,
     connector_queue: Mutex<VecDeque<ConnectorKey>>,
     schedulers: Mutex<Vec<JoinHandle<()>>>,
     // Conditions to determine whether the runtime can exit
     scheduler_notifier: Condvar,  // coupled to mutex on `connector_queue`.
     // TODO: Figure out if we can simply merge the counters?
     active_connectors: AtomicU32, // active connectors (if sleeping, then still considered active)
     active_interfaces: AtomicU32, // active API interfaces that can add connectors/channels
     should_exit: AtomicBool,
+}
 impl RuntimeInner {
     // --- Managing the components queued for execution
     /// Wait until there is a connector to run. If there is one, then `Some`
     /// will be returned. If there is no more work, then `None` will be
     /// returned.
     pub(crate) fn wait_for_work(&self) -> Option<ConnectorKey> {
         let mut lock = self.connector_queue.lock().unwrap();
         while lock.is_empty() && !self.should_exit.load(Ordering::Acquire) {
             lock = self.scheduler_notifier.wait(lock).unwrap();
+        }
         return lock.pop_front();

src/runtime2/scheduler.rs

➞

Show inline comments

 use std::collections::VecDeque;
 use std::sync::Arc;
 use std::sync::atomic::Ordering;
 use crate::runtime2::inbox2::{DataMessageFancy, MessageFancy};
 use super::{ScheduledConnector, RuntimeInner, ConnectorId, ConnectorKey, ConnectorVariant};
 use super::port::{Port, PortState, PortIdLocal};
 use super::native::Connector;
 use super::connector::{BranchId, ConnectorPDL, ConnectorScheduling};
 use super::inbox::{
     Message, MessageContents, ControlMessageVariant,
     DataMessage, ControlMessage, SolutionMessage, SyncMessage
 };
 // 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 cur_schedule == ConnectorScheduling::Immediate {
                 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()));
                     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_fancy);
                     self.debug_conn(connector_id, "Finished running");
                     // 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_fancy.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);
+                        }
+                    }
                     if scheduled.router.num_pending_acks() == 0 {
                         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_fancy.id;
         while let Some(message) = scheduled.public.inbox.take_message() {
             // Check for rerouting
             self.debug_conn(connector_id, &format!("Handling message from conn({}) at port({})\n --- {:?}", message.sending_connector.0, message.receiving_port.index, message));
             if let Some(other_connector_id) = scheduled.router.should_reroute(message.sending_connector, message.receiving_port) {
                 self.debug_conn(connector_id, &format!(" ... Rerouting to connector {}", other_connector_id.0));
                 self.runtime.send_message(other_connector_id, message);
                 continue;
+            }
             // Handle special messages here, messages for the component
             // will be added to the inbox.
             self.debug_conn(connector_id, " ... Handling message myself");
             match message.contents {
                 MessageContents::Control(content) => {
                     match content.content {
                         ControlMessageVariant::ChangePortPeer(port_id, new_target_connector_id) => {
                             // Need to change port target
                             let port = scheduled.ctx_fancy.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_fancy.outbox.is_empty());
                             // And respond with an Ack
                             let ack_message = Message{
                                 sending_connector: connector_id,
                                 receiving_port: PortIdLocal::new_invalid(),
                                 contents: MessageContents::Control(ControlMessage{
                                     id: content.id,
                                     content: ControlMessageVariant::Ack,
                                 }),
                             };
                             self.debug_conn(connector_id, &format!("Sending message [pp ack]\n --- {:?}", ack_message));
                             self.runtime.send_message(message.sending_connector, ack_message);
                         },
                         ControlMessageVariant::CloseChannel(port_id) => {
                             // Mark the port as being closed
                             let port = scheduled.ctx_fancy.get_port_mut_by_id(port_id).unwrap();
                             port.state = PortState::Closed;
                             // Send an Ack
                             let ack_message = Message{
                                 sending_connector: connector_id,
                                 receiving_port: PortIdLocal::new_invalid(),
                                 contents: MessageContents::Control(ControlMessage{
                                     id: content.id,
                                     content: ControlMessageVariant::Ack,
                                 }),
                             };
                             self.debug_conn(connector_id, &format!("Sending message [cc ack] \n --- {:?}", ack_message));
                             self.runtime.send_message(message.sending_connector, ack_message);
                         },
                         ControlMessageVariant::Ack => {
                             scheduled.router.handle_ack(content.id);
+                        }
+                    }
                 },
                 MessageContents::Ping => {
                     // Pings are sent just to wake up a component, so
                     // nothing to do here.
                 },
                 _ => {
                     // All other cases have to be handled by the component
@@ @@ -216,491 +217,490 @@ impl Scheduler { @@
                     let port = scheduled.ctx_fancy.get_port_by_id(contents.sending_port).unwrap();
                     (port.peer_connector, contents.sending_port, port.peer_id)
                 },
                 MessageContents::Sync(contents) => {
                     let connector = contents.to_visit.pop().unwrap();
                     (connector, PortIdLocal::new_invalid(), PortIdLocal::new_invalid())
                 },
                 MessageContents::RequestCommit(contents)=> {
                     let connector = contents.to_visit.pop().unwrap();
                     (connector, PortIdLocal::new_invalid(), PortIdLocal::new_invalid())
                 },
                 MessageContents::ConfirmCommit(contents) => {
                     for to_visit in &contents.to_visit {
                         let message = Message{
                             sending_connector: scheduled.ctx_fancy.id,
                             receiving_port: PortIdLocal::new_invalid(),
                             contents: MessageContents::ConfirmCommit(contents.clone()),
                         };
                         self.runtime.send_message(*to_visit, message);
+                    }
                     (ConnectorId::new_invalid(), PortIdLocal::new_invalid(), PortIdLocal::new_invalid())
                 },
                 MessageContents::Control(_) | MessageContents::Ping => {
                     // Never generated by the user's code
                     unreachable!();
+                }
             };
             // TODO: Maybe clean this up, perhaps special case for
             //  ConfirmCommit can be handled differently.
             if peer_connector.is_valid() {
                 if peer_port.is_valid() {
                     // Sending a message to a port, so the port may not be
                     // closed.
                     let port = scheduled.ctx_fancy.get_port_by_id(self_port).unwrap();
                     match port.state {
                         PortState::Open => {},
                         PortState::Closed => {
                             todo!("Handling sending over a closed port");
+                        }
+                    }
+                }
                 let message = Message {
                     sending_connector: scheduled.ctx_fancy.id,
                     receiving_port: peer_port,
                     contents: message,
                 };
                 self.runtime.send_message(peer_connector, message);
+            }
+        }
         while let Some(state_change) = scheduled.ctx_fancy.state_changes.pop_front() {
             match state_change {
                 ComponentStateChange::CreatedComponent(component) => {
                     // Add the new connector to the global registry
                     let new_key = self.runtime.create_pdl_component(component, false);
                     let new_connector = self.runtime.get_component_private(&new_key);
                     // Transfer ports
                     // TODO: Clean this up the moment native components are somewhat
                     //  properly implemented. We need to know about the ports that
                     //  are "owned by the PDL code", and then make sure that the
                     //  context contains a description of those ports.
                     let ports = if let ConnectorVariant::UserDefined(connector) = &new_connector.connector {
                         &connector.ports.owned_ports
                     } else {
                         unreachable!();
                     };
                     for port_id in ports {
                         // Transfer messages associated with the transferred port
                         let mut message_idx = 0;
                         while message_idx < scheduled.ctx_fancy.inbox_messages.len() {
                             let message = &scheduled.ctx_fancy.inbox_messages[message_idx];
                             if message.receiving_port == *port_id {
                                 // Need to transfer this message
                                 let taken_message = scheduled.ctx_fancy.inbox_messages.remove(message_idx);
                                 new_connector.ctx_fancy.inbox_messages.push(taken_message);
                             } else {
                                 message_idx += 1;
+                            }
+                        }
                         // Transfer the port itself
                         let port_index = scheduled.ctx_fancy.ports.iter()
                             .position(|v| v.self_id == *port_id)
                             .unwrap();
                         let port = scheduled.ctx_fancy.ports.remove(port_index);
                         new_connector.ctx_fancy.ports.push(port.clone());
                         // Notify the peer that the port has changed
                         let reroute_message = scheduled.router.prepare_reroute(
                             port.self_id, port.peer_id, scheduled.ctx_fancy.id,
                             port.peer_connector, new_connector.ctx_fancy.id
                         );
                         self.debug_conn(connector_id, &format!("Sending message [newcon]\n --- {:?}", reroute_message));
                         self.runtime.send_message(port.peer_connector, reroute_message);
+                    }
                     // Schedule new connector to run
                     self.runtime.push_work(new_key);
                 },
                 ComponentStateChange::CreatedPort(port) => {
                     scheduled.ctx_fancy.ports.push(port);
                 },
                 ComponentStateChange::ChangedPort(port_change) => {
                     if port_change.is_acquired {
                         scheduled.ctx_fancy.ports.push(port_change.port);
                     } else {
                         let index = scheduled.ctx_fancy.ports
                             .iter()
                             .position(|v| v.self_id == port_change.port.self_id)
                             .unwrap();
                         scheduled.ctx_fancy.ports.remove(index);
+                    }
+                }
+            }
+        }
         // Finally, check if we just entered or just left a sync region
         if scheduled.ctx_fancy.changed_in_sync {
             if scheduled.ctx_fancy.is_in_sync {
                 // Just entered sync region
             } else {
                 // Just left sync region. So clear inbox
                 scheduled.ctx_fancy.inbox_messages.clear();
                 scheduled.ctx_fancy.inbox_len_read = 0;
+            }
             scheduled.ctx_fancy.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_fancy.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)
+            }
+        }
+    }
     // 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),
     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 ComponentCtxFancy {
     // Mostly managed by the scheduler
     pub(crate) id: ConnectorId,
     ports: Vec<Port>,
-    inbox_messages: Vec<Message>, // never control or ping messages
+    inbox_messages: Vec<MessageFancy>, // never control or ping messages
     inbox_len_read: usize,
     // Submitted by the component
     is_in_sync: bool,
     changed_in_sync: bool,
     outbox: VecDeque<MessageContents>,
     state_changes: VecDeque<ComponentStateChange>
+}
 pub(crate) enum ReceivedMessage {
     Data((PortIdLocal, DataMessage)),
     Sync(SyncMessage),
     RequestCommit(SolutionMessage),
     ConfirmCommit(SolutionMessage),
+}
 impl ComponentCtxFancy {
     pub(crate) fn new_empty() -> Self {
         return Self{
             id: ConnectorId::new_invalid(),
             ports: Vec::new(),
             inbox_messages: Vec::new(),
             inbox_len_read: 0,
             is_in_sync: false,
             changed_in_sync: false,
             outbox: VecDeque::new(),
             state_changes: VecDeque::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) {
         debug_assert!(!self.is_in_sync);
         self.state_changes.push_back(ComponentStateChange::CreatedComponent(component));
+    }
     /// 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))
+    }
     pub(crate) fn get_port_by_id(&self, id: PortIdLocal) -> Option<&Port> {
         return self.ports.iter().find(|v| v.self_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 `Component::run` function with an
     /// appropriate scheduling value.
     pub(crate) fn notify_sync_start(&mut self) -> &[Port] {
         debug_assert!(!self.is_in_sync);
         self.is_in_sync = true;
         self.changed_in_sync = true;
         return &self.ports
+    }
     #[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: MessageContents) {
         debug_assert!(self.is_in_sync);
         self.outbox.push_back(contents);
+    }
     /// 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, match_prev_branch_id: BranchId) -> MessagesIter {
     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, match_prev_branch_id
             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<ReceivedMessage> {
+    pub(crate) fn read_next_message(&mut self) -> Option<MessageFancy> {
         if !self.is_in_sync { return None; }
         if self.inbox_len_read == self.inbox_messages.len() { return None; }
         let message = &self.inbox_messages[self.inbox_len_read];
         if let MessageContents::Data(contents) = &message.contents {
             self.inbox_len_read += 1;
             return Some(ReceivedMessage::Data((message.receiving_port, contents.clone())));
         } else {
             // Must be a sync/solution message
             let message = self.inbox_messages.remove(self.inbox_len_read);
             return match message.contents {
                 MessageContents::Sync(v) => Some(ReceivedMessage::Sync(v)),
                 MessageContents::RequestCommit(v) => Some(ReceivedMessage::RequestCommit(v)),
                 MessageContents::ConfirmCommit(v) => Some(ReceivedMessage::ConfirmCommit(v)),
                 _ => unreachable!(), // because we only put data/synclike messages in the inbox
+            }
+        }
+    }
+}
 pub(crate) struct MessagesIter<'a> {
-    messages: &'a [Message],
+    messages: &'a [MessageFancy],
     next_index: usize,
     max_index: usize,
     match_port_id: PortIdLocal,
     match_prev_branch_id: BranchId,
+}
 impl<'a> Iterator for MessagesIter<'a> {
-    type Item = &'a DataMessage;
+    type Item = &'a DataMessageFancy;
     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 MessageContents::Data(data_message) = &message.contents {
                 if message.receiving_port == self.match_port_id && data_message.sender_prev_branch_id == self.match_prev_branch_id {
             if let MessageFancy::Data(message) = &message {
                 if message.data_header.target_port == self.match_port_id {
                     // Found a match
                     self.next_index += 1;
                     return Some(data_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;
+    }
+}
 // -----------------------------------------------------------------------------
 // Control messages
 // -----------------------------------------------------------------------------
 struct ControlEntry {
     id: u32,
     variant: ControlVariant,
+}
 enum ControlVariant {
     ChangedPort(ControlChangedPort),
     ClosedChannel(ControlClosedChannel),
+}
 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
+}
 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
     ) -> Message {
         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 Message{
             sending_connector: self_connector_id,
             receiving_port: peer_port_id,
             contents: MessageContents::Control(ControlMessage{
                 id,
                 content: ControlMessageVariant::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
     ) -> Message {
         let id = self.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,
             }),
         });
         return Message{
             sending_connector: self_connector_id,
             receiving_port: peer_port_id,
             contents: MessageContents::Control(ControlMessage{
                 id,
                 content: ControlMessageVariant::ChangePortPeer(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, sending_connector: ConnectorId, 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) {
         let index = self.active.iter()
             .position(|v| v.id == id);
         match index {
             Some(index) => { self.active.remove(index); },
             None => { todo!("handling of nefarious ACKs"); },
+        }
+    }
     /// 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 @@

0 comments (0 inline, 0 general)