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

Changeset - 088be7630245

Parent rev.

Child rev.

[Not reviewed]

0 9 0

mh - 4 years ago 2021-11-14 15:34:10
contact@maxhenger.nl

Implement multi-put and multi-get

9 files changed with 206 insertions and 173 deletions:

src/protocol/eval/executor.rs

src/protocol/mod.rs

src/runtime2/branch.rs

src/runtime2/connector.rs

src/runtime2/consensus.rs

src/runtime2/inbox.rs

src/runtime2/native.rs

src/runtime2/scheduler.rs

src/runtime2/tests/mod.rs

0 comments (0 inline, 0 general)

src/protocol/eval/executor.rs

➞

Show inline comments

@@ @@ -564,82 +564,83 @@ impl Prompt { @@
                             let subject = cur_frame.expr_values.pop_back().unwrap();
                             match apply_casting(&mut self.store, output_type, &subject) {
                                 Ok(value) => cur_frame.expr_values.push_back(value),
                                 Err(msg) => {
                                     return Err(EvalError::new_error_at_expr(self, modules, heap, expr.this.upcast(), msg));
+                                }
+                            }
                             self.store.drop_value(subject.get_heap_pos());
+                        }
                         Expression::Call(expr) => {
                             // If we're dealing with a builtin we don't do any
                             // fancy shenanigans at all, just push the result.
                             match expr.method {
                                 Method::Get => {
                                     let value = cur_frame.expr_values.pop_front().unwrap();
                                     let value = self.store.maybe_read_ref(&value).clone();
                                     let port_id = if let Value::Input(port_id) = value {
                                         port_id
                                     } else {
                                         unreachable!("executor calling 'get' on value {:?}", value)
                                     };
-                                    match ctx.get(port_id) {
+                                    match ctx.performed_get(port_id) {
                                         Some(result) => {
                                             // We have the result. Merge the `ValueGroup` with the
                                             // stack/heap storage.
                                             debug_assert_eq!(result.values.len(), 1);
                                             result.into_stack(&mut cur_frame.expr_values, &mut self.store);
                                         },
                                         None => {
                                             // Don't have the result yet, prepare the expression to
                                             // get run again after we've received a message.
                                             cur_frame.expr_values.push_front(value.clone());
                                             cur_frame.expr_stack.push_back(ExprInstruction::EvalExpr(expr_id));
                                             return Ok(EvalContinuation::BlockGet(port_id));
+                                        }
+                                    }
                                 },
                                 Method::Put => {
                                     let port_value = cur_frame.expr_values.pop_front().unwrap();
                                     let deref_port_value = self.store.maybe_read_ref(&port_value).clone();
                                     let port_id = if let Value::Output(port_id) = deref_port_value {
                                         port_id
                                     } else {
                                         unreachable!("executor calling 'put' on value {:?}", deref_port_value)
                                     };
                                     let msg_value = cur_frame.expr_values.pop_front().unwrap();
                                     let deref_msg_value = self.store.maybe_read_ref(&msg_value).clone();
-                                    if ctx.did_put(port_id) {
+                                    if ctx.performed_put(port_id) {
                                         // We're fine, deallocate in case the expression value stack
                                         // held an owned value
                                         self.store.drop_value(msg_value.get_heap_pos());
                                     } else {
                                         // Prepare to execute again
                                         cur_frame.expr_values.push_front(msg_value);
                                         cur_frame.expr_values.push_front(port_value);
                                         cur_frame.expr_stack.push_back(ExprInstruction::EvalExpr(expr_id));
                                         return Ok(EvalContinuation::Put(port_id, deref_msg_value));
+                                    }
                                 },
                                 Method::Fires => {
                                     let port_value = cur_frame.expr_values.pop_front().unwrap();
                                     let port_value_deref = self.store.maybe_read_ref(&port_value).clone();
                                     let port_id = match port_value_deref {
                                         Value::Input(port_id) => port_id,
                                         Value::Output(port_id) => port_id,
                                         _ => unreachable!("executor calling 'fires' on value {:?}", port_value_deref),
                                     };
                                     match ctx.fires(port_id) {
                                         None => {
                                             cur_frame.expr_values.push_front(port_value);
                                             cur_frame.expr_stack.push_back(ExprInstruction::EvalExpr(expr_id));
                                             return Ok(EvalContinuation::BlockFires(port_id));
                                         },
                                         Some(value) => {
                                             cur_frame.expr_values.push_back(value);
@@ @@ -790,49 +791,49 @@ impl Prompt { @@
             Statement::Block(stmt) => {
                 debug_assert!(stmt.statements.is_empty() || stmt.next == stmt.statements[0]);
                 cur_frame.position = stmt.next;
                 Ok(EvalContinuation::Stepping)
             },
             Statement::EndBlock(stmt) => {
                 let block = &heap[stmt.start_block];
                 self.store.clear_stack(block.first_unique_id_in_scope as usize);
                 cur_frame.position = stmt.next;
                 Ok(EvalContinuation::Stepping)
             },
             Statement::Local(stmt) => {
                 match stmt {
                     LocalStatement::Memory(stmt) => {
                         let variable = &heap[stmt.variable];
                         self.store.write(ValueId::Stack(variable.unique_id_in_scope as u32), Value::Unassigned);
                         cur_frame.position = stmt.next;
                         Ok(EvalContinuation::Stepping)
                     },
                     LocalStatement::Channel(stmt) => {
                         // Need to create a new channel by requesting it from
                         // the runtime.
-                        match ctx.get_channel() {
+                        match ctx.created_channel() {
                             None => {
                                 // No channel is pending. So request one
                                 Ok(EvalContinuation::NewChannel)
                             },
                             Some((put_port, get_port)) => {
                                 self.store.write(ValueId::Stack(heap[stmt.from].unique_id_in_scope as u32), put_port);
                                 self.store.write(ValueId::Stack(heap[stmt.to].unique_id_in_scope as u32), get_port);
                                 cur_frame.position = stmt.next;
                                 Ok(EvalContinuation::Stepping)
+                            }
+                        }
+                    }
+                }
             },
             Statement::Labeled(stmt) => {
                 cur_frame.position = stmt.body;
                 Ok(EvalContinuation::Stepping)
             },
             Statement::If(stmt) => {
                 debug_assert_eq!(cur_frame.expr_values.len(), 1, "expected one expr value for if statement");
                 let test_value = cur_frame.expr_values.pop_back().unwrap();
                 let test_value = self.store.maybe_read_ref(&test_value).as_bool();
                 if test_value {
@@ @@ -872,49 +873,49 @@ impl Prompt { @@
                 Ok(EvalContinuation::Stepping)
             },
             Statement::Continue(stmt) => {
                 cur_frame.position = stmt.target.unwrap().upcast();
                 Ok(EvalContinuation::Stepping)
             },
             Statement::Synchronous(stmt) => {
                 cur_frame.position = stmt.body.upcast();
                 Ok(EvalContinuation::SyncBlockStart)
             },
             Statement::EndSynchronous(stmt) => {
                 cur_frame.position = stmt.next;
                 Ok(EvalContinuation::SyncBlockEnd)
             },
             Statement::Fork(stmt) => {
                 if stmt.right_body.is_none() {
                     // No reason to fork
                     cur_frame.position = stmt.left_body.upcast();
                 } else {
                     // Need to fork
-                    if let Some(go_left) = ctx.get_fork() {
+                    if let Some(go_left) = ctx.performed_fork() {
                         // Runtime has created a fork
                         if go_left {
                             cur_frame.position = stmt.left_body.upcast();
                         } else {
                             cur_frame.position = stmt.right_body.unwrap().upcast();
+                        }
                     } else {
                         // Request the runtime to create a fork of the current
                         // branch
                         return Ok(EvalContinuation::NewFork);
+                    }
+                }
                 Ok(EvalContinuation::Stepping)
             },
             Statement::EndFork(stmt) => {
                 cur_frame.position = stmt.next;
                 Ok(EvalContinuation::Stepping)
+            }
             Statement::Return(_stmt) => {
                 debug_assert!(heap[cur_frame.definition].is_function());
                 debug_assert_eq!(cur_frame.expr_values.len(), 1, "expected one expr value for return statement");

src/protocol/mod.rs

➞

Show inline comments

@@ @@ -263,99 +263,99 @@ impl ProtocolDescription { @@
                 if let Value::Array(heap_pos) = argument {
                     let heap_pos = *heap_pos;
                     for element in &arguments.regions[heap_pos as usize] {
                         if !self.verify_same_type(expected, expected_idx + 1, arguments, element) {
                             return false;
+                        }
+                    }
                     return true;
                 } else {
                     return false;
+                }
             },
             CTP::Input => if let Value::Input(_) = argument { true } else { false },
             CTP::Output => if let Value::Output(_) = argument { true } else { false },
             CTP::Instance(_definition_id, _num_embedded) => {
                 todo!("implement full type checking on user-supplied arguments");
                 return false;
             },
+        }
+    }
+}
 // TODO: @temp Should just become a concrete thing that is passed in
 pub trait RunContext {
     fn did_put(&mut self, port: PortId) -> bool;
     fn get(&mut self, port: PortId) -> Option<ValueGroup>; // None if still waiting on message
     fn performed_put(&mut self, port: PortId) -> bool;
     fn performed_get(&mut self, port: PortId) -> Option<ValueGroup>; // None if still waiting on message
     fn fires(&mut self, port: PortId) -> Option<Value>; // None if not yet branched
     fn get_fork(&mut self) -> Option<bool>; // None if not yet forked
     fn get_channel(&mut self) -> Option<(Value, Value)>; // None if not yet prepared
     fn performed_fork(&mut self) -> Option<bool>; // None if not yet forked
     fn created_channel(&mut self) -> Option<(Value, Value)>; // None if not yet prepared
+}
 #[derive(Debug)]
 pub enum RunResult {
     // Can only occur outside sync blocks
     ComponentTerminated, // component has exited its procedure
     ComponentAtSyncStart,
     NewComponent(DefinitionId, i32, ValueGroup), // should also be possible inside sync
     NewChannel, // should also be possible inside sync
     // Can only occur inside sync blocks
     BranchInconsistent, // branch has inconsistent behaviour
     BranchMissingPortState(PortId), // branch doesn't know about port firing
-    BranchMissingPortValue(PortId), // branch hasn't received message on input port yet
+    BranchGet(PortId), // branch hasn't received message on input port yet
     BranchAtSyncEnd,
     BranchFork,
     BranchPut(PortId, ValueGroup),
+}
 impl ComponentState {
     pub(crate) fn run(&mut self, ctx: &mut impl RunContext, pd: &ProtocolDescription) -> RunResult {
         use EvalContinuation as EC;
         use RunResult as RR;
         loop {
             let step_result = self.prompt.step(&pd.types, &pd.heap, &pd.modules, ctx);
             match step_result {
                 Err(reason) => {
                     // TODO: @temp
                     println!("Evaluation error:\n{}", reason);
                     todo!("proper error handling/bubbling up");
                 },
                 Ok(continuation) => match continuation {
                     // TODO: Probably want to remove this translation
                     EC::Stepping => continue,
                     EC::Inconsistent => return RR::BranchInconsistent,
                     EC::Terminal => return RR::ComponentTerminated,
                     EC::SyncBlockStart => return RR::ComponentAtSyncStart,
                     EC::SyncBlockEnd => return RR::BranchAtSyncEnd,
                     EC::NewComponent(definition_id, monomorph_idx, args) =>
                         return RR::NewComponent(definition_id, monomorph_idx, args),
                     EC::NewChannel =>
                         return RR::NewChannel,
                     EC::NewFork =>
                         return RR::BranchFork,
                     EC::BlockFires(port_id) => return RR::BranchMissingPortState(port_id),
-                    EC::BlockGet(port_id) => return RR::BranchMissingPortValue(port_id),
+                    EC::BlockGet(port_id) => return RR::BranchGet(port_id),
                     EC::Put(port_id, value) => {
                         let value_group = ValueGroup::from_store(&self.prompt.store, &[value]);
                         return RR::BranchPut(port_id, value_group);
                     },
+                }
+            }
+        }
+    }
+}
 // TODO: @remove the old stuff
 impl ComponentState {
     pub(crate) fn nonsync_run<'a: 'b, 'b>(
         &'a mut self,
         context: &'b mut NonsyncProtoContext<'b>,
         pd: &'a ProtocolDescription,
     ) -> NonsyncBlocker {
         let mut context = EvalContext::Nonsync(context);
         loop {
             let result = self.prompt.step(&pd.types, &pd.heap, &pd.modules, &mut context);
             match result {
                 Err(err) => {
                     println!("Evaluation error:\n{}", err);
                     panic!("proper error handling when component fails");
@@ @@ -448,111 +448,111 @@ impl ComponentState { @@
                         match message {
                             Value::Null => {
                                 return SyncBlocker::Inconsistent;
                             },
                             Value::Message(heap_pos) => {
                                 // Create a copy of the payload
                                 let values = &self.prompt.store.heap_regions[heap_pos as usize].values;
                                 let mut bytes = Vec::with_capacity(values.len());
                                 for value in values {
                                     bytes.push(value.as_uint8());
+                                }
                                 payload = Payload(Arc::new(bytes));
+                            }
                             _ => unreachable!(),
+                        }
                         return SyncBlocker::PutMsg(port, payload);
+                    }
                 },
+            }
+        }
+    }
+}
 impl RunContext for EvalContext<'_> {
-    fn did_put(&mut self, port: PortId) -> bool {
+    fn performed_put(&mut self, port: PortId) -> bool {
         match self {
             EvalContext::None => unreachable!(),
             EvalContext::Nonsync(_) => unreachable!(),
             EvalContext::Sync(ctx) => {
                 ctx.did_put_or_get(port)
+            }
+        }
+    }
-    fn get(&mut self, port: PortId) -> Option<ValueGroup> {
+    fn performed_get(&mut self, port: PortId) -> Option<ValueGroup> {
         match self {
             EvalContext::None => unreachable!(),
             EvalContext::Nonsync(_) => unreachable!(),
             EvalContext::Sync(ctx) => {
                 let payload = ctx.read_msg(port);
                 if payload.is_none() {
                     return None;
+                }
                 let payload = payload.unwrap();
                 let mut transformed = Vec::with_capacity(payload.len());
                 for byte in payload.0.iter() {
                     transformed.push(Value::UInt8(*byte));
+                }
                 let value_group = ValueGroup{
                     values: vec![Value::Message(0)],
                     regions: vec![transformed],
                 };
                 return Some(value_group);
+            }
+        }
+    }
     fn fires(&mut self, port: PortId) -> Option<Value> {
         match self {
             EvalContext::None => unreachable!(),
             EvalContext::Nonsync(_) => unreachable!(),
             EvalContext::Sync(context) => {
                 match context.is_firing(port) {
                     Some(did_fire) => Some(Value::Bool(did_fire)),
                     None => None,
+                }
+            }
+        }
+    }
-    fn get_channel(&mut self) -> Option<(Value, Value)> {
+    fn created_channel(&mut self) -> Option<(Value, Value)> {
         match self {
             EvalContext::None => unreachable!(),
             EvalContext::Nonsync(context) => {
                 let [from, to] = context.new_port_pair();
                 let from = Value::Output(from);
                 let to = Value::Input(to);
                 return Some((from, to));
             },
             EvalContext::Sync(_) => unreachable!(),
+        }
+    }
-    fn get_fork(&mut self) -> Option<bool> {
+    fn performed_fork(&mut self) -> Option<bool> {
         // Never actually used in the old runtime
         return None;
+    }
+}
 // TODO: @remove once old runtime has disappeared
 impl EvalContext<'_> {
     // fn random(&mut self) -> LongValue {
     //     match self {
     //         // EvalContext::None => unreachable!(),
     //         EvalContext::Nonsync(_context) => todo!(),
     //         EvalContext::Sync(_) => unreachable!(),
     //     }
     // }
     fn new_component(&mut self, moved_ports: HashSet<PortId>, init_state: ComponentState) -> () {
         match self {
             EvalContext::None => unreachable!(),
             EvalContext::Nonsync(context) => {
                 context.new_component(moved_ports, init_state)
+            }
             EvalContext::Sync(_) => unreachable!(),
+        }
+    }
     fn new_channel(&mut self) -> [Value; 2] {

src/runtime2/branch.rs

➞

Show inline comments

@@ @@ -31,117 +31,130 @@ impl BranchId { @@
     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
+}
 #[derive(Debug)]
 pub(crate) enum PreparedStatement {
     CreatedChannel((Value, Value)),
     ForkedExecution(bool),
     PerformedPut,
     PerformedGet(ValueGroup),
     None,
+}
 impl PreparedStatement {
     pub(crate) fn is_none(&self) -> bool {
         if let PreparedStatement::None = self {
             return true;
         } else {
             return false;
+        }
+    }
     pub(crate) fn take(&mut self) -> PreparedStatement {
         if let PreparedStatement::None = self {
             return PreparedStatement::None;
         } else {
             let mut replacement = PreparedStatement::None;
             std::mem::swap(self, &mut replacement);
             return replacement;
+        }
+    }
+}
 /// 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?
     pub prepared_fork: Option<bool>, // TODO: See above
     pub prepared: PreparedStatement,
+}
 impl BranchListItem for Branch {
     #[inline] fn get_id(&self) -> BranchId { return self.id; }
     #[inline] fn set_next_id(&mut self, id: BranchId) { self.next_in_queue = id; }
     #[inline] fn get_next_id(&self) -> BranchId { return self.next_in_queue; }
+}
 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,
             prepared_fork: None,
             prepared: PreparedStatement::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());
         debug_assert!(parent_branch.prepared_fork.is_none());
         debug_assert!(parent_branch.prepared.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,
             prepared_fork: None,
             prepared: PreparedStatement::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();
+    }
+}
@@ @@ -274,50 +287,49 @@ impl ExecTree { @@
         let parent_branch = &self[parent_branch_id];
         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());
         // 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());
         debug_assert!(branch.prepared.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];
+    }
+}

src/runtime2/connector.rs

➞

Show inline comments

@@ @@ -12,120 +12,129 @@ @@
 // So currently the code is organized as following:
 // - The scheduler that is running the component is the authoritative source on
 //     ports during *non-sync* mode. The consensus algorithm is the
 //     authoritative source during *sync* mode. They retrieve each other's
 //     state during the transitions. Hence port data exists duplicated between
 //     these two datastructures.
 // - The execution tree is where executed branches reside. But the execution
 //     tree is only aware of the tree shape itself (and keeps track of some
 //     queues of branches that are in a particular state), and tends to store
 //     the PDL program state. The consensus algorithm is also somewhat aware
 //     of the execution tree, but only in terms of what is needed to complete
 //     a sync round (for now, that means the port mapping in each branch).
 //     Hence once more we have properties conceptually associated with branches
 //     in two places.
 // - TODO: Write about handling messages, consensus wrapping data
 // - TODO: Write about way information is exchanged between PDL/component and scheduler through ctx
 use std::collections::HashMap;
 use std::sync::atomic::AtomicBool;
 use crate::PortId;
 use crate::common::ComponentState;
 use crate::protocol::eval::{Prompt, Value, ValueGroup};
 use crate::protocol::{RunContext, RunResult};
 use crate::runtime2::branch::PreparedStatement;
 use super::branch::{BranchId, ExecTree, QueueKind, SpeculativeState};
 use super::consensus::{Consensus, Consistency, find_ports_in_value_group};
 use super::inbox::{DataMessage, DataContent, Message, SyncMessage, PublicInbox};
 use super::native::Connector;
 use super::port::{PortKind, PortIdLocal};
 use super::scheduler::{ComponentCtx, 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(Debug, 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,
     last_finished_handled: Option<BranchId>,
+}
 // TODO: Remove remaining fields once 'fires()' is removed from language.
 struct ConnectorRunContext<'a> {
     branch_id: BranchId,
     consensus: &'a Consensus,
     received: &'a HashMap<PortIdLocal, ValueGroup>,
     scheduler: SchedulerCtx<'a>,
     prepared_channel: Option<(Value, Value)>,
     prepared_fork: Option<bool>,
     prepared: PreparedStatement,
+}
 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 performed_put(&mut self, _port: PortId) -> bool {
         return match self.prepared.take() {
             PreparedStatement::None => false,
             PreparedStatement::PerformedPut => true,
             taken => unreachable!("prepared statement is '{:?}' during 'performed_put()'", taken)
         };
+    }
     fn 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 performed_get(&mut self, _port: PortId) -> Option<ValueGroup> {
         return match self.prepared.take() {
             PreparedStatement::None => None,
             PreparedStatement::PerformedGet(value) => Some(value),
             taken => unreachable!("prepared statement is '{:?}' during 'performed_get()'", taken),
         };
+    }
     fn fires(&mut self, port: PortId) -> Option<Value> {
         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();
     fn created_channel(&mut self) -> Option<(Value, Value)> {
         return match self.prepared.take() {
             PreparedStatement::None => None,
             PreparedStatement::CreatedChannel(ports) => Some(ports),
             taken => unreachable!("prepared statement is '{:?}' during 'created_channel)_'", taken),
         };
+    }
     fn get_fork(&mut self) -> Option<bool> {
         return self.prepared_fork.take();
     fn performed_fork(&mut self) -> Option<bool> {
         return match self.prepared.take() {
             PreparedStatement::None => None,
             PreparedStatement::ForkedExecution(path) => Some(path),
             taken => unreachable!("prepared statement is '{:?}' during 'performed_fork()'", taken),
         };
+    }
+}
 impl Connector for ConnectorPDL {
     fn run(&mut self, sched_ctx: SchedulerCtx, comp_ctx: &mut ComponentCtx) -> ConnectorScheduling {
         self.handle_new_messages(comp_ctx);
         if self.tree.is_in_sync() {
             // Run in sync mode
             let scheduling = self.run_in_sync_mode(sched_ctx, comp_ctx);
             // Handle any new finished branches
             let mut iter_id = self.last_finished_handled.or(self.tree.get_queue_first(QueueKind::FinishedSync));
             while let Some(branch_id) = iter_id {
                 iter_id = self.tree.get_queue_next(branch_id);
                 self.last_finished_handled = Some(branch_id);
                 if let Some(solution_branch_id) = self.consensus.handle_new_finished_sync_branch(branch_id, comp_ctx) {
                     // Actually found a solution
                     self.collapse_sync_to_solution_branch(solution_branch_id, comp_ctx);
                     return ConnectorScheduling::Immediate;
+                }
                 self.last_finished_handled = Some(branch_id);
@@ @@ -160,267 +169,253 @@ impl ConnectorPDL { @@
+        }
+    }
     pub fn handle_new_data_message(&mut self, message: DataMessage, ctx: &mut ComponentCtx) {
         // Go through all branches that are awaiting new messages and see if
         // there is one that can receive this message.
         if !self.consensus.handle_new_data_message(&message, ctx) {
             // Old message, so drop it
             return;
+        }
         let mut iter_id = self.tree.get_queue_first(QueueKind::AwaitingMessage);
         while let Some(branch_id) = iter_id {
             iter_id = self.tree.get_queue_next(branch_id);
             let branch = &self.tree[branch_id];
             if branch.awaiting_port != message.data_header.target_port { continue; }
             if !self.consensus.branch_can_receive(branch_id, &message) { continue; }
             // This branch can receive, so fork and given it the message
             let receiving_branch_id = self.tree.fork_branch(branch_id);
             self.consensus.notify_of_new_branch(branch_id, receiving_branch_id);
             let receiving_branch = &mut self.tree[receiving_branch_id];
             receiving_branch.insert_message(message.data_header.target_port, message.content.as_message().unwrap().clone());
             debug_assert!(receiving_branch.awaiting_port == message.data_header.target_port);
             receiving_branch.awaiting_port = PortIdLocal::new_invalid();
             receiving_branch.prepared = PreparedStatement::PerformedGet(message.content.as_message().unwrap().clone());
             self.consensus.notify_of_received_message(receiving_branch_id, &message);
             // And prepare the branch for running
             self.tree.push_into_queue(QueueKind::Runnable, receiving_branch_id);
+        }
+    }
     pub fn handle_new_sync_message(&mut self, message: SyncMessage, ctx: &mut ComponentCtx) {
         if let Some(solution_branch_id) = self.consensus.handle_new_sync_message(message, ctx) {
             self.collapse_sync_to_solution_branch(solution_branch_id, ctx);
+        }
+    }
     // --- Running code
     pub fn run_in_sync_mode(&mut self, sched_ctx: SchedulerCtx, comp_ctx: &mut ComponentCtx) -> ConnectorScheduling {
         // Check if we have any branch that needs running
         debug_assert!(self.tree.is_in_sync() && self.consensus.is_in_sync());
         let branch_id = self.tree.pop_from_queue(QueueKind::Runnable);
         if branch_id.is_none() {
             return ConnectorScheduling::NotNow;
+        }
         // Retrieve the branch and run it
         let branch_id = branch_id.unwrap();
         let branch = &mut self.tree[branch_id];
         let mut run_context = ConnectorRunContext{
             branch_id,
             consensus: &self.consensus,
             received: &branch.inbox,
             scheduler: sched_ctx,
             prepared_channel: branch.prepared_channel.take(),
             prepared_fork: branch.prepared_fork.take(),
             prepared: branch.prepared.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) => {
+            RunResult::BranchGet(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_message_received = false;
                     for message in comp_ctx.get_read_data_messages(port_id) {
                         if self.consensus.branch_can_receive(branch_id, &message) {
                             // This branch can receive the message, so we do the
                             // fork-and-receive dance
                             let receiving_branch_id = self.tree.fork_branch(branch_id);
                             let branch = &mut self.tree[receiving_branch_id];
                             branch.insert_message(port_id, message.content.as_message().unwrap().clone());
                             self.consensus.notify_of_new_branch(branch_id, receiving_branch_id);
                             self.consensus.notify_of_received_message(receiving_branch_id, &message);
                             self.tree.push_into_queue(QueueKind::Runnable, receiving_branch_id);
                             any_message_received = true;
+                        }
+                    }
                     if any_message_received {
                         return ConnectorScheduling::Immediate;
                 branch.sync_state = SpeculativeState::HaltedAtBranchPoint;
                 branch.awaiting_port = port_id;
                 self.tree.push_into_queue(QueueKind::AwaitingMessage, branch_id);
                 // Note: we only know that a branch is waiting on a message when
                 // it reaches the `get` call. But we might have already received
                 // a message that targets this branch, so check now.
                 let mut any_message_received = false;
                 for message in comp_ctx.get_read_data_messages(port_id) {
                     if self.consensus.branch_can_receive(branch_id, &message) {
                         // This branch can receive the message, so we do the
                         // fork-and-receive dance
                         let receiving_branch_id = self.tree.fork_branch(branch_id);
                         let branch = &mut self.tree[receiving_branch_id];
                         branch.awaiting_port = PortIdLocal::new_invalid();
                         branch.prepared = PreparedStatement::PerformedGet(message.content.as_message().unwrap().clone());
                         self.consensus.notify_of_new_branch(branch_id, receiving_branch_id);
                         self.consensus.notify_of_received_message(receiving_branch_id, &message);
                         self.tree.push_into_queue(QueueKind::Runnable, receiving_branch_id);
                         any_message_received = true;
+                    }
                 } else {
                     branch.sync_state = SpeculativeState::Inconsistent;
+                }
                 if any_message_received {
                     return ConnectorScheduling::Immediate;
+                }
+            }
             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 {
                     branch.sync_state = SpeculativeState::Inconsistent;
+                }
             },
             RunResult::BranchFork => {
                 // Like the `NewChannel` result. This means we're setting up
                 // a branch and putting a marker inside the RunContext for the
                 // next time we run the PDL code
                 let left_id = branch_id;
                 let right_id = self.tree.fork_branch(left_id);
                 self.consensus.notify_of_new_branch(left_id, right_id);
                 self.tree.push_into_queue(QueueKind::Runnable, left_id);
                 self.tree.push_into_queue(QueueKind::Runnable, right_id);
                 let left_branch = &mut self.tree[left_id];
-                left_branch.prepared_fork = Some(true);
+                left_branch.prepared = PreparedStatement::ForkedExecution(true);
                 let right_branch = &mut self.tree[right_id];
-                right_branch.prepared_fork = Some(false);
+                right_branch.prepared = PreparedStatement::ForkedExecution(false);
+            }
             RunResult::BranchPut(port_id, content) => {
                 // Branch is attempting to send data
                 let port_id = PortIdLocal::new(port_id.0.u32_suffix);
                 let consistency = self.consensus.notify_of_speculative_mapping(branch_id, port_id, true);
                 if consistency == Consistency::Valid {
                     // `put()` is valid.
                     let (sync_header, data_header) = self.consensus.handle_message_to_send(branch_id, port_id, &content, comp_ctx);
                     comp_ctx.submit_message(Message::Data(DataMessage {
                         sync_header, data_header,
                         content: DataContent::Message(content),
                     }));
                     self.tree.push_into_queue(QueueKind::Runnable, branch_id);
                     return ConnectorScheduling::Immediate;
                 } else {
                     branch.sync_state = SpeculativeState::Inconsistent;
+                }
                 let (sync_header, data_header) = self.consensus.handle_message_to_send(branch_id, port_id, &content, comp_ctx);
                 comp_ctx.submit_message(Message::Data(DataMessage {
                     sync_header, data_header,
                     content: DataContent::Message(content),
                 }));
                 branch.prepared = PreparedStatement::PerformedPut;
                 self.tree.push_into_queue(QueueKind::Runnable, branch_id);
                 return ConnectorScheduling::Immediate;
             },
             _ => unreachable!("unexpected run result {:?} in sync mode", run_result),
+        }
         // If here then the run result did not require a particular action. We
         // return whether we have more active branches to run or not.
         if self.tree.queue_is_empty(QueueKind::Runnable) {
             return ConnectorScheduling::NotNow;
         } else {
             return ConnectorScheduling::Later;
+        }
+    }
     pub fn run_in_deterministic_mode(&mut self, sched_ctx: SchedulerCtx, comp_ctx: &mut ComponentCtx) -> ConnectorScheduling {
         debug_assert!(!self.tree.is_in_sync() && !self.consensus.is_in_sync());
         let branch = self.tree.base_branch_mut();
         debug_assert!(branch.sync_state == SpeculativeState::RunningNonSync);
         let mut run_context = ConnectorRunContext{
             branch_id: branch.id,
             consensus: &self.consensus,
             received: &branch.inbox,
             scheduler: sched_ctx,
             prepared_channel: branch.prepared_channel.take(),
             prepared_fork: branch.prepared_fork.take(),
             prepared: branch.prepared.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();
                 debug_assert!(self.last_finished_handled.is_none());
                 self.consensus.start_sync(comp_ctx);
                 self.consensus.notify_of_new_branch(BranchId::new_invalid(), sync_branch_id);
                 self.tree.push_into_queue(QueueKind::Runnable, sync_branch_id);
                 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((
+                branch.prepared = PreparedStatement::CreatedChannel((
                     Value::Output(PortId::new(putter.self_id.index)),
                     Value::Input(PortId::new(getter.self_id.index)),
                 ));
                 comp_ctx.push_port(putter);
                 comp_ctx.push_port(getter);
                 return ConnectorScheduling::Immediate;
             },
             _ => unreachable!("unexpected run result '{:?}' while running in non-sync mode", run_result),
+        }
+    }
     pub fn collapse_sync_to_solution_branch(&mut self, solution_branch_id: BranchId, ctx: &mut ComponentCtx) {
         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(&[]);

src/runtime2/consensus.rs

➞

Show inline comments

 use crate::collections::VecSet;
 use crate::protocol::eval::ValueGroup;
 use crate::runtime2::inbox::BranchMarker;
 use super::ConnectorId;
 use super::branch::BranchId;
 use super::port::{ChannelId, PortIdLocal};
 use super::inbox::{
     Message, PortAnnotation,
     DataMessage, DataContent, DataHeader,
     SyncMessage, SyncContent, SyncHeader,
 };
 use super::scheduler::ComponentCtx;
 struct BranchAnnotation {
     port_mapping: Vec<PortAnnotation>,
     cur_marker: BranchMarker,
+}
 #[derive(Debug)]
 pub(crate) struct LocalSolution {
     component: ConnectorId,
     final_branch_id: BranchId,
-    port_mapping: Vec<(ChannelId, BranchId)>,
+    port_mapping: Vec<(ChannelId, BranchMarker)>,
+}
 #[derive(Debug, Clone)]
 pub(crate) struct GlobalSolution {
     component_branches: Vec<(ConnectorId, BranchId)>,
-    channel_mapping: Vec<(ChannelId, BranchId)>, // TODO: This can go, is debugging info
+    channel_mapping: Vec<(ChannelId, BranchMarker)>, // TODO: This can go, is debugging info
+}
 // -----------------------------------------------------------------------------
 // Consensus
 // -----------------------------------------------------------------------------
 struct Peer {
     id: ConnectorId,
     encountered_this_round: bool,
     expected_sync_round: u32,
+}
 /// The consensus algorithm. Currently only implemented to find the component
 /// with the highest ID within the sync region and letting it handle all the
 /// local solutions.
 ///
 /// The type itself serves as an experiment to see how code should be organized.
 // TODO: Flatten all datastructures
 // TODO: Have a "branch+port position hint" in case multiple operations are
 //  performed on the same port to prevent repeated lookups
 // TODO: A lot of stuff should be batched. Like checking all the sync headers
 //  and sending "I have a higher ID" messages. Should reduce locking by quite a
 //  bit.
 pub(crate) struct Consensus {
     // --- State that is cleared after each round
     // Local component's state
     highest_connector_id: ConnectorId,
     branch_annotations: Vec<BranchAnnotation>,
     branch_annotations: Vec<BranchAnnotation>, // index is branch ID
     branch_markers: Vec<BranchId>, // index is branch marker, maps to branch
     // Gathered state from communication
     encountered_ports: VecSet<PortIdLocal>, // to determine if we should send "port remains silent" messages.
     solution_combiner: SolutionCombiner,
     // --- Persistent state
     peers: Vec<Peer>,
     sync_round: u32,
     // --- Workspaces
     workspace_ports: Vec<PortIdLocal>,
+}
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub(crate) enum Consistency {
     Valid,
     Inconsistent,
+}
 impl Consensus {
     pub fn new() -> Self {
         return Self {
             highest_connector_id: ConnectorId::new_invalid(),
             branch_annotations: Vec::new(),
             branch_markers: Vec::new(),
             encountered_ports: VecSet::new(),
             solution_combiner: SolutionCombiner::new(),
             peers: Vec::new(),
             sync_round: 0,
             workspace_ports: Vec::new(),
+        }
+    }
     // --- Controlling sync round and branches
     /// Returns whether the consensus algorithm is running in sync mode
     pub fn is_in_sync(&self) -> bool {
         return !self.branch_annotations.is_empty();
+    }
     /// TODO: Remove this once multi-fire is in place
     pub fn get_annotation(&self, branch_id: BranchId, port_id: PortIdLocal) -> &PortAnnotation {
         let branch = &self.branch_annotations[branch_id.index as usize];
         let port = branch.port_mapping.iter().find(|v| v.port_id == port_id).unwrap();
         return port;
+    }
     /// Sets up the consensus algorithm for a new synchronous round. The
     /// provided ports should be the ports the component owns at the start of
     /// the sync round.
     pub fn start_sync(&mut self, ctx: &ComponentCtx) {
         debug_assert!(!self.highest_connector_id.is_valid());
         debug_assert!(self.branch_annotations.is_empty());
         debug_assert!(self.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(),
             cur_marker: BranchMarker::new_invalid(),
         });
         self.branch_markers.push(BranchId::new_invalid());
         self.highest_connector_id = ctx.id;
+    }
     /// Notifies the consensus algorithm that a new branch has appeared. Must be
     /// called for each forked branch in the execution tree.
     pub fn notify_of_new_branch(&mut self, parent_branch_id: BranchId, new_branch_id: BranchId) {
         // If called correctly. Then each time we are notified the new branch's
         // index is the length in `branch_annotations`.
         debug_assert!(self.branch_annotations.len() == new_branch_id.index as usize);
         let parent_branch_annotations = &self.branch_annotations[parent_branch_id.index as usize];
         let new_marker = BranchMarker::new(self.branch_markers.len() as u32);
         let new_branch_annotations = BranchAnnotation{
             port_mapping: parent_branch_annotations.port_mapping.clone(),
             cur_marker: new_marker,
         };
         self.branch_annotations.push(new_branch_annotations);
         self.branch_markers.push(new_branch_id);
+    }
     /// Notifies the consensus algorithm that a 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;
+    }
@@ @@ -193,58 +202,58 @@ impl Consensus { @@
     pub(crate) fn handle_new_finished_sync_branch(&mut self, branch_id: BranchId, ctx: &mut ComponentCtx) -> Option<BranchId> {
         // Turn the port mapping into a local solution
         let source_mapping = &self.branch_annotations[branch_id.index as usize].port_mapping;
         let mut target_mapping = Vec::with_capacity(source_mapping.len());
         for port in source_mapping {
             // Note: if the port is silent, and we've never communicated
             // over the port, then we need to do so now, to let the peer
             // component know about our sync leader state.
             let port_desc = ctx.get_port_by_id(port.port_id).unwrap();
             let peer_port_id = port_desc.peer_id;
             let channel_id = port_desc.channel_id;
             if !self.encountered_ports.contains(&port.port_id) {
                 ctx.submit_message(Message::Data(DataMessage {
                     sync_header: SyncHeader{
                         sending_component_id: ctx.id,
                         highest_component_id: self.highest_connector_id,
                         sync_round: self.sync_round
                     },
                     data_header: DataHeader{
                         expected_mapping: source_mapping.clone(),
                         sending_port: port.port_id,
                         target_port: peer_port_id,
-                        new_mapping: BranchId::new_invalid(),
+                        new_mapping: BranchMarker::new_invalid(),
                     },
                     content: DataContent::SilentPortNotification,
                 }));
                 self.encountered_ports.push(port.port_id);
+            }
             target_mapping.push((
                 channel_id,
-                port.registered_id.unwrap_or(BranchId::new_invalid())
+                port.registered_id.unwrap_or(BranchMarker::new_invalid())
             ));
+        }
         let local_solution = LocalSolution{
             component: ctx.id,
             final_branch_id: branch_id,
             port_mapping: target_mapping,
         };
         let solution_branch = self.send_or_store_local_solution(local_solution, ctx);
         return solution_branch;
+    }
     /// Notifies the consensus algorithm about the chosen branch to commit to
     /// memory.
     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);
+        }
@@ @@ -269,65 +278,69 @@ impl Consensus { @@
     /// sending the message is consistent with the speculative state.
     pub fn handle_message_to_send(&mut self, branch_id: BranchId, source_port_id: PortIdLocal, content: &ValueGroup, ctx: &mut ComponentCtx) -> (SyncHeader, DataHeader) {
         debug_assert!(self.is_in_sync());
         let branch = &mut self.branch_annotations[branch_id.index as usize];
         if cfg!(debug_assertions) {
             // Check for consistent mapping
             let port = branch.port_mapping.iter()
                 .find(|v| v.port_id == source_port_id)
                 .unwrap();
             debug_assert!(port.expected_firing == None || port.expected_firing == Some(true));
+        }
         // Check for ports that are being sent
         debug_assert!(self.workspace_ports.is_empty());
         find_ports_in_value_group(content, &mut self.workspace_ports);
         if !self.workspace_ports.is_empty() {
             todo!("handle sending ports");
             self.workspace_ports.clear();
+        }
         // Construct data header
         // TODO: Handle multiple firings. Right now we just assign the current
         //  branch to the `None` value because we know we can only send once.
         debug_assert!(branch.port_mapping.iter().find(|v| v.port_id == source_port_id).unwrap().registered_id.is_none());
         let port_info = ctx.get_port_by_id(source_port_id).unwrap();
         let data_header = DataHeader{
             expected_mapping: branch.port_mapping.clone(),
             sending_port: port_info.self_id,
             target_port: port_info.peer_id,
-            new_mapping: branch_id
+            new_mapping: branch.cur_marker,
         };
         // 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);
+                mapping.registered_id = Some(branch.cur_marker);
+            }
+        }
         // Update branch marker
         let new_marker = BranchMarker::new(self.branch_markers.len() as u32);
         branch.cur_marker = new_marker;
         self.branch_markers.push(branch_id);
         self.encountered_ports.push(source_port_id);
         return (self.create_sync_header(ctx), data_header);
+    }
     /// Handles a new data message by handling the sync header. The caller is
     /// responsible for checking for branches that might be able to receive
     /// the message.
     pub fn handle_new_data_message(&mut self, message: &DataMessage, ctx: &mut ComponentCtx) -> bool {
         return self.handle_received_sync_header(&message.sync_header, ctx)
+    }
     /// Handles a new sync message by handling the sync header and the contents
     /// of the message. Returns `Some` with the branch ID of the global solution
     /// if the sync solution has been found.
     pub fn handle_new_sync_message(&mut self, message: SyncMessage, ctx: &mut ComponentCtx) -> Option<BranchId> {
         if !self.handle_received_sync_header(&message.sync_header, ctx) {
             return None;
+        }
         // And handle the contents
         debug_assert_eq!(message.target_component_id, ctx.id);
         match message.content {
             SyncContent::Notification => {
@@ @@ -525,49 +538,49 @@ impl Consensus { @@
     fn forward_local_solutions(&mut self, ctx: &mut ComponentCtx) {
         debug_assert_ne!(self.highest_connector_id, ctx.id);
         for local_solution in self.solution_combiner.drain() {
             let message = SyncMessage {
                 sync_header: self.create_sync_header(ctx),
                 target_component_id: self.highest_connector_id,
                 content: SyncContent::LocalSolution(local_solution),
             };
             ctx.submit_message(Message::Sync(message));
+        }
+    }
+}
 // -----------------------------------------------------------------------------
 // Solution storage and algorithms
 // -----------------------------------------------------------------------------
 // TODO: Remove all debug derives
 #[derive(Debug)]
 struct MatchedLocalSolution {
     final_branch_id: BranchId,
-    channel_mapping: Vec<(ChannelId, BranchId)>,
+    channel_mapping: Vec<(ChannelId, BranchMarker)>,
     matches: Vec<ComponentMatches>,
+}
 #[derive(Debug)]
 struct ComponentMatches {
     target_id: ConnectorId,
     target_index: usize,
     match_indices: Vec<usize>, // of local solution in connector
+}
 #[derive(Debug)]
 struct ComponentPeer {
     target_id: ConnectorId,
     target_index: usize, // in array of global solution components
     involved_channels: Vec<ChannelId>,
+}
 #[derive(Debug)]
 struct ComponentLocalSolutions {
     component: ConnectorId,
     peers: Vec<ComponentPeer>,
     solutions: Vec<MatchedLocalSolution>,
     all_peers_present: bool,
+}

src/runtime2/inbox.rs

➞

Show inline comments

 use std::sync::Mutex;
 use std::collections::VecDeque;
 use crate::protocol::eval::ValueGroup;
 use super::ConnectorId;
 use super::branch::BranchId;
 use super::consensus::{GlobalSolution, LocalSolution};
 use super::port::PortIdLocal;
 // TODO: Remove Debug derive from all types
 #[derive(Debug, Copy, Clone)]
 pub(crate) struct PortAnnotation {
     pub port_id: PortIdLocal,
-    pub registered_id: Option<BranchId>,
+    pub registered_id: Option<BranchMarker>,
     pub expected_firing: Option<bool>,
+}
 /// Marker for a branch in a port mapping. A marker is, like a branch ID, a
 /// unique identifier for a branch, but differs in that a branch only has one
 /// branch ID, but might have multiple associated markers (i.e. one branch
 /// performing a `put` three times will generate three markers.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub(crate) struct BranchMarker{
     marker: u32,
+}
 impl BranchMarker {
     #[inline]
     pub(crate) fn new(marker: u32) -> Self {
         debug_assert!(marker != 0);
         return Self{ marker };
+    }
     #[inline]
     pub(crate) fn new_invalid() -> Self {
         return Self{ marker: 0 }
+    }
+}
 /// The header added by the synchronization algorithm to all.
 #[derive(Debug, Clone)]
 pub(crate) struct SyncHeader {
     pub sending_component_id: ConnectorId,
     pub highest_component_id: ConnectorId,
     pub sync_round: u32,
+}
 /// The header added to data messages
 #[derive(Debug, Clone)]
 pub(crate) struct DataHeader {
     pub expected_mapping: Vec<PortAnnotation>,
     pub sending_port: PortIdLocal,
     pub target_port: PortIdLocal,
-    pub new_mapping: BranchId,
+    pub new_mapping: BranchMarker,
+}
 // 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,

src/runtime2/native.rs

➞

Show inline comments

@@ @@ -162,96 +162,88 @@ impl ConnectorApplication { @@
+        }
         let branch_id = branch_id.unwrap();
         let branch = &mut self.tree[branch_id];
         let mut instruction_idx = self.branch_extra[branch_id.index as usize];
         if instruction_idx >= self.sync_desc.len() {
             // Performed last instruction, so this branch is officially at the
             // end of the synchronous interaction.
             let consistency = self.consensus.notify_of_finished_branch(branch_id);
             if consistency == Consistency::Valid {
                 branch.sync_state = SpeculativeState::ReachedSyncEnd;
                 self.tree.push_into_queue(QueueKind::FinishedSync, branch_id);
             } else {
                 branch.sync_state = SpeculativeState::Inconsistent;
+            }
         } else {
             // We still have instructions to perform
             let cur_instruction = &self.sync_desc[instruction_idx];
             self.branch_extra[branch_id.index as usize] += 1;
             match &cur_instruction {
                 ApplicationSyncAction::Put(port_id, content) => {
                     let port_id = *port_id;
                     let consistency = self.consensus.notify_of_speculative_mapping(branch_id, port_id, true);
                     if consistency == Consistency::Valid {
                         let (sync_header, data_header) = self.consensus.handle_message_to_send(branch_id, port_id, &content, comp_ctx);
                         let message = Message::Data(DataMessage {
                             sync_header,
                             data_header,
                             content: DataContent::Message(content.clone()),
                         });
                         comp_ctx.submit_message(message);
                         self.tree.push_into_queue(QueueKind::Runnable, branch_id);
                         return ConnectorScheduling::Immediate;
                     } else {
                         branch.sync_state = SpeculativeState::Inconsistent;
+                    }
                     let (sync_header, data_header) = self.consensus.handle_message_to_send(branch_id, port_id, &content, comp_ctx);
                     let message = Message::Data(DataMessage {
                         sync_header,
                         data_header,
                         content: DataContent::Message(content.clone()),
                     });
                     comp_ctx.submit_message(message);
                     self.tree.push_into_queue(QueueKind::Runnable, branch_id);
                     return ConnectorScheduling::Immediate;
                 },
                 ApplicationSyncAction::Get(port_id) => {
                     let port_id = *port_id;
                     let consistency = self.consensus.notify_of_speculative_mapping(branch_id, port_id, true);
                     if consistency == Consistency::Valid {
                         branch.sync_state = SpeculativeState::HaltedAtBranchPoint;
                         branch.awaiting_port = port_id;
                         self.tree.push_into_queue(QueueKind::AwaitingMessage, branch_id);
                         let mut any_message_received = false;
                         for message in comp_ctx.get_read_data_messages(port_id) {
                             if self.consensus.branch_can_receive(branch_id, &message) {
                                 // This branch can receive the message, so we do the
                                 // fork-and-receive dance
                                 let receiving_branch_id = self.tree.fork_branch(branch_id);
                                 let branch = &mut self.tree[receiving_branch_id];
                                 debug_assert!(receiving_branch_id.index as usize == self.branch_extra.len());
                                 self.branch_extra.push(instruction_idx + 1);
                                 branch.insert_message(port_id, message.content.as_message().unwrap().clone());
                                 self.consensus.notify_of_new_branch(branch_id, receiving_branch_id);
                                 self.consensus.notify_of_received_message(receiving_branch_id, &message);
                                 self.tree.push_into_queue(QueueKind::Runnable, receiving_branch_id);
                                 any_message_received = true;
+                            }
+                        }
                         if any_message_received {
                             return ConnectorScheduling::Immediate;
                     branch.sync_state = SpeculativeState::HaltedAtBranchPoint;
                     branch.awaiting_port = port_id;
                     self.tree.push_into_queue(QueueKind::AwaitingMessage, branch_id);
                     let mut any_message_received = false;
                     for message in comp_ctx.get_read_data_messages(port_id) {
                         if self.consensus.branch_can_receive(branch_id, &message) {
                             // This branch can receive the message, so we do the
                             // fork-and-receive dance
                             let receiving_branch_id = self.tree.fork_branch(branch_id);
                             let branch = &mut self.tree[receiving_branch_id];
                             debug_assert!(receiving_branch_id.index as usize == self.branch_extra.len());
                             self.branch_extra.push(instruction_idx + 1);
                             branch.insert_message(port_id, message.content.as_message().unwrap().clone());
                             self.consensus.notify_of_new_branch(branch_id, receiving_branch_id);
                             self.consensus.notify_of_received_message(receiving_branch_id, &message);
                             self.tree.push_into_queue(QueueKind::Runnable, receiving_branch_id);
                             any_message_received = true;
+                        }
                     } else {
                         branch.sync_state = SpeculativeState::Inconsistent;
+                    }
                     if any_message_received {
                         return ConnectorScheduling::Immediate;
+                    }
+                }
+            }
+        }
         if self.tree.queue_is_empty(QueueKind::Runnable) {
             return ConnectorScheduling::NotNow;
         } else {
             return ConnectorScheduling::Later;
+        }
+    }
     fn run_in_deterministic_mode(&mut self, _sched_ctx: SchedulerCtx, comp_ctx: &mut ComponentCtx) -> ConnectorScheduling {
         debug_assert!(!self.is_in_sync);
         // In non-sync mode the application component doesn't really do anything
         // except performing jobs submitted from the API. This is the only
         // case where we expect to be woken up.
         // Note that we have to communicate to the scheduler when we've received
         // ports or created components (hence: given away ports) *before* we
         // enter a sync round.
         let mut queue = self.job_queue.lock().unwrap();
         while let Some(job) = queue.pop_front() {
             match job {

src/runtime2/scheduler.rs

➞

Show inline comments

@@ @@ -327,53 +327,53 @@ impl Scheduler { @@
                 self.runtime.push_work(connector_key)
+            }
+        }
+    }
     #[inline]
     fn get_message_target_port(message: &Message) -> Option<PortIdLocal> {
         match message {
             Message::Data(data) => return Some(data.data_header.target_port),
             Message::Sync(_) => {},
             Message::Control(control) => {
                 match &control.content {
                     ControlContent::PortPeerChanged(port_id, _) => return Some(*port_id),
                     ControlContent::CloseChannel(port_id) => return Some(*port_id),
                     ControlContent::Ping | ControlContent::Ack => {},
+                }
             },
+        }
         return None
+    }
     // TODO: Remove, this is debugging stuff
     fn debug(&self, message: &str) {
         println!("DEBUG [thrd:{:02} conn:  ]: {}", self.scheduler_id, message);
+        // 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);
+        // 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.

src/runtime2/tests/mod.rs

➞

Show inline comments

 mod network_shapes;
 mod api_component;
 mod speculation_basic;
 mod basics;
 use super::*;
 use crate::{PortId, ProtocolDescription};
 use crate::common::Id;
 use crate::protocol::eval::*;
 use crate::runtime2::native::{ApplicationSyncAction};
 //
 // Generic testing constants, use when appropriate to simplify stress-testing
 pub(crate) const NUM_THREADS: u32 = 1;     // number of threads in runtime
 pub(crate) const NUM_INSTANCES: u32 = 1;   // number of test instances constructed
 pub(crate) const NUM_LOOPS: u32 = 1;       // number of loops within a single test (not used by all tests)
 pub(crate) const NUM_THREADS: u32 = 3;     // number of threads in runtime
 pub(crate) const NUM_INSTANCES: u32 = 7;   // number of test instances constructed
 pub(crate) const NUM_LOOPS: u32 = 8;       // number of loops within a single test (not used by all tests)
 fn create_runtime(pdl: &str) -> Runtime {
     let protocol = ProtocolDescription::parse(pdl.as_bytes()).expect("parse pdl");
     let runtime = Runtime::new(NUM_THREADS, protocol);
     return runtime;
+}
 fn run_test_in_runtime<F: Fn(&mut ApplicationInterface)>(pdl: &str, constructor: F) {
     let protocol = ProtocolDescription::parse(pdl.as_bytes())
         .expect("parse PDL");
     let runtime = Runtime::new(NUM_THREADS, protocol);
     let mut api = runtime.create_interface();
     for _ in 0..NUM_INSTANCES {
         constructor(&mut api);
+    }
+}
 pub(crate) struct TestTimer {
     name: &'static str,
     started: std::time::Instant
+}

0 comments (0 inline, 0 general)