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

Changeset - ac804a4a3d70

Parent rev.

Child rev.

[Not reviewed]

0 8 0

mh - 3 years ago 2022-04-26 10:54:14
contact@maxhenger.nl

More granularity in debug logging

8 files changed with 81 insertions and 47 deletions:

src/runtime2/component/component.rs

src/runtime2/component/component_internet.rs

src/runtime2/component/component_pdl.rs

src/runtime2/component/component_random.rs

src/runtime2/component/consensus.rs

src/runtime2/poll/mod.rs

src/runtime2/runtime.rs

src/runtime2/scheduler.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);
+        peer_info.handle.send_message_logged(sched_ctx, 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);
     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);
+            peer.handle.send_message_logged(sched_ctx, 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);
     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:{}) closed the port", peer_id.0)
         ));
+    }
     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 and the port is not blocked.
         // 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);
     debug_assert_ne!(port_info.last_instruction, PortInstruction::None); // set by caller
     debug_assert_ne!(port_info.state, PortState::Closed); // checked by caller
     // 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);
+        peer_info.handle.send_message_logged(sched_ctx, 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;
                 if exec_state.mode.is_in_sync_block() {
                     let closed_during_sync_round = content.closed_in_sync_round && port_was_used;
                     let closed_before_sync_round = !content.closed_in_sync_round && !port_has_had_message;
                     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");
+    sched_ctx.info("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)");
+    sched_ctx.info("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(&format!("Component starting exit (reason: {:?})", exec_state.exit_reason));
+    sched_ctx.info(&format!("Component starting exit (reason: {:?})", exec_state.exit_reason));
     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 || 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);
+        peer_info.handle.send_message_logged(sched_ctx, 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");
+        sched_ctx.info("Component busy exiting, still has `Ack`s remaining");
         return CompScheduling::Sleep;
     } else {
-        sched_ctx.log("Component busy exiting, now shutting down");
+        sched_ctx.info("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));
+    sched_ctx.info(&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);
+                handle.send_message_logged(sched_ctx, 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{
+    peer_info.handle.send_message_logged(sched_ctx, 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);
+        peer_info.handle.send_message_logged(sched_ctx, 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_internet.rs

➞

Show inline comments

 use crate::protocol::eval::{ValueGroup, Value};
 use crate::runtime2::*;
 use crate::runtime2::component::{CompCtx, CompId, PortInstruction};
 use crate::runtime2::stdlib::internet::*;
 use crate::runtime2::poll::*;
 use super::component::{self, *};
 use super::control_layer::*;
 use super::consensus::*;
 use std::io::ErrorKind as IoErrorKind;
 enum SocketState {
     Connected(SocketTcpClient),
     Error,
+}
 impl SocketState {
     fn get_socket(&self) -> &SocketTcpClient {
         match self {
             SocketState::Connected(v) => v,
             SocketState::Error => unreachable!(),
+        }
+    }
+}
 /// States from the point of view of the component that is connecting to this
 /// TCP component (i.e. from the point of view of attempting to interface with
 /// a socket).
 #[derive(PartialEq, Debug)]
 enum SyncState {
     AwaitingCmd,
     Getting,
     Putting,
     FinishSync,
     FinishSyncThenQuit,
+}
 pub struct ComponentTcpClient {
     // Properties for the tcp socket
     socket_state: SocketState,
     sync_state: SyncState,
     poll_ticket: Option<PollTicket>,
     inbox_main: InboxMain,
     inbox_backup: InboxBackup,
     pdl_input_port_id: PortId, // input from PDL, so transmitted over socket
     pdl_output_port_id: PortId, // output towards PDL, so received over socket
     input_union_send_tag_value: i64,
     input_union_receive_tag_value: i64,
     input_union_finish_tag_value: i64,
     input_union_shutdown_tag_value: i64,
     // Generic component state
     exec_state: CompExecState,
     control: ControlLayer,
     consensus: Consensus,
     // Temporary variables
     byte_buffer: Vec<u8>,
+}
 impl Component for ComponentTcpClient {
     fn on_creation(&mut self, id: CompId, sched_ctx: &SchedulerCtx) {
         // Retrieve type information for messages we're going to receive
         let pd = &sched_ctx.runtime.protocol;
         let cmd_type = pd.find_type(b"std.internet", b"Cmd")
             .expect("'Cmd' type in the 'std.internet' module");
         let cmd_type = cmd_type
             .as_union();
         self.input_union_send_tag_value = cmd_type.get_variant_tag_value(b"Send").unwrap();
         self.input_union_receive_tag_value = cmd_type.get_variant_tag_value(b"Receive").unwrap();
         self.input_union_finish_tag_value = cmd_type.get_variant_tag_value(b"Finish").unwrap();
         self.input_union_shutdown_tag_value = cmd_type.get_variant_tag_value(b"Shutdown").unwrap();
         // Register socket for async events
         if let SocketState::Connected(socket) = &self.socket_state {
             let self_handle = sched_ctx.runtime.get_component_public(id);
             let poll_ticket = sched_ctx.polling.register(socket, self_handle, true, true)
                 .expect("registering tcp component");
             debug_assert!(self.poll_ticket.is_none());
             self.poll_ticket = Some(poll_ticket);
+        }
+    }
     fn on_shutdown(&mut self, sched_ctx: &SchedulerCtx) {
         if let Some(poll_ticket) = self.poll_ticket.take() {
             sched_ctx.polling.unregister(poll_ticket)
                 .expect("unregistering tcp component");
+        }
+    }
     fn adopt_message(&mut self, _comp_ctx: &mut CompCtx, message: DataMessage) {
         let slot = &mut self.inbox_main[0];
         if slot.is_none() {
             *slot = Some(message);
         } else {
             self.inbox_backup.push(message);
+        }
+    }
     fn handle_message(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx, message: Message) {
         match message {
             Message::Data(message) => {
                 self.handle_incoming_data_message(sched_ctx, comp_ctx, message);
             },
             Message::Sync(message) => {
                 let decision = self.consensus.receive_sync_message(sched_ctx, comp_ctx, message);
                 component::default_handle_sync_decision(sched_ctx, &mut self.exec_state, comp_ctx, decision, &mut self.consensus);
             },
             Message::Control(message) => {
                 if let Err(location_and_message) = component::default_handle_control_message(
                     &mut self.exec_state, &mut self.control, &mut self.consensus,
                     message, sched_ctx, comp_ctx
                 ) {
                     component::default_handle_error_for_builtin(&mut self.exec_state, sched_ctx, location_and_message);
+                }
             },
             Message::Poll => {
-                sched_ctx.log("Received polling event");
+                sched_ctx.info("Received polling event");
             },
+        }
+    }
     fn run(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx) -> CompScheduling {
-        sched_ctx.log(&format!("Running component ComponentTcpClient (mode: {:?}, sync state: {:?})", self.exec_state.mode, self.sync_state));
+        sched_ctx.info(&format!("Running component ComponentTcpClient (mode: {:?}, sync state: {:?})", self.exec_state.mode, self.sync_state));
         match self.exec_state.mode {
             CompMode::BlockedSelect => {
                 // Not possible: we never enter this state
                 unreachable!();
             },
             CompMode::NonSync => {
                 // When in non-sync mode
                 match &mut self.socket_state {
                     SocketState::Connected(_socket) => {
                         if self.sync_state == SyncState::FinishSyncThenQuit {
                             // Previous request was to let the component shut down
                             self.exec_state.set_as_start_exit(ExitReason::Termination);
                         } else {
                             // Reset for a new request
                             self.sync_state = SyncState::AwaitingCmd;
                             component::default_handle_sync_start(
                                 &mut self.exec_state, &mut self.inbox_main, sched_ctx, comp_ctx, &mut self.consensus
                             );
+                        }
                         return CompScheduling::Immediate;
                     },
                     SocketState::Error => {
                         // Could potentially send an error message to the
                         // connected component.
                         self.exec_state.set_as_start_exit(ExitReason::ErrorNonSync);
                         return CompScheduling::Immediate;
+                    }
+                }
             },
             CompMode::Sync => {
                 // When in sync mode: wait for a command to come in
                 match self.sync_state {
                     SyncState::AwaitingCmd => {
                         match component::default_attempt_get(
                             &mut self.exec_state, self.pdl_input_port_id, PortInstruction::NoSource,
                             &mut self.inbox_main, &mut self.inbox_backup, sched_ctx, comp_ctx,
                             &mut self.control, &mut self.consensus
                         ) {
                             GetResult::Received(message) => {
                                 let (tag_value, embedded_heap_pos) = message.content.values[0].as_union();
                                 if tag_value == self.input_union_send_tag_value {
                                     // Retrieve bytes from the message
                                     self.byte_buffer.clear();
                                     let union_content = &message.content.regions[embedded_heap_pos as usize];
                                     debug_assert_eq!(union_content.len(), 1);
                                     let array_heap_pos = union_content[0].as_array();
                                     let array_values = &message.content.regions[array_heap_pos as usize];
                                     self.byte_buffer.reserve(array_values.len());
                                     for value in array_values {
                                         self.byte_buffer.push(value.as_uint8());
+                                    }
                                     self.sync_state = SyncState::Putting;
                                 } else if tag_value == self.input_union_receive_tag_value {
                                     // Component requires a `recv`
                                     self.sync_state = SyncState::Getting;
                                 } else if tag_value == self.input_union_finish_tag_value {
                                     // Component requires us to end the sync round
                                     self.sync_state = SyncState::FinishSync;
                                 } else if tag_value == self.input_union_shutdown_tag_value {
                                     // Component wants to close the connection
                                     self.sync_state = SyncState::FinishSyncThenQuit;
                                 } else {
                                     unreachable!("got tag_value {}", tag_value)
+                                }
                                 return CompScheduling::Immediate;
                             },
                             GetResult::NoMessage => {
                                 return CompScheduling::Sleep;
                             },
                             GetResult::Error(location_and_message) => {
                                 component::default_handle_error_for_builtin(&mut self.exec_state, sched_ctx, location_and_message);
                                 return CompScheduling::Immediate;
+                            }
+                        }
                     },
                     SyncState::Putting => {
                         // We're supposed to send a user-supplied message fully
                         // over the socket. But we might end up blocking. In
                         // that case the component goes to sleep until it is
                         // polled.
                         let socket = self.socket_state.get_socket();
                         while !self.byte_buffer.is_empty() {
                             match socket.send(&self.byte_buffer) {
                                 Ok(bytes_sent) => {
                                     self.byte_buffer.drain(..bytes_sent);
                                 },
                                 Err(err) => {
                                     if err.kind() == IoErrorKind::WouldBlock {
                                         return CompScheduling::Sleep; // wait until notified
                                     } else {
                                         todo!("handle socket.send error {:?}", err)
+                                    }
+                                }
+                            }
+                        }
                         // If here then we're done putting the data, we can
                         // finish the sync round
                         component::default_handle_sync_end(&mut self.exec_state, sched_ctx, comp_ctx, &mut self.consensus);
                         return CompScheduling::Requeue;
                     },
                     SyncState::Getting => {
                         // We're going to try and receive a single message. If
                         // this causes us to end up blocking the component
                         // goes to sleep until it is polled.
                         const BUFFER_SIZE: usize = 1024; // TODO: Move to config
                         let socket = self.socket_state.get_socket();
                         self.byte_buffer.resize(BUFFER_SIZE, 0);
                         match socket.receive(&mut self.byte_buffer) {
                             Ok(num_received) => {
                                 self.byte_buffer.resize(num_received, 0);
                                 let message_content = self.bytes_to_data_message_content(&self.byte_buffer);
                                 let send_result = component::default_send_data_message(&mut self.exec_state, self.pdl_output_port_id, PortInstruction::NoSource, message_content, sched_ctx, &mut self.consensus, comp_ctx);
                                 if let Err(location_and_message) = send_result {
                                     component::default_handle_error_for_builtin(&mut self.exec_state, sched_ctx, location_and_message);
                                     return CompScheduling::Immediate;
                                 } else {
                                     let scheduling = send_result.unwrap();
                                     self.sync_state = SyncState::AwaitingCmd;
                                     return scheduling;
+                                }
                             },
                             Err(err) => {
                                 if err.kind() == IoErrorKind::WouldBlock {
                                     return CompScheduling::Sleep; // wait until polled
                                 } else {
                                     todo!("handle socket.receive error {:?}", err)
+                                }
+                            }
+                        }
                     },
                     SyncState::FinishSync | SyncState::FinishSyncThenQuit => {
                         component::default_handle_sync_end(&mut self.exec_state, sched_ctx, comp_ctx, &mut self.consensus);
                         return CompScheduling::Requeue;
                     },
+                }
             },
             CompMode::BlockedGet => {
                 // Entered when awaiting a new command
                 debug_assert_eq!(self.sync_state, SyncState::AwaitingCmd);
                 return CompScheduling::Sleep;
             },
             CompMode::SyncEnd | CompMode::BlockedPut =>
                 return CompScheduling::Sleep,
             CompMode::StartExit =>
                 return component::default_handle_start_exit(&mut self.exec_state, &mut self.control, sched_ctx, comp_ctx, &mut self.consensus),
             CompMode::BusyExit =>
                 return component::default_handle_busy_exit(&mut self.exec_state, &mut self.control, sched_ctx),
             CompMode::Exit =>
                 return component::default_handle_exit(&self.exec_state),
+        }
+    }
+}
 impl ComponentTcpClient {
     pub(crate) fn new(arguments: ValueGroup) -> Self {
         use std::net::{IpAddr, Ipv4Addr};
         debug_assert_eq!(arguments.values.len(), 4);
         // Parsing arguments
         let ip_heap_pos = arguments.values[0].as_array();
         let ip_elements = &arguments.regions[ip_heap_pos as usize];
         if ip_elements.len() != 4 {
             todo!("friendly error reporting: ip contains 4 octects");
+        }
         let ip_address = IpAddr::V4(Ipv4Addr::new(
             ip_elements[0].as_uint8(), ip_elements[1].as_uint8(),
             ip_elements[2].as_uint8(), ip_elements[3].as_uint8()
         ));
         let port = arguments.values[1].as_uint16();
         let input_port = component::port_id_from_eval(arguments.values[2].as_input());
         let output_port = component::port_id_from_eval(arguments.values[3].as_output());
         let socket = SocketTcpClient::new(ip_address, port);
         if let Err(socket) = socket {
             todo!("friendly error reporting: failed to open socket (reason: {:?})", socket);
+        }
         return Self{
             socket_state: SocketState::Connected(socket.unwrap()),
             sync_state: SyncState::AwaitingCmd,
             poll_ticket: None,
             inbox_main: vec![None],
             inbox_backup: Vec::new(),
             input_union_send_tag_value: -1,
             input_union_receive_tag_value: -1,
             input_union_finish_tag_value: -1,
             input_union_shutdown_tag_value: -1,
             pdl_input_port_id: input_port,
             pdl_output_port_id: output_port,
             exec_state: CompExecState::new(),
             control: ControlLayer::default(),
             consensus: Consensus::new(),
             byte_buffer: Vec::new(),
+        }
+    }
     // Handles incoming data from the PDL side (hence, going into the socket)
     fn handle_incoming_data_message(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, message: DataMessage) {
         if self.exec_state.mode.is_in_sync_block() {
             self.consensus.handle_incoming_data_message(comp_ctx, &message);
+        }
         match component::default_handle_incoming_data_message(
             &mut self.exec_state, &mut self.inbox_main, comp_ctx, message, sched_ctx, &mut self.control
         ) {
             IncomingData::PlacedInSlot => {},
             IncomingData::SlotFull(message) => {
                 self.inbox_backup.push(message);
+            }
+        }
+    }
     fn data_message_to_bytes(&self, message: DataMessage, bytes: &mut Vec<u8>) {
         debug_assert_eq!(message.data_header.target_port, self.pdl_input_port_id);
         debug_assert_eq!(message.content.values.len(), 1);
         if let Value::Array(array_pos) = message.content.values[0] {
             let region = &message.content.regions[array_pos as usize];
             bytes.reserve(region.len());
             for value in region {
                 bytes.push(value.as_uint8());
+            }
         } else {
             unreachable!();
+        }
+    }
     fn bytes_to_data_message_content(&self, buffer: &[u8]) -> ValueGroup {
         // Turn bytes into silly executor-style array
         let mut values = Vec::with_capacity(buffer.len());
         for byte in buffer.iter().copied() {
             values.push(Value::UInt8(byte));
+        }
         // Put in a value group
         let mut value_group = ValueGroup::default();
         value_group.regions.push(values);
         value_group.values.push(Value::Array(0));
         return value_group;
+    }
+}
@@ \ No newline at end of file @@

src/runtime2/component/component_pdl.rs

➞

Show inline comments

 use crate::random::Random;
 use crate::protocol::*;
 use crate::protocol::ast::ProcedureDefinitionId;
 use crate::protocol::eval::{
     PortId as EvalPortId, Prompt,
     ValueGroup, Value,
     EvalContinuation, EvalResult, EvalError
 };
 use crate::runtime2::runtime::CompId;
 use crate::runtime2::scheduler::SchedulerCtx;
 use crate::runtime2::communication::*;
 use super::component::{
     self,
     InboxMain, InboxBackup, GetResult,
     CompExecState, Component, CompScheduling, CompError, CompMode, ExitReason,
     port_id_from_eval, port_id_to_eval
 };
 use super::component_context::*;
 use super::control_layer::*;
 use super::consensus::Consensus;
 pub enum ExecStmt {
     CreatedChannel((Value, Value)),
     PerformedPut,
     PerformedGet(ValueGroup),
     PerformedSelectWait(u32),
     None,
+}
 impl ExecStmt {
     fn take(&mut self) -> ExecStmt {
         let mut value = ExecStmt::None;
         std::mem::swap(self, &mut value);
         return value;
+    }
     fn is_none(&self) -> bool {
         match self {
             ExecStmt::None => return true,
             _ => return false,
+        }
+    }
+}
 pub struct ExecCtx {
     stmt: ExecStmt,
+}
 impl RunContext for ExecCtx {
     fn performed_put(&mut self, _port: EvalPortId) -> bool {
         match self.stmt.take() {
             ExecStmt::None => return false,
             ExecStmt::PerformedPut => return true,
             _ => unreachable!(),
+        }
+    }
     fn performed_get(&mut self, _port: EvalPortId) -> Option<ValueGroup> {
         match self.stmt.take() {
             ExecStmt::None => return None,
             ExecStmt::PerformedGet(value) => return Some(value),
             _ => unreachable!(),
+        }
+    }
     fn fires(&mut self, _port: EvalPortId) -> Option<Value> {
         todo!("remove fires")
+    }
     fn performed_fork(&mut self) -> Option<bool> {
         todo!("remove fork")
+    }
     fn created_channel(&mut self) -> Option<(Value, Value)> {
         match self.stmt.take() {
             ExecStmt::None => return None,
             ExecStmt::CreatedChannel(ports) => return Some(ports),
             _ => unreachable!(),
+        }
+    }
     fn performed_select_wait(&mut self) -> Option<u32> {
         match self.stmt.take() {
             ExecStmt::None => return None,
             ExecStmt::PerformedSelectWait(selected_case) => Some(selected_case),
             _v => unreachable!(),
+        }
+    }
+}
 struct SelectCase {
     involved_ports: Vec<LocalPortHandle>,
+}
 // TODO: @Optimize, flatten cases into single array, have index-pointers to next case
 struct SelectState {
     cases: Vec<SelectCase>,
     next_case: u32,
     num_cases: u32,
     random: Random,
     candidates_workspace: Vec<usize>,
+}
 enum SelectDecision {
     None,
     Case(u32), // contains case index, should be passed along to PDL code
+}
 impl SelectState {
     fn new() -> Self {
         return Self{
             cases: Vec::new(),
             next_case: 0,
             num_cases: 0,
             random: Random::new(),
             candidates_workspace: Vec::new(),
+        }
+    }
     fn handle_select_start(&mut self, num_cases: u32) {
         self.cases.clear();
         self.next_case = 0;
         self.num_cases = num_cases;
+    }
     /// Register a port as belonging to a particular case. As for correctness of
     /// PDL code one cannot register the same port twice, this function might
     /// return an error
     fn register_select_case_port(&mut self, comp_ctx: &CompCtx, case_index: u32, _port_index: u32, port_id: PortId) -> Result<(), PortId> {
         // Retrieve case and port handle
         self.ensure_at_case(case_index);
         let cur_case = &mut self.cases[case_index as usize];
         let port_handle = comp_ctx.get_port_handle(port_id);
         debug_assert_eq!(cur_case.involved_ports.len(), _port_index as usize);
         // Make sure port wasn't added before, we disallow having the same port
         // in the same select guard twice.
         if cur_case.involved_ports.contains(&port_handle) {
             return Err(port_id);
+        }
         cur_case.involved_ports.push(port_handle);
         return Ok(());
+    }
     /// Notification that all ports have been registered and we should now wait
     /// until the appropriate messages have come in.
     fn handle_select_waiting_point(&mut self, inbox: &InboxMain, comp_ctx: &CompCtx) -> SelectDecision {
         if self.num_cases != self.next_case {
             // This happens when there are >=1 select cases written at the end
             // of the select block.
             self.ensure_at_case(self.num_cases - 1);
+        }
         return self.has_decision(inbox, comp_ctx);
+    }
     fn handle_updated_inbox(&mut self, inbox: &InboxMain, comp_ctx: &CompCtx) -> SelectDecision {
         return self.has_decision(inbox, comp_ctx);
+    }
     /// Internal helper, pushes empty cases inbetween last case and provided new
     /// case index.
     fn ensure_at_case(&mut self, new_case_index: u32) {
         // Push an empty case for all intermediate cases that were not
         // registered with a port.
         debug_assert!(new_case_index >= self.next_case && new_case_index < self.num_cases);
         for _ in self.next_case..new_case_index + 1 {
             self.cases.push(SelectCase{ involved_ports: Vec::new() });
+        }
         self.next_case = new_case_index + 1;
+    }
     /// Checks if a decision can be reached
     fn has_decision(&mut self, inbox: &InboxMain, comp_ctx: &CompCtx) -> SelectDecision {
         self.candidates_workspace.clear();
         if self.cases.is_empty() {
             // If there are no cases then we can immediately reach a "bogus
             // decision".
             return SelectDecision::Case(0);
+        }
         // Need to check for valid case
         'case_loop: for (case_index, case) in self.cases.iter().enumerate() {
             for port_handle in case.involved_ports.iter().copied() {
                 let port_index = comp_ctx.get_port_index(port_handle);
                 if inbox[port_index].is_none() {
                     // Condition not satisfied
                     continue 'case_loop;
+                }
+            }
             // If here then the case guard is satisfied
             self.candidates_workspace.push(case_index);
+        }
         if self.candidates_workspace.is_empty() {
             return SelectDecision::None;
         } else {
             let candidate_index = self.random.get_u64() as usize % self.candidates_workspace.len();
             return SelectDecision::Case(self.candidates_workspace[candidate_index] as u32);
+        }
+    }
+}
 pub(crate) struct CompPDL {
     pub exec_state: CompExecState,
     select_state: SelectState,
     pub prompt: Prompt,
     pub control: ControlLayer,
     pub consensus: Consensus,
     pub sync_counter: u32,
     pub exec_ctx: ExecCtx,
     // TODO: Temporary field, simulates future plans of having one storage place
     //  reserved per port.
     // Should be same length as the number of ports. Corresponding indices imply
     // message is intended for that port.
     pub inbox_main: InboxMain,
     pub inbox_backup: Vec<DataMessage>,
+}
 impl Component for CompPDL {
     fn on_creation(&mut self, _id: CompId, _sched_ctx: &SchedulerCtx) {
         // Intentionally empty
+    }
     fn on_shutdown(&mut self, _sched_ctx: &SchedulerCtx) {
         // Intentionally empty
+    }
     fn adopt_message(&mut self, comp_ctx: &mut CompCtx, message: DataMessage) {
         let port_handle = comp_ctx.get_port_handle(message.data_header.target_port);
         let port_index = comp_ctx.get_port_index(port_handle);
         if self.inbox_main[port_index].is_none() {
             self.inbox_main[port_index] = Some(message);
         } else {
             self.inbox_backup.push(message);
+        }
+    }
     fn handle_message(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx, mut message: Message) {
         // sched_ctx.log(&format!("handling message: {:?}", message));
         if let Some(new_target) = self.control.should_reroute(&mut message) {
             let mut target = sched_ctx.runtime.get_component_public(new_target); // TODO: @NoDirectHandle
-            target.send_message(&sched_ctx.runtime, message, false); // not waking up: we schedule once we've received all PortPeerChanged Acks
+            target.send_message_logged(sched_ctx, message, false); // not waking up: we schedule once we've received all PortPeerChanged Acks
             let _should_remove = target.decrement_users();
             debug_assert!(_should_remove.is_none());
             return;
+        }
         match message {
             Message::Data(message) => {
                 self.handle_incoming_data_message(sched_ctx, comp_ctx, message);
             },
             Message::Control(message) => {
                 if let Err(location_and_message) = component::default_handle_control_message(
                     &mut self.exec_state, &mut self.control, &mut self.consensus,
                     message, sched_ctx, comp_ctx
                 ) {
                     self.handle_generic_component_error(sched_ctx, location_and_message);
+                }
             },
             Message::Sync(message) => {
                 self.handle_incoming_sync_message(sched_ctx, comp_ctx, message);
             },
             Message::Poll => {
                 unreachable!(); // because we never register at the polling thread
+            }
+        }
+    }
     fn run(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx) -> CompScheduling {
         use EvalContinuation as EC;
-        sched_ctx.log(&format!("Running component (mode: {:?})", self.exec_state.mode));
+        sched_ctx.info(&format!("Running component (mode: {:?})", self.exec_state.mode));
         // Depending on the mode don't do anything at all, take some special
         // actions, or fall through and run the PDL code.
         match self.exec_state.mode {
             CompMode::NonSync | CompMode::Sync => {
                 // continue and run PDL code
             },
             CompMode::SyncEnd | CompMode::BlockedGet | CompMode::BlockedPut | CompMode::BlockedSelect => {
                 return CompScheduling::Sleep;
+            }
             CompMode::StartExit => return component::default_handle_start_exit(
                 &mut self.exec_state, &mut self.control, sched_ctx, comp_ctx, &mut self.consensus
             ),
             CompMode::BusyExit => return component::default_handle_busy_exit(
                 &mut self.exec_state, &self.control, sched_ctx
             ),
             CompMode::Exit => return component::default_handle_exit(&self.exec_state),
+        }
         let run_result = self.execute_prompt(&sched_ctx);
         if let Err(error) = run_result {
             self.handle_component_error(sched_ctx, CompError::Executor(error));
             return CompScheduling::Immediate;
+        }
         let run_result = run_result.unwrap();
         match run_result {
             EC::Stepping => unreachable!(), // execute_prompt runs until this is no longer returned
             EC::BranchInconsistent | EC::NewFork | EC::BlockFires(_) => todo!("remove these"),
             // Results that can be returned in sync mode
             EC::SyncBlockEnd => {
                 component::default_handle_sync_end(&mut self.exec_state, sched_ctx, comp_ctx, &mut self.consensus);
                 return CompScheduling::Immediate;
             },
             EC::BlockGet(expr_id, port_id) => {
                 debug_assert_eq!(self.exec_state.mode, CompMode::Sync);
                 debug_assert!(self.exec_ctx.stmt.is_none());
                 let port_id = port_id_from_eval(port_id);
                 match component::default_attempt_get(
                     &mut self.exec_state, port_id, PortInstruction::SourceLocation(expr_id),
                     &mut self.inbox_main, &mut self.inbox_backup, sched_ctx, comp_ctx,
                     &mut self.control, &mut self.consensus
                 ) {
                     GetResult::Received(message) => {
                         self.exec_ctx.stmt = ExecStmt::PerformedGet(message.content);
                         return CompScheduling::Immediate;
                     },
                     GetResult::NoMessage => {
                         return CompScheduling::Sleep;
                     },
                     GetResult::Error(location_and_message) => {
                         self.handle_generic_component_error(sched_ctx, location_and_message);
                         return CompScheduling::Immediate;
+                    }
+                }
             },
             EC::Put(expr_id, port_id, value) => {
                 debug_assert_eq!(self.exec_state.mode, CompMode::Sync);
-                sched_ctx.log(&format!("Putting value {:?}", value));
+                sched_ctx.info(&format!("Putting value {:?}", value));
                 // Send the message
                 let target_port_id = port_id_from_eval(port_id);
                 let send_result = component::default_send_data_message(
                     &mut self.exec_state, target_port_id,
                     PortInstruction::SourceLocation(expr_id), value,
                     sched_ctx, &mut self.consensus, comp_ctx
                 );
                 if let Err(location_and_message) = send_result {
                     self.handle_generic_component_error(sched_ctx, location_and_message);
                     return CompScheduling::Immediate;
                 } else {
                     // When `run` is called again (potentially after becoming
                     // unblocked) we need to instruct the executor that we performed
                     // the `put`
                     let scheduling = send_result.unwrap();
                     self.exec_ctx.stmt = ExecStmt::PerformedPut;
                     return scheduling;
+                }
             },
             EC::SelectStart(num_cases, _num_ports) => {
                 debug_assert_eq!(self.exec_state.mode, CompMode::Sync);
                 self.select_state.handle_select_start(num_cases);
                 return CompScheduling::Requeue;
             },
             EC::SelectRegisterPort(expr_id, case_index, port_index, port_id) => {
                 debug_assert_eq!(self.exec_state.mode, CompMode::Sync);
                 let port_id = port_id_from_eval(port_id);
                 let port_handle = comp_ctx.get_port_handle(port_id);
                 // Note: we register the "last_instruction" here already. This
                 // way if we get a `ClosePort` message, the condition to fail
                 // the synchronous round is satisfied.
                 let port_info = comp_ctx.get_port_mut(port_handle);
                 port_info.last_instruction = PortInstruction::SourceLocation(expr_id);
                 let port_is_closed = port_info.state == PortState::Closed;
                 // Register port as part of select guard
                 if let Err(_err) = self.select_state.register_select_case_port(comp_ctx, case_index, port_index, port_id) {
                     // Failure occurs if a port is used twice in the same guard
                     let protocol = &sched_ctx.runtime.protocol;
                     self.handle_component_error(sched_ctx, CompError::Executor(EvalError::new_error_at_expr(
                         &self.prompt, &protocol.modules, &protocol.heap, expr_id,
                         String::from("Cannot have the one port appear in the same guard twice")
                     )));
                 } else if port_is_closed {
                     // Port is closed
                     let peer_id = comp_ctx.get_port(port_handle).peer_comp_id;
                     let protocol = &sched_ctx.runtime.protocol;
                     self.handle_component_error(sched_ctx, CompError::Executor(EvalError::new_error_at_expr(
                         &self.prompt, &protocol.modules, &protocol.heap, expr_id,
                         format!("Cannot register port, as the peer component (id:{}) has shut down", peer_id.0)
                     )));
+                }
                 return CompScheduling::Immediate;
             },
             EC::SelectWait => {
                 debug_assert_eq!(self.exec_state.mode, CompMode::Sync);
                 let select_decision = self.select_state.handle_select_waiting_point(&self.inbox_main, comp_ctx);
                 if let SelectDecision::Case(case_index) = select_decision {
                     // Reached a conclusion, so we can continue immediately
                     self.exec_ctx.stmt = ExecStmt::PerformedSelectWait(case_index);
                     self.exec_state.mode = CompMode::Sync;
                     return CompScheduling::Immediate;
                 } else {
                     // No decision yet
                     self.exec_state.mode = CompMode::BlockedSelect;
                     return CompScheduling::Sleep;
+                }
             },
             // Results that can be returned outside of sync mode
             EC::ComponentTerminated => {
                 self.exec_state.set_as_start_exit(ExitReason::Termination);
                 return CompScheduling::Immediate;
             },
             EC::SyncBlockStart => {
                 component::default_handle_sync_start(
                     &mut self.exec_state, &mut self.inbox_main, sched_ctx, comp_ctx, &mut self.consensus
                 );
                 return CompScheduling::Immediate;
             },
             EC::NewComponent(definition_id, type_id, arguments) => {
                 debug_assert_eq!(self.exec_state.mode, CompMode::NonSync);
                 self.create_component_and_transfer_ports(
                     sched_ctx, comp_ctx,
                     definition_id, type_id, arguments
                 );
                 return CompScheduling::Requeue;
             },
             EC::NewChannel => {
                 debug_assert_eq!(self.exec_state.mode, CompMode::NonSync);
                 debug_assert!(self.exec_ctx.stmt.is_none());
                 let channel = comp_ctx.create_channel();
                 self.exec_ctx.stmt = ExecStmt::CreatedChannel((
                     Value::Output(port_id_to_eval(channel.putter_id)),
                     Value::Input(port_id_to_eval(channel.getter_id))
                 ));
                 self.inbox_main.push(None);
                 self.inbox_main.push(None);
                 return CompScheduling::Immediate;
+            }
+        }
+    }
+}
 impl CompPDL {
     pub(crate) fn new(initial_state: Prompt, num_ports: usize) -> Self {
         let mut inbox_main = Vec::new();
         inbox_main.reserve(num_ports);
         for _ in 0..num_ports {
             inbox_main.push(None);
+        }
         return Self{
             exec_state: CompExecState::new(),
             select_state: SelectState::new(),
             prompt: initial_state,
             control: ControlLayer::default(),
             consensus: Consensus::new(),
             sync_counter: 0,
             exec_ctx: ExecCtx{
                 stmt: ExecStmt::None,
             },
             inbox_main,
             inbox_backup: Vec::new(),
+        }
+    }
     // -------------------------------------------------------------------------
     // Running component and handling changes in global component state
     // -------------------------------------------------------------------------
     fn execute_prompt(&mut self, sched_ctx: &SchedulerCtx) -> EvalResult {
         let mut step_result = EvalContinuation::Stepping;
         while let EvalContinuation::Stepping = step_result {
             step_result = self.prompt.step(
                 &sched_ctx.runtime.protocol.types, &sched_ctx.runtime.protocol.heap,
                 &sched_ctx.runtime.protocol.modules, &mut self.exec_ctx,
             )?;
+        }
         return Ok(step_result)
+    }
     // -------------------------------------------------------------------------
     // Handling messages
     // -------------------------------------------------------------------------
     /// Handles a message that came in through the public inbox. This function
     /// will handle putting it in the correct place, and potentially blocking
     /// the port in case too many messages are being received.
     fn handle_incoming_data_message(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, message: DataMessage) {
         use component::IncomingData;
         // Whatever we do, glean information from headers in message
         if self.exec_state.mode.is_in_sync_block() {
             self.consensus.handle_incoming_data_message(comp_ctx, &message);
+        }
         match component::default_handle_incoming_data_message(
             &mut self.exec_state, &mut self.inbox_main, comp_ctx, message,
             sched_ctx, &mut self.control
         ) {
             IncomingData::PlacedInSlot => {
                 if self.exec_state.mode == CompMode::BlockedSelect {
                     let select_decision = self.select_state.handle_updated_inbox(&self.inbox_main, comp_ctx);
                     if let SelectDecision::Case(case_index) = select_decision {
                         self.exec_ctx.stmt = ExecStmt::PerformedSelectWait(case_index);
                         self.exec_state.mode = CompMode::Sync;
+                    }
+                }
             },
             IncomingData::SlotFull(message) => {
                 self.inbox_backup.push(message);
+            }
+        }
+    }
     fn handle_incoming_sync_message(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, message: SyncMessage) {
         let decision = self.consensus.receive_sync_message(sched_ctx, comp_ctx, message);
         component::default_handle_sync_decision(sched_ctx, &mut self.exec_state, comp_ctx, decision, &mut self.consensus);
+    }
     /// Handles an error coming from the generic `component::handle_xxx`
     /// functions. Hence accepts argument as a tuple.
     fn handle_generic_component_error(&mut self, sched_ctx: &SchedulerCtx, location_and_message: (PortInstruction, String)) {
         // Retrieve location and message, display in terminal
         let (location, message) = location_and_message;
         let error = match location {
             PortInstruction::None => CompError::Component(message),
             PortInstruction::NoSource => unreachable!(), // for debugging: all in-sync errors are associated with a source location
             PortInstruction::SourceLocation(expression_id) => {
                 let protocol = &sched_ctx.runtime.protocol;
                 CompError::Executor(EvalError::new_error_at_expr(
                     &self.prompt, &protocol.modules, &protocol.heap,
                     expression_id, message
                 ))
+            }
         };
         self.handle_component_error(sched_ctx, error);
+    }
     fn handle_component_error(&mut self, sched_ctx: &SchedulerCtx, error: CompError) {
         sched_ctx.error(&format!("{}", error));
         // Set state to handle subsequent error
         let exit_reason = if self.exec_state.mode.is_in_sync_block() {
             ExitReason::ErrorInSync
         } else {
             ExitReason::ErrorNonSync
         };
         self.exec_state.set_as_start_exit(exit_reason);
+    }
     // -------------------------------------------------------------------------
     // Handling ports
     // -------------------------------------------------------------------------
     /// Creates a new component and transfers ports. Because of the stepwise
     /// process in which memory is allocated, ports are transferred, messages
     /// are exchanged, component lifecycle methods are called, etc. This
     /// function facilitates a lot of implicit assumptions (e.g. when the
     /// `Component::on_creation` method is called, the component is already
     /// registered at the runtime).
     fn create_component_and_transfer_ports(
         &mut self,
         sched_ctx: &SchedulerCtx, creator_ctx: &mut CompCtx,
         definition_id: ProcedureDefinitionId, type_id: TypeId, mut arguments: ValueGroup
     ) {
         struct PortPair{
             creator_handle: LocalPortHandle,
             creator_id: PortId,
             created_handle: LocalPortHandle,
             created_id: PortId,
+        }
         let mut opened_port_id_pairs = Vec::new();
         let mut closed_port_id_pairs = Vec::new();
         let reservation = sched_ctx.runtime.start_create_pdl_component();
         let mut created_ctx = CompCtx::new(&reservation);
         let other_proc = &sched_ctx.runtime.protocol.heap[definition_id];
         let self_proc = &sched_ctx.runtime.protocol.heap[self.prompt.frames[0].definition];
         // dbg_code!({
         //     sched_ctx.log(&format!(
         //         "DEBUG: Comp '{}' (ID {:?}) is creating comp '{}' (ID {:?})",
         //         self_proc.identifier.value.as_str(), creator_ctx.id,
         //         other_proc.identifier.value.as_str(), reservation.id()
         //     ));
         // });
         // Take all the ports ID that are in the `args` (and currently belong to
         // the creator component) and translate them into new IDs that are
         // associated with the component we're about to create
         let mut arg_iter = ValueGroupPortIter::new(&mut arguments);
         while let Some(port_reference) = arg_iter.next() {
             // Create port entry for new component
             let creator_port_id = port_reference.id;
             let creator_port_handle = creator_ctx.get_port_handle(creator_port_id);
             let creator_port = creator_ctx.get_port(creator_port_handle);
             let created_port_handle = created_ctx.add_port(
                 creator_port.peer_comp_id, creator_port.peer_port_id,
                 creator_port.kind, creator_port.state
             );
             let created_port = created_ctx.get_port(created_port_handle);
             let created_port_id = created_port.self_id;
             let port_id_pair = PortPair {
                 creator_handle: creator_port_handle,
                 creator_id: creator_port_id,
                 created_handle: created_port_handle,
                 created_id: created_port_id,
             };
             if creator_port.state == PortState::Closed {
                 closed_port_id_pairs.push(port_id_pair)
             } else {
                 opened_port_id_pairs.push(port_id_pair);
+            }
             // Modify value in arguments (bit dirty, but double vec in ValueGroup causes lifetime issues)
             let arg_value = if let Some(heap_pos) = port_reference.heap_pos {
                 &mut arg_iter.group.regions[heap_pos][port_reference.index]
             } else {
                 &mut arg_iter.group.values[port_reference.index]
             };
             match arg_value {
                 Value::Input(id) => *id = port_id_to_eval(created_port_id),
                 Value::Output(id) => *id = port_id_to_eval(created_port_id),
                 _ => unreachable!(),
+            }
+        }
         // For each transferred port pair set their peer components to the
         // correct values. This will only change the values for the ports of
         // the new component.
         let mut created_component_has_remote_peers = false;
         for pair in opened_port_id_pairs.iter() {
             let creator_port_info = creator_ctx.get_port(pair.creator_handle);
             let created_port_info = created_ctx.get_port_mut(pair.created_handle);
             if created_port_info.peer_comp_id == creator_ctx.id {
                 // Port peer is owned by the creator as well
                 let created_peer_port_index = opened_port_id_pairs
                     .iter()
                     .position(|v| v.creator_id == creator_port_info.peer_port_id);
                 match created_peer_port_index {
                     Some(created_peer_port_index) => {
                         // Peer port moved to the new component as well. So
                         // adjust IDs appropriately.
                         let peer_pair = &opened_port_id_pairs[created_peer_port_index];
                         created_port_info.peer_port_id = peer_pair.created_id;
                         created_port_info.peer_comp_id = reservation.id();
                         todo!("either add 'self peer', or remove that idea from Ctx altogether")
                     },
                     None => {
                         // Peer port remains with creator component.
                         created_port_info.peer_comp_id = creator_ctx.id;
                         created_ctx.add_peer(pair.created_handle, sched_ctx, creator_ctx.id, None);
+                    }
+                }
             } else {
                 // Peer is a different component. We'll deal with sending the
                 // appropriate messages later
                 let peer_handle = creator_ctx.get_peer_handle(created_port_info.peer_comp_id);
                 let peer_info = creator_ctx.get_peer(peer_handle);
                 created_ctx.add_peer(pair.created_handle, sched_ctx, peer_info.id, Some(&peer_info.handle));
                 created_component_has_remote_peers = true;
+            }
+        }
         // We'll now actually turn our reservation for a new component into an
         // actual component. Note that we initialize it as "not sleeping" as
         // its initial scheduling might be performed based on `Ack`s in response
         // to message exchanges between remote peers.
         let total_num_ports = opened_port_id_pairs.len() + closed_port_id_pairs.len();
         let component = component::create_component(&sched_ctx.runtime.protocol, definition_id, type_id, arguments, total_num_ports);
         let (created_key, component) = sched_ctx.runtime.finish_create_pdl_component(
             reservation, component, created_ctx, false,
         );
         component.component.on_creation(created_key.downgrade(), sched_ctx);
         // Now modify the creator's ports: remove every transferred port and
         // potentially remove the peer component.
         for pair in opened_port_id_pairs.iter() {
             // Remove peer if appropriate
             let creator_port_info = creator_ctx.get_port(pair.creator_handle);
             let creator_port_index = creator_ctx.get_port_index(pair.creator_handle);
             let creator_peer_comp_id = creator_port_info.peer_comp_id;
             creator_ctx.remove_peer(sched_ctx, pair.creator_handle, creator_peer_comp_id, false);
             creator_ctx.remove_port(pair.creator_handle);
             // Transfer any messages
             if let Some(mut message) = self.inbox_main.remove(creator_port_index) {
                 message.data_header.target_port = pair.created_id;
                 component.component.adopt_message(&mut component.ctx, message)
+            }
             let mut message_index = 0;
             while message_index < self.inbox_backup.len() {
                 let message = &self.inbox_backup[message_index];
                 if message.data_header.target_port == pair.creator_id {
                     // transfer message
                     let mut message = self.inbox_backup.remove(message_index);
                     message.data_header.target_port = pair.created_id;
                     component.component.adopt_message(&mut component.ctx, message);
                 } else {
                     message_index += 1;
+                }
+            }
             // Handle potential channel between creator and created component
             let created_port_info = component.ctx.get_port(pair.created_handle);
             if created_port_info.peer_comp_id == creator_ctx.id {
                 let peer_port_handle = creator_ctx.get_port_handle(created_port_info.peer_port_id);
                 let peer_port_info = creator_ctx.get_port_mut(peer_port_handle);
                 peer_port_info.peer_comp_id = component.ctx.id;
                 peer_port_info.peer_port_id = created_port_info.self_id;
                 creator_ctx.add_peer(peer_port_handle, sched_ctx, component.ctx.id, None);
+            }
+        }
         // Do the same for the closed ports. Note that we might still have to
         // transfer messages that cause the new owner of the port to fail.
         for pair in closed_port_id_pairs.iter() {
             let port_index = creator_ctx.get_port_index(pair.creator_handle);
             creator_ctx.remove_port(pair.creator_handle);
             if let Some(mut message) = self.inbox_main.remove(port_index) {
                 message.data_header.target_port = pair.created_id;
                 component.component.adopt_message(&mut component.ctx, message);
+            }
             let mut message_index = 0;
             while message_index < self.inbox_backup.len() {
                 let message = &self.inbox_backup[message_index];
                 if message.data_header.target_port == pair.created_id {
                     // Transfer message
                     let mut message = self.inbox_backup.remove(message_index);
                     message.data_header.target_port = pair.created_id;
                     component.component.adopt_message(&mut component.ctx, message);
                 } else {
                     message_index += 1;
+                }
+            }
+        }
         // By now all ports and messages have been transferred. If there are any
         // peers that need to be notified about this new component, then we
         // initiate the protocol that will notify everyone here.
         if created_component_has_remote_peers {
             let created_ctx = &component.ctx;
             let schedule_entry_id = self.control.add_schedule_entry(created_ctx.id);
             for pair in opened_port_id_pairs.iter() {
                 let port_info = created_ctx.get_port(pair.created_handle);
                 if port_info.peer_comp_id != creator_ctx.id && port_info.peer_comp_id != created_ctx.id {
                     let message = self.control.add_reroute_entry(
                         creator_ctx.id, port_info.peer_port_id, port_info.peer_comp_id,
                         pair.creator_id, pair.created_id, created_ctx.id,
                         schedule_entry_id
                     );
                     let peer_handle = created_ctx.get_peer_handle(port_info.peer_comp_id);
                     let peer_info = created_ctx.get_peer(peer_handle);
-                    peer_info.handle.send_message(&sched_ctx.runtime, message, true);
+                    peer_info.handle.send_message_logged(sched_ctx, message, true);
+                }
+            }
         } else {
             // Peer can be scheduled immediately
             sched_ctx.runtime.enqueue_work(created_key);
+        }
+    }
+}
 /// Recursively goes through the value group, attempting to find ports.
 /// Duplicates will only be added once.
 pub(crate) fn find_ports_in_value_group(value_group: &ValueGroup, ports: &mut Vec<PortId>) {
     // Helper to check a value for a port and recurse if needed.
     fn find_port_in_value(group: &ValueGroup, value: &Value, ports: &mut Vec<PortId>) {
         match value {
             Value::Input(port_id) | Value::Output(port_id) => {
                 // This is an actual port
                 let cur_port = PortId(port_id.id);
                 for prev_port in ports.iter() {
                     if *prev_port == cur_port {
                         // Already added
                         return;
+                    }
+                }
                 ports.push(cur_port);
             },
             Value::Array(heap_pos) |
             Value::Message(heap_pos) |
             Value::String(heap_pos) |
             Value::Struct(heap_pos) |
             Value::Union(_, heap_pos) => {
                 // Reference to some dynamic thing which might contain ports,
                 // so recurse
                 let heap_region = &group.regions[*heap_pos as usize];
                 for embedded_value in heap_region {
                     find_port_in_value(group, embedded_value, ports);
+                }
             },
             _ => {}, // values we don't care about
+        }
+    }
     // Clear the ports, then scan all the available values
     ports.clear();
     for value in &value_group.values {
         find_port_in_value(value_group, value, ports);
+    }
+}
 struct ValueGroupPortIter<'a> {
     group: &'a mut ValueGroup,
     heap_stack: Vec<(usize, usize)>,
     index: usize,
+}
 impl<'a> ValueGroupPortIter<'a> {
     fn new(group: &'a mut ValueGroup) -> Self {
         return Self{ group, heap_stack: Vec::new(), index: 0 }
+    }
+}
 struct ValueGroupPortRef {
     id: PortId,
     heap_pos: Option<usize>, // otherwise: on stack
     index: usize,
+}
 impl<'a> Iterator for ValueGroupPortIter<'a> {
     type Item = ValueGroupPortRef;
     fn next(&mut self) -> Option<Self::Item> {
         // Enter loop that keeps iterating until a port is found
         loop {
             if let Some(pos) = self.heap_stack.last() {
                 let (heap_pos, region_index) = *pos;
                 if region_index >= self.group.regions[heap_pos].len() {
                     self.heap_stack.pop();
                     continue;
+                }
                 let value = &self.group.regions[heap_pos][region_index];
                 self.heap_stack.last_mut().unwrap().1 += 1;
                 match value {
                     Value::Input(id) | Value::Output(id) => {
                         let id = PortId(id.id);
                         return Some(ValueGroupPortRef{
                             id,
                             heap_pos: Some(heap_pos),
                             index: region_index,
                         });
                     },
                     _ => {},
+                }
                 if let Some(heap_pos) = value.get_heap_pos() {
                     self.heap_stack.push((heap_pos as usize, 0));
+                }
             } else {
                 if self.index >= self.group.values.len() {
                     return None;
+                }
                 let value = &mut self.group.values[self.index];
                 self.index += 1;
                 match value {
                     Value::Input(id) | Value::Output(id) => {
                         let id = PortId(id.id);
                         return Some(ValueGroupPortRef{
                             id,
                             heap_pos: None,
                             index: self.index - 1
                         });
                     },
                     _ => {},
+                }
                 // Not a port, check if we need to enter a heap region
                 if let Some(heap_pos) = value.get_heap_pos() {
                     self.heap_stack.push((heap_pos as usize, 0));
                 } // else: just consider the next value
+            }
+        }
+    }
+}
@@ \ No newline at end of file @@

src/runtime2/component/component_random.rs

➞

Show inline comments

 use rand::prelude as random;
 use rand::RngCore;
 use crate::protocol::eval::{ValueGroup, Value};
 use crate::runtime2::*;
 use super::*;
 use super::component::{
     self,
     Component, CompExecState, CompScheduling,
     CompMode, ExitReason
 };
 use super::control_layer::*;
 use super::consensus::*;
 /// TODO: Temporary component to figure out what to do with custom components.
 ///     This component sends random numbers between two u32 limits
 pub struct ComponentRandomU32 {
     // Properties for this specific component
     output_port_id: PortId,
     random_minimum: u32,
     random_maximum: u32,
     num_sends: u32,
     max_num_sends: u32,
     generator: random::ThreadRng,
     // Generic state-tracking
     exec_state: CompExecState,
     did_perform_send: bool, // when in sync mode
     control: ControlLayer,
     consensus: Consensus,
+}
 impl Component for ComponentRandomU32 {
     fn on_creation(&mut self, _id: CompId, _sched_ctx: &SchedulerCtx) {}
     fn on_shutdown(&mut self, sched_ctx: &SchedulerCtx) {}
     fn adopt_message(&mut self, _comp_ctx: &mut CompCtx, _message: DataMessage) {
         // Impossible since this component does not have any input ports in its
         // signature.
         unreachable!();
+    }
     fn handle_message(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx, message: Message) {
         match message {
             Message::Data(_message) => unreachable!(),
             Message::Sync(message) => {
                 let decision = self.consensus.receive_sync_message(sched_ctx, comp_ctx, message);
                 component::default_handle_sync_decision(sched_ctx, &mut self.exec_state, comp_ctx, decision, &mut self.consensus);
             },
             Message::Control(message) => {
                 if let Err(location_and_message) = component::default_handle_control_message(
                     &mut self.exec_state, &mut self.control, &mut self.consensus,
                     message, sched_ctx, comp_ctx
                 ) {
                     component::default_handle_error_for_builtin(&mut self.exec_state, sched_ctx, location_and_message);
+                }
             },
             Message::Poll => unreachable!(),
+        }
+    }
     fn run(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx) -> CompScheduling {
-        sched_ctx.log(&format!("Running component ComponentRandomU32 (mode: {:?})", self.exec_state.mode));
+        sched_ctx.info(&format!("Running component ComponentRandomU32 (mode: {:?})", self.exec_state.mode));
         match self.exec_state.mode {
             CompMode::BlockedGet | CompMode::BlockedSelect => {
                 // impossible for this component, no input ports and no select
                 // blocks
                 unreachable!();
+            }
             CompMode::NonSync => {
                 // If in non-sync mode then we check if the arguments make sense
                 // (at some point in the future, this is just a testing
                 // component).
                 if self.random_minimum >= self.random_maximum {
                     // Could throw an evaluation error, but lets just panic
                     panic!("going to crash 'n burn your system now, please provide valid arguments");
+                }
                 if self.num_sends >= self.max_num_sends {
                     self.exec_state.set_as_start_exit(ExitReason::Termination);
                 } else {
-                    sched_ctx.log("Entering sync mode");
+                    sched_ctx.info("Entering sync mode");
                     self.did_perform_send = false;
                     component::default_handle_sync_start(
                         &mut self.exec_state, &mut [], sched_ctx, comp_ctx, &mut self.consensus
                     );
+                }
                 return CompScheduling::Immediate;
             },
             CompMode::Sync => {
                 // This component just sends a single message, then waits until
                 // consensus has been reached
                 if !self.did_perform_send {
-                    sched_ctx.log("Sending random message");
+                    sched_ctx.info("Sending random message");
                     let mut random = self.generator.next_u32() - self.random_minimum;
                     let random_delta = self.random_maximum - self.random_minimum;
                     random %= random_delta;
                     random += self.random_minimum;
                     let value_group = ValueGroup::new_stack(vec![Value::UInt32(random)]);
                     let send_result = component::default_send_data_message(
                         &mut self.exec_state, self.output_port_id,
                         PortInstruction::NoSource, value_group,
                         sched_ctx, &mut self.consensus, comp_ctx
                     );
                     if let Err(location_and_message) = send_result {
                         component::default_handle_error_for_builtin(&mut self.exec_state, sched_ctx, location_and_message);
                         return CompScheduling::Immediate
                     } else {
                         // Blocked or not, we set `did_perform_send` to true. If
                         // blocked then the moment we become unblocked (and are back
                         // at the `Sync` mode) we have sent the message.
                         let scheduling = send_result.unwrap();
                         self.did_perform_send = true;
                         self.num_sends += 1;
                         return scheduling
+                    }
                 } else {
                     // Message was sent, finish this sync round
                     component::default_handle_sync_end(&mut self.exec_state, sched_ctx, comp_ctx, &mut self.consensus);
                     return CompScheduling::Requeue;
+                }
             },
             CompMode::SyncEnd | CompMode::BlockedPut => return CompScheduling::Sleep,
             CompMode::StartExit => return component::default_handle_start_exit(
                 &mut self.exec_state, &mut self.control, sched_ctx, comp_ctx, &mut self.consensus
             ),
             CompMode::BusyExit => return component::default_handle_busy_exit(
                 &mut self.exec_state, &self.control, sched_ctx
             ),
             CompMode::Exit => return component::default_handle_exit(&self.exec_state),
+        }
+    }
+}
 impl ComponentRandomU32 {
     pub(crate) fn new(arguments: ValueGroup) -> Self {
         debug_assert_eq!(arguments.values.len(), 4);
         debug_assert!(arguments.regions.is_empty());
         let port_id = component::port_id_from_eval(arguments.values[0].as_port_id());
         let minimum = arguments.values[1].as_uint32();
         let maximum = arguments.values[2].as_uint32();
         let num_sends = arguments.values[3].as_uint32();
         return Self{
             output_port_id: port_id,
             random_minimum: minimum,
             random_maximum: maximum,
             num_sends: 0,
             max_num_sends: num_sends,
             generator: random::thread_rng(),
             exec_state: CompExecState::new(),
             did_perform_send: false,
             control: ControlLayer::default(),
             consensus: Consensus::new(),
+        }
+    }
+}
@@ \ 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.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, 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, 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, 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);
+                peer.handle.send_message_logged(sched_ctx, 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);
+            peer_info.handle.send_message_logged(sched_ctx, 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, 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 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);
+            handle.send_message_logged(sched_ctx, 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);
+        leader_info.send_message_logged(sched_ctx, 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 @@

src/runtime2/poll/mod.rs

➞

Show inline comments

 use libc::{self, c_int};
 use std::{io, ptr, time, thread};
 use std::sync::Arc;
 use std::sync::atomic::{AtomicU32, Ordering};
 use std::collections::HashMap;
 use crate::runtime2::RtError;
 use crate::runtime2::runtime::{CompHandle, RuntimeInner};
+use crate::runtime2::runtime::{CompHandle, RuntimeInner, LogLevel};
 use crate::runtime2::store::queue_mpsc::*;
 pub(crate) type FileDescriptor = c_int;
 pub(crate) trait AsFileDescriptor {
     fn as_file_descriptor(&self) -> FileDescriptor;
+}
 #[derive(Copy, Clone)]
 pub(crate) struct UserData(u64);
 // -----------------------------------------------------------------------------
 // Poller
 // -----------------------------------------------------------------------------
 #[cfg(unix)]
 pub(crate) struct Poller {
     handle: c_int,
+}
 // All of this is gleaned from the `mio` crate.
 #[cfg(unix)]
 impl Poller {
     pub fn new() -> io::Result<Self> {
         let handle = syscall_result(unsafe{ libc::epoll_create1(libc::EPOLL_CLOEXEC) })?;
         return Ok(Self{
             handle,
         })
+    }
     fn register(&self, fd: FileDescriptor, user: UserData, read: bool, write: bool) -> io::Result<()> {
         let mut event = libc::epoll_event{
             events: Self::events_from_rw_flags(read, write),
             u64: user.0,
         };
         syscall_result(unsafe{
             libc::epoll_ctl(self.handle, libc::EPOLL_CTL_ADD, fd, &mut event)
         })?;
         return Ok(());
+    }
     fn unregister(&self, fd: FileDescriptor) -> io::Result<()> {
         syscall_result(unsafe{
             libc::epoll_ctl(self.handle, libc::EPOLL_CTL_DEL, fd, ptr::null_mut())
         })?;
         return Ok(());
+    }
     /// Performs `epoll_wait`, waiting for the provided timeout or until events
     /// are reported. They are stored in the `events` variable (up to
     /// `events.cap()` are reported, so ensure it is preallocated).
     pub fn wait(&self, events: &mut Vec<libc::epoll_event>, timeout: time::Duration) -> io::Result<()> {
         // See `mio` for the reason. Works around a linux bug
         #[cfg(target_pointer_width = "32")]
         const MAX_TIMEOUT: u128 = 1789569;
         #[cfg(not(target_pointer_width = "32"))]
         const MAX_TIMEOUT: u128 = c_int::MAX as u128;
         let timeout_millis = timeout.as_millis();
         let timeout_millis = if timeout_millis > MAX_TIMEOUT {
             -1 // effectively infinite
         } else {
             timeout_millis as c_int
         };
         debug_assert!(events.is_empty());
         debug_assert!(events.capacity() > 0 && events.capacity() < i32::MAX as usize);
         let num_events = syscall_result(unsafe{
             libc::epoll_wait(self.handle, events.as_mut_ptr(), events.capacity() as i32, timeout_millis)
         })?;
         unsafe{
             debug_assert!(num_events >= 0);
             events.set_len(num_events as usize);
+        }
         return Ok(());
+    }
     fn events_from_rw_flags(read: bool, write: bool) -> u32 {
         let mut events = libc::EPOLLET;
         if read {
             events |= libc::EPOLLIN | libc::EPOLLRDHUP;
+        }
         if write {
             events |= libc::EPOLLOUT;
+        }
         return events as u32;
+    }
+}
 #[cfg(unix)]
 impl Drop for Poller {
     fn drop(&mut self) {
         unsafe{ libc::close(self.handle); }
+    }
+}
 #[inline]
 fn syscall_result(result: c_int) -> io::Result<c_int> {
     if result < 0 {
         return Err(io::Error::last_os_error());
     } else {
         return Ok(result);
+    }
+}
 #[cfg(not(unix))]
 struct Poller {
     // Not implemented for OS's other than unix
+}
 // -----------------------------------------------------------------------------
 // Polling Thread
 // -----------------------------------------------------------------------------
 enum PollCmd {
     Register(CompHandle, UserData),
     Unregister(FileDescriptor, UserData),
     Shutdown,
+}
 pub struct PollingThread {
     poller: Arc<Poller>,
     runtime: Arc<RuntimeInner>,
     queue: QueueDynMpsc<PollCmd>,
-    logging_enabled: bool,
+    log_level: LogLevel,
+}
 impl PollingThread {
-    pub(crate) fn new(runtime: Arc<RuntimeInner>, logging_enabled: bool) -> Result<(PollingThreadHandle, PollingClientFactory), RtError> {
+    pub(crate) fn new(runtime: Arc<RuntimeInner>, log_level: LogLevel) -> Result<(PollingThreadHandle, PollingClientFactory), RtError> {
         let poller = Poller::new()
             .map_err(|e| rt_error!("failed to create poller, because: {}", e))?;
         let poller = Arc::new(poller);
         let queue = QueueDynMpsc::new(64);
         let queue_producers = queue.producer_factory();
         let mut thread_data = PollingThread{
             poller: poller.clone(),
             runtime: runtime.clone(),
             queue,
-            logging_enabled,
+            log_level,
         };
         let thread_handle = thread::spawn(move || { thread_data.run() });
         let thread_handle = thread::Builder::new()
             .name(String::from("poller"))
             .spawn(move || { thread_data.run() })
             .map_err(|reason|
                 rt_error!("failed to start polling thread, because: {}", reason)
             )?;
         let thread_handle = PollingThreadHandle{
             queue: Some(queue_producers.producer()),
             handle: Some(thread_handle),
         };
         let client_factory = PollingClientFactory{
             poller,
             generation_counter: Arc::new(AtomicU32::new(0)),
             queue_factory: queue_producers,
         };
         return Ok((thread_handle, client_factory));
+    }
     pub(crate) fn run(&mut self) {
         use crate::runtime2::communication::Message;
         const NUM_EVENTS: usize = 256;
         const EPOLL_DURATION: time::Duration = time::Duration::from_millis(250);
         // @performance: Lot of improvements possible here, a HashMap is likely
         // a horrible way to do this.
         let mut events = Vec::with_capacity(NUM_EVENTS);
         let mut lookup = HashMap::with_capacity(64);
         self.log("Starting polling thread");
         loop {
             // Retrieve events first (because the PollingClient will first
             // register at epoll, and then push a command into the queue).
             self.poller.wait(&mut events, EPOLL_DURATION).unwrap();
             // Then handle everything in the command queue.
             while let Some(command) = self.queue.pop() {
                 match command {
                     PollCmd::Register(handle, user_data) => {
                         self.log(&format!("Registering component {:?} as {}", handle.id(), user_data.0));
                         let key = Self::user_data_as_key(user_data);
                         debug_assert!(!lookup.contains_key(&key));
                         lookup.insert(key, handle);
                     },
                     PollCmd::Unregister(_file_descriptor, user_data) => {
                         let key = Self::user_data_as_key(user_data);
                         debug_assert!(lookup.contains_key(&key));
                         let mut handle = lookup.remove(&key).unwrap();
                         self.log(&format!("Unregistering component {:?} as {}", handle.id(), user_data.0));
                         if let Some(key) = handle.decrement_users() {
                             self.runtime.destroy_component(key);
+                        }
                     },
                     PollCmd::Shutdown => {
                         // The contract is that all scheduler threads shutdown
                         // before the polling thread. This happens when all
                         // components are removed.
                         self.log("Received shutdown signal");
                         debug_assert!(lookup.is_empty());
                         return;
+                    }
+                }
+            }
             // Now process all of the events. Because we might have had a
             // `Register` command followed by an `Unregister` command (e.g. a
             // component has died), we might get events that are not associated
             // with an entry in the lookup.
             for event in events.drain(..) {
                 let key = event.u64;
                 if let Some(handle) = lookup.get(&key) {
                     let events = event.events;
                     self.log(&format!("Sending poll to {:?} (event: {:x})", handle.id(), events));
                     handle.send_message(&self.runtime, Message::Poll, true);
+                }
+            }
+        }
+    }
     #[inline]
     fn user_data_as_key(data: UserData) -> u64 {
         return data.0;
+    }
     fn log(&self, message: &str) {
-        if self.logging_enabled {
+        if self.log_level >= LogLevel::Info {
             println!("[polling] {}", message);
+        }
+    }
+}
 // bit convoluted, but it works
 pub(crate) struct PollingThreadHandle {
     // requires Option, because:
     queue: Option<QueueDynProducer<PollCmd>>, // destructor needs to be called
     handle: Option<thread::JoinHandle<()>>, // we need to call `join`
+}
 impl PollingThreadHandle {
     pub(crate) fn shutdown(&mut self) -> thread::Result<()> {
         debug_assert!(self.handle.is_some(), "polling thread already destroyed");
         self.queue.take().unwrap().push(PollCmd::Shutdown);
         return self.handle.take().unwrap().join();
+    }
+}
 impl Drop for PollingThreadHandle {
     fn drop(&mut self) {
         debug_assert!(self.queue.is_none() && self.handle.is_none());
+    }
+}
 // oh my god, now I'm writing factory objects. I'm not feeling too well
 pub(crate) struct PollingClientFactory {
     poller: Arc<Poller>,
     generation_counter: Arc<AtomicU32>,
     queue_factory: QueueDynProducerFactory<PollCmd>,
+}
 impl PollingClientFactory {
     pub(crate) fn client(&self) -> PollingClient {
         return PollingClient{
             poller: self.poller.clone(),
             generation_counter: self.generation_counter.clone(),
             queue: self.queue_factory.producer(),
         };
+    }
+}
 pub(crate) struct PollTicket(FileDescriptor, u64);
 /// A structure that allows the owner to register components at the polling
 /// thread. Because of assumptions in the communication queue all of these
 /// clients should be dropped before stopping the polling thread.
 pub(crate) struct PollingClient {
     poller: Arc<Poller>,
     generation_counter: Arc<AtomicU32>,
     queue: QueueDynProducer<PollCmd>,
+}
 impl PollingClient {
     pub(crate) fn register<F: AsFileDescriptor>(&self, entity: &F, handle: CompHandle, read: bool, write: bool) -> Result<PollTicket, RtError> {
         let generation = self.generation_counter.fetch_add(1, Ordering::Relaxed);
         let user_data = user_data_for_component(handle.id().0, generation);
         self.queue.push(PollCmd::Register(handle, user_data));
         let file_descriptor = entity.as_file_descriptor();
         self.poller.register(file_descriptor, user_data, read, write)
             .map_err(|e| rt_error!("failed to register for polling, because: {}", e))?;
         return Ok(PollTicket(file_descriptor, user_data.0));
+    }
     pub(crate) fn unregister(&self, ticket: PollTicket) -> Result<(), RtError> {
         let file_descriptor = ticket.0;
         let user_data = UserData(ticket.1);
         self.queue.push(PollCmd::Unregister(file_descriptor, user_data));
         self.poller.unregister(file_descriptor)
             .map_err(|e| rt_error!("failed to unregister polling, because: {}", e))?;
         return Ok(());
+    }
+}
 #[inline]
 fn user_data_for_component(component_id: u32, generation: u32) -> UserData {
     return UserData((generation as u64) << 32 | (component_id as u64));
+}
@@ \ No newline at end of file @@

src/runtime2/runtime.rs

➞

Show inline comments

 use std::sync::{Arc, Mutex, Condvar};
 use std::sync::atomic::{AtomicU32, AtomicBool, Ordering};
 use std::thread;
 use std::collections::VecDeque;
 use crate::protocol::*;
 use crate::runtime2::poll::{PollingThread, PollingThreadHandle};
 use crate::runtime2::RtError;
 use super::communication::Message;
 use super::component::{Component, wake_up_if_sleeping, CompPDL, CompCtx};
 use super::store::{ComponentStore, ComponentReservation, QueueDynMpsc, QueueDynProducer};
 use super::scheduler::*;
 #[derive(PartialOrd, PartialEq, Copy, Clone)]
 pub enum LogLevel {
     None, // no logging
     Debug, // all logging (includes messages)
     Info, // rough logging
+}
 // -----------------------------------------------------------------------------
 // Component
 // -----------------------------------------------------------------------------
 /// Key to a component. Type system somewhat ensures that there can only be one
 /// of these. Only with a key one may retrieve privately-accessible memory for
 /// a component. Practically just a generational index, like `CompId` is.
 #[derive(Debug, Copy, Clone, PartialEq, Eq)]
 pub(crate) struct CompKey(pub u32);
 impl CompKey {
     pub(crate) fn downgrade(&self) -> CompId {
         return CompId(self.0);
+    }
+}
 /// Generational ID of a component.
 #[derive(Debug, Copy, Clone, PartialEq, Eq)]
 pub struct CompId(pub u32);
 impl CompId {
     pub(crate) fn new_invalid() -> CompId {
         return CompId(u32::MAX);
+    }
     /// Upgrade component ID to component key. Unsafe because the caller needs
     /// to make sure that only one component key can exist at a time (to ensure
     /// a component can only be scheduled/executed by one thread).
     pub(crate) unsafe fn upgrade(&self) -> CompKey {
         return CompKey(self.0);
+    }
+}
 /// Handle to a component that is being created.
 pub(crate) struct CompReserved {
     reservation: ComponentReservation,
+}
 impl CompReserved {
     pub(crate) fn id(&self) -> CompId {
         return CompId(self.reservation.index)
+    }
+}
 /// Representation of a runtime component. Contains the bookkeeping variables
 /// for the schedulers, the publicly accessible fields, and the private fields
 /// that should only be accessed by the thread running the component's routine.
 pub(crate) struct RuntimeComp {
     pub public: CompPublic,
     pub component: Box<dyn Component>,
     pub ctx: CompCtx,
     pub inbox: QueueDynMpsc<Message>,
     pub exiting: bool,
+}
 /// Should contain everything that is accessible in a thread-safe manner
 // TODO: Do something about the `num_handles` thing. This needs to be a bit more
 //  "foolproof" to lighten the mental burden of using the `num_handles`
 //  variable.
 pub(crate) struct CompPublic {
     pub sleeping: AtomicBool,
     pub num_handles: AtomicU32, // manually modified (!)
     inbox: QueueDynProducer<Message>,
+}
 /// Handle to public part of a component. Would be nice if we could
 /// automagically manage the `num_handles` counter. But when it reaches zero we
 /// need to manually remove the handle from the runtime. So we just have debug
 /// code to make sure this actually happens.
 pub(crate) struct CompHandle {
     target: *const CompPublic,
     id: CompId,
     #[cfg(debug_assertions)] decremented: bool,
+}
 impl CompHandle {
     fn new(id: CompId, public: &CompPublic) -> CompHandle {
         let handle = CompHandle{
             target: public,
             id,
             #[cfg(debug_assertions)] decremented: false,
         };
         handle.increment_users();
         return handle;
+    }
     pub(crate) fn send_message(&self, runtime: &RuntimeInner, message: Message, try_wake_up: bool) {
         self.inbox.push(message);
         if try_wake_up {
             wake_up_if_sleeping(runtime, self.id, self);
+        }
+    }
     #[inline]
     pub(crate) fn send_message_logged(&self, sched_ctx: &SchedulerCtx, message: Message, try_wake_up: bool) {
         sched_ctx.debug(&format!("Sending message to comp:{} ... {:?}", self.id.0, message));
         self.send_message(&sched_ctx.runtime, message, try_wake_up);
+    }
     pub(crate) fn id(&self) -> CompId {
         return self.id;
+    }
     fn increment_users(&self) {
         let old_count = self.num_handles.fetch_add(1, Ordering::AcqRel);
         debug_assert!(old_count > 0); // because we should never be able to retrieve a handle when the component is (being) destroyed
+    }
     /// Returns the `CompKey` to the component if it should be destroyed
     pub(crate) fn decrement_users(&mut self) -> Option<CompKey> {
         dbg_code!(assert!(!self.decremented, "illegal to 'decrement_users' twice"));
         let old_count = self.num_handles.fetch_sub(1, Ordering::AcqRel);
         let new_count = old_count - 1;
         dbg_code!(self.decremented = true);
         if new_count == 0 {
             return Some(unsafe{ self.id.upgrade() });
+        }
         return None;
+    }
+}
 impl Clone for CompHandle {
     fn clone(&self) -> Self {
         dbg_code!(assert!(!self.decremented, "illegal to clone after 'decrement_users'"));
         self.increment_users();
         return CompHandle{
             target: self.target,
             id: self.id,
             #[cfg(debug_assertions)] decremented: false,
         };
+    }
+}
 impl std::ops::Deref for CompHandle {
     type Target = CompPublic;
     fn deref(&self) -> &Self::Target {
         dbg_code!(assert!(!self.decremented)); // cannot access if control is relinquished
         return unsafe{ &*self.target };
+    }
+}
 impl Drop for CompHandle {
     fn drop(&mut self) {
         dbg_code!(assert!(self.decremented, "need call to 'decrement_users' before dropping"));
+    }
+}
 // -----------------------------------------------------------------------------
 // Runtime
 // -----------------------------------------------------------------------------
 pub struct Runtime {
     pub(crate) inner: Arc<RuntimeInner>,
     scheduler_threads: Vec<thread::JoinHandle<()>>,
     polling_handle: PollingThreadHandle,
+}
 impl Runtime {
     // TODO: debug_logging should be removed at some point
-    pub fn new(num_threads: u32, debug_logging: bool, protocol_description: ProtocolDescription) -> Result<Runtime, RtError> {
+    pub fn new(num_threads: u32, log_level: LogLevel, protocol_description: ProtocolDescription) -> Result<Runtime, RtError> {
         if num_threads == 0 {
             return Err(rt_error!("need at least one thread to create the runtime"));
+        }
         let runtime_inner = Arc::new(RuntimeInner {
             protocol: protocol_description,
             components: ComponentStore::new(128),
             work_queue: Mutex::new(VecDeque::with_capacity(128)),
             work_condvar: Condvar::new(),
             active_elements: AtomicU32::new(1),
         });
         let (polling_handle, polling_clients) = rt_error_try!(
-            PollingThread::new(runtime_inner.clone(), debug_logging),
+            PollingThread::new(runtime_inner.clone(), log_level),
             "failed to build polling thread"
         );
         let mut scheduler_threads = Vec::with_capacity(num_threads as usize);
         for thread_index in 0..num_threads {
             let mut scheduler = Scheduler::new(
                 runtime_inner.clone(), polling_clients.client(),
-                thread_index, debug_logging
+                thread_index, log_level
             );
             let thread_handle = thread::spawn(move || {
                 scheduler.run();
             });
             let thread_handle = thread::Builder::new()
                 .name(format!("scheduler:{}", thread_index))
                 .spawn(move || {
                     scheduler.run();
                 })
                 .map_err(|reason|
                     rt_error!(
                         "failed to spawn scheduler thread {}, because: {}",
                         thread_index, reason
+                    )
                 )?;
             scheduler_threads.push(thread_handle);
+        }
         return Ok(Runtime{
             inner: runtime_inner,
             scheduler_threads,
             polling_handle,
         });
+    }
     pub fn create_component(&self, module_name: &[u8], routine_name: &[u8]) -> Result<(), ComponentCreationError> {
         use crate::protocol::eval::ValueGroup;
         let prompt = self.inner.protocol.new_component(
             module_name, routine_name,
             ValueGroup::new_stack(Vec::new())
         )?;
         let reserved = self.inner.start_create_pdl_component();
         let ctx = CompCtx::new(&reserved);
         let component = Box::new(CompPDL::new(prompt, 0));
         let (key, _) = self.inner.finish_create_pdl_component(reserved, component, ctx, false);
         self.inner.enqueue_work(key);
         return Ok(())
+    }
+}
 impl Drop for Runtime {
     fn drop(&mut self) {
         self.inner.decrement_active_components();
         for handle in self.scheduler_threads.drain(..) {
             handle.join().expect("join scheduler thread");
+        }
         self.polling_handle.shutdown().expect("shutdown polling thread");
+    }
+}
 /// Memory that is maintained by "the runtime". In practice it is maintained by
 /// multiple schedulers, and this serves as the common interface to that memory.
 pub(crate) struct RuntimeInner {
     pub protocol: ProtocolDescription,
     components: ComponentStore<RuntimeComp>,
     work_queue: Mutex<VecDeque<CompKey>>, // TODO: should be MPMC queue
     work_condvar: Condvar,
     active_elements: AtomicU32, // active components and APIs (i.e. component creators)
+}
 impl RuntimeInner {
     // Scheduling and retrieving work
     pub(crate) fn take_work(&self) -> Option<CompKey> {
         let mut lock = self.work_queue.lock().unwrap();
         while lock.is_empty() && self.active_elements.load(Ordering::Acquire) != 0 {
             lock = self.work_condvar.wait(lock).unwrap();
+        }
         // We have work, or the schedulers should exit.
         return lock.pop_front();
+    }
     pub(crate) fn enqueue_work(&self, key: CompKey) {
         let mut lock = self.work_queue.lock().unwrap();
         lock.push_back(key);
         self.work_condvar.notify_one();
+    }
     // Creating/destroying components
     pub(crate) fn start_create_pdl_component(&self) -> CompReserved {
         self.increment_active_components();
         let reservation = self.components.reserve();
         return CompReserved{ reservation };
+    }
     pub(crate) fn finish_create_pdl_component(
         &self, reserved: CompReserved,
         component: Box<dyn Component>, mut context: CompCtx, initially_sleeping: bool,
     ) -> (CompKey, &mut RuntimeComp) {
         let inbox_queue = QueueDynMpsc::new(16);
         let inbox_producer = inbox_queue.producer();
         let _id = reserved.id();
         context.id = reserved.id();
         let component = RuntimeComp {
             public: CompPublic{
                 sleeping: AtomicBool::new(initially_sleeping),
                 num_handles: AtomicU32::new(1), // the component itself acts like a handle
                 inbox: inbox_producer,
             },
             component,
             ctx: context,
             inbox: inbox_queue,
             exiting: false,
         };
         let index = self.components.submit(reserved.reservation, component);
         debug_assert_eq!(index, _id.0);
         let component = self.components.get_mut(index);
         return (CompKey(index), component);
+    }
     pub(crate) fn get_component(&self, key: CompKey) -> &mut RuntimeComp {
         let component = self.components.get_mut(key.0);
         return component;
+    }
     pub(crate) fn get_component_public(&self, id: CompId) -> CompHandle {
         let component = self.components.get(id.0);
         return CompHandle::new(id, &component.public);
+    }
     /// Will remove a component and its memory from the runtime. May only be
     /// called if the necessary conditions for destruction have been met.
     pub(crate) fn destroy_component(&self, key: CompKey) {
         dbg_code!({
             let component = self.get_component(key);
             debug_assert!(component.exiting);
             debug_assert_eq!(component.public.num_handles.load(Ordering::Acquire), 0);
         });
         self.decrement_active_components();
         self.components.destroy(key.0);
+    }
     // Tracking number of active interfaces and the active components
     #[inline]
     fn increment_active_components(&self) {
         let _old_val = self.active_elements.fetch_add(1, Ordering::AcqRel);
         debug_assert!(_old_val > 0); // can only create a component from a API/component, so can never be 0.
+    }
     fn decrement_active_components(&self) {
         let old_val = self.active_elements.fetch_sub(1, Ordering::AcqRel);
         debug_assert!(old_val > 0); // something wrong with incr/decr logic
         let new_val = old_val - 1;
         if new_val == 0 {
             // Just to be sure, in case the last thing that gets destroyed is an
             // API instead of a thread.
             let _lock = self.work_queue.lock();
             self.work_condvar.notify_all();
+        }
+    }
+}

src/runtime2/scheduler.rs

➞

Show inline comments

 use std::sync::Arc;
 use std::sync::atomic::Ordering;
 use crate::runtime2::poll::PollingClient;
 use super::component::*;
 use super::runtime::*;
 /// Data associated with a scheduler thread
 pub(crate) struct Scheduler {
     runtime: Arc<RuntimeInner>,
     polling: PollingClient,
     scheduler_id: u32,
-    debug_logging: bool,
+    log_level: LogLevel,
+}
 pub(crate) struct SchedulerCtx<'a> {
     pub runtime: &'a RuntimeInner,
     pub polling: &'a PollingClient,
     pub id: u32,
     pub comp: u32,
-    pub logging_enabled: bool,
+    pub log_level: LogLevel,
+}
 impl<'a> SchedulerCtx<'a> {
-    pub fn new(runtime: &'a RuntimeInner, polling: &'a PollingClient, id: u32, logging_enabled: bool) -> Self {
+    pub fn new(runtime: &'a RuntimeInner, polling: &'a PollingClient, id: u32, log_level: LogLevel) -> Self {
         return Self {
             runtime,
             polling,
             id,
             comp: 0,
-            logging_enabled,
+            log_level,
+        }
+    }
     pub(crate) fn log(&self, text: &str) {
         if self.logging_enabled {
     pub(crate) fn debug(&self, text: &str) {
         // TODO: Probably not always use colour
         if self.log_level >= LogLevel::Debug {
             println!("[s:{:02}, c:{:03}] \x1b[0;34m{}\x1b[0m", self.id, self.comp, text);
+        }
+    }
     pub(crate) fn info(&self, text: &str) {
         if self.log_level >= LogLevel::Info {
             println!("[s:{:02}, c:{:03}] {}", self.id, self.comp, text);
+        }
+    }
     pub(crate) fn error(&self, text: &str) {
         // TODO: Probably not always use colour
         println!("[s:{:02}, c:{:03}] \x1b[0;31m{}\x1b[0m", self.id, self.comp, text);
+    }
+}
 impl Scheduler {
     // public interface to thread
     pub fn new(runtime: Arc<RuntimeInner>, polling: PollingClient, scheduler_id: u32, debug_logging: bool) -> Self {
         return Scheduler{ runtime, polling, scheduler_id, debug_logging }
     pub fn new(runtime: Arc<RuntimeInner>, polling: PollingClient, scheduler_id: u32, log_level: LogLevel) -> Self {
         return Scheduler{ runtime, polling, scheduler_id, log_level }
+    }
     pub fn run(&mut self) {
-        let mut scheduler_ctx = SchedulerCtx::new(&*self.runtime, &self.polling, self.scheduler_id, self.debug_logging);
+        let mut scheduler_ctx = SchedulerCtx::new(&*self.runtime, &self.polling, self.scheduler_id, self.log_level);
         'run_loop: loop {
             // Wait until we have something to do (or need to quit)
             let comp_key = self.runtime.take_work();
             if comp_key.is_none() {
                 break 'run_loop;
+            }
             let comp_key = comp_key.unwrap();
             let component = self.runtime.get_component(comp_key);
             scheduler_ctx.comp = comp_key.0;
             // Run the component until it no longer indicates that it needs to
             // be re-executed immediately.
             let mut new_scheduling = CompScheduling::Immediate;
             while let CompScheduling::Immediate = new_scheduling {
                 while let Some(message) = component.inbox.pop() {
                     component.component.handle_message(&mut scheduler_ctx, &mut component.ctx, message);
+                }
                 new_scheduling = component.component.run(&mut scheduler_ctx, &mut component.ctx);
+            }
             // Handle the new scheduling
             match new_scheduling {
                 CompScheduling::Immediate => unreachable!(),
                 CompScheduling::Requeue => { self.runtime.enqueue_work(comp_key); },
                 CompScheduling::Sleep => { self.mark_component_as_sleeping(comp_key, component); },
                 CompScheduling::Exit => {
                     component.component.on_shutdown(&scheduler_ctx);
                     self.mark_component_as_exiting(&scheduler_ctx, component);
+                }
+            }
+        }
+    }
     // local utilities
     /// Marks component as sleeping, if after marking itself as sleeping the
     /// inbox contains messages then the component will be immediately
     /// rescheduled. After calling this function the component should not be
     /// executed anymore.
     fn mark_component_as_sleeping(&self, key: CompKey, component: &mut RuntimeComp) {
         debug_assert_eq!(key.downgrade(), component.ctx.id); // make sure component matches key
         debug_assert_eq!(component.public.sleeping.load(Ordering::Acquire), false); // we're executing it, so it cannot be sleeping
         component.public.sleeping.store(true, Ordering::Release);
         if component.inbox.can_pop() {
             let should_reschedule = component.public.sleeping
                 .compare_exchange(true, false, Ordering::AcqRel, Ordering::Relaxed)
                 .is_ok();
             if should_reschedule {
                 self.runtime.enqueue_work(key);
+            }
+        }
+    }
     /// Marks the component as exiting by removing the reference it holds to
     /// itself. Afterward the component will enter "normal" sleeping mode (if it
     /// has not yet been destroyed)
     fn mark_component_as_exiting(&self, sched_ctx: &SchedulerCtx, component: &mut RuntimeComp) {
         // If we didn't yet decrement our reference count, do so now
         let comp_key = unsafe{ component.ctx.id.upgrade() };
         if !component.exiting {
             component.exiting = true;
             let old_count = component.public.num_handles.fetch_sub(1, Ordering::AcqRel);
             let new_count = old_count - 1;
             if new_count == 0 {
                 sched_ctx.runtime.destroy_component(comp_key);
                 return;
+            }
+        }
         // Enter "regular" sleeping mode
         self.mark_component_as_sleeping(comp_key, component);
+    }
+}
@@ \ No newline at end of file @@

0 comments (0 inline, 0 general)