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

Changeset - e62f669c1e03

Parent rev.

Child rev.

[Not reviewed]

0 3 0

mh - 3 years ago 2022-05-09 13:37:54
contact@maxhenger.nl

WIP on refactoring port transmission code

3 files changed with 105 insertions and 6 deletions:

src/runtime2/component/component.rs

src/runtime2/component/component_context.rs

src/runtime2/component/control_layer.rs

0 comments (0 inline, 0 general)

src/runtime2/component/component.rs

➞

Show inline comments

+}
 // -----------------------------------------------------------------------------
 // Internal messaging/state utilities
 // -----------------------------------------------------------------------------
 /// Sends a message without any transmitted ports. Does not check if sending
 /// is actually valid.
 fn send_message_without_ports(
     sending_port_handle: LocalPortHandle, value: ValueGroup,
     comp_ctx: &CompCtx, sched_ctx: &SchedulerCtx, consensus: &mut Consensus,
 ) {
     let port_info = comp_ctx.get_port(sending_port_handle);
     debug_assert!(port_info.state.can_send());
     let peer_handle = comp_ctx.get_peer_handle(port_info.peer_comp_id);
     let peer_info = comp_ctx.get_peer(peer_handle);
     let annotated_message = consensus.annotate_data_message(comp_ctx, port_info, value);
     peer_info.handle.send_message_logged(sched_ctx, Message::Data(annotated_message), true);
+}
 fn send_message_with_ports(
     sending_port_handle: LocalPortHandle, sending_port_instruction: PortInstruction,
     value: ValueGroup,
     exec_state: &mut CompExecState, comp_ctx: &mut CompCtx, sched_ctx: &SchedulerCtx,
     control: &mut ControlLayer, consensus: &mut Consensus
 ) -> Result<(), (PortInstruction, String)> {
     debug_assert_eq!(exec_state.mode, CompMode::Sync); // busy in sync, trying to send
     let sending_port_info = comp_ctx.get_port_mut(sending_port_handle);
     sending_port_info.last_instruction = sending_port_instruction;
     let mut transmit_ports = Vec::new();
     find_ports_in_value_group(&value, &mut transmit_ports);
     debug_assert!(!transmit_ports.is_empty()); // requisite for calling this function
     // Set up the final Ack that triggers us to send our final message
     let unblock_put_control_id = control.add_unblock_put_with_ports_entry();
     for (_, port_id) in &transmit_ports {
         let transmit_port_handle = comp_ctx.get_port_handle(*port_id);
         let transmit_port_info = comp_ctx.get_port_mut(transmit_port_handle);
         // Note: we checked earlier that we are currently in sync mode. Now we
         // will check if we've already used the port we're about to transmit.
         if !transmit_port_info.last_instruction.is_none() {
             return Err((
                 sending_port_instruction,
                 String::from("Cannot transmit one of the ports, as it is used in this sync round")
             ));
+        }
         if transmit_port_info.state.is_set(PortStateFlag::Transmitted) {
             return Err((
                 sending_port_instruction,
                 String::from("Cannot transmit one of the ports, as that port is already transmitted")
             ));
+        }
         transmit_port_info.state.set(PortStateFlag::Transmitted);
+    }
     let port_info = comp_ctx.get_port(sending_port_handle);
     return Ok(());
+}
 /// Handles an `Ack` for the control layer.
 fn default_handle_ack(
     control: &mut ControlLayer, control_id: ControlId,
     sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx
 ) {
     // Since an `Ack` may cause another one, handle them in a loop
 #[inline]
 pub(crate) fn port_id_to_eval(port_id: PortId) -> EvalPortId {
     return EvalPortId{ id: port_id.0 };
+}
 // TODO: Optimize double vec
 type EncounteredPorts = Vec<(Vec<ValueId>, PortId)>;
 /// Recursively goes through the value group, attempting to find ports.
 /// Duplicates will only be added once.
-pub(crate) fn find_ports_in_value_group(value_group: &ValueGroup, ports: &mut Vec<(ValueId, PortId)>) {
+pub(crate) fn find_ports_in_value_group(value_group: &ValueGroup, ports: &mut EncounteredPorts) {
     // Helper to check a value for a port and recurse if needed.
-    fn find_port_in_value(group: &ValueGroup, value: &Value, value_location: ValueId, ports: &mut Vec<(ValueId, PortId)>) {
+    fn find_port_in_value(group: &ValueGroup, value: &Value, value_location: ValueId, ports: &mut EncounteredPorts) {
         match value {
             Value::Input(port_id) | Value::Output(port_id) => {
                 // This is an actual port
                 let cur_port = PortId(port_id.id);
                 for prev_port in ports.iter() {
                     if *prev_port == cur_port.1 {
                 for prev_port in ports.iter_mut() {
                     if prev_port.1 == cur_port {
                         // Already added
                         prev_port.0.push(value_location);
                         return;
+                    }
+                }
                 ports.push((value_location, cur_port));
+                ports.push((vec![value_location], cur_port));
             },
             Value::Array(heap_pos) |
             Value::Message(heap_pos) |
             Value::String(heap_pos) |
             Value::Struct(heap_pos) |
             Value::Union(_, heap_pos) => {

src/runtime2/component/component_context.rs

➞

Show inline comments

@@ @@ -11,12 +11,21 @@ use crate::protocol::ExpressionId; @@
 pub enum PortInstruction {
     None,
     NoSource,
     SourceLocation(ExpressionId),
 }
 impl PortInstruction {
     pub fn is_none(&self) -> bool {
         match self {
             PortInstruction::None => return true,
             _ => return false,
+        }
+    }
+}
 /// Directionality of a port
 #[derive(Debug, PartialEq, Eq, Clone, Copy)]
 pub enum PortKind {
     Putter,
     Getter,
 }
@@ @@ -26,13 +35,13 @@ pub enum PortKind { @@
 #[repr(u32)]
 #[derive(Debug, Copy, Clone)]
 pub enum PortStateFlag {
     Closed = 0x01, // If not closed, then the port is open
     BlockedDueToPeerChange = 0x02, // busy changing peers, hence use of port is temporarily blocked
     BlockedDueToFullBuffers = 0x04,
     Transmitted, // Transmitted, so cannot be used anymore
+    Transmitted = 0x08, // Transmitted, so cannot be used anymore
 }
 #[derive(Copy, Clone)]
 pub struct PortState {
     flags: u32
 }
@@ @@ -46,12 +55,19 @@ impl PortState { @@
     #[inline]
     pub fn is_open(&self) -> bool {
         return !self.is_closed();
     }
     #[inline]
     pub fn can_send(&self) -> bool {
         return
             !self.is_set(PortStateFlag::Closed) &&
             !self.is_set(PortStateFlag::Transmitted);
+    }
     #[inline]
     pub fn is_closed(&self) -> bool {
         return self.is_set(PortStateFlag::Closed);
     }
     #[inline]

src/runtime2/component/control_layer.rs

➞

Show inline comments

@@ @@ -22,12 +22,13 @@ struct ControlEntry { @@
+}
 enum ControlContent {
     PeerChange(ContentPeerChange),
     ScheduleComponent(CompId),
     ClosedPort(PortId),
     UnblockPutWithPorts
+}
 struct ContentPeerChange {
     source_port: PortId,
     source_comp: CompId,
     old_target_port: PortId,
@@ @@ -42,12 +43,13 @@ struct ControlClosedPort { @@
+}
 pub(crate) enum AckAction {
     None,
     SendMessage(CompId, ControlMessage),
     ScheduleComponent(CompId),
     UnblockPutWithPorts,
+}
 /// Handling/sending control messages.
 pub(crate) struct ControlLayer {
     id_counter: ControlId,
     entries: Vec<ControlEntry>,
@@ @@ -125,12 +127,15 @@ impl ControlLayer { @@
                 // that component.
                 let port_handle = comp_ctx.get_port_handle(closed_port);
                 debug_assert!(comp_ctx.get_port(port_handle).state.is_closed());
                 comp_ctx.change_port_peer(sched_ctx, port_handle, None);
                 return (AckAction::None, None);
             },
             ControlContent::UnblockPutWithPorts => {
                 return (AckAction::UnblockPutWithPorts, None);
+            }
+        }
+    }
     pub(crate) fn has_acks_remaining(&self) -> bool {
         return !self.entries.is_empty();
@@ @@ -149,12 +154,24 @@ impl ControlLayer { @@
             content: ControlContent::ScheduleComponent(to_schedule_id),
         });
         return entry_id;
+    }
     /// Adds an entry that returns the similarly named Ack action
     pub(crate) fn add_unblock_put_with_ports_entry(&mut self) -> ControlId {
         let entry_id = self.take_id();
         self.entries.push(ControlEntry{
             id: entry_id,
             ack_countdown: 1, // incremented by calls to `add_reroute_entry`
             content: ControlContent::UnblockPutWithPorts,
         });
         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_by_id(schedule_entry_id).unwrap();
         debug_assert_eq!(self.entries[index].ack_countdown, 0);

0 comments (0 inline, 0 general)