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

Changeset - 58fb633d3fa1

Parent rev.

Child rev.

[Not reviewed]

0 5 0

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

WIP on figuring out protocol for port transmission

5 files changed with 87 insertions and 51 deletions:

src/protocol/eval/mod.rs

src/runtime2/communication.rs

src/runtime2/component/component.rs

src/runtime2/component/component_context.rs

src/runtime2/component/component_pdl.rs

0 comments (0 inline, 0 general)

src/protocol/eval/mod.rs

➞

Show inline comments

@@ @@ -24,8 +24,7 @@ pub(crate) mod value; @@
 pub(crate) mod store;
 pub(crate) mod executor;
 pub(crate) mod error;
 pub use error::EvalError;
 pub use value::{PortId, Value, ValueGroup};
+pub use value::{PortId, Value, ValueId, ValueGroup};
 pub use executor::{EvalContinuation, EvalResult, Prompt};

src/runtime2/communication.rs

➞

Show inline comments

 use crate::protocol::eval::*;
 use crate::protocol::eval::value::ValueId;
 use super::runtime::*;
 use super::component::*;
@@ @@ -31,6 +32,7 @@ pub struct DataMessage { @@
     pub data_header: MessageDataHeader,
     pub sync_header: MessageSyncHeader,
     pub content: ValueGroup,
     pub ports: Vec<TransmittedPort>,
+}
 #[derive(Debug)]
@@ @@ -61,6 +63,14 @@ pub struct MessageDataHeader { @@
     pub target_port: PortId,
+}
 #[derive(Debug)]
 pub struct TransmittedPort {
     location: ValueId, // within `content`
     messages: Vec<DataMessage>, // owned by previous component
     peer_comp: CompId,
     peer_port: PortId,
+}
 // -----------------------------------------------------------------------------
 // Sync messages
 // -----------------------------------------------------------------------------

src/runtime2/component/component.rs

➞

Show inline comments

 use std::fmt::{Display as FmtDisplay, Result as FmtResult, Formatter};
 use crate::protocol::eval::{Prompt, EvalError, ValueGroup, PortId as EvalPortId};
+use crate::protocol::eval::{Prompt, EvalError, ValueGroup, Value, ValueId, PortId as EvalPortId};
 use crate::protocol::*;
 use crate::runtime2::*;
 use crate::runtime2::communication::*;
 use crate::runtime2::component::component_pdl::find_ports_in_value_group;
 use super::{CompCtx, CompPDL, CompId};
 use super::component_context::*;
@@ @@ -78,6 +79,7 @@ pub(crate) enum CompMode { @@
     BlockedGet, // blocked because we need to receive a message on a particular port
     BlockedPut, // component is blocked because the port is blocked
     BlockedSelect, // waiting on message to complete the select statement
     BlockedPutPorts,// blocked because we're waiting to send a data message containing ports
     StartExit, // temporary state: if encountered then we start the shutdown process.
     BusyExit, // temporary state: waiting for Acks for all the closed ports, potentially waiting for sync round to finish
     Exit, // exiting: shutdown process started, now waiting until the reference count drops to 0
@@ @@ -88,7 +90,7 @@ impl CompMode { @@
         use CompMode::*;
         match self {
             Sync | SyncEnd | BlockedGet | BlockedPut | BlockedSelect => true,
+            Sync | SyncEnd | BlockedGet | BlockedPut | BlockedSelect | BlockedPutPorts => true,
             NonSync | StartExit | BusyExit | Exit => false,
+        }
+    }
@@ @@ -97,7 +99,7 @@ impl CompMode { @@
         use CompMode::*;
         match self {
             NonSync | Sync | SyncEnd | BlockedGet | BlockedPut | BlockedSelect => false,
+            NonSync | Sync | SyncEnd | BlockedGet | BlockedPut | BlockedSelect | BlockedPutPorts => false,
             StartExit | BusyExit => true,
             Exit => false,
+        }
@@ @@ -256,6 +258,30 @@ pub(crate) fn default_send_data_message( @@
         return Ok(CompScheduling::Sleep);
     } else {
         // Check if there are any ports that are being transmitted
         let mut ports = Vec::new();
         find_ports_in_value_group(&value, &mut ports);
         if !ports.is_empty() {
             for (value_location, port_id) in ports {
                 let transmitted_port_handle = comp_ctx.get_port_handle(port_id);
                 let transmitted_port = comp_ctx.get_port(transmitted_port_handle);
                 if transmitted_port.state.is_set(PortStateFlag::Transmitted) {
                     // Note: We could also attach where the port has been
                     //  transferred
                     return Err((
                         port_info.last_instruction,
                         String::from("Cannot send this message, as it contains a previously transmitted port")
                     ));
+                }
                 // Prepare ack for PPC
                 // Prepare PPC message
+            }
+        }
         // Port is not blocked, so send to the peer
         let peer_handle = comp_ctx.get_peer_handle(port_info.peer_comp_id);
         let peer_info = comp_ctx.get_peer(peer_handle);
@@ @@ -800,6 +826,10 @@ fn default_handle_recently_unblocked_port( @@
     debug_assert!(!port_info.state.is_blocked()); // should have been done by the caller
     if exec_state.is_blocked_on_put(port_id) {
         // Return to the regular execution mode
         exec_state.mode = CompMode::Sync;
         exec_state.mode_port = PortId::new_invalid();
         // Annotate the message that we're going to send
         let port_info = comp_ctx.get_port(port_handle); // for immutable access
         debug_assert_eq!(port_info.kind, PortKind::Putter);
 pub(crate) fn port_id_to_eval(port_id: PortId) -> EvalPortId {
     return EvalPortId{ id: port_id.0 };
+}
 /// Recursively goes through the value group, attempting to find ports.
 /// Duplicates will only be added once.
 pub(crate) fn find_ports_in_value_group(value_group: &ValueGroup, ports: &mut Vec<(ValueId, PortId)>) {
     // Helper to check a value for a port and recurse if needed.
     fn find_port_in_value(group: &ValueGroup, value: &Value, value_location: ValueId, ports: &mut Vec<(ValueId, PortId)>) {
         match value {
             Value::Input(port_id) | Value::Output(port_id) => {
                 // This is an actual port
                 let cur_port = PortId(port_id.id);
                 for prev_port in ports.iter() {
                     if *prev_port == cur_port.1 {
                         // Already added
                         return;
+                    }
+                }
                 ports.push((value_location, cur_port));
             },
             Value::Array(heap_pos) |
             Value::Message(heap_pos) |
             Value::String(heap_pos) |
             Value::Struct(heap_pos) |
             Value::Union(_, heap_pos) => {
                 // Reference to some dynamic thing which might contain ports,
                 // so recurse
                 let heap_region = &group.regions[*heap_pos as usize];
                 for (value_index, embedded_value) in heap_region.iter().enumerate() {
                     let value_location = ValueId::Heap(*heap_pos, value_index as u32);
                     find_port_in_value(group, embedded_value, value_location, ports);
+                }
             },
             _ => {}, // values we don't care about
+        }
+    }
     // Clear the ports, then scan all the available values
     ports.clear();
     for (value_index, value) in &value_group.values.iter().enumerate() {
         find_port_in_value(value_group, value, ValueId::Stack(value_index as u32), ports);
+    }
+}
@@ \ No newline at end of file @@

src/runtime2/component/component_context.rs

➞

Show inline comments

@@ @@ -29,6 +29,7 @@ pub enum PortStateFlag { @@
     Closed = 0x01, // If not closed, then the port is open
     BlockedDueToPeerChange = 0x02, // busy changing peers, hence use of port is temporarily blocked
     BlockedDueToFullBuffers = 0x04,
     Transmitted, // Transmitted, so cannot be used anymore
 }
 #[derive(Copy, Clone)]
@@ @@ -307,11 +308,6 @@ impl CompCtx { @@
     // Local utilities
     // -------------------------------------------------------------------------
     #[inline]
     fn requires_peer_reference(port: &Port, self_id: CompId, required_if_closed: bool) -> bool {
         return (!port.state.is_closed() || required_if_closed) && port.peer_comp_id != self_id;
     }
     fn must_get_port_index(&self, handle: LocalPortHandle) -> usize {
         for (index, port) in self.ports.iter().enumerate() {
             if port.self_id == handle.0 {

src/runtime2/component/component_pdl.rs

➞

Show inline comments

@@ @@ -774,47 +774,6 @@ impl CompPDL { @@
+    }
+}
 /// Recursively goes through the value group, attempting to find ports.
 /// Duplicates will only be added once.
 pub(crate) fn find_ports_in_value_group(value_group: &ValueGroup, ports: &mut Vec<PortId>) {
     // Helper to check a value for a port and recurse if needed.
     fn find_port_in_value(group: &ValueGroup, value: &Value, ports: &mut Vec<PortId>) {
         match value {
             Value::Input(port_id) | Value::Output(port_id) => {
                 // This is an actual port
                 let cur_port = PortId(port_id.id);
                 for prev_port in ports.iter() {
                     if *prev_port == cur_port {
                         // Already added
                         return;
+                    }
+                }
                 ports.push(cur_port);
             },
             Value::Array(heap_pos) |
             Value::Message(heap_pos) |
             Value::String(heap_pos) |
             Value::Struct(heap_pos) |
             Value::Union(_, heap_pos) => {
                 // Reference to some dynamic thing which might contain ports,
                 // so recurse
                 let heap_region = &group.regions[*heap_pos as usize];
                 for embedded_value in heap_region {
                     find_port_in_value(group, embedded_value, ports);
+                }
             },
             _ => {}, // values we don't care about
+        }
+    }
     // Clear the ports, then scan all the available values
     ports.clear();
     for value in &value_group.values {
         find_port_in_value(value_group, value, ports);
+    }
+}
 struct ValueGroupPortIter<'a> {
     group: &'a mut ValueGroup,
     heap_stack: Vec<(usize, usize)>,

0 comments (0 inline, 0 general)