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

Changeset - 6f608e19fa8a

Parent rev.

Child rev.

[Not reviewed]

0 5 0

MH - 4 years ago 2021-11-22 18:52:41
contact@maxhenger.nl

Fix part of the error-handling code

5 files changed with 19 insertions and 10 deletions:

src/runtime2/consensus.rs

src/runtime2/mod.rs

src/runtime2/scheduler.rs

src/runtime2/tests/mod.rs

src/runtime2/tests/sync_failure.rs

0 comments (0 inline, 0 general)

src/runtime2/consensus.rs

➞

Show inline comments

 use crate::collections::VecSet;
 use crate::protocol::eval::ValueGroup;
 use super::ConnectorId;
 use super::branch::BranchId;
 use super::port::{ChannelId, PortIdLocal};
 use super::inbox::{
     Message, DataHeader, SyncHeader, ChannelAnnotation, BranchMarker,
     DataMessage,
     SyncCompMessage, SyncCompContent,
     SyncPortMessage, SyncPortContent,
     SyncControlMessage, SyncControlContent
 };
 use super::scheduler::{ComponentCtx, ComponentPortChange};
 struct BranchAnnotation {
     channel_mapping: Vec<ChannelAnnotation>,
     cur_marker: BranchMarker,
+}
 #[derive(Debug)]
 pub(crate) struct LocalSolution {
     component: ConnectorId,
     final_branch_id: BranchId,
     port_mapping: Vec<(ChannelId, BranchMarker)>,
+}
 #[derive(Debug, Clone)]
 pub(crate) struct GlobalSolution {
     component_branches: Vec<(ConnectorId, BranchId)>,
     channel_mapping: Vec<(ChannelId, BranchMarker)>, // TODO: This can go, is debugging info
+}
 #[derive(Debug, PartialEq, Eq)]
 pub enum RoundConclusion {
     Failure,
     Success(BranchId),
+}
 // -----------------------------------------------------------------------------
 // Consensus
 // -----------------------------------------------------------------------------
 struct Peer {
     id: ConnectorId,
     encountered_this_round: bool,
     expected_sync_round: u32,
+}
 /// The consensus algorithm. Currently only implemented to find the component
 /// with the highest ID within the sync region and letting it handle all the
 /// local solutions.
 ///
 /// The type itself serves as an experiment to see how code should be organized.
 // TODO: Flatten all datastructures
 // TODO: Have a "branch+port position hint" in case multiple operations are
 //  performed on the same port to prevent repeated lookups
 // TODO: A lot of stuff should be batched. Like checking all the sync headers
 //  and sending "I have a higher ID" messages. Should reduce locking by quite a
 //  bit.
 pub(crate) struct Consensus {
     // --- State that is cleared after each round
     // Local component's state
     highest_connector_id: ConnectorId,
     branch_annotations: Vec<BranchAnnotation>, // index is branch ID
     branch_markers: Vec<BranchId>, // index is branch marker, maps to branch
     // Gathered state from communication
     encountered_ports: VecSet<PortIdLocal>, // to determine if we should send "port remains silent" messages.
     solution_combiner: SolutionCombiner,
     handled_wave: bool, // encountered notification wave in this round
     conclusion: Option<RoundConclusion>,
     ack_remaining: u32,
     // --- Persistent state
     peers: Vec<Peer>,
     sync_round: u32,
     // --- Workspaces
     workspace_ports: Vec<PortIdLocal>,
+}
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub(crate) enum Consistency {
     Valid,
     Inconsistent,
+}
 impl Consensus {
     pub fn new() -> Self {
         return Self {
             highest_connector_id: ConnectorId::new_invalid(),
             branch_annotations: Vec::new(),
             branch_markers: Vec::new(),
             encountered_ports: VecSet::new(),
             solution_combiner: SolutionCombiner::new(),
             handled_wave: false,
             conclusion: None,
             ack_remaining: 0,
             peers: Vec::new(),
             sync_round: 0,
             workspace_ports: Vec::new(),
+        }
+    }
     // --- Controlling sync round and branches
     /// Returns whether the consensus algorithm is running in sync mode
     pub fn is_in_sync(&self) -> bool {
         return !self.branch_annotations.is_empty();
+    }
     /// TODO: Remove this once multi-fire is in place
     #[deprecated]
     pub fn get_annotation(&self, branch_id: BranchId, channel_id: PortIdLocal) -> &ChannelAnnotation {
         let branch = &self.branch_annotations[branch_id.index as usize];
         let port = branch.channel_mapping.iter().find(|v| v.channel_id.index == channel_id.index).unwrap();
         return port;
+    }
     /// Sets up the consensus algorithm for a new synchronous round. The
     /// provided ports should be the ports the component owns at the start of
     /// the sync round.
     pub fn start_sync(&mut self, ctx: &ComponentCtx) {
         debug_assert!(!self.highest_connector_id.is_valid());
         debug_assert!(self.branch_annotations.is_empty());
         debug_assert!(self.solution_combiner.local.is_empty());
         // We'll use the first "branch" (the non-sync one) to store our ports,
         // this allows cloning if we created a new branch.
         self.branch_annotations.push(BranchAnnotation{
             channel_mapping: ctx.get_ports().iter()
                 .map(|v| ChannelAnnotation {
                     channel_id: v.channel_id,
                     registered_id: None,
                     expected_firing: None,
                 })
                 .collect(),
             cur_marker: BranchMarker::new_invalid(),
         });
         self.branch_markers.push(BranchId::new_invalid());
         self.highest_connector_id = ctx.id;
+    }
     /// Notifies the consensus algorithm that a new branch has appeared. Must be
     /// called for each forked branch in the execution tree.
     pub fn notify_of_new_branch(&mut self, parent_branch_id: BranchId, new_branch_id: BranchId) {
         // If called correctly. Then each time we are notified the new branch's
         // index is the length in `branch_annotations`.
         debug_assert!(self.branch_annotations.len() == new_branch_id.index as usize);
         let parent_branch_annotations = &self.branch_annotations[parent_branch_id.index as usize];
         let new_marker = BranchMarker::new(self.branch_markers.len() as u32);
         let new_branch_annotations = BranchAnnotation{
             channel_mapping: parent_branch_annotations.channel_mapping.clone(),
             cur_marker: new_marker,
         };
         self.branch_annotations.push(new_branch_annotations);
         self.branch_markers.push(new_branch_id);
+    }
     /// Notifies the consensus algorithm that a particular branch has
     /// encountered an unrecoverable error.
     pub fn notify_of_fatal_branch(&mut self, failed_branch_id: BranchId, ctx: &mut ComponentCtx) -> Option<RoundConclusion> {
         debug_assert!(self.is_in_sync());
         // Check for trivial case, where branch has not yet communicated within
         // the consensus algorithm
         let branch = &self.branch_annotations[failed_branch_id.index as usize];
         if branch.channel_mapping.iter().all(|v| v.registered_id.is_none()) {
             return Some(RoundConclusion::Failure);
+        }
         // We need to go through the hassle of notifying all participants in the
         // sync round that we've encountered an error.
         // --- notify leader
         let maybe_conclusion = self.send_to_leader_or_handle_as_leader(SyncCompContent::LocalFailure, ctx);
         // --- initiate discovery wave (to let leader know about all components)
         self.handled_wave = true;
         for mapping in &self.branch_annotations[0].channel_mapping {
             let channel_id = mapping.channel_id;
             let port_info = ctx.get_port_by_channel_id(channel_id).unwrap();
             let message = SyncPortMessage{
                 sync_header: self.create_sync_header(ctx),
                 source_port: port_info.self_id,
                 target_port: port_info.peer_id,
                 content: SyncPortContent::NotificationWave,
             };
             ctx.submit_message(Message::SyncPort(message));
+        }
         return maybe_conclusion;
+    }
     /// Notifies the consensus algorithm that a branch has reached the end of
     /// the sync block. A final check for consistency will be performed that the
     /// caller has to handle. Note that
     pub fn notify_of_finished_branch(&self, branch_id: BranchId) -> Consistency {
         debug_assert!(self.is_in_sync());
         let branch = &self.branch_annotations[branch_id.index as usize];
         for mapping in &branch.channel_mapping {
             match mapping.expected_firing {
                 Some(expected) => {
                     if expected != mapping.registered_id.is_some() {
                         // Inconsistent speculative state and actual state
                         debug_assert!(mapping.registered_id.is_none()); // because if we did fire on a silent port, we should've caught that earlier
                         return Consistency::Inconsistent;
+                    }
                 },
                 None => {},
+            }
+        }
         return Consistency::Valid;
+    }
     /// Notifies the consensus algorithm that a particular branch has assumed
     /// a speculative value for its port mapping.
     pub fn notify_of_speculative_mapping(&mut self, branch_id: BranchId, port_id: PortIdLocal, does_fire: bool, ctx: &ComponentCtx) -> Consistency {
         debug_assert!(self.is_in_sync());
         let port_desc = ctx.get_port_by_id(port_id).unwrap();
         let channel_id = port_desc.channel_id;
         let branch = &mut self.branch_annotations[branch_id.index as usize];
         for mapping in &mut branch.channel_mapping {
             if mapping.channel_id == channel_id {
                 match mapping.expected_firing {
                     None => {
                         // Not yet mapped, perform speculative mapping
                         mapping.expected_firing = Some(does_fire);
                         return Consistency::Valid;
                     },
                     Some(current) => {
                         // Already mapped
                         if current == does_fire {
                             return Consistency::Valid;
                         } else {
                             return Consistency::Inconsistent;
+                        }
+                    }
+                }
+            }
+        }
         unreachable!("notify_of_speculative_mapping called with unowned port");
+    }
     /// Generates a new local solution from a finished branch. If the component
     /// is not the leader of the sync region then it will be sent to the
     /// appropriate component. If it is the leader then there is a chance that
     /// this solution completes a global solution. In that case the solution
     /// branch ID will be returned.
     pub(crate) fn handle_new_finished_sync_branch(&mut self, branch_id: BranchId, ctx: &mut ComponentCtx) -> Option<RoundConclusion> {
         // Turn the port mapping into a local solution
         let source_mapping = &self.branch_annotations[branch_id.index as usize].channel_mapping;
         let mut target_mapping = Vec::with_capacity(source_mapping.len());
         for port in source_mapping {
             // Note: if the port is silent, and we've never communicated
             // over the port, then we need to do so now, to let the peer
             // component know about our sync leader state.
             let port_desc = ctx.get_port_by_channel_id(port.channel_id).unwrap();
             let self_port_id = port_desc.self_id;
             let peer_port_id = port_desc.peer_id;
             let channel_id = port_desc.channel_id;
             if !self.encountered_ports.contains(&self_port_id) {
                 ctx.submit_message(Message::SyncPort(SyncPortMessage {
                     sync_header: SyncHeader{
                         sending_component_id: ctx.id,
                         highest_component_id: self.highest_connector_id,
                         sync_round: self.sync_round
                     },
                     source_port: self_port_id,
                     target_port: peer_port_id,
                     content: SyncPortContent::SilentPortNotification,
                 }));
                 self.encountered_ports.push(self_port_id);
+            }
             target_mapping.push((
                 channel_id,
                 port.registered_id.unwrap_or(BranchMarker::new_invalid())
             ));
+        }
         let local_solution = LocalSolution{
             component: ctx.id,
             final_branch_id: branch_id,
             port_mapping: target_mapping,
         };
         let maybe_conclusion = self.send_to_leader_or_handle_as_leader(SyncCompContent::LocalSolution(local_solution), ctx);
         return maybe_conclusion;
+    }
     /// Notifies the consensus algorithm about the chosen branch to commit to
     /// memory (may be the invalid "start" branch)
     pub fn end_sync(&mut self, branch_id: BranchId, final_ports: &mut Vec<ComponentPortChange>) {
         debug_assert!(self.is_in_sync());
         // TODO: Handle sending and receiving ports
         // Set final ports
         let branch = &self.branch_annotations[branch_id.index as usize];
         // Clear out internal storage to defaults
         self.highest_connector_id = ConnectorId::new_invalid();
         self.branch_annotations.clear();
         self.branch_markers.clear();
         self.encountered_ports.clear();
         self.solution_combiner.clear();
         self.handled_wave = false;
         self.conclusion = None;
         self.ack_remaining = 0;
         // And modify persistent storage
         self.sync_round += 1;
         for peer in self.peers.iter_mut() {
             peer.encountered_this_round = false;
             peer.expected_sync_round += 1;
+        }
+    }
     // --- 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 ComponentCtx) -> (SyncHeader, DataHeader) {
         debug_assert!(self.is_in_sync());
         let branch = &mut self.branch_annotations[branch_id.index as usize];
         let port_info = ctx.get_port_by_id(source_port_id).unwrap();
         if cfg!(debug_assertions) {
             // Check for consistent mapping
             let port = branch.channel_mapping.iter()
                 .find(|v| v.channel_id == port_info.channel_id)
                 .unwrap();
             debug_assert!(port.expected_firing == None || port.expected_firing == Some(true));
+        }
         // Check for ports that are being sent
         debug_assert!(self.workspace_ports.is_empty());
         find_ports_in_value_group(content, &mut self.workspace_ports);
         if !self.workspace_ports.is_empty() {
             todo!("handle sending ports");
             self.workspace_ports.clear();
+        }
         // Construct data header
         // TODO: Handle multiple firings. Right now we just assign the current
         //  branch to the `None` value because we know we can only send once.
         let data_header = DataHeader{
             expected_mapping: branch.channel_mapping.clone(),
             expected_mapping: branch.channel_mapping.iter()
                 .filter(|v| v.registered_id.is_some() || v.channel_id == port_info.channel_id)
                 .copied()
                 .collect(),
             sending_port: port_info.self_id,
             target_port: port_info.peer_id,
             new_mapping: branch.cur_marker,
         };
         // Update port mapping
         for mapping in &mut branch.channel_mapping {
             if mapping.channel_id == port_info.channel_id {
                 mapping.expected_firing = Some(true);
                 mapping.registered_id = Some(branch.cur_marker);
+            }
+        }
         // Update branch marker
         let new_marker = BranchMarker::new(self.branch_markers.len() as u32);
         branch.cur_marker = new_marker;
         self.branch_markers.push(branch_id);
         self.encountered_ports.push(source_port_id);
         return (self.create_sync_header(ctx), data_header);
+    }
     /// Handles a new data message by handling the sync header. The caller is
     /// responsible for checking for branches that might be able to receive
     /// the message.
     pub fn handle_new_data_message(&mut self, message: &DataMessage, ctx: &mut ComponentCtx) -> bool {
         let handled = self.handle_received_sync_header(&message.sync_header, ctx);
         if handled {
             self.encountered_ports.push(message.data_header.target_port);
+        }
         return handled;
+    }
     /// Handles a new sync message by handling the sync header and the contents
     /// of the message. Returns `Some` with the branch ID of the global solution
     /// if the sync solution has been found.
     pub fn handle_new_sync_comp_message(&mut self, message: SyncCompMessage, ctx: &mut ComponentCtx) -> Option<RoundConclusion> {
         if !self.handle_received_sync_header(&message.sync_header, ctx) {
             return None;
+        }
         // And handle the contents
         debug_assert_eq!(message.target_component_id, ctx.id);
         match &message.content {
             SyncCompContent::LocalFailure |
             SyncCompContent::LocalSolution(_) |
             SyncCompContent::PartialSolution(_) |
             SyncCompContent::AckFailure |
             SyncCompContent::Presence(_, _) => {
                 // Needs to be handled by the leader
                 return self.send_to_leader_or_handle_as_leader(message.content, ctx);
             },
             SyncCompContent::GlobalSolution(solution) => {
                 // Found a global solution
                 debug_assert_ne!(self.highest_connector_id, ctx.id); // not the leader
                 let (_, branch_id) = solution.component_branches.iter()
                     .find(|(component_id, _)| *component_id == ctx.id)
                     .unwrap();
                 return Some(RoundConclusion::Success(*branch_id));
             },
             SyncCompContent::GlobalFailure => {
                 // Global failure of round, send Ack to leader
                 debug_assert_ne!(self.highest_connector_id, ctx.id); // not the leader
                 let _result = self.send_to_leader_or_handle_as_leader(SyncCompContent::AckFailure, ctx);
                 debug_assert!(_result.is_none());
                 return Some(RoundConclusion::Failure);
             },
             SyncCompContent::Notification => {
                 // We were just interested in the sync header we handled above
                 return None;
+            }
+        }
+    }
     pub fn handle_new_sync_port_message(&mut self, message: SyncPortMessage, ctx: &mut ComponentCtx) {
         if !self.handle_received_sync_header(&message.sync_header, ctx) {
             return;
+        }
         debug_assert!(self.is_in_sync());
         debug_assert!(ctx.get_port_by_id(message.target_port).is_some());
         match message.content {
             SyncPortContent::SilentPortNotification => {
                 // The point here is to let us become part of the sync round and
                 // take note of the leader in case all of our ports are silent.
                 self.encountered_ports.push(message.target_port);
+            }
             SyncPortContent::NotificationWave => {
                 // Wave to discover everyone in the network, handling sync
                 // header takes care of leader discovery, here we need to make
                 // sure we propagate the wave
                 if self.handled_wave {
                     return;
+                }
                 self.handled_wave = true;
                 // Propagate wave to all peers except the one that has sent us
                 // the wave.
                 for mapping in &self.branch_annotations[0].channel_mapping {
                     let channel_id = mapping.channel_id;
                     let port_desc = ctx.get_port_by_channel_id(channel_id).unwrap();
                     if port_desc.self_id == message.target_port {
                         // Wave came from this port, no need to send one back
                         continue;
+                    }
                     let message = SyncPortMessage{
                         sync_header: self.create_sync_header(ctx),
                         source_port: port_desc.self_id,
                         target_port: port_desc.peer_id,
                         content: SyncPortContent::NotificationWave,
                     };
                     ctx.submit_message(Message::SyncPort(message)).unwrap();
+                }
+            }
+        }
+    }
     pub fn handle_new_sync_control_message(&mut self, message: SyncControlMessage, ctx: &mut ComponentCtx) -> Option<RoundConclusion> {
         if message.in_response_to_sync_round < self.sync_round {
             // Old message
             return None
+        }
         match message.content {
             SyncControlContent::ChannelIsClosed(_) => {
                 return Some(RoundConclusion::Failure);
+            }
+        }
+    }
     pub fn notify_of_received_message(&mut self, branch_id: BranchId, message: &DataMessage, ctx: &ComponentCtx) {
         debug_assert!(self.branch_can_receive(branch_id, message));
         let target_port = ctx.get_port_by_id(message.data_header.target_port).unwrap();
         let branch = &mut self.branch_annotations[branch_id.index as usize];
         for mapping in &mut branch.channel_mapping {
             if mapping.channel_id == target_port.channel_id {
                 // Found the port in which the message should be inserted
                 mapping.registered_id = Some(message.data_header.new_mapping);
                 // Check for sent ports
                 debug_assert!(self.workspace_ports.is_empty());
                 find_ports_in_value_group(&message.content, &mut self.workspace_ports);
                 if !self.workspace_ports.is_empty() {
                     todo!("handle received ports");
                     self.workspace_ports.clear();
+                }
                 return;
+            }
+        }
         // If here, then the branch didn't actually own the port? Means the
         // caller made a mistake
         unreachable!("incorrect notify_of_received_message");
+    }
     /// Matches the mapping between the branch and the data message. If they
     /// match then the branch can receive the message.
     pub fn branch_can_receive(&self, branch_id: BranchId, message: &DataMessage) -> bool {
         if let Some(peer) = self.peers.iter().find(|v| v.id == message.sync_header.sending_component_id) {
             if message.sync_header.sync_round < peer.expected_sync_round {
                 return false;
+            }
+        }
         let annotation = &self.branch_annotations[branch_id.index as usize];
         for expected in &message.data_header.expected_mapping {
             // If we own the port, then we have an entry in the
             // annotation, check if the current mapping matches
             for current in &annotation.channel_mapping {
                 if expected.channel_id == current.channel_id {
                     if expected.registered_id != current.registered_id {
                         // IDs do not match, we cannot receive the
                         // message in this branch
                         return false;
+                    }
+                }
+            }
+        }
         return true;
+    }
     // --- Internal helpers
     fn handle_received_sync_header(&mut self, sync_header: &SyncHeader, ctx: &mut ComponentCtx) -> bool {
         debug_assert!(sync_header.sending_component_id != ctx.id); // not sending to ourselves
         if !self.handle_peer(sync_header) {
             // We can drop this package
             return false;
+        }
         if sync_header.highest_component_id > self.highest_connector_id {
             // Sender has higher component ID. So should be the target of our
             // messages. We should also let all of our peers know
             self.highest_connector_id = sync_header.highest_component_id;
             for peer in self.peers.iter() {
                 if peer.id == sync_header.sending_component_id || !peer.encountered_this_round {
                     // Don't need to send it to this one
                     continue
+                }
                 let message = SyncCompMessage {
                     sync_header: self.create_sync_header(ctx),
                     target_component_id: peer.id,
                     content: SyncCompContent::Notification,
                 };
                 ctx.submit_message(Message::SyncComp(message));
+            }
             // But also send our locally combined solution
             self.forward_local_data_to_new_leader(ctx);
         } else if sync_header.highest_component_id < self.highest_connector_id {
             // Sender has lower leader ID, so it should know about our higher
             // one.
             let message = SyncCompMessage {
                 sync_header: self.create_sync_header(ctx),
                 target_component_id: sync_header.sending_component_id,
                 content: SyncCompContent::Notification
             };
             ctx.submit_message(Message::SyncComp(message));
         } // else: exactly equal, so do nothing
         return true;
+    }
     /// Handles a (potentially new) peer. Returns `false` if the provided sync
     /// number is different then the expected one.
     fn handle_peer(&mut self, sync_header: &SyncHeader) -> bool {
         let position = self.peers.iter().position(|v| v.id == sync_header.sending_component_id);
         match position {
             Some(index) => {
                 let entry = &mut self.peers[index];
                 entry.encountered_this_round = true;
                 // TODO: Proper handling of potential overflow
                 if sync_header.sync_round >= entry.expected_sync_round {
                     entry.expected_sync_round = sync_header.sync_round;
                     return true;
                 } else {
                     return false;
+                }
             },
             None => {
                 self.peers.push(Peer{
                     id: sync_header.sending_component_id,
                     encountered_this_round: true,
                     expected_sync_round: sync_header.sync_round,
                 });
                 return true;
+            }
+        }
+    }
     /// Sends a message towards the leader, if already the leader then the
     /// message will be handled immediately.
     fn send_to_leader_or_handle_as_leader(&mut self, content: SyncCompContent, ctx: &mut ComponentCtx) -> Option<RoundConclusion> {
         if self.highest_connector_id == ctx.id {
             // We are the leader
             match content {
                 SyncCompContent::LocalFailure => {
                     if self.solution_combiner.mark_failure_and_check_for_global_failure() {
                         return self.handle_global_failure_as_leader(ctx);
+                    }
                 },
                 SyncCompContent::LocalSolution(local_solution) => {
                     if let Some(global_solution) = self.solution_combiner.add_solution_and_check_for_global_solution(local_solution) {
                         return self.handle_global_solution_as_leader(global_solution, ctx);
+                    }
                 },
                 SyncCompContent::PartialSolution(partial_solution) => {
                     if let Some(conclusion) = self.solution_combiner.combine(partial_solution) {
                         match conclusion {
                             LeaderConclusion::Solution(global_solution) => {
                                 return self.handle_global_solution_as_leader(global_solution, ctx);
                             },
                             LeaderConclusion::Failure => {
                                 return self.handle_global_failure_as_leader(ctx);
+                            }
+                        }
+                    }
                 },
                 SyncCompContent::Presence(component_id, presence) => {
                     if self.solution_combiner.add_presence_and_check_for_global_failure(component_id, &presence) {
                         return self.handle_global_failure_as_leader(ctx);
+                    }
                 },
                 SyncCompContent::AckFailure => {
                     debug_assert_eq!(Some(RoundConclusion::Failure), self.conclusion);
                     debug_assert!(self.ack_remaining > 0);
                     self.ack_remaining -= 1;
                     if self.ack_remaining == 0 {
                         return Some(RoundConclusion::Failure);
+                    }
+                }
                 SyncCompContent::Notification | SyncCompContent::GlobalSolution(_) |
                 SyncCompContent::GlobalFailure => {
                     unreachable!("unexpected message content for leader");
                 },
+            }
         } else {
             // Someone else is the leader
             let message = SyncCompMessage {
                 sync_header: self.create_sync_header(ctx),
                 target_component_id: self.highest_connector_id,
                 content,
             };
             ctx.submit_message(Message::SyncComp(message));
+        }
         return None;
+    }
     fn handle_global_solution_as_leader(&mut self, global_solution: GlobalSolution, ctx: &mut ComponentCtx) -> Option<RoundConclusion> {
         if self.conclusion.is_some() {
             return None;
+        }
         // Handle the global solution
         let mut my_final_branch_id = BranchId::new_invalid();
         for (connector_id, branch_id) in global_solution.component_branches.iter().copied() {
             if connector_id == ctx.id {
                 // This is our solution branch
                 my_final_branch_id = branch_id;
                 continue;
+            }
             let message = SyncCompMessage {
                 sync_header: self.create_sync_header(ctx),
                 target_component_id: connector_id,
                 content: SyncCompContent::GlobalSolution(global_solution.clone()),
             };
             ctx.submit_message(Message::SyncComp(message));
+        }
         debug_assert!(my_final_branch_id.is_valid());
         self.conclusion = Some(RoundConclusion::Success(my_final_branch_id));
         return Some(RoundConclusion::Success(my_final_branch_id));
+    }
     fn handle_global_failure_as_leader(&mut self, ctx: &mut ComponentCtx) -> Option<RoundConclusion> {
         debug_assert!(self.solution_combiner.failure_reported && self.solution_combiner.check_for_global_failure());
         if self.conclusion.is_some() {
             return None;
+        }
         // TODO: Performance
         let mut encountered = VecSet::new();
         for presence in &self.solution_combiner.presence {
             if presence.added_by != ctx.id {
                 // Did not add it ourselves
                 if encountered.push(presence.added_by) {
                     // Not yet sent a message
                     let message = SyncCompMessage{
                         sync_header: self.create_sync_header(ctx),
                         target_component_id: presence.added_by,
                         content: SyncCompContent::GlobalFailure,
                     };
                     ctx.submit_message(Message::SyncComp(message));
+                }
+            }
+        }
         self.conclusion = Some(RoundConclusion::Failure);
         if encountered.is_empty() {
             // We don't have to wait on Acks
             return Some(RoundConclusion::Failure);
         } else {
             return None;
+        }
+    }
     #[inline]
     fn create_sync_header(&self, ctx: &ComponentCtx) -> SyncHeader {
         return SyncHeader{
             sending_component_id: ctx.id,
             highest_component_id: self.highest_connector_id,
             sync_round: self.sync_round,
+        }
+    }

src/runtime2/mod.rs

➞

Show inline comments

@@ @@ -50,429 +50,429 @@ impl ConnectorKey { @@
+}
 /// A kind of token that allows shared access to a connector. Multiple threads
 /// may hold this
 #[derive(Debug, Copy, Clone, PartialEq, Eq, PartialOrd, Ord)]
 pub struct ConnectorId(pub u32);
 impl ConnectorId {
     // TODO: Like the other `new_invalid`, maybe remove
     #[inline]
     pub fn new_invalid() -> ConnectorId {
         return ConnectorId(u32::MAX);
+    }
     #[inline]
     pub(crate) fn is_valid(&self) -> bool {
         return self.0 != u32::MAX;
+    }
+}
 // TODO: Change this, I hate this. But I also don't want to put `public` and
 //  `router` of `ScheduledConnector` back into `Connector`. The reason I don't
 //  want `Box<dyn Connector>` everywhere is because of the v-table overhead. But
 //  to truly design this properly I need some benchmarks.
 pub(crate) enum ConnectorVariant {
     UserDefined(ConnectorPDL),
     Native(Box<dyn Connector>),
+}
 impl Connector for ConnectorVariant {
     fn run(&mut self, scheduler_ctx: SchedulerCtx, comp_ctx: &mut ComponentCtx) -> ConnectorScheduling {
         match self {
             ConnectorVariant::UserDefined(c) => c.run(scheduler_ctx, comp_ctx),
             ConnectorVariant::Native(c) => c.run(scheduler_ctx, comp_ctx),
+        }
+    }
+}
 pub(crate) struct ScheduledConnector {
     pub connector: ConnectorVariant, // access by connector
     pub ctx: ComponentCtx,
     pub public: ConnectorPublic, // accessible by all schedulers and connectors
     pub router: ControlMessageHandler,
     pub shutting_down: bool,
+}
 // -----------------------------------------------------------------------------
 // Runtime
 // -----------------------------------------------------------------------------
 /// Externally facing runtime.
 pub struct Runtime {
     inner: Arc<RuntimeInner>,
+}
 impl Runtime {
     pub fn new(num_threads: u32, protocol_description: ProtocolDescription) -> Runtime {
         // Setup global state
         assert!(num_threads > 0, "need a thread to run connectors");
         let runtime_inner = Arc::new(RuntimeInner{
             protocol_description,
             port_counter: AtomicU32::new(0),
             connectors: RwLock::new(ConnectorStore::with_capacity(32)),
             connector_queue: Mutex::new(VecDeque::with_capacity(32)),
             schedulers: Mutex::new(Vec::new()),
             scheduler_notifier: Condvar::new(),
             active_connectors: AtomicU32::new(0),
             active_interfaces: AtomicU32::new(1), // this `Runtime` instance
             should_exit: AtomicBool::new(false),
         });
         // Launch threads
+        {
             let mut schedulers = Vec::with_capacity(num_threads as usize);
             for thread_index in 0..num_threads {
                 let cloned_runtime_inner = runtime_inner.clone();
                 let thread = thread::Builder::new()
                     .name(format!("thread-{}", thread_index))
                     .spawn(move || {
                         let mut scheduler = Scheduler::new(cloned_runtime_inner, thread_index);
                         scheduler.run();
                     })
                     .unwrap();
                 schedulers.push(thread);
+            }
             let mut lock = runtime_inner.schedulers.lock().unwrap();
             *lock = schedulers;
+        }
         // Return runtime
         return Runtime{ inner: runtime_inner };
+    }
     /// Returns a new interface through which channels and connectors can be
     /// created.
     pub fn create_interface(&self) -> ApplicationInterface {
         self.inner.increment_active_interfaces();
         let (connector, mut interface) = ConnectorApplication::new(self.inner.clone());
         let connector_key = self.inner.create_interface_component(connector);
         interface.set_connector_id(connector_key.downcast());
         // Note that we're not scheduling. That is done by the interface in case
         // it is actually needed.
         return interface;
+    }
+}
 impl Drop for Runtime {
     fn drop(&mut self) {
         self.inner.decrement_active_interfaces();
         let mut lock = self.inner.schedulers.lock().unwrap();
         for handle in lock.drain(..) {
             handle.join().unwrap();
+        }
+    }
+}
 // -----------------------------------------------------------------------------
 // RuntimeInner
 // -----------------------------------------------------------------------------
 pub(crate) struct RuntimeInner {
     // Protocol
     pub(crate) protocol_description: ProtocolDescription,
     // Regular counter for port IDs
     port_counter: AtomicU32,
     // Storage of connectors and the work queue
     connectors: RwLock<ConnectorStore>,
     connector_queue: Mutex<VecDeque<ConnectorKey>>,
     schedulers: Mutex<Vec<JoinHandle<()>>>,
     // Conditions to determine whether the runtime can exit
     scheduler_notifier: Condvar,  // coupled to mutex on `connector_queue`.
     // TODO: Figure out if we can simply merge the counters?
     active_connectors: AtomicU32, // active connectors (if sleeping, then still considered active)
     active_interfaces: AtomicU32, // active API interfaces that can add connectors/channels
     should_exit: AtomicBool,
+}
 impl RuntimeInner {
     // --- Managing the components queued for execution
     /// Wait until there is a connector to run. If there is one, then `Some`
     /// will be returned. If there is no more work, then `None` will be
     /// returned.
     pub(crate) fn wait_for_work(&self) -> Option<ConnectorKey> {
         let mut lock = self.connector_queue.lock().unwrap();
         while lock.is_empty() && !self.should_exit.load(Ordering::Acquire) {
             lock = self.scheduler_notifier.wait(lock).unwrap();
+        }
         return lock.pop_front();
+    }
     pub(crate) fn push_work(&self, key: ConnectorKey) {
         let mut lock = self.connector_queue.lock().unwrap();
         lock.push_back(key);
         self.scheduler_notifier.notify_one();
+    }
     // --- Creating/using ports
     /// Creates a new port pair. Note that these are stored globally like the
     /// connectors are. Ports stored by components belong to those components.
     pub(crate) fn create_channel(&self, creating_connector: ConnectorId) -> (Port, Port) {
         use port::{PortIdLocal, PortKind};
         let getter_id = self.port_counter.fetch_add(2, Ordering::SeqCst);
         let channel_id = ChannelId::new(getter_id);
         let putter_id = PortIdLocal::new(getter_id + 1);
         let getter_id = PortIdLocal::new(getter_id);
         let getter_port = Port{
             self_id: getter_id,
             peer_id: putter_id,
             channel_id,
             kind: PortKind::Getter,
             state: PortState::Open,
             peer_connector: creating_connector,
         };
         let putter_port = Port{
             self_id: putter_id,
             peer_id: getter_id,
             channel_id,
             kind: PortKind::Putter,
             state: PortState::Open,
             peer_connector: creating_connector,
         };
         return (getter_port, putter_port);
+    }
     /// Sends a message to a particular connector. If the connector happened to
     /// be sleeping then it will be scheduled for execution.
     pub(crate) fn send_message(&self, target_id: ConnectorId, message: 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
     /// Creates an initially sleeping application connector.
     fn create_interface_component(&self, component: ConnectorApplication) -> ConnectorKey {
         // Initialize as sleeping, as it will be scheduled by the programmer.
         let mut lock = self.connectors.write().unwrap();
         let key = lock.create(ConnectorVariant::Native(Box::new(component)), true);
         self.increment_active_components();
         return key;
+    }
     /// Creates a new PDL component. This function just creates the component.
     /// If you create it initially awake, then you must add it to the work
     /// queue. Other aspects of correctness (i.e. setting initial ports) are
     /// relinquished to the caller!
     pub(crate) fn create_pdl_component(&self, connector: ConnectorPDL, initially_sleeping: bool) -> ConnectorKey {
         // Create as not sleeping, as we'll schedule it immediately
         let key = {
             let mut lock = self.connectors.write().unwrap();
             lock.create(ConnectorVariant::UserDefined(connector), initially_sleeping)
         };
         self.increment_active_components();
         return key;
+    }
     #[inline]
     pub(crate) fn get_component_private(&self, connector_key: &ConnectorKey) -> &'static mut ScheduledConnector {
         let lock = self.connectors.read().unwrap();
         return lock.get_private(connector_key);
+    }
     #[inline]
     pub(crate) fn get_component_public(&self, connector_id: ConnectorId) -> &'static ConnectorPublic {
         let lock = self.connectors.read().unwrap();
         return lock.get_public(connector_id);
+    }
     pub(crate) fn destroy_component(&self, connector_key: ConnectorKey) {
         let mut lock = self.connectors.write().unwrap();
         lock.destroy(connector_key);
         self.decrement_active_components();
+    }
     // --- Managing exit condition
     #[inline]
     pub(crate) fn increment_active_interfaces(&self) {
         let _old_num = self.active_interfaces.fetch_add(1, Ordering::SeqCst);
         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);
         debug_assert!(old_num > 0);
         if old_num == 1 { // such that active interfaces is now 0
             let num_connectors = self.active_connectors.load(Ordering::Acquire);
             if num_connectors == 0 {
                 self.signal_for_shutdown();
+            }
+        }
+    }
     #[inline]
     fn increment_active_components(&self) {
         let _old_num = self.active_connectors.fetch_add(1, Ordering::SeqCst);
+    }
     fn decrement_active_components(&self) {
         let old_num = self.active_connectors.fetch_sub(1, Ordering::SeqCst);
         debug_assert!(old_num > 0);
         if old_num == 1 { // such that we have no more active connectors (for now!)
             let num_interfaces = self.active_interfaces.load(Ordering::Acquire);
             if num_interfaces == 0 {
                 self.signal_for_shutdown();
+            }
+        }
+    }
     #[inline]
     fn signal_for_shutdown(&self) {
         debug_assert_eq!(self.active_interfaces.load(Ordering::Acquire), 0);
         debug_assert_eq!(self.active_connectors.load(Ordering::Acquire), 0);
         let _lock = self.connector_queue.lock().unwrap();
         let should_signal = self.should_exit
             .compare_exchange(false, true, Ordering::SeqCst, Ordering::Acquire)
             .is_ok();
         if should_signal {
             self.scheduler_notifier.notify_all();
+        }
+    }
+}
 // TODO: Come back to this at some point
 unsafe impl Send for RuntimeInner {}
 unsafe impl Sync for RuntimeInner {}
 // -----------------------------------------------------------------------------
 // ConnectorStore
 // -----------------------------------------------------------------------------
 struct ConnectorStore {
     // Freelist storage of connectors. Storage should be pointer-stable as
     // someone might be mutating the vector while we're executing one of the
     // connectors.
     connectors: RawVec<*mut ScheduledConnector>,
     free: Vec<usize>,
+}
 impl ConnectorStore {
     fn with_capacity(capacity: usize) -> Self {
         Self {
             connectors: RawVec::with_capacity(capacity),
             free: Vec::with_capacity(capacity),
+        }
+    }
     /// Retrieves public part of connector - accessible by many threads at once.
     fn get_public(&self, id: ConnectorId) -> &'static ConnectorPublic {
         unsafe {
             debug_assert!(!self.free.contains(&(id.0 as usize)));
             let connector = self.connectors.get(id.0 as usize);
             debug_assert!(!connector.is_null());
             return &(**connector).public;
+        }
+    }
     /// Retrieves private part of connector - accessible by one thread at a
     /// time.
     fn get_private(&self, key: &ConnectorKey) -> &'static mut ScheduledConnector {
         unsafe {
             debug_assert!(!self.free.contains(&(key.index as usize)));
             let connector = self.connectors.get_mut(key.index as usize);
             debug_assert!(!connector.is_null());
             return &mut (**connector);
+        }
+    }
     /// Creates a new connector. Caller should ensure ports are set up correctly
     /// and the connector is queued for execution if needed.
     fn create(&mut self, connector: ConnectorVariant, initially_sleeping: bool) -> ConnectorKey {
         let mut connector = ScheduledConnector {
             connector,
             ctx: ComponentCtx::new_empty(),
             public: ConnectorPublic::new(initially_sleeping),
             router: ControlMessageHandler::new(),
             shutting_down: false,
         };
         let index;
         let key;
         if self.free.is_empty() {
             // No free entries, allocate new entry
             index = self.connectors.len();
             key = ConnectorKey{ index: index as u32 };
             connector.ctx.id = key.downcast();
             let connector = Box::into_raw(Box::new(connector));
             self.connectors.push(connector);
         } else {
             // Free spot available
             index = self.free.pop().unwrap();
             key = ConnectorKey{ index: index as u32 };
             connector.ctx.id = key.downcast();
             unsafe {
                 let target = self.connectors.get_mut(index);
                 std::ptr::write(*target, connector);
+            }
+        }
         println!("DEBUG [ global store  ] Created component at {}", key.index);
+        // println!("DEBUG [ global store  ] Created component at {}", key.index);
         return key;
+    }
     /// Destroys a connector. Caller should make sure it is not scheduled for
     /// execution. Otherwise one experiences "bad stuff" (tm).
     fn destroy(&mut self, key: ConnectorKey) {
         unsafe {
             let target = self.connectors.get_mut(key.index as usize);
             std::ptr::drop_in_place(*target);
             // Note: but not deallocating!
+        }
         println!("DEBUG [ global store  ] Destroyed component at {}", key.index);
+        // println!("DEBUG [ global store  ] Destroyed component at {}", key.index);
         self.free.push(key.index as usize);
+    }
+}
 impl Drop for ConnectorStore {
     fn drop(&mut self) {
         // Everything in the freelist already had its destructor called, so only
         // has to be deallocated
         for free_idx in self.free.iter().copied() {
             unsafe {
                 let memory = self.connectors.get_mut(free_idx);
                 let layout = std::alloc::Layout::for_value(&**memory);
                 std::alloc::dealloc(*memory as *mut u8, layout);
                 // mark as null for the remainder
                 *memory = std::ptr::null_mut();
+            }
+        }
         // With the deallocated stuff marked as null, clear the remainder that
         // is not null
         for idx in 0..self.connectors.len() {
             unsafe {
                 let memory = *self.connectors.get_mut(idx);
                 if !memory.is_null() {
                     let _ = Box::from_raw(memory); // take care of deallocation, bit dirty, but meh
+                }
+            }
+        }
+    }
+}
@@ \ No newline at end of file @@

src/runtime2/scheduler.rs

➞

Show inline comments

 use std::collections::VecDeque;
 use std::sync::Arc;
 use std::sync::atomic::Ordering;
 use crate::protocol::eval::EvalError;
 use crate::runtime2::port::ChannelId;
 use super::{ScheduledConnector, RuntimeInner, ConnectorId, ConnectorKey};
 use super::port::{Port, PortState, PortIdLocal};
 use super::native::Connector;
 use super::branch::{BranchId};
 use super::connector::{ConnectorPDL, ConnectorScheduling};
 use super::inbox::{
     Message, DataMessage, SyncHeader,
     ControlMessage, ControlContent,
     SyncControlMessage, SyncControlContent,
 };
 // Because it contains pointers we're going to do a copy by value on this one
 #[derive(Clone, Copy)]
 pub(crate) struct SchedulerCtx<'a> {
     pub(crate) runtime: &'a RuntimeInner
+}
 pub(crate) struct Scheduler {
     runtime: Arc<RuntimeInner>,
     scheduler_id: u32,
+}
 impl Scheduler {
     pub fn new(runtime: Arc<RuntimeInner>, scheduler_id: u32) -> Self {
         return Self{ runtime, scheduler_id };
+    }
     pub fn run(&mut self) {
         // Setup global storage and workspaces that are reused for every
         // connector that we run
         'thread_loop: loop {
             // Retrieve a unit of work
             self.debug("Waiting for work");
             let connector_key = self.runtime.wait_for_work();
             if connector_key.is_none() {
                 // We should exit
                 self.debug(" ... No more work, quitting");
                 break 'thread_loop;
+            }
             // We have something to do
             let connector_key = connector_key.unwrap();
             let connector_id = connector_key.downcast();
             self.debug_conn(connector_id, &format!(" ... Got work, running {}", connector_key.index));
             let scheduled = self.runtime.get_component_private(&connector_key);
             // Keep running until we should no longer immediately schedule the
             // connector.
             let mut cur_schedule = ConnectorScheduling::Immediate;
             while let ConnectorScheduling::Immediate = cur_schedule {
                 self.handle_inbox_messages(scheduled);
                 // Run the main behaviour of the connector, depending on its
                 // current state.
                 if scheduled.shutting_down {
                     // Nothing to do. But we're stil waiting for all our pending
                     // control messages to be answered.
                     self.debug_conn(connector_id, &format!("Shutting down, {} Acks remaining", scheduled.router.num_pending_acks()));
                     self.handle_inbox_while_shutting_down(scheduled);
                     if scheduled.router.num_pending_acks() == 0 {
                         // We're actually done, we can safely destroy the
                         // currently running connector
                         self.runtime.destroy_component(connector_key);
                         continue 'thread_loop;
                     } else {
                         cur_schedule = ConnectorScheduling::NotNow;
+                    }
                 } else {
                     self.debug_conn(connector_id, "Running ...");
                     let scheduler_ctx = SchedulerCtx{ runtime: &*self.runtime };
                     let new_schedule = scheduled.connector.run(scheduler_ctx, &mut scheduled.ctx);
                     self.debug_conn(connector_id, &format!("Finished running (new scheduling is {:?})", new_schedule));
                     // Handle all of the output from the current run: messages to
                     // send and connectors to instantiate.
                     self.handle_changes_in_context(scheduled);
                     cur_schedule = new_schedule;
+                }
+            }
             // If here then the connector does not require immediate execution.
             // So enqueue it if requested, and otherwise put it in a sleeping
             // state.
             match cur_schedule {
                 ConnectorScheduling::Immediate => unreachable!(),
                 ConnectorScheduling::Later => {
                     // Simply queue it again later
                     self.runtime.push_work(connector_key);
                 },
                 ConnectorScheduling::NotNow => {
                     // Need to sleep, note that we are the only ones which are
                     // allows to set the sleeping state to `true`, and since
                     // we're running it must currently be `false`.
                     self.try_go_to_sleep(connector_key, scheduled);
                 },
                 ConnectorScheduling::Exit => {
                     // Prepare for exit. Set the shutdown flag and broadcast
                     // messages to notify peers of closing channels
                     scheduled.shutting_down = true;
                     for port in &scheduled.ctx.ports {
                         if port.state != PortState::Closed {
                             let message = scheduled.router.prepare_closing_channel(
                                 port.self_id, port.peer_id,
                                 connector_id
                             );
                             self.debug_conn(connector_id, &format!("Sending message [ exit ] \n --- {:?}", message));
                             self.runtime.send_message(port.peer_connector, Message::Control(message));
+                        }
+                    }
                     if scheduled.router.num_pending_acks() == 0 {
                         // All ports (if any) already closed
                         self.runtime.destroy_component(connector_key);
                         continue 'thread_loop;
+                    }
                     self.try_go_to_sleep(connector_key, scheduled);
                 },
+            }
+        }
+    }
     /// Receiving messages from the public inbox and handling them or storing
     /// them in the component's private inbox
     fn handle_inbox_messages(&mut self, scheduled: &mut ScheduledConnector) {
         let connector_id = scheduled.ctx.id;
         while let Some(message) = scheduled.public.inbox.take_message() {
             // Check if the message has to be rerouted because we have moved the
             // target port to another component.
             self.debug_conn(connector_id, &format!("Handling message\n --- {:#?}", message));
             if let Some(target_port) = Self::get_message_target_port(&message) {
                 if let Some(other_component_id) = scheduled.router.should_reroute(target_port) {
                     self.debug_conn(connector_id, " ... Rerouting the message");
                     self.runtime.send_message(other_component_id, message);
                     continue;
+                }
+            }
             // If here, then we should handle the message
             self.debug_conn(connector_id, " ... Handling the message");
             match message {
                 Message::Control(message) => {
                     match message.content {
                         ControlContent::PortPeerChanged(port_id, new_target_connector_id) => {
                             // Need to change port target
                             let port = scheduled.ctx.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.outbox.is_empty());
                             // And respond with an Ack
                             let ack_message = Message::Control(ControlMessage {
                                 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.get_port_mut_by_id(port_id).unwrap();
                             port.state = PortState::Closed;
                             // Send an Ack
                             let ack_message = Message::Control(ControlMessage {
                                 id: message.id,
                                 sending_component_id: connector_id,
                                 content: ControlContent::Ack,
                             });
                             self.debug_conn(connector_id, &format!("Sending message [cc ack] \n --- {:?}", ack_message));
                             self.runtime.send_message(message.sending_component_id, ack_message);
                         },
                         ControlContent::Ack => {
                             scheduled.router.handle_ack(message.id);
                         },
                         ControlContent::Ping => {},
+                    }
                 },
                 _ => {
                     // All other cases have to be handled by the component
                     scheduled.ctx.inbox_messages.push(message);
+                }
+            }
+        }
+    }
     /// Handles inbox messages while shutting down. This intends to handle the
     /// case where a component cleanly exited outside of a sync region, but a
     /// peer, before receiving the `CloseChannel` message, sent a message inside
     /// a sync region. This peer should be notified that its message is not
     /// received by a component in a sync region.
     fn handle_inbox_while_shutting_down(&mut self, scheduled: &mut ScheduledConnector) {
         // Note: we're not handling the public inbox, we're dealing with the
         // private one!
         debug_assert!(scheduled.shutting_down);
-        while let Some(message) = scheduled.ctx.read_next_message() {
+        while let Some(message) = scheduled.ctx.read_next_message_even_if_not_in_sync() {
             let target_port_and_round_number = match &message {
                 Message::Data(msg) => Some((msg.data_header.target_port, msg.sync_header.sync_round)),
                 Message::SyncComp(_) => None,
                 Message::SyncPort(msg) => Some((msg.target_port, msg.sync_header.sync_round)),
                 Message::SyncControl(_) => None,
                 Message::Control(_) => None,
             };
             if let Some((target_port, sync_round)) = target_port_and_round_number {
                 // This message is aimed at a port, but we're shutting down, so
                 // notify the peer that its was not received properly.
                 // (also: since we're shutting down, we're not in sync mode and
                 // the context contains the definitive set of owned ports)
                 let port = scheduled.ctx.get_port_by_id(target_port).unwrap();
                 let message = SyncControlMessage{
                     in_response_to_sync_round: sync_round,
                     target_component_id: port.peer_connector,
                     content: SyncControlContent::ChannelIsClosed(port.peer_id),
                 };
                 self.debug_conn(scheduled.ctx.id, &format!("Sending message [shutdown]\n --- {:?}", message));
                 self.runtime.send_message(port.peer_connector, Message::SyncControl(message));
+            }
+        }
+    }
     /// Handles changes to the context that were made by the component. This is
     /// the way (due to Rust's borrowing rules) that we bubble up changes in the
     /// component's state that the scheduler needs to know about (e.g. a message
     /// that the component wants to send, a port that has been added).
     fn handle_changes_in_context(&mut self, scheduled: &mut ScheduledConnector) {
         let connector_id = scheduled.ctx.id;
         // Handling any messages that were sent
         while let Some(message) = scheduled.ctx.outbox.pop_front() {
             self.debug_conn(connector_id, &format!("Sending message [outbox] \n --- {:#?}", message));
             let target_component_id = match &message {
                 Message::Data(content) => {
                     // Data messages are always sent to a particular port, and
                     // may end up being rerouted.
                     let port_desc = scheduled.ctx.get_port_by_id(content.data_header.sending_port).unwrap();
                     debug_assert_eq!(port_desc.peer_id, content.data_header.target_port);
                     if port_desc.state == PortState::Closed {
                         todo!("handle sending over a closed port")
+                    }
                     port_desc.peer_connector
                 },
                 Message::SyncComp(content) => {
                     // Sync messages are always sent to a particular component,
                     // the sender must make sure it actually wants to send to
                     // the specified component (and is not using an inconsistent
                     // component ID associated with a port).
                     content.target_component_id
                 },
                 Message::SyncPort(content) => {
                     let port_desc = scheduled.ctx.get_port_by_id(content.source_port).unwrap();
                     debug_assert_eq!(port_desc.peer_id, content.target_port);
                     if port_desc.state == PortState::Closed {
                         todo!("handle sending over a closed port")
+                    }
                     port_desc.peer_connector
                 },
                 Message::SyncControl(_) => unreachable!("component sending 'SyncControl' messages directly"),
                 Message::Control(_) => unreachable!("component sending 'Control' messages directly"),
             };
             self.runtime.send_message(target_component_id, message);
+        }
         while let Some(state_change) = scheduled.ctx.state_changes.pop_front() {
             match state_change {
                 ComponentStateChange::CreatedComponent(component, initial_ports) => {
                     // Creating a new component. The creator needs to relinquish
                     // ownership of the ports that are given to the new
                     // component. All data messages that were intended for that
                     // port also needs to be transferred.
                     let new_key = self.runtime.create_pdl_component(component, false);
                     let new_connector = self.runtime.get_component_private(&new_key);
                     for port_id in initial_ports {
                         // Transfer messages associated with the transferred port
                         let mut message_idx = 0;
                         while message_idx < scheduled.ctx.inbox_messages.len() {
                             let message = &scheduled.ctx.inbox_messages[message_idx];
                             if Self::get_message_target_port(message) == Some(port_id) {
                                 // Need to transfer this message
                                 // TODO: Revise messages, this is becoming messy and error-prone
                                 let message = scheduled.ctx.inbox_messages.remove(message_idx);
                                 if message_idx < scheduled.ctx.inbox_len_read {
                                     scheduled.ctx.inbox_len_read -= 1;
+                                }
                                 new_connector.ctx.inbox_messages.push(message);
                             } else {
                                 message_idx += 1;
+                            }
+                        }
                         // Transfer the port itself
                         let port_index = scheduled.ctx.ports.iter()
                             .position(|v| v.self_id == port_id)
                             .unwrap();
                         let port = scheduled.ctx.ports.remove(port_index);
                         new_connector.ctx.ports.push(port.clone());
                         // Notify the peer that the port has changed
                         let reroute_message = scheduled.router.prepare_reroute(
                             port.self_id, port.peer_id, scheduled.ctx.id,
                             port.peer_connector, new_connector.ctx.id
                         );
                         self.debug_conn(connector_id, &format!("Sending message [newcon]\n --- {:?}", reroute_message));
                         self.runtime.send_message(port.peer_connector, Message::Control(reroute_message));
+                    }
                     // Schedule new connector to run
                     self.runtime.push_work(new_key);
                 },
                 ComponentStateChange::CreatedPort(port) => {
                     scheduled.ctx.ports.push(port);
                 },
                 ComponentStateChange::ChangedPort(port_change) => {
                     if port_change.is_acquired {
                         scheduled.ctx.ports.push(port_change.port);
                     } else {
                         let index = scheduled.ctx.ports
                             .iter()
                             .position(|v| v.self_id == port_change.port.self_id)
                             .unwrap();
                         scheduled.ctx.ports.remove(index);
+                    }
+                }
+            }
+        }
         // Finally, check if we just entered or just left a sync region
         if scheduled.ctx.changed_in_sync {
             if scheduled.ctx.is_in_sync {
                 // Just entered sync region
             } else {
                 // Just left sync region. So clear inbox up until the last
                 // message that was read.
                 scheduled.ctx.inbox_messages.drain(0..scheduled.ctx.inbox_len_read);
                 scheduled.ctx.inbox_len_read = 0;
+            }
             scheduled.ctx.changed_in_sync = false; // reset flag
+        }
+    }
     fn try_go_to_sleep(&self, connector_key: ConnectorKey, connector: &mut ScheduledConnector) {
         debug_assert_eq!(connector_key.index, connector.ctx.id.0);
         debug_assert_eq!(connector.public.sleeping.load(Ordering::Acquire), false);
         // This is the running connector, and only the running connector may
         // decide it wants to sleep again.
         connector.public.sleeping.store(true, Ordering::Release);
         // But due to reordering we might have received messages from peers who
         // did not consider us sleeping. If so, then we wake ourselves again.
         if !connector.public.inbox.is_empty() {
             // Try to wake ourselves up (needed because someone might be trying
             // the exact same atomic compare-and-swap at this point in time)
             let should_wake_up_again = connector.public.sleeping
                 .compare_exchange(true, false, Ordering::SeqCst, Ordering::Acquire)
                 .is_ok();
             if should_wake_up_again {
                 self.runtime.push_work(connector_key)
+            }
+        }
+    }
     #[inline]
     fn get_message_target_port(message: &Message) -> Option<PortIdLocal> {
         match message {
             Message::Data(data) => return Some(data.data_header.target_port),
             Message::SyncComp(_) => {},
             Message::SyncPort(content) => return Some(content.target_port),
             Message::SyncControl(_) => return None,
             Message::Control(control) => {
                 match &control.content {
                     ControlContent::PortPeerChanged(port_id, _) => return Some(*port_id),
                     ControlContent::CloseChannel(port_id) => return Some(*port_id),
                     ControlContent::Ping | ControlContent::Ack => {},
+                }
             },
+        }
         return None
+    }
     // TODO: Remove, this is debugging stuff
     fn debug(&self, message: &str) {
         println!("DEBUG [thrd:{:02} conn:  ]: {}", self.scheduler_id, message);
+        // println!("DEBUG [thrd:{:02} conn:  ]: {}", self.scheduler_id, message);
+    }
     fn debug_conn(&self, conn: ConnectorId, message: &str) {
         println!("DEBUG [thrd:{:02} conn:{:02}]: {}", self.scheduler_id, conn.0, message);
+        // println!("DEBUG [thrd:{:02} conn:{:02}]: {}", self.scheduler_id, conn.0, message);
+    }
+}
 // -----------------------------------------------------------------------------
 // ComponentCtx
 // -----------------------------------------------------------------------------
 enum ComponentStateChange {
     CreatedComponent(ConnectorPDL, Vec<PortIdLocal>),
     CreatedPort(Port),
     ChangedPort(ComponentPortChange),
+}
 #[derive(Clone)]
 pub(crate) struct ComponentPortChange {
     pub is_acquired: bool, // otherwise: released
     pub port: Port,
+}
 /// The component context (better name may be invented). This was created
 /// because part of the component's state is managed by the scheduler, and part
 /// of it by the component itself. When the component starts a sync block or
 /// exits a sync block the partially managed state by both component and
 /// scheduler need to be exchanged.
 pub(crate) struct ComponentCtx {
     // Mostly managed by the scheduler
     pub(crate) id: ConnectorId,
     ports: Vec<Port>,
     inbox_messages: Vec<Message>,
     inbox_len_read: usize,
     // Submitted by the component
     is_in_sync: bool,
     changed_in_sync: bool,
     outbox: VecDeque<Message>,
     state_changes: VecDeque<ComponentStateChange>,
     // Workspaces that may be used by components to (generally) prevent
     // allocations. Be a good scout and leave it empty after you've used it.
     // TODO: Move to scheduler ctx, this is the wrong place
     pub workspace_ports: Vec<PortIdLocal>,
     pub workspace_branches: Vec<BranchId>,
+}
 impl ComponentCtx {
     pub(crate) fn new_empty() -> Self {
         return Self{
             id: ConnectorId::new_invalid(),
             ports: Vec::new(),
             inbox_messages: Vec::new(),
             inbox_len_read: 0,
             is_in_sync: false,
             changed_in_sync: false,
             outbox: VecDeque::new(),
             state_changes: VecDeque::new(),
             workspace_ports: Vec::new(),
             workspace_branches: Vec::new(),
         };
+    }
     /// Notify the runtime that the component has created a new component. May
     /// only be called outside of a sync block.
     pub(crate) fn push_component(&mut self, component: ConnectorPDL, initial_ports: Vec<PortIdLocal>) {
         debug_assert!(!self.is_in_sync);
         self.state_changes.push_back(ComponentStateChange::CreatedComponent(component, initial_ports));
+    }
     /// Notify the runtime that the component has created a new port. May only
     /// be called outside of a sync block (for ports received during a sync
     /// block, pass them when calling `notify_sync_end`).
     pub(crate) fn push_port(&mut self, port: Port) {
         debug_assert!(!self.is_in_sync);
         self.state_changes.push_back(ComponentStateChange::CreatedPort(port))
+    }
     /// Notify the runtime of an error. Note that this will not perform any
     /// special action beyond printing the error. The component is responsible
     /// for waiting until it is appropriate to shut down (i.e. being outside
     /// of a sync region) and returning the `Exit` scheduling code.
     pub(crate) fn push_error(&mut self, error: EvalError) {
         println!("ERROR: Component ({}) encountered a critical error:\n{}", self.id.0, error);
+    }
     #[inline]
     pub(crate) fn get_ports(&self) -> &[Port] {
         return self.ports.as_slice();
+    }
     pub(crate) fn get_port_by_id(&self, id: PortIdLocal) -> Option<&Port> {
         return self.ports.iter().find(|v| v.self_id == id);
+    }
     pub(crate) fn get_port_by_channel_id(&self, id: ChannelId) -> Option<&Port> {
         return self.ports.iter().find(|v| v.channel_id == id);
+    }
     fn get_port_mut_by_id(&mut self, id: PortIdLocal) -> Option<&mut Port> {
         return self.ports.iter_mut().find(|v| v.self_id == id);
+    }
     /// Notify that component will enter a sync block. Note that after calling
     /// this function you must allow the scheduler to pick up the changes in the
     /// context by exiting your code-executing loop, and to continue executing
     /// code the next time the scheduler picks up the component.
     pub(crate) fn notify_sync_start(&mut self) {
         debug_assert!(!self.is_in_sync);
         self.is_in_sync = true;
         self.changed_in_sync = true;
+    }
     #[inline]
     pub(crate) fn is_in_sync(&self) -> bool {
         return self.is_in_sync;
+    }
     /// Submit a message for the scheduler to send to the appropriate receiver.
     /// May only be called inside of a sync block.
     pub(crate) fn submit_message(&mut self, contents: Message) -> Result<(), ()> {
         debug_assert!(self.is_in_sync);
         if let Some(port_id) = contents.source_port() {
             let port_info = self.get_port_by_id(port_id);
             let is_valid = match port_info {
                 Some(port_info) => {
                     port_info.state == PortState::Open
                 },
                 None => false,
             };
             if !is_valid {
                 // We don't own the port
                 println!(" ****** DEBUG ****** : Sending through closed port!!! {}", port_id.index);
                 return Err(());
+            }
+        }
         self.outbox.push_back(contents);
         return Ok(());
+    }
     /// Notify that component just finished a sync block. Like
     /// `notify_sync_start`: drop out of the `Component::Run` function.
     pub(crate) fn notify_sync_end(&mut self, changed_ports: &[ComponentPortChange]) {
         debug_assert!(self.is_in_sync);
         self.is_in_sync = false;
         self.changed_in_sync = true;
         self.state_changes.reserve(changed_ports.len());
         for changed_port in changed_ports {
             self.state_changes.push_back(ComponentStateChange::ChangedPort(changed_port.clone()));
+        }
+    }
     /// Retrieves messages matching a particular port and branch id. But only
     /// those messages that have been previously received with
     /// `read_next_message`.
     pub(crate) fn get_read_data_messages(&self, match_port_id: PortIdLocal) -> MessagesIter {
         return MessagesIter {
             messages: &self.inbox_messages,
             next_index: 0,
             max_index: self.inbox_len_read,
             match_port_id
         };
+    }
     /// Retrieves the next unread message from the inbox `None` if there are no
     /// (new) messages to read.
     // TODO: Fix the clone of the data message, entirely unnecessary
     pub(crate) fn read_next_message(&mut self) -> Option<Message> {
         if !self.is_in_sync { return None; }
         return self.read_next_message_even_if_not_in_sync();
+    }
     pub(crate) fn read_next_message_even_if_not_in_sync(&mut self) -> Option<Message> {
         if self.inbox_len_read == self.inbox_messages.len() { return None; }
         // We want to keep data messages in the inbox, because we need to check
         // them in the future. We don't want to keep sync messages around, we
         // should only handle them once. Control messages should never be in
         // here.
         let message = &self.inbox_messages[self.inbox_len_read];
         match &message {
             Message::Data(content) => {
                 // Keep message in inbox for later reading
                 self.inbox_len_read += 1;
                 return Some(Message::Data(content.clone()));
             },
             Message::SyncComp(_) | Message::SyncPort(_) | Message::SyncControl(_) => {
                 // Remove message from inbox
                 let message = self.inbox_messages.remove(self.inbox_len_read);
                 return Some(message);
             },
             Message::Control(_) => unreachable!("control message ended up in component inbox"),
+        }
+    }
+}
 pub(crate) struct MessagesIter<'a> {
     messages: &'a [Message],
     next_index: usize,
     max_index: usize,
     match_port_id: PortIdLocal,
+}
 impl<'a> Iterator for MessagesIter<'a> {
     type Item = &'a DataMessage;
     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 Message::Data(message) = &message {
                 if message.data_header.target_port == self.match_port_id {
                     // Found a match
                     self.next_index += 1;
                     return Some(message);
+                }
             } else {
                 // Unreachable because:
                 //  1. We only iterate over messages that were previously retrieved by `read_next_message`.
                 //  2. Inbox does not contain control/ping messages.
                 //  3. If `read_next_message` encounters anything else than a data message, it is removed from the inbox.
                 unreachable!();
+            }
             self.next_index += 1;
+        }
         // No more messages
         return None;
+    }
+}
 // -----------------------------------------------------------------------------
 // Control messages
 // -----------------------------------------------------------------------------
 struct ControlEntry {
     id: u32,
     variant: ControlVariant,
+}
 enum ControlVariant {
     ChangedPort(ControlChangedPort),
     ClosedChannel(ControlClosedChannel),
+}
 struct ControlChangedPort {
     target_port: PortIdLocal,       // if send to this port, then reroute
     source_connector: ConnectorId,  // connector we expect messages from
     target_connector: ConnectorId,  // connector we need to reroute to
+}
 struct ControlClosedChannel {
     source_port: PortIdLocal,
     target_port: PortIdLocal,
+}
 pub(crate) struct ControlMessageHandler {
     id_counter: u32,
     active: Vec<ControlEntry>,
+}
 impl ControlMessageHandler {
     pub fn new() -> Self {
         ControlMessageHandler {
             id_counter: 0,
             active: Vec::new(),
+        }
+    }
     /// Prepares a message indicating that a channel has closed, we keep a local
     /// entry to match against the (hopefully) returned `Ack` message.
     pub fn prepare_closing_channel(
         &mut self, self_port_id: PortIdLocal, peer_port_id: PortIdLocal,
         self_connector_id: ConnectorId
     ) -> ControlMessage {
         let id = self.take_id();
         self.active.push(ControlEntry{
             id,
             variant: ControlVariant::ClosedChannel(ControlClosedChannel{
                 source_port: self_port_id,
                 target_port: peer_port_id,
             }),
         });
         return ControlMessage {
             id,
             sending_component_id: self_connector_id,
             content: ControlContent::CloseChannel(peer_port_id),
         };
+    }
     /// Prepares rerouting messages due to changed ownership of a port. The
     /// control message returned by this function must be sent to the
     /// transferred port's peer connector.
     pub fn prepare_reroute(
         &mut self,
         port_id: PortIdLocal, peer_port_id: PortIdLocal,
         self_connector_id: ConnectorId, peer_connector_id: ConnectorId,
         new_owner_connector_id: ConnectorId
     ) -> ControlMessage {
         let id = self.take_id();
         self.active.push(ControlEntry{
             id,
             variant: ControlVariant::ChangedPort(ControlChangedPort{
                 target_port: port_id,
                 source_connector: peer_connector_id,
                 target_connector: new_owner_connector_id,
             }),
         });
         return ControlMessage {
             id,
             sending_component_id: self_connector_id,
             content: ControlContent::PortPeerChanged(peer_port_id, new_owner_connector_id),
         };
+    }
     /// Returns true if the supplied message should be rerouted. If so then this
     /// function returns the connector that should retrieve this message.
     pub fn should_reroute(&self, target_port: PortIdLocal) -> Option<ConnectorId> {
         for entry in &self.active {
             if let ControlVariant::ChangedPort(entry) = &entry.variant {
                 if entry.target_port == target_port {
                     // Need to reroute this message
                     return Some(entry.target_connector);
+                }
+            }
+        }
         return None;
+    }
     /// Handles an Ack as an answer to a previously sent control message
     pub fn handle_ack(&mut self, id: u32) {
         let index = self.active.iter()
             .position(|v| v.id == id);
         match index {
             Some(index) => { self.active.remove(index); },
             None => { todo!("handling of nefarious ACKs"); },
+        }
+    }
     /// Retrieves the number of responses we still expect to receive from our
     /// peers
     #[inline]
     pub fn num_pending_acks(&self) -> usize {
         return self.active.len();
+    }
     fn take_id(&mut self) -> u32 {
         let generated_id = self.id_counter;
         let (new_id, _) = self.id_counter.overflowing_add(1);
         self.id_counter = new_id;
         return generated_id;
+    }
+}
@@ \ No newline at end of file @@

src/runtime2/tests/mod.rs

➞

Show inline comments

 mod network_shapes;
 mod api_component;
 mod speculation;
 mod data_transmission;
 mod sync_failure;
 use super::*;
 use crate::{PortId, ProtocolDescription};
 use crate::common::Id;
 use crate::protocol::eval::*;
 use crate::runtime2::native::{ApplicationSyncAction};
 // Generic testing constants, use when appropriate to simplify stress-testing
 // pub(crate) const NUM_THREADS: u32 = 3;     // number of threads in runtime
 // pub(crate) const NUM_INSTANCES: u32 = 7;   // number of test instances constructed
 // pub(crate) const NUM_LOOPS: u32 = 8;       // number of loops within a single test (not used by all tests)
 pub(crate) const NUM_THREADS: u32 = 1;
 pub(crate) const NUM_INSTANCES: u32 = 1;
 pub(crate) const NUM_LOOPS: u32 = 1;
 pub(crate) const NUM_INSTANCES: u32 = 5;
 pub(crate) const NUM_LOOPS: u32 = 5;
 fn create_runtime(pdl: &str) -> Runtime {
     let protocol = ProtocolDescription::parse(pdl.as_bytes()).expect("parse pdl");
     let runtime = Runtime::new(NUM_THREADS, protocol);
     return runtime;
+}
 fn run_test_in_runtime<F: Fn(&mut ApplicationInterface)>(pdl: &str, constructor: F) {
     let protocol = ProtocolDescription::parse(pdl.as_bytes())
         .expect("parse PDL");
     let runtime = Runtime::new(NUM_THREADS, protocol);
     let mut api = runtime.create_interface();
     for _ in 0..NUM_INSTANCES {
         constructor(&mut api);
+    }
+}
 pub(crate) struct TestTimer {
     name: &'static str,
     started: std::time::Instant
+}
 impl TestTimer {
     pub(crate) fn new(name: &'static str) -> Self {
         Self{ name, started: std::time::Instant::now() }
+    }
+}
 impl Drop for TestTimer {
     fn drop(&mut self) {
         let delta = std::time::Instant::now() - self.started;
         let nanos = (delta.as_secs_f64() * 1_000_000.0) as u64;
         let millis = nanos / 1000;
         let nanos = nanos % 1000;
         println!("[{}] Took {:>4}.{:03} ms", self.name, millis, nanos);
+    }
+}

src/runtime2/tests/sync_failure.rs

➞

Show inline comments

 // sync_failure.rs
 //
 // Various tests to ensure that failing components fail in a consistent way.
 use super::*;
 #[test]
 fn test_local_sync_failure() {
     // If the component exits cleanly, then the runtime exits cleanly, and the
     // test will finish
     const CODE: &'static str = "
     primitive immediate_failure_inside_sync() {
         u32[] only_allows_index_0 = { 1 };
         while (true) sync { // note the infinite loop
             auto value = only_allows_index_0[1];
+        }
+    }
     primitive immediate_failure_outside_sync() {
         u32[] only_allows_index_0 = { 1 };
         auto never_gonna_get = only_allows_index_0[1];
         while (true) sync {}
+    }
     ";
     // let thing = TestTimer::new("local_sync_failure");
     run_test_in_runtime(CODE, |api| {
         api.create_connector("", "immediate_failure_outside_sync", ValueGroup::new_stack(Vec::new()))
             .expect("create component");
         api.create_connector("", "immediate_failure_inside_sync", ValueGroup::new_stack(Vec::new()))
             .expect("create component");
     })
+}
 #[test]
 fn test_shared_sync_failure() {
     // Same as above. One of the components should fail, the other should follow
     // suit because it cannot complete a sync round.
     // suit because it cannot complete a sync round. We intentionally have an
     // infinite loop in the while condition because we need at least two loops
     // for the last error to get picked up.
     const CODE: &'static str = "
     enum Location { BeforeSync, AfterPut, AfterGet, AfterSync, Never }
     primitive failing_at_location(in<bool> input, out<bool> output, Location loc) {
         u32[] failure_array = {};
         while (true) {
             if (loc == Location::BeforeSync) failure_array[0];
             sync {
                 put(output, true);
                 if (loc == Location::AfterPut) failure_array[0];
                 auto received = get(input);
                 assert(received);
                 if (loc == Location::AfterGet) failure_array[0];
+            }
             if (loc == Location::AfterSync) failure_array[0];
+        }
+    }
     composite constructor(Location loc) {
         channel output_a -> input_a;
         channel output_b -> input_b;
         new failing_at_location(input_a, output_b, Location::Never);
         new failing_at_location(input_b, output_a, loc);
+    }
     ";
     run_test_in_runtime(CODE, |api| {
-        for variant in 0..1 {
+        for variant in 0..4 { // all `Location` enum variants, except `Never`.
             // Create the channels
             api.create_connector("", "constructor", ValueGroup::new_stack(vec![
                 Value::Enum(variant)
             ])).expect("create connector");
+        }
     })
+}
@@ \ No newline at end of file @@

0 comments (0 inline, 0 general)