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

Changeset - 98aadfccbafd

Parent rev.

Child rev.

[Not reviewed]

0 7 0

MH - 4 years ago 2021-10-22 21:01:38
contact@maxhenger.nl

solving problem of connectors shutting down

7 files changed with 164 insertions and 73 deletions:

src/common.rs

src/runtime2/connector.rs

src/runtime2/mod.rs

src/runtime2/native.rs

src/runtime2/port.rs

src/runtime2/scheduler.rs

src/runtime2/tests/mod.rs

0 comments (0 inline, 0 general)

src/common.rs

➞

Show inline comments

@@ @@ -33,48 +33,59 @@ pub type ConnectorId = u32; @@
 pub type U32Suffix = u32;
 #[derive(Copy, Clone, Eq, PartialEq, Ord, Hash, PartialOrd)]
 /// Generalization of a port/component identifier
 #[derive(serde::Serialize, serde::Deserialize)]
 #[repr(C)]
 pub struct Id {
     pub(crate) connector_id: ConnectorId,
     pub(crate) u32_suffix: U32Suffix,
+}
 #[derive(Clone, Debug, Default)]
 pub struct U32Stream {
     next: u32,
+}
 /// Identifier of a component in a session
 #[derive(Copy, Clone, Eq, PartialEq, Ord, Hash, PartialOrd, serde::Serialize, serde::Deserialize)]
 pub struct ComponentId(Id); // PUB because it can be returned by errors
 /// Identifier of a port in a session
 #[derive(Copy, Clone, Eq, PartialEq, Ord, Hash, PartialOrd, serde::Serialize, serde::Deserialize)]
 #[repr(transparent)]
 pub struct PortId(pub(crate) Id);
 impl PortId {
     // TODO: Remove concept of ComponentId and PortId in this file
     #[deprecated]
     pub fn new(port: u32) -> Self {
         return PortId(Id{
             connector_id: u32::MAX,
             u32_suffix: port,
         });
+    }
+}
 /// A safely aliasable heap-allocated payload of message bytes
 #[derive(Default, Eq, PartialEq, Clone, Ord, PartialOrd)]
 pub struct Payload(pub Arc<Vec<u8>>);
 #[derive(Debug, Eq, PartialEq, Clone, Hash, Copy, Ord, PartialOrd)]
 /// "Orientation" of a port, determining whether they can send or receive messages with `put` and `get` respectively.
 #[repr(C)]
 #[derive(serde::Serialize, serde::Deserialize)]
 pub enum Polarity {
     Putter, // output port (from the perspective of the component)
     Getter, // input port (from the perspective of the component)
+}
 #[derive(Debug, Eq, PartialEq, Clone, Hash, Copy, Ord, PartialOrd)]
 /// "Orientation" of a transport-layer network endpoint, dictating how it's connection procedure should
 /// be conducted. Corresponds with connect() / accept() familiar to TCP socket programming.
 #[repr(C)]
 pub enum EndpointPolarity {
     Active,  // calls connect()
     Passive, // calls bind() listen() accept()
+}
 #[derive(Debug, Clone)]
 pub(crate) enum NonsyncBlocker {

src/runtime2/connector.rs

➞

Show inline comments

@@ @@ -38,82 +38,86 @@ impl BranchId { @@
     #[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
+}
 pub(crate) struct Branch {
     index: BranchId,
     parent_index: BranchId,
     // Code execution state
     code_state: ComponentState,
     prepared_channel: Option<(Value, Value)>,
     sync_state: SpeculativeState,
     next_branch_in_queue: Option<u32>,
     // Message/port state
     received: HashMap<PortIdLocal, DataMessage>, // TODO: @temporary, remove together with fires()
     ports_delta: Vec<PortOwnershipDelta>,
+}
 impl Branch {
     /// Constructs a non-sync branch. It is assumed that the code is at the
     /// first instruction
     pub(crate) fn new_initial_branch(component_state: ComponentState) -> Self {
         Branch{
             index: BranchId::new_invalid(),
             parent_index: BranchId::new_invalid(),
             code_state: component_state,
             prepared_channel: None,
             sync_state: SpeculativeState::RunningNonSync,
             next_branch_in_queue: None,
             received: HashMap::new(),
             ports_delta: Vec::new(),
+        }
+    }
     /// Constructs a sync branch. The provided branch is assumed to be the
     /// parent of the new branch within the execution tree.
     fn new_sync_branching_from(new_index: u32, parent_branch: &Branch) -> Self {
         debug_assert!(
             (parent_branch.sync_state == SpeculativeState::RunningNonSync && !parent_branch.parent_index.is_valid()) ||
             (parent_branch.sync_state == SpeculativeState::HaltedAtBranchPoint)
         );
         debug_assert!(parent_branch.prepared_channel.is_none());
         Branch{
             index: BranchId::new(new_index),
             parent_index: parent_branch.index,
             code_state: parent_branch.code_state.clone(),
             prepared_channel: None,
             sync_state: SpeculativeState::RunningInSync,
             next_branch_in_queue: None,
             received: parent_branch.received.clone(),
             ports_delta: parent_branch.ports_delta.clone(),
+        }
+    }
     fn commit_to_sync(&mut self) {
         self.index = BranchId::new(0);
         self.parent_index = BranchId::new_invalid();
         self.sync_state = SpeculativeState::RunningNonSync;
         self.next_branch_in_queue = None;
         self.received.clear();
         self.ports_delta.clear();
+    }
+}
 #[derive(Clone)]
 struct PortAssignment {
     is_assigned: bool,
     last_registered_branch_id: BranchId, // invalid branch ID implies not assigned yet
     num_times_fired: u32,
+}
@@ @@ -247,48 +251,49 @@ impl ConnectorPorts { @@
     fn get_port(&self, branch_idx: u32, port_idx: usize) -> &PortAssignment {
         let mapped_idx = self.mapped_index(branch_idx, port_idx);
         return &self.port_mapping[mapped_idx];
+    }
     #[inline]
     fn get_port_mut(&mut self, branch_idx: u32, port_idx: usize) -> &mut PortAssignment {
         let mapped_idx = self.mapped_index(branch_idx, port_idx);
         return &mut self.port_mapping[mapped_idx];
+    }
     #[inline]
     fn num_ports(&self) -> usize {
         return self.owned_ports.len();
+    }
     // Function for internal use: retrieve index in flattened port mapping array
     // based on branch/port index.
     #[inline]
     fn mapped_index(&self, branch_idx: u32, port_idx: usize) -> usize {
         let branch_idx = branch_idx as usize;
         let num_ports = self.owned_ports.len();
         println!("port_idx = {}, branch_idx = {}, num_ports = {}, port_mapping.len() = {}", port_idx, branch_idx, num_ports, self.port_mapping.len());
         debug_assert!(port_idx < num_ports);
         debug_assert!((branch_idx + 1) * num_ports <= self.port_mapping.len());
         return branch_idx * num_ports + port_idx;
+    }
+}
 struct BranchQueue {
     first: u32,
     last: u32,
+}
 impl BranchQueue {
     #[inline]
     fn new() -> Self {
         Self{ first: 0, last: 0 }
+    }
     #[inline]
     fn is_empty(&self) -> bool {
         debug_assert!((self.first == 0) == (self.last == 0));
         return self.first == 0;
+    }
@@ @@ -312,163 +317,164 @@ impl ConnectorPublic { @@
             inbox: PublicInbox::new(),
             sleeping: AtomicBool::new(initialize_as_sleeping),
+        }
+    }
+}
 // TODO: Maybe prevent false sharing by aligning `public` to next cache line.
 // TODO: Do this outside of the connector, create a wrapping struct
 pub(crate) struct ConnectorPDL {
     // State and properties of connector itself
     in_sync: bool,
     // Branch management
     branches: Vec<Branch>, // first branch is always non-speculative one
     sync_active: BranchQueue,
     sync_pending_get: BranchQueue,
     sync_finished: BranchQueue,
     sync_finished_last_handled: u32, // TODO: Change to BranchId?
     cur_round: u32,
     // Port/message management
     pub committed_to: Option<(ConnectorId, u64)>,
     pub inbox: PrivateInbox,
     pub ports: ConnectorPorts,
+}
 // TODO: Remove this monstrosity
 struct ConnectorRunContext<'a> {
-    inbox: &'a PrivateInbox,
+    branch_index: u32,
     ports: &'a ConnectorPorts,
     branch: &'a Branch,
     ports_delta: &'a Vec<PortOwnershipDelta>,
     received: &'a HashMap<PortIdLocal, DataMessage>,
     scheduler: SchedulerCtx<'a>,
     prepared_channel: Option<(Value, Value)>,
+}
 impl<'a> RunContext for ConnectorRunContext<'a> {
     fn did_put(&mut self, port: PortId) -> bool {
-        if self.branch.ports_delta.iter().any(|v| v.port_id.index == port.0.u32_suffix) {
         if self.ports_delta.iter().any(|v| v.port_id.index == port.0.u32_suffix) {
             // Either acquired or released, must be silent
             return false;
+        }
         let port_index = self.ports.get_port_index(PortIdLocal::new(port.0.u32_suffix)).unwrap();
-        let mapping = self.ports.get_port(self.branch.index.index, port_index);
+        let mapping = self.ports.get_port(self.branch_index, port_index);
         return mapping.is_assigned;
+    }
     fn get(&mut self, port: PortId) -> Option<ValueGroup> {
         let port_id = PortIdLocal::new(port.0.u32_suffix);
-        match self.branch.received.get(&port_id) {
         match self.received.get(&port_id) {
             Some(message) => Some(message.message.clone()),
             None => None,
+        }
+    }
     fn fires(&mut self, port: PortId) -> Option<Value> {
         let port_id = PortIdLocal::new(port.0.u32_suffix);
-        if self.branch.ports_delta.iter().any(|v| v.port_id == port_id) {
         if self.ports_delta.iter().any(|v| v.port_id == port_id) {
             return None
+        }
         let port_index = self.ports.get_port_index(port_id).unwrap();
-        let mapping = self.ports.get_port(self.branch.index.index, port_index);
+        let mapping = self.ports.get_port(self.branch_index, port_index);
         if mapping.is_assigned {
             return Some(Value::Bool(mapping.num_times_fired != 0));
         } else {
             return None;
+        }
+    }
     fn get_channel(&mut self) -> Option<(Value, Value)> {
         let (getter, putter) = self.scheduler.runtime.create_channel();
         debug_assert_eq!(getter.kind, PortKind::Getter);
         return self.prepared_channel.take();
+    }
+}
 impl Connector for ConnectorPDL {
     fn handle_message(&mut self, message: Message, ctx: &ConnectorCtx, delta_state: &mut RunDeltaState) {
         use MessageContents as MC;
         match message.contents {
             MC::Data(content) => self.handle_data_message(message.receiving_port, content),
             MC::Sync(content) => self.handle_sync_message(content, ctx, delta_state),
             MC::RequestCommit(content) => self.handle_request_commit_message(content, ctx, delta_state),
             MC::ConfirmCommit(content) => self.handle_confirm_commit_message(content, ctx, delta_state),
             MC::Control(_) | MC::Ping => {},
+        }
+    }
-    fn run(&mut self, sched_ctx: &SchedulerCtx, conn_ctx: &ConnectorCtx, delta_state: &mut RunDeltaState) -> ConnectorScheduling {
     fn run(&mut self, sched_ctx: SchedulerCtx, conn_ctx: &ConnectorCtx, delta_state: &mut RunDeltaState) -> ConnectorScheduling {
         if self.in_sync {
-            let scheduling = self.run_in_speculative_mode(pd, conn_ctx, delta_state);
+            let scheduling = self.run_in_speculative_mode(sched_ctx, conn_ctx, delta_state);
             // When in speculative mode we might have generated new sync
             // solutions, we need to turn them into proposed solutions here.
             if self.sync_finished_last_handled != self.sync_finished.last {
                 // Retrieve first element in queue
                 let mut next_id;
                 if self.sync_finished_last_handled == 0 {
                     next_id = self.sync_finished.first;
                 } else {
                     let last_handled = &self.branches[self.sync_finished_last_handled as usize];
                     debug_assert!(last_handled.next_branch_in_queue.is_some()); // because "last handled" != "last in queue"
                     next_id = last_handled.next_branch_in_queue.unwrap();
+                }
                 loop {
                     let branch_id = BranchId::new(next_id);
                     let branch = &self.branches[next_id as usize];
                     let branch_next = branch.next_branch_in_queue;
                     // Turn local solution into a message and send it along
                     // TODO: Like `ports` access, also revise the construction of this `key`, should not be needed
                     let solution_message = self.generate_initial_solution_for_branch(branch_id, conn_ctx);
                     if let Some(valid_solution) = solution_message {
                         self.submit_sync_solution(valid_solution, conn_ctx, delta_state);
                     } else {
                         // Branch is actually invalid, but we only just figured
                         // it out. We need to mark it as invalid to prevent
                         // future use
                         Self::remove_branch_from_queue(&mut self.branches, &mut self.sync_finished, branch_id);
                         if branch_id.index == self.sync_finished_last_handled {
                             self.sync_finished_last_handled = self.sync_finished.last;
+                        }
                         let branch = &mut self.branches[next_id as usize];
                         branch.sync_state = SpeculativeState::Inconsistent;
+                    }
                     match branch_next {
                         Some(id) => next_id = id,
                         None => break,
+                    }
+                }
                 self.sync_finished_last_handled = next_id;
+            }
             return scheduling;
         } else {
-            let scheduling = self.run_in_deterministic_mode(pd, conn_ctx, delta_state);
+            let scheduling = self.run_in_deterministic_mode(sched_ctx, conn_ctx, delta_state);
             return scheduling;
+        }
+    }
+}
 impl ConnectorPDL {
     /// Constructs a representation of a connector. The assumption is that the
     /// initial branch is at the first instruction of the connector's code,
     /// hence is in a non-sync state.
     pub fn new(initial_branch: Branch, owned_ports: Vec<PortIdLocal>) -> Self {
         Self{
             in_sync: false,
             branches: vec![initial_branch],
             sync_active: BranchQueue::new(),
             sync_pending_get: BranchQueue::new(),
             sync_finished: BranchQueue::new(),
             sync_finished_last_handled: 0, // none at all
             cur_round: 0,
             committed_to: None,
             inbox: PrivateInbox::new(),
             ports: ConnectorPorts::new(owned_ports),
+        }
+    }
@@ @@ -698,57 +704,68 @@ impl ConnectorPDL { @@
         self.committed_to = None;
         self.inbox.clear();
         self.ports.commit_to_sync();
         // Add/remove any of the ports we lost during the sync phase
         for port_delta in &solution.ports_delta {
             if port_delta.acquired {
                 self.ports.add_port(port_delta.port_id);
             } else {
                 self.ports.remove_port(port_delta.port_id);
+            }
+        }
         solution.commit_to_sync();
+    }
     // -------------------------------------------------------------------------
     // Executing connector code
     // -------------------------------------------------------------------------
     /// Runs the connector in synchronous mode. Potential changes to the global
     /// system's state are added to the `RunDeltaState` object by the connector,
     /// where it is the caller's responsibility to immediately take care of
     /// those changes. The return value indicates when (and if) the connector
     /// needs to be scheduled again.
-    pub fn run_in_speculative_mode(&mut self, pd: &ProtocolDescription, _context: &ConnectorCtx, results: &mut RunDeltaState) -> ConnectorScheduling {
+    pub fn run_in_speculative_mode(&mut self, sched_ctx: SchedulerCtx, _context: &ConnectorCtx, results: &mut RunDeltaState) -> ConnectorScheduling {
         debug_assert!(self.in_sync);
         debug_assert!(!self.sync_active.is_empty());
         if self.sync_active.is_empty() {
             return ConnectorScheduling::NotNow;
+        }
         let branch = Self::pop_branch_from_queue(&mut self.branches, &mut self.sync_active);
         // Run the branch to the next blocking point
         let mut run_context = ConnectorRunContext {};
         let run_result = branch.code_state.run(&mut run_context, pd);
         debug_assert!(branch.prepared_channel.is_none());
         let mut run_context = ConnectorRunContext {
             branch_index: branch.index.index,
             ports: &self.ports,
             ports_delta: &branch.ports_delta,
             scheduler: sched_ctx,
             prepared_channel: None,
             received: &branch.received,
         };
         let run_result = branch.code_state.run(&mut run_context, &sched_ctx.runtime.protocol_description);
         // Match statement contains `return` statements only if the particular
         // run result behind handled requires an immediate re-run of the
         // connector.
         match run_result {
             RunResult::BranchInconsistent => {
                 // Speculative branch became inconsistent
                 branch.sync_state = SpeculativeState::Inconsistent;
             },
             RunResult::BranchMissingPortState(port_id) => {
                 // Branch called `fires()` on a port that does not yet have an
                 // assigned speculative value. So we need to create those
                 // branches
                 let local_port_id = PortIdLocal::new(port_id.0.u32_suffix);
                 let local_port_index = self.ports.get_port_index(local_port_id).unwrap();
                 debug_assert!(self.ports.owned_ports.contains(&local_port_id));
                 // Create two copied branches, one silent and one firing
                 branch.sync_state = SpeculativeState::HaltedAtBranchPoint;
                 let parent_branch_id = branch.index;
                 let parent_branch = &self.branches[parent_branch_id.index as usize];
                 let silent_index = self.branches.len() as u32;
@@ @@ -887,128 +904,147 @@ impl ConnectorPDL { @@
                 } else {
                     // Not yet assigned, do so now
                     true
                 };
                 if is_valid_put {
                     // Put in run results for thread to pick up and transfer to
                     // the correct connector inbox.
                     port_mapping.mark_definitive(branch.index, 1);
                     let message = DataMessage{
                         sending_port: local_port_id,
                         sender_prev_branch_id: BranchId::new_invalid(),
                         sender_cur_branch_id: branch.index,
                         message: value_group,
                     };
                     // If the message contains any ports then we release our
                     // ownership over them in this branch
                     debug_assert!(results.ports.is_empty());
                     find_ports_in_value_group(&message.message, &mut results.ports);
                     Self::release_ports_during_sync(&mut self.ports, branch, &results.ports).unwrap();
                     results.ports.clear();
                     results.outbox.push(MessageContents::Data(message));
                     let branch_index = branch.index;
                     Self::push_branch_into_queue(&mut self.branches, &mut self.sync_active, branch_index);
                     return ConnectorScheduling::Immediate
                 } else {
                     branch.sync_state = SpeculativeState::Inconsistent;
+                }
             },
             _ => unreachable!("unexpected run result '{:?}' while running in sync mode", run_result),
+        }
         // Not immediately scheduling, so schedule again if there are more
         // branches to run
         if self.sync_active.is_empty() {
             return ConnectorScheduling::NotNow;
         } else {
             return ConnectorScheduling::Later;
+        }
+    }
     /// Runs the connector in non-synchronous mode.
-    pub fn run_in_deterministic_mode(&mut self, sched_ctx: &SchedulerCtx, _context: &ConnectorCtx, results: &mut RunDeltaState) -> ConnectorScheduling {
+    pub fn run_in_deterministic_mode(&mut self, sched_ctx: SchedulerCtx, conn_ctx: &ConnectorCtx, results: &mut RunDeltaState) -> ConnectorScheduling {
         debug_assert!(!self.in_sync);
         debug_assert!(self.sync_active.is_empty() && self.sync_pending_get.is_empty() && self.sync_finished.is_empty());
         debug_assert!(self.branches.len() == 1);
         let branch = &mut self.branches[0];
         debug_assert!(branch.sync_state == SpeculativeState::RunningNonSync);
         let mut run_context = ConnectorRunContext{
-            inbox: &self.inbox,
+            branch_index: branch.index.index,
             ports: &self.ports,
             branch: &Branch {}
             ports_delta: &branch.ports_delta,
             scheduler: sched_ctx,
             prepared_channel: branch.prepared_channel.take(),
             received: &branch.received,
         };
-        let run_result = branch.code_state.run(&mut run_context, pd);
+        let run_result = branch.code_state.run(&mut run_context, &sched_ctx.runtime.protocol_description);
         match run_result {
             RunResult::ComponentTerminated => {
                 // Need to wait until all children are terminated
                 // TODO: Think about how to do this?
                 branch.sync_state = SpeculativeState::Finished;
                 return ConnectorScheduling::Exit;
             },
             RunResult::ComponentAtSyncStart => {
                 // Prepare for sync execution and reschedule immediately
                 self.in_sync = true;
                 let first_sync_branch = Branch::new_sync_branching_from(1, branch);
                 let first_sync_branch_id = first_sync_branch.index;
                 self.ports.prepare_sync_branch(0, 1);
                 self.branches.push(first_sync_branch);
                 Self::push_branch_into_queue(&mut self.branches, &mut self.sync_active, first_sync_branch_id);
                 return ConnectorScheduling::Later;
             },
             RunResult::NewComponent(definition_id, monomorph_idx, arguments) => {
                 // Construction of a new component. Find all references to ports
                 // inside of the arguments
                 debug_assert!(results.ports.is_empty());
                 find_ports_in_value_group(&arguments, &mut results.ports);
                 if !results.ports.is_empty() {
                     // Ports changing ownership
                     if let Err(_) = Self::release_ports_during_non_sync(&mut self.ports, branch, &results.ports) {
                         todo!("fatal error handling");
+                    }
+                }
                 // Add connector for later execution
                 let new_connector_state = ComponentState {
                     prompt: Prompt::new(&pd.types, &pd.heap, definition_id, monomorph_idx, arguments)
                     prompt: Prompt::new(
                         &sched_ctx.runtime.protocol_description.types,
                         &sched_ctx.runtime.protocol_description.heap,
                         definition_id, monomorph_idx, arguments
+                    )
                 };
                 let new_connector_ports = results.ports.clone(); // TODO: Do something with this
                 let new_connector_branch = Branch::new_initial_branch(new_connector_state);
                 let new_connector = ConnectorPDL::new(new_connector_branch, new_connector_ports);
                 results.new_connectors.push(new_connector);
                 return ConnectorScheduling::Later;
             },
             RunResult::NewChannel => {
                 // Need to prepare a new channel
                 todo!("adding channels to some global context");
                 let (getter, putter) = sched_ctx.runtime.create_channel(conn_ctx.id);
                 debug_assert_eq!(getter.kind, PortKind::Getter);
                 branch.prepared_channel = Some((
                     Value::Input(PortId::new(putter.self_id.index)),
                     Value::Output(PortId::new(getter.self_id.index))
                 ));
                 return ConnectorScheduling::Later;
                 results.new_ports.push(putter);
                 results.new_ports.push(getter);
                 return ConnectorScheduling::Immediate;
             },
             _ => unreachable!("unexpected run result '{:?}' while running in non-sync mode", run_result),
+        }
+    }
     // -------------------------------------------------------------------------
     // Internal helpers
     // -------------------------------------------------------------------------
     // Helpers for management of the branches and their internally stored
     // `next_branch_in_queue` and the `BranchQueue` objects. Essentially forming
     // linked lists inside of the vector of branches.
     /// Pops from front of linked-list branch queue.
     fn pop_branch_from_queue<'a>(branches: &'a mut Vec<Branch>, queue: &mut BranchQueue) -> &'a mut Branch {
         debug_assert!(queue.first != 0);
         let branch = &mut branches[queue.first as usize];
         queue.first = branch.next_branch_in_queue.unwrap_or(0);
         branch.next_branch_in_queue = None;
         if queue.first == 0 {
             // No more entries in queue
             debug_assert_eq!(queue.last, branch.index.index);
             queue.last = 0;
@@ @@ -1085,48 +1121,52 @@ impl ConnectorPDL { @@
+    }
     // Helpers for local port management. Specifically for adopting/losing
     // ownership over ports, and for checking if specific ports can be sent
     // over another port.
     /// Releasing ownership of ports while in non-sync mode. This only occurs
     /// while instantiating new connectors
     fn release_ports_during_non_sync(ports: &mut ConnectorPorts, branch: &mut Branch, port_ids: &[PortIdLocal]) -> Result<(), PortOwnershipError> {
         debug_assert!(!branch.index.is_valid()); // branch in non-sync mode
         for port_id in port_ids {
             // We must own the port, or something is wrong with our code
             todo!("Set up some kind of message router");
             debug_assert!(ports.get_port_index(*port_id).is_some());
             ports.remove_port(*port_id);
+        }
         return Ok(())
+    }
     /// Releasing ownership of ports during a sync-session. Will provide an
     /// error if the port was already used during a sync block.
     fn release_ports_during_sync(ports: &mut ConnectorPorts, branch: &mut Branch, port_ids: &[PortIdLocal]) -> Result<(), PortOwnershipError> {
         if port_ids.is_empty() {
             return Ok(())
+        }
         todo!("unfinished: add port properties during final solution-commit msgs");
         debug_assert!(branch.index.is_valid()); // branch in sync mode
         for port_id in port_ids {
             match ports.get_port_index(*port_id) {
                 Some(port_index) => {
                     // We (used to) own the port. Make sure it is not given away
                     // already and not used to put/get data.
                     let port_mapping = ports.get_port(branch.index.index, port_index);
                     if port_mapping.is_assigned && port_mapping.num_times_fired != 0 {
                         // Already used
                         return Err(PortOwnershipError::UsedInInteraction(*port_id));
+                    }
                     for delta in &branch.ports_delta {
                         if delta.port_id == *port_id {
                             // We cannot have acquired this port, because the
                             // call to `ports.get_port_index` returned an index.
                             debug_assert!(!delta.acquired);
                             return Err(PortOwnershipError::AlreadyGivenAway(*port_id));
+                        }
+                    }
                     branch.ports_delta.push(PortOwnershipDelta{
@@ @@ -1136,48 +1176,52 @@ impl ConnectorPDL { @@
                 },
                 None => {
                     // Not in port mapping, so we must have acquired it before,
                     // remove the acquirement.
                     let mut to_delete_index: isize = -1;
                     for (delta_idx, delta) in branch.ports_delta.iter().enumerate() {
                         if delta.port_id == *port_id {
                             debug_assert!(delta.acquired);
                             to_delete_index = delta_idx as isize;
                             break;
+                        }
+                    }
                     debug_assert!(to_delete_index != -1);
                     branch.ports_delta.remove(to_delete_index as usize);
+                }
+            }
+        }
         return Ok(())
+    }
     /// Acquiring ports during a sync-session.
     fn acquire_ports_during_sync(ports: &mut ConnectorPorts, branch: &mut Branch, port_ids: &[PortIdLocal]) -> Result<(), PortOwnershipError> {
         if port_ids.is_empty() {
             return Ok(())
+        }
         todo!("unfinished: add port properties during final solution-commit msgs");
         debug_assert!(branch.index.is_valid()); // branch in sync mode
         'port_loop: for port_id in port_ids {
             for (delta_idx, delta) in branch.ports_delta.iter().enumerate() {
                 if delta.port_id == *port_id {
                     if delta.acquired {
                         // Somehow already received this port.
                         // TODO: @security
                         todo!("take care of nefarious peers");
                     } else {
                         // Sending ports to ourselves
                         debug_assert!(ports.get_port_index(*port_id).is_some());
                         branch.ports_delta.remove(delta_idx);
                         continue 'port_loop;
+                    }
+                }
+            }
             // If here then we can safely acquire the new port
             branch.ports_delta.push(PortOwnershipDelta{
                 acquired: true,
                 port_id: *port_id,
             });

src/runtime2/mod.rs

➞

Show inline comments

 // Structure of module
 mod runtime;
 mod messages;
 mod connector;
 mod native;
 mod port;
 mod scheduler;
 mod inbox;
 #[cfg(test)] mod tests;
 // Imports
 use std::collections::VecDeque;
 use std::sync::{Arc, Condvar, Mutex, RwLock};
 use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
 use std::thread::{self, JoinHandle};
 use crate::collections::RawVec;
 use crate::ProtocolDescription;
 use inbox::Message;
 use connector::{ConnectorPDL, ConnectorPublic, ConnectorScheduling, RunDeltaState};
-use scheduler::{Scheduler, ConnectorCtx, Router};
+use scheduler::{Scheduler, ConnectorCtx, ControlMessageHandler};
 use native::{Connector, ConnectorApplication, ApplicationInterface};
 use crate::runtime2::port::Port;
 use crate::runtime2::scheduler::SchedulerCtx;
 /// A kind of token that, once obtained, allows mutable access to a connector.
 /// We're trying to use move semantics as much as possible: the owner of this
 /// key is the only one that may execute the connector's code.
 pub(crate) struct ConnectorKey {
     pub index: u32, // of connector
+}
 impl ConnectorKey {
     /// Downcasts the `ConnectorKey` type, which can be used to obtain mutable
     /// access, to a "regular ID" which can be used to obtain immutable access.
     #[inline]
     pub fn downcast(&self) -> ConnectorId {
         return ConnectorId(self.index);
+    }
     /// Turns the `ConnectorId` into a `ConnectorKey`, marked as unsafe as it
     /// bypasses the type-enforced `ConnectorKey`/`ConnectorId` system
     #[inline]
     pub unsafe fn from_id(id: ConnectorId) -> ConnectorKey {
         return ConnectorKey{ index: id.0 };
+    }
+}
@@ @@ -63,61 +64,63 @@ impl ConnectorId { @@
     #[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 handle_message(&mut self, message: Message, ctx: &ConnectorCtx, delta_state: &mut RunDeltaState) {
         match self {
             ConnectorVariant::UserDefined(c) => c.handle_message(message, ctx, delta_state),
             ConnectorVariant::Native(c) => c.handle_message(message, ctx, delta_state),
+        }
+    }
-    fn run(&mut self, protocol_description: &ProtocolDescription, ctx: &ConnectorCtx, delta_state: &mut RunDeltaState) -> ConnectorScheduling {
+    fn run(&mut self, scheduler_ctx: SchedulerCtx, conn_ctx: &ConnectorCtx, delta_state: &mut RunDeltaState) -> ConnectorScheduling {
         match self {
             ConnectorVariant::UserDefined(c) => c.run(protocol_description, ctx, delta_state),
             ConnectorVariant::Native(c) => c.run(protocol_description, ctx, delta_state),
             ConnectorVariant::UserDefined(c) => c.run(scheduler_ctx, conn_ctx, delta_state),
             ConnectorVariant::Native(c) => c.run(scheduler_ctx, conn_ctx, delta_state),
+        }
+    }
+}
 pub(crate) struct ScheduledConnector {
     pub connector: ConnectorVariant, // access by connector
     pub context: ConnectorCtx, // mutable access by scheduler, immutable by connector
     pub public: ConnectorPublic, // accessible by all schedulers and connectors
     pub router: Router,
     pub router: ControlMessageHandler,
     pub shutting_down: bool,
     pub pending_acks: u32,
+}
 // -----------------------------------------------------------------------------
 // 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
@@ @@ -193,122 +196,140 @@ pub(crate) struct RuntimeInner { @@
     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 ports
+    // --- 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) -> (Port, Port) {
+    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 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,
             kind: PortKind::Getter,
-            peer_connector: self.connector_id,
+            peer_connector: creating_connector,
         };
         let putter_port = Port{
             self_id: putter_id,
             peer_id: getter_id,
             kind: PortKind::Putter,
-            peer_connector: self.connector_id,
+            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: Message) {
         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
     pub(crate) 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. The caller MUST make sure to schedule the
     /// connector.
     // TODO: Nicer code, not forcing the caller to schedule, perhaps?
     pub(crate) fn create_pdl_component(&self, created_by: &mut ScheduledConnector, connector: ConnectorPDL) -> 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), true)
+            lock.create(ConnectorVariant::UserDefined(connector), false)
         };
         // Transfer the ports
+        {
             let lock = self.connectors.read().unwrap();
             let created = lock.get_private(&key);
             match &created.connector {
                 ConnectorVariant::UserDefined(connector) => {
                     for port_id in connector.ports.owned_ports.iter().copied() {
                         println!("DEBUG: Transferring port {:?} from {} to {}", port_id, created_by.context.id.0, key.index);
                         let mut port = created_by.context.remove_port(port_id);
                         created.context.add_port(port);
+                    }
                 },
                 ConnectorVariant::Native(_) => unreachable!(),
+            }
+        }
         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);
@@ @@ -385,49 +406,51 @@ impl ConnectorStore { @@
         unsafe {
             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 {
             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,
             context: ConnectorCtx::new(),
             public: ConnectorPublic::new(initially_sleeping),
             router: Router::new(),
             router: ControlMessageHandler::new(),
             shutting_down: false,
             pending_acks: 0,
         };
         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.context.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.context.id = key.downcast();
             unsafe {
                 let target = self.connectors.get_mut(index);
                 std::ptr::write(*target, connector);
+            }
+        }

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 crate::ProtocolDescription;
 use super::{ConnectorKey, ConnectorId, RuntimeInner, ConnectorCtx};
 use super::scheduler::SchedulerCtx;
 use super::port::{Port, PortIdLocal, Channel, PortKind};
 use super::connector::{Branch, ConnectorScheduling, RunDeltaState, ConnectorPDL};
 use super::connector::find_ports_in_value_group;
 use super::inbox::{Message, MessageContents};
 /// Generic connector interface from the scheduler's point of view.
 pub(crate) trait Connector {
     /// Handle a new message (preprocessed by the scheduler). You probably only
     /// want to handle `Data`, `Sync`, and `Solution` messages. The others are
     /// intended for the scheduler itself.
     fn handle_message(&mut self, message: Message, ctx: &ConnectorCtx, delta_state: &mut RunDeltaState);
     /// Should run the connector's behaviour up until the next blocking point.
-    fn run(&mut self, sched_ctx: &SchedulerCtx, conn_ctx: &ConnectorCtx, delta_state: &mut RunDeltaState) -> ConnectorScheduling;
     fn run(&mut self, sched_ctx: SchedulerCtx, conn_ctx: &ConnectorCtx, delta_state: &mut RunDeltaState) -> ConnectorScheduling;
+}
 type SyncDone = Arc<(Mutex<bool>, Condvar)>;
 type JobQueue = Arc<Mutex<VecDeque<ApplicationJob>>>;
 enum ApplicationJob {
     NewChannel((Port, Port)),
     NewConnector(ConnectorPDL),
     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 handle_message(&mut self, message: Message, _ctx: &ConnectorCtx, _delta_state: &mut RunDeltaState) {
         use MessageContents as MC;
         match message.contents {
             MC::Data(_) => unreachable!("data message in API connector"),
             MC::Sync(_) | MC::RequestCommit(_) | MC::ConfirmCommit(_) => {
                 // Handling sync in API
             },
             MC::Control(_) => {},
             MC::Ping => {},
+        }
+    }
-    fn run(&mut self, _sched_ctx: &SchedulerCtx, _conn_ctx: &ConnectorCtx, delta_state: &mut RunDeltaState) -> ConnectorScheduling {
     fn run(&mut self, _sched_ctx: SchedulerCtx, _conn_ctx: &ConnectorCtx, delta_state: &mut RunDeltaState) -> ConnectorScheduling {
         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");
                     delta_state.new_ports.reserve(2);
                     delta_state.new_ports.push(endpoint_a);
                     delta_state.new_ports.push(endpoint_b);
+                }
                 ApplicationJob::NewConnector(connector) => {
                     println!("DEBUG: API creating connector");
                     delta_state.new_connectors.push(connector);
                 },
                 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();
+        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: Optimize by yanking 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

src/runtime2/port.rs

➞

Show inline comments

 use super::ConnectorId;
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
 pub struct PortIdLocal {
     pub index: u32,
+}
 impl PortIdLocal {
     pub fn new(id: u32) -> Self {
         Self{ index: id }
+    }
     // TODO: Unsure about this, maybe remove, then also remove all struct
     //  instances where I call this
     pub fn new_invalid() -> Self {
         Self{ index: u32::MAX }
+    }
     pub fn is_valid(&self) -> bool {
         return self.index != u32::MAX;
+    }
+}
 #[derive(Eq, PartialEq)]
+#[derive(Debug, Eq, PartialEq)]
 pub enum PortKind {
     Putter,
     Getter,
+}
 /// Represents a port inside of the runtime. May be without owner if it is
 /// created by the application interfacing with the runtime, instead of being
 /// created by a connector.
 pub struct Port {
     pub self_id: PortIdLocal,
     pub peer_id: PortIdLocal,
     pub kind: PortKind,
     pub peer_connector: ConnectorId, // might be temporarily inconsistent while peer port is sent around in non-sync phase.
+}
 // TODO: Turn port ID into its own type
 pub struct Channel {
     pub putter_id: PortIdLocal, // can put on it, so from the connector's point of view, this is an output
     pub getter_id: PortIdLocal, // vice versa: can get on it, so an input for the connector
+}
@@ \ No newline at end of file @@

src/runtime2/scheduler.rs

➞

Show inline comments

@@ @@ -32,61 +32,59 @@ impl ConnectorCtx { @@
         return self.ports.remove(index);
+    }
     pub(crate) fn get_port(&self, id: PortIdLocal) -> &Port {
         let index = self.port_id_to_index(id);
         return &self.ports[index];
+    }
     pub(crate) fn get_port_mut(&mut self, id: PortIdLocal) -> &mut Port {
         let index = self.port_id_to_index(id);
         return &mut self.ports[index];
+    }
     fn port_id_to_index(&self, id: PortIdLocal) -> usize {
         for (idx, port) in self.ports.iter().enumerate() {
             if port.self_id == id {
                 return idx;
+            }
+        }
         panic!("port {:?}, not owned by connector", id);
+    }
+}
 // 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,
+}
 // Thinking aloud: actual ports should be accessible by connector, but managed
 // by the scheduler (to handle rerouting messages). We could just give a read-
 // only context, instead of an extra call on the "Connector" trait.
 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
         let scheduler_id = self.scheduler_id;
         let mut delta_state = RunDeltaState::new();
         'thread_loop: loop {
             // Retrieve a unit of work
             println!("DEBUG [{}]: Waiting for work", scheduler_id);
             let connector_key = self.runtime.wait_for_work();
             if connector_key.is_none() {
                 // We should exit
                 println!("DEBUG [{}]: ... No more work, quitting", scheduler_id);
                 break 'thread_loop;
+            }
             // We have something to do
             let connector_key = connector_key.unwrap();
             println!("DEBUG [{}]: ... Got work, running {}", scheduler_id, connector_key.index);
@@ @@ -123,246 +121,261 @@ impl Scheduler { @@
                                 // clear the rerouting entry of the `ack`-receiver.
                                 self.send_message_and_wake_up_if_sleeping(
                                     message.sending_connector,
                                     Message{
                                         sending_connector: connector_key.downcast(),
                                         receiving_port: PortIdLocal::new_invalid(),
                                         contents: MessageContents::Control(ControlMessage{
                                             id: content.id,
                                             content: ControlMessageVariant::Ack,
                                         }),
+                                    }
                                 );
                             },
                             ControlMessageVariant::Ack => {
                                 scheduled.router.handle_ack(content.id);
+                            }
+                        }
                     } else {
                         // Let connector handle message
                         scheduled.connector.handle_message(message, &scheduled.context, &mut delta_state);
+                    }
+                }
                 // Actually run the connector
                 println!("DEBUG [{}]: Running {} ...", scheduler_id, connector_key.index);
                 let scheduler_ctx = SchedulerCtx{ runtime: &*self.runtime };
                 let new_schedule = scheduled.connector.run(
-                    &self.runtime.protocol_description, &scheduled.context, &mut delta_state
+                    scheduler_ctx, &scheduled.context, &mut delta_state
                 );
                 println!("DEBUG [{}]: ... Finished running {}", scheduler_id, connector_key.index);
                 // Handle all of the output from the current run: messages to
                 // send and connectors to instantiate.
                 self.handle_delta_state(&connector_key, &mut scheduled.context, &mut delta_state);
                 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`.
                     debug_assert_eq!(scheduled.public.sleeping.load(Ordering::Acquire), false);
                     scheduled.public.sleeping.store(true, Ordering::Release);
                     // We might have received a message in the meantime from a
                     // thread that did not see the sleeping flag set to `true`,
                     // so:
                     if !scheduled.public.inbox.is_empty() {
                         let should_reschedule_self = scheduled.public.sleeping
                             .compare_exchange(true, false, Ordering::SeqCst, Ordering::Acquire)
                             .is_ok();
                         if should_reschedule_self {
                             self.runtime.push_work(connector_key);
+                        }
+                    }
                 },
                 ConnectorScheduling::Exit => {
                     // TODO: Better way of doing this, when exiting then
                     //  connected components must know their channels are invalid
                     self.runtime.destroy_component(connector_key);
                     // Prepare for exit. Set the shutdown flag and broadcast
                     // messages to notify peers of closing channels
                     scheduled.shutting_down = true;
                     for port in &scheduled.context.ports {
                         self.runtime.send_message(port.peer_connector, Message{
                             sending_connector: connector_key.downcast(),
                             receiving_port: port.peer_id,
                             contents: MessageContents::Control(ControlMessage{
                                 id: 0,
                                 content: ControlMessageVariant::Ack
                             })
                         })
+                    }
+                }
+            }
+        }
+    }
     fn handle_delta_state(&mut self, connector_key: &ConnectorKey, context: &mut ConnectorCtx, delta_state: &mut RunDeltaState) {
         // Handling any messages that were sent
         let connector_id = connector_key.downcast();
         if !delta_state.outbox.is_empty() {
             for mut message in delta_state.outbox.drain(..) {
                 // Based on the message contents, decide where the message
                 // should be sent to. This might end up modifying the message.
                 let (peer_connector, peer_port) = match &mut message {
                     MessageContents::Data(contents) => {
                         let port = context.get_port(contents.sending_port);
                         (port.peer_connector, port.peer_id)
                     },
                     MessageContents::Sync(contents) => {
                         let connector = contents.to_visit.pop().unwrap();
                         (connector, PortIdLocal::new_invalid())
                     },
                     MessageContents::RequestCommit(contents)=> {
                         let connector = contents.to_visit.pop().unwrap();
                         (connector, PortIdLocal::new_invalid())
                     },
                     MessageContents::ConfirmCommit(contents) => {
                         for to_visit in &contents.to_visit {
                             let message = Message{
                                 sending_connector: connector_id,
                                 receiving_port: PortIdLocal::new_invalid(),
                                 contents: MessageContents::ConfirmCommit(contents.clone()),
                             };
-                            self.send_message_and_wake_up_if_sleeping(*to_visit, message);
+                            self.runtime.send_message(*to_visit, message);
+                        }
                         (ConnectorId::new_invalid(), PortIdLocal::new_invalid())
                     },
                     MessageContents::Control(_) | MessageContents::Ping => {
                         // Never generated by the user's code
                         unreachable!();
+                    }
                 };
                 // TODO: Maybe clean this up, perhaps special case for
                 //  ConfirmCommit can be handled differently.
                 if peer_connector.is_valid() {
                     let message = Message {
                         sending_connector: connector_id,
                         receiving_port: peer_port,
                         contents: message,
                     };
-                    self.send_message_and_wake_up_if_sleeping(peer_connector, message);
+                    self.runtime.send_message(peer_connector, message);
+                }
+            }
+        }
         if !delta_state.new_ports.is_empty() {
             for port in delta_state.new_ports.drain(..) {
                 context.ports.push(port);
+            }
+        }
         // Handling any new connectors that were scheduled
         // TODO: Pool outgoing messages to reduce atomic access
         if !delta_state.new_connectors.is_empty() {
             let cur_connector = self.runtime.get_component_private(connector_key);
             for new_connector in delta_state.new_connectors.drain(..) {
                 // Add to global registry to obtain key
                 let new_key = self.runtime.create_pdl_component(cur_connector, new_connector);
                 let new_connector = self.runtime.get_component_private(&new_key);
                 // Call above changed ownership of ports, but we still have to
                 // let the other end of the channel know that the port has
                 // changed location.
                 for port in &new_connector.context.ports {
                     cur_connector.pending_acks += 1;
                     let reroute_message = cur_connector.router.prepare_reroute(
                         port.self_id, port.peer_id, cur_connector.context.id,
                         port.peer_connector, new_connector.context.id
                     );
-                    self.send_message_and_wake_up_if_sleeping(port.peer_connector, reroute_message);
+                    self.runtime.send_message(port.peer_connector, reroute_message);
+                }
                 // Schedule new connector to run
                 self.runtime.push_work(new_key);
+            }
+        }
         debug_assert!(delta_state.outbox.is_empty());
         debug_assert!(delta_state.new_ports.is_empty());
         debug_assert!(delta_state.new_connectors.is_empty());
+    }
+}
     fn send_message_and_wake_up_if_sleeping(&self, connector_id: ConnectorId, message: Message) {
         let connector = self.runtime.get_component_public(connector_id);
 // -----------------------------------------------------------------------------
 // Control messages
 // -----------------------------------------------------------------------------
         connector.inbox.insert_message(message);
         let should_wake_up = connector.sleeping
             .compare_exchange(true, false, Ordering::SeqCst, Ordering::Acquire)
             .is_ok();
 struct ControlEntry {
     id: u32,
     variant: ControlVariant,
+}
         if should_wake_up {
             let key = unsafe { ConnectorKey::from_id(connector_id) };
             self.runtime.push_work(key);
+        }
+    }
 enum ControlVariant {
     ChangedPort(ControlChangedPort),
     ClosedChannel(ControlClosedChannel),
+}
 /// Represents a rerouting entry due to a moved port
 // TODO: Optimize
 struct ReroutedTraffic {
     id: u32,                        // ID of control message
     target_port: PortIdLocal,       // targeted port
 struct ControlChangedPort {
     target_port: PortIdLocal,       // if send to this port, then reroute
     source_connector: ConnectorId,  // connector we expect messages from
     target_connector: ConnectorId,  // connector they should be rerouted to
     target_connector: ConnectorId,  // connector we need to reroute to
+}
 struct ControlClosedChannel {
+}
-pub(crate) struct Router {
+pub(crate) struct ControlMessageHandler {
     id_counter: u32,
-    active: Vec<ReroutedTraffic>,
+    active: Vec<ControlEntry>,
+}
-impl Router {
+impl ControlMessageHandler {
     pub fn new() -> Self {
-        Router{
+        ControlMessageHandler {
             id_counter: 0,
             active: Vec::new(),
+        }
+    }
     /// Prepares rerouting messages due to changed ownership of a port. The
     /// control message returned by this function must be sent to the
     /// transferred port's peer connector.
     pub fn prepare_reroute(
         &mut self,
         port_id: PortIdLocal, peer_port_id: PortIdLocal,
         self_connector_id: ConnectorId, peer_connector_id: ConnectorId,
         new_owner_connector_id: ConnectorId
     ) -> Message {
         let id = self.id_counter;
         let (new_id_counter, _) = self.id_counter.overflowing_add(1);
         self.id_counter = new_id_counter;
         self.active.push(ReroutedTraffic{
             id,
             target_port: port_id,
             source_connector: peer_connector_id,
             target_connector: new_owner_connector_id,
         });
         return Message{
             sending_connector: self_connector_id,
-            receiving_port: PortIdLocal::new_invalid(),
+            receiving_port: peer_port_id,
             contents: MessageContents::Control(ControlMessage{
                 id,
                 content: ControlMessageVariant::ChangePortPeer(peer_port_id, new_owner_connector_id),
             })
         };
+    }
     /// Returns true if the supplied message should be rerouted. If so then this
     /// function returns the connector that should retrieve this message.
     pub fn should_reroute(&self, sending_connector: ConnectorId, target_port: PortIdLocal) -> Option<ConnectorId> {
         for reroute in &self.active {
             if reroute.source_connector == sending_connector &&
                 reroute.target_port == target_port {
                 // Need to reroute this message
                 return Some(reroute.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()

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::*;
 fn runtime_for(num_threads: u32, pdl: &str) -> Runtime {
     let protocol = ProtocolDescription::parse(pdl.as_bytes()).expect("parse pdl");
     let runtime = Runtime::new(num_threads, protocol);
     return runtime;
+}
 #[test]
 fn test_put_and_get() {
-    let rt = runtime_for(4, "
+    let rt = runtime_for(1, "
 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;
+    }
+}
     ");

0 comments (0 inline, 0 general)