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

 pub(crate) use core::{
     cmp::Ordering,
     fmt::{Debug, Formatter},
     hash::Hash,
     ops::Range,
     time::Duration,
 };
 // pub(crate) use indexmap::IndexSet;
 pub(crate) use maplit::hashmap;
 pub(crate) use mio::{
     net::{TcpListener, TcpStream},
     Events, Interest, Poll, Token,
 };
 pub(crate) use std::{
     collections::{BTreeMap, HashMap, HashSet},
     convert::TryInto,
     io::{Read, Write},
     net::SocketAddr,
     sync::Arc,
     time::Instant,
 };
 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 {
     Inconsistent,
     ComponentExit,
     SyncBlockStart,
+}
 #[derive(Debug, Clone)]
 pub(crate) enum SyncBlocker {
     Inconsistent,
     SyncBlockEnd,
     CouldntReadMsg(PortId),
     CouldntCheckFiring(PortId),
     PutMsg(PortId, Payload),
+}
 ///////////////////// IMPL /////////////////////
 impl U32Stream {
     pub(crate) fn next(&mut self) -> u32 {
         if self.next == u32::MAX {
             panic!("NO NEXT!")
+        }
         self.next += 1;
         self.next - 1
+    }
+}
 impl From<Id> for PortId {
@@ @@ -129,58 +129,63 @@ impl Payload { @@
+    }
     pub fn as_mut_slice(&mut self) -> &mut [u8] {
         Arc::make_mut(&mut self.0) as _
+    }
     pub fn concatenate_with(&mut self, other: &Self) {
         let bytes = other.as_slice().iter().copied();
         let me = Arc::make_mut(&mut self.0);
         me.extend(bytes);
+    }
+}
 impl serde::Serialize for Payload {
     fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
     where
         S: serde::Serializer,
+    {
         let inner: &Vec<u8> = &self.0;
         inner.serialize(serializer)
+    }
+}
 impl<'de> serde::Deserialize<'de> for Payload {
     fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
     where
         D: serde::Deserializer<'de>,
+    {
         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

@@ @@ -48,96 +48,97 @@ pub(crate) struct SyncProtoContext<'a> { @@
 #[derive(Debug)]
 struct RoundOk {
     batch_index: usize,
     gotten: HashMap<PortId, Payload>,
+}
 #[derive(Default)]
 struct VecSet<T: std::cmp::Ord> {
     // invariant: ordered, deduplicated
     vec: Vec<T>,
+}
 #[derive(Debug, Clone, Copy, Eq, PartialEq, Hash, serde::Serialize, serde::Deserialize)]
 enum ComponentId {
     Native,
     Proto(ProtoComponentId),
+}
 #[derive(Debug, Clone, Copy, Eq, PartialEq, Hash, serde::Serialize, serde::Deserialize)]
 enum Route {
     LocalComponent(ComponentId),
     Endpoint { index: usize },
+}
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 struct MyPortInfo {
     polarity: Polarity,
     port: PortId,
+}
 #[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,
     Equivalent,
     New(Predicate),
     Nonexistant,
+}
 struct Endpoint {
     inbox: Vec<u8>,
     stream: TcpStream,
+}
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 struct ProtoComponent {
     state: ComponentState,
     ports: HashSet<PortId>,
+}
 #[derive(Debug, Clone)]
 struct EndpointSetup {
     sock_addr: SocketAddr,
     endpoint_polarity: EndpointPolarity,
+}
 #[derive(Debug)]
 struct EndpointExt {
     endpoint: Endpoint,
     getter_for_incoming: PortId,
+}
 #[derive(Debug)]

src/runtime/setup.rs

➞

Show inline comments

@@ @@ -598,129 +598,139 @@ fn session_optimize( @@
             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(SessionGather { unoptimized_map: child_unoptimized_map }) => {
                 if !awaiting.remove(&recv_index) {
                     log!(
                         cu.logger,
                         "Wasn't expecting session info from {:?}. Got {:?}",
                         recv_index,
                         &child_unoptimized_map
                     );
                     return Err(SetupAlgMisbehavior);
+                }
                 unoptimized_map.extend(child_unoptimized_map.into_iter());
+            }
             msg @ S(YouAreMyParent)
             | msg @ S(MyPortInfo(..))
             | msg @ S(LeaderAnnounce { .. })
             | msg @ S(LeaderWave { .. }) => {
                 log!(cu.logger, "discarding old message {:?} during election", msg);
+            }
             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);
+                    }
                     break 'scatter_loop optimized_map;
+                }
                 msg @ Msg::CommMsg { .. } => {
                     log!(cu.logger, "delaying msg {:?} during scatter recv", msg);
                     comm.endpoint_manager.delayed_messages.push((recv_index, msg));
+                }
                 msg @ S(SessionGather { .. })
                 | msg @ S(YouAreMyParent)
                 | msg @ S(MyPortInfo(..))
                 | msg @ S(LeaderAnnounce { .. })
                 | msg @ S(LeaderWave { .. }) => {
                     log!(cu.logger, "discarding old message {:?} during election", msg);
+                }
+            }
+        }
     } else {
         // by computing it myself
         log!(cu.logger, "I am the leader! I will optimize this session");
         leader_session_map_optimize(&mut *cu.logger, unoptimized_map)?
     };
     log!(
         cu.logger,
         "Optimized info map is {:?}. Sending to children {:?}",
         &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)