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

Changeset - edb4c4be7e45

Parent rev.

Child rev.

[Not reviewed]

0 9 0

MH - 4 years ago 2021-11-08 21:21:31
contact@maxhenger.nl

fixing compilation errors

9 files changed with 176 insertions and 127 deletions:

src/collections/sets.rs

src/runtime2/branch.rs

src/runtime2/connector2.rs

src/runtime2/consensus.rs

src/runtime2/inbox.rs

src/runtime2/inbox2.rs

src/runtime2/mod.rs

src/runtime2/native.rs

src/runtime2/scheduler.rs

0 comments (0 inline, 0 general)

src/collections/sets.rs

➞

Show inline comments

@@ @@ -80,25 +80,25 @@ impl<T: Eq> VecSet<T> { @@
+            }
+        }
         self.inner.push(to_push);
+    }
     #[inline]
     pub fn clear(&mut self) {
         self.inner.clear();
+    }
     #[inline]
     pub fn iter(&self) -> impl Iterator<Item=T> {
+    pub fn iter(&self) -> impl Iterator<Item=&T> {
         return self.inner.iter();
+    }
     #[inline]
     pub fn is_empty(&self) -> bool {
         self.inner.is_empty()
+    }
     #[inline]
     pub fn into_vec(self) -> Vec<T> {
         return self.inner;
+    }

src/runtime2/branch.rs

➞

Show inline comments

@@ @@ -274,36 +274,39 @@ impl ExecTree { @@
         debug_assert!(self.is_in_sync());
         let parent_branch = &self[parent_branch_id];
         let new_branch = Branch::new_sync(1, parent_branch);
         let new_branch_id = new_branch.id;
         return new_branch_id;
+    }
     /// Collapses the speculative execution tree back into a deterministic one,
     /// using the provided branch as the final sync result.
     pub fn end_sync(&mut self, branch_id: BranchId) {
         debug_assert!(self.is_in_sync());
         debug_assert!(self.iter_queue(QueueKind::FinishedSync).any(|v| v.id == branch_id));
+        debug_assert!(self.iter_queue(QueueKind::FinishedSync, None).any(|v| v.id == branch_id));
         // Swap indicated branch into the first position
         self.branches.swap(0, branch_id.index as usize);
         self.branches.truncate(1);
         // Reset all values to non-sync defaults
         let branch = &mut self.branches[0];
         branch.id = BranchId::new_invalid();
         branch.parent_id = BranchId::new_invalid();
         branch.sync_state = SpeculativeState::RunningNonSync;
         debug_assert!(!branch.awaiting_port.is_valid());
         branch.next_in_queue = BranchId::new_invalid();
         branch.inbox.clear();
         debug_assert!(branch.prepared_channel.is_none());
         // Clear out all the queues
         for queue_idx in 0..NUM_QUEUES {
             self.queues[queue_idx] = BranchQueue::new();
+        }
+    }
+}
 impl Index<BranchId> for ExecTree {
     type Output = Branch;
     fn index(&self, index: BranchId) -> &Self::Output {

src/runtime2/connector2.rs

➞

Show inline comments

@@ @@ -29,26 +29,25 @@ 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::consensus::find_ports_in_value_group;
 use crate::runtime2::port::PortKind;
 use super::branch::{Branch, BranchId, ExecTree, QueueKind, SpeculativeState};
 use super::consensus::{Consensus, Consistency};
 use super::inbox2::{DataMessageFancy, MessageFancy, SyncMessageFancy};
 use super::inbox::PublicInbox;
 use super::inbox2::{DataMessageFancy, MessageFancy, SyncMessageFancy, PublicInbox};
 use super::native::Connector;
 use super::port::PortIdLocal;
 use super::scheduler::{ComponentCtxFancy, SchedulerCtx};
 pub(crate) struct ConnectorPublic {
     pub inbox: PublicInbox,
     pub sleeping: AtomicBool,
+}
 impl ConnectorPublic {
     pub fn new(initialize_as_sleeping: bool) -> Self {
         ConnectorPublic{
@@ @@ -70,25 +69,25 @@ pub(crate) struct ConnectorPDL { @@
     tree: ExecTree,
     consensus: Consensus,
+}
 struct ConnectorRunContext<'a> {
     branch_id: BranchId,
     consensus: &'a Consensus,
     received: &'a HashMap<PortIdLocal, ValueGroup>,
     scheduler: SchedulerCtx<'a>,
     prepared_channel: Option<(Value, Value)>,
+}
 impl RunContext for ConnectorRunContext{
+impl<'a> RunContext for ConnectorRunContext<'a>{
     fn did_put(&mut self, port: PortId) -> bool {
         let port_id = PortIdLocal::new(port.0.u32_suffix);
         let annotation = self.consensus.get_annotation(self.branch_id, port_id);
         return annotation.registered_id.is_some();
+    }
     fn get(&mut self, port: PortId) -> Option<ValueGroup> {
         let port_id = PortIdLocal::new(port.0.u32_suffix);
         match self.received.get(&port_id) {
             Some(data) => Some(data.clone()),
             None => None,
+        }
@@ @@ -101,77 +100,80 @@ impl RunContext for ConnectorRunContext{ @@
+    }
     fn get_channel(&mut self) -> Option<(Value, Value)> {
         return self.prepared_channel.take();
+    }
+}
 impl Connector for ConnectorPDL {
     fn run(&mut self, sched_ctx: SchedulerCtx, comp_ctx: &mut ComponentCtxFancy) -> ConnectorScheduling {
         self.handle_new_messages(comp_ctx);
         if self.tree.is_in_sync() {
             let scheduling = self.run_in_sync_mode(sched_ctx, comp_ctx);
             self.consensus.handle_new_finished_sync_branches();
             if let Some(solution_branch_id) = self.consensus.handle_new_finished_sync_branches(&self.tree, comp_ctx) {
                 todo!("call handler");
+            }
             return scheduling;
         } else {
             let scheduling = self.run_in_deterministic_mode(sched_ctx, comp_ctx);
             return scheduling;
+        }
+    }
+}
 impl ConnectorPDL {
     pub fn new(initial: ComponentState) -> Self {
         Self{
             tree: ExecTree::new(initial),
             consensus: Consensus::new(),
+        }
+    }
     // --- Handling messages
     pub fn handle_new_messages(&mut self, ctx: &mut ComponentCtxFancy) {
         while let Some(message) = ctx.read_next_message() {
             match message {
                 MessageFancy::Data(message) => handle_new_data_message(message, ctx),
                 MessageFancy::Sync(message) => handle_new_sync_message(message, ctx),
                 MessageFancy::Data(message) => self.handle_new_data_message(message, ctx),
                 MessageFancy::Sync(message) => self.handle_new_sync_message(message, ctx),
                 MessageFancy::Control(_) => unreachable!("control message in component"),
+            }
+        }
+    }
     pub fn handle_new_data_message(&mut self, message: DataMessageFancy, ctx: &mut ComponentCtxFancy) {
         // Go through all branches that are awaiting new messages and see if
         // there is one that can receive this message.
         debug_assert!(ctx.workspace_branches.is_empty());
         self.consensus.handle_received_sync_header(&message.sync_header, ctx);
         self.consensus.handle_received_data_header(&self.tree, &message.data_header, &mut ctx.workspace_branches);
         self.consensus.handle_new_data_message(&self.tree, &message, ctx, &mut ctx.workspace_branches);
         for branch_id in ctx.workspace_branches.drain(..) {
             // This branch can receive, so fork and given it the message
             let receiving_branch_id = self.tree.fork_branch(branch_id);
             self.consensus.notify_of_new_branch(branch_id, receiving_branch_id);
             let receiving_branch = &mut self.tree[receiving_branch_id];
             receiving_branch.insert_message(message.data_header.target_port, message.content.clone());
             self.consensus.notify_of_received_message(branch_id, &message.data_header, &message.content);
             // And prepare the branch for running
             self.tree.push_into_queue(QueueKind::Runnable, receiving_branch_id);
+        }
+    }
     pub fn handle_new_sync_message(&mut self, message: SyncMessageFancy, ctx: &mut ComponentCtxFancy) {
         self.consensus.handle_received_sync_header(&message.sync_header, ctx);
         self.consensus.handle_received_sync_message(message, ctx);
         if let Some(solution_branch_id) = self.consensus.handle_new_sync_message(message, ctx) {
+        }
+    }
     // --- Running code
     pub fn run_in_sync_mode(&mut self, sched_ctx: SchedulerCtx, comp_ctx: &mut ComponentCtxFancy) -> ConnectorScheduling {
         // Check if we have any branch that needs running
         debug_assert!(self.tree.is_in_sync() && self.consensus.is_in_sync());
         let branch_id = self.tree.pop_from_queue(QueueKind::Runnable);
         if branch_id.is_none() {
             return ConnectorScheduling::NotNow;
+        }
@@ @@ -294,42 +296,42 @@ impl ConnectorPDL { @@
         } else {
             return ConnectorScheduling::Later;
+        }
+    }
     pub fn run_in_deterministic_mode(&mut self, sched_ctx: SchedulerCtx, comp_ctx: &mut ComponentCtxFancy) -> ConnectorScheduling {
         debug_assert!(!self.tree.is_in_sync() && !self.consensus.is_in_sync());
         let branch = self.tree.base_branch_mut();
         debug_assert!(branch.sync_state == SpeculativeState::RunningNonSync);
         let mut run_context = ConnectorRunContext{
             branch_id,
+            branch_id: branch.id,
             consensus: &self.consensus,
             received: &branch.inbox,
             scheduler: *sched_ctx,
             prepared_channel: branch.prepared_channel.take(),
         };
         let run_result = branch.code_state.run(&mut run_context, &sched_ctx.runtime.protocol_description);
         match run_result {
             RunResult::ComponentTerminated => {
                 branch.sync_state = SpeculativeState::Finished;
                 return ConnectorScheduling::Exit;
             },
             RunResult::ComponentAtSyncStart => {
                 let current_ports = comp_ctx.notify_sync_start();
                 let sync_branch_id = self.tree.start_sync();
                 self.consensus.start_sync(current_ports);
+                self.consensus.start_sync(current_ports, comp_ctx);
                 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 {
@@ @@ -352,13 +354,26 @@ impl ConnectorPDL { @@
                     Value::Input(PortId::new(putter.self_id.index)),
                     Value::Output(PortId::new(getter.self_id.index)),
                 ));
                 comp_ctx.push_port(putter);
                 comp_ctx.push_port(getter);
                 return ConnectorScheduling::Immediate;
             },
             _ => unreachable!("unexpected run result '{:?}' while running in non-sync mode", run_result),
+        }
+    }
     pub fn collapse_sync_to_solution_branch(&mut self, solution_branch_id: BranchId, ctx: &mut ComponentCtxFancy) {
         let mut fake_vec = Vec::new();
         self.tree.end_sync(solution_branch_id);
         self.consensus.end_sync(solution_branch_id, &mut fake_vec);
         for port in fake_vec {
             // TODO: Handle sent/received ports
             debug_assert!(ctx.get_port_by_id(port).is_some());
+        }
         ctx.notify_sync_end(&[]);
+    }
+}
@@ \ No newline at end of file @@

src/runtime2/consensus.rs

➞

Show inline comments

 use std::path::Component;
 use std::str::pattern::Pattern;
 use crate::collections::VecSet;
 use crate::protocol::eval::ValueGroup;
 use crate::runtime2::branch::{BranchId, ExecTree, QueueKind};
 use crate::runtime2::ConnectorId;
 use crate::runtime2::inbox2::{DataHeader, DataMessageFancy, MessageFancy, SyncContent, SyncHeader, SyncMessageFancy};
 use crate::runtime2::inbox::SyncMessage;
 use crate::runtime2::port::{ChannelId, Port, PortIdLocal};
 use crate::runtime2::scheduler::ComponentCtxFancy;
 use super::inbox2::PortAnnotation;
 use super::branch::{BranchId, ExecTree, QueueKind};
 use super::ConnectorId;
 use super::port::{ChannelId, Port, PortIdLocal};
 use super::inbox2::{
     DataHeader, DataMessageFancy, MessageFancy,
     SyncContent, SyncHeader, SyncMessageFancy, PortAnnotation
 };
 use super::scheduler::ComponentCtxFancy;
 struct BranchAnnotation {
     port_mapping: Vec<PortAnnotation>,
+}
 pub(crate) struct LocalSolution {
     component: ConnectorId,
     final_branch_id: BranchId,
     port_mapping: Vec<(ChannelId, BranchId)>,
+}
 #[derive(Clone)]
 pub(crate) struct GlobalSolution {
     branches: Vec<(ConnectorId, BranchId)>,
     port_mapping: Vec<(ChannelId, BranchId)>, // TODO: This can go, is debugging info
     component_branches: Vec<(ConnectorId, BranchId)>,
     channel_mapping: Vec<(ChannelId, BranchId)>, // TODO: This can go, is debugging info
+}
 // -----------------------------------------------------------------------------
 // Consensus
 // -----------------------------------------------------------------------------
 /// The consensus algorithm. Currently only implemented to find the component
 /// with the highest ID within the sync region and letting it handle all the
 /// local solutions.
 ///
 /// The type itself serves as an experiment to see how code should be organized.
 // TODO: Flatten all datastructures
@@ @@ -79,40 +79,45 @@ impl Consensus { @@
+    }
     /// 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, ports: &[Port]) {
+    pub fn start_sync(&mut self, ports: &[Port], ctx: &ComponentCtxFancy) {
         debug_assert!(!self.highest_connector_id.is_valid());
         debug_assert!(self.branch_annotations.is_empty());
         debug_assert!(self.last_finished_handled.is_none());
         debug_assert!(self.encountered_peers.is_empty());
         debug_assert!(self.solution_combiner.local.is_empty());
         // We'll use the first "branch" (the non-sync one) to store our ports,
         // this allows cloning if we created a new branch.
         self.branch_annotations.push(BranchAnnotation{
             port_mapping: ports.iter()
                 .map(|v| PortAnnotation{
                     port_id: v.self_id,
                     registered_id: None,
                     expected_firing: None,
                 })
                 .collect(),
         });
         self.highest_connector_id = ctx.id;
+    }
     /// Notifies the consensus algorithm that a new branch has appeared. Must be
     /// called for each forked branch in the execution tree.
     pub fn notify_of_new_branch(&mut self, parent_branch_id: BranchId, new_branch_id: BranchId) {
         // If called correctly. Then each time we are notified the new branch's
         // index is the length in `branch_annotations`.
         debug_assert!(self.branch_annotations.len() == new_branch_id.index as usize);
         let parent_branch_annotations = &self.branch_annotations[parent_branch_id.index as usize];
         let new_branch_annotations = BranchAnnotation{
             port_mapping: parent_branch_annotations.port_mapping.clone(),
         };
@@ @@ -163,66 +168,76 @@ impl Consensus { @@
+                        }
+                    }
+                }
+            }
+        }
         unreachable!("notify_of_speculative_mapping called with unowned port");
+    }
     /// Generates sync messages for any branches that are at the end of the
     /// sync block. To find these branches, they should've been put in the
     /// "finished" queue in the execution tree.
     pub fn handle_new_finished_sync_branches(&mut self, tree: &ExecTree, ctx: &mut ComponentCtxFancy) {
+    pub fn handle_new_finished_sync_branches(&mut self, tree: &ExecTree, ctx: &mut ComponentCtxFancy) -> Option<BranchId> {
         debug_assert!(self.is_in_sync());
         let mut last_branch_id = self.last_finished_handled;
         for branch in tree.iter_queue(QueueKind::FinishedSync, last_branch_id) {
             // Turn the port mapping into a local solution
             let source_mapping = &self.branch_annotations[branch.id.index as usize].port_mapping;
             let mut target_mapping = Vec::with_capacity(source_mapping.len());
             for port in source_mapping {
                 let port_desc = ctx.get_port_by_id(port.port_id).unwrap();
                 target_mapping.push((
                     port_desc.channel_id,
                     port.registered_id.unwrap_or(BranchId::new_invalid())
                 ));
+            }
             let local_solution = LocalSolution{
                 component: ctx.id,
                 final_branch_id: branch.id,
                 port_mapping: target_mapping,
             };
             self.send_or_store_local_solution(local_solution, ctx);
             let solution_branch = self.send_or_store_local_solution(local_solution, ctx);
             if solution_branch.is_some() {
                 // No need to continue iterating, we've found the solution
                 return solution_branch;
+            }
             last_branch_id = Some(branch.id);
+        }
         self.last_finished_handled = last_branch_id;
         return None;
+    }
     pub fn end_sync(&mut self, branch_id: BranchId, final_ports: &mut Vec<PortIdLocal>) {
         debug_assert!(self.is_in_sync());
         // TODO: Handle sending and receiving ports
         // Set final ports
         final_ports.clear();
         let branch = &self.branch_annotations[branch_id.index as usize];
         for port in &branch.port_mapping {
             final_ports.push(port.port_id);
+        }
         // Clear out internal storage
         // Clear out internal storage to defaults
         self.highest_connector_id = ConnectorId::new_invalid();
         self.branch_annotations.clear();
         self.last_finished_handled = None;
         self.encountered_peers.clear();
         self.solution_combiner.clear();
+    }
     // --- Handling messages
     /// Prepares a message for sending. Caller should have made sure that
     /// sending the message is consistent with the speculative state.
     pub fn handle_message_to_send(&mut self, branch_id: BranchId, source_port_id: PortIdLocal, content: &ValueGroup, ctx: &mut ComponentCtxFancy) -> (SyncHeader, DataHeader) {
         debug_assert!(self.is_in_sync());
         let branch = &mut self.branch_annotations[branch_id.index as usize];
         if cfg!(debug_assertions) {
             let port = branch.port_mapping.iter()
@@ @@ -280,30 +295,29 @@ impl Consensus { @@
         self.handle_received_sync_header(&message.sync_header, ctx);
         // And handle the contents
         debug_assert_eq!(message.target_component_id, ctx.id);
         match message.content {
             SyncContent::Notification => {
                 // We were just interested in the header
                 return None;
             },
             SyncContent::LocalSolution(solution) => {
                 // We might be the leader, or earlier messages caused us to not
                 // be the leader anymore.
                 self.send_or_store_local_solution(solution, ctx);
                 return None;
                 return self.send_or_store_local_solution(solution, ctx);
             },
             SyncContent::GlobalSolution(solution) => {
                 // Take branch of interest and return it.
-                let (_, branch_id) = solution.branches.iter()
+                let (_, branch_id) = solution.component_branches.iter()
                     .find(|(connector_id, _)| connector_id == ctx.id)
                     .unwrap();
                 return Some(*branch_id);
+            }
+        }
+    }
     pub fn notify_of_received_message(&mut self, branch_id: BranchId, data_header: &DataHeader, content: &ValueGroup) {
         debug_assert!(self.branch_can_receive(branch_id, data_header));
         let branch = &mut self.branch_annotations[branch_id.index as usize];
         for mapping in &mut branch.port_mapping {
             if mapping.port_id == data_header.target_port {
@@ @@ -393,45 +407,58 @@ impl Consensus { @@
         } else if sync_header.highest_component_id < self.highest_connector_id {
             // Sender has lower leader ID, so it should know about our higher
             // one.
             let message = SyncMessageFancy{
                 sync_header: self.create_sync_header(ctx),
                 target_component_id: sync_header.sending_component_id,
                 content: SyncContent::Notification
             };
             ctx.submit_message(MessageFancy::Sync(message));
         } // else: exactly equal, so do nothing
+    }
     fn send_or_store_local_solution(&mut self, solution: LocalSolution, ctx: &mut ComponentCtxFancy) {
+    fn send_or_store_local_solution(&mut self, solution: LocalSolution, ctx: &mut ComponentCtxFancy) -> Option<BranchId> {
         if self.highest_connector_id == ctx.id {
             // We are the leader
             if let Some(global_solution) = self.solution_combiner.add_solution_and_check_for_global_solution(solution) {
                 for (connector_id, _) in global_solution.branches.iter().copied() {
                 let mut my_final_branch_id = BranchId::new_invalid();
                 for (connector_id, branch_id) in global_solution.component_branches.iter().copied() {
                     if connector_id == ctx.id {
                         // This is our solution branch
                         my_final_branch_id = branch_id;
                         continue;
+                    }
                     let message = SyncMessageFancy{
                         sync_header: self.create_sync_header(ctx),
                         target_component_id: connector_id,
                         content: SyncContent::GlobalSolution(global_solution.clone()),
                     };
                     ctx.submit_message(MessageFancy::Sync(message));
+                }
                 debug_assert!(my_final_branch_id.is_valid());
                 return Some(my_final_branch_id);
             } else {
                 return None;
+            }
         } else {
             // Someone else is the leader
             let message = SyncMessageFancy{
                 sync_header: self.create_sync_header(ctx),
                 target_component_id: self.highest_connector_id,
                 content: SyncContent::LocalSolution(solution),
             };
             ctx.submit_message(MessageFancy::Sync(message));
             return None;
+        }
+    }
     #[inline]
     fn create_sync_header(&self, ctx: &ComponentCtxFancy) -> SyncHeader {
         return SyncHeader{
             sending_component_id: ctx.id,
             highest_component_id: self.highest_connector_id,
+        }
+    }
     fn forward_local_solutions(&mut self, ctx: &mut ComponentCtxFancy) {
@@ @@ -445,25 +472,25 @@ impl Consensus { @@
             };
             ctx.submit_message(MessageFancy::Sync(message));
+        }
+    }
+}
 // -----------------------------------------------------------------------------
 // Solution storage and algorithms
 // -----------------------------------------------------------------------------
 struct MatchedLocalSolution {
     final_branch_id: BranchId,
-    port_mapping: Vec<(ChannelId, BranchId)>,
+    channel_mapping: Vec<(ChannelId, BranchId)>,
     matches: Vec<ComponentMatches>,
+}
 struct ComponentMatches {
     target_id: ConnectorId,
     target_index: usize,
     match_indices: Vec<usize>, // of local solution in connector
+}
 struct ComponentPeer {
     target_id: ConnectorId,
     target_index: usize, // in array of global solution components
@@ @@ -487,25 +514,25 @@ impl SolutionCombiner { @@
         return Self{
             local: Vec::new(),
         };
+    }
     /// Adds a new local solution to the global solution storage. Will check the
     /// new local solutions for matching against already stored local solutions
     /// of peer connectors.
     fn add_solution_and_check_for_global_solution(&mut self, solution: LocalSolution) -> Option<GlobalSolution> {
         let component_id = solution.component;
         let solution = MatchedLocalSolution{
             final_branch_id: solution.final_branch_id,
-            port_mapping: solution.port_mapping,
+            channel_mapping: solution.port_mapping,
             matches: Vec::new(),
         };
         // Create an entry for the solution for the particular component
         let component_exists = self.local.iter_mut()
             .enumerate()
             .find(|(_, v)| v.component == component_id);
         let (component_index, solution_index, new_component) = match component_exists {
             Some((component_index, storage)) => {
                 // Entry for component exists, so add to solutions
                 let solution_index = storage.solutions.len();
                 storage.solutions.push(solution);
@@ @@ -520,75 +547,75 @@ impl SolutionCombiner { @@
                     peers: Vec::new(),
                     solutions: vec![solution],
                     all_peers_present: false,
                 });
                 (component_index, 0, true)
+            }
         };
         // If this is a solution of a component that is new to us, then we check
         // in the stored solutions which other components are peers of the new
         // one.
         if new_component {
-            let cur_ports = &self.local[component_index].solutions[0].port_mapping;
+            let cur_ports = &self.local[component_index].solutions[0].channel_mapping;
             let mut component_peers = Vec::new();
             // Find the matching components
             for (other_index, other_component) in self.local.iter().enumerate() {
                 if other_index == component_index {
                     // Don't match against ourselves
                     continue;
+                }
                 let mut matching_ports = Vec::new();
                 for (cur_port_id, _) in cur_ports {
                     for (other_port_id, _) in &other_component.solutions[0].port_mapping {
                         if cur_port_id == other_port_id {
                 let mut matching_channels = Vec::new();
                 for (cur_channel_id, _) in cur_ports {
                     for (other_channel_id, _) in &other_component.solutions[0].channel_mapping {
                         if cur_channel_id == other_channel_id {
                             // We have a shared port
-                            matching_ports.push(*port_id);
+                            matching_channels.push(*cur_channel_id);
+                        }
+                    }
+                }
-                if !matching_ports.is_empty() {
+                if !matching_channels.is_empty() {
                     // We share some ports
                     component_peers.push(ComponentPeer{
                         target_id: other_component.component,
                         target_index: other_index,
-                        involved_channels: matching_ports,
+                        involved_channels: matching_channels,
                     });
+                }
+            }
             let mut num_ports_in_peers = 0;
             for peer in component_peers {
                 num_ports_in_peers += peer.involved_channels.len();
+            }
             if num_ports_in_peers == cur_ports.len() {
                 // Newly added component has all required peers present
                 self.local[component_index].all_peers_present = true;
+            }
             // Add the found component pairing entries to the solution entries
             // for the two involved components
             for component_match in component_peers {
                 // Check the other component for having all peers present
                 let mut num_ports_in_peers = component_match.involved_channels.len();
                 let other_component = &mut self.local[component_match.target_index];
                 for existing_peer in &other_component.peers {
                     num_ports_in_peers += existing_peer.involved_channels.len();
+                }
-                if num_ports_in_peers == other_component.solutions[0].port_mapping.len() {
+                if num_ports_in_peers == other_component.solutions[0].channel_mapping.len() {
                     other_component.all_peers_present = true;
+                }
                 other_component.peers.push(ComponentPeer{
                     target_id: component_id,
                     target_index: component_index,
                     involved_channels: component_match.involved_channels.clone(),
                 });
                 let new_component = &mut self.local[component_index];
                 new_component.peers.push(component_match);
+            }
@@ @@ -600,26 +627,26 @@ impl SolutionCombiner { @@
         let mut new_component_matches = Vec::new();
         let cur_component = &self.local[component_index];
         let cur_solution = &cur_component.solutions[solution_index];
         for peer in &cur_component.peers {
             let mut new_solution_matches = Vec::new();
             let other_component = &self.local[peer.target_index];
             for (other_solution_index, other_solution) in other_component.solutions.iter().enumerate() {
                 // Check the port mappings between the pair of solutions.
                 let mut all_matched = true;
                 'mapping_check_loop: for (cur_port, cur_branch) in &cur_solution.port_mapping {
                     for (other_port, other_branch) in &other_solution.port_mapping {
                 'mapping_check_loop: for (cur_port, cur_branch) in &cur_solution.channel_mapping {
                     for (other_port, other_branch) in &other_solution.channel_mapping {
                         if cur_port == other_port {
                             if cur_branch == other_branch {
                                 // Same port mapping, go to next port
                                 break;
                             } else {
                                 // Different port mapping, not a match
                                 all_matched = false;
                                 break 'mapping_check_loop;
+                            }
+                        }
+                    }
+                }
@@ @@ -762,64 +789,64 @@ impl SolutionCombiner { @@
                         check_stack.push((cur_match.target_index, 0))
+                    }
+                }
+            }
             check_from = new_check_from;
+        }
         // Because of our earlier checking if we can have a solution at
         // all (all components have their peers), and the exit condition of the
         // while loop: if we're here, then we have a global solution
         debug_assert_eq!(check_stack.len(), self.local.len());
-        let mut global_solution = Vec::with_capacity(check_stack.len());
+        let mut final_branches = Vec::with_capacity(check_stack.len());
         for (component_index, solution_index) in check_stack.iter().copied() {
             let component = &self.local[component_index];
             let solution = &component.solutions[solution_index];
-            global_solution.push((component.component, solution.final_branch_id));
+            final_branches.push((component.component, solution.final_branch_id));
+        }
         // Just debugging here, TODO: @remove
-        let mut total_num_ports = 0;
+        let mut total_num_channels = 0;
         for (component_index, _) in check_stack.iter().copied() {
             let component = &self.local[component_index];
-            total_num_ports += component.solutions[0].port_mapping.len();
+            total_num_channels += component.solutions[0].channel_mapping.len();
+        }
         total_num_ports /= 2;
         let mut final_mapping = Vec::with_capacity(total_num_ports);
         total_num_channels /= 2;
         let mut final_mapping = Vec::with_capacity(total_num_channels);
         let mut total_num_checked = 0;
         for (component_index, solution_index) in check_stack.iter().copied() {
             let component = &self.local[component_index];
             let solution = &component.solutions[solution_index];
-            for (channel_id, branch_id) in solution.port_mapping.iter().copied() {
+            for (channel_id, branch_id) in solution.channel_mapping.iter().copied() {
                 match final_mapping.iter().find(|(v, _)| *v == channel_id) {
                     Some((_, encountered_branch_id)) => {
                         debug_assert_eq!(encountered_branch_id, branch_id);
                         total_num_checked += 1;
                     },
                     None => {
                         final_mapping.push((channel_id, branch_id));
+                    }
+                }
+            }
+        }
-        debug_assert_eq!(total_num_checked, total_num_ports);
+        debug_assert_eq!(total_num_checked, total_num_channels);
         return Some(GlobalSolution{
             branches: global_solution,
             port_mapping: final_mapping,
             component_branches: final_branches,
             channel_mapping: final_mapping,
         });
+    }
     /// Simple test if a solution is at all possible. If this returns true it
     /// does not mean there actually is a solution.
     fn can_have_solution(&self) -> bool {
         for component in &self.local {
             if !component.all_peers_present {
                 return false;
+            }
+        }
@@ @@ -833,31 +860,35 @@ impl SolutionCombiner { @@
     fn drain(&mut self) -> Vec<LocalSolution> {
         let mut reserve_len = 0;
         for component in &self.local {
             reserve_len += component.solutions.len();
+        }
         let mut solutions = Vec::with_capacity(reserve_len);
         for component in self.local.drain(..) {
             for solution in component.solutions {
                 solutions.push(LocalSolution{
                     component: component.component,
                     final_branch_id: solution.final_branch_id,
-                    port_mapping: solution.port_mapping,
+                    port_mapping: solution.channel_mapping,
                 });
+            }
+        }
         return solutions;
+    }
     fn clear(&mut self) {
         self.local.clear();
+    }
+}
 // -----------------------------------------------------------------------------
 // Generic Helpers
 // -----------------------------------------------------------------------------
 /// Recursively goes through the value group, attempting to find ports.
 /// Duplicates will only be added once.
 pub(crate) fn find_ports_in_value_group(value_group: &ValueGroup, ports: &mut Vec<PortIdLocal>) {
     // Helper to check a value for a port and recurse if needed.
     use crate::protocol::eval::Value;

src/runtime2/inbox.rs

➞

Show inline comments

@@ @@ -200,44 +200,12 @@ pub enum MessageContents { @@
     RequestCommit(SolutionMessage), // solution message, requesting participants to commit
     ConfirmCommit(SolutionMessage), // solution message, confirming a solution everyone committed to
     Control(ControlMessage),        // control message, handled by scheduler
     Ping,                           // ping message, intentionally waking up a connector (used for native connectors)
+}
 #[derive(Debug)]
 pub struct Message {
     pub sending_connector: ConnectorId,
     pub receiving_port: PortIdLocal, // may be invalid (in case of messages targeted at the connector)
     pub contents: MessageContents,
+}
@@ \ No newline at end of file @@
 /// The public inbox of a connector. The thread running the connector that owns
 /// this inbox may retrieved from it. Non-owning threads may only put new
 /// messages inside of it.
 // TODO: @Optimize, lazy concurrency. Probably ringbuffer with read/write heads.
 //  Should behave as a MPSC queue.
 pub struct PublicInbox {
     messages: Mutex<VecDeque<MessageFancy>>,
+}
 impl PublicInbox {
     pub fn new() -> Self {
         Self{
             messages: Mutex::new(VecDeque::new()),
+        }
+    }
     pub fn insert_message(&self, message: MessageFancy) {
         let mut lock = self.messages.lock().unwrap();
         lock.push_back(message);
+    }
     pub fn take_message(&self) -> Option<MessageFancy> {
         let mut lock = self.messages.lock().unwrap();
         return lock.pop_front();
+    }
     pub fn is_empty(&self) -> bool {
         let lock = self.messages.lock().unwrap();
         return lock.is_empty();
+    }
+}
@@ \ No newline at end of file @@

src/runtime2/inbox2.rs

➞

Show inline comments

 use std::sync::Mutex;
 use std::collections::VecDeque;
 use crate::protocol::eval::ValueGroup;
 use crate::runtime2::branch::BranchId;
 use crate::runtime2::ConnectorId;
 use crate::runtime2::consensus::{GlobalSolution, LocalSolution};
 use crate::runtime2::port::PortIdLocal;
 // TODO: Remove Debug derive from all types
 #[derive(Debug, Copy, Clone)]
 pub(crate) struct PortAnnotation {
     pub port_id: PortIdLocal,
     pub registered_id: Option<BranchId>,
@@ @@ -69,12 +72,44 @@ pub(crate) enum ControlContent { @@
     CloseChannel(PortIdLocal),
     Ack,
     Ping,
+}
 /// Combination of data message and control messages.
 #[derive(Debug)]
 pub(crate) enum MessageFancy {
     Data(DataMessageFancy),
     Sync(SyncMessageFancy),
     Control(ControlMessageFancy),
+}
 /// The public inbox of a connector. The thread running the connector that owns
 /// this inbox may retrieved from it. Non-owning threads may only put new
 /// messages inside of it.
 // TODO: @Optimize, lazy concurrency. Probably ringbuffer with read/write heads.
 //  Should behave as a MPSC queue.
 pub struct PublicInbox {
     messages: Mutex<VecDeque<MessageFancy>>,
+}
 impl PublicInbox {
     pub fn new() -> Self {
         Self{
             messages: Mutex::new(VecDeque::new()),
+        }
+    }
     pub fn insert_message(&self, message: MessageFancy) {
         let mut lock = self.messages.lock().unwrap();
         lock.push_back(message);
+    }
     pub fn take_message(&self) -> Option<MessageFancy> {
         let mut lock = self.messages.lock().unwrap();
         return lock.pop_front();
+    }
     pub fn is_empty(&self) -> bool {
         let lock = self.messages.lock().unwrap();
         return lock.is_empty();
+    }
+}
@@ \ No newline at end of file @@

src/runtime2/mod.rs

➞

Show inline comments

 // Structure of module
 mod runtime;
 mod messages;
 mod connector;
+// mod connector;
 mod branch;
 mod native;
 mod port;
 mod scheduler;
 mod inbox;
+// mod inbox;
 mod consensus;
 mod inbox2;
 #[cfg(test)] mod tests;
 mod connector2;
 // Imports
 use std::collections::VecDeque;
 use std::sync::{Arc, Condvar, Mutex, RwLock};
 use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
 use std::thread::{self, JoinHandle};
 use crate::collections::RawVec;
 use crate::ProtocolDescription;
 use inbox::Message;
 use connector2::{ConnectorPDL, ConnectorPublic, ConnectorScheduling};
 use scheduler::{Scheduler, ControlMessageHandler};
+use scheduler::{Scheduler, ComponentCtxFancy, SchedulerCtx, ControlMessageHandler};
 use native::{Connector, ConnectorApplication, ApplicationInterface};
 use crate::runtime2::inbox2::MessageFancy;
 use crate::runtime2::port::{ChannelId, Port, PortState};
 use crate::runtime2::scheduler::{ComponentCtxFancy, SchedulerCtx};
 use inbox2::MessageFancy;
 use port::{ChannelId, Port, PortState};
 /// A kind of token that, once obtained, allows mutable access to a connector.
 /// We're trying to use move semantics as much as possible: the owner of this
 /// key is the only one that may execute the connector's code.
 pub(crate) struct ConnectorKey {
     pub index: u32, // of connector
+}
 impl ConnectorKey {
     /// Downcasts the `ConnectorKey` type, which can be used to obtain mutable
     /// access, to a "regular ID" which can be used to obtain immutable access.
     #[inline]

src/runtime2/native.rs

➞

Show inline comments

 use std::collections::VecDeque;
 use std::sync::{Arc, Mutex, Condvar};
 use std::sync::atomic::Ordering;
 use crate::protocol::ComponentCreationError;
 use crate::protocol::eval::ValueGroup;
 use super::{ConnectorKey, ConnectorId, RuntimeInner};
-use super::scheduler::{SchedulerCtx, ComponentCtxFancy, ReceivedMessage};
 use super::scheduler::{SchedulerCtx, ComponentCtxFancy};
 use super::port::{Port, PortIdLocal, Channel, PortKind};
 use super::connector::{Branch, ConnectorScheduling, ConnectorPDL};
 use super::connector::find_ports_in_value_group;
 use super::inbox::{Message, MessageContents};
 use super::branch::{Branch};
 use super::consensus::find_ports_in_value_group;
 use super::connector2::{ConnectorScheduling, ConnectorPDL};
 use super::inbox2::{MessageFancy, ControlContent, ControlMessageFancy};
 /// Generic connector interface from the scheduler's point of view.
 pub(crate) trait Connector {
     /// Should run the connector's behaviour up until the next blocking point.
     /// One should generally request and handle new messages from the component
     /// context. Then perform any logic the component has to do, and in the
     /// process perhaps queue up some state changes using the same context.
     fn run(&mut self, sched_ctx: SchedulerCtx, comp_ctx: &mut ComponentCtxFancy) -> ConnectorScheduling;
+}
 type SyncDone = Arc<(Mutex<bool>, Condvar)>;
 type JobQueue = Arc<Mutex<VecDeque<ApplicationJob>>>;
 enum ApplicationJob {
     NewChannel((Port, Port)),
     NewConnector(ConnectorPDL),
+    NewConnector(ConnectorPDL, Vec<PortIdLocal>),
     Shutdown,
+}
 /// The connector which an application can directly interface with. Once may set
 /// up the next synchronous round, and retrieve the data afterwards.
 pub struct ConnectorApplication {
     sync_done: SyncDone,
     job_queue: JobQueue,
+}
 impl ConnectorApplication {
     pub(crate) fn new(runtime: Arc<RuntimeInner>) -> (Self, ApplicationInterface) {
@@ @@ -48,44 +49,43 @@ impl ConnectorApplication { @@
         };
         let interface = ApplicationInterface::new(sync_done, job_queue, runtime);
         return (connector, interface);
+    }
+}
 impl Connector for ConnectorApplication {
     fn run(&mut self, _sched_ctx: SchedulerCtx, comp_ctx: &mut ComponentCtxFancy) -> ConnectorScheduling {
         // Handle any incoming messages if we're participating in a round
         while let Some(message) = comp_ctx.read_next_message() {
             match message {
                 ReceivedMessage::Data(_) => todo!("data message in API connector"),
                 ReceivedMessage::Sync(_) | ReceivedMessage::RequestCommit(_) | ReceivedMessage::ConfirmCommit(_) => {
                     todo!("sync message in API connector");
+                }
                 MessageFancy::Data(_) => todo!("data message in API connector"),
                 MessageFancy::Sync(_)  => todo!("sync message in API connector"),
                 MessageFancy::Control(_) => todo!("impossible control message"),
+            }
+        }
         // Handle requests coming from the API
+        {
             let mut queue = self.job_queue.lock().unwrap();
             while let Some(job) = queue.pop_front() {
                 match job {
                     ApplicationJob::NewChannel((endpoint_a, endpoint_b)) => {
                         println!("DEBUG: API adopting ports");
                         comp_ctx.push_port(endpoint_a);
                         comp_ctx.push_port(endpoint_b);
+                    }
                     ApplicationJob::NewConnector(connector) => {
+                    ApplicationJob::NewConnector(connector, initial_ports) => {
                         println!("DEBUG: API creating connector");
                         comp_ctx.push_component(connector);
+                        comp_ctx.push_component(connector, initial_ports);
                     },
                     ApplicationJob::Shutdown => {
                         debug_assert!(queue.is_empty());
                         return ConnectorScheduling::Exit;
+                    }
+                }
+            }
+        }
         return ConnectorScheduling::NotNow;
+    }
+}
@@ @@ -130,44 +130,43 @@ impl ApplicationInterface { @@
+    }
     /// Creates a new connector. Note that it is not scheduled immediately, but
     /// depends on the `ApplicationConnector` to run, followed by the created
     /// connector being scheduled.
     // TODO: Yank out scheduler logic for common use.
     pub fn create_connector(&mut self, module: &str, routine: &str, arguments: ValueGroup) -> Result<(), ComponentCreationError> {
         // Retrieve ports and make sure that we own the ones that are currently
         // specified. This is also checked by the scheduler, but that is done
         // asynchronously.
         let mut initial_ports = Vec::new();
         find_ports_in_value_group(&arguments, &mut initial_ports);
         for port_to_remove in &initial_ports {
             match self.owned_ports.iter().position(|v| v == port_to_remove) {
                 Some(index_to_remove) => {
                     // We own the port, so continue
                     self.owned_ports.remove(index_to_remove);
                 },
                 None => {
                     // We don't own the port
         for initial_port in &initial_ports {
             if !self.owned_ports.iter().any(|v| v == initial_port) {
                 return Err(ComponentCreationError::UnownedPort);
+            }
+        }
         // We own all ports, so remove them on this side
         for initial_port in &initial_ports {
             let position = self.owned_ports.iter().position(|v| *v == initial_port).unwrap();
             self.owned_ports.remove(position);
+        }
         let state = self.runtime.protocol_description.new_component_v2(module.as_bytes(), routine.as_bytes(), arguments)?;
-        let connector = ConnectorPDL::new(Branch::new_initial_branch(state), initial_ports);
         let connector = ConnectorPDL::new(state);
         // Put on job queue
+        {
             let mut queue = self.job_queue.lock().unwrap();
             queue.push_back(ApplicationJob::NewConnector(connector));
+            queue.push_back(ApplicationJob::NewConnector(connector, initial_ports));
+        }
         self.wake_up_connector_with_ping();
         return Ok(());
+    }
     /// Check if the next sync-round is finished.
     pub fn try_wait(&self) -> bool {
         let (is_done, _) = &*self.sync_done;
         let lock = is_done.lock().unwrap();
         return *lock;
@@ @@ -178,29 +177,29 @@ impl ApplicationInterface { @@
         let (is_done, condition) = &*self.sync_done;
         let lock = is_done.lock().unwrap();
         condition.wait_while(lock, |v| !*v).unwrap(); // wait while not done
+    }
     /// Called by runtime to set associated connector's ID.
     pub(crate) fn set_connector_id(&mut self, id: ConnectorId) {
         self.connector_id = id;
+    }
     fn wake_up_connector_with_ping(&self) {
         let connector = self.runtime.get_component_public(self.connector_id);
         connector.inbox.insert_message(Message{
             sending_connector: ConnectorId::new_invalid(),
             receiving_port: PortIdLocal::new_invalid(),
             contents: MessageContents::Ping,
         });
         connector.inbox.insert_message(MessageFancy::Control(ControlMessageFancy{
             id: 0,
             sending_component_id: self.connector_id,
             content: ControlContent::Ack
         }));
         let should_wake_up = connector.sleeping
             .compare_exchange(true, false, Ordering::SeqCst, Ordering::Acquire)
             .is_ok();
         if should_wake_up {
             println!("DEBUG: Waking up connector");
             let key = unsafe{ ConnectorKey::from_id(self.connector_id) };
             self.runtime.push_work(key);
         } else {
             println!("DEBUG: NOT waking up connector");
+        }

src/runtime2/scheduler.rs

➞

Show inline comments

@@ @@ -145,47 +145,47 @@ impl Scheduler { @@
                             // Need to change port target
                             let port = scheduled.ctx_fancy.get_port_mut_by_id(port_id).unwrap();
                             port.peer_connector = new_target_connector_id;
                             // Note: for simplicity we program the scheduler to always finish
                             // running a connector with an empty outbox. If this ever changes
                             // then accepting the "port peer changed" message implies we need
                             // to change the recipient of the message in the outbox.
                             debug_assert!(scheduled.ctx_fancy.outbox.is_empty());
                             // And respond with an Ack
                             let ack_message = MessageFancy::Control(ControlMessageFancy{
-                                id: content.id,
+                                id: message.id,
                                 sending_component_id: connector_id,
                                 content: ControlContent::Ack,
                             });
                             self.debug_conn(connector_id, &format!("Sending message [pp ack]\n --- {:?}", ack_message));
                             self.runtime.send_message(message.sending_component_id, ack_message);
                         },
                         ControlContent::CloseChannel(port_id) => {
                             // Mark the port as being closed
                             let port = scheduled.ctx_fancy.get_port_mut_by_id(port_id).unwrap();
                             port.state = PortState::Closed;
                             // Send an Ack
                             let ack_message = MessageFancy::Control(ControlMessageFancy{
-                                id: content.id,
+                                id: message.id,
                                 sending_component_id: connector_id,
                                 content: ControlContent::Ack,
                             });
                             self.debug_conn(connector_id, &format!("Sending message [cc ack] \n --- {:?}", ack_message));
                             self.runtime.send_message(message.sending_component_id, ack_message);
                         },
                         ControlContent::Ack => {
-                            scheduled.router.handle_ack(content.id);
+                            scheduled.router.handle_ack(message.id);
                         },
                         ControlContent::Ping => {},
+                    }
                 },
                 _ => {
                     // All other cases have to be handled by the component
                     scheduled.ctx_fancy.inbox_messages.push(message);
+                }
+            }
+        }
+    }
@@ @@ -497,25 +497,25 @@ pub(crate) struct MessagesIter<'a> { @@
 impl<'a> Iterator for MessagesIter<'a> {
     type Item = &'a DataMessageFancy;
     fn next(&mut self) -> Option<Self::Item> {
         // Loop until match is found or at end of messages
         while self.next_index < self.max_index {
             let message = &self.messages[self.next_index];
             if let MessageFancy::Data(message) = &message {
                 if message.data_header.target_port == self.match_port_id {
                     // Found a match
                     self.next_index += 1;
-                    return Some(data_message);
+                    return Some(message);
+                }
             } else {
                 // Unreachable because:
                 //  1. We only iterate over messages that were previously retrieved by `read_next_message`.
                 //  2. Inbox does not contain control/ping messages.
                 //  3. If `read_next_message` encounters anything else than a data message, it is removed from the inbox.
                 unreachable!();
+            }
             self.next_index += 1;
+        }

0 comments (0 inline, 0 general)