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

Changeset - bd16d99233dd

Parent rev.

Child rev.

[Not reviewed]

v0.4.0

0 4 0

Christopher Esterhuyse - 5 years ago 2020-07-01 10:16:17
christopher.esterhuyse@gmail.com

added endpoint->port field to session transform structure

4 files changed with 21 insertions and 6 deletions:

src/common.rs

src/macros.rs

src/runtime/mod.rs

src/runtime/setup.rs

0 comments (0 inline, 0 general)

src/common.rs

➞

Show inline comments

@@ @@ -29,49 +29,49 @@ pub(crate) use Polarity::*; @@
 pub type ConnectorId = u32;
 pub type PortSuffix = u32;
 #[derive(
     Copy, Clone, Eq, PartialEq, Ord, Hash, PartialOrd, serde::Serialize, serde::Deserialize,
 )]
 // acquired via error in the Rust API
 pub struct ProtoComponentId(Id);
 #[derive(
     Copy, Clone, Eq, PartialEq, Ord, Hash, PartialOrd, serde::Serialize, serde::Deserialize,
 )]
 #[repr(C)]
 pub struct Id {
     pub(crate) connector_id: ConnectorId,
     pub(crate) u32_suffix: PortSuffix,
+}
 #[derive(Debug, Default)]
 pub struct U32Stream {
     next: u32,
+}
 #[derive(
     Copy, Clone, Eq, PartialEq, Ord, Hash, PartialOrd, serde::Serialize, serde::Deserialize,
 )]
 #[repr(transparent)]
 pub struct PortId(Id);
-#[derive(Default, Debug, Clone, Eq, PartialEq, Ord, PartialOrd)]
 #[derive(Default, Clone, Eq, PartialEq, Ord, PartialOrd)]
 pub struct Payload(Arc<Vec<u8>>);
 #[derive(
     Debug, Eq, PartialEq, Clone, Hash, Copy, Ord, PartialOrd, serde::Serialize, serde::Deserialize,
 )]
 #[repr(C)]
 pub enum Polarity {
     Putter, // output port (from the perspective of the component)
     Getter, // input port (from the perspective of the component)
+}
 #[derive(
     Debug, Eq, PartialEq, Clone, Hash, Copy, Ord, PartialOrd, serde::Serialize, serde::Deserialize,
 )]
 #[repr(C)]
 pub enum EndpointPolarity {
     Active,  // calls connect()
     Passive, // calls bind() listen() accept()
+}
 #[derive(
     Copy, Clone, Eq, PartialEq, Ord, Hash, PartialOrd, serde::Serialize, serde::Deserialize,
 )]
 pub(crate) struct FiringVar(pub(crate) PortId);
 #[derive(Debug, Clone)]
 pub(crate) enum NonsyncBlocker {
@@ @@ -153,34 +153,39 @@ impl<'de> serde::Deserialize<'de> for Payload { @@
         let inner: Vec<u8> = Vec::deserialize(deserializer)?;
         Ok(Self(Arc::new(inner)))
+    }
+}
 impl From<Vec<u8>> for Payload {
     fn from(s: Vec<u8>) -> Self {
         Self(s.into())
+    }
+}
 impl Debug for PortId {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         write!(f, "ptID({}'{})", self.0.connector_id, self.0.u32_suffix)
+    }
+}
 impl Debug for FiringVar {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         write!(f, "fvID({}'{})", (self.0).0.connector_id, (self.0).0.u32_suffix)
+    }
+}
 impl Debug for ProtoComponentId {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         write!(f, "pcID({}'{})", self.0.connector_id, self.0.u32_suffix)
+    }
+}
 impl Debug for Payload {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         write!(f, "Payload{:x?}", self.as_slice())
+    }
+}
 impl std::ops::Not for Polarity {
     type Output = Self;
     fn not(self) -> Self::Output {
         use Polarity::*;
         match self {
             Putter => Getter,
             Getter => Putter,
+        }
+    }
+}

src/macros.rs

➞

Show inline comments

 macro_rules! endptlog {
     ($logger:expr, $($arg:tt)*) => {{
         let w = $logger.line_writer();
         let _ = write!(w, "[ENDPT]");
         let _ = writeln!(w, $($arg)*);
         // let w = $logger.line_writer();
         // let _ = writeln!(w, $($arg)*);
     }};
+}
 macro_rules! log {
     ($logger:expr, $($arg:tt)*) => {{
         let _ = writeln!($logger.line_writer(), $($arg)*);
     }};
+}

src/runtime/mod.rs

➞

Show inline comments

@@ @@ -72,48 +72,49 @@ struct MyPortInfo { @@
+}
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 enum Decision {
     Failure,
     Success(Predicate),
+}
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 enum Msg {
     SetupMsg(SetupMsg),
     CommMsg(CommMsg),
+}
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 enum SetupMsg {
     MyPortInfo(MyPortInfo),
     LeaderWave { wave_leader: ConnectorId },
     LeaderAnnounce { tree_leader: ConnectorId },
     YouAreMyParent,
     SessionGather { unoptimized_map: HashMap<ConnectorId, SessionInfo> },
     SessionScatter { optimized_map: HashMap<ConnectorId, SessionInfo> },
+}
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 struct SessionInfo {
     serde_proto_description: SerdeProtocolDescription,
     port_info: PortInfo,
     getter_for_incoming: Vec<PortId>,
     proto_components: HashMap<ProtoComponentId, ProtoComponent>,
+}
 #[derive(Debug, Clone)]
 struct SerdeProtocolDescription(Arc<ProtocolDescription>);
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 struct CommMsg {
     round_index: usize,
     contents: CommMsgContents,
+}
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 enum CommMsgContents {
     SendPayload(SendPayloadMsg),
     Suggest { suggestion: Decision }, // SINKWARD
     Announce { decision: Decision },  // SINKAWAYS
+}
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 struct SendPayloadMsg {
     predicate: Predicate,
     payload: Payload,
+}
 #[derive(Debug, PartialEq)]
 enum CommonSatResult {
     FormerNotLatter,
     LatterNotFormer,

src/runtime/setup.rs

➞

Show inline comments

@@ @@ -622,48 +622,54 @@ fn session_optimize( @@
             msg @ S(SessionScatter { .. }) => {
                 log!(
                     cu.logger,
                     "Endpoint {:?} sent unexpected scatter! {:?} I've not contributed yet!",
                     recv_index,
                     &msg
                 );
                 return Err(SetupAlgMisbehavior);
+            }
             msg @ Msg::CommMsg(..) => {
                 log!(cu.logger, "delaying msg {:?} during session optimization", msg);
                 comm.endpoint_manager.delayed_messages.push((recv_index, msg));
+            }
+        }
+    }
     log!(
         cu.logger,
         "Gathered all children's maps. ConnectorId set is... {:?}",
         unoptimized_map.keys()
     );
     let my_session_info = SessionInfo {
         port_info: cu.port_info.clone(),
         proto_components: cu.proto_components.clone(),
         serde_proto_description: SerdeProtocolDescription(cu.proto_description.clone()),
         getter_for_incoming: comm
             .endpoint_manager
             .endpoint_exts
             .iter()
             .map(|ee| ee.getter_for_incoming)
             .collect(),
     };
     unoptimized_map.insert(cu.id_manager.connector_id, my_session_info);
     log!(cu.logger, "Inserting my own info. Unoptimized subtree map is {:?}", &unoptimized_map);
     // acquire the optimized info...
     let optimized_map = if let Some(parent) = comm.neighborhood.parent {
         // ... as a message from my parent
         log!(cu.logger, "Forwarding gathered info to parent {:?}", parent);
         let msg = S(SessionGather { unoptimized_map });
         comm.endpoint_manager.send_to_setup(parent, &msg)?;
         'scatter_loop: loop {
             log!(
                 cu.logger,
                 "Session scatter recv loop. awaiting info from children {:?}...",
                 awaiting.iter()
             );
             let (recv_index, msg) =
                 comm.endpoint_manager.try_recv_any_setup(&mut *cu.logger, deadline)?;
             log!(cu.logger, "Received from index {:?} msg {:?}", &recv_index, &msg);
             match msg {
                 S(SessionScatter { optimized_map }) => {
                     if recv_index != parent {
                         log!(cu.logger, "I expected the scatter from my parent only!");
                         return Err(SetupAlgMisbehavior);
@@ @@ -694,33 +700,37 @@ fn session_optimize( @@
         &optimized_map,
         comm.neighborhood.children.iter()
     );
     log!(cu.logger, "All session info dumped!: {:#?}", &optimized_map);
     let optimized_info =
         optimized_map.get(&cu.id_manager.connector_id).expect("HEY NO INFO FOR ME?").clone();
     let msg = S(SessionScatter { optimized_map });
     for &child in comm.neighborhood.children.iter() {
         comm.endpoint_manager.send_to_setup(child, &msg)?;
+    }
     apply_optimizations(cu, comm, optimized_info)?;
     log!(cu.logger, "Session optimizations applied");
     Ok(())
+}
 fn leader_session_map_optimize(
     logger: &mut dyn Logger,
     unoptimized_map: HashMap<ConnectorId, SessionInfo>,
 ) -> Result<HashMap<ConnectorId, SessionInfo>, ConnectError> {
     log!(logger, "Session map optimize START");
     log!(logger, "Session map optimize END");
     Ok(unoptimized_map)
+}
 fn apply_optimizations(
     cu: &mut ConnectorUnphased,
-    _comm: &mut ConnectorCommunication,
+    comm: &mut ConnectorCommunication,
     session_info: SessionInfo,
 ) -> Result<(), ConnectError> {
     let SessionInfo { proto_components, port_info, serde_proto_description } = session_info;
     let SessionInfo { proto_components, port_info, serde_proto_description, getter_for_incoming } =
         session_info;
     cu.port_info = port_info;
     cu.proto_components = proto_components;
     cu.proto_description = serde_proto_description.0;
     for (ee, getter) in comm.endpoint_manager.endpoint_exts.iter_mut().zip(getter_for_incoming) {
         ee.getter_for_incoming = getter;
+    }
     Ok(())
+}

0 comments (0 inline, 0 general)