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

Changeset - 1e0c33498fac

Parent rev.

Child rev.

[Not reviewed]

0 2 0

MH - 3 years ago 2022-04-22 13:02:36
contact@maxhenger.nl

Test and fix pre-sync ClosePort case

2 files changed with 20 insertions and 9 deletions:

src/runtime2/component/component.rs

src/runtime2/component/consensus.rs

0 comments (0 inline, 0 general)

src/runtime2/component/component.rs

➞

Show inline comments

 use std::fmt::{Display as FmtDisplay, Result as FmtResult, Formatter};
 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,
+}
 /// Potential error emitted by a component
 pub enum CompError {
     /// Error originating from the code executor. Hence has an associated
     /// source location.
     Executor(EvalError),
     /// Error originating from a component, but not necessarily associated with
     /// a location in the source.
     Component(String), // TODO: Maybe a different embedded value in the future?
     /// Pure runtime error. Not necessarily originating from the component
     /// itself. Should be treated as a very severe runtime-compromising error.
     Runtime(RtError),
+}
 impl FmtDisplay for CompError {
     fn fmt(&self, f: &mut Formatter<'_>) -> FmtResult {
         match self {
             CompError::Executor(v) => v.fmt(f),
             CompError::Component(v) => v.fmt(f),
             CompError::Runtime(v) => v.fmt(f),
+        }
+    }
+}
 /// 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) -> CompScheduling;
+}
 /// Representation of the generic operating mode of a component. Although not
 /// every state may be used by every kind of (builtin) component, this allows
 /// writing standard handlers for particular events in a component's lifetime.
 #[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, potentially waiting for sync round to finish
     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,
+        }
+    }
     pub(crate) fn is_busy_exiting(&self) -> bool {
         use CompMode::*;
         match self {
             NonSync | Sync | SyncEnd | BlockedGet | BlockedPut | BlockedSelect => false,
             StartExit | BusyExit => true,
             Exit => false,
+        }
+    }
+}
 #[derive(Debug)]
 pub(crate) enum ExitReason {
     Termination, // regular termination of component
     ErrorInSync,
     ErrorNonSync,
+}
 impl ExitReason {
     pub(crate) fn is_in_sync(&self) -> bool {
         use ExitReason::*;
         match self {
             Termination | ErrorNonSync => false,
             ErrorInSync => true,
+        }
+    }
     pub(crate) fn is_error(&self) -> bool {
         use ExitReason::*;
         match self {
             Termination => false,
             ErrorInSync | ErrorNonSync => true,
+        }
+    }
+}
 /// 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
     pub exit_reason: ExitReason, // valid if in StartExit/BusyExit/Exit mode
+}
 impl CompExecState {
     pub(crate) fn new() -> Self {
         return Self{
             mode: CompMode::NonSync,
             mode_port: PortId::new_invalid(),
             mode_value: ValueGroup::default(),
             exit_reason: ExitReason::Termination,
+        }
+    }
     pub(crate) fn set_as_start_exit(&mut self, reason: ExitReason) {
         self.mode = CompMode::StartExit;
         self.exit_reason = reason;
+    }
     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;
+    }
+}
 // TODO: Replace when implementing port sending. Should probably be incorporated
 //  into CompCtx (and rename CompCtx into CompComms)
 pub(crate) type InboxMain = Vec<Option<DataMessage>>;
 pub(crate) type InboxMainRef = [Option<DataMessage>];
 pub(crate) type InboxBackup = Vec<DataMessage>;
 /// 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,
     port_instruction: PortInstruction, value: ValueGroup,
     sched_ctx: &SchedulerCtx, consensus: &mut Consensus, comp_ctx: &mut CompCtx
 ) -> Result<CompScheduling, (PortInstruction, String)> {
     debug_assert_eq!(exec_state.mode, CompMode::Sync);
     let port_handle = comp_ctx.get_port_handle(transmitting_port_id);
     let port_info = comp_ctx.get_port_mut(port_handle);
     port_info.last_instruction = port_instruction;
     let port_info = comp_ctx.get_port(port_handle);
     debug_assert_eq!(port_info.kind, PortKind::Putter);
     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((
             port_info.last_instruction,
             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 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 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, inbox_main: &mut InboxMain,
     comp_ctx: &mut CompCtx, incoming_message: DataMessage,
     sched_ctx: &SchedulerCtx, control: &mut ControlLayer
 ) -> IncomingData {
     let port_handle = comp_ctx.get_port_handle(incoming_message.data_header.target_port);
     let port_index = comp_ctx.get_port_index(port_handle);
     sched_ctx.log("DEBUG: Setting received_message_for_sync");
     comp_ctx.get_port_mut(port_handle).received_message_for_sync = true;
     let port_value_slot = &mut inbox_main[port_index];
     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.
             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_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)
+    }
+}
 pub(crate) enum GetResult {
     Received(DataMessage),
     NoMessage,
     Error((PortInstruction, String)),
+}
 /// Default attempt at trying to receive from a port (i.e. through a `get`, or
 /// the equivalent operation for a builtin component). `target_port` is the port
 /// we're trying to receive from, and the `target_port_instruction` is the
 /// instruction we're attempting on this port.
 pub(crate) fn default_attempt_get(
     exec_state: &mut CompExecState, target_port: PortId, target_port_instruction: PortInstruction,
     inbox_main: &mut InboxMainRef, inbox_backup: &mut InboxBackup, sched_ctx: &SchedulerCtx,
     comp_ctx: &mut CompCtx, control: &mut ControlLayer, consensus: &mut Consensus
 ) -> GetResult {
     let port_handle = comp_ctx.get_port_handle(target_port);
     let port_index = comp_ctx.get_port_index(port_handle);
     if let Some(message) = &inbox_main[port_index] {
         if consensus.try_receive_data_message(sched_ctx, comp_ctx, message) {
             // We're allowed to receive this message
             let message = inbox_main[port_index].take().unwrap();
             debug_assert_eq!(target_port, message.data_header.target_port);
             // Note: we can still run into an unrecoverable error when actually
             // receiving this message
             match default_handle_received_data_message(
                 target_port, target_port_instruction, inbox_main, inbox_backup,
                 comp_ctx, sched_ctx, control,
             ) {
                 Ok(()) => return GetResult::Received(message),
                 Err(location_and_message) => return GetResult::Error(location_and_message)
+            }
         } else {
             // We're not allowed to receive this message. This means that the
             // receiver is attempting to receive something out of order with
             // respect to the sender.
             return GetResult::Error((target_port_instruction, String::from(
                 "Cannot get from this port, as this causes a deadlock. This happens if you `get` in a different order as another component `put`s"
             )));
+        }
     } else {
         // We don't have a message waiting for us.
         let port_info = comp_ctx.get_port(port_handle);
         let port_info = comp_ctx.get_port_mut(port_handle);
         port_info.last_instruction = target_port_instruction;
         let port_is_closed = port_info.state == PortState::Closed;
         if port_is_closed {
             let peer_id = port_info.peer_comp_id;
             return GetResult::Error((
                 target_port_instruction,
                 format!("Cannot get from this port, as the peer component (id:{}) shut down", peer_id.0)
             ));
+        }
         // No error ocurred, so enter the BlockedGet state
         exec_state.set_as_blocked_get(target_port);
         return GetResult::NoMessage;
+    }
+}
 /// 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, port_instruction: PortInstruction,
     inbox_main: &mut InboxMainRef, inbox_backup: &mut InboxBackup,
     comp_ctx: &mut CompCtx, sched_ctx: &SchedulerCtx, control: &mut ControlLayer
 ) -> Result<(), (PortInstruction, String)> {
     let port_handle = comp_ctx.get_port_handle(targeted_port);
     let port_index = comp_ctx.get_port_index(port_handle);
     let slot = &mut inbox_main[port_index];
     debug_assert!(slot.is_none()); // because we've just received from it
     // Modify last-known location where port instruction was retrieved
     let port_info = comp_ctx.get_port_mut(port_handle);
     port_info.last_instruction = port_instruction;
     if port_info.state == PortState::Closed {
         return Err((
             port_info.last_instruction,
             format!("Cannot 'get' because the channel is closed"))
         );
+    }
     // Check if there are any more messages in the backup buffer
     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 Ok(());
+        }
+    }
     // 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);
+    }
     return Ok(());
+}
 /// 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<(), (PortInstruction, 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(content) => {
             // Request to close the port. We immediately comply and remove
             // the component handle as well
             let port_handle = comp_ctx.get_port_handle(content.port_to_close);
             // We're closing the port, so we will always update the peer of the
             // port (in case of error messages)
             let port_info = comp_ctx.get_port_mut(port_handle);
             port_info.peer_comp_id = message.sender_comp_id;
             port_info.close_at_sync_end = true; // might be redundant (we might set it closed now)
             let port_info = comp_ctx.get_port(port_handle);
             let peer_comp_id = port_info.peer_comp_id;
             let peer_handle = comp_ctx.get_peer_handle(peer_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. So we don't care about the
                 // content of the `ClosePort` message.
                 default_handle_ack(control, control_id, sched_ctx, comp_ctx);
             } else {
                 // Respond to the message
                 let last_instruction = port_info.last_instruction;
                 let port_has_had_message = port_info.received_message_for_sync;
                 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
                 // Handle any possible error conditions (which boil down to: the
                 // port has been used, but the peer has died). If not in sync
                 // mode then we close the port immediately.
                 // Note that `port_was_used` does not mean that any messages
                 // were actually received. It might also mean that e.g. the
                 // component attempted a `get`, but there were no messages, so
                 // now it is in the `BlockedGet` state.
                 let port_was_used = last_instruction != PortInstruction::None;
                 sched_ctx.log(&format!("DEBUG: last_instruction = {:?}, mode = {:?}, was_used = {}, has_had_message = {}", last_instruction, exec_state.mode, port_was_used, port_has_had_message));
                 if exec_state.mode.is_in_sync_block() {
                     let closed_during_sync_round = content.closed_in_sync_round && port_was_used;
-                    // TODO: Finish this
+                    let closed_before_sync_round = !content.closed_in_sync_round && !port_has_had_message;
                     if closed_during_sync_round {
+                    if closed_during_sync_round || closed_before_sync_round {
                         return Err((
                             last_instruction,
                             format!("Peer component (id:{}) shut down, so previous communication cannot have succeeded", peer_comp_id.0)
                         ));
+                    }
                 } else {
                     comp_ctx.set_port_state(port_handle, PortState::Closed);
+                }
+            }
         },
         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 entering the synchronous block. Will ensure that the
 /// `Consensus` and the `ComponentCtx` are initialized properly.
 pub(crate) fn default_handle_sync_start(
     exec_state: &mut CompExecState, inbox_main: &mut InboxMainRef,
     sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, consensus: &mut Consensus
 ) {
     sched_ctx.log("Component starting sync mode");
     // If any messages are present for this sync round, set the appropriate flag
     // and notify the consensus handler of the present messages
     consensus.notify_sync_start(comp_ctx);
     for (port_index, message) in inbox_main.iter().enumerate() {
         if let Some(message) = message {
             consensus.handle_incoming_data_message(comp_ctx, message);
             let port_info = comp_ctx.get_port_by_index_mut(port_index);
             port_info.received_message_for_sync = true;
+        }
+    }
     // Modify execution state
     debug_assert_eq!(exec_state.mode, CompMode::NonSync);
     exec_state.mode = CompMode::Sync;
+}
 /// Handles a component that has reached the end of the sync block. This does
 /// not necessarily mean that the component will go into the `NonSync` mode, as
 /// it might have to wait for the leader to finish the round for everyone (see
 /// `default_handle_sync_decision`)
 pub(crate) fn default_handle_sync_end(
     exec_state: &mut CompExecState, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx,
     consensus: &mut Consensus
 ) {
     sched_ctx.log("Component ending sync mode (but possibly waiting for a solution)");
     debug_assert_eq!(exec_state.mode, CompMode::Sync);
     let decision = consensus.notify_sync_end_success(sched_ctx, comp_ctx);
     exec_state.mode = CompMode::SyncEnd;
     default_handle_sync_decision(sched_ctx, exec_state, comp_ctx, decision, consensus);
+}
 /// 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, consensus: &mut Consensus
 ) -> CompScheduling {
     debug_assert_eq!(exec_state.mode, CompMode::StartExit);
     sched_ctx.log("Component starting exit");
     exec_state.mode = CompMode::BusyExit;
     let exit_inside_sync = exec_state.exit_reason.is_in_sync();
     // If exiting while inside sync mode, report to the leader of the current
     // round that we've failed.
     if exit_inside_sync {
         let decision = consensus.notify_sync_end_failure(sched_ctx, comp_ctx);
         default_handle_sync_decision(sched_ctx, exec_state, comp_ctx, decision, consensus);
+    }
     // Iterating over ports 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 {
+        if port.state == PortState::Closed || port.close_at_sync_end {
             // 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, exit_inside_sync, 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`.
 ///
 /// Might be called in two cases:
 /// 1. The component is in regular execution mode, at the end of a sync round,
 ///     and is waiting for a solution to the round.
 /// 2. The component has encountered an error during a sync round and is
 ///     exiting, hence is waiting for a "Failure" message from the leader.
 pub(crate) fn default_handle_sync_decision(
     sched_ctx: &SchedulerCtx, exec_state: &mut CompExecState, comp_ctx: &mut CompCtx,
     decision: SyncRoundDecision, consensus: &mut Consensus
 ) -> Option<bool> {
     let success = match decision {
         SyncRoundDecision::None => return None,
         SyncRoundDecision::Solution => true,
         SyncRoundDecision::Failure => false,
     };
     debug_assert!(
         exec_state.mode == CompMode::SyncEnd || (
             exec_state.mode.is_busy_exiting() && exec_state.exit_reason.is_error()
         ) || (
             exec_state.mode.is_in_sync_block() && decision == SyncRoundDecision::Failure
+        )
     );
     sched_ctx.log(&format!("Handling decision {:?} (in mode: {:?})", decision, exec_state.mode));
     consensus.notify_sync_decision(decision);
     if success {
         // We cannot get a success message if the component has encountered an
         // error.
         for port_index in 0..comp_ctx.num_ports() {
             let port_info = comp_ctx.get_port_by_index_mut(port_index);
             if port_info.close_at_sync_end {
                 port_info.state = PortState::Closed;
+            }
+        }
         debug_assert_eq!(exec_state.mode, CompMode::SyncEnd);
         exec_state.mode = CompMode::NonSync;
         return Some(true);
     } else {
         // We may get failure both in all possible cases. But we should only
         // modify the execution state if we're not already in exit mode
         if !exec_state.mode.is_busy_exiting() {
             sched_ctx.error("failed synchronous round, initiating exit");
             exec_state.set_as_start_exit(ExitReason::ErrorNonSync);
+        }
         return Some(false);
+    }
+}
 /// Performs the default action of printing the provided error, and then putting
 /// the component in the state where it will shut down. Only to be used for
 /// builtin components: their error message construction is simpler (and more
 /// common) as they don't have any source code.
 pub(crate) fn default_handle_error_for_builtin(
     exec_state: &mut CompExecState, sched_ctx: &SchedulerCtx,
     location_and_message: (PortInstruction, String)
 ) {
     let (_location, message) = location_and_message;
     sched_ctx.error(&message);
     let exit_reason = if exec_state.mode.is_in_sync_block() {
         ExitReason::ErrorInSync
     } else {
         ExitReason::ErrorNonSync
     };
     exec_state.set_as_start_exit(exit_reason);
+}
 #[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/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.failed &&
             !getter.failed &&
             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_success(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &CompCtx) -> SyncRoundDecision {
         debug_assert_eq!(self.mode, Mode::SyncBusy);
         self.mode = Mode::SyncAwaitingSolution;
         let local_solution = self.generate_local_solution(comp_ctx, false);
         let decision = self.handle_local_solution(sched_ctx, comp_ctx, comp_ctx.id, local_solution);
+        let decision = self.handle_local_solution(sched_ctx, comp_ctx, comp_ctx.id, local_solution, false);
         return decision;
+    }
     /// Notifies the consensus management that the component has encountered a
     /// critical error during the synchronous round. Hence we should report that
     /// we've failed and wait until all the participants have been notified of
     /// the error.
     pub(crate) fn notify_sync_end_failure(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &CompCtx) -> SyncRoundDecision {
         debug_assert_eq!(self.mode, Mode::SyncBusy);
         self.mode = Mode::SyncAwaitingSolution;
         let local_solution = self.generate_local_solution(comp_ctx, true);
         let decision = self.handle_local_solution(sched_ctx, comp_ctx, comp_ctx.id, local_solution);
+        let decision = self.handle_local_solution(sched_ctx, comp_ctx, comp_ctx.id, local_solution, true);
         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.
     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);
+                return self.handle_local_solution(sched_ctx, comp_ctx, solution_generator_id, local_solution, false);
             },
             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 {
+    fn handle_local_solution(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &CompCtx, solution_sender_id: CompId, solution: SyncLocalSolution, fail_if_empty: bool) -> 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();
+            let mut round_decision = self.solution.get_decision();
             if round_decision != SyncRoundDecision::None {
                 if fail_if_empty && self.solution.matched_channels == 0 {
                     // TODO: Not sure about this, bit of a hack. Situation is that a component
                     //  cannot interact with other components, but it is in a sync round, and has
                     //  failed that sync round.
                     round_decision = SyncRoundDecision::Failure;
+                }
                 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, // 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);
+        }
+    }
     // -------------------------------------------------------------------------
     // Small utilities
     // -------------------------------------------------------------------------
     fn generate_local_solution(&self, comp_ctx: &CompCtx, failed: bool) -> SyncLocalSolution {
         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,
                         failed
                     }),
                     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,
                         failed
                     })
                 };
                 local_solution.push(new_entry);
+            }
+        }
         return local_solution;
+    }
     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)