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

Changeset - 58fb633d3fa1

Parent rev.

Child rev.

[Not reviewed]

0 5 0

MH - 3 years ago 2022-05-06 18:03:23
contact@maxhenger.nl

WIP on figuring out protocol for port transmission

5 files changed with 87 insertions and 51 deletions:

src/protocol/eval/mod.rs

src/runtime2/communication.rs

src/runtime2/component/component.rs

src/runtime2/component/component_context.rs

src/runtime2/component/component_pdl.rs

0 comments (0 inline, 0 general)

src/protocol/eval/mod.rs

➞

Show inline comments

 /// eval
 ///
 /// Evaluator of the generated AST. Note that we use some misappropriated terms
 /// to describe where values live and what they do. This is a temporary
 /// implementation of an evaluator until some kind of appropriate bytecode or
 /// machine code is generated.
 ///
 /// Code is always executed within a "frame". For Reowolf the first frame is
 /// usually an executed component. All subsequent frames are function calls.
 /// Simple values live on the "stack". Each variable/parameter has a place on
 /// the stack where its values are stored. If the value is not a primitive, then
 /// its value will be stored in the "heap". Expressions are treated differently
 /// and use a separate "stack" for their evaluation.
 ///
 /// Since this is a value-based language, most values are copied. One has to be
 /// careful with values that reside in the "heap" and make sure that copies are
 /// properly removed from the heap..
 ///
 /// Just to reiterate: this is a temporary wasteful implementation. A proper
 /// implementation would fully fill out the type table with alignment/size/
 /// offset information and lay out bytecode.
 pub(crate) mod value;
 pub(crate) mod store;
 pub(crate) mod executor;
 pub(crate) mod error;
 pub use error::EvalError;
 pub use value::{PortId, Value, ValueGroup};
+pub use value::{PortId, Value, ValueId, ValueGroup};
 pub use executor::{EvalContinuation, EvalResult, Prompt};

src/runtime2/communication.rs

➞

Show inline comments

 use crate::protocol::eval::*;
 use crate::protocol::eval::value::ValueId;
 use super::runtime::*;
 use super::component::*;
 // -----------------------------------------------------------------------------
 // Generic types
 // -----------------------------------------------------------------------------
 #[derive(Debug, Copy, Clone, PartialEq, Eq)]
 pub struct PortId(pub u32);
 impl PortId {
     /// This value is not significant, it is chosen to make debugging easier: a
     /// very large port number is more likely to shine a light on bugs.
     pub fn new_invalid() -> Self {
         return Self(u32::MAX);
+    }
+}
 pub struct Channel {
     pub putter_id: PortId,
     pub getter_id: PortId,
+}
 // -----------------------------------------------------------------------------
 // Data messages
 // -----------------------------------------------------------------------------
 #[derive(Debug)]
 pub struct DataMessage {
     pub data_header: MessageDataHeader,
     pub sync_header: MessageSyncHeader,
     pub content: ValueGroup,
     pub ports: Vec<TransmittedPort>,
+}
 #[derive(Debug)]
 pub enum PortAnnotationKind {
     Getter(PortAnnotationGetter),
     Putter(PortAnnotationPutter),
+}
 #[derive(Debug)]
 pub struct PortAnnotationGetter {
     pub self_comp_id: CompId,
     pub self_port_id: PortId,
     pub peer_comp_id: CompId,
     pub peer_port_id: PortId,
+}
 #[derive(Debug)]
 pub struct PortAnnotationPutter {
     pub self_comp_id: CompId,
     pub self_port_id: PortId,
+}
 #[derive(Debug)]
 pub struct MessageDataHeader {
     pub expected_mapping: Vec<(PortAnnotationKind, Option<u32>)>,
     pub new_mapping: u32,
     pub source_port: PortId,
     pub target_port: PortId,
+}
 #[derive(Debug)]
 pub struct TransmittedPort {
     location: ValueId, // within `content`
     messages: Vec<DataMessage>, // owned by previous component
     peer_comp: CompId,
     peer_port: PortId,
+}
 // -----------------------------------------------------------------------------
 // Sync messages
 // -----------------------------------------------------------------------------
 #[derive(Debug)]
 pub struct SyncMessage {
     pub sync_header: MessageSyncHeader,
     pub content: SyncMessageContent,
+}
 #[derive(Debug)]
 pub enum SyncLocalSolutionEntry {
     Putter(SyncSolutionPutterPort),
     Getter(SyncSolutionGetterPort),
+}
 pub type SyncLocalSolution = Vec<SyncLocalSolutionEntry>;
 /// Getter port in a solution. Upon receiving a message it is certain about who
 /// its peer is.
 #[derive(Debug)]
 pub struct SyncSolutionGetterPort {
     pub self_comp_id: CompId,
     pub self_port_id: PortId,
     pub peer_comp_id: CompId,
     pub peer_port_id: PortId,
     pub mapping: u32,
     pub failed: bool,
+}
 /// Putter port in a solution. A putter may not be certain about who its peer
 /// component/port is.
 #[derive(Debug)]
 pub struct SyncSolutionPutterPort {
     pub self_comp_id: CompId,
     pub self_port_id: PortId,
     pub mapping: u32,
     pub failed: bool,
+}
 #[derive(Debug)]
 pub struct SyncSolutionChannel {
     pub putter: Option<SyncSolutionPutterPort>,
     pub getter: Option<SyncSolutionGetterPort>,
+}
 #[derive(Debug, Copy, Clone, PartialEq, Eq)]
 pub enum SyncRoundDecision {
     None,
     Solution,
     Failure,
+}
 #[derive(Debug)]
 pub struct SyncPartialSolution {
     pub channel_mapping: Vec<SyncSolutionChannel>,
     pub decision: SyncRoundDecision,
+}
 impl Default for SyncPartialSolution {
     fn default() -> Self {
         return Self{
             channel_mapping: Vec::new(),
             decision: SyncRoundDecision::None,
+        }
+    }
+}
 #[derive(Debug)]
 pub enum SyncMessageContent {
     NotificationOfLeader,
     LocalSolution(CompId, SyncLocalSolution), // local solution of the specified component
     PartialSolution(SyncPartialSolution), // partial solution of multiple components
     GlobalSolution,
     GlobalFailure,
+}
 // -----------------------------------------------------------------------------
 // Control messages
 // -----------------------------------------------------------------------------
 #[derive(Debug)]
 pub struct ControlMessage {
     pub(crate) id: ControlId,
     pub sender_comp_id: CompId,
     pub target_port_id: Option<PortId>,
     pub content: ControlMessageContent,
+}
 /// Content of a control message. If the content refers to a port then the
 /// `target_port_id` field is the one that it refers to.
 #[derive(Copy, Clone, Debug)]
 pub enum ControlMessageContent {
     Ack,
     BlockPort,
     UnblockPort,

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::eval::{Prompt, EvalError, ValueGroup, Value, ValueId, PortId as EvalPortId};
 use crate::protocol::*;
 use crate::runtime2::*;
 use crate::runtime2::communication::*;
 use crate::runtime2::component::component_pdl::find_ports_in_value_group;
 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
     BlockedPutPorts,// blocked because we're waiting to send a data message containing ports
     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,
+            Sync | SyncEnd | BlockedGet | BlockedPut | BlockedSelect | BlockedPutPorts => 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,
+            NonSync | Sync | SyncEnd | BlockedGet | BlockedPut | BlockedSelect | BlockedPutPorts => 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.is_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 {
         // Check if there are any ports that are being transmitted
         let mut ports = Vec::new();
         find_ports_in_value_group(&value, &mut ports);
         if !ports.is_empty() {
             for (value_location, port_id) in ports {
                 let transmitted_port_handle = comp_ctx.get_port_handle(port_id);
                 let transmitted_port = comp_ctx.get_port(transmitted_port_handle);
                 if transmitted_port.state.is_set(PortStateFlag::Transmitted) {
                     // Note: We could also attach where the port has been
                     //  transferred
                     return Err((
                         port_info.last_instruction,
                         String::from("Cannot send this message, as it contains a previously transmitted port")
                     ));
+                }
                 // Prepare ack for PPC
                 // Prepare PPC message
+            }
+        }
         // 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_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);
         if port_info.state.is_open() {
             port_info.state.set(PortStateFlag::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_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;
     if port_info.state.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)
         ));
+    }
@@ @@ -707,121 +733,167 @@ pub(crate) fn default_handle_sync_decision( @@
+}
 /// 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_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_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_recently_unblocked_port(
     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()); // should have been done by the caller
     if exec_state.is_blocked_on_put(port_id) {
         // Return to the regular execution mode
         exec_state.mode = CompMode::Sync;
         exec_state.mode_port = PortId::new_invalid();
         // 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_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 };
+}
 /// 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<(ValueId, PortId)>) {
     // Helper to check a value for a port and recurse if needed.
     fn find_port_in_value(group: &ValueGroup, value: &Value, value_location: ValueId, ports: &mut Vec<(ValueId, 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.1 {
                         // Already added
                         return;
+                    }
+                }
                 ports.push((value_location, 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 (value_index, embedded_value) in heap_region.iter().enumerate() {
                     let value_location = ValueId::Heap(*heap_pos, value_index as u32);
                     find_port_in_value(group, embedded_value, value_location, ports);
+                }
             },
             _ => {}, // values we don't care about
+        }
+    }
     // Clear the ports, then scan all the available values
     ports.clear();
     for (value_index, value) in &value_group.values.iter().enumerate() {
         find_port_in_value(value_group, value, ValueId::Stack(value_index as u32), ports);
+    }
+}
@@ \ No newline at end of file @@

src/runtime2/component/component_context.rs

➞

Show inline comments

 use std::fmt::{Debug, Formatter, Result as FmtResult};
 use crate::runtime2::scheduler::*;
 use crate::runtime2::runtime::*;
 use crate::runtime2::communication::*;
 use crate::protocol::ExpressionId;
 /// Helper struct to remember when the last operation on the port took place.
 #[derive(Debug, PartialEq, Copy, Clone)]
 pub enum PortInstruction {
     None,
     NoSource,
     SourceLocation(ExpressionId),
 }
 /// Directionality of a port
 #[derive(Debug, PartialEq, Eq, Clone, Copy)]
 pub enum PortKind {
     Putter,
     Getter,
 }
 /// Bitflags for port
 // TODO: Incorporate remaining flags from `Port` struct
 #[repr(u32)]
 #[derive(Debug, Copy, Clone)]
 pub enum PortStateFlag {
     Closed = 0x01, // If not closed, then the port is open
     BlockedDueToPeerChange = 0x02, // busy changing peers, hence use of port is temporarily blocked
     BlockedDueToFullBuffers = 0x04,
     Transmitted, // Transmitted, so cannot be used anymore
 }
 #[derive(Copy, Clone)]
 pub struct PortState {
     flags: u32
 }
 impl PortState {
     pub(crate) fn new() -> PortState {
         return PortState{ flags: 0 }
     }
     // high-level
     #[inline]
     pub fn is_open(&self) -> bool {
         return !self.is_closed();
     }
     #[inline]
     pub fn is_closed(&self) -> bool {
         return self.is_set(PortStateFlag::Closed);
     }
     #[inline]
     pub fn is_blocked(&self) -> bool {
         return
             self.is_set(PortStateFlag::BlockedDueToPeerChange) ||
             self.is_set(PortStateFlag::BlockedDueToFullBuffers);
     }
     // lower-level utils
     #[inline]
     pub fn set(&mut self, flag: PortStateFlag) {
         self.flags |= flag as u32;
     }
     #[inline]
     pub fn clear(&mut self, flag: PortStateFlag) {
         self.flags &= !(flag as u32);
     }
     #[inline]
     pub const fn is_set(&self, flag: PortStateFlag) -> bool {
         return (self.flags & (flag as u32)) != 0;
     }
 }
 impl Debug for PortState {
     fn fmt(&self, f: &mut Formatter<'_>) -> FmtResult {
         use PortStateFlag::*;
         let mut s = f.debug_struct("PortState");
         for (flag_name, flag_value) in &[
             ("closed", Closed),
             ("blocked_peer_change", BlockedDueToPeerChange),
             ("blocked_full_buffers", BlockedDueToFullBuffers)
         ] {
             s.field(flag_name, &self.is_set(*flag_value));
         }
         return s.finish();
     }
 }
 #[derive(Debug)]
 pub struct Port {
     // Identifiers
     pub self_id: PortId,
     pub peer_comp_id: CompId, // eventually consistent
     pub peer_port_id: PortId, // eventually consistent
     // Generic operating state
     pub kind: PortKind,
     pub state: PortState,
     // State tracking for error detection and error handling
     pub last_instruction: PortInstruction, // used during sync round to detect port-closed-during-sync errors
     pub received_message_for_sync: bool, // used during sync round to detect port-closed-before-sync errors
     pub close_at_sync_end: bool, // set during sync round when receiving a port-closed-after-sync message
     pub(crate) associated_with_peer: bool,
 }
 pub struct Peer {
     pub id: CompId,
     pub num_associated_ports: u32,
     pub(crate) handle: CompHandle,
 }
 /// Port and peer management structure. Will keep a local reference counter to
 /// the ports associate with peers, additionally manages the atomic reference
 /// counter associated with the peers' component handles.
 pub struct CompCtx {
     pub id: CompId,
     ports: Vec<Port>,
     peers: Vec<Peer>,
     port_id_counter: u32,
 }
@@ @@ -214,137 +215,132 @@ impl CompCtx { @@
             if peer.num_associated_ports == 0 {
                 let mut peer = self.peers.remove(peer_index);
                 if let Some(key) = peer.handle.decrement_users() {
                     sched_ctx.runtime.destroy_component(key);
                 }
             }
         }
         // If there is a new peer, then set it as the peer associated with the
         // port
         if let Some(peer_id) = new_peer_comp_id {
             let port = &mut self.ports[port_index];
             port.peer_comp_id = peer_id;
             if peer_id != self.id && !port_is_closed {
                 port.associated_with_peer = true;
                 match self.get_peer_index_by_id(peer_id) {
                     Some(index) => {
                         let peer = &mut self.peers[index];
                         peer.num_associated_ports += 1;
                     },
                     None => {
                         let handle = sched_ctx.runtime.get_component_public(peer_id);
                         self.peers.push(Peer {
                             id: peer_id,
                             num_associated_ports: 1,
                             handle
                         })
                     }
                 }
             }
         }
     }
     pub(crate) fn get_port_handle(&self, port_id: PortId) -> LocalPortHandle {
         return LocalPortHandle(port_id);
     }
     // should perhaps be revised, used in main inbox
     pub(crate) fn get_port_index(&self, port_handle: LocalPortHandle) -> usize {
         return self.must_get_port_index(port_handle);
     }
     pub(crate) fn get_peer_handle(&self, peer_id: CompId) -> LocalPeerHandle {
         return LocalPeerHandle(peer_id);
     }
     pub(crate) fn get_port(&self, port_handle: LocalPortHandle) -> &Port {
         let index = self.must_get_port_index(port_handle);
         return &self.ports[index];
     }
     pub(crate) fn get_port_mut(&mut self, port_handle: LocalPortHandle) -> &mut Port {
         let index = self.must_get_port_index(port_handle);
         return &mut self.ports[index];
     }
     pub(crate) fn get_port_by_index_mut(&mut self, index: usize) -> &mut Port {
         return &mut self.ports[index];
     }
     pub(crate) fn get_peer(&self, peer_handle: LocalPeerHandle) -> &Peer {
         let index = self.must_get_peer_index(peer_handle);
         return &self.peers[index];
     }
     pub(crate) fn get_peer_mut(&mut self, peer_handle: LocalPeerHandle) -> &mut Peer {
         let index = self.must_get_peer_index(peer_handle);
         return &mut self.peers[index];
     }
     #[inline]
     pub(crate) fn iter_ports(&self) -> impl Iterator<Item=&Port> {
         return self.ports.iter();
     }
     #[inline]
     pub(crate) fn iter_ports_mut(&mut self) -> impl Iterator<Item=&mut Port> {
         return self.ports.iter_mut();
     }
     #[inline]
     pub(crate) fn iter_peers(&self) -> impl Iterator<Item=&Peer> {
         return self.peers.iter();
     }
     #[inline]
     pub(crate) fn num_ports(&self) -> usize {
         return self.ports.len();
     }
     // -------------------------------------------------------------------------
     // Local utilities
     // -------------------------------------------------------------------------
     #[inline]
     fn requires_peer_reference(port: &Port, self_id: CompId, required_if_closed: bool) -> bool {
         return (!port.state.is_closed() || required_if_closed) && port.peer_comp_id != self_id;
     }
     fn must_get_port_index(&self, handle: LocalPortHandle) -> usize {
         for (index, port) in self.ports.iter().enumerate() {
             if port.self_id == handle.0 {
                 return index;
             }
         }
         unreachable!()
     }
     fn must_get_peer_index(&self, handle: LocalPeerHandle) -> usize {
         for (index, peer) in self.peers.iter().enumerate() {
             if peer.id == handle.0 {
                 return index;
             }
         }
         unreachable!()
     }
     fn get_peer_index_by_id(&self, comp_id: CompId) -> Option<usize> {
         for (index, peer) in self.peers.iter().enumerate() {
             if peer.id == comp_id {
                 return Some(index);
             }
         }
         return None;
     }
     fn take_port_id(&mut self) -> u32 {
         let port_id = self.port_id_counter;
         self.port_id_counter = self.port_id_counter.wrapping_add(1);
         return port_id;
     }
 }
@@ \ No newline at end of file @@

src/runtime2/component/component_pdl.rs

➞

Show inline comments

@@ @@ -681,214 +681,173 @@ impl CompPDL { @@
         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_index = creator_ctx.get_port_index(pair.creator_handle);
             creator_ctx.change_port_peer(sched_ctx, pair.creator_handle, None);
             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;
+                }
+            }
             let created_port_info = component.ctx.get_port(pair.created_handle);
             if created_port_info.peer_comp_id == creator_ctx.id {
                 // This handles the creation of a channel between the creator
                 // component and the newly created component. So if the creator
                 // had a `a -> b` channel, and `b` is moved to the new
                 // component, then `a` needs to set its peer component.
                 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.change_port_peer(sched_ctx, peer_port_handle, Some(component.ctx.id));
+            }
+        }
         // 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_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 @@

0 comments (0 inline, 0 general)