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

Changeset - 32d91577e090

Parent rev.

Child rev.

[Not reviewed]

0 10 0

MH - 4 years ago 2021-11-09 17:39:28
contact@maxhenger.nl

initial multithreaded runtime

10 files changed with 125 insertions and 44 deletions:

src/collections/raw_vec.rs

src/collections/sets.rs

src/runtime2/branch.rs

src/runtime2/connector2.rs

src/runtime2/consensus.rs

src/runtime2/inbox2.rs

src/runtime2/mod.rs

src/runtime2/native.rs

src/runtime2/scheduler.rs

src/runtime2/tests/mod.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.
 pub struct RawVec<T: Sized> {
     base: *mut T,
     cap: usize,
     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;
+    }
     pub unsafe fn get(&self, idx: usize) -> *const T {
         debug_assert!(idx < self.len);
         return self.base.add(idx);
+    }
     pub unsafe fn get_mut(&self, idx: usize) -> *mut T {
         debug_assert!(idx < self.len);
         return self.base.add(idx);
+    }
     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;
+        }
+    }
     pub fn len(&self) -> usize {
         return 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(new_base, old_base, old_size);
+                    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/collections/sets.rs

➞

Show inline comments

 #![allow(dead_code)] // For now, come back to this when compiler has matured -- MH 27/05/21
 use std::collections::VecDeque;
 /// Simple double ended queue that ensures that all elements are unique. Queue
 /// elements are not ordered (we expect the queue to be rather small).
 pub struct DequeSet<T: Eq> {
     inner: VecDeque<T>,
+}
 impl<T: Eq> DequeSet<T> {
     pub fn new() -> Self {
         Self{ inner: VecDeque::new() }
+    }
     #[inline]
     pub fn pop_front(&mut self) -> Option<T> {
         self.inner.pop_front()
+    }
     #[inline]
     pub fn pop_back(&mut self) -> Option<T> {
         self.inner.pop_back()
+    }
     #[inline]
     pub fn push_back(&mut self, to_push: T) {
         for element in self.inner.iter() {
             if *element == to_push {
                 return;
+            }
+        }
         self.inner.push_back(to_push);
+    }
     #[inline]
     pub fn push_front(&mut self, to_push: T) {
         for element in self.inner.iter() {
             if *element == to_push {
                 return;
+            }
+        }
         self.inner.push_front(to_push);
+    }
     #[inline]
     pub fn clear(&mut self) {
         self.inner.clear();
+    }
     #[inline]
     pub fn is_empty(&self) -> bool {
         self.inner.is_empty()
+    }
+}
 /// Simple vector set that ensures that all elements are unique. Elements are
 /// not ordered (we expect the vector to be small).
 pub struct VecSet<T: Eq> {
     inner: Vec<T>,
+}
 impl<T: Eq> VecSet<T> {
     pub fn new() -> Self {
         Self{ inner: Vec::new() }
+    }
     #[inline]
     pub fn pop(&mut self) -> Option<T> {
         self.inner.pop()
+    }
     #[inline]
     pub fn push(&mut self, to_push: T) {
         for element in self.inner.iter() {
             if *element == to_push {
                 return;
+            }
+        }
         self.inner.push(to_push);
+    }
     #[inline]
     pub fn clear(&mut self) {
         self.inner.clear();
+    }
     #[inline]
     pub fn iter(&self) -> impl Iterator<Item=&T> {
         return self.inner.iter();
+    }
     #[inline]
     pub fn contains(&self, item: &T) -> bool {
         return self.inner.contains(item);
+    }
     #[inline]
     pub fn is_empty(&self) -> bool {
         self.inner.is_empty()
+    }
     #[inline]
     pub fn into_vec(self) -> Vec<T> {
         return self.inner;
+    }
+}
@@ \ 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::{Value, 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]
     pub(crate) 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
     pub prepared_channel: Option<(Value, Value)>, // TODO: Maybe remove?
+}
 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(),
             prepared_channel: None,
+        }
+    }
     /// 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_id.is_valid()) ||
             (parent_branch.sync_state == SpeculativeState::HaltedAtBranchPoint)
         ); // forking from non-sync, or forking from a branching point
         debug_assert!(parent_branch.prepared_channel.is_none());
         Branch {
             id: BranchId::new(new_index),
             parent_id: parent_branch.id,
             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(),
             prepared_channel: None,
+        }
+    }
     /// 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;
 #[derive(Debug, PartialEq, Eq)]
 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>,
     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> {
         debug_assert_ne!(kind, QueueKind::FinishedSync); // for purposes of logic we expect the queue to grow during a sync round
         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;
+        }
+    }
     /// Returns the non-sync branch (TODO: better name?)
     pub fn base_branch_mut(&mut self) -> &mut Branch {
         debug_assert!(!self.is_in_sync());
         return &mut self.branches[0];
+    }
     /// Returns an iterator over all the elements in the queue of the given
     /// kind. One can start the iteration at the branch *after* the provided
     /// branch. Just make sure it actually is in the provided queue.
     pub fn iter_queue(&self, kind: QueueKind, start_at: Option<BranchId>) -> BranchQueueIter {
         let queue = &self.queues[kind.as_index()];
         let index = match start_at {
             Some(branch_id) => {
                 debug_assert!(self.iter_queue(kind, None).any(|v| v.id == branch_id));
                 let branch = &self.branches[branch_id.index as usize];
                 branch.next_in_queue.index as usize
             },
             None => {
                 queue.first.index as usize
+            }
         };
         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. The id of this root sync branch is
     /// returned.
     pub fn start_sync(&mut self) -> BranchId {
         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);
         return 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) -> 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 = Branch::new_sync(self.branches.len() as u32, parent_branch);
         let new_branch_id = new_branch.id;
         self.branches.push(new_branch);
         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, None).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;
         debug_assert!(!branch.awaiting_port.is_valid());
         branch.next_in_queue = BranchId::new_invalid();
         branch.inbox.clear();
         debug_assert!(branch.prepared_channel.is_none());
         // 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(crate) struct BranchQueueIter<'a> {
     branches: &'a [Branch],
     index: usize,
+}
 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(crate) 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

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

src/runtime2/consensus.rs

➞

Show inline comments

 use crate::collections::VecSet;
 use crate::protocol::eval::ValueGroup;
 use crate::runtime2::inbox2::DataContent;
 use super::branch::{BranchId, ExecTree, QueueKind};
 use super::ConnectorId;
 use super::port::{ChannelId, Port, PortIdLocal};
 use super::inbox2::{
     DataHeader, DataMessageFancy, MessageFancy,
     SyncContent, SyncHeader, SyncMessageFancy, PortAnnotation
 };
 use super::scheduler::ComponentCtxFancy;
 struct BranchAnnotation {
     port_mapping: Vec<PortAnnotation>,
+}
 #[derive(Debug)]
 pub(crate) struct LocalSolution {
     component: ConnectorId,
     final_branch_id: BranchId,
     port_mapping: Vec<(ChannelId, BranchId)>,
+}
 #[derive(Debug, Clone)]
 pub(crate) struct GlobalSolution {
     component_branches: Vec<(ConnectorId, BranchId)>,
     channel_mapping: Vec<(ChannelId, BranchId)>, // TODO: This can go, is debugging info
+}
 // -----------------------------------------------------------------------------
 // Consensus
 // -----------------------------------------------------------------------------
 /// The consensus algorithm. Currently only implemented to find the component
 /// with the highest ID within the sync region and letting it handle all the
 /// local solutions.
 ///
 /// The type itself serves as an experiment to see how code should be organized.
 // TODO: Flatten all datastructures
 // TODO: Have a "branch+port position hint" in case multiple operations are
 //  performed on the same port to prevent repeated lookups
 // TODO: A lot of stuff should be batched. Like checking all the sync headers
 //  and sending "I have a higher ID" messages.
 pub(crate) struct Consensus {
     // --- State that is cleared after each round
     // Local component's state
     highest_connector_id: ConnectorId,
     branch_annotations: Vec<BranchAnnotation>,
     last_finished_handled: Option<BranchId>,
     // Gathered state (in case we are currently the leader of the distributed
     // consensus protocol)
     encountered_peers: VecSet<ConnectorId>,
     // Gathered state from communication
     encountered_peers: VecSet<ConnectorId>, // to determine when we should send "found a higher ID" messages.
     encountered_ports: VecSet<PortIdLocal>, // to determine if we should send "port remains silent" messages.
     solution_combiner: SolutionCombiner,
     // Workspaces
     // --- Persistent state
     // TODO: Tracking sync round numbers
     // --- Workspaces
     workspace_ports: Vec<PortIdLocal>,
+}
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub(crate) enum Consistency {
     Valid,
     Inconsistent,
+}
 impl Consensus {
     pub fn new() -> Self {
         return Self {
             highest_connector_id: ConnectorId::new_invalid(),
             branch_annotations: Vec::new(),
             last_finished_handled: None,
             encountered_peers: VecSet::new(),
             encountered_ports: VecSet::new(),
             solution_combiner: SolutionCombiner::new(),
             workspace_ports: Vec::new(),
+        }
+    }
     // --- Controlling sync round and branches
     /// Returns whether the consensus algorithm is running in sync mode
     pub fn is_in_sync(&self) -> bool {
         return !self.branch_annotations.is_empty();
+    }
     /// TODO: Remove this once multi-fire is in place
     pub fn get_annotation(&self, branch_id: BranchId, port_id: PortIdLocal) -> &PortAnnotation {
         let branch = &self.branch_annotations[branch_id.index as usize];
         let port = branch.port_mapping.iter().find(|v| v.port_id == port_id).unwrap();
         return port;
+    }
     /// Sets up the consensus algorithm for a new synchronous round. The
     /// provided ports should be the ports the component owns at the start of
     /// the sync round.
     pub fn start_sync(&mut self, ctx: &ComponentCtxFancy) {
         debug_assert!(!self.highest_connector_id.is_valid());
         debug_assert!(self.branch_annotations.is_empty());
         debug_assert!(self.last_finished_handled.is_none());
         debug_assert!(self.encountered_peers.is_empty());
         debug_assert!(self.solution_combiner.local.is_empty());
         // We'll use the first "branch" (the non-sync one) to store our ports,
         // this allows cloning if we created a new branch.
         self.branch_annotations.push(BranchAnnotation{
             port_mapping: ctx.get_ports().iter()
                 .map(|v| PortAnnotation{
                     port_id: v.self_id,
                     registered_id: None,
                     expected_firing: None,
                 })
                 .collect(),
         });
         self.highest_connector_id = ctx.id;
+    }
     /// Notifies the consensus algorithm that a new branch has appeared. Must be
     /// called for each forked branch in the execution tree.
     pub fn notify_of_new_branch(&mut self, parent_branch_id: BranchId, new_branch_id: BranchId) {
         // If called correctly. Then each time we are notified the new branch's
         // index is the length in `branch_annotations`.
         debug_assert!(self.branch_annotations.len() == new_branch_id.index as usize);
         let parent_branch_annotations = &self.branch_annotations[parent_branch_id.index as usize];
         let new_branch_annotations = BranchAnnotation{
             port_mapping: parent_branch_annotations.port_mapping.clone(),
         };
         self.branch_annotations.push(new_branch_annotations);
+    }
     /// Notifies the consensus algorithm that a branch has reached the end of
     /// the sync block. A final check for consistency will be performed that the
     /// caller has to handle. Note that
     pub fn notify_of_finished_branch(&self, branch_id: BranchId) -> Consistency {
         debug_assert!(self.is_in_sync());
         let branch = &self.branch_annotations[branch_id.index as usize];
         for mapping in &branch.port_mapping {
             match mapping.expected_firing {
                 Some(expected) => {
                     if expected != mapping.registered_id.is_some() {
                         // Inconsistent speculative state and actual state
                         debug_assert!(mapping.registered_id.is_none()); // because if we did fire on a silent port, we should've caught that earlier
                         return Consistency::Inconsistent;
+                    }
                 },
                 None => {},
+            }
+        }
         return Consistency::Valid;
+    }
     /// Notifies the consensus algorithm that a particular branch has assumed
     /// a speculative value for its port mapping.
     pub fn notify_of_speculative_mapping(&mut self, branch_id: BranchId, port_id: PortIdLocal, does_fire: bool) -> Consistency {
         debug_assert!(self.is_in_sync());
         let branch = &mut self.branch_annotations[branch_id.index as usize];
         for mapping in &mut branch.port_mapping {
             if mapping.port_id == port_id {
                 match mapping.expected_firing {
                     None => {
                         // Not yet mapped, perform speculative mapping
                         mapping.expected_firing = Some(does_fire);
                         return Consistency::Valid;
                     },
                     Some(current) => {
                         // Already mapped
                         if current == does_fire {
                             return Consistency::Valid;
                         } else {
                             return Consistency::Inconsistent;
+                        }
+                    }
+                }
+            }
+        }
         unreachable!("notify_of_speculative_mapping called with unowned port");
+    }
     /// Generates sync messages for any branches that are at the end of the
     /// sync block. To find these branches, they should've been put in the
     /// "finished" queue in the execution tree.
     pub fn handle_new_finished_sync_branches(&mut self, tree: &ExecTree, ctx: &mut ComponentCtxFancy) -> Option<BranchId> {
         debug_assert!(self.is_in_sync());
         let mut last_branch_id = self.last_finished_handled;
         for branch in tree.iter_queue(QueueKind::FinishedSync, last_branch_id) {
             // Turn the port mapping into a local solution
             let source_mapping = &self.branch_annotations[branch.id.index as usize].port_mapping;
             let mut target_mapping = Vec::with_capacity(source_mapping.len());
             for port in source_mapping {
                 // Note: if the port is silent, and we've never communicated
                 // over the port, then we need to do so now, to let the peer
                 // component know about our sync leader state.
                 let port_desc = ctx.get_port_by_id(port.port_id).unwrap();
                 let peer_port_id = port_desc.peer_id;
                 let channel_id = port_desc.channel_id;
                 if !self.encountered_ports.contains(&port.port_id) {
                     ctx.submit_message(MessageFancy::Data(DataMessageFancy{
                         sync_header: SyncHeader{
                             sending_component_id: ctx.id,
                             highest_component_id: self.highest_connector_id,
                         },
                         data_header: DataHeader{
                             expected_mapping: source_mapping.clone(),
                             sending_port: port.port_id,
                             target_port: peer_port_id,
                             new_mapping: BranchId::new_invalid(),
                         },
                         content: DataContent::SilentPortNotification,
                     }));
                     self.encountered_ports.push(port.port_id);
+                }
                 target_mapping.push((
-                    port_desc.channel_id,
                     channel_id,
                     port.registered_id.unwrap_or(BranchId::new_invalid())
                 ));
+            }
             let local_solution = LocalSolution{
                 component: ctx.id,
                 final_branch_id: branch.id,
                 port_mapping: target_mapping,
             };
             let solution_branch = self.send_or_store_local_solution(local_solution, ctx);
             if solution_branch.is_some() {
                 // No need to continue iterating, we've found the solution
                 return solution_branch;
+            }
             last_branch_id = Some(branch.id);
+        }
         self.last_finished_handled = last_branch_id;
         return None;
+    }
     pub fn end_sync(&mut self, branch_id: BranchId, final_ports: &mut Vec<PortIdLocal>) {
         debug_assert!(self.is_in_sync());
         // TODO: Handle sending and receiving ports
         // Set final ports
         final_ports.clear();
         let branch = &self.branch_annotations[branch_id.index as usize];
         for port in &branch.port_mapping {
             final_ports.push(port.port_id);
+        }
         // Clear out internal storage to defaults
         self.highest_connector_id = ConnectorId::new_invalid();
         self.branch_annotations.clear();
         self.last_finished_handled = None;
         self.encountered_peers.clear();
         self.encountered_ports.clear();
         self.solution_combiner.clear();
+    }
     // --- 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 ComponentCtxFancy) -> (SyncHeader, DataHeader) {
         debug_assert!(self.is_in_sync());
         let sync_header = self.create_sync_header(ctx);
         let branch = &mut self.branch_annotations[branch_id.index as usize];
         if cfg!(debug_assertions) {
             // Check for consistent mapping
             let port = branch.port_mapping.iter()
                 .find(|v| v.port_id == source_port_id)
                 .unwrap();
             debug_assert!(port.expected_firing == None || port.expected_firing == Some(true));
+        }
         // Check for ports that are being sent
         debug_assert!(self.workspace_ports.is_empty());
         find_ports_in_value_group(content, &mut self.workspace_ports);
         if !self.workspace_ports.is_empty() {
             todo!("handle sending ports");
             self.workspace_ports.clear();
+        }
         // Construct data header
         // TODO: Handle multiple firings. Right now we just assign the current
         //  branch to the `None` value because we know we can only send once.
         debug_assert!(branch.port_mapping.iter().find(|v| v.port_id == source_port_id).unwrap().registered_id.is_none());
         let port_info = ctx.get_port_by_id(source_port_id).unwrap();
         let data_header = DataHeader{
             expected_mapping: branch.port_mapping.clone(),
-            sending_port: port_info.peer_id,
+            sending_port: port_info.self_id,
             target_port: port_info.peer_id,
             new_mapping: branch_id
         };
         // Update port mapping
         for mapping in &mut branch.port_mapping {
             if mapping.port_id == source_port_id {
                 mapping.expected_firing = Some(true);
                 mapping.registered_id = Some(branch_id);
+            }
+        }
         return (sync_header, data_header);
         self.encountered_ports.push(source_port_id);
         return (self.create_sync_header(ctx), data_header);
+    }
     /// Handles a new data message by handling the data and sync header, and
     /// checking which *existing* branches *can* receive the message. So two
     /// cautionary notes:
     /// 1. A future branch might also be able to receive this message, see the
     ///     `branch_can_receive` function.
     /// 2. We return the branches that *can* receive the message, you still
     ///     have to explicitly call `notify_of_received_message`.
     pub fn handle_new_data_message(&mut self, exec_tree: &ExecTree, message: &DataMessageFancy, ctx: &mut ComponentCtxFancy, target_ids: &mut Vec<BranchId>) {
         self.handle_received_data_header(exec_tree, &message.data_header, target_ids);
+        self.handle_received_data_header(exec_tree, &message.data_header, &message.content, target_ids);
         self.handle_received_sync_header(&message.sync_header, ctx);
+    }
     /// Handles a new sync message by handling the sync header and the contents
     /// of the message. Returns `Some` with the branch ID of the global solution
     /// if the sync solution has been found.
     pub fn handle_new_sync_message(&mut self, message: SyncMessageFancy, ctx: &mut ComponentCtxFancy) -> Option<BranchId> {
         self.handle_received_sync_header(&message.sync_header, ctx);
         // And handle the contents
         debug_assert_eq!(message.target_component_id, ctx.id);
         match message.content {
             SyncContent::Notification => {
                 // We were just interested in the header
                 return None;
             },
             SyncContent::LocalSolution(solution) => {
                 // We might be the leader, or earlier messages caused us to not
                 // be the leader anymore.
                 return self.send_or_store_local_solution(solution, ctx);
             },
             SyncContent::GlobalSolution(solution) => {
                 // Take branch of interest and return it.
                 let (_, branch_id) = solution.component_branches.iter()
                     .find(|(connector_id, _)| *connector_id == ctx.id)
                     .unwrap();
                 return Some(*branch_id);
+            }
+        }
+    }
     pub fn notify_of_received_message(&mut self, branch_id: BranchId, data_header: &DataHeader, content: &ValueGroup) {
         debug_assert!(self.branch_can_receive(branch_id, data_header));
     pub fn notify_of_received_message(&mut self, branch_id: BranchId, data_header: &DataHeader, content: &DataContent) {
         debug_assert!(self.branch_can_receive(branch_id, data_header, content));
         let branch = &mut self.branch_annotations[branch_id.index as usize];
         for mapping in &mut branch.port_mapping {
             if mapping.port_id == data_header.target_port {
                 // Found the port in which the message should be inserted
                 mapping.registered_id = Some(data_header.new_mapping);
                 // Check for sent ports
                 debug_assert!(self.workspace_ports.is_empty());
                 find_ports_in_value_group(content, &mut self.workspace_ports);
+                find_ports_in_value_group(content.as_message().unwrap(), &mut self.workspace_ports);
                 if !self.workspace_ports.is_empty() {
                     todo!("handle received ports");
                     self.workspace_ports.clear();
+                }
                 return;
+            }
+        }
         // If here, then the branch didn't actually own the port? Means the
         // caller made a mistake
         unreachable!("incorrect notify_of_received_message");
+    }
     /// Matches the mapping between the branch and the data message. If they
     /// match then the branch can receive the message.
     pub fn branch_can_receive(&self, branch_id: BranchId, data_header: &DataHeader) -> bool {
     pub fn branch_can_receive(&self, branch_id: BranchId, data_header: &DataHeader, content: &DataContent) -> bool {
         if let DataContent::SilentPortNotification = content {
             // No port can receive a "silent" notification.
             return false;
+        }
         let annotation = &self.branch_annotations[branch_id.index as usize];
         for expected in &data_header.expected_mapping {
             // If we own the port, then we have an entry in the
             // annotation, check if the current mapping matches
             for current in &annotation.port_mapping {
                 if expected.port_id == current.port_id {
                     if expected.registered_id != current.registered_id {
                         // IDs do not match, we cannot receive the
                         // message in this branch
                         return false;
+                    }
+                }
+            }
+        }
         return true;
+    }
     // --- Internal helpers
     /// Checks data header and consults the stored port mapping and the
     /// execution tree to see which branches may receive the data message's
     /// contents.
     fn handle_received_data_header(&mut self, exec_tree: &ExecTree, data_header: &DataHeader, target_ids: &mut Vec<BranchId>) {
+    fn handle_received_data_header(&mut self, exec_tree: &ExecTree, data_header: &DataHeader, content: &DataContent, target_ids: &mut Vec<BranchId>) {
         for branch in exec_tree.iter_queue(QueueKind::AwaitingMessage, None) {
             if branch.awaiting_port == data_header.target_port {
                 // Found a branch awaiting the message, but we need to make sure
                 // the mapping is correct
                 if self.branch_can_receive(branch.id, data_header) {
+                if self.branch_can_receive(branch.id, data_header, content) {
                     target_ids.push(branch.id);
+                }
+            }
+        }
+    }
     fn handle_received_sync_header(&mut self, sync_header: &SyncHeader, ctx: &mut ComponentCtxFancy) {
         debug_assert!(sync_header.sending_component_id != ctx.id); // not sending to ourselves
         self.encountered_peers.push(sync_header.sending_component_id);
         if sync_header.highest_component_id > self.highest_connector_id {
             // Sender has higher component ID. So should be the target of our
             // messages. We should also let all of our peers know
             self.highest_connector_id = sync_header.highest_component_id;
             for encountered_id in self.encountered_peers.iter() {
                 if *encountered_id == sync_header.sending_component_id {
                     // Don't need to send it to this one
                     continue
+                }
                 let message = SyncMessageFancy{
                     sync_header: self.create_sync_header(ctx),
                     target_component_id: *encountered_id,
                     content: SyncContent::Notification,
                 };
                 ctx.submit_message(MessageFancy::Sync(message));
+            }
             // But also send our locally combined solution
             self.forward_local_solutions(ctx);
         } else if sync_header.highest_component_id < self.highest_connector_id {
             // Sender has lower leader ID, so it should know about our higher
             // one.
             let message = SyncMessageFancy{
                 sync_header: self.create_sync_header(ctx),
                 target_component_id: sync_header.sending_component_id,
                 content: SyncContent::Notification
             };
             ctx.submit_message(MessageFancy::Sync(message));
         } // else: exactly equal, so do nothing
+    }
     fn send_or_store_local_solution(&mut self, solution: LocalSolution, ctx: &mut ComponentCtxFancy) -> Option<BranchId> {
         println!("DEBUG [....:.. conn:{:02}]: Storing local solution for component {}, branch {}", ctx.id.0, solution.component.0, solution.final_branch_id.index);
         if self.highest_connector_id == ctx.id {
             // We are the leader
             if let Some(global_solution) = self.solution_combiner.add_solution_and_check_for_global_solution(solution) {
                 let mut my_final_branch_id = BranchId::new_invalid();
                 for (connector_id, branch_id) in global_solution.component_branches.iter().copied() {
                     if connector_id == ctx.id {
                         // This is our solution branch
                         my_final_branch_id = branch_id;
                         continue;
+                    }
                     let message = SyncMessageFancy{
                         sync_header: self.create_sync_header(ctx),
                         target_component_id: connector_id,
                         content: SyncContent::GlobalSolution(global_solution.clone()),
                     };
                     ctx.submit_message(MessageFancy::Sync(message));
+                }
                 debug_assert!(my_final_branch_id.is_valid());
                 return Some(my_final_branch_id);
             } else {
                 return None;
+            }
         } else {
             // Someone else is the leader
             let message = SyncMessageFancy{
                 sync_header: self.create_sync_header(ctx),
                 target_component_id: self.highest_connector_id,
                 content: SyncContent::LocalSolution(solution),
             };
             ctx.submit_message(MessageFancy::Sync(message));
             return None;
+        }
+    }
     #[inline]
     fn create_sync_header(&self, ctx: &ComponentCtxFancy) -> SyncHeader {
         return SyncHeader{
             sending_component_id: ctx.id,
             highest_component_id: self.highest_connector_id,
+        }
+    }
     fn forward_local_solutions(&mut self, ctx: &mut ComponentCtxFancy) {
         debug_assert_ne!(self.highest_connector_id, ctx.id);
         for local_solution in self.solution_combiner.drain() {
             let message = SyncMessageFancy{
                 sync_header: self.create_sync_header(ctx),
                 target_component_id: self.highest_connector_id,
                 content: SyncContent::LocalSolution(local_solution),
             };
             ctx.submit_message(MessageFancy::Sync(message));
+        }
+    }
+}
 // -----------------------------------------------------------------------------
 // Solution storage and algorithms
 // -----------------------------------------------------------------------------
 struct MatchedLocalSolution {
     final_branch_id: BranchId,
     channel_mapping: Vec<(ChannelId, BranchId)>,
     matches: Vec<ComponentMatches>,
+}
 struct ComponentMatches {
     target_id: ConnectorId,
     target_index: usize,
     match_indices: Vec<usize>, // of local solution in connector
+}
 struct ComponentPeer {
     target_id: ConnectorId,
     target_index: usize, // in array of global solution components
     involved_channels: Vec<ChannelId>,
+}
 struct ComponentLocalSolutions {
     component: ConnectorId,
     peers: Vec<ComponentPeer>,
     solutions: Vec<MatchedLocalSolution>,
     all_peers_present: bool,
+}
 // TODO: Flatten? Flatten. Flatten everything.
 pub(crate) struct SolutionCombiner {
     local: Vec<ComponentLocalSolutions>
+}
 impl SolutionCombiner {
     fn new() -> Self {
         return Self{
             local: Vec::new(),
         };
+    }
     /// Adds a new local solution to the global solution storage. Will check the
     /// new local solutions for matching against already stored local solutions
     /// of peer connectors.
     fn add_solution_and_check_for_global_solution(&mut self, solution: LocalSolution) -> Option<GlobalSolution> {
         let component_id = solution.component;
         let solution = MatchedLocalSolution{
             final_branch_id: solution.final_branch_id,
             channel_mapping: solution.port_mapping,
             matches: Vec::new(),
         };
         // Create an entry for the solution for the particular component
         let component_exists = self.local.iter_mut()
             .enumerate()
             .find(|(_, v)| v.component == component_id);
         let (component_index, solution_index, new_component) = match component_exists {
             Some((component_index, storage)) => {
                 // Entry for component exists, so add to solutions
                 let solution_index = storage.solutions.len();
                 storage.solutions.push(solution);
                 (component_index, solution_index, false)
+            }
             None => {
                 // Entry for component does not exist yet
                 let component_index = self.local.len();
                 self.local.push(ComponentLocalSolutions{
                     component: component_id,
                     peers: Vec::new(),
                     solutions: vec![solution],
                     all_peers_present: false,
                 });
                 (component_index, 0, true)
+            }
         };
         // If this is a solution of a component that is new to us, then we check
         // in the stored solutions which other components are peers of the new
         // one.
         if new_component {
             let cur_ports = &self.local[component_index].solutions[0].channel_mapping;
             let mut component_peers = Vec::new();
             // Find the matching components
             for (other_index, other_component) in self.local.iter().enumerate() {
                 if other_index == component_index {
                     // Don't match against ourselves
                     continue;
+                }
                 let mut matching_channels = Vec::new();
                 for (cur_channel_id, _) in cur_ports {
                     for (other_channel_id, _) in &other_component.solutions[0].channel_mapping {
                         if cur_channel_id == other_channel_id {
                             // We have a shared port
                             matching_channels.push(*cur_channel_id);
+                        }
+                    }
+                }
                 if !matching_channels.is_empty() {
                     // We share some ports
                     component_peers.push(ComponentPeer{
                         target_id: other_component.component,
                         target_index: other_index,
                         involved_channels: matching_channels,
                     });
+                }
+            }
             let mut num_ports_in_peers = 0;
             for peer in &component_peers {
                 num_ports_in_peers += peer.involved_channels.len();
+            }
             if num_ports_in_peers == cur_ports.len() {
                 // Newly added component has all required peers present
                 self.local[component_index].all_peers_present = true;
+            }
             // Add the found component pairing entries to the solution entries
             // for the two involved components
             for component_match in component_peers {
                 // Check the other component for having all peers present
                 let mut num_ports_in_peers = component_match.involved_channels.len();
                 let other_component = &mut self.local[component_match.target_index];
                 for existing_peer in &other_component.peers {
                     num_ports_in_peers += existing_peer.involved_channels.len();
+                }
                 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_new_solution(component_index, solution_index);
+    }
     /// Checks if, starting at the provided local solution, a global solution
     /// can be formed.
     fn check_new_solution(&self, component_index: usize, solution_index: usize) -> Option<GlobalSolution> {
         if !self.can_have_solution() {
             return None;
+        }
         // By now we're certain that all peers are present. So once our
         // backtracking solution stack is as long as the number of components,
         // then we have found a global solution.
         let mut check_stack = Vec::new();
         let mut check_from = 0;
         check_stack.push((component_index, solution_index));
         'checking_loop: while check_from < check_stack.len() {
             // Prepare for next iteration
             let new_check_from = check_stack.len();
             // Go through all entries on the checking stack. Each entry
             // corresponds to a component's solution. We check that one against
             // previously added ones on the stack, and if they're not already
             // added we push them onto the check stack.
             for check_idx in check_from..new_check_from {
                 // Take the current solution
                 let (component_index, solution_index) = check_stack[check_idx];
                 debug_assert!(!self.local[component_index].solutions.is_empty());
                 let cur_solution = &self.local[component_index].solutions[solution_index];
                 // Go through the matches and check if they're on the stack or
                 // should be added to the stack.
                 for cur_match in &cur_solution.matches {
                     let mut is_already_on_stack = false;
                     let mut has_same_solution = false;
                     for existing_check_idx in 0..check_from {
                         let (existing_component_index, existing_solution_index) = check_stack[existing_check_idx];
                         if existing_component_index == cur_match.target_index {
                             // Already lives on the stack, so the match MUST
                             // contain the same solution index if the checked
                             // local solution is agreeable with the (partially
                             // determined) global solution.
                             is_already_on_stack = true;
                             if cur_match.match_indices.contains(&existing_solution_index) {
                                 has_same_solution = true;
                                 break;
+                            }
+                        }
+                    }
                     if is_already_on_stack {
                         if !has_same_solution {
                             // We have an inconsistency, so we need to go back
                             // in our stack, and try the next solution
                             let (last_component_index, last_solution_index) = check_stack[check_from];
                             check_stack.truncate(check_from);
                             if check_stack.is_empty() {
                                 // The starting point does not yield a valid
                                 // solution
                                 return None;
+                            }
                             // Try the next one
                             let last_component = &self.local[last_component_index];
                             let new_solution_index = last_solution_index + 1;
                             if new_solution_index >= last_component.solutions.len() {
                                 // No more things to try, again: no valid
                                 // solution
                                 return None;
+                            }
                             check_stack.push((last_component_index, new_solution_index));
                             continue 'checking_loop;
                         } // else: we're fine, the solution is agreeable
                     } else {
                         check_stack.push((cur_match.target_index, 0))
+                    }
+                }
+            }
             check_from = new_check_from;
+        }
         // Because of our earlier checking if we can have a solution at
         // all (all components have their peers), and the exit condition of the
         // while loop: if we're here, then we have a global solution
         debug_assert_eq!(check_stack.len(), self.local.len());
         let mut final_branches = Vec::with_capacity(check_stack.len());
         for (component_index, solution_index) in check_stack.iter().copied() {
             let component = &self.local[component_index];
             let solution = &component.solutions[solution_index];
             final_branches.push((component.component, solution.final_branch_id));
+        }
         // Just debugging here, TODO: @remove
         let mut total_num_channels = 0;
         for (component_index, _) in check_stack.iter().copied() {
             let component = &self.local[component_index];
             total_num_channels += component.solutions[0].channel_mapping.len();
+        }
         total_num_channels /= 2;
         let mut final_mapping = Vec::with_capacity(total_num_channels);
         let mut total_num_checked = 0;
         for (component_index, solution_index) in check_stack.iter().copied() {
             let component = &self.local[component_index];
             let solution = &component.solutions[solution_index];
             for (channel_id, branch_id) in solution.channel_mapping.iter().copied() {
                 match final_mapping.iter().find(|(v, _)| *v == channel_id) {
                     Some((_, encountered_branch_id)) => {
                         debug_assert_eq!(*encountered_branch_id, branch_id);
                         total_num_checked += 1;
                     },
                     None => {
                         final_mapping.push((channel_id, branch_id));
+                    }
+                }
+            }
+        }
         debug_assert_eq!(total_num_checked, total_num_channels);
         return Some(GlobalSolution{
             component_branches: final_branches,
             channel_mapping: final_mapping,
         });
+    }
     /// Simple test if a solution is at all possible. If this returns true it
     /// does not mean there actually is a solution.
     fn can_have_solution(&self) -> bool {
         for component in &self.local {
             if !component.all_peers_present {
                 return false;
+            }
+        }
         return true;
+    }
     /// Turns the entire (partially resolved) global solution back into local
     /// solutions to ship to another component.
     // TODO: Don't do this, kind of wasteful since a lot of processing has
     //  already been performed.
     fn drain(&mut self) -> Vec<LocalSolution> {
         let mut reserve_len = 0;
         for component in &self.local {
             reserve_len += component.solutions.len();
+        }
         let mut solutions = Vec::with_capacity(reserve_len);
         for component in self.local.drain(..) {
             for solution in component.solutions {
                 solutions.push(LocalSolution{
                     component: component.component,
                     final_branch_id: solution.final_branch_id,
                     port_mapping: solution.channel_mapping,
                 });
+            }
+        }
         return solutions;
+    }
     fn clear(&mut self) {
         self.local.clear();
+    }
+}
 // -----------------------------------------------------------------------------
 // Generic Helpers
 // -----------------------------------------------------------------------------
 /// Recursively goes through the value group, attempting to find ports.
 /// Duplicates will only be added once.
 pub(crate) fn find_ports_in_value_group(value_group: &ValueGroup, ports: &mut Vec<PortIdLocal>) {
     // Helper to check a value for a port and recurse if needed.
     use crate::protocol::eval::Value;
     fn find_port_in_value(group: &ValueGroup, value: &Value, ports: &mut Vec<PortIdLocal>) {
         match value {
             Value::Input(port_id) | Value::Output(port_id) => {
                 // This is an actual port
                 let cur_port = PortIdLocal::new(port_id.0.u32_suffix);
                 for prev_port in ports.iter() {
                     if *prev_port == cur_port {
                         // Already added
                         return;
+                    }
+                }
                 ports.push(cur_port);
             },
             Value::Array(heap_pos) |
             Value::Message(heap_pos) |
             Value::String(heap_pos) |
             Value::Struct(heap_pos) |
             Value::Union(_, heap_pos) => {
                 // Reference to some dynamic thing which might contain ports,
                 // so recurse
                 let heap_region = &group.regions[*heap_pos as usize];
                 for embedded_value in heap_region {
                     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/inbox2.rs

➞

Show inline comments

 use std::sync::Mutex;
 use std::collections::VecDeque;
 use crate::protocol::eval::ValueGroup;
 use crate::runtime2::branch::BranchId;
 use crate::runtime2::ConnectorId;
 use crate::runtime2::consensus::{GlobalSolution, LocalSolution};
 use crate::runtime2::port::PortIdLocal;
 // TODO: Remove Debug derive from all types
 #[derive(Debug, Copy, Clone)]
 pub(crate) struct PortAnnotation {
     pub port_id: PortIdLocal,
     pub registered_id: Option<BranchId>,
     pub expected_firing: Option<bool>,
+}
 /// The header added by the synchronization algorithm to all.
 #[derive(Debug, Clone)]
 pub(crate) struct SyncHeader {
     pub sending_component_id: ConnectorId,
     pub highest_component_id: ConnectorId,
+}
 /// The header added to data messages
 #[derive(Debug, Clone)]
 pub(crate) struct DataHeader {
     pub expected_mapping: Vec<PortAnnotation>,
     pub sending_port: PortIdLocal,
     pub target_port: PortIdLocal,
     pub new_mapping: BranchId,
+}
 // TODO: Very much on the fence about this. On one hand I thought making it a
 //  data message was neat because "silent port notification" should be rerouted
 //  like any other data message to determine the component ID of the receiver
 //  and to make it part of the leader election algorithm for the sync leader.
 //  However: it complicates logic quite a bit. Really it might be easier to
 //  create `Message::SyncAtComponent` and `Message::SyncAtPort` messages...
 #[derive(Debug, Clone)]
 pub(crate) enum DataContent {
     SilentPortNotification,
     Message(ValueGroup),
+}
 impl DataContent {
     pub(crate) fn as_message(&self) -> Option<&ValueGroup> {
         match self {
             DataContent::SilentPortNotification => None,
             DataContent::Message(message) => Some(message),
+        }
+    }
+}
 /// A data message is a message that is intended for the receiver's PDL code,
 /// but will also be handled by the consensus algorithm
 #[derive(Debug, Clone)]
 pub(crate) struct DataMessageFancy {
     pub sync_header: SyncHeader,
     pub data_header: DataHeader,
-    pub content: ValueGroup,
+    pub content: DataContent,
+}
 #[derive(Debug)]
 pub(crate) enum SyncContent {
     LocalSolution(LocalSolution), // sending a local solution to the leader
     GlobalSolution(GlobalSolution), // broadcasting to everyone
     Notification, // just a notification (so purpose of message is to send the SyncHeader)
+}
 /// A sync message is a message that is intended only for the consensus
 /// algorithm.
 #[derive(Debug)]
 pub(crate) struct SyncMessageFancy {
     pub sync_header: SyncHeader,
     pub target_component_id: ConnectorId,
     pub content: SyncContent,
+}
 /// A control message is a message intended for the scheduler that is executing
 /// a component.
 #[derive(Debug)]
 pub(crate) struct ControlMessageFancy {
     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 MessageFancy {
     Data(DataMessageFancy),
     Sync(SyncMessageFancy),
     Control(ControlMessageFancy),
+}
 /// 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<MessageFancy>>,
+}
 impl PublicInbox {
     pub fn new() -> Self {
         Self{
             messages: Mutex::new(VecDeque::new()),
+        }
+    }
     pub(crate) fn insert_message(&self, message: MessageFancy) {
         let mut lock = self.messages.lock().unwrap();
         lock.push_back(message);
+    }
     pub(crate) fn take_message(&self) -> Option<MessageFancy> {
         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/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 connector2::{ConnectorPDL, ConnectorPublic, ConnectorScheduling};
 use scheduler::{Scheduler, ComponentCtxFancy, SchedulerCtx, ControlMessageHandler};
 use native::{Connector, ConnectorApplication, ApplicationInterface};
 use inbox2::MessageFancy;
 use port::{ChannelId, Port, PortState};
 /// 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, PartialOrd, Ord)]
 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();
+    }
     pub(crate) fn push_work(&self, key: ConnectorKey) {
         let mut lock = self.connector_queue.lock().unwrap();
         lock.push_back(key);
         self.scheduler_notifier.notify_one();
+    }
     // --- Creating/using ports
     /// Creates a new port pair. Note that these are stored globally like the
     /// connectors are. Ports stored by components belong to those components.
     pub(crate) fn create_channel(&self, creating_connector: ConnectorId) -> (Port, Port) {
         use port::{PortIdLocal, PortKind};
         let getter_id = self.port_counter.fetch_add(2, Ordering::SeqCst);
         let channel_id = ChannelId::new(getter_id);
         let putter_id = PortIdLocal::new(getter_id + 1);
         let getter_id = PortIdLocal::new(getter_id);
         let getter_port = Port{
             self_id: getter_id,
             peer_id: putter_id,
             channel_id,
             kind: PortKind::Getter,
             state: PortState::Open,
             peer_connector: creating_connector,
         };
         let putter_port = Port{
             self_id: putter_id,
             peer_id: getter_id,
             channel_id,
             kind: PortKind::Putter,
             state: PortState::Open,
             peer_connector: creating_connector,
         };
         return (getter_port, putter_port);
+    }
     /// Sends a message to a particular connector. If the connector happened to
     /// be sleeping then it will be scheduled for execution.
     pub(crate) fn send_message(&self, target_id: ConnectorId, message: MessageFancy) {
         let target = self.get_component_public(target_id);
         target.inbox.insert_message(message);
         let should_wake_up = target.sleeping
             .compare_exchange(true, false, Ordering::SeqCst, Ordering::Acquire)
             .is_ok();
         if should_wake_up {
             let key = unsafe{ ConnectorKey::from_id(target_id) };
             self.push_work(key);
+        }
+    }
     // --- Creating/retrieving/destroying components
     /// Creates an initially sleeping application connector.
     fn create_interface_component(&self, component: ConnectorApplication) -> ConnectorKey {
         // Initialize as sleeping, as it will be scheduled by the programmer.
         let mut lock = self.connectors.write().unwrap();
         let key = lock.create(ConnectorVariant::Native(Box::new(component)), true);
         self.increment_active_components();
         return key;
+    }
     /// Creates a new PDL component. This function just creates the component.
     /// If you create it initially awake, then you must add it to the work
     /// queue. Other aspects of correctness (i.e. setting initial ports) are
     /// relinquished to the caller!
     pub(crate) fn create_pdl_component(&self, connector: ConnectorPDL, initially_sleeping: bool) -> ConnectorKey {
         // Create as not sleeping, as we'll schedule it immediately
         let key = {
             let mut lock = self.connectors.write().unwrap();
             lock.create(ConnectorVariant::UserDefined(connector), initially_sleeping)
         };
         self.increment_active_components();
         return key;
+    }
     #[inline]
     pub(crate) fn get_component_private(&self, connector_key: &ConnectorKey) -> &'static mut ScheduledConnector {
         let lock = self.connectors.read().unwrap();
         return lock.get_private(connector_key);
+    }
     #[inline]
     pub(crate) fn get_component_public(&self, connector_id: ConnectorId) -> &'static ConnectorPublic {
         let lock = self.connectors.read().unwrap();
         return lock.get_public(connector_id);
+    }
     pub(crate) fn destroy_component(&self, connector_key: ConnectorKey) {
         let mut lock = self.connectors.write().unwrap();
         lock.destroy(connector_key);
         self.decrement_active_components();
+    }
     // --- Managing exit condition
     #[inline]
     pub(crate) fn increment_active_interfaces(&self) {
         let _old_num = self.active_interfaces.fetch_add(1, Ordering::SeqCst);
         println!("DEBUG: Incremented active interfaces to {}", _old_num + 1);
         debug_assert_ne!(_old_num, 0); // once it hits 0, it stays zero
+    }
     pub(crate) fn decrement_active_interfaces(&self) {
         let old_num = self.active_interfaces.fetch_sub(1, Ordering::SeqCst);
         println!("DEBUG: Decremented active interfaces to {}", old_num - 1);
         debug_assert!(old_num > 0);
         if old_num == 1 { // such that active interfaces is now 0
             let num_connectors = self.active_connectors.load(Ordering::Acquire);
             if num_connectors == 0 {
                 self.signal_for_shutdown();
+            }
+        }
+    }
     #[inline]
     fn increment_active_components(&self) {
         let _old_num = self.active_connectors.fetch_add(1, Ordering::SeqCst);
         println!("DEBUG: Incremented components to {}", _old_num + 1);
+    }
     fn decrement_active_components(&self) {
         let old_num = self.active_connectors.fetch_sub(1, Ordering::SeqCst);
         println!("DEBUG: Decremented components to {}", old_num - 1);
         debug_assert!(old_num > 0);
         if old_num == 1 { // such that we have no more active connectors (for now!)
             let num_interfaces = self.active_interfaces.load(Ordering::Acquire);
             if num_interfaces == 0 {
                 self.signal_for_shutdown();
+            }
+        }
+    }
     #[inline]
     fn signal_for_shutdown(&self) {
         debug_assert_eq!(self.active_interfaces.load(Ordering::Acquire), 0);
         debug_assert_eq!(self.active_connectors.load(Ordering::Acquire), 0);
         println!("DEBUG: Signaling for shutdown");
         let _lock = self.connector_queue.lock().unwrap();
         let should_signal = self.should_exit
             .compare_exchange(false, true, Ordering::SeqCst, Ordering::Acquire)
             .is_ok();
         if should_signal {
             println!("DEBUG: Notifying all waiting schedulers");
             self.scheduler_notifier.notify_all();
+        }
+    }
+}
 // TODO: Come back to this at some point
 unsafe impl Send for RuntimeInner {}
 unsafe impl Sync for RuntimeInner {}
 // -----------------------------------------------------------------------------
 // ConnectorStore
 // -----------------------------------------------------------------------------
 struct ConnectorStore {
     // Freelist storage of connectors. Storage should be pointer-stable as
     // someone might be mutating the vector while we're executing one of the
     // connectors.
     connectors: RawVec<*mut ScheduledConnector>,
     free: Vec<usize>,
+}
 impl ConnectorStore {
     fn with_capacity(capacity: usize) -> Self {
         Self {
             connectors: RawVec::with_capacity(capacity),
             free: Vec::with_capacity(capacity),
+        }
+    }
     /// Retrieves public part of connector - accessible by many threads at once.
     fn get_public(&self, id: ConnectorId) -> &'static ConnectorPublic {
         unsafe {
             debug_assert!(!self.free.contains(&(id.0 as usize)));
             let connector = self.connectors.get(id.0 as usize);
             debug_assert!(!connector.is_null());
             return &(**connector).public;
+        }
+    }
     /// Retrieves private part of connector - accessible by one thread at a
     /// time.
     fn get_private(&self, key: &ConnectorKey) -> &'static mut ScheduledConnector {
         unsafe {
             debug_assert!(!self.free.contains(&(key.index as usize)));
             let connector = self.connectors.get_mut(key.index as usize);
             debug_assert!(!connector.is_null());
             return &mut (**connector);
+        }
+    }
     /// Creates a new connector. Caller should ensure ports are set up correctly
     /// and the connector is queued for execution if needed.
     fn create(&mut self, connector: ConnectorVariant, initially_sleeping: bool) -> ConnectorKey {
         let mut connector = ScheduledConnector {
             connector,
             ctx_fancy: ComponentCtxFancy::new_empty(),
             public: ConnectorPublic::new(initially_sleeping),
             router: ControlMessageHandler::new(),
             shutting_down: false,
         };
         let index;
         let key;
         if self.free.is_empty() {
             // No free entries, allocate new entry
             index = self.connectors.len();
             key = ConnectorKey{ index: index as u32 };
             connector.ctx_fancy.id = key.downcast();
             let connector = Box::into_raw(Box::new(connector));
             self.connectors.push(connector);
         } else {
             // Free spot available
             index = self.free.pop().unwrap();
             key = ConnectorKey{ index: index as u32 };
             connector.ctx_fancy.id = key.downcast();
             unsafe {
                 let target = self.connectors.get_mut(index);
                 std::ptr::write(*target, connector);
+            }
+        }
         return key;
+    }
     /// Destroys a connector. Caller should make sure it is not scheduled for
     /// execution. Otherwise one experiences "bad stuff" (tm).
     fn destroy(&mut self, key: ConnectorKey) {
         unsafe {
             let target = self.connectors.get_mut(key.index as usize);
             std::ptr::drop_in_place(*target);
             // Note: but not deallocating!
+        }
         self.free.push(key.index as usize);
+    }
+}
 impl Drop for ConnectorStore {
     fn drop(&mut self) {
         // Everything in the freelist already had its destructor called, so only
         // has to be deallocated
         for free_idx in self.free.iter().copied() {
             unsafe {
                 let memory = self.connectors.get_mut(free_idx);
                 let layout = std::alloc::Layout::for_value(&**memory);
                 std::alloc::dealloc(*memory as *mut u8, layout);
                 // mark as null for the remainder
                 *memory = std::ptr::null_mut();
+            }
+        }
         // With the deallocated stuff marked as null, clear the remainder that
         // is not null
         for idx in 0..self.connectors.len() {
             unsafe {
                 let memory = *self.connectors.get_mut(idx);
                 if !memory.is_null() {
                     let _ = Box::from_raw(memory); // take care of deallocation, bit dirty, but meh
+                }
+            }
+        }
+    }
+}
@@ \ 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 crate::protocol::ComponentCreationError;
 use crate::protocol::eval::ValueGroup;
 use super::{ConnectorKey, ConnectorId, RuntimeInner};
 use super::scheduler::{SchedulerCtx, ComponentCtxFancy};
 use super::port::{Port, PortIdLocal, Channel, PortKind};
 use super::consensus::find_ports_in_value_group;
 use super::connector2::{ConnectorScheduling, ConnectorPDL};
 use super::inbox2::{MessageFancy, ControlContent, ControlMessageFancy};
 /// 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 ComponentCtxFancy) -> ConnectorScheduling;
+}
 type SyncDone = Arc<(Mutex<bool>, Condvar)>;
 type JobQueue = Arc<Mutex<VecDeque<ApplicationJob>>>;
 enum ApplicationJob {
     NewChannel((Port, Port)),
     NewConnector(ConnectorPDL, Vec<PortIdLocal>),
     Shutdown,
+}
 /// The connector which an application can directly interface with. Once may set
 /// up the next synchronous round, and retrieve the data afterwards.
 pub struct ConnectorApplication {
     sync_done: SyncDone,
     job_queue: JobQueue,
+}
 impl ConnectorApplication {
     pub(crate) fn new(runtime: Arc<RuntimeInner>) -> (Self, ApplicationInterface) {
         let sync_done = Arc::new(( Mutex::new(false), 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()
         };
         let interface = ApplicationInterface::new(sync_done, job_queue, runtime);
         return (connector, interface);
+    }
+}
 impl Connector for ConnectorApplication {
     fn run(&mut self, _sched_ctx: SchedulerCtx, comp_ctx: &mut ComponentCtxFancy) -> ConnectorScheduling {
         // Handle any incoming messages if we're participating in a round
         while let Some(message) = comp_ctx.read_next_message() {
             match message {
                 MessageFancy::Data(_) => todo!("data message in API connector"),
                 MessageFancy::Sync(_)  => todo!("sync message in API connector"),
                 MessageFancy::Control(_) => todo!("impossible control message"),
+            }
+        }
         // Handle requests coming from the API
+        {
             let mut queue = self.job_queue.lock().unwrap();
             while let Some(job) = queue.pop_front() {
                 match job {
                     ApplicationJob::NewChannel((endpoint_a, endpoint_b)) => {
                         println!("DEBUG: API adopting ports");
                         comp_ctx.push_port(endpoint_a);
                         comp_ctx.push_port(endpoint_b);
+                    }
                     ApplicationJob::NewConnector(connector, initial_ports) => {
                         println!("DEBUG: API creating connector");
                         comp_ctx.push_component(connector, initial_ports);
                     },
                     ApplicationJob::Shutdown => {
                         debug_assert!(queue.is_empty());
                         return ConnectorScheduling::Exit;
+                    }
+                }
+            }
+        }
         return ConnectorScheduling::NotNow;
+    }
+}
 /// 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>,
     connector_id: ConnectorId,
     owned_ports: Vec<PortIdLocal>,
+}
 impl ApplicationInterface {
     fn new(sync_done: SyncDone, job_queue: JobQueue, runtime: Arc<RuntimeInner>) -> Self {
         return Self{
             sync_done, job_queue, runtime,
             connector_id: ConnectorId::new_invalid(),
             owned_ports: Vec::new(),
+        }
+    }
     /// Creates a new channel.
     pub fn create_channel(&mut self) -> Channel {
         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(putter_id);
         self.owned_ports.push(getter_id);
         return 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.
     // TODO: Yank out scheduler logic for common use.
     pub fn create_connector(&mut self, module: &str, routine: &str, arguments: ValueGroup) -> Result<(), ComponentCreationError> {
         // 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 state = self.runtime.protocol_description.new_component_v2(module.as_bytes(), routine.as_bytes(), arguments)?;
         let connector = ConnectorPDL::new(state);
         // 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(());
+    }
     /// Check if the next sync-round is finished.
     pub fn try_wait(&self) -> bool {
         let (is_done, _) = &*self.sync_done;
         let lock = is_done.lock().unwrap();
         return *lock;
+    }
     /// Wait until the next sync-round is finished
     pub fn wait(&self) {
         let (is_done, condition) = &*self.sync_done;
         let lock = is_done.lock().unwrap();
         condition.wait_while(lock, |v| !*v).unwrap(); // wait while not done
+    }
     /// 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(MessageFancy::Control(ControlMessageFancy{
             id: 0,
             sending_component_id: self.connector_id,
-            content: ControlContent::Ack
+            content: ControlContent::Ping,
         }));
         let should_wake_up = connector.sleeping
             .compare_exchange(true, false, Ordering::SeqCst, Ordering::Acquire)
             .is_ok();
         if should_wake_up {
             println!("DEBUG: Waking up connector");
             let key = unsafe{ ConnectorKey::from_id(self.connector_id) };
             self.runtime.push_work(key);
         } else {
             println!("DEBUG: NOT waking up connector");
+        }
+    }
+}
 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::sync::Arc;
 use std::sync::atomic::Ordering;
 use crate::runtime2::inbox2::ControlContent;
 use super::{ScheduledConnector, RuntimeInner, ConnectorId, ConnectorKey};
 use super::port::{Port, PortState, PortIdLocal};
 use super::native::Connector;
 use super::branch::{BranchId};
 use super::connector2::{ConnectorPDL, ConnectorScheduling};
 use super::inbox2::{MessageFancy, DataMessageFancy, ControlMessageFancy};
 // 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, MessageFancy::Control(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 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_data_message_target_port(&message) {
+            if let Some(target_port) = Self::get_message_target_port(&message) {
                 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 {
                 MessageFancy::Control(message) => {
                     match message.content {
                         ControlContent::PortPeerChanged(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 = MessageFancy::Control(ControlMessageFancy{
                                 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_fancy.get_port_mut_by_id(port_id).unwrap();
                             port.state = PortState::Closed;
                             // Send an Ack
                             let ack_message = MessageFancy::Control(ControlMessageFancy{
                                 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);
                         },
                         ControlContent::Ping => {},
+                    }
                 },
                 _ => {
                     // All other cases have to be handled by the component
                     scheduled.ctx_fancy.inbox_messages.push(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_fancy.id;
         // Handling any messages that were sent
         while let Some(message) = scheduled.ctx_fancy.outbox.pop_front() {
             self.debug_conn(connector_id, &format!("Sending message [outbox] \n --- {:?}", message));
             let target_component_id = match &message {
                 MessageFancy::Data(content) => {
                     // Data messages are always sent to a particular port, and
                     // may end up being rerouted.
                     let port_desc = scheduled.ctx_fancy.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
                 },
                 MessageFancy::Sync(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.sync_header.highest_component_id
+                    content.target_component_id
                 },
                 MessageFancy::Control(_) => {
                     unreachable!("component sending control messages directly");
+                }
             };
             self.runtime.send_message(target_component_id, message);
+        }
         while let Some(state_change) = scheduled.ctx_fancy.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_fancy.inbox_messages.len() {
                             let message = &scheduled.ctx_fancy.inbox_messages[message_idx];
-                            if Self::get_data_message_target_port(message) == Some(port_id) {
+                            if Self::get_message_target_port(message) == Some(port_id) {
                                 // Need to transfer this message
                                 let message = scheduled.ctx_fancy.inbox_messages.remove(message_idx);
                                 new_connector.ctx_fancy.inbox_messages.push(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, MessageFancy::Control(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)
+            }
+        }
+    }
     #[inline]
     fn get_data_message_target_port(message: &MessageFancy) -> Option<PortIdLocal> {
         if let MessageFancy::Data(message) = message {
             return Some(message.data_header.target_port)
     fn get_message_target_port(message: &MessageFancy) -> Option<PortIdLocal> {
         match message {
             MessageFancy::Data(data) => return Some(data.data_header.target_port),
             MessageFancy::Sync(_) => {},
             MessageFancy::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 ComponentCtxFancy {
     // Mostly managed by the scheduler
     pub(crate) id: ConnectorId,
     ports: Vec<Port>,
     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<MessageFancy>,
     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 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(),
             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))
+    }
     #[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);
+    }
     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: MessageFancy) {
         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) -> MessagesIter {
         return MessagesIter {
             messages: &self.inbox_messages,
             next_index: 0,
             max_index: self.inbox_len_read,
             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<MessageFancy> {
         if !self.is_in_sync { return None; }
         if self.inbox_len_read == self.inbox_messages.len() { return None; }
         // 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 {
             MessageFancy::Data(content) => {
                 self.inbox_len_read += 1;
                 return Some(MessageFancy::Data(content.clone()));
             },
             MessageFancy::Sync(_) => {
                 let message = self.inbox_messages.remove(self.inbox_len_read);
                 return Some(message);
             },
             MessageFancy::Control(_) => unreachable!("control message ended up in component inbox"),
+        }
+    }
+}
 pub(crate) struct MessagesIter<'a> {
     messages: &'a [MessageFancy],
     next_index: usize,
     max_index: usize,
     match_port_id: PortIdLocal,
+}
 impl<'a> Iterator for MessagesIter<'a> {
     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 MessageFancy::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;
+    }
+}
 // -----------------------------------------------------------------------------
 // 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
     ) -> ControlMessageFancy {
         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 ControlMessageFancy{
             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
     ) -> ControlMessageFancy {
         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 ControlMessageFancy{
             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) {
         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 @@

src/runtime2/tests/mod.rs

➞

Show inline comments

 use std::sync::Arc;
 use super::*;
 use crate::{PortId, ProtocolDescription};
 use crate::common::Id;
 use crate::protocol::eval::*;
-const NUM_THREADS: u32 = 1;     // number of threads in runtime
+const NUM_THREADS: u32 = 10;  // number of threads in runtime
 const NUM_INSTANCES: u32 = 10;  // number of test instances constructed
 const NUM_LOOPS: u32 = 10;      // number of loops within a single test (not used by all tests)
 fn create_runtime(pdl: &str) -> Runtime {
     let protocol = ProtocolDescription::parse(pdl.as_bytes()).expect("parse pdl");
     let runtime = Runtime::new(NUM_THREADS, protocol);
     return runtime;
+}
 fn run_test_in_runtime<F: Fn(&mut ApplicationInterface)>(pdl: &str, constructor: F) {
     let protocol = ProtocolDescription::parse(pdl.as_bytes())
         .expect("parse PDL");
     let runtime = Runtime::new(NUM_THREADS, protocol);
     let mut api = runtime.create_interface();
     for _ in 0..NUM_INSTANCES {
         constructor(&mut api);
+    }
     // Wait until done :)
+}
 #[test]
 fn test_put_and_get() {
     const CODE: &'static str = "
     primitive putter(out<bool> sender, u32 loops) {
         u32 index = 0;
         while (index < loops) {
             synchronous {
                 print(\"putting!\");
                 put(sender, true);
+            }
             index += 1;
+        }
+    }
     primitive getter(in<bool> receiver, u32 loops) {
         u32 index = 0;
         while (index < loops) {
             synchronous {
                 print(\"getting!\");
                 auto result = get(receiver);
                 assert(result);
+            }
             index += 1;
+        }
+    }
     ";
     run_test_in_runtime(CODE, |api| {
         let channel = api.create_channel();
         api.create_connector("", "putter", ValueGroup::new_stack(vec![
             Value::Output(PortId(Id{ connector_id: 0, u32_suffix: channel.putter_id.index })),
             Value::UInt32(NUM_LOOPS)
         ])).expect("create putter");
         api.create_connector("", "getter", ValueGroup::new_stack(vec![
             Value::Input(PortId(Id{ connector_id: 0, u32_suffix: channel.getter_id.index })),
             Value::UInt32(NUM_LOOPS)
         ])).expect("create getter");
     });
+}
@@ \ No newline at end of file @@

0 comments (0 inline, 0 general)