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

Changeset - bf4c0ee5ba65

Parent rev.

Child rev.

[Not reviewed]

0 7 0

mh - 4 years ago 2022-01-25 16:46:08
contact@maxhenger.nl

WIP: Changed component creation logic once more

7 files changed with 389 insertions and 108 deletions:

src/runtime2/communication.rs

src/runtime2/component/component_pdl.rs

263

src/runtime2/component/control_layer.rs

src/runtime2/runtime.rs

src/runtime2/store/component.rs

src/runtime2/store/mod.rs

src/runtime2/tests/mod.rs

0 comments (0 inline, 0 general)

src/runtime2/communication.rs

➞

Show inline comments

 use crate::protocol::eval::*;
 use super::runtime::*;
 use super::component::*;
 // -----------------------------------------------------------------------------
 // Generic types
 // -----------------------------------------------------------------------------
 #[derive(Debug, Copy, Clone, PartialEq, Eq)]
 pub struct PortId(pub u32);
 impl PortId {
     /// This value is not significant, it is chosen to make debugging easier: a
     /// very large port number is more likely to shine a light on bugs.
     pub fn new_invalid() -> Self {
         return Self(u32::MAX);
+    }
+}
 pub struct Peer {
     pub id: CompId,
     pub num_associated_ports: u32,
     pub(crate) handle: CompHandle,
+}
 #[derive(Debug, PartialEq, Eq, Clone, Copy)]
 pub enum PortKind {
     Putter,
     Getter,
+}
 #[derive(Debug, PartialEq, Eq)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum PortState {
     Open,
     Blocked,
     Closed,
+}
 pub struct Port {
     pub self_id: PortId,
     pub peer_id: PortId,
     pub kind: PortKind,
     pub state: PortState,
     pub peer_comp_id: CompId,
+}
 pub struct Channel {
     pub putter_id: PortId,
     pub getter_id: PortId,
+}
 // -----------------------------------------------------------------------------
 // Data messages
 // -----------------------------------------------------------------------------
 #[derive(Debug)]
 pub struct DataMessage {
     pub data_header: MessageDataHeader,
     pub sync_header: MessageSyncHeader,
     pub content: ValueGroup,
+}
 #[derive(Debug)]
 pub struct MessageDataHeader {
     pub expected_mapping: Vec<(PortId, Option<u32>)>,
     pub new_mapping: u32,
     pub source_port: PortId,
     pub target_port: PortId,
+}
 // -----------------------------------------------------------------------------
 // Sync messages
 // -----------------------------------------------------------------------------
 #[derive(Debug)]
 pub struct SyncMessage {
     pub sync_header: MessageSyncHeader,
     pub content: SyncMessageContent,
+}
 #[derive(Debug)]
 pub struct SyncLocalSolutionEntry {
     pub self_port_id: PortId,
     pub peer_comp_id: CompId,
     pub peer_port_id: PortId,
     pub mapping: u32,
     pub port_kind: PortKind,
+}
 pub type SyncLocalSolution = Vec<SyncLocalSolutionEntry>;
 #[derive(Debug)]
 pub struct SyncSolutionPort {
     pub self_comp_id: CompId,
     pub self_port_id: PortId,
     pub peer_comp_id: CompId,
     pub peer_port_id: PortId,
     pub mapping: u32,
     pub port_kind: PortKind,
+}
 #[derive(Debug)]
 pub struct SyncSolutionChannel {
     pub putter: Option<SyncSolutionPort>,
     pub getter: Option<SyncSolutionPort>,
+}
 #[derive(Debug, Copy, Clone, PartialEq, Eq)]
 pub enum SyncRoundDecision {
     None,
     Solution,
     Failure,
+}
 #[derive(Debug)]
 pub struct SyncPartialSolution {
     pub submissions_by: Vec<(CompId, bool)>,
     pub channel_mapping: Vec<SyncSolutionChannel>,
     pub decision: SyncRoundDecision,
+}
 impl Default for SyncPartialSolution {
     fn default() -> Self {
         return Self{
             submissions_by: Vec::new(),
             channel_mapping: Vec::new(),
             decision: SyncRoundDecision::None,
+        }
+    }
+}
 #[derive(Debug)]
 pub enum SyncMessageContent {
     NotificationOfLeader,
     LocalSolution(CompId, SyncLocalSolution), // local solution of the specified component
     PartialSolution(SyncPartialSolution), // partial solution of multiple components
     GlobalSolution,
     GlobalFailure,
+}
 // -----------------------------------------------------------------------------
 // Control messages
 // -----------------------------------------------------------------------------
 #[derive(Debug)]
 pub struct ControlMessage {
     pub(crate) id: ControlId,
     pub sender_comp_id: CompId,
     pub target_port_id: Option<PortId>,
     pub content: ControlMessageContent,
+}
 #[derive(Copy, Clone, Debug)]
 pub enum ControlMessageContent {
     Ack,
     BlockPort(PortId),
     UnblockPort(PortId),
     ClosePort(PortId),
     PortPeerChangedBlock(PortId),
     PortPeerChangedUnblock(PortId, CompId),
+}
 // -----------------------------------------------------------------------------
 // Messages (generic)
 // -----------------------------------------------------------------------------
 #[derive(Debug)]
 pub struct MessageSyncHeader {
     pub sync_round: u32,
     pub sending_id: CompId,
     pub highest_id: CompId,
+}
 #[derive(Debug)]
 pub enum Message {
     Data(DataMessage),
     Sync(SyncMessage),
     Control(ControlMessage),
+}
 impl Message {
     pub(crate) fn target_port(&self) -> Option<PortId> {
         match self {
             Message::Data(v) =>
                 return Some(v.data_header.target_port),
             Message::Control(v) =>
                 return v.target_port_id,
             Message::Sync(_) =>
                 return None,
+        }
+    }
     pub(crate) fn modify_target_port(&mut self, port_id: PortId) {
         match self {
             Message::Data(v) =>
                 v.data_header.target_port = port_id,
             Message::Control(v) =>
                 v.target_port_id = Some(port_id),
             Message::Sync(_) => unreachable!(), // should never be called for this message type
+        }
+    }
+}

src/runtime2/component/component_pdl.rs

➞

Show inline comments

 use crate::protocol::*;
 use crate::protocol::ast::DefinitionId;
 use crate::protocol::eval::{
     PortId as EvalPortId, Prompt,
     ValueGroup, Value,
     EvalContinuation, EvalResult, EvalError
 };
 use crate::runtime2::runtime::*;
 use crate::runtime2::scheduler::SchedulerCtx;
 use crate::runtime2::communication::*;
 use super::*;
 use super::control_layer::*;
 use super::consensus::Consensus;
 pub enum CompScheduling {
     Immediate,
     Requeue,
     Sleep,
     Exit,
+}
 pub struct CompCtx {
     pub id: CompId,
     pub ports: Vec<Port>,
     pub peers: Vec<Peer>,
     pub messages: Vec<Option<DataMessage>>, // same size as "ports"
     pub port_id_counter: u32,
+}
 impl Default for CompCtx {
     fn default() -> Self {
 impl CompCtx {
     pub(crate) fn new(reservation: &CompReserved) -> Self {
         return Self{
-            id: CompId(0),
+            id: reservation.id(),
             ports: Vec::new(),
             peers: Vec::new(),
             messages: Vec::new(),
             port_id_counter: 0,
+        }
+    }
+}
 struct MessageView<'a> {
     index: usize,
     pub message: &'a DataMessage,
+}
 impl CompCtx {
     /// Creates a new channel that is fully owned by the component associated
     /// with this context.
     fn create_channel(&mut self) -> Channel {
         let putter_id = PortId(self.take_port_id());
         let getter_id = PortId(self.take_port_id());
         self.ports.push(Port{
             self_id: putter_id,
             peer_id: getter_id,
             kind: PortKind::Putter,
             state: PortState::Open,
             peer_comp_id: self.id,
         });
         self.ports.push(Port{
             self_id: getter_id,
             peer_id: putter_id,
             kind: PortKind::Getter,
             state: PortState::Closed,
             peer_comp_id: self.id,
         });
         return Channel{ putter_id, getter_id };
+    }
     /// Adopts a port transferred by another component. Essentially copies all
     /// port data but creates a new ID. Caller should ensure that the other
     /// endpoint becomes aware of this ID.
     fn adopt_port(&mut self, to_transfer: &Port) -> &mut Port {
         let port_id = PortId(self.take_port_id());
         let port_index = self.ports.len();
         self.ports.push(Port{
             self_id: port_id,
             peer_id: to_transfer.peer_id,
             kind: to_transfer.kind,
             state: to_transfer.state,
             peer_comp_id: to_transfer.peer_comp_id,
         });
         return &mut self.ports[port_index];
+    }
     /// Adds a peer (or increments the "associated port" counter). Hence caller
     /// must make sure that this makes sense.
     fn add_peer(&mut self, sched_ctx: &SchedulerCtx, peer_id: CompId, peer_handle: Option<&CompHandle>) {
         match self.get_peer_index(peer_id) {
             Some(peer_index) => {
                 let peer_info = &mut self.peers[peer_index];
                 peer_info.num_associated_ports += 1;
             },
             None => {
                 let handle = if let Some(handle) = peer_handle {
                     handle.clone()
                 } else {
                     sched_ctx.runtime.get_component_public(peer_id)
                 };
                 self.peers.push(Peer{
                     id: peer_id,
                     num_associated_ports: 1,
                     handle,
                 })
+            }
+        }
+    }
     /// Removes a peer (or decrements the "associated port" counter). If there
     /// are no more references to the peer then the handle will be destroyed.
     fn remove_peer(&mut self, sched_ctx: &SchedulerCtx, peer_id: CompId) {
         let peer_index = self.get_peer_index(peer_id).unwrap();
         let peer_info = &mut self.peers[peer_index];
         peer_info.num_associated_ports -= 1;
         if peer_info.num_associated_ports == 0 {
             let mut peer = self.peers.remove(peer_index);
             let should_remove = peer.handle.decrement_users();
             if should_remove {
                 let key = unsafe{ peer.id.upgrade() };
                 sched_ctx.runtime.destroy_component(key);
+            }
+        }
+    }
     pub(crate) fn get_port(&self, port_id: PortId) -> &Port {
         let index = self.get_port_index(port_id).unwrap();
         return &self.ports[index];
+    }
     pub(crate) fn get_port_mut(&mut self, port_id: PortId) -> &mut Port {
         let index = self.get_port_index(port_id).unwrap();
         return &mut self.ports[index];
+    }
     pub(crate) fn get_port_index(&self, port_id: PortId) -> Option<usize> {
         for (index, port) in self.ports.iter().enumerate() {
             if port.self_id == port_id {
                 return Some(index);
+            }
+        }
         return None;
+    }
     pub(crate) fn get_peer(&self, peer_id: CompId) -> &Peer {
         let index = self.get_peer_index(peer_id).unwrap();
         return &self.peers[index];
+    }
     fn get_peer_mut(&mut self, peer_id: CompId) -> &mut Peer {
         let index = self.get_peer_index(peer_id).unwrap();
         return &mut self.peers[index];
+    }
     pub(crate) fn get_peer_index(&self, peer_id: CompId) -> Option<usize> {
         for (index, peer) in self.peers.iter().enumerate() {
             if peer.id == peer_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;
+    }
+}
 pub enum ExecStmt {
     CreatedChannel((Value, Value)),
     PerformedPut,
     PerformedGet(ValueGroup),
     None,
+}
 impl ExecStmt {
     fn take(&mut self) -> ExecStmt {
         let mut value = ExecStmt::None;
         std::mem::swap(self, &mut value);
         return value;
+    }
     fn is_none(&self) -> bool {
         match self {
             ExecStmt::None => return true,
             _ => return false,
+        }
+    }
+}
 pub struct ExecCtx {
     stmt: ExecStmt,
+}
 impl RunContext for ExecCtx {
     fn performed_put(&mut self, _port: EvalPortId) -> bool {
         match self.stmt.take() {
             ExecStmt::None => return false,
             ExecStmt::PerformedPut => return true,
             _ => unreachable!(),
+        }
+    }
     fn performed_get(&mut self, _port: EvalPortId) -> Option<ValueGroup> {
         match self.stmt.take() {
             ExecStmt::None => return None,
             ExecStmt::PerformedGet(value) => return Some(value),
             _ => unreachable!(),
+        }
+    }
     fn fires(&mut self, _port: EvalPortId) -> Option<Value> {
         todo!("remove fires")
+    }
     fn performed_fork(&mut self) -> Option<bool> {
         todo!("remove fork")
+    }
     fn created_channel(&mut self) -> Option<(Value, Value)> {
         match self.stmt.take() {
             ExecStmt::None => return None,
             ExecStmt::CreatedChannel(ports) => return Some(ports),
             _ => unreachable!(),
+        }
+    }
+}
 #[derive(Debug, Copy, Clone, PartialEq, Eq)]
 pub(crate) enum Mode {
     NonSync, // not in sync mode
     Sync, // in sync mode, can interact with other components
     SyncFail, // something went wrong during sync mode (deadlocked, error, whatever)
     SyncEnd, // awaiting a solution, i.e. encountered the end of the sync block
     BlockedGet,
     BlockedPut,
     Exit,
+}
 impl Mode {
     fn can_run(&self) -> bool {
         match self {
             Mode::NonSync | Mode::Sync =>
                 return true,
             Mode::SyncFail | Mode::SyncEnd | Mode::BlockedGet | Mode::BlockedPut | Mode::Exit =>
                 return false,
+        }
+    }
+}
 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
     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: Vec<Option<DataMessage>>,
     pub inbox_backup: Vec<DataMessage>,
+}
 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(),
             prompt: initial_state,
             control: ControlLayer::default(),
             consensus: Consensus::new(),
             sync_counter: 0,
             exec_ctx: ExecCtx{
                 stmt: ExecStmt::None,
             },
             inbox_main,
             inbox_backup: Vec::new(),
+        }
+    }
     pub(crate) fn handle_message(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx, message: Message) {
+    pub(crate) fn handle_message(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx, mut message: Message) {
         sched_ctx.log(&format!("handling message: {:?}", message));
         if let Some(new_target) = self.control.should_reroute(&message) {
+        if let Some(new_target) = self.control.should_reroute(&mut message) {
             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);
             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);
             },
             Message::Sync(message) => {
                 self.handle_incoming_sync_message(sched_ctx, comp_ctx, message);
+            }
+        }
+    }
     // -------------------------------------------------------------------------
     // Running component and handling changes in global component state
     // -------------------------------------------------------------------------
     pub(crate) fn run(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx) -> Result<CompScheduling, EvalError> {
         use EvalContinuation as EC;
         let can_run = self.mode.can_run();
         sched_ctx.log(&format!("Running component (mode: {:?}, can run: {})", self.mode, can_run));
         if !can_run {
             return Ok(CompScheduling::Sleep);
+        }
         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);
                 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!(self.exec_ctx.stmt.is_none());
                 let port_id = port_id_from_eval(port_id);
                 let port_index = comp_ctx.get_port_index(port_id).unwrap();
                 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.exec_ctx.stmt = ExecStmt::PerformedGet(message.content);
                         return Ok(CompScheduling::Immediate);
                     } else {
                         self.mode = Mode::SyncFail;
                         return Ok(CompScheduling::Sleep);
+                    }
                 } else {
                     // We need to wait
                     self.mode = Mode::BlockedGet;
                     self.mode_port = port_id;
                     return Ok(CompScheduling::Sleep);
+                }
             },
             EC::Put(port_id, value) => {
                 debug_assert_eq!(self.mode, Mode::Sync);
                 let port_id = port_id_from_eval(port_id);
                 let port_info = comp_ctx.get_port(port_id);
                 if port_info.state == PortState::Blocked {
                     todo!("handle blocked port");
+                }
                 self.send_data_message_and_wake_up(sched_ctx, comp_ctx, port_id, value);
                 self.exec_ctx.stmt = ExecStmt::PerformedPut;
                 return Ok(CompScheduling::Immediate);
             },
             // Results that can be returned outside of sync mode
             EC::ComponentTerminated => {
                 self.handle_component_exit(sched_ctx, comp_ctx);
                 return Ok(CompScheduling::Exit);
             },
             EC::SyncBlockStart => {
                 debug_assert_eq!(self.mode, Mode::NonSync);
                 self.handle_sync_start(sched_ctx, comp_ctx);
                 return Ok(CompScheduling::Immediate);
             },
             EC::NewComponent(definition_id, monomorph_idx, arguments) => {
                 debug_assert_eq!(self.mode, Mode::NonSync);
                 let mut ports = Vec::new(); // TODO: Optimize
                 let protocol = &sched_ctx.runtime.protocol;
                 find_ports_in_value_group(&arguments, &mut ports);
                 let prompt = Prompt::new(
                     &protocol.types, &protocol.heap,
                 self.create_component_and_transfer_ports2(
                     sched_ctx, comp_ctx,
                     definition_id, monomorph_idx, arguments
                 );
                 self.create_component_and_transfer_ports(sched_ctx, comp_ctx, prompt, &ports);
                 return Ok(CompScheduling::Requeue);
             },
             EC::NewChannel => {
                 debug_assert_eq!(self.mode, Mode::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);
+            }
+        }
+    }
     fn execute_prompt(&mut self, sched_ctx: &SchedulerCtx) -> EvalResult {
         let mut step_result = EvalContinuation::Stepping;
         while let EvalContinuation::Stepping = step_result {
             step_result = self.prompt.step(
                 &sched_ctx.runtime.protocol.types, &sched_ctx.runtime.protocol.heap,
                 &sched_ctx.runtime.protocol.modules, &mut self.exec_ctx,
             )?;
+        }
         return Ok(step_result)
+    }
     fn handle_sync_start(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx) {
         self.consensus.notify_sync_start(comp_ctx);
         debug_assert_eq!(self.mode, Mode::NonSync);
         self.mode = Mode::Sync;
+    }
     fn handle_sync_end(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx) {
         self.consensus.notify_sync_end(sched_ctx, comp_ctx);
         debug_assert_eq!(self.mode, Mode::Sync);
         self.mode = Mode::SyncEnd;
+    }
     fn handle_component_exit(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx) {
         debug_assert_eq!(self.mode, Mode::NonSync); // not a perfect assert, but just to remind myself: cannot exit while in sync
         // Note: for now we have that the scheduler handles exiting. I don't
         // know if that is a good idea, we'll see
         self.mode = Mode::Exit;
+    }
     // -------------------------------------------------------------------------
     // Handling messages
     // -------------------------------------------------------------------------
     fn send_data_message_and_wake_up(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &CompCtx, source_port_id: PortId, value: ValueGroup) {
         let port_info = comp_ctx.get_port(source_port_id);
         let peer_info = comp_ctx.get_peer(port_info.peer_comp_id);
         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) {
         // 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_index = comp_ctx.get_port_index(target_port_id).unwrap();
         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_ne!(comp_ctx.ports[port_index].state, PortState::Blocked); // because we could insert directly
             if self.mode == Mode::BlockedGet && self.mode_port == target_port_id {
                 // We were indeed blocked
                 self.mode = Mode::Sync;
                 self.mode_port = PortId::new_invalid();
+            }
             return;
+        }
         // The direct inbox is full, so the port will become (or was already) blocked
         let port_info = &mut comp_ctx.ports[port_index];
         debug_assert!(port_info.state == PortState::Open || port_info.state == PortState::Blocked);
         let _peer_comp_id = port_info.peer_comp_id;
         if port_info.state == PortState::Open {
             let (target_comp_id, block_message) =
                 self.control.set_port_and_peer_blocked(target_port_id, comp_ctx);
             debug_assert_eq!(_peer_comp_id, target_comp_id);
             let peer = comp_ctx.get_peer(target_comp_id);
             peer.handle.send_message(sched_ctx, Message::Control(block_message), true);
+        }
         // But we still need to remember the message, so:
         self.inbox_backup.push(message);
+    }
     /// Handles when a message has been handed off from the inbox to the PDL
     /// code. We check to see if there are more messages waiting and, if not,
     /// then we handle the case where the port might have been blocked
     /// previously.
     fn handle_received_data_message(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, port_id: PortId) {
         let port_index = comp_ctx.get_port_index(port_id).unwrap();
         debug_assert!(self.inbox_main[port_index].is_none()); // because we just received it
         // Check for any more messages
         for message_index in 0..self.inbox_backup.len() {
             let message = &self.inbox_backup[message_index];
             if message.data_header.target_port == port_id {
                 // One more message for this port
                 let message = self.inbox_backup.remove(message_index);
                 debug_assert_eq!(comp_ctx.get_port(port_id).state, PortState::Blocked); // since we had >1 message on the port
                 self.inbox_main[port_index] = Some(message);
                 return;
+            }
+        }
         // Did not have any more messages. So if we were blocked, then we need
         // to send the "unblock" message.
         let port_info = &comp_ctx.ports[port_index];
         if port_info.state == PortState::Blocked {
             let (peer_comp_id, message) = self.control.set_port_and_peer_unblocked(port_id, comp_ctx);
             let peer_info = comp_ctx.get_peer(peer_comp_id);
             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_port_id: PortId, peer_comp_id: CompId) {
             let peer_info = comp_ctx.get_peer(peer_comp_id);
             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 => {
                 let mut to_ack = message.id;
                 loop {
                     let action = self.control.handle_ack(to_ack, sched_ctx, comp_ctx);
                     match action {
                         AckAction::SendMessageAndAck(target_comp, message, new_to_ack) => {
                             // 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);
                             to_ack = new_to_ack;
                         },
                         AckAction::ScheduleComponent(to_schedule) => {
                             // FIX @NoDirectHandle
                             let mut handle = sched_ctx.runtime.get_component_public(to_schedule);
                             wake_up_if_sleeping(sched_ctx, to_schedule, &handle);
                             let _should_remove = handle.decrement_users();
                             debug_assert!(!_should_remove);
                             break;
                         },
                         AckAction::None => {
                             break;
+                        }
+                    }
+                }
             },
             ControlMessageContent::BlockPort(port_id) => {
                 // On of our messages was accepted, but the port should be
                 // blocked.
                 let port_info = comp_ctx.get_port_mut(port_id);
                 debug_assert_eq!(port_info.kind, PortKind::Putter);
                 if port_info.state != PortState::Closed {
                     debug_assert_ne!(port_info.state, PortState::Blocked); // implies unnecessary messages
                     port_info.state = PortState::Blocked;
+                }
             },
             ControlMessageContent::ClosePort(port_id) => {
                 // Request to close the port. We immediately comply and remove
                 // the component handle as well
                 let port_index = comp_ctx.get_port_index(port_id).unwrap();
                 let port_info = &mut comp_ctx.ports[port_index];
                 let peer_port_id = port_info.peer_id;
                 let peer_comp_id = port_info.peer_comp_id;
                 port_info.state = PortState::Closed;
                 let peer_index = comp_ctx.get_peer_index(peer_comp_id).unwrap();
                 let peer_info = &mut comp_ctx.peers[peer_index];
                 peer_info.num_associated_ports -= 1;
                 if peer_info.num_associated_ports == 0 {
                     // TODO: @Refactor clean up all these uses of "num_associated_ports"
                     let should_remove = peer_info.handle.decrement_users();
                     if should_remove {
                         let comp_key = unsafe{ peer_info.id.upgrade() };
                         sched_ctx.runtime.destroy_component(comp_key);
+                    }
                     comp_ctx.peers.remove(peer_index);
+                }
                 send_control_ack_message(sched_ctx, comp_ctx, message.id, peer_port_id, peer_comp_id);
+            }
             ControlMessageContent::UnblockPort(port_id) => {
                 // We were previously blocked (or already closed)
                 let port_info = comp_ctx.get_port(port_id);
                 debug_assert_eq!(port_info.kind, PortKind::Putter);
                 debug_assert!(port_info.state == PortState::Blocked || port_info.state == PortState::Closed);
                 if port_info.state == PortState::Blocked {
                     self.unblock_local_port(sched_ctx, comp_ctx, port_id);
+                }
             },
             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_info = comp_ctx.get_port_mut(port_id);
                 debug_assert!(port_info.state == PortState::Open || port_info.state == PortState::Blocked);
                 if port_info.state == PortState::Open {
                     port_info.state = PortState::Blocked;
+                }
                 let peer_port_id = port_info.peer_id;
                 let peer_comp_id = port_info.peer_comp_id;
                 send_control_ack_message(sched_ctx, comp_ctx, message.id, peer_port_id, peer_comp_id);
             },
             ControlMessageContent::PortPeerChangedUnblock(port_id, new_comp_id) => {
                 debug_assert_eq!(message.target_port_id, Some(port_id));
                 let port_info = comp_ctx.get_port_mut(port_id);
                 debug_assert!(port_info.state == PortState::Blocked);
                 port_info.peer_comp_id = new_comp_id;
                 self.unblock_local_port(sched_ctx, comp_ctx, port_id);
+            }
+        }
+    }
     fn handle_incoming_sync_message(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, message: SyncMessage) -> Option<CompScheduling> {
         let decision = self.consensus.receive_sync_message(sched_ctx, comp_ctx, message);
         let is_success = match decision {
             SyncRoundDecision::None => {
                 // No decision yet
                 return None;
             },
             SyncRoundDecision::Solution => true,
             SyncRoundDecision::Failure => false,
         };
         // If here then we've reached a conclusion
         debug_assert_eq!(self.mode, Mode::SyncEnd);
         self.mode = Mode::NonSync;
         if is_success {
             // We can simply continue executing. So we do nothing extra!
         } else {
             todo!("handle this better, show some kind of error");
             self.handle_component_exit(sched_ctx, comp_ctx);
             return Some(CompScheduling::Exit);
+        }
         return None;
+    }
     // -------------------------------------------------------------------------
     // Handling ports
     // -------------------------------------------------------------------------
     /// Marks the local port as being unblocked. If the execution was blocked on
     /// sending a message over this port, then execution will continue and the
     /// message will be sent.
     fn unblock_local_port(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, port_id: PortId) {
         let port_info = comp_ctx.get_port_mut(port_id);
         debug_assert_eq!(port_info.state, PortState::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_id, 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, prompt: Prompt, ports: &[PortId]) {
         let component = CompPDL::new(prompt, ports.len());
         let (comp_key, component) = sched_ctx.runtime.create_pdl_component(component, true);
         let created_ctx = &mut component.ctx;
         let mut has_reroute_entry = false;
         let schedule_entry_id = self.control.add_schedule_entry(created_ctx.id);
         for port_id in ports.iter().copied() {
             // Create temporary reroute entry if the peer is another component
             let port_info = creator_ctx.get_port(port_id);
             debug_assert_ne!(port_info.state, PortState::Blocked);
             if port_info.peer_comp_id == creator_ctx.id {
                 // We own the peer port. So retrieve it and modify the peer directly
                 let peer_port_id = port_info.peer_id;
                 let port_info = creator_ctx.get_port_mut(peer_port_id);
                 port_info.peer_comp_id = created_ctx.id;
                 Self::add_peer_associated_port_to_component(sched_ctx, creator_ctx, created_ctx.id);
     fn create_component_and_transfer_ports2(
         &mut self,
         sched_ctx: &SchedulerCtx, creator_ctx: &mut CompCtx,
         definition_id: DefinitionId, monomorph_index: i32, mut arguments: ValueGroup
     ) {
         struct PortPair{ creator: PortId, created: PortId }
         let mut port_id_pairs = Vec::new();
         let reservation = sched_ctx.runtime.start_create_pdl_component();
         let mut created_ctx = CompCtx::new(&reservation);
         // 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 = ValueGroupIter::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 = creator_ctx.get_port(creator_port_id);
             let created_port = created_ctx.adopt_port(creator_port);
             let created_port_id = created_port.self_id;
             port_id_pairs.push(PortPair{
                 creator: creator_port_id,
                 created: created_port_id,
             });
             // Modify value in arguments (bit dirty, but double vec in ValueGroup causes lifetime issues)
             let arg_value = if let Some(heap_pos) = port_reference.heap_pos {
                 &mut arg_iter.group.regions[heap_pos][port_reference.index]
             } else {
                 // We don't own the peer port, so send the appropriate messages
                 // to the peer component and notify the control layer
                 has_reroute_entry = true;
                 let message = self.control.add_reroute_entry(
                     creator_ctx.id, port_info.peer_id, port_info.peer_comp_id,
                     port_info.self_id, created_ctx.id, schedule_entry_id
                 );
                 let peer_info = creator_ctx.get_peer(port_info.peer_comp_id);
                 peer_info.handle.send_message(sched_ctx, message, true);
                 &mut arg_iter.group.values[port_reference.index]
             };
             match arg_value {
                 Value::Input(id) => *id = port_id_to_eval(created_port_id),
                 Value::Output(id) => *id = port_id_to_eval(created_port_id),
                 _ => unreachable!(),
+            }
+        }
             // Take out any potential messages for the peer
             let creator_port_index = creator_ctx.get_port_index(port_id).unwrap();
             let port_main_message = self.inbox_main[creator_port_index].take();
         // For each transferred port pair set their peer components to the
         // correct values. This will only change the values for the ports of
         // the new component.
         let mut created_component_has_remote_peers = false;
         for pair in port_id_pairs.iter() {
             let creator_port_info = creator_ctx.get_port(pair.creator);
             let created_port_info = created_ctx.get_port_mut(pair.created);
             if created_port_info.peer_comp_id == creator_ctx.id {
                 // Port peer is owned by the creator as well
                 let created_peer_port_index = port_id_pairs
                     .iter()
                     .position(|v| v.creator == creator_port_info.peer_id);
                 match created_peer_port_index {
                     Some(created_peer_port_index) => {
                         // Peer port moved to the new component as well
                         let peer_pair = &port_id_pairs[created_peer_port_index];
                         created_port_info.peer_id = peer_pair.created;
                         created_port_info.peer_comp_id = reservation.id();
                     },
                     None => {
                         // Peer port remains with creator component.
                         created_port_info.peer_comp_id = creator_ctx.id;
                         created_ctx.add_peer(sched_ctx, creator_ctx.id, None);
+                    }
+                }
             } else {
                 // Peer is a different component
                 let peer_info = creator_ctx.get_peer(created_port_info.peer_comp_id);
                 created_ctx.add_peer(sched_ctx, peer_info.id, Some(&peer_info.handle));
                 created_component_has_remote_peers = true;
+            }
+        }
             // Transfer port and create temporary reroute entry
             let (port_info, peer_info) = Self::remove_port_from_component(creator_ctx, port_id);
             if port_info.state == PortState::Blocked {
                 todo!("Think about this when you're not tired!");
         // 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 prompt = Prompt::new(
             &sched_ctx.runtime.protocol.types, &sched_ctx.runtime.protocol.heap,
             definition_id, monomorph_index, arguments,
         );
         let component = CompPDL::new(prompt, port_id_pairs.len());
         let (created_key, component) = sched_ctx.runtime.finish_create_pdl_component(
             reservation, component, created_ctx, false,
         );
         let created_ctx = &component.ctx;
         // Now modify the creator's ports: remove every transferred port and
         // potentially remove the peer component
         for pair in port_id_pairs.iter() {
             let creator_port_index = creator_ctx.get_port_index(pair.creator).unwrap();
             let creator_port_info = creator_ctx.ports.remove(creator_port_index);
             if creator_port_info.peer_comp_id != creator_ctx.id {
                 creator_ctx.remove_peer(sched_ctx, creator_port_info.peer_comp_id);
+            }
             Self::add_port_to_component(sched_ctx, created_ctx, port_info);
             // Transfer the taken messages
             let created_port_index = created_ctx.get_port_index(port_id).unwrap();
             component.code.inbox_main[created_port_index] = port_main_message;
             let mut message_index = 0;
             while message_index < self.inbox_backup.len() {
                 if self.inbox_backup[message_index].data_header.target_port == port_id {
                     // Move this message
                     let message = self.inbox_backup.remove(message_index);
                     component.code.inbox_backup.push(message);
                 } else {
                     message_index += 1;
+                }
             let created_port_info = created_ctx.get_port(pair.created);
             if created_port_info.peer_comp_id == creator_ctx.id {
                 // This is the cause where the creator obtains a reference
                 // to the created component
                 let peer_port_info = creator_ctx.get_port_mut(created_port_info.peer_id);
                 peer_port_info.peer_comp_id = created_ctx.id;
                 creator_ctx.add_peer(sched_ctx, created_ctx.id, None);
+            }
+        }
             // Maybe remove peer from the creator
             if let Some(mut peer_info) = peer_info {
                 let remove_from_runtime = peer_info.handle.decrement_users();
                 if remove_from_runtime {
                     let removed_comp_key = unsafe{ peer_info.id.upgrade() };
                     sched_ctx.runtime.destroy_component(removed_comp_key);
         // By now all ports have been transferred. We'll now do any of the setup
         // for rerouting/messaging
         if created_component_has_remote_peers {
             let schedule_entry_id = self.control.add_schedule_entry(created_ctx.id);
             for pair in port_id_pairs.iter() {
                 let port_info = created_ctx.get_port(pair.created);
                 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_id, port_info.peer_comp_id,
                         pair.creator, pair.created, created_ctx.id,
                         schedule_entry_id
                     );
                     let peer_info = created_ctx.get_peer(port_info.peer_comp_id);
                     peer_info.handle.send_message(sched_ctx, message, true);
+                }
+            }
         } else {
             // Peer can be scheduled immediately
             sched_ctx.runtime.enqueue_work(created_key);
+        }
         if !has_reroute_entry {
             // We can schedule the component immediately
             self.control.remove_schedule_entry(schedule_entry_id);
             component.public.sleeping.store(false, std::sync::atomic::Ordering::Release);
             sched_ctx.runtime.enqueue_work(comp_key);
         } // else: wait for the `Ack`s, they will trigger the scheduling of the component
+    }
     /// Removes a port from a component. Also decrements the port counter in
     /// the peer component's entry. If that hits 0 then it will be removed and
     /// returned. If returned then the caller is responsible for decrementing
     /// the atomic counters of the peer component's handle.
     fn remove_port_from_component(comp_ctx: &mut CompCtx, port_id: PortId) -> (Port, Option<Peer>) {
         let port_index = comp_ctx.get_port_index(port_id).unwrap();
         let port_info = comp_ctx.ports.remove(port_index);
         // If the component owns the peer, then we don't have to decrement the
         // number of peers (because we don't have an entry for ourselves)
         if port_info.peer_comp_id == comp_ctx.id {
             return (port_info, None);
+        }
         let peer_index = comp_ctx.get_peer_index(port_info.peer_comp_id).unwrap();
         let peer_info = &mut comp_ctx.peers[peer_index];
         peer_info.num_associated_ports -= 1;
         // Check if we still have other ports referencing this peer
         if peer_info.num_associated_ports != 0 {
             return (port_info, None);
+        }
         let peer_info = comp_ctx.peers.remove(peer_index);
         return (port_info, Some(peer_info));
+    }
     /// Adds a port to the component context. The peer (or its counter) will be
     /// updated accordingly.
     fn add_port_to_component(sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, port_info: Port) {
         // Add the port info
         let peer_comp_id = port_info.peer_comp_id;
         debug_assert!(!comp_ctx.ports.iter().any(|v| v.self_id == port_info.self_id));
         comp_ctx.ports.push(port_info);
         Self::add_peer_associated_port_to_component(sched_ctx, comp_ctx, peer_comp_id);
+    }
     /// Only adds/updates a peer for a given port. This function assumes (but
     /// does not check!) that the port was not considered to belong to that peer
     /// before calling this function.
     fn add_peer_associated_port_to_component(sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, peer_id: CompId) {
         match comp_ctx.get_peer_index(peer_id) {
             Some(peer_index) => {
                 let peer_info = &mut comp_ctx.peers[peer_index];
                 peer_info.num_associated_ports += 1;
             },
             None => {
                 let handle = sched_ctx.runtime.get_component_public(peer_id);
                 comp_ctx.peers.push(Peer{
                     id: peer_id,
                     num_associated_ports: 1,
                     handle,
                 });
+            }
+        }
+    }
     fn change_port_peer_component(
         &mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx,
         port_id: PortId, new_peer_comp_id: CompId
     ) {
         let port_info = comp_ctx.get_port_mut(port_id);
         let cur_peer_comp_id = port_info.peer_comp_id;
         let cur_peer_info = comp_ctx.get_peer_mut(cur_peer_comp_id);
         cur_peer_info.num_associated_ports -= 1;
         if cur_peer_info.num_associated_ports == 0 {
             let should_remove = cur_peer_info.handle.decrement_users();
             if should_remove {
                 let cur_peer_comp_key = unsafe{ cur_peer_comp_id.upgrade() };
                 sched_ctx.runtime.destroy_component(cur_peer_comp_key);
+            }
+        }
+    }
+}
 #[inline]
 fn port_id_from_eval(port_id: EvalPortId) -> PortId {
     return PortId(port_id.id);
+}
 #[inline]
 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<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 ValueGroupIter<'a> {
     group: &'a mut ValueGroup,
     heap_stack: Vec<(usize, usize)>,
     index: usize,
+}
 impl<'a> ValueGroupIter<'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 ValueGroupIter<'a> {
     type Item = ValueGroupPortRef;
     fn next(&mut self) -> Option<Self::Item> {
         // Enter loop that keeps iterating until a port is found
         loop {
             if let Some(pos) = self.heap_stack.last() {
                 let (heap_pos, region_index) = *pos;
                 if region_index >= self.group.regions[heap_pos].len() {
                     self.heap_stack.pop();
                     continue;
+                }
                 let value = &self.group.regions[heap_pos][region_index];
                 self.heap_stack.last_mut().unwrap().1 += 1;
                 match value {
                     Value::Input(id) | Value::Output(id) => {
                         let id = PortId(id.id);
                         return Some(ValueGroupPortRef{
                             id,
                             heap_pos: Some(heap_pos),
                             index: region_index,
                         });
                     },
                     _ => {},
+                }
                 if let Some(heap_pos) = value.get_heap_pos() {
                     self.heap_stack.push((heap_pos as usize, 0));
+                }
             } else {
                 if self.index >= self.group.values.len() {
                     return None;
+                }
                 let value = &mut self.group.values[self.index];
                 self.index += 1;
                 match value {
                     Value::Input(id) | Value::Output(id) => {
                         let id = PortId(id.id);
                         return Some(ValueGroupPortRef{
                             id,
                             heap_pos: None,
                             index: self.index - 1
                         });
                     },
                     _ => {},
+                }
                 // Not a port, check if we need to enter a heap region
                 if let Some(heap_pos) = value.get_heap_pos() {
                     self.heap_stack.push((heap_pos as usize, 0));
                 } // else: just consider the next value
+            }
+        }
+    }
+}
@@ \ No newline at end of file @@

src/runtime2/component/control_layer.rs

➞

Show inline comments

 use crate::runtime2::runtime::*;
 use crate::runtime2::communication::*;
 use crate::runtime2::component::*;
 #[derive(Debug, PartialEq, Eq, Clone, Copy)]
 pub(crate) struct ControlId(u32);
 impl ControlId {
     /// Like other invalid IDs, this one doesn't care any significance, but is
     /// just set at u32::MAX to hopefully bring out bugs sooner.
     fn new_invalid() -> Self {
         return ControlId(u32::MAX);
+    }
+}
 struct ControlEntry {
     id: ControlId,
     ack_countdown: u32,
     content: ControlContent,
+}
 enum ControlContent {
     PeerChange(ContentPeerChange),
     ScheduleComponent(CompId),
     BlockedPort(PortId),
     ClosedPort(PortId),
+}
 struct ContentPeerChange {
     source_port: PortId,
     source_comp: CompId,
     target_port: PortId,
     old_target_port: PortId,
     new_target_port: PortId,
     new_target_comp: CompId,
     schedule_entry_id: ControlId,
+}
 pub(crate) enum AckAction {
     None,
     SendMessageAndAck(CompId, ControlMessage, ControlId),
     ScheduleComponent(CompId),
+}
 /// Handling/sending control messages.
 pub(crate) struct ControlLayer {
     id_counter: ControlId,
     entries: Vec<ControlEntry>,
+}
 impl ControlLayer {
     pub(crate) fn should_reroute(&self, message: &Message) -> Option<CompId> {
+    pub(crate) fn should_reroute(&self, message: &mut Message) -> Option<CompId> {
         // Safety note: rerouting should occur during the time when we're
         // notifying a peer of a new component. During this period that
         // component hasn't been executed yet, so cannot have died yet.
         // FIX @NoDirectHandle
         let target_port = message.target_port();
         if target_port.is_none() {
             return None;
+        }
         let target_port = target_port.unwrap();
         for entry in &self.entries {
             if let ControlContent::PeerChange(entry) = &entry.content {
                 if entry.target_port == target_port {
                 if entry.old_target_port == target_port {
                     message.modify_target_port(entry.new_target_port);
                     return Some(entry.new_target_comp);
+                }
+            }
+        }
         return None;
+    }
     pub(crate) fn handle_ack(&mut self, entry_id: ControlId, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx) -> AckAction {
         let entry_index = self.get_entry_index(entry_id).unwrap();
         let entry = &mut self.entries[entry_index];
         debug_assert!(entry.ack_countdown > 0);
         entry.ack_countdown -= 1;
         if entry.ack_countdown != 0 {
             return AckAction::None;
+        }
         // All `Ack`s received, take action based on the kind of entry
         match &entry.content {
             ControlContent::PeerChange(content) => {
                 // If change of peer is ack'd. Then we are certain we have
                 // rerouted all of the messages, and the sender's port can now
                 // be unblocked again.
                 let target_comp_id = content.source_comp;
                 let message_to_send = ControlMessage{
                     id: ControlId::new_invalid(),
                     sender_comp_id: comp_ctx.id,
                     target_port_id: Some(content.source_port),
                     content: ControlMessageContent::PortPeerChangedUnblock(
-                        content.source_port,
+                        content.new_target_port,
                         content.new_target_comp
+                    )
                 };
                 let to_ack = content.schedule_entry_id;
                 self.entries.remove(entry_index);
                 return AckAction::SendMessageAndAck(target_comp_id, message_to_send, to_ack);
             },
             ControlContent::ScheduleComponent(to_schedule) => {
                 // If all change-of-peers are `Ack`d, then we're ready to
                 // schedule the component!
                 return AckAction::ScheduleComponent(*to_schedule);
             },
             ControlContent::BlockedPort(_) => unreachable!(),
             ControlContent::ClosedPort(port_id) => {
                 // If a closed port is Ack'd, then we remove the reference to
                 // that component.
                 let port_index = comp_ctx.get_port_index(*port_id).unwrap();
                 debug_assert_eq!(comp_ctx.ports[port_index].state, PortState::Blocked);
                 let peer_id = comp_ctx.ports[port_index].peer_comp_id;
                 let peer_index = comp_ctx.get_peer_index(peer_id).unwrap();
                 let peer_info = &mut comp_ctx.peers[peer_index];
                 peer_info.num_associated_ports -= 1;
                 if peer_info.num_associated_ports == 0 {
                     let should_remove = peer_info.handle.decrement_users();
                     if should_remove {
                         let comp_key = unsafe{ peer_info.id.upgrade() };
                         sched_ctx.runtime.destroy_component(comp_key);
+                    }
                     comp_ctx.peers.remove(peer_index);
+                }
                 return AckAction::None;
+            }
+        }
+    }
     // -------------------------------------------------------------------------
     // Port transfer (due to component creation)
     // -------------------------------------------------------------------------
     /// Adds an entry that, when completely ack'd, will schedule a component.
     pub(crate) fn add_schedule_entry(&mut self, to_schedule_id: CompId) -> ControlId {
         let entry_id = self.take_id();
         self.entries.push(ControlEntry{
             id: entry_id,
             ack_countdown: 0, // incremented by calls to `add_reroute_entry`
             content: ControlContent::ScheduleComponent(to_schedule_id),
         });
         return entry_id;
+    }
     /// Removes a schedule entry. Only used if the caller preemptively called
     /// `add_schedule_entry`, but ended up not calling `add_reroute_entry`,
     /// hence the `ack_countdown` in the scheduling entry is at 0.
     pub(crate) fn remove_schedule_entry(&mut self, schedule_entry_id: ControlId) {
         let index = self.get_entry_index(schedule_entry_id).unwrap();
         debug_assert_eq!(self.entries[index].ack_countdown, 0);
         self.entries.remove(index);
+    }
     pub(crate) fn add_reroute_entry(
         &mut self, creator_comp_id: CompId,
         source_port_id: PortId, source_comp_id: CompId,
-        target_port_id: PortId, new_comp_id: CompId,
+        old_target_port_id: PortId, new_target_port_id: PortId, new_comp_id: CompId,
         schedule_entry_id: ControlId,
     ) -> Message {
         let entry_id = self.take_id();
         self.entries.push(ControlEntry{
             id: entry_id,
             ack_countdown: 1,
             content: ControlContent::PeerChange(ContentPeerChange{
                 source_port: source_port_id,
                 source_comp: source_comp_id,
                 target_port: target_port_id,
                 old_target_port: old_target_port_id,
                 new_target_port: new_target_port_id,
                 new_target_comp: new_comp_id,
                 schedule_entry_id,
             }),
         });
         // increment counter on schedule entry
         for entry in &mut self.entries {
             if entry.id == schedule_entry_id {
                 entry.ack_countdown += 1;
                 break;
+            }
+        }
         return Message::Control(ControlMessage{
             id: entry_id,
             sender_comp_id: creator_comp_id,
             target_port_id: Some(source_port_id),
             content: ControlMessageContent::PortPeerChangedBlock(source_port_id)
         })
+    }
     // -------------------------------------------------------------------------
     // Blocking, unblocking, and closing ports
     // -------------------------------------------------------------------------
     pub(crate) fn mark_port_closed<'a>(&mut self, port_id: PortId, comp_ctx: &mut CompCtx) -> Option<(CompId, ControlMessage)> {
         let port = comp_ctx.get_port_mut(port_id);
         let peer_port_id = port.peer_id;
         let peer_comp_id = port.peer_comp_id;
         debug_assert!(port.state == PortState::Open || port.state == PortState::Blocked);
         port.state = PortState::Closed;
         if peer_comp_id == comp_ctx.id {
             // We own the other end of the channel as well.
             return None;
+        }
         let entry_id = self.take_id();
         self.entries.push(ControlEntry{
             id: entry_id,
             ack_countdown: 1,
             content: ControlContent::ClosedPort(port_id),
         });
         return Some((
             peer_comp_id,
             ControlMessage{
                 id: entry_id,
                 sender_comp_id: comp_ctx.id,
                 target_port_id: Some(peer_port_id),
                 content: ControlMessageContent::ClosePort(peer_port_id),
+            }
         ));
+    }
     pub(crate) fn set_port_and_peer_blocked(&mut self, port_id: PortId, comp_ctx: &mut CompCtx) -> (CompId, ControlMessage) {
         // TODO: Feels like this shouldn't be an entry. Hence this class should
         //  be renamed. Lets see where the code ends up being
         let entry_id = self.take_id();
         let port_info = comp_ctx.get_port_mut(port_id);
         let peer_port_id = port_info.peer_id;
         let peer_comp_id = port_info.peer_comp_id;
         debug_assert_eq!(port_info.state, PortState::Open); // prevent unforeseen issues
         port_info.state = PortState::Blocked;
         self.entries.push(ControlEntry{
             id: entry_id,
             ack_countdown: 0,
             content: ControlContent::BlockedPort(port_id),
         });
         return (
             peer_comp_id,
             ControlMessage{
                 id: entry_id,
                 sender_comp_id: comp_ctx.id,
                 target_port_id: Some(peer_port_id),
                 content: ControlMessageContent::BlockPort(peer_port_id),
+            }
         );
+    }
     pub(crate) fn set_port_and_peer_unblocked(&mut self, port_id: PortId, comp_ctx: &mut CompCtx) -> (CompId, ControlMessage) {
         // Find the entry that contains the blocking entry for the port
         let mut entry_index = usize::MAX;
         let mut entry_id = ControlId::new_invalid();
         for (index, entry) in self.entries.iter().enumerate() {
             if let ControlContent::BlockedPort(blocked_port) = &entry.content {
                 if *blocked_port == port_id {
                     entry_index = index;
                     entry_id = entry.id;
                     break;
+                }
+            }
+        }
         let port_info = comp_ctx.get_port_mut(port_id);
         let peer_port_id = port_info.peer_id;
         let peer_comp_id = port_info.peer_comp_id;
         debug_assert_eq!(port_info.state, PortState::Blocked);
         debug_assert_eq!(port_info.kind, PortKind::Getter); // because we blocked it because of receiving too many messages
         port_info.state = PortState::Open;
         return (
             peer_comp_id,
             ControlMessage{
                 id: entry_id,
                 sender_comp_id: comp_ctx.id,
                 target_port_id: Some(peer_port_id),
                 content: ControlMessageContent::UnblockPort(peer_port_id),
+            }
+        )
+    }
     // -------------------------------------------------------------------------
     // Internal utilities
     // -------------------------------------------------------------------------
     fn take_id(&mut self) -> ControlId {
         let id = self.id_counter;
         self.id_counter.0 = self.id_counter.0.wrapping_add(1);
         return id;
+    }
     fn get_entry_index(&self, entry_id: ControlId) -> Option<usize> {
         for (index, entry) in self.entries.iter().enumerate() {
             if entry.id == entry_id {
                 return Some(index);
+            }
+        }
         return None;
+    }
+}
 impl Default for ControlLayer {
     fn default() -> Self {
         return ControlLayer{
             id_counter: ControlId(0),
             entries: Vec::new(),
+        }
+    }
+}
@@ \ No newline at end of file @@

src/runtime2/runtime.rs

➞

Show inline comments

 use std::sync::{Arc, Mutex, Condvar};
 use std::sync::atomic::{AtomicU32, AtomicBool, Ordering};
 use std::collections::VecDeque;
 use crate::protocol::*;
 use crate::runtime2::component::wake_up_if_sleeping;
 use super::communication::Message;
 use super::component::{CompCtx, CompPDL};
 use super::store::{ComponentStore, QueueDynMpsc, QueueDynProducer};
+use super::store::{ComponentStore, ComponentReservation, QueueDynMpsc, QueueDynProducer};
 use super::scheduler::*;
 // -----------------------------------------------------------------------------
 // Component
 // -----------------------------------------------------------------------------
 /// Key to a component. Type system somewhat ensures that there can only be one
 /// of these. Only with a key one may retrieve privately-accessible memory for
 /// a component. Practically just a generational index, like `CompId` is.
 #[derive(Debug, Copy, Clone, PartialEq, Eq)]
 pub(crate) struct CompKey(pub u32);
 impl CompKey {
     pub(crate) fn downgrade(&self) -> CompId {
         return CompId(self.0);
+    }
+}
 /// Generational ID of a component
 #[derive(Debug, Copy, Clone, PartialEq, Eq)]
 pub struct CompId(pub u32);
 impl CompId {
     pub(crate) fn new_invalid() -> CompId {
         return CompId(u32::MAX);
+    }
     /// Upgrade component ID to component key. Unsafe because the caller needs
     /// to make sure that only one component key can exist at a time (to ensure
     /// a component can only be scheduled/executed by one thread).
     pub(crate) unsafe fn upgrade(&self) -> CompKey {
         return CompKey(self.0);
+    }
+}
 /// Handle to a component that is being created.
 pub(crate) struct CompReserved {
     reservation: ComponentReservation,
+}
 impl CompReserved {
     pub(crate) fn id(&self) -> CompId {
         return CompId(self.reservation.index)
+    }
+}
 /// Private fields of a component, may only be modified by a single thread at
 /// a time.
 pub(crate) struct RuntimeComp {
     pub public: CompPublic,
     pub code: CompPDL,
     pub ctx: CompCtx,
     pub inbox: QueueDynMpsc<Message>
+}
 /// Should contain everything that is accessible in a thread-safe manner
 // TODO: Do something about the `num_handles` thing. This needs to be a bit more
 //  "foolproof" to lighten the mental burden of using the `num_handles`
 //  variable.
 pub(crate) struct CompPublic {
     pub sleeping: AtomicBool,
     pub num_handles: AtomicU32, // manually modified (!)
     inbox: QueueDynProducer<Message>,
+}
 /// Handle to public part of a component. Would be nice if we could
 /// automagically manage the `num_handles` counter. But when it reaches zero we
 /// need to manually remove the handle from the runtime. So we just have debug
 /// code to make sure this actually happens.
 pub(crate) struct CompHandle {
     target: *const CompPublic,
     id: CompId, // TODO: @Remove after debugging
     #[cfg(debug_assertions)] decremented: bool,
+}
 impl CompHandle {
     fn new(id: CompId, public: &CompPublic) -> CompHandle {
         let handle = CompHandle{
             target: public,
             id,
             #[cfg(debug_assertions)] decremented: false,
         };
         handle.increment_users();
         return handle;
+    }
     pub(crate) fn send_message(&self, sched_ctx: &SchedulerCtx, message: Message, try_wake_up: bool) {
         sched_ctx.log(&format!("Sending message to [c:{:03}, wakeup:{}]: {:?}", self.id.0, try_wake_up, message));
         self.inbox.push(message);
         if try_wake_up {
             wake_up_if_sleeping(sched_ctx, self.id, self);
+        }
+    }
     fn increment_users(&self) {
         let old_count = self.num_handles.fetch_add(1, Ordering::AcqRel);
         debug_assert!(old_count > 0); // because we should never be able to retrieve a handle when the component is (being) destroyed
+    }
     /// Returns true if the component should be destroyed
     pub(crate) fn decrement_users(&mut self) -> bool {
         debug_assert!(!self.decremented, "illegal to 'decrement_users' twice");
         dbg_code!(self.decremented = true);
         let old_count = self.num_handles.fetch_sub(1, Ordering::AcqRel);
         return old_count == 1;
+    }
+}
 impl Clone for CompHandle {
     fn clone(&self) -> Self {
         debug_assert!(!self.decremented, "illegal to clone after 'decrement_users'");
         self.increment_users();
         return CompHandle{
             target: self.target,
             id: self.id,
             #[cfg(debug_assertions)] decremented: false,
         };
+    }
+}
 impl std::ops::Deref for CompHandle {
     type Target = CompPublic;
     fn deref(&self) -> &Self::Target {
         return unsafe{ &*self.target };
+    }
+}
 impl Drop for CompHandle {
     fn drop(&mut self) {
         debug_assert!(self.decremented, "need call to 'decrement_users' before dropping");
+    }
+}
 // -----------------------------------------------------------------------------
 // Runtime
 // -----------------------------------------------------------------------------
 pub struct Runtime {
     pub(crate) inner: Arc<RuntimeInner>,
     threads: Vec<std::thread::JoinHandle<()>>,
+}
 impl Runtime {
     pub fn new(num_threads: u32, protocol_description: ProtocolDescription) -> Runtime {
         assert!(num_threads > 0, "need a thread to perform work");
         let runtime_inner = Arc::new(RuntimeInner {
             protocol: protocol_description,
             components: ComponentStore::new(128),
             work_queue: Mutex::new(VecDeque::with_capacity(128)),
             work_condvar: Condvar::new(),
             active_elements: AtomicU32::new(1),
         });
         let mut runtime = Runtime {
             inner: runtime_inner,
             threads: Vec::with_capacity(num_threads as usize),
         };
         for thread_index in 0..num_threads {
             let mut scheduler = Scheduler::new(runtime.inner.clone(), thread_index);
             let thread_handle = std::thread::spawn(move || {
                 scheduler.run();
             });
             runtime.threads.push(thread_handle);
+        }
         return runtime;
+    }
+}
 impl Drop for Runtime {
     fn drop(&mut self) {
         self.inner.decrement_active_components();
         for handle in self.threads.drain(..) {
             handle.join().expect("join scheduler thread");
+        }
+    }
+}
 /// Memory that is maintained by "the runtime". In practice it is maintained by
 /// multiple schedulers, and this serves as the common interface to that memory.
 pub(crate) struct RuntimeInner {
     pub protocol: ProtocolDescription,
     components: ComponentStore<RuntimeComp>,
     work_queue: Mutex<VecDeque<CompKey>>,
     work_condvar: Condvar,
     active_elements: AtomicU32, // active components and APIs (i.e. component creators)
+}
 impl RuntimeInner {
     // Scheduling and retrieving work
     pub(crate) fn take_work(&self) -> Option<CompKey> {
         let mut lock = self.work_queue.lock().unwrap();
         while lock.is_empty() && self.active_elements.load(Ordering::Acquire) != 0 {
             lock = self.work_condvar.wait(lock).unwrap();
+        }
         // We have work, or the schedulers should exit.
         return lock.pop_front();
+    }
     pub(crate) fn enqueue_work(&self, key: CompKey) {
         let mut lock = self.work_queue.lock().unwrap();
         lock.push_back(key);
         self.work_condvar.notify_one();
+    }
     // Creating/destroying components
     pub(crate) fn start_create_pdl_component(&self) -> CompReserved {
         self.increment_active_components();
         let reservation = self.components.reserve();
         return CompReserved{ reservation };
+    }
     pub(crate) fn finish_create_pdl_component(
         &self, reserved: CompReserved,
         component: CompPDL, mut context: CompCtx, initially_sleeping: bool,
     ) -> (CompKey, &mut RuntimeComp) {
         let inbox_queue = QueueDynMpsc::new(16);
         let inbox_producer = inbox_queue.producer();
         let _id = reserved.id();
         context.id = reserved.id();
         let component = RuntimeComp {
             public: CompPublic{
                 sleeping: AtomicBool::new(initially_sleeping),
                 num_handles: AtomicU32::new(1), // the component itself acts like a handle
                 inbox: inbox_producer,
             },
             code: component,
             ctx: context,
             inbox: inbox_queue,
         };
         let index = self.components.submit(reserved.reservation, component);
         debug_assert_eq!(index, _id.0);
         let component = self.components.get_mut(index);
         return (CompKey(index), component);
+    }
     /// Creates a new component. Note that the public part will be properly
     /// initialized, but the private fields (e.g. owned ports, peers, etc.)
     /// are not.
     pub(crate) fn create_pdl_component(&self, comp: CompPDL, initially_sleeping: bool) -> (CompKey, &mut RuntimeComp) {
     /// initialized, but not all private fields are.
     pub(crate) fn create_pdl_component(&self, comp: CompPDL, ctx: CompCtx, initially_sleeping: bool) -> (CompKey, &mut RuntimeComp) {
         let inbox_queue = QueueDynMpsc::new(16);
         let inbox_producer = inbox_queue.producer();
         let comp = RuntimeComp{
             public: CompPublic{
                 sleeping: AtomicBool::new(initially_sleeping),
                 num_handles: AtomicU32::new(1), // the component itself acts like a handle
                 inbox: inbox_producer,
             },
             code: comp,
-            ctx: CompCtx::default(),
             ctx,
             inbox: inbox_queue,
         };
         let index = self.components.create(comp);
         // TODO: just do a reserve_index followed by a consume_index or something
         self.increment_active_components();
         let component = self.components.get_mut(index);
         component.ctx.id = CompId(index);
         return (CompKey(index), component);
+    }
     pub(crate) fn get_component(&self, key: CompKey) -> &mut RuntimeComp {
         let component = self.components.get_mut(key.0);
         return component;
+    }
     pub(crate) fn get_component_public(&self, id: CompId) -> CompHandle {
         let component = self.components.get(id.0);
         return CompHandle::new(id, &component.public);
+    }
     pub(crate) fn destroy_component(&self, key: CompKey) {
         debug_assert_eq!(self.get_component(key).public.num_handles.load(Ordering::Acquire), 0);
         self.decrement_active_components();
         self.components.destroy(key.0);
+    }
     // Tracking number of active interfaces and the active components
     #[inline]
     fn increment_active_components(&self) {
         let _old_val = self.active_elements.fetch_add(1, Ordering::AcqRel);
         debug_assert!(_old_val > 0); // can only create a component from a API/component, so can never be 0.
+    }
     fn decrement_active_components(&self) {
         let old_val = self.active_elements.fetch_sub(1, Ordering::AcqRel);
         debug_assert!(old_val > 0); // something wrong with incr/decr logic
         let new_val = old_val - 1;
         if new_val == 0 {
             // Just to be sure, in case the last thing that gets destroyed is an
             // API instead of a thread.
             let lock = self.work_queue.lock();
             self.work_condvar.notify_all();
+        }
+    }
+}

src/runtime2/store/component.rs

➞

Show inline comments

 /*
  * Component Store
+ *
  * Concurrent datastructure for creating/destroying/retrieving components using
  * their ID. It is essentially a variation on a concurrent freelist. We store an
  * array of (potentially null) pointers to data. Indices into this array that
  * are unused (but may be left allocated) are in a freelist. So creating a new
  * bit of data involves taking an index from this freelist. Destruction involves
  * putting the index back.
+ *
  * This datastructure takes care of the threadsafe implementation of the
  * freelist and calling the data's destructor when needed. Note that it is not
  * completely safe (in Rust's sense of the word) because it is possible to
  * get more than one mutable reference to a piece of data. Likewise it is
  * possible to put back bogus indices into the freelist, which will destroy the
  * integrity of the datastructure.
+ *
  * Some underlying assumptions that led to this design (note that I haven't
  * actually checked these conditions or performed any real profiling, yet):
  *  - Resizing the freelist should be very rare. The datastructure should grow
  *    to some kind of maximum size and stay at that size.
  *  - Creation should (preferably) be faster than deletion of data. Reason being
  *    that creation implies we're creating a component that has code to be
  *    executed. Better to quickly be able to execute code than being able to
  *    quickly tear down finished components.
  *  - Retrieval is much more likely than creation/destruction.
+ *
  * Some obvious flaws with this implementation:
  *  - Because of the freelist implementation we will generally allocate all of
  *    the data pointers that are available (i.e. if we have a buffer of size
  *    64, but we generally use 33 elements, than we'll have 64 elements
  *    allocated), which might be wasteful at larger array sizes (which are
  *    always powers of two).
  *  - A lot of concurrent operations are not necessary: we may move some of the
  *    access to the global concurrent datastructure by an initial access to some
  *    kind of thread-local datastructure.
  */
 use std::mem::transmute;
 use std::alloc::{dealloc, Layout};
 use std::ptr;
 use std::sync::atomic::{AtomicUsize, Ordering};
 use super::unfair_se_lock::{UnfairSeLock, UnfairSeLockSharedGuard};
 /// Generic store of components. Essentially a resizable freelist (implemented
 /// as a ringbuffer) combined with an array of actual elements.
 pub struct ComponentStore<T: Sized> {
     inner: UnfairSeLock<Inner<T>>,
     read_head: AtomicUsize,
     write_head: AtomicUsize,
     limit_head: AtomicUsize,
+}
 unsafe impl<T: Sized> Send for ComponentStore<T>{}
 unsafe impl<T: Sized> Sync for ComponentStore<T>{}
 /// Contents of the `ComponentStore` that require a shared/exclusive locking
 /// mechanism for consistency.
 struct Inner<T: Sized> {
     freelist: Vec<u32>,
     data: Vec<*mut T>,
     size: usize,
     compare_mask: usize,
     index_mask: usize,
+}
 type InnerShared<'a, T> = UnfairSeLockSharedGuard<'a, Inner<T>>;
 /// Reservation of a slot in the component store. Corresponds to the case where
 /// an index has been taken from the freelist, but the element has not yet been
 /// initialized
 pub struct ComponentReservation {
     pub(crate) index: u32,
     #[cfg(debug_assertions)] submitted: bool,
+}
 impl ComponentReservation {
     fn new(index: u32) -> Self {
         return Self{
             index,
             #[cfg(debug_assertions)] submitted: false,
+        }
+    }
+}
 impl Drop for ComponentReservation {
     fn drop(&mut self) {
         debug_assert!(self.submitted);
+    }
+}
 impl<T: Sized> ComponentStore<T> {
     pub fn new(initial_size: usize) -> Self {
         Self::assert_valid_size(initial_size);
         // Fill initial freelist and preallocate data array
         let mut initial_freelist = Vec::with_capacity(initial_size);
         for idx in 0..initial_size {
             initial_freelist.push(idx as u32)
+        }
         let mut initial_data = Vec::new();
         initial_data.resize(initial_size, ptr::null_mut());
         // Return initial store
         return Self{
             inner: UnfairSeLock::new(Inner{
                 freelist: initial_freelist,
                 data: initial_data,
                 size: initial_size,
                 compare_mask: 2*initial_size - 1,
                 index_mask: initial_size - 1,
             }),
             read_head: AtomicUsize::new(0),
             write_head: AtomicUsize::new(initial_size),
             limit_head: AtomicUsize::new(initial_size),
         };
+    }
     /// Creates a new element initialized to the provided `value`. This returns
     /// the index at which the element can be retrieved.
     pub fn create(&self, value: T) -> u32 {
         let lock = self.inner.lock_shared();
         let (lock, index) = self.pop_freelist_index(lock);
-        self.initialize_at_index(lock, index, value);
+        Self::initialize_at_index(lock, index, value);
         return index;
+    }
     pub fn reserve(&self) -> ComponentReservation {
         let lock = self.inner.lock_shared();
         let (lock, index) = self.pop_freelist_index(lock);
         return ComponentReservation::new(index);
+    }
     pub fn submit(&self, mut reservation: ComponentReservation, value: T) -> u32 {
         dbg_code!({ reservation.submitted = true; });
         let lock = self.inner.lock_shared();
         Self::initialize_at_index(lock, reservation.index, value);
         return reservation.index;
+    }
     /// Destroys an element at the provided `index`. The caller must make sure
     /// that it does not use any previously received references to the data at
     /// this index, and that no more calls to `get` are performed using this
     /// index. This is allowed again if the index has been reacquired using
     /// `create`.
     pub fn destroy(&self, index: u32) {
         let lock = self.inner.lock_shared();
         self.destruct_at_index(&lock, index);
         self.push_freelist_index(&lock, index);
+    }
     /// Retrieves an element by reference
     pub fn get(&self, index: u32) -> &T {
         let lock = self.inner.lock_shared();
         let value = lock.data[index as usize];
         unsafe {
             debug_assert!(!value.is_null());
             return &*value;
+        }
+    }
     /// Retrieves an element by mutable reference. The caller should ensure that
     /// use of that mutability is thread-safe
     pub fn get_mut(&self, index: u32) -> &mut T {
         let lock = self.inner.lock_shared();
         let value = lock.data[index as usize];
         unsafe {
             debug_assert!(!value.is_null());
             return &mut *value;
+        }
+    }
     #[inline]
     fn pop_freelist_index<'a>(&'a self, mut shared_lock: InnerShared<'a, T>) -> (InnerShared<'a, T>, u32) {
         'attempt_read: loop {
             // Load indices and check for reallocation condition
             let current_size = shared_lock.size;
             let mut read_index = self.read_head.load(Ordering::Relaxed);
             let limit_index = self.limit_head.load(Ordering::Acquire);
             if read_index == limit_index {
                 shared_lock = self.reallocate(current_size, shared_lock);
                 continue 'attempt_read;
+            }
             loop {
                 let preemptive_read = shared_lock.freelist[read_index & shared_lock.index_mask];
                 if let Err(actual_read_index) = self.read_head.compare_exchange(
                     read_index, (read_index + 1) & shared_lock.compare_mask,
                     Ordering::AcqRel, Ordering::Acquire
                 ) {
                     // We need to try again
                     read_index = actual_read_index;
                     continue 'attempt_read;
+                }
                 // If here then we performed the read
                 return (shared_lock, preemptive_read);
+            }
+        }
+    }
     #[inline]
-    fn initialize_at_index(&self, read_lock: InnerShared<T>, index: u32, value: T) {
     fn initialize_at_index(read_lock: InnerShared<T>, index: u32, value: T) {
         let mut target_ptr = read_lock.data[index as usize];
         unsafe {
             if target_ptr.is_null() {
                 let layout = Layout::for_value(&value);
                 target_ptr = std::alloc::alloc(layout).cast();
                 let rewrite: *mut *mut T = transmute(read_lock.data.as_ptr());
                 *rewrite.add(index as usize) = target_ptr;
+            }
             std::ptr::write(target_ptr, value);
+        }
+    }
     #[inline]
     fn push_freelist_index(&self, read_lock: &InnerShared<T>, index_to_put_back: u32) {
         // Acquire an index in the freelist to which we can write
         let mut cur_write_index = self.write_head.load(Ordering::Relaxed);
         let mut new_write_index = (cur_write_index + 1) & read_lock.compare_mask;
         while let Err(actual_write_index) = self.write_head.compare_exchange(
             cur_write_index, new_write_index,
             Ordering::AcqRel, Ordering::Acquire
         ) {
             cur_write_index = actual_write_index;
             new_write_index = (cur_write_index + 1) & read_lock.compare_mask;
+        }
         // We own the data at the index, write to it and notify reader through
         // limit_head that it can be read from. Note that we cheat around the
         // rust mutability system here :)
         unsafe {
             let target: *mut u32 = transmute(read_lock.freelist.as_ptr());
             *(target.add(cur_write_index & read_lock.index_mask)) = index_to_put_back;
+        }
         // Essentially spinlocking, relaxed failure ordering because the logic
         // is that a write first moves the `write_head`, then the `limit_head`.
         while let Err(_) = self.limit_head.compare_exchange(
             cur_write_index, new_write_index,
             Ordering::AcqRel, Ordering::Relaxed
         ) {};
+    }
     #[inline]
     fn destruct_at_index(&self, read_lock: &InnerShared<T>, index: u32) {
         let target_ptr = read_lock.data[index as usize];
         unsafe{ ptr::drop_in_place(target_ptr); }
+    }
     // NOTE: Bit of a mess, and could have a cleanup with better logic for the
     // resizing. Maybe even a different indexing scheme...
     fn reallocate(&self, old_size: usize, inner: InnerShared<T>) -> InnerShared<T> {
         drop(inner);
+        {
             // After dropping read lock, acquire write lock
             let mut lock = self.inner.lock_exclusive();
             if old_size == lock.size {
                 // We are the thread that is supposed to reallocate
                 let new_size = old_size * 2;
                 Self::assert_valid_size(new_size);
                 // Note that the atomic indices are in the range [0, new_size)
                 // already, so we need to be careful
                 let new_index_mask = new_size - 1;
                 let new_compare_mask = (2 * new_size) - 1;
                 lock.data.resize(new_size, ptr::null_mut());
                 lock.freelist.resize(new_size, 0);
                 for idx in 0..old_size {
                     lock.freelist[old_size + idx] = lock.freelist[idx];
+                }
                 // We need to fill the freelist with the indices of all of the
                 // new elements that we have just created.
                 debug_assert_eq!(self.limit_head.load(Ordering::SeqCst), self.write_head.load(Ordering::SeqCst));
                 let old_read_index = self.read_head.load(Ordering::SeqCst);
                 let old_write_index = self.write_head.load(Ordering::SeqCst);
                 if old_read_index > old_write_index {
                     // Read index wraps, so keep it as-is and fill
                     let new_read_index = old_read_index + old_size;
                     for index in 0..old_size {
                         let target_idx = (new_read_index + index) & new_index_mask;
                         lock.freelist[target_idx] = (old_size + index) as u32;
+                    }
                     self.read_head.store(new_read_index, Ordering::SeqCst);
                     debug_assert!(new_read_index < 2*new_size);
                     debug_assert!(old_write_index.wrapping_sub(new_read_index) & new_compare_mask <= new_size);
                 } else {
                     // No wrapping, so increment write index
                     let new_write_index = old_write_index + old_size;
                     for index in 0..old_size {
                         let target_idx = (old_write_index + index) & new_index_mask;
                         lock.freelist[target_idx] = (old_size + index) as u32;
+                    }
                     // Update write/limit heads
                     self.write_head.store(new_write_index, Ordering::SeqCst);
                     self.limit_head.store(new_write_index, Ordering::SeqCst);
                     debug_assert!(new_write_index < 2*new_size);
                     debug_assert!(new_write_index.wrapping_sub(old_read_index) & new_compare_mask <= new_size);
+                }
                 // Update sizes and masks
                 lock.size = new_size;
                 lock.compare_mask = new_compare_mask;
                 lock.index_mask = new_index_mask;
             } // else: someone else allocated, so we don't have to
+        }
         // We've dropped the write lock, acquire the read lock again
         return self.inner.lock_shared();
+    }
     #[inline]
     fn assert_valid_size(size: usize) {
         // Condition the size needs to adhere to. Some are a bit excessive, but
         // we don't hit this check very often
         assert!(
             size.is_power_of_two() &&
                 size >= 4 &&
                 size <= usize::MAX / 2 &&
                 size <= u32::MAX as usize
         );
+    }
+}
 impl<T: Sized> Drop for ComponentStore<T> {
     fn drop(&mut self) {
         let value_layout = Layout::from_size_align(
             std::mem::size_of::<T>(), std::mem::align_of::<T>()
         ).unwrap();
         // Note that if the indices exist in the freelist then the destructor
         // has already been called. So handle them first
         let mut lock = self.inner.lock_exclusive();
         let read_index = self.read_head.load(Ordering::Acquire);
         let write_index = self.write_head.load(Ordering::Acquire);
         debug_assert_eq!(write_index, self.limit_head.load(Ordering::Acquire));
         let mut index = read_index;
         while index != write_index {
             let dealloc_index = lock.freelist[index & lock.index_mask] as usize;
             let target_ptr = lock.data[dealloc_index];
             unsafe {
                 dealloc(target_ptr.cast(), value_layout);
                 lock.data[dealloc_index] = ptr::null_mut();
+            }
             index += 1;
             index &= lock.compare_mask;
+        }
         // With all of those set to null, we'll just iterate through all
         // pointers and destruct+deallocate the ones not set to null yet
         for target_ptr in lock.data.iter().copied() {
             if !target_ptr.is_null() {
                 unsafe {
                     ptr::drop_in_place(target_ptr);
                     dealloc(target_ptr.cast(), value_layout);
+                }
+            }
+        }
+    }
+}
 #[cfg(test)]
 mod tests {
     use super::*;
     use super::super::tests::*;
     use rand::prelude::*;
     use rand_pcg::Pcg32;
     fn seeds() -> Vec<[u8;16]> {
         return vec![
             [241, 47, 70, 87, 240, 246, 20, 173, 219, 143, 74, 23, 158, 58, 205, 172],
             [178, 112, 230, 205, 230, 178, 2, 90, 162, 218, 49, 196, 224, 222, 208, 43],
             [245, 42, 35, 167, 153, 205, 221, 144, 200, 253, 144, 117, 176, 231, 17, 70],
             [143, 39, 177, 216, 124, 96, 225, 39, 30, 82, 239, 193, 133, 58, 255, 193],
             [25, 105, 10, 52, 161, 212, 190, 112, 178, 193, 68, 249, 167, 153, 172, 144],
+        ]
+    }
     #[test]
     fn test_ctor_dtor_simple_unthreaded() {
         const NUM_ROUNDS: usize = 5;
         const NUM_ELEMENTS: usize = 1024;
         let store = ComponentStore::new(32);
         let counters = Counters::new();
         let mut indices = Vec::with_capacity(NUM_ELEMENTS);
         for _round_index in 0..NUM_ROUNDS {
             // Creation round
             for value in 0..NUM_ELEMENTS {
                 let new_resource = Resource::new(&counters, value as u64);
                 let new_index = store.create(new_resource);
                 indices.push(new_index);
+            }
             // Checking round
             for el_index in indices.iter().copied() {
                 let element = store.get(el_index);
                 assert_eq!(element.val, el_index as u64);
+            }
             // Destruction round
             for el_index in indices.iter().copied() {
                 store.destroy(el_index);
+            }
             indices.clear();
+        }
         let num_expected = (NUM_ROUNDS * NUM_ELEMENTS) as u64;
         assert_ctor_eq!(counters, num_expected);
         assert_dtor_eq!(counters, num_expected);
+    }
     #[test]
     fn test_ctor_dtor_simple_threaded() {
         const MAX_SIZE: usize = 1024;
         const NUM_THREADS: usize = 4;
         const NUM_PER_THREAD: usize = MAX_SIZE / NUM_THREADS;
         const NUM_ROUNDS: usize = 4;
         assert!(MAX_SIZE % NUM_THREADS == 0);
         let store = Arc::new(ComponentStore::new(16));
         let counters = Counters::new();
         let mut threads = Vec::with_capacity(NUM_THREADS);
         for thread_index in 0..NUM_THREADS {
             // Setup local clones to move into the thread
             let store = store.clone();
             let first_index = thread_index * NUM_PER_THREAD;
             let last_index = (thread_index + 1) * NUM_PER_THREAD;
             let counters = counters.clone();
             let handle = std::thread::spawn(move || {
                 let mut indices = Vec::with_capacity(last_index - first_index);
                 for _round_index in 0..NUM_ROUNDS {
                     // Creation round
                     for value in first_index..last_index {
                         let el_index = store.create(Resource::new(&counters, value as u64));
                         indices.push(el_index);
+                    }
                     // Checking round
                     for (value_offset, el_index) in indices.iter().copied().enumerate() {
                         let element = store.get(el_index);
                         assert_eq!(element.val, (first_index + value_offset) as u64);
+                    }
                     // Destruction round
                     for el_index in indices.iter().copied() {
                         store.destroy(el_index);
+                    }
                     indices.clear();
+                }
             });
             threads.push(handle);
+        }
         for thread in threads {
             thread.join().expect("clean exit");
+        }
         let num_expected = (NUM_ROUNDS * MAX_SIZE) as u64;
         assert_ctor_eq!(counters, num_expected);
         assert_dtor_eq!(counters, num_expected);
+    }
     #[test]
     fn test_ctor_dtor_random_threaded() {
         const NUM_ROUNDS: usize = 4;
         const NUM_THREADS: usize = 4;
         const NUM_OPERATIONS: usize = 1024;
         const NUM_OPS_PER_THREAD: usize = NUM_OPERATIONS / NUM_THREADS;
         const NUM_OPS_PER_ROUND: usize = NUM_OPS_PER_THREAD / NUM_ROUNDS;
         const NUM_STORED_PER_THREAD: usize = 32;
         assert!(NUM_OPERATIONS % NUM_THREADS == 0);
         assert!(NUM_OPS_PER_THREAD / 2 > NUM_STORED_PER_THREAD);
         let seeds = seeds();
         for seed_index in 0..seeds.len() {
             // Setup store, counters and threads
             let store = Arc::new(ComponentStore::new(16));
             let counters = Counters::new();
             let mut threads = Vec::with_capacity(NUM_THREADS);
             for thread_index in 0..NUM_THREADS {
                 // Setup local clones to move into the thread
                 let store = store.clone();
                 let counters = counters.clone();
                 // Setup local rng
                 let mut seed = seeds[seed_index];
                 for seed_val_idx in 0..16 {
                     seed[seed_val_idx] ^= thread_index as u8; // blegh
+                }
                 let mut rng = Pcg32::from_seed(seed);
                 let handle = std::thread::spawn(move || {
                     let mut stored = Vec::with_capacity(NUM_STORED_PER_THREAD);
                     for _round_index in 0..NUM_ROUNDS {
                         // Modify store elements in the store randomly, for some
                         // silly definition of random
                         for _op_index in 0..NUM_OPS_PER_ROUND {
                             // Perform a single operation, depending on current
                             // size of the number of values owned by this thread
                             let new_value = rng.next_u64();
                             let should_create = rng.next_u32() % 2 == 0;
                             let is_empty = stored.is_empty();
                             let is_full = stored.len() == NUM_STORED_PER_THREAD;
                             if is_empty || (!is_full && should_create) {
                                 // Must create
                                 let el_index = store.create(Resource::new(&counters, new_value));
                                 stored.push((el_index, new_value));
                             } else {
                                 // Must destroy
                                 let stored_index = new_value as usize % stored.len();
                                 let (el_index, el_value) = stored.remove(stored_index);
                                 store.destroy(el_index);
+                            }
+                        }
                         // Checking if the values we own still make sense
                         for (el_index, value) in stored.iter().copied() {
                             let gotten = store.get(el_index);
                             assert_eq!(value, gotten.val, "failed at thread {} value {}", thread_index, el_index);
+                        }
+                    }
                     return stored.len(); // return number of remaining elements
                 });
                 threads.push(handle);
+            }
             // Done with the current round
             let mut total_left_allocated = 0;
             for thread in threads {
                 let num_still_stored = thread.join().unwrap();
                 total_left_allocated += num_still_stored as u64;
+            }
             // Before store is dropped
             // note: cannot determine number of creations, creation/destructor
             // is random
             let num_ctor = counters.ctor.load(Ordering::Acquire);
             assert_dtor_eq!(counters, num_ctor - total_left_allocated);
             // After store is dropped
             drop(store);
             assert_dtor_eq!(counters, num_ctor);
+        }
+    }
+}
@@ \ No newline at end of file @@

src/runtime2/store/mod.rs

➞

Show inline comments

 #[macro_use]
 #[cfg(test)]
 mod tests;
 pub mod unfair_se_lock;
 pub mod component;
 pub mod queue_mpsc;
 pub(crate) use component::ComponentStore;
+pub(crate) use component::{ComponentStore, ComponentReservation};
 pub(crate) use queue_mpsc::{QueueDynMpsc, QueueDynProducer};
@@ \ No newline at end of file @@

src/runtime2/tests/mod.rs

➞

Show inline comments

 use crate::protocol::*;
 use crate::protocol::eval::*;
 use crate::runtime2::runtime::*;
 use crate::runtime2::component::CompPDL;
 use crate::runtime2::component::{CompCtx, CompPDL};
 fn create_component(rt: &Runtime, module_name: &str, routine_name: &str, args: ValueGroup) {
     let prompt = rt.inner.protocol.new_component(
         module_name.as_bytes(), routine_name.as_bytes(), args
     ).expect("create prompt");
     let reserved = rt.inner.start_create_pdl_component();
     let ctx = CompCtx::new(&reserved);
     let (key, _) = rt.inner.finish_create_pdl_component(reserved, CompPDL::new(prompt, 0), ctx, false);
     rt.inner.enqueue_work(key);
+}
 fn no_args() -> ValueGroup { ValueGroup::new_stack(Vec::new()) }
 #[test]
 fn test_component_creation() {
     let pd = ProtocolDescription::parse(b"
     primitive nothing_at_all() {
         s32 a = 5;
         auto b = 5 + a;
+    }
     ").expect("compilation");
     let rt = Runtime::new(1, pd);
     for i in 0..20 {
         let prompt = rt.inner.protocol.new_component(b"", b"nothing_at_all", ValueGroup::new_stack(Vec::new()))
             .expect("component creation");
         let comp = CompPDL::new(prompt, 0);
         let (key, _) = rt.inner.create_pdl_component(comp, false);
         rt.inner.enqueue_work(key);
         create_component(&rt, "", "nothing_at_all", no_args());
+    }
+}
 #[test]
 fn test_component_communication() {
     let pd = ProtocolDescription::parse(b"
     primitive sender(out<u32> o) {
         print(\"sender\");
         sync put(o, 1);
+    }
     primitive receiver(in<u32> i) {
         print(\"receiver\");
         sync get(i);
+    }
     composite constructor() {
         channel o -> i;
         print(\"creating sender\");
         new sender(o);
         print(\"creating receiver\");
         new receiver(i);
         print(\"done\");
+    }
     ").expect("compilation");
     let rt = Runtime::new(1, pd);
     let prompt = rt.inner.protocol.new_component(b"", b"constructor", ValueGroup::new_stack(Vec::new()))
         .expect("creation");
     let (key, _) = rt.inner.create_pdl_component(CompPDL::new(prompt, 0), false);
     rt.inner.enqueue_work(key);
     create_component(&rt, "", "constructor", no_args());
+}
@@ \ No newline at end of file @@

0 comments (0 inline, 0 general)