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

Changeset - aefbf606d736

Parent rev.

Child rev.

[Not reviewed]

0 4 0

mh - 3 years ago 2022-03-30 13:14:10
contact@maxhenger.nl

Factor out execution state and control message handling

4 files changed with 324 insertions and 228 deletions:

src/protocol/eval/value.rs

src/runtime2/component/component.rs

244

src/runtime2/component/component_ip.rs

src/runtime2/component/component_pdl.rs

224

0 comments (0 inline, 0 general)

src/protocol/eval/value.rs

➞

Show inline comments

@@ @@ -174,38 +174,48 @@ pub struct ValueGroup { @@
     pub(crate) values: Vec<Value>,
     pub(crate) regions: Vec<Vec<Value>>
+}
 impl ValueGroup {
     pub(crate) fn new_stack(values: Vec<Value>) -> Self {
         debug_assert!(values.iter().all(|v| v.get_heap_pos().is_none()));
         Self{
             values,
             regions: Vec::new(),
+        }
+    }
     pub(crate) fn from_store(store: &Store, values: &[Value]) -> Self {
         let mut group = ValueGroup{
             values: Vec::with_capacity(values.len()),
             regions: Vec::with_capacity(values.len()), // estimation
         };
         for value in values {
             let transferred = group.retrieve_value(value, store);
             group.values.push(transferred);
+        }
         group
+    }
     /// Creates a clone of the value group, but leaves the memory inside of the
     /// ValueGroup vectors allocated.
     pub(crate) fn take(&mut self) -> ValueGroup {
         let cloned = self.clone();
         self.values.clear();
         self.regions.clear();
         return cloned;
+    }
     /// Transfers a provided value from a store into a local value with its
     /// heap allocations (if any) stored in the ValueGroup. Calling this
     /// function will not store the returned value in the `values` member.
     fn retrieve_value(&mut self, value: &Value, from_store: &Store) -> Value {
         let value = from_store.maybe_read_ref(value);
         if let Some(heap_pos) = value.get_heap_pos() {
             // Value points to a heap allocation, so transfer the heap values
             // internally.
             let from_region = &from_store.heap_regions[heap_pos as usize].values;
             let mut new_region = Vec::with_capacity(from_region.len());
             for value in from_region {
                 let transferred = self.retrieve_value(value, from_store);

src/runtime2/component/component.rs

➞

Show inline comments

-use crate::protocol::eval::*;
+use crate::protocol::eval::{Prompt, EvalError, ValueGroup, PortId as EvalPortId};
 use crate::protocol::*;
 use crate::runtime2::*;
 use crate::runtime2::communication::*;
 use super::{CompCtx, CompPDL};
 use super::component_context::*;
 use super::component_ip::*;
 use super::control_layer::*;
 use super::consensus::*;
 pub enum CompScheduling {
     Immediate,
     Requeue,
     Sleep,
     Exit,
+}
 /// Generic representation of a component (as viewed by a scheduler).
 pub(crate) trait Component {
     /// Called if the component is created by another component and the messages
     /// are being transferred between the two.
     fn adopt_message(&mut self, comp_ctx: &mut CompCtx, message: DataMessage);
     /// Called if the component receives a new message. The component is
     /// responsible for deciding where that messages goes.
     fn handle_message(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx, message: Message);
     /// Called if the component's routine should be executed. The return value
     /// can be used to indicate when the routine should be run again.
     fn run(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx) -> Result<CompScheduling, EvalError>;
+}
 /// Representation of the generic operating mode of a component.
 #[derive(Debug, Copy, Clone, PartialEq, Eq)]
 pub(crate) enum CompMode {
     NonSync, // not in sync mode
     Sync, // in sync mode, can interact with other components
     SyncEnd, // awaiting a solution, i.e. encountered the end of the sync block
     BlockedGet, // blocked because we need to receive a message on a particular port
     BlockedPut, // component is blocked because the port is blocked
     BlockedSelect, // waiting on message to complete the select statement
     StartExit, // temporary state: if encountered then we start the shutdown process
     BusyExit, // temporary state: waiting for Acks for all the closed ports
     Exit, // exiting: shutdown process started, now waiting until the reference count drops to 0
+}
 impl CompMode {
     pub(crate) fn is_in_sync_block(&self) -> bool {
         use CompMode::*;
         match self {
             Sync | SyncEnd | BlockedGet | BlockedPut | BlockedSelect => true,
             NonSync | StartExit | BusyExit | Exit => false,
+        }
+    }
+}
 /// Component execution state: the execution mode along with some descriptive
 /// fields. Fields are public for ergonomic reasons, use member functions when
 /// appropriate.
 pub(crate) struct CompExecState {
     pub mode: CompMode,
     pub mode_port: PortId, // valid if blocked on a port (put/get)
     pub mode_value: ValueGroup, // valid if blocked on a put
+}
 impl CompExecState {
     pub(crate) fn new() -> Self {
         return Self{
             mode: CompMode::NonSync,
             mode_port: PortId::new_invalid(),
             mode_value: ValueGroup::default(),
+        }
+    }
     pub(crate) fn set_as_blocked_get(&mut self, port: PortId) {
         self.mode = CompMode::BlockedGet;
         self.mode_port = port;
         debug_assert!(self.mode_value.values.is_empty());
+    }
     pub(crate) fn is_blocked_on_get(&self, port: PortId) -> bool {
         return
             self.mode == CompMode::BlockedGet &&
             self.mode_port == port;
+    }
     pub(crate) fn set_as_blocked_put(&mut self, port: PortId, value: ValueGroup) {
         self.mode = CompMode::BlockedPut;
         self.mode_port = port;
         self.mode_value = value;
+    }
     pub(crate) fn is_blocked_on_put(&self, port: PortId) -> bool {
         return
             self.mode == CompMode::BlockedPut &&
             self.mode_port == port;
+    }
+}
 /// Creates a new component based on its definition. Meaning that if it is a
 /// user-defined component then we set up the PDL code state. Otherwise we
 /// construct a custom component. This does NOT take care of port and message
 /// management.
 pub(crate) fn create_component(
     protocol: &ProtocolDescription,
     definition_id: ProcedureDefinitionId, type_id: TypeId,
     arguments: ValueGroup, num_ports: usize
 ) -> Box<dyn Component> {
     let definition = &protocol.heap[definition_id];
     debug_assert!(definition.kind == ProcedureKind::Primitive || definition.kind == ProcedureKind::Composite);
@@ @@ -48,22 +121,192 @@ pub(crate) fn create_component( @@
         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)
 // -----------------------------------------------------------------------------
 /// 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
 ) {
     match message.content {
         ControlMessageContent::Ack => {
             default_handle_ack(control, message.id, sched_ctx, comp_ctx);
         },
         ControlMessageContent::BlockPort(port_id) => {
             // One of our messages was accepted, but the port should be
             // blocked.
             let port_handle = comp_ctx.get_port_handle(port_id);
             let port_info = comp_ctx.get_port(port_handle);
             debug_assert_eq!(port_info.kind, PortKind::Putter);
             if port_info.state == PortState::Open {
                 // only when open: we don't do this when closed, and we we don't do this if we're blocked due to peer changes
                 comp_ctx.set_port_state(port_handle, PortState::BlockedDueToFullBuffers);
+            }
         },
         ControlMessageContent::ClosePort(port_id) => {
             // Request to close the port. We immediately comply and remove
             // the component handle as well
             let port_handle = comp_ctx.get_port_handle(port_id);
             let peer_comp_id = comp_ctx.get_port(port_handle).peer_comp_id;
             let 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) {
                 default_handle_ack(control, control_id, sched_ctx, comp_ctx);
             } else {
                 default_send_ack(message.id, peer_handle, sched_ctx, comp_ctx);
                 comp_ctx.remove_peer(sched_ctx, port_handle, peer_comp_id, false); // do not remove if closed
                 comp_ctx.set_port_state(port_handle, PortState::Closed); // now set to closed
+            }
         },
         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);
+        }
+    }
+}
 // -----------------------------------------------------------------------------
 // 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, 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, 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
         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, 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_ip.rs

➞

Show inline comments

@@ @@ -4,39 +4,49 @@ use super::*; @@
 use super::component::*;
 /// TODO: Temporary component to figure out what to do with custom components.
 ///     This component sends random numbers between two u32 limits
 pub struct ComponentRandomU32 {
     output_port_id: PortId,
     random_minimum: u32,
     random_maximum: u32,
+}
 impl Component for ComponentRandomU32 {
     fn adopt_message(&mut self, _comp_ctx: &mut CompCtx, _message: DataMessage) {
         unreachable!("should not adopt messages");
         // 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) {
         todo!()
         match message {
             Message::Data(message) => unreachable!(),
             Message::Sync(message) => {
             },
             Message::Control(message) => {
+            }
+        }
+    }
     fn run(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx) -> Result<CompScheduling, EvalError> {
         todo!()
+    }
+}
 impl ComponentRandomU32 {
     pub(crate) fn new(arguments: ValueGroup) -> Self {
         debug_assert_eq!(arguments.values.len(), 3);
         debug_assert!(arguments.regions.is_empty());
         let port_id = port_id_from_eval(arguments.values[0].as_port_id());
         let minimum = arguments.values[1].as_uint32();
         let maximum = arguments.values[2].as_uint32();
-        return ComponentRandomU32{
+        return Self{
             output_port_id: port_id,
             random_minimum: minimum,
             random_maximum: maximum,
+        }
+    }
+}
@@ \ 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::scheduler::SchedulerCtx;
 use crate::runtime2::communication::*;
 use super::component::*;
 use super::component::{
     self,
     CompExecState, Component, CompScheduling, CompMode,
     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,
+}
@@ @@ -75,48 +79,24 @@ impl RunContext for ExecCtx { @@
+        }
+    }
     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!(),
+        }
+    }
+}
 #[derive(Debug, Copy, Clone, PartialEq, Eq)]
 pub(crate) enum Mode {
     NonSync, // not in sync mode
     Sync, // in sync mode, can interact with other components
     SyncEnd, // awaiting a solution, i.e. encountered the end of the sync block
     BlockedGet, // blocked because we need to receive a message on a particular port
     BlockedPut, // component is blocked because the port is blocked
     BlockedSelect, // waiting on message to complete the select statement
     StartExit, // temporary state: if encountered then we start the shutdown process
     BusyExit, // temporary state: waiting for Acks for all the closed ports
     Exit, // exiting: shutdown process started, now waiting until the reference count drops to 0
+}
 impl Mode {
     fn is_in_sync_block(&self) -> bool {
         use Mode::*;
         match self {
             Sync | SyncEnd | BlockedGet | BlockedPut | BlockedSelect => true,
             NonSync | StartExit | BusyExit | Exit => false,
+        }
+    }
+}
 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>,
+}
@@ @@ -217,28 +197,26 @@ impl SelectState { @@
+        }
         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 mode: Mode,
     pub mode_port: PortId, // when blocked on a port
     pub mode_value: ValueGroup, // when blocked on a put
     select: SelectState,
     pub exec_state: CompExecState,
     pub 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>,
+}
@@ @@ -260,194 +238,192 @@ impl Component for CompPDL { @@
             let mut target = sched_ctx.runtime.get_component_public(new_target);
             target.send_message(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) => {
                 self.handle_incoming_control_message(sched_ctx, comp_ctx, message);
                 component::default_handle_control_message(
                     &mut self.exec_state, &mut self.control, &mut self.consensus,
                     message, sched_ctx, comp_ctx
                 );
             },
             Message::Sync(message) => {
                 self.handle_incoming_sync_message(sched_ctx, comp_ctx, message);
+            }
+        }
+    }
     fn run(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx) -> Result<CompScheduling, EvalError> {
         use EvalContinuation as EC;
         sched_ctx.log(&format!("Running component (mode: {:?})", self.mode));
+        sched_ctx.log(&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.mode {
             Mode::NonSync | Mode::Sync => {
         match self.exec_state.mode {
             CompMode::NonSync | CompMode::Sync => {
                 // continue and run PDL code
             },
-            Mode::SyncEnd | Mode::BlockedGet | Mode::BlockedPut | Mode::BlockedSelect => {
+            CompMode::SyncEnd | CompMode::BlockedGet | CompMode::BlockedPut | CompMode::BlockedSelect => {
                 return Ok(CompScheduling::Sleep);
+            }
-            Mode::StartExit => {
+            CompMode::StartExit => {
                 self.handle_component_exit(sched_ctx, comp_ctx);
                 return Ok(CompScheduling::Immediate);
             },
-            Mode::BusyExit => {
+            CompMode::BusyExit => {
                 if self.control.has_acks_remaining() {
                     return Ok(CompScheduling::Sleep);
                 } else {
-                    self.mode = Mode::Exit;
+                    self.exec_state.mode = CompMode::Exit;
                     return Ok(CompScheduling::Exit);
+                }
             },
-            Mode::Exit => {
+            CompMode::Exit => {
                 return Ok(CompScheduling::Exit);
+            }
+        }
         let run_result = self.execute_prompt(&sched_ctx)?;
         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 => {
-                debug_assert_eq!(self.mode, Mode::Sync);
+                debug_assert_eq!(self.exec_state.mode, CompMode::Sync);
                 self.handle_sync_end(sched_ctx, comp_ctx);
                 return Ok(CompScheduling::Immediate);
             },
             EC::BlockGet(port_id) => {
-                debug_assert_eq!(self.mode, Mode::Sync);
+                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);
                 let port_handle = comp_ctx.get_port_handle(port_id);
                 let port_index = comp_ctx.get_port_index(port_handle);
                 if let Some(message) = &self.inbox_main[port_index] {
                     // Check if we can actually receive the message
                     if self.consensus.try_receive_data_message(sched_ctx, comp_ctx, message) {
                         // Message was received. Make sure any blocked peers and
                         // pending messages are handled.
                         let message = self.inbox_main[port_index].take().unwrap();
                         self.handle_received_data_message(sched_ctx, comp_ctx, port_handle);
                         self.exec_ctx.stmt = ExecStmt::PerformedGet(message.content);
                         return Ok(CompScheduling::Immediate);
                     } else {
                         todo!("handle sync failure due to message deadlock");
                         return Ok(CompScheduling::Sleep);
+                    }
                 } else {
                     // We need to wait
                     self.mode = Mode::BlockedGet;
                     self.mode_port = port_id;
                     self.exec_state.set_as_blocked_get(port_id);
                     return Ok(CompScheduling::Sleep);
+                }
             },
             EC::Put(port_id, value) => {
-                debug_assert_eq!(self.mode, Mode::Sync);
+                debug_assert_eq!(self.exec_state.mode, CompMode::Sync);
                 sched_ctx.log(&format!("Putting value {:?}", value));
                 let port_id = port_id_from_eval(port_id);
                 let port_handle = comp_ctx.get_port_handle(port_id);
                 let port_info = comp_ctx.get_port(port_handle);
                 if port_info.state.is_blocked() {
                     self.mode = Mode::BlockedPut;
                     self.mode_port = port_id;
                     self.mode_value = value;
                     self.exec_state.set_as_blocked_put(port_id, value);
                     self.exec_ctx.stmt = ExecStmt::PerformedPut; // prepare for when we become unblocked
                     return Ok(CompScheduling::Sleep);
                 } else {
                     self.send_data_message_and_wake_up(sched_ctx, comp_ctx, port_handle, value);
                     self.exec_ctx.stmt = ExecStmt::PerformedPut;
                     return Ok(CompScheduling::Immediate);
+                }
             },
             EC::SelectStart(num_cases, _num_ports) => {
                 debug_assert_eq!(self.mode, Mode::Sync);
                 self.select.handle_select_start(num_cases);
                 debug_assert_eq!(self.exec_state.mode, CompMode::Sync);
                 self.select_state.handle_select_start(num_cases);
                 return Ok(CompScheduling::Requeue);
             },
             EC::SelectRegisterPort(case_index, port_index, port_id) => {
-                debug_assert_eq!(self.mode, Mode::Sync);
+                debug_assert_eq!(self.exec_state.mode, CompMode::Sync);
                 let port_id = port_id_from_eval(port_id);
-                if let Err(_err) = self.select.register_select_case_port(comp_ctx, case_index, port_index, port_id) {
+                if let Err(_err) = self.select_state.register_select_case_port(comp_ctx, case_index, port_index, port_id) {
                     todo!("handle registering a port multiple times");
+                }
                 return Ok(CompScheduling::Immediate);
             },
             EC::SelectWait => {
                 debug_assert_eq!(self.mode, Mode::Sync);
                 let select_decision = self.select.handle_select_waiting_point(&self.inbox_main, comp_ctx);
                 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.mode = Mode::Sync;
+                    self.exec_state.mode = CompMode::Sync;
                     return Ok(CompScheduling::Immediate);
                 } else {
                     // No decision yet
-                    self.mode = Mode::BlockedSelect;
+                    self.exec_state.mode = CompMode::BlockedSelect;
                     return Ok(CompScheduling::Sleep);
+                }
             },
             // Results that can be returned outside of sync mode
             EC::ComponentTerminated => {
-                self.mode = Mode::StartExit; // next call we'll take care of the exit
+                self.exec_state.mode = CompMode::StartExit; // next call we'll take care of the exit
                 return Ok(CompScheduling::Immediate);
             },
             EC::SyncBlockStart => {
-                debug_assert_eq!(self.mode, Mode::NonSync);
+                debug_assert_eq!(self.exec_state.mode, CompMode::NonSync);
                 self.handle_sync_start(sched_ctx, comp_ctx);
                 return Ok(CompScheduling::Immediate);
             },
             EC::NewComponent(definition_id, type_id, arguments) => {
-                debug_assert_eq!(self.mode, Mode::NonSync);
+                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 Ok(CompScheduling::Requeue);
             },
             EC::NewChannel => {
-                debug_assert_eq!(self.mode, Mode::NonSync);
+                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 Ok(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{
             mode: Mode::NonSync,
             mode_port: PortId::new_invalid(),
             mode_value: ValueGroup::default(),
             select: SelectState::new(),
             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(),
+        }
+    }
@@ @@ -466,67 +442,67 @@ impl CompPDL { @@
         return Ok(step_result)
+    }
     fn handle_sync_start(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx) {
         sched_ctx.log("Component starting sync mode");
         self.consensus.notify_sync_start(comp_ctx);
         for message in self.inbox_main.iter() {
             if let Some(message) = message {
                 self.consensus.handle_new_data_message(comp_ctx, message);
+            }
+        }
         debug_assert_eq!(self.mode, Mode::NonSync);
         self.mode = Mode::Sync;
         debug_assert_eq!(self.exec_state.mode, CompMode::NonSync);
         self.exec_state.mode = CompMode::Sync;
+    }
     /// Handles end of sync. The conclusion to the sync round might arise
     /// immediately (and be handled immediately), or might come later through
     /// messaging. In any case the component should be scheduled again
     /// immediately
     fn handle_sync_end(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx) {
         sched_ctx.log("Component ending sync mode (now waiting for solution)");
         let decision = self.consensus.notify_sync_end(sched_ctx, comp_ctx);
-        self.mode = Mode::SyncEnd;
+        self.exec_state.mode = CompMode::SyncEnd;
         self.handle_sync_decision(sched_ctx, comp_ctx, decision);
+    }
     /// Handles decision from the consensus round. This will cause a change in
     /// the internal `Mode`, such that the next call to `run` can take the
     /// appropriate next steps.
     fn handle_sync_decision(&mut self, sched_ctx: &SchedulerCtx, _comp_ctx: &mut CompCtx, decision: SyncRoundDecision) {
         sched_ctx.log(&format!("Handling sync decision: {:?} (in mode {:?})", decision, self.mode));
         let is_success = match decision {
             SyncRoundDecision::None => {
                 // No decision yet
                 return;
             },
             SyncRoundDecision::Solution => true,
             SyncRoundDecision::Failure => false,
         };
         // If here then we've reached a decision
-        debug_assert_eq!(self.mode, Mode::SyncEnd);
+        debug_assert_eq!(self.exec_state.mode, CompMode::SyncEnd);
         if is_success {
-            self.mode = Mode::NonSync;
+            self.exec_state.mode = CompMode::NonSync;
             self.consensus.notify_sync_decision(decision);
         } else {
-            self.mode = Mode::StartExit;
+            self.exec_state.mode = CompMode::StartExit;
+        }
+    }
     fn handle_component_exit(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx) {
         sched_ctx.log("Component exiting");
         debug_assert_eq!(self.mode, Mode::StartExit);
         self.mode = Mode::BusyExit;
         debug_assert_eq!(self.exec_state.mode, CompMode::StartExit);
         self.exec_state.mode = CompMode::BusyExit;
         // Doing this by index, then retrieving the handle is a bit rediculous,
         // but Rust is being Rust with its borrowing rules.
         for port_index in 0..comp_ctx.num_ports() {
             let port = comp_ctx.get_port_by_index_mut(port_index);
             if port.state == PortState::Closed {
                 // Already closed, or in the process of being closed
                 continue;
+            }
             // Mark as closed
             let port_id = port.self_id;
@@ @@ -548,48 +524,48 @@ impl CompPDL { @@
         let port_info = comp_ctx.get_port(source_port_handle);
         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 = self.consensus.annotate_data_message(comp_ctx, port_info, value);
         peer_info.handle.send_message(sched_ctx, Message::Data(annotated_message), true);
+    }
     /// 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) {
         // Whatever we do, glean information from headers in message
         if self.mode.is_in_sync_block() {
+        if self.exec_state.mode.is_in_sync_block() {
             self.consensus.handle_new_data_message(comp_ctx, &message);
+        }
         // Check if we can insert it directly into the storage associated with
         // the port
         let target_port_id = message.data_header.target_port;
         let port_handle = comp_ctx.get_port_handle(target_port_id);
         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);
             // After direct insertion, check if this component's execution is
             // blocked on receiving a message on that port
             debug_assert!(!comp_ctx.get_port(port_handle).state.is_blocked()); // because we could insert directly
-            if self.mode == Mode::BlockedGet && self.mode_port == target_port_id {
+            if self.exec_state.is_blocked_on_get(target_port_id) {
                 // We were indeed blocked
                 self.mode = Mode::Sync;
                 self.mode_port = PortId::new_invalid();
             } else if self.mode == Mode::BlockedSelect {
                 let select_decision = self.select.handle_updated_inbox(&self.inbox_main, comp_ctx);
                 self.exec_state.mode = CompMode::Sync;
                 self.exec_state.mode_port = PortId::new_invalid();
             } else 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.mode = Mode::Sync;
+                    self.exec_state.mode = CompMode::Sync;
+                }
+            }
             return;
+        }
         // The direct inbox is full, so the port will become (or was already) blocked
         let port_info = comp_ctx.get_port_mut(port_handle);
         debug_assert!(port_info.state == PortState::Open || port_info.state.is_blocked());
         if port_info.state == PortState::Open {
             comp_ctx.set_port_state(port_handle, PortState::BlockedDueToFullBuffers);
@@ @@ -627,201 +603,58 @@ impl CompPDL { @@
+        }
         // Did not have any more messages. So if we were blocked, then we need
         // to send the "unblock" message.
         if port_info.state == PortState::BlockedDueToFullBuffers {
             comp_ctx.set_port_state(port_handle, PortState::Open);
             let (peer_handle, message) = self.control.cancel_port_blocking(comp_ctx, port_handle);
             let peer_info = comp_ctx.get_peer(peer_handle);
             peer_info.handle.send_message(sched_ctx, Message::Control(message), true);
+        }
+    }
     fn handle_incoming_control_message(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, message: ControlMessage) {
         // Little local utility to send an Ack
         fn send_control_ack_message(sched_ctx: &SchedulerCtx, comp_ctx: &CompCtx, causer_id: ControlId, peer_handle: LocalPeerHandle) {
             let peer_info = comp_ctx.get_peer(peer_handle);
             peer_info.handle.send_message(sched_ctx, Message::Control(ControlMessage{
                 id: causer_id,
                 sender_comp_id: comp_ctx.id,
                 target_port_id: None,
                 content: ControlMessageContent::Ack,
             }), true);
+        }
         // Handle the content of the control message, and optionally Ack it
         match message.content {
             ControlMessageContent::Ack => {
                 self.handle_ack(sched_ctx, comp_ctx, message.id);
             },
             ControlMessageContent::BlockPort(port_id) => {
                 // On of our messages was accepted, but the port should be
                 // blocked.
                 let port_handle = comp_ctx.get_port_handle(port_id);
                 let port_info = comp_ctx.get_port(port_handle);
                 debug_assert_eq!(port_info.kind, PortKind::Putter);
                 if port_info.state == PortState::Open {
                     // only when open: we don't do this when closed, and we we don't do this if we're blocked due to peer changes
                     comp_ctx.set_port_state(port_handle, PortState::BlockedDueToFullBuffers);
+                }
             },
             ControlMessageContent::ClosePort(port_id) => {
                 // Request to close the port. We immediately comply and remove
                 // the component handle as well
                 let port_handle = comp_ctx.get_port_handle(port_id);
                 let peer_comp_id = comp_ctx.get_port(port_handle).peer_comp_id;
                 let 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) = self.control.has_close_port_entry(port_handle, comp_ctx) {
                     self.handle_ack(sched_ctx, comp_ctx, control_id);
                 } else {
                     send_control_ack_message(sched_ctx, comp_ctx, message.id, peer_handle);
                     comp_ctx.remove_peer(sched_ctx, port_handle, peer_comp_id, false); // do not remove if closed
                     comp_ctx.set_port_state(port_handle, PortState::Closed); // now set to closed
+                }
             },
             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 {
                     self.handle_unblock_port_instruction(sched_ctx, comp_ctx, port_handle);
+                }
             },
             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);
                 send_control_ack_message(sched_ctx, comp_ctx, message.id, peer_handle);
             },
             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);
                 self.handle_unblock_port_instruction(sched_ctx, comp_ctx, port_handle);
+            }
+        }
+    }
     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);
         self.handle_sync_decision(sched_ctx, comp_ctx, decision);
+    }
     /// Little helper that notifies the control layer of an `Ack`, and takes the
     /// appropriate subsequent action
     fn handle_ack(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, control_id: ControlId) {
         let mut to_ack = control_id;
         loop {
             let (action, new_to_ack) = self.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, 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,
+            }
+        }
+    }
     // -------------------------------------------------------------------------
     // Handling ports
     // -------------------------------------------------------------------------
     /// Unblocks a port, potentially continuing execution of the component, in
     /// response to a message that told us to unblock a previously blocked
     fn handle_unblock_port_instruction(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, port_handle: LocalPortHandle) {
         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 self.mode == Mode::BlockedPut && port_id == self.mode_port {
             // We were blocked on the port that just became unblocked, so
             // send the message.
             debug_assert_eq!(port_info.kind, PortKind::Putter);
             let mut replacement = ValueGroup::default();
             std::mem::swap(&mut replacement, &mut self.mode_value);
             self.send_data_message_and_wake_up(sched_ctx, comp_ctx, port_handle, replacement);
             self.mode = Mode::Sync;
             self.mode_port = PortId::new_invalid();
+        }
+    }
     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();
         // TODO: @Nocommit
         let other_proc = &sched_ctx.runtime.protocol.heap[definition_id];
         let self_proc = &sched_ctx.runtime.protocol.heap[self.prompt.frames[0].definition];
         let reservation = sched_ctx.runtime.start_create_pdl_component();
         let mut created_ctx = CompCtx::new(&reservation);
         println!(
             "DEBUG: Comp '{}' is creating comp '{}' at ID {:?}",
             self_proc.identifier.value.as_str(), other_proc.identifier.value.as_str(),
             reservation.id()
             "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(
@@ @@ -892,25 +725,25 @@ impl CompPDL { @@
                 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 = create_component(&sched_ctx.runtime.protocol, definition_id, type_id, arguments, total_num_ports);
+        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,
         );
         // 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);

0 comments (0 inline, 0 general)