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

Changeset - 60057e2acf9e

Parent rev.

Child rev.

[Not reviewed]

0 4 0

MH - 3 years ago 2022-04-19 18:16:33
contact@maxhenger.nl

WIP on error-handling implementation

4 files changed with 70 insertions and 11 deletions:

src/runtime2/communication.rs

src/runtime2/component/component.rs

src/runtime2/component/component_context.rs

src/runtime2/component/consensus.rs

0 comments (0 inline, 0 general)

src/runtime2/communication.rs

➞

Show inline comments

 use crate::protocol::eval::*;
 use super::runtime::*;
 use super::component::*;
 // -----------------------------------------------------------------------------
 // Generic types
 // -----------------------------------------------------------------------------
 #[derive(Debug, Copy, Clone, PartialEq, Eq)]
 pub struct PortId(pub u32);
 impl PortId {
     /// This value is not significant, it is chosen to make debugging easier: a
     /// very large port number is more likely to shine a light on bugs.
     pub fn new_invalid() -> Self {
         return Self(u32::MAX);
+    }
+}
 #[derive(Debug, PartialEq, Eq, Clone, Copy)]
 pub enum PortKind {
     Putter,
     Getter,
+}
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum PortState {
     Open,
     BlockedDueToPeerChange,
     BlockedDueToFullBuffers,
     Closed,
+}
 impl PortState {
     pub fn is_blocked(&self) -> bool {
         match self {
             PortState::BlockedDueToPeerChange | PortState::BlockedDueToFullBuffers => true,
             PortState::Open | PortState::Closed => false,
+        }
+    }
+}
 pub struct Channel {
     pub putter_id: PortId,
     pub getter_id: PortId,
+}
 // -----------------------------------------------------------------------------
 // Data messages
 // -----------------------------------------------------------------------------
 #[derive(Debug)]
 pub struct DataMessage {
     pub data_header: MessageDataHeader,
     pub sync_header: MessageSyncHeader,
     pub content: ValueGroup,
+}
 #[derive(Debug)]
 pub enum PortAnnotationKind {
     Getter(PortAnnotationGetter),
     Putter(PortAnnotationPutter),
+}
 #[derive(Debug)]
 pub struct PortAnnotationGetter {
     pub self_comp_id: CompId,
     pub self_port_id: PortId,
     pub peer_comp_id: CompId,
     pub peer_port_id: PortId,
+}
 #[derive(Debug)]
 pub struct PortAnnotationPutter {
     pub self_comp_id: CompId,
     pub self_port_id: PortId,
+}
 #[derive(Debug)]
 pub struct MessageDataHeader {
     pub expected_mapping: Vec<(PortAnnotationKind, Option<u32>)>,
     pub new_mapping: u32,
     pub source_port: PortId,
     pub target_port: PortId,
+}
 // -----------------------------------------------------------------------------
 // Sync messages
 // -----------------------------------------------------------------------------
 #[derive(Debug)]
 pub struct SyncMessage {
     pub sync_header: MessageSyncHeader,
     pub content: SyncMessageContent,
+}
 #[derive(Debug)]
 pub enum SyncLocalSolutionEntry {
     Putter(SyncSolutionPutterPort),
     Getter(SyncSolutionGetterPort),
+}
 pub type SyncLocalSolution = Vec<SyncLocalSolutionEntry>;
 /// Getter port in a solution. Upon receiving a message it is certain about who
 /// its peer is.
 #[derive(Debug)]
 pub struct SyncSolutionGetterPort {
     pub self_comp_id: CompId,
     pub self_port_id: PortId,
     pub peer_comp_id: CompId,
     pub peer_port_id: PortId,
     pub mapping: u32,
+}
 /// Putter port in a solution. A putter may not be certain about who its peer
 /// component/port is.
 #[derive(Debug)]
 pub struct SyncSolutionPutterPort {
     pub self_comp_id: CompId,
     pub self_port_id: PortId,
     pub mapping: u32,
+}
 #[derive(Debug)]
 pub struct SyncSolutionChannel {
     pub putter: Option<SyncSolutionPutterPort>,
     pub getter: Option<SyncSolutionGetterPort>,
+}
 #[derive(Debug, Copy, Clone, PartialEq, Eq)]
 pub enum SyncRoundDecision {
     None,
     Solution,
     Failure,
+}
 #[derive(Debug)]
 pub struct SyncPartialSolution {
     pub channel_mapping: Vec<SyncSolutionChannel>,
     pub decision: SyncRoundDecision,
+}
 impl Default for SyncPartialSolution {
     fn default() -> Self {
         return Self{
             channel_mapping: Vec::new(),
             decision: SyncRoundDecision::None,
+        }
+    }
+}
 #[derive(Debug)]
 pub enum SyncMessageContent {
     NotificationOfLeader,
     LocalSolution(CompId, SyncLocalSolution), // local solution of the specified component
     PartialSolution(SyncPartialSolution), // partial solution of multiple components
     GlobalSolution,
     GlobalFailure,
+}
 // -----------------------------------------------------------------------------
 // Control messages
 // -----------------------------------------------------------------------------
 #[derive(Debug)]
 pub struct ControlMessage {
     pub(crate) id: ControlId,
     pub sender_comp_id: CompId,
     pub target_port_id: Option<PortId>,
     pub content: ControlMessageContent,
+}
 #[derive(Copy, Clone, Debug)]
 pub enum ControlMessageContent {
     Ack,
     BlockPort(PortId),
     UnblockPort(PortId),
-    ClosePort(PortId),
+    ClosePort(ControlMessageClosePort),
     PortPeerChangedBlock(PortId),
     PortPeerChangedUnblock(PortId, CompId),
+}
 #[derive(Debug)]
 pub struct ControlMessageClosePort {
     pub port_to_close: PortId, // ID of the receiving port
     pub closed_in_sync_round: bool, // needed to ensure correct handling of errors
+}
 // -----------------------------------------------------------------------------
 // Messages (generic)
 // -----------------------------------------------------------------------------
 #[derive(Debug)]
 pub struct MessageSyncHeader {
     pub sync_round: u32,
     pub sending_id: CompId,
     pub highest_id: CompId,
+}
 #[derive(Debug)]
 pub enum Message {
     Data(DataMessage),
     Sync(SyncMessage),
     Control(ControlMessage),
     Poll,
+}
 impl Message {
     pub(crate) fn target_port(&self) -> Option<PortId> {
         match self {
             Message::Data(v) =>
                 return Some(v.data_header.target_port),
             Message::Control(v) =>
                 return v.target_port_id,
             Message::Sync(_) =>
                 return None,
             Message::Poll =>
                 return None,
+        }
+    }
     pub(crate) fn modify_target_port(&mut self, port_id: PortId) {
         match self {
             Message::Data(v) =>
                 v.data_header.target_port = port_id,
             Message::Control(v) =>
                 v.target_port_id = Some(port_id),
             Message::Sync(_) => unreachable!(), // should never be called for this message type
             Message::Poll => unreachable!(),
+        }
+    }
+}

src/runtime2/component/component.rs

➞

Show inline comments

 use crate::protocol::eval::{Prompt, EvalError, ValueGroup, PortId as EvalPortId};
 use crate::protocol::*;
 use crate::runtime2::*;
 use crate::runtime2::communication::*;
 use super::{CompCtx, CompPDL, CompId};
 use super::component_context::*;
 use super::component_random::*;
 use super::component_internet::*;
 use super::control_layer::*;
 use super::consensus::*;
 pub enum CompScheduling {
     Immediate,
     Requeue,
     Sleep,
     Exit,
+}
 /// Generic representation of a component (as viewed by a scheduler).
 pub(crate) trait Component {
     /// Called upon the creation of the component. Note that the scheduler
     /// context is officially running another component (the component that is
     /// creating the new component).
     fn on_creation(&mut self, comp_id: CompId, sched_ctx: &SchedulerCtx);
     /// Called when a component crashes or wishes to exit. So is not called
     /// right before destruction, other components may still hold a handle to
     /// the component and send it messages!
     fn on_shutdown(&mut self, sched_ctx: &SchedulerCtx);
     /// Called if the component is created by another component and the messages
     /// are being transferred between the two.
     fn adopt_message(&mut self, comp_ctx: &mut CompCtx, message: DataMessage);
     /// Called if the component receives a new message. The component is
     /// responsible for deciding where that messages goes.
     fn handle_message(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx, message: Message);
     /// Called if the component's routine should be executed. The return value
     /// can be used to indicate when the routine should be run again.
     fn run(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx) -> Result<CompScheduling, EvalError>;
+}
 /// Representation of the generic operating mode of a component.
 #[derive(Debug, Copy, Clone, PartialEq, Eq)]
 pub(crate) enum CompMode {
     NonSync, // not in sync mode
     Sync, // in sync mode, can interact with other components
     SyncEnd, // awaiting a solution, i.e. encountered the end of the sync block
     BlockedGet, // blocked because we need to receive a message on a particular port
     BlockedPut, // component is blocked because the port is blocked
     BlockedSelect, // waiting on message to complete the select statement
     StartExit, // temporary state: if encountered then we start the shutdown process
     BusyExit, // temporary state: waiting for Acks for all the closed ports
     Exit, // exiting: shutdown process started, now waiting until the reference count drops to 0
+}
 impl CompMode {
     pub(crate) fn is_in_sync_block(&self) -> bool {
         use CompMode::*;
         match self {
             Sync | SyncEnd | BlockedGet | BlockedPut | BlockedSelect => true,
             NonSync | StartExit | BusyExit | Exit => false,
+        }
+    }
+}
 /// Component execution state: the execution mode along with some descriptive
 /// fields. Fields are public for ergonomic reasons, use member functions when
 /// appropriate.
 pub(crate) struct CompExecState {
     pub mode: CompMode,
     pub mode_port: PortId, // valid if blocked on a port (put/get)
     pub mode_value: ValueGroup, // valid if blocked on a put
+}
 impl CompExecState {
     pub(crate) fn new() -> Self {
         return Self{
             mode: CompMode::NonSync,
             mode_port: PortId::new_invalid(),
             mode_value: ValueGroup::default(),
+        }
+    }
     pub(crate) fn set_as_blocked_get(&mut self, port: PortId) {
         self.mode = CompMode::BlockedGet;
         self.mode_port = port;
         debug_assert!(self.mode_value.values.is_empty());
+    }
     pub(crate) fn is_blocked_on_get(&self, port: PortId) -> bool {
         return
             self.mode == CompMode::BlockedGet &&
             self.mode_port == port;
+    }
     pub(crate) fn set_as_blocked_put(&mut self, port: PortId, value: ValueGroup) {
         self.mode = CompMode::BlockedPut;
         self.mode_port = port;
         self.mode_value = value;
+    }
     pub(crate) fn is_blocked_on_put(&self, port: PortId) -> bool {
         return
             self.mode == CompMode::BlockedPut &&
             self.mode_port == port;
+    }
+}
 /// Creates a new component based on its definition. Meaning that if it is a
 /// user-defined component then we set up the PDL code state. Otherwise we
 /// construct a custom component. This does NOT take care of port and message
 /// management.
 pub(crate) fn create_component(
     protocol: &ProtocolDescription,
     definition_id: ProcedureDefinitionId, type_id: TypeId,
     arguments: ValueGroup, num_ports: usize
 ) -> Box<dyn Component> {
     let definition = &protocol.heap[definition_id];
     debug_assert!(definition.kind == ProcedureKind::Primitive || definition.kind == ProcedureKind::Composite);
     if definition.source.is_builtin() {
         // Builtin component
         let component: Box<dyn Component> = match definition.source {
             ProcedureSource::CompRandomU32 => Box::new(ComponentRandomU32::new(arguments)),
             ProcedureSource::CompTcpClient => Box::new(ComponentTcpClient::new(arguments)),
             _ => unreachable!(),
         };
         return component;
     } else {
         // User-defined component
         let prompt = Prompt::new(
             &protocol.types, &protocol.heap,
             definition_id, type_id, arguments
         );
         let component = CompPDL::new(prompt, num_ports);
         return Box::new(component);
+    }
+}
 // -----------------------------------------------------------------------------
 // Generic component messaging utilities (for sending and receiving)
 // -----------------------------------------------------------------------------
 /// Default handling of sending a data message. In case the port is blocked then
 /// the `ExecState` will become blocked as well. Note that
 /// `default_handle_control_message` will ensure that the port becomes
 /// unblocked if so instructed by the receiving component. The returned
 /// scheduling value must be used.
 #[must_use]
 pub(crate) fn default_send_data_message(
     exec_state: &mut CompExecState, transmitting_port_id: PortId, value: ValueGroup,
     sched_ctx: &SchedulerCtx, consensus: &mut Consensus, comp_ctx: &mut CompCtx
 ) -> CompScheduling {
+) -> Result<CompScheduling, String> { // @nocommit: Something better than Err(String)
     debug_assert_eq!(exec_state.mode, CompMode::Sync);
     // TODO: Handle closed ports
     let port_handle = comp_ctx.get_port_handle(transmitting_port_id);
     let port_info = comp_ctx.get_port(port_handle);
     debug_assert_eq!(port_info.kind, PortKind::Putter);
     if port_info.state.is_blocked() {
     if port_info.state == PortState::Closed {
         // Note: normally peer is eventually consistent, but if it has shut down
         // then we can be sure it is consistent (I think?)
         return Err(format!("Cannot send on this port, as the peer (id:{}) has shut down", port_info.peer_comp_id.0))
     } else if port_info.state.is_blocked() {
         // Port is blocked, so we cannot send
         exec_state.set_as_blocked_put(transmitting_port_id, value);
         return CompScheduling::Sleep;
+        return Ok(CompScheduling::Sleep);
     } else {
         // Port is not blocked, so send to the peer
         let peer_handle = comp_ctx.get_peer_handle(port_info.peer_comp_id);
         let peer_info = comp_ctx.get_peer(peer_handle);
         let annotated_message = consensus.annotate_data_message(comp_ctx, port_info, value);
         peer_info.handle.send_message(&sched_ctx.runtime, Message::Data(annotated_message), true);
         return CompScheduling::Immediate;
+        return Ok(CompScheduling::Immediate);
+    }
+}
 pub(crate) enum IncomingData {
     PlacedInSlot,
     SlotFull(DataMessage),
+}
 /// Default handling of receiving a data message. In case there is no room for
 /// the message it is returned from this function. Note that this function is
 /// different from PDL code performing a `get` on a port; this is the case where
 /// the message first arrives at the component.
 // NOTE: This is supposed to be a somewhat temporary implementation. It would be
 //  nicest if the sending component can figure out it cannot send any more data.
 #[must_use]
 pub(crate) fn default_handle_incoming_data_message(
     exec_state: &mut CompExecState, port_value_slot: &mut Option<DataMessage>,
     comp_ctx: &mut CompCtx, incoming_message: DataMessage,
     sched_ctx: &SchedulerCtx, control: &mut ControlLayer
 ) -> IncomingData {
     let target_port_id = incoming_message.data_header.target_port;
     if port_value_slot.is_none() {
         // We can put the value in the slot
         *port_value_slot = Some(incoming_message);
         // Check if we're blocked on receiving this message.
         dbg_code!({
             // Our port cannot have been blocked itself, because we're able to
             // directly insert the message into its slot.
             let port_handle = comp_ctx.get_port_handle(target_port_id);
             assert!(!comp_ctx.get_port(port_handle).state.is_blocked());
         });
         if exec_state.is_blocked_on_get(target_port_id) {
             // Return to normal operation
             exec_state.mode = CompMode::Sync;
             exec_state.mode_port = PortId::new_invalid();
             debug_assert!(exec_state.mode_value.values.is_empty());
+        }
         return IncomingData::PlacedInSlot
     } else {
         // Slot is already full, so if the port was previously opened, it will
         // now become closed
         let port_handle = comp_ctx.get_port_handle(target_port_id);
         let port_info = comp_ctx.get_port_mut(port_handle);
         debug_assert!(port_info.state == PortState::Open || port_info.state.is_blocked()); // i.e. not closed, but will go off if more states are added in the future
         if port_info.state == PortState::Open {
             comp_ctx.set_port_state(port_handle, PortState::BlockedDueToFullBuffers);
             let (peer_handle, message) =
                 control.initiate_port_blocking(comp_ctx, port_handle);
             let peer = comp_ctx.get_peer(peer_handle);
             peer.handle.send_message(&sched_ctx.runtime, Message::Control(message), true);
+        }
         return IncomingData::SlotFull(incoming_message)
+    }
+}
 /// Default handling that has been received through a `get`. Will check if any
 /// more messages are waiting, and if the corresponding port was blocked because
 /// of full buffers (hence, will use the control layer to make sure the peer
 /// will become unblocked).
 pub(crate) fn default_handle_received_data_message(
     targeted_port: PortId, slot: &mut Option<DataMessage>, inbox_backup: &mut Vec<DataMessage>,
     comp_ctx: &mut CompCtx, sched_ctx: &SchedulerCtx, control: &mut ControlLayer
 ) {
     debug_assert!(slot.is_none()); // because we've just received from it
     // Check if there are any more messages in the backup buffer
     let port_handle = comp_ctx.get_port_handle(targeted_port);
     let port_info = comp_ctx.get_port(port_handle);
     for message_index in 0..inbox_backup.len() {
         let message = &inbox_backup[message_index];
         if message.data_header.target_port == targeted_port {
             // One more message, place it in the slot
             let message = inbox_backup.remove(message_index);
             debug_assert!(port_info.state.is_blocked()); // since we're removing another message from the backup
             *slot = Some(message);
             return;
+        }
+    }
     // Did not have any more messages, so if we were blocked, then we need to
     // unblock the port now (and inform the peer of this unblocking)
     if port_info.state == PortState::BlockedDueToFullBuffers {
         comp_ctx.set_port_state(port_handle, PortState::Open);
         let (peer_handle, message) = control.cancel_port_blocking(comp_ctx, port_handle);
         let peer_info = comp_ctx.get_peer(peer_handle);
         peer_info.handle.send_message(&sched_ctx.runtime, Message::Control(message), true);
+    }
+}
 /// Handles control messages in the default way. Note that this function may
 /// take a lot of actions in the name of the caller: pending messages may be
 /// sent, ports may become blocked/unblocked, etc. So the execution
 /// (`CompExecState`), control (`ControlLayer`) and consensus (`Consensus`)
 /// state may all change.
 pub(crate) fn default_handle_control_message(
     exec_state: &mut CompExecState, control: &mut ControlLayer, consensus: &mut Consensus,
     message: ControlMessage, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx
 ) {
+) -> Result<(), String> { // @nocommit, use something else than Err(String)
     match message.content {
         ControlMessageContent::Ack => {
             default_handle_ack(control, message.id, sched_ctx, comp_ctx);
         },
         ControlMessageContent::BlockPort(port_id) => {
             // One of our messages was accepted, but the port should be
             // blocked.
             let port_handle = comp_ctx.get_port_handle(port_id);
             let port_info = comp_ctx.get_port(port_handle);
             debug_assert_eq!(port_info.kind, PortKind::Putter);
             if port_info.state == PortState::Open {
                 // only when open: we don't do this when closed, and we we don't do this if we're blocked due to peer changes
                 comp_ctx.set_port_state(port_handle, PortState::BlockedDueToFullBuffers);
+            }
         },
-        ControlMessageContent::ClosePort(port_id) => {
+        ControlMessageContent::ClosePort(content) => {
             // Request to close the port. We immediately comply and remove
             // the component handle as well
             let port_handle = comp_ctx.get_port_handle(port_id);
             let peer_comp_id = comp_ctx.get_port(port_handle).peer_comp_id;
             let port_handle = comp_ctx.get_port_handle(content.port_id);
             let port_info = comp_ctx.get_port_mut(port_handle);
             let peer_comp_id = port_info.peer_comp_id;
             let peer_handle = comp_ctx.get_peer_handle(peer_comp_id);
             // We're closing the port, so we will always update the peer of the
             // port (in case of error messages)
             port_info.peer_comp_id = message.sender_comp_id;
             // One exception to sending an `Ack` is if we just closed the
             // port ourselves, meaning that the `ClosePort` messages got
             // sent to one another.
             if let Some(control_id) = control.has_close_port_entry(port_handle, comp_ctx) {
                 // The two components (sender and this component) are closing
                 // the channel at the same time.
                 default_handle_ack(control, control_id, sched_ctx, comp_ctx);
             } else {
                 // Respond to the message
                 let last_instruction = port_info.last_instruction;
                 let port_was_used = last_instruction != PortInstruction::None;
                 default_send_ack(message.id, peer_handle, sched_ctx, comp_ctx);
                 comp_ctx.remove_peer(sched_ctx, port_handle, peer_comp_id, false); // do not remove if closed
                 comp_ctx.set_port_state(port_handle, PortState::Closed); // now set to closed
                 // Make sure that we've not reached an error condition. Note
                 // that if this condition is not met, then we don't error out
                 // now, but we may error out in the next sync block when we
                 // try to `put`/`get` on the port. This condition makes sure
                 // that if we have a successful sync round, followed by the peer
                 // closing the port, that we don't consider the sync round to
                 // have failed by mistake.
                 if content.closed_in_sync_round && exec_state.mode.is_in_sync_block() && port_was_used {
                     let error_message = match last_instruction {
                         PortInstruction::None => unreachable!(), // port was used
                         PortInstruction::NoSource => format!(
                             "Peer component (id:{}) shut down, so operation on port cannot have succeeded",
                             message.sender_comp_id.0
                         ),
                         PortInstruction::SourceLocation(source_location) => format!(
                             "Peer component (id:{}) shut down, so this operation cannot have succeeded",
                             message.sender_comp_id.0
                         ),
                     };
                     return Err(error_message);
+                }
+            }
         },
         ControlMessageContent::UnblockPort(port_id) => {
             // We were previously blocked (or already closed)
             let port_handle = comp_ctx.get_port_handle(port_id);
             let port_info = comp_ctx.get_port(port_handle);
             debug_assert_eq!(port_info.kind, PortKind::Putter);
             if port_info.state == PortState::BlockedDueToFullBuffers {
                 default_handle_unblock_put(exec_state, consensus, port_handle, sched_ctx, comp_ctx);
+            }
         },
         ControlMessageContent::PortPeerChangedBlock(port_id) => {
             // The peer of our port has just changed. So we are asked to
             // temporarily block the port (while our original recipient is
             // potentially rerouting some of the in-flight messages) and
             // Ack. Then we wait for the `unblock` call.
             debug_assert_eq!(message.target_port_id, Some(port_id));
             let port_handle = comp_ctx.get_port_handle(port_id);
             comp_ctx.set_port_state(port_handle, PortState::BlockedDueToPeerChange);
             let port_info = comp_ctx.get_port(port_handle);
             let peer_handle = comp_ctx.get_peer_handle(port_info.peer_comp_id);
             default_send_ack(message.id, peer_handle, sched_ctx, comp_ctx);
         },
         ControlMessageContent::PortPeerChangedUnblock(new_port_id, new_comp_id) => {
             let port_handle = comp_ctx.get_port_handle(message.target_port_id.unwrap());
             let port_info = comp_ctx.get_port(port_handle);
             debug_assert!(port_info.state == PortState::BlockedDueToPeerChange);
             let old_peer_id = port_info.peer_comp_id;
             comp_ctx.remove_peer(sched_ctx, port_handle, old_peer_id, false);
             let port_info = comp_ctx.get_port_mut(port_handle);
             port_info.peer_comp_id = new_comp_id;
             port_info.peer_port_id = new_port_id;
             comp_ctx.add_peer(port_handle, sched_ctx, new_comp_id, None);
             default_handle_unblock_put(exec_state, consensus, port_handle, sched_ctx, comp_ctx);
+        }
+    }
     return Ok(());
+}
 /// Handles a component initiating the exiting procedure, and closing all of its
 /// ports. Should only be called once per component (which is ensured by
 /// checking and modifying the mode in the execution state).
 #[must_use]
 pub(crate) fn default_handle_start_exit(
     exec_state: &mut CompExecState, control: &mut ControlLayer,
     sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx
 ) -> CompScheduling {
     debug_assert_eq!(exec_state.mode, CompMode::StartExit);
     sched_ctx.log("Component starting exit");
     exec_state.mode = CompMode::BusyExit;
     // Iterating by index to work around borrowing rules
     for port_index in 0..comp_ctx.num_ports() {
         let port = comp_ctx.get_port_by_index_mut(port_index);
         if port.state == PortState::Closed {
             // Already closed, or in the process of being closed
             continue;
+        }
         // Mark as closed
         let port_id = port.self_id;
         port.state = PortState::Closed;
         // Notify peer of closing
         let port_handle = comp_ctx.get_port_handle(port_id);
         let (peer, message) = control.initiate_port_closing(port_handle, comp_ctx);
         let peer_info = comp_ctx.get_peer(peer);
         peer_info.handle.send_message(&sched_ctx.runtime, Message::Control(message), true);
+    }
     return CompScheduling::Immediate; // to check if we can shut down immediately
+}
 /// Handles a component waiting until all peers are notified that it is quitting
 /// (i.e. after calling `default_handle_start_exit`).
 #[must_use]
 pub(crate) fn default_handle_busy_exit(
     exec_state: &mut CompExecState, control: &ControlLayer,
     sched_ctx: &SchedulerCtx
 ) -> CompScheduling {
     debug_assert_eq!(exec_state.mode, CompMode::BusyExit);
     if control.has_acks_remaining() {
         sched_ctx.log("Component busy exiting, still has `Ack`s remaining");
         return CompScheduling::Sleep;
     } else {
         sched_ctx.log("Component busy exiting, now shutting down");
         exec_state.mode = CompMode::Exit;
         return CompScheduling::Exit;
+    }
+}
 /// Handles a potential synchronous round decision. If there was a decision then
 /// the `Some(success)` value indicates whether the round succeeded or not.
 /// Might also end up changing the `ExecState`.
 pub(crate) fn default_handle_sync_decision(
     exec_state: &mut CompExecState, decision: SyncRoundDecision,
     consensus: &mut Consensus
 ) -> Option<bool> {
     debug_assert_eq!(exec_state.mode, CompMode::SyncEnd);
     let success = match decision {
         SyncRoundDecision::None => return None,
         SyncRoundDecision::Solution => true,
         SyncRoundDecision::Failure => false,
     };
     debug_assert_eq!(exec_state.mode, CompMode::SyncEnd);
     if success {
         exec_state.mode = CompMode::NonSync;
         consensus.notify_sync_decision(decision);
         return Some(true);
     } else {
         exec_state.mode = CompMode::StartExit;
         return Some(false);
+    }
+}
 #[inline]
 pub(crate) fn default_handle_exit(_exec_state: &CompExecState) -> CompScheduling {
     debug_assert_eq!(_exec_state.mode, CompMode::Exit);
     return CompScheduling::Exit;
+}
 // -----------------------------------------------------------------------------
 // Internal messaging/state utilities
 // -----------------------------------------------------------------------------
 /// Handles an `Ack` for the control layer.
 fn default_handle_ack(
     control: &mut ControlLayer, control_id: ControlId,
     sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx
 ) {
     // Since an `Ack` may cause another one, handle them in a loop
     let mut to_ack = control_id;
     loop {
         let (action, new_to_ack) = control.handle_ack(to_ack, sched_ctx, comp_ctx);
         match action {
             AckAction::SendMessage(target_comp, message) => {
                 // FIX @NoDirectHandle
                 let mut handle = sched_ctx.runtime.get_component_public(target_comp);
                 handle.send_message(&sched_ctx.runtime, Message::Control(message), true);
                 let _should_remove = handle.decrement_users();
                 debug_assert!(_should_remove.is_none());
             },
             AckAction::ScheduleComponent(to_schedule) => {
                 // FIX @NoDirectHandle
                 let mut handle = sched_ctx.runtime.get_component_public(to_schedule);
                 // Note that the component is intentionally not
                 // sleeping, so we just wake it up
                 debug_assert!(!handle.sleeping.load(std::sync::atomic::Ordering::Acquire));
                 let key = unsafe { to_schedule.upgrade() };
                 sched_ctx.runtime.enqueue_work(key);
                 let _should_remove = handle.decrement_users();
                 debug_assert!(_should_remove.is_none());
             },
             AckAction::None => {}
+        }
         match new_to_ack {
             Some(new_to_ack) => to_ack = new_to_ack,
             None => break,
+        }
+    }
+}
 /// Little helper for sending the most common kind of `Ack`
 fn default_send_ack(
     causer_of_ack_id: ControlId, peer_handle: LocalPeerHandle,
     sched_ctx: &SchedulerCtx, comp_ctx: &CompCtx
 ) {
     let peer_info = comp_ctx.get_peer(peer_handle);
     peer_info.handle.send_message(&sched_ctx.runtime, Message::Control(ControlMessage{
         id: causer_of_ack_id,
         sender_comp_id: comp_ctx.id,
         target_port_id: None,
         content: ControlMessageContent::Ack
     }), true);
+}
 /// Handles the unblocking of a putter port. In case there is a pending message
 /// on that port then it will be sent.
 fn default_handle_unblock_put(
     exec_state: &mut CompExecState, consensus: &mut Consensus,
     port_handle: LocalPortHandle, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx,
 ) {
     let port_info = comp_ctx.get_port_mut(port_handle);
     let port_id = port_info.self_id;
     debug_assert!(port_info.state.is_blocked());
     port_info.state = PortState::Open;
     if exec_state.is_blocked_on_put(port_id) {
         // Annotate the message that we're going to send
         let port_info = comp_ctx.get_port(port_handle); // for immutable access
         debug_assert_eq!(port_info.kind, PortKind::Putter);
         let to_send = exec_state.mode_value.take();
         let to_send = consensus.annotate_data_message(comp_ctx, port_info, to_send);
         // Retrieve peer to send the message
         let peer_handle = comp_ctx.get_peer_handle(port_info.peer_comp_id);
         let peer_info = comp_ctx.get_peer(peer_handle);
         peer_info.handle.send_message(&sched_ctx.runtime, Message::Data(to_send), true);
         exec_state.mode = CompMode::Sync; // because we're blocked on a `put`, we must've started in the sync state.
         exec_state.mode_port = PortId::new_invalid();
+    }
+}
 #[inline]
 pub(crate) fn port_id_from_eval(port_id: EvalPortId) -> PortId {
     return PortId(port_id.id);
+}
 #[inline]
 pub(crate) fn port_id_to_eval(port_id: PortId) -> EvalPortId {
     return EvalPortId{ id: port_id.0 };
+}

src/runtime2/component/component_context.rs

➞

Show inline comments

 use crate::runtime2::scheduler::*;
 use crate::runtime2::runtime::*;
 use crate::runtime2::communication::*;
 use crate::protocol::ExpressionId;
 /// Helper struct to remember when the last operation on the port took place
 #[derive(Debug, PartialEq, Copy, Clone)]
 pub enum PortInstruction {
     None,
     NoSource,
     SourceLocation(ExpressionId),
+}
 #[derive(Debug)]
 pub struct Port {
     pub self_id: PortId,
     pub peer_comp_id: CompId, // eventually consistent
     pub peer_port_id: PortId, // eventually consistent
     pub kind: PortKind,
     pub state: PortState,
     pub last_instruction: PortInstruction,
     #[cfg(debug_assertions)] pub(crate) associated_with_peer: bool,
 }
 pub struct Peer {
     pub id: CompId,
     pub num_associated_ports: u32,
     pub(crate) handle: CompHandle,
 }
 /// Port and peer management structure. Will keep a local reference counter to
 /// the ports associate with peers, additionally manages the atomic reference
 /// counter associated with the peers' component handles.
 pub struct CompCtx {
     pub id: CompId,
     ports: Vec<Port>,
     peers: Vec<Peer>,
     port_id_counter: u32,
 }
 #[derive(Copy, Clone, PartialEq, Eq)]
 pub struct LocalPortHandle(PortId);
 #[derive(Copy, Clone)]
 pub struct LocalPeerHandle(CompId);
 impl CompCtx {
     /// Creates a new component context based on a reserved entry in the
     /// component store. This reservation is used such that we already know our
     /// assigned ID.
     pub(crate) fn new(reservation: &CompReserved) -> Self {
         return Self{
             id: reservation.id(),
             ports: Vec::new(),
             peers: Vec::new(),
             port_id_counter: 0,
         }
     }
     /// Creates a new channel that is fully owned by the component associated
     /// with this context.
     pub(crate) fn create_channel(&mut self) -> Channel {
         let putter_id = PortId(self.take_port_id());
         let getter_id = PortId(self.take_port_id());
         self.ports.push(Port{
             self_id: putter_id,
             peer_port_id: getter_id,
             kind: PortKind::Putter,
             state: PortState::Open,
             peer_comp_id: self.id,
             last_instruction: PortInstruction::None,
             #[cfg(debug_assertions)] associated_with_peer: false,
         });
         self.ports.push(Port{
             self_id: getter_id,
             peer_port_id: putter_id,
             kind: PortKind::Getter,
             state: PortState::Open,
             peer_comp_id: self.id,
             last_instruction: PortInstruction::None,
             #[cfg(debug_assertions)] associated_with_peer: false,
         });
         return Channel{ putter_id, getter_id };
     }
     /// Adds a new port. Make sure to call `add_peer` afterwards.
     pub(crate) fn add_port(&mut self, peer_comp_id: CompId, peer_port_id: PortId, kind: PortKind, state: PortState) -> LocalPortHandle {
         let self_id = PortId(self.take_port_id());
         self.ports.push(Port{
             self_id, peer_comp_id, peer_port_id, kind, state,
             last_instruction: PortInstruction::None,
             #[cfg(debug_assertions)] associated_with_peer: false,
         });
         return LocalPortHandle(self_id);
     }
     /// Removes a port. Make sure you called `remove_peer` first.
     pub(crate) fn remove_port(&mut self, port_handle: LocalPortHandle) -> Port {
         let port_index = self.must_get_port_index(port_handle);
         let port = self.ports.remove(port_index);
         dbg_code!(assert!(!port.associated_with_peer));
         return port;
     }
     /// Adds a new peer. This must be called for every port, no matter the
     /// component the channel is connected to. If a `CompHandle` is supplied,
     /// then it will be used to add the peer. Otherwise it will be retrieved
     /// from the runtime using its ID.
     pub(crate) fn add_peer(&mut self, port_handle: LocalPortHandle, sched_ctx: &SchedulerCtx, peer_comp_id: CompId, handle: Option<&CompHandle>) {
         let self_id = self.id;
         let port = self.get_port_mut(port_handle);
         debug_assert_eq!(port.peer_comp_id, peer_comp_id);
         dbg_code!(assert!(!port.associated_with_peer));
         if !Self::requires_peer_reference(port, self_id, false) {
             return;
         }
         dbg_code!(port.associated_with_peer = true);
         match self.get_peer_index_by_id(peer_comp_id) {
             Some(peer_index) => {
                 let peer = &mut self.peers[peer_index];
                 peer.num_associated_ports += 1;
             },
             None => {
                 let handle = match handle {
                     Some(handle) => handle.clone(),
                     None => sched_ctx.runtime.get_component_public(peer_comp_id)
                 };
                 self.peers.push(Peer{
                     id: peer_comp_id,
                     num_associated_ports: 1,
                     handle,
                 });
             }
         }
     }
     /// Removes a peer associated with a port.
     pub(crate) fn remove_peer(&mut self, sched_ctx: &SchedulerCtx, port_handle: LocalPortHandle, peer_id: CompId, also_remove_if_closed: bool) {
         let self_id = self.id;
         let port = self.get_port_mut(port_handle);
         debug_assert_eq!(port.peer_comp_id, peer_id);
         if !Self::requires_peer_reference(port, self_id, also_remove_if_closed) {
             return;
         }
         dbg_code!(assert!(port.associated_with_peer));
         dbg_code!(port.associated_with_peer = false);
         let peer_index = self.get_peer_index_by_id(peer_id).unwrap();
         let peer = &mut self.peers[peer_index];
         peer.num_associated_ports -= 1;
         if peer.num_associated_ports == 0 {
             let mut peer = self.peers.remove(peer_index);
             if let Some(key) = peer.handle.decrement_users() {
                 debug_assert_ne!(key.downgrade(), self.id); // should be upheld by the code that shuts down a component
                 sched_ctx.runtime.destroy_component(key);
             }
         }
     }
     pub(crate) fn set_port_state(&mut self, port_handle: LocalPortHandle, new_state: PortState) {
         let port_info = self.get_port_mut(port_handle);
         debug_assert_ne!(port_info.state, PortState::Closed); // because then we do not expect to change the state
         port_info.state = new_state;
     }
     pub(crate) fn get_port_handle(&self, port_id: PortId) -> LocalPortHandle {
         return LocalPortHandle(port_id);
     }
     // should perhaps be revised, used in main inbox
     pub(crate) fn get_port_index(&self, port_handle: LocalPortHandle) -> usize {
         return self.must_get_port_index(port_handle);
     }
     pub(crate) fn get_peer_handle(&self, peer_id: CompId) -> LocalPeerHandle {
         return LocalPeerHandle(peer_id);
     }
     pub(crate) fn get_port(&self, port_handle: LocalPortHandle) -> &Port {
         let index = self.must_get_port_index(port_handle);
         return &self.ports[index];
     }
     pub(crate) fn get_port_mut(&mut self, port_handle: LocalPortHandle) -> &mut Port {
         let index = self.must_get_port_index(port_handle);
         return &mut self.ports[index];
     }
     pub(crate) fn get_port_by_index_mut(&mut self, index: usize) -> &mut Port {
         return &mut self.ports[index];
     }
     pub(crate) fn get_peer(&self, peer_handle: LocalPeerHandle) -> &Peer {
         let index = self.must_get_peer_index(peer_handle);
         return &self.peers[index];
     }
     pub(crate) fn get_peer_mut(&mut self, peer_handle: LocalPeerHandle) -> &mut Peer {
         let index = self.must_get_peer_index(peer_handle);
         return &mut self.peers[index];
     }
     #[inline]
     pub(crate) fn iter_ports(&self) -> impl Iterator<Item=&Port> {
         return self.ports.iter();
     }
     #[inline]
     pub(crate) fn iter_ports_mut(&mut self) -> impl Iterator<Item=&mut Port> {
         return self.ports.iter_mut();
     }
     #[inline]
     pub(crate) fn iter_peers(&self) -> impl Iterator<Item=&Peer> {
         return self.peers.iter();
     }
     #[inline]
     pub(crate) fn num_ports(&self) -> usize {
         return self.ports.len();
     }
     // -------------------------------------------------------------------------
     // Local utilities
     // -------------------------------------------------------------------------
     #[inline]
     fn requires_peer_reference(port: &Port, self_id: CompId, required_if_closed: bool) -> bool {
         return (port.state != PortState::Closed || required_if_closed) && port.peer_comp_id != self_id;
     }
     fn must_get_port_index(&self, handle: LocalPortHandle) -> usize {
         for (index, port) in self.ports.iter().enumerate() {
             if port.self_id == handle.0 {
                 return index;
             }
         }
         unreachable!()
     }
     fn must_get_peer_index(&self, handle: LocalPeerHandle) -> usize {
         for (index, peer) in self.peers.iter().enumerate() {
             if peer.id == handle.0 {
                 return index;
             }
         }
         unreachable!()
     }
     fn get_peer_index_by_id(&self, comp_id: CompId) -> Option<usize> {
         for (index, peer) in self.peers.iter().enumerate() {
             if peer.id == comp_id {
                 return Some(index);
             }
         }
         return None;
     }
     fn take_port_id(&mut self) -> u32 {
         let port_id = self.port_id_counter;
         self.port_id_counter = self.port_id_counter.wrapping_add(1);
         return port_id;
     }
 }
@@ \ No newline at end of file @@

src/runtime2/component/consensus.rs

➞

Show inline comments

 use crate::protocol::eval::ValueGroup;
 use crate::runtime2::scheduler::*;
 use crate::runtime2::runtime::*;
 use crate::runtime2::communication::*;
 use super::component_context::*;
 pub struct PortAnnotation {
     self_comp_id: CompId,
     self_port_id: PortId,
     peer_comp_id: CompId, // only valid for getter ports
     peer_port_id: PortId, // only valid for getter ports
     peer_discovered: bool, // only valid for getter ports
     mapping: Option<u32>,
     kind: PortKind,
+}
 impl PortAnnotation {
     fn new(comp_id: CompId, port_id: PortId, kind: PortKind) -> Self {
         return Self{
             self_comp_id: comp_id,
             self_port_id: port_id,
             peer_comp_id: CompId::new_invalid(),
             peer_port_id: PortId::new_invalid(),
             peer_discovered: false,
             mapping: None,
             kind,
+        }
+    }
+}
 #[derive(Debug, Eq, PartialEq)]
 enum Mode {
     NonSync,
     SyncBusy,
     SyncAwaitingSolution,
     SelectBusy,
     SelectWait,
+}
 struct SolutionCombiner {
     solution: SyncPartialSolution,
     matched_channels: usize,
+}
 impl SolutionCombiner {
     fn new() -> Self {
         return Self {
             solution: SyncPartialSolution::default(),
             matched_channels: 0,
+        }
+    }
     #[inline]
     fn has_contributions(&self) -> bool {
         return !self.solution.channel_mapping.is_empty();
+    }
     /// Returns a decision for the current round. If there is no decision (yet)
     /// then `RoundDecision::None` is returned.
     fn get_decision(&self) -> SyncRoundDecision {
         if self.matched_channels == self.solution.channel_mapping.len() {
             debug_assert_ne!(self.solution.decision, SyncRoundDecision::None);
             return self.solution.decision;
+        }
         return SyncRoundDecision::None; // even in case of failure: wait for everyone.
+    }
     fn combine_with_partial_solution(&mut self, partial: SyncPartialSolution) {
         debug_assert_ne!(self.solution.decision, SyncRoundDecision::Solution);
         debug_assert_ne!(partial.decision, SyncRoundDecision::Solution);
         if partial.decision == SyncRoundDecision::Failure {
             self.solution.decision = SyncRoundDecision::Failure;
+        }
         for entry in partial.channel_mapping {
             let channel_index = if entry.getter.is_some() && entry.putter.is_some() {
                 let channel_index = self.solution.channel_mapping.len();
                 self.solution.channel_mapping.push(entry);
                 channel_index
             } else if let Some(putter) = entry.putter {
                 self.combine_with_putter_port(putter)
             } else if let Some(getter) = entry.getter {
                 self.combine_with_getter_port(getter)
             } else {
                 unreachable!(); // both putter and getter are None
             };
             let channel = &self.solution.channel_mapping[channel_index];
             if let Some(consistent) = Self::channel_is_consistent(channel) {
                 if !consistent {
                     self.solution.decision = SyncRoundDecision::Failure;
+                }
                 self.matched_channels += 1;
+            }
+        }
         self.update_solution();
+    }
     /// Combines the currently stored global solution (if any) with the newly
     /// provided local solution. Make sure to check the `has_decision` return
     /// value afterwards.
     fn combine_with_local_solution(&mut self, _comp_id: CompId, solution: SyncLocalSolution) {
         debug_assert_ne!(self.solution.decision, SyncRoundDecision::Solution);
         // Combine partial solution with the local solution entries
         for entry in solution {
             // Match the current entry up with its peer endpoint, or add a new
             // entry.
             let channel_index = match entry {
                 SyncLocalSolutionEntry::Putter(putter) => {
                     self.combine_with_putter_port(putter)
                 },
                 SyncLocalSolutionEntry::Getter(getter) => {
                     self.combine_with_getter_port(getter)
+                }
             };
             // Check if channel is now consistent
             let channel = &self.solution.channel_mapping[channel_index];
             if let Some(consistent) = Self::channel_is_consistent(channel) {
                 if !consistent {
                     self.solution.decision = SyncRoundDecision::Failure;
+                }
                 self.matched_channels += 1;
+            }
+        }
         self.update_solution();
+    }
     /// Takes whatever partial solution is present in the solution combiner and
     /// returns it. The solution combiner's solution will end up being empty.
     /// This is used when a new leader is found and we need to pass along our
     /// partial results.
     fn take_partial_solution(&mut self) -> SyncPartialSolution {
         let mut partial_solution = SyncPartialSolution::default();
         std::mem::swap(&mut partial_solution, &mut self.solution);
         self.clear();
         return partial_solution;
+    }
     fn clear(&mut self) {
         self.solution.channel_mapping.clear();
         self.solution.decision = SyncRoundDecision::None;
         self.matched_channels = 0;
+    }
     // --- Small utilities for combining solutions
     fn combine_with_putter_port(&mut self, putter: SyncSolutionPutterPort) -> usize {
         let channel_index = self.get_channel_index_for_putter(putter.self_comp_id, putter.self_port_id);
         if let Some(channel_index) = channel_index {
             let channel = &mut self.solution.channel_mapping[channel_index];
             debug_assert!(channel.putter.is_none());
             channel.putter = Some(putter);
             return channel_index;
         } else {
             let channel_index = self.solution.channel_mapping.len();
             self.solution.channel_mapping.push(SyncSolutionChannel{
                 putter: Some(putter),
                 getter: None,
             });
             return channel_index;
+        }
+    }
     fn combine_with_getter_port(&mut self, getter: SyncSolutionGetterPort) -> usize {
         let channel_index = self.get_channel_index_for_getter(getter.peer_comp_id, getter.peer_port_id);
         if let Some(channel_index) = channel_index {
             let channel = &mut self.solution.channel_mapping[channel_index];
             debug_assert!(channel.getter.is_none());
             channel.getter = Some(getter);
             return channel_index;
         } else {
             let channel_index = self.solution.channel_mapping.len();
             self.solution.channel_mapping.push(SyncSolutionChannel{
                 putter: None,
                 getter: Some(getter)
             });
             return channel_index;
+        }
+    }
     /// Retrieve index of the channel containing a getter port that has received
     /// from the specified putter port.
     fn get_channel_index_for_putter(&self, putter_comp_id: CompId, putter_port_id: PortId) -> Option<usize> {
         for (channel_index, channel) in self.solution.channel_mapping.iter().enumerate() {
             if let Some(getter) = &channel.getter {
                 if getter.peer_comp_id == putter_comp_id && getter.peer_port_id == putter_port_id {
                     return Some(channel_index);
+                }
+            }
+        }
         return None;
+    }
     /// Retrieve index of the channel for a getter port. To find this channel
     /// the **peer** component/port IDs of the getter port are used.
     fn get_channel_index_for_getter(&self, peer_comp_id: CompId, peer_port_id: PortId) -> Option<usize> {
         for (channel_index, channel) in self.solution.channel_mapping.iter().enumerate() {
             if let Some(putter) = &channel.putter {
                 if putter.self_comp_id == peer_comp_id && putter.self_port_id == peer_port_id {
                     return Some(channel_index);
+                }
+            }
+        }
         return None;
+    }
     fn channel_is_consistent(channel: &SyncSolutionChannel) -> Option<bool> {
         if channel.putter.is_none() || channel.getter.is_none() {
             return None;
+        }
         let putter = channel.putter.as_ref().unwrap();
         let getter = channel.getter.as_ref().unwrap();
         return Some(putter.mapping == getter.mapping);
+    }
     /// Determines the global solution if all components have contributed their
     /// local solutions.
     fn update_solution(&mut self) {
         if self.matched_channels == self.solution.channel_mapping.len() {
             if self.solution.decision != SyncRoundDecision::Failure {
                 self.solution.decision = SyncRoundDecision::Solution;
+            }
+        }
+    }
+}
 /// Tracking consensus state
 pub struct Consensus {
     // General state of consensus manager
     mapping_counter: u32,
     mode: Mode,
     // State associated with sync round
     round_index: u32,
     highest_id: CompId,
     ports: Vec<PortAnnotation>,
     // State associated with arriving at a solution and being a (temporary)
     // leader in the consensus round
     solution: SolutionCombiner,
+}
 impl Consensus {
     pub(crate) fn new() -> Self {
         return Self{
             round_index: 0,
             highest_id: CompId::new_invalid(),
             ports: Vec::new(),
             mapping_counter: 0,
             mode: Mode::NonSync,
             solution: SolutionCombiner::new(),
+        }
+    }
     // -------------------------------------------------------------------------
     // Managing sync state
     // -------------------------------------------------------------------------
     /// Notifies the consensus management that the PDL code has reached the
     /// start of a sync block.
     pub(crate) fn notify_sync_start(&mut self, comp_ctx: &CompCtx) {
         debug_assert_eq!(self.mode, Mode::NonSync);
         self.highest_id = comp_ctx.id;
         self.mapping_counter = 0;
         self.mode = Mode::SyncBusy;
         // Make the internally stored port annotation array consistent with the
         // ports that the component currently owns. They should match by index
         // (i.e. annotation at index `i` corresponds to port `i` in `comp_ctx`).
         let mut needs_setting_ports = false;
         if comp_ctx.num_ports() != self.ports.len() {
             needs_setting_ports = true;
         } else {
             for (idx, port) in comp_ctx.iter_ports().enumerate() {
                 let comp_port_id = port.self_id;
                 let cons_port_id = self.ports[idx].self_port_id;
                 if comp_port_id != cons_port_id {
                     needs_setting_ports = true;
                     break;
+                }
+            }
+        }
         if needs_setting_ports {
             // Reset all ports
             self.ports.clear();
             self.ports.reserve(comp_ctx.num_ports());
             for port in comp_ctx.iter_ports() {
                 self.ports.push(PortAnnotation::new(comp_ctx.id, port.self_id, port.kind));
+            }
         } else {
             // Make sure that we consider all peers as undiscovered again
             for annotation in self.ports.iter_mut() {
                 annotation.peer_discovered = false;
+            }
+        }
+    }
     /// Notifies the consensus management that the PDL code has reached the end
     /// of a sync block. A local solution will be submitted, after which we wait
     /// until the participants in the round (hopefully) reach a conclusion.
     pub(crate) fn notify_sync_end(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &CompCtx) -> SyncRoundDecision {
         debug_assert_eq!(self.mode, Mode::SyncBusy);
         self.mode = Mode::SyncAwaitingSolution;
         // Submit our port mapping as a solution
         let mut local_solution = Vec::with_capacity(self.ports.len());
         for port in &self.ports {
             if let Some(mapping) = port.mapping {
                 let port_handle = comp_ctx.get_port_handle(port.self_port_id);
                 let port_info = comp_ctx.get_port(port_handle);
                 let new_entry = match port_info.kind {
                     PortKind::Putter => SyncLocalSolutionEntry::Putter(SyncSolutionPutterPort{
                         self_comp_id: comp_ctx.id,
                         self_port_id: port_info.self_id,
                         mapping
                     }),
                     PortKind::Getter => SyncLocalSolutionEntry::Getter(SyncSolutionGetterPort{
                         self_comp_id: comp_ctx.id,
                         self_port_id: port_info.self_id,
                         peer_comp_id: port.peer_comp_id,
                         peer_port_id: port.peer_port_id,
                         mapping
                     })
                 };
                 local_solution.push(new_entry);
+            }
+        }
         let decision = self.handle_local_solution(sched_ctx, comp_ctx, comp_ctx.id, local_solution);
         return decision;
+    }
     /// Notifies that a decision has been reached. Note that the caller should
     /// still take the appropriate actions based on the decision it is supplying
     /// to the consensus layer.
     pub(crate) fn notify_sync_decision(&mut self, _decision: SyncRoundDecision) {
         // Reset everything for the next round
         debug_assert_eq!(self.mode, Mode::SyncAwaitingSolution);
         self.mode = Mode::NonSync;
         self.round_index = self.round_index.wrapping_add(1);
         for port in self.ports.iter_mut() {
             port.mapping = None;
+        }
         self.solution.clear();
+    }
     // -------------------------------------------------------------------------
     // Handling inbound and outbound messages
     // -------------------------------------------------------------------------
     /// Prepares a set of values to be sent of a channel.
     pub(crate) fn annotate_data_message(&mut self, comp_ctx: &CompCtx, port_info: &Port, content: ValueGroup) -> DataMessage {
         debug_assert_eq!(self.mode, Mode::SyncBusy); // can only send between sync start and sync end
         debug_assert!(self.ports.iter().any(|v| v.self_port_id == port_info.self_id));
         let data_header = self.create_data_header_and_update_mapping(port_info);
         let sync_header = self.create_sync_header(comp_ctx);
         return DataMessage{ data_header, sync_header, content };
+    }
     /// Handles the arrival of a new data message (needs to be called for every
     /// new data message, even though it might not end up being received). This
     /// is used to determine peers of `get`ter ports.
     // TODO: The use of this function is rather ugly. Find a more robust
     //  scheme about owners of `get`ter ports not knowing about their peers.
     //  (also, figure out why this was written again, I forgot).
     pub(crate) fn handle_incoming_data_message(&mut self, comp_ctx: &CompCtx, message: &DataMessage) {
         let target_handle = comp_ctx.get_port_handle(message.data_header.target_port);
         let target_index = comp_ctx.get_port_index(target_handle);
         let annotation = &mut self.ports[target_index];
         debug_assert!(
             !annotation.peer_discovered || (
                 annotation.peer_comp_id == message.sync_header.sending_id &&
                 annotation.peer_port_id == message.data_header.source_port
+            )
         );
         annotation.peer_comp_id = message.sync_header.sending_id;
         annotation.peer_port_id = message.data_header.source_port;
         annotation.peer_discovered = true;
+    }
     /// Checks if the data message can be received (due to port annotations), if
     /// it can then `true` is returned and the caller is responsible for handing
     /// the message of to the PDL code. Otherwise the message cannot be
     /// received.
     pub(crate) fn try_receive_data_message(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, message: &DataMessage) -> bool {
         debug_assert_eq!(self.mode, Mode::SyncBusy);
         debug_assert!(self.ports.iter().any(|v| v.self_port_id == message.data_header.target_port));
         // Make sure the expected mapping matches the currently stored mapping
         for (peer_port_kind, expected_annotation) in &message.data_header.expected_mapping {
             // Determine our annotation, in order to do so we need to find the
             // port matching the peer ports
             let mut self_annotation = None;
             let mut self_annotation_found = false;
             match peer_port_kind {
                 PortAnnotationKind::Putter(peer_port) => {
                     for self_port in &self.ports {
                         if self_port.kind == PortKind::Getter &&
                             self_port.peer_discovered &&
                             self_port.peer_comp_id == peer_port.self_comp_id &&
                             self_port.peer_port_id == peer_port.self_port_id
+                        {
                             self_annotation = self_port.mapping;
                             self_annotation_found = true;
                             break;
+                        }
+                    }
                 },
                 PortAnnotationKind::Getter(peer_port) => {
                     if peer_port.peer_comp_id == comp_ctx.id {
                         // Peer indicates that we talked to it
                         let self_port_handle = comp_ctx.get_port_handle(peer_port.peer_port_id);
                         let self_port_index = comp_ctx.get_port_index(self_port_handle);
                         self_annotation = self.ports[self_port_index].mapping;
                         self_annotation_found = true;
+                    }
+                }
+            }
             if !self_annotation_found {
                 continue
+            }
             if self_annotation != *expected_annotation {
                 return false;
+            }
+        }
         // Expected mapping matches current mapping, so we will receive the message
         self.set_annotation(message.sync_header.sending_id, &message.data_header);
         // Handle the sync header embedded within the data message
         self.handle_sync_header(sched_ctx, comp_ctx, &message.sync_header);
         return true;
+    }
     /// Receives the sync message and updates the consensus state appropriately.
     pub(crate) fn receive_sync_message(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, message: SyncMessage) -> SyncRoundDecision {
         // Whatever happens: handle the sync header (possibly changing the
         // currently registered leader)
         self.handle_sync_header(sched_ctx, comp_ctx, &message.sync_header);
         match message.content {
             SyncMessageContent::NotificationOfLeader => {
                 return SyncRoundDecision::None;
             },
             SyncMessageContent::LocalSolution(solution_generator_id, local_solution) => {
                 return self.handle_local_solution(sched_ctx, comp_ctx, solution_generator_id, local_solution);
             },
             SyncMessageContent::PartialSolution(partial_solution) => {
                 return self.handle_partial_solution(sched_ctx, comp_ctx, partial_solution);
             },
             SyncMessageContent::GlobalSolution => {
                 debug_assert_eq!(self.mode, Mode::SyncAwaitingSolution); // leader can only find global- if we submitted local solution
                 return SyncRoundDecision::Solution;
             },
             SyncMessageContent::GlobalFailure => {
                 debug_assert_eq!(self.mode, Mode::SyncAwaitingSolution);
                 return SyncRoundDecision::Failure;
+            }
+        }
+    }
     fn handle_sync_header(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, header: &MessageSyncHeader) {
         if header.highest_id.0 > self.highest_id.0 {
             // Sender knows of someone with a higher ID. So store highest ID,
             // notify all peers, and forward local solutions
             self.highest_id = header.highest_id;
             for peer in comp_ctx.iter_peers() {
                 if peer.id == header.sending_id {
                     continue; // do not send to sender: it has the higher ID
+                }
                 // also: only send if we received a message in this round
                 let mut performed_communication = false; // TODO: Revise, temporary fix
                 for port in self.ports.iter() {
                     if port.peer_comp_id == peer.id && port.mapping.is_some() {
                         performed_communication = true;
                         break;
+                    }
+                }
                 if !performed_communication {
                     continue;
+                }
                 let message = SyncMessage{
                     sync_header: self.create_sync_header(comp_ctx),
                     content: SyncMessageContent::NotificationOfLeader,
                 };
                 peer.handle.send_message(&sched_ctx.runtime, Message::Sync(message), true);
+            }
             self.forward_partial_solution(sched_ctx, comp_ctx);
         } else if header.highest_id.0 < self.highest_id.0 {
             // Sender has a lower ID, so notify it of our higher one
             let message = SyncMessage{
                 sync_header: self.create_sync_header(comp_ctx),
                 content: SyncMessageContent::NotificationOfLeader,
             };
             let peer_handle = comp_ctx.get_peer_handle(header.sending_id);
             let peer_info = comp_ctx.get_peer(peer_handle);
             peer_info.handle.send_message(&sched_ctx.runtime, Message::Sync(message), true);
         } // else: exactly equal
+    }
     fn set_annotation(&mut self, source_comp_id: CompId, data_header: &MessageDataHeader) {
         for annotation in self.ports.iter_mut() {
             if annotation.self_port_id == data_header.target_port {
                 // Message should have already passed the `handle_new_data_message` function, so we
                 // should have already annotated the peer of the port.
                 debug_assert!(
                     annotation.peer_discovered &&
                     annotation.peer_comp_id == source_comp_id &&
                     annotation.peer_port_id == data_header.source_port
                 );
                 annotation.mapping = Some(data_header.new_mapping);
+            }
+        }
+    }
     // -------------------------------------------------------------------------
     // Leader-related methods
     // -------------------------------------------------------------------------
     fn forward_partial_solution(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx) {
         debug_assert_ne!(self.highest_id, comp_ctx.id); // not leader
         // Make sure that we have something to send
         if !self.solution.has_contributions() {
             return;
+        }
         // Swap the container with the partial solution and then send it along
         let partial_solution = self.solution.take_partial_solution();
         self.send_to_leader(sched_ctx, comp_ctx, Message::Sync(SyncMessage{
             sync_header: self.create_sync_header(comp_ctx),
             content: SyncMessageContent::PartialSolution(partial_solution),
         }));
+    }
     fn handle_local_solution(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &CompCtx, solution_sender_id: CompId, solution: SyncLocalSolution) -> SyncRoundDecision {
         if self.highest_id == comp_ctx.id {
             // We are the leader
             self.solution.combine_with_local_solution(solution_sender_id, solution);
             let round_decision = self.solution.get_decision();
             if round_decision != SyncRoundDecision::None {
                 self.broadcast_decision(sched_ctx, comp_ctx, round_decision);
+            }
             return round_decision;
         } else {
             // Forward the solution
             let message = SyncMessage{
                 sync_header: self.create_sync_header(comp_ctx),
                 content: SyncMessageContent::LocalSolution(solution_sender_id, solution),
             };
             self.send_to_leader(sched_ctx, comp_ctx, Message::Sync(message));
             return SyncRoundDecision::None;
+        }
+    }
     fn handle_partial_solution(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, solution: SyncPartialSolution) -> SyncRoundDecision {
         if self.highest_id == comp_ctx.id {
             // We are the leader, combine existing and new solution
             self.solution.combine_with_partial_solution(solution);
             let round_decision = self.solution.get_decision();
             if round_decision != SyncRoundDecision::None {
                 self.broadcast_decision(sched_ctx, comp_ctx, round_decision);
+            }
             return round_decision;
         } else {
             // Forward the partial solution
             let message = SyncMessage{
                 sync_header: self.create_sync_header(comp_ctx),
                 content: SyncMessageContent::PartialSolution(solution),
             };
             self.send_to_leader(sched_ctx, comp_ctx, Message::Sync(message));
             return SyncRoundDecision::None;
+        }
+    }
     fn broadcast_decision(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &CompCtx, decision: SyncRoundDecision) {
         debug_assert_eq!(self.highest_id, comp_ctx.id);
         let is_success = match decision {
             SyncRoundDecision::None => unreachable!(),
             SyncRoundDecision::Solution => true,
             SyncRoundDecision::Failure => false,
         };
         let mut peers = Vec::with_capacity(self.solution.solution.channel_mapping.len()); // TODO: @Performance
         for channel in self.solution.solution.channel_mapping.iter() {
             let getter = channel.getter.as_ref().unwrap();
             if getter.self_comp_id != comp_ctx.id && !peers.contains(&getter.self_comp_id) {
                 peers.push(getter.self_comp_id);
+            }
             if getter.peer_comp_id != comp_ctx.id && !peers.contains(&getter.peer_comp_id) {
                 peers.push(getter.peer_comp_id);
+            }
+        }
         for peer in peers {
             let mut handle = sched_ctx.runtime.get_component_public(peer);
             let message = Message::Sync(SyncMessage{
                 sync_header: self.create_sync_header(comp_ctx),
                 content: if is_success { SyncMessageContent::GlobalSolution } else { SyncMessageContent::GlobalFailure },
             });
             handle.send_message(&sched_ctx.runtime, message, true);
             let _should_remove = handle.decrement_users();
             debug_assert!(_should_remove.is_none());
+        }
+    }
     fn send_to_leader(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &CompCtx, message: Message) {
         debug_assert_ne!(self.highest_id, comp_ctx.id); // we're not the leader
+        debug_assert_ne!(self.highest_id, comp_ctx.id); // we're not the leader, // TODO: @NoDirectHandle
         let mut leader_info = sched_ctx.runtime.get_component_public(self.highest_id);
         leader_info.send_message(&sched_ctx.runtime, message, true);
         let should_remove = leader_info.decrement_users();
         if let Some(key) = should_remove {
             sched_ctx.runtime.destroy_component(key);
+        }
+    }
     // -------------------------------------------------------------------------
     // Creating message headers
     // -------------------------------------------------------------------------
     fn create_data_header_and_update_mapping(&mut self, port_info: &Port) -> MessageDataHeader {
         let mut expected_mapping = Vec::with_capacity(self.ports.len());
         let mut port_index = usize::MAX;
         for (index, port) in self.ports.iter().enumerate() {
             if port.self_port_id == port_info.self_id {
                 port_index = index; // remember for later updating
+            }
             // Add all of the
             let annotation_kind = match port.kind {
                 PortKind::Putter => {
                     PortAnnotationKind::Putter(PortAnnotationPutter{
                         self_comp_id: port.self_comp_id,
                         self_port_id: port.self_port_id
                     })
                 },
                 PortKind::Getter => {
                     if !port.peer_discovered {
                         continue;
+                    }
                     PortAnnotationKind::Getter(PortAnnotationGetter{
                         self_comp_id: port.self_comp_id,
                         self_port_id: port.self_port_id,
                         peer_comp_id: port.peer_comp_id,
                         peer_port_id: port.peer_port_id,
                     })
+                }
             };
             expected_mapping.push((annotation_kind, port.mapping));
+        }
         let new_mapping = self.take_mapping();
         self.ports[port_index].mapping = Some(new_mapping);
         debug_assert_eq!(port_info.kind, PortKind::Putter);
         return MessageDataHeader{
             expected_mapping,
             new_mapping,
             source_port: port_info.self_id,
             target_port: port_info.peer_port_id,
         };
+    }
     #[inline]
     fn create_sync_header(&self, comp_ctx: &CompCtx) -> MessageSyncHeader {
         return MessageSyncHeader{
             sync_round: self.round_index,
             sending_id: comp_ctx.id,
             highest_id: self.highest_id,
         };
+    }
     #[inline]
     fn take_mapping(&mut self) -> u32 {
         let mapping = self.mapping_counter;
         self.mapping_counter = self.mapping_counter.wrapping_add(1);
         return mapping;
+    }
+}
@@ \ No newline at end of file @@

0 comments (0 inline, 0 general)