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

Changeset - 65390fb1cdbc

Parent rev.

Child rev.

[Not reviewed]

0 9 1

Christopher Esterhuyse - 5 years ago 2020-06-24 12:12:27
christopher.esterhuyse@gmail.com

bugfixing and rewrote dist algs to be more readable

10 files changed with 446 insertions and 243 deletions:

src/common.rs

src/lib.rs

src/macros.rs

src/runtime/communication.rs

src/runtime/endpoints.rs

src/runtime/error.rs

src/runtime/logging.rs

src/runtime/mod.rs

src/runtime/setup.rs

170

108

src/runtime/tests.rs

106

0 comments (0 inline, 0 general)

src/common.rs

➞

Show inline comments

@@ @@ -19,32 +19,32 @@ pub use mio::{ @@
 pub use std::{
     collections::{hash_map::Entry, BTreeMap, HashMap, HashSet},
     convert::TryInto,
     io::{Read, Write},
     net::SocketAddr,
     sync::Arc,
     time::Instant,
 };
 pub use Polarity::*;
 ///////////////////// DEFS /////////////////////
-pub type ControllerId = u32;
+pub type ConnectorId = u32;
 pub type PortSuffix = u32;
 #[derive(
     Copy, Clone, Eq, PartialEq, Ord, Hash, PartialOrd, serde::Serialize, serde::Deserialize,
 )]
 pub struct Id {
-    pub(crate) controller_id: ControllerId,
+    pub(crate) connector_id: ConnectorId,
     pub(crate) u32_suffix: PortSuffix,
+}
 #[derive(Debug, Default)]
 pub struct U32Stream {
     next: u32,
+}
 // globally unique
 #[derive(
     Copy, Clone, Eq, PartialEq, Ord, Hash, PartialOrd, serde::Serialize, serde::Deserialize,
 )]
@@ @@ -170,35 +170,35 @@ impl<'de> serde::Deserialize<'de> for Payload { @@
 impl std::iter::FromIterator<u8> for Payload {
     fn from_iter<I: IntoIterator<Item = u8>>(it: I) -> Self {
         Self(Arc::new(it.into_iter().collect()))
+    }
+}
 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, "PID<{},{}>", self.0.controller_id, self.0.u32_suffix)
+        write!(f, "PID<{},{}>", self.0.connector_id, self.0.u32_suffix)
+    }
+}
 impl Debug for FiringVar {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
-        write!(f, "VID<{},{}>", (self.0).0.controller_id, (self.0).0.u32_suffix)
+        write!(f, "VID<{},{}>", (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, "ProtoComponentId({},{})", self.0.controller_id, self.0.u32_suffix)
+        write!(f, "ProtoComponentId({},{})", self.0.connector_id, self.0.u32_suffix)
+    }
+}
 impl std::ops::Not for Polarity {
     type Output = Self;
     fn not(self) -> Self::Output {
         use Polarity::*;
         match self {
             Putter => Getter,
             Getter => Putter,
+        }
+    }
+}

src/lib.rs

➞

Show inline comments

 #[macro_use]
 mod macros;
 mod common;
 mod protocol;
 mod runtime;
 // #[cfg(test)]
 // mod test;
-pub use common::{ControllerId, Polarity, PortId};
+pub use common::{ConnectorId, Polarity, PortId};
 pub use protocol::ProtocolDescription;
 pub use runtime::{error, Connector, EndpointSetup, FileLogger, VecLogger};
 // #[cfg(feature = "ffi")]
 // pub use runtime::ffi;

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)*);
     }};
+}
 macro_rules! log {
     ($logger:expr, $($arg:tt)*) => {{
-        let _ = write!($logger.line_writer(), $($arg)*).unwrap();
+        let _ = writeln!($logger.line_writer(), $($arg)*);
     }};
+}
 // macro_rules! assert_let {
 //     ($pat:pat = $expr:expr => $work:expr) => {
 //         if let $pat = $expr {
 //             $work
 //         } else {
 //             panic!("assert_let failed");
 //         }
 //     };
 // }
 // #[test]
 // fn assert_let() {
 //     let x = Some(5);
 //     let z = assert_let![Some(y) = x => {
 //         println!("{:?}", y);
 //         3
 //     }];
 //     println!("{:?}", z);
 // }
 // #[test]
 // #[should_panic]
 // fn must_let_panic() {
 //     let x: Option<u32> = None;
 //     assert_let![Some(y) = x => {
 //         println!("{:?}", y);
 //     }];
 // }

src/runtime/communication.rs

➞

Show inline comments

@@ @@ -339,25 +339,25 @@ impl Connector { @@
                             None => {
                                 log!(logger, "No parent. Deciding on solution {:?}", &solution);
                                 break 'undecided Decision::Success(solution);
+                            }
+                        }
+                    }
                     // stuck! make progress by receiving a msg
                     // try recv messages arriving through endpoints
                     log!(logger, "No decision yet. Let's recv an endpoint msg...");
+                    {
                         let (endpoint_index, msg) = loop {
                             match endpoint_manager.try_recv_any_comms(deadline)? {
+                            match endpoint_manager.try_recv_any_comms(logger, deadline)? {
                                 None => {
                                     log!(
                                         logger,
                                         "Reached user-defined deadling without decision..."
                                     );
                                     if let Some(parent) = neighborhood.parent {
                                         log!(
                                             logger,
                                             "Sending failure request to parent index {}",
                                             parent
                                         );
                                         let msg = Msg::CommMsg(CommMsg {
@@ @@ -413,25 +413,25 @@ impl Connector { @@
                                 let getter = endpoint_manager.endpoint_exts[endpoint_index]
                                     .getter_for_incoming;
                                 assert!(self.port_info.polarities.get(&getter) == Some(&Getter));
                                 log!(
                                     logger,
                                     "Msg routed to getter port {:?}. Buffer for recv loop",
                                     getter,
                                 );
                                 payloads_to_get.push((getter, send_payload_msg));
+                            }
                             CommMsgContents::Suggest { suggestion } => {
                                 // only accept this control msg through a child endpoint
-                                if neighborhood.children.binary_search(&endpoint_index).is_ok() {
+                                if neighborhood.children.contains(&endpoint_index) {
                                     match suggestion {
                                         Decision::Success(predicate) => {
                                             // child solution contributes to local solution
                                             log!(
                                                 logger,
                                                 "Child provided solution {:?}",
                                                 &predicate
                                             );
                                             let route = Route::Endpoint { index: endpoint_index };
                                             solution_storage.submit_and_digest_subtree_solution(
                                                 logger, route, predicate,
                                             );
@@ @@ -483,25 +483,25 @@ impl Connector { @@
                 });
                 log!(
                     logger,
                     "Announcing decision {:?} through child endpoints {:?}",
                     &msg,
                     &neighborhood.children
                 );
                 for &child in neighborhood.children.iter() {
                     endpoint_manager.send_to(child, &msg).unwrap();
+                }
                 *round_result = match decision {
-                    Decision::Failure => Err(DistributedTimeout),
+                    Decision::Failure => Err(RoundFailure),
                     Decision::Success(predicate) => {
                         // commit changes to component states
                         self.proto_components.clear();
                         self.proto_components.extend(
                             branching_proto_components
                                 .into_iter()
                                 .map(|(id, bpc)| (id, bpc.collapse_with(&predicate))),
                         );
                         Ok(Some(branching_native.collapse_with(&predicate)))
+                    }
                 };
                 log!(logger, "Updated round_result to {:?}", round_result);

src/runtime/endpoints.rs

➞

Show inline comments

@@ @@ -4,31 +4,39 @@ struct MonitoredReader<R: Read> { @@
     bytes: usize,
     r: R,
+}
 #[derive(Debug)]
 enum TryRecyAnyError {
     Timeout,
     PollFailed,
     EndpointError { error: EndpointError, index: usize },
+}
 /////////////////////
 fn would_block(err: &std::io::Error) -> bool {
     err.kind() == std::io::ErrorKind::WouldBlock
+}
 impl Endpoint {
     pub fn try_recv<T: serde::de::DeserializeOwned>(&mut self) -> Result<Option<T>, EndpointError> {
     pub fn try_recv<T: serde::de::DeserializeOwned>(
         &mut self,
         logger: &mut dyn Logger,
     ) -> Result<Option<T>, EndpointError> {
         use EndpointError::*;
         // populate inbox as much as possible
         'read_loop: loop {
             match self.stream.read_to_end(&mut self.inbox) {
                 Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => break 'read_loop,
             let res = self.stream.read_to_end(&mut self.inbox);
             endptlog!(logger, "Stream read to end {:?}", &res);
             match res {
                 Err(e) if would_block(&e) => break 'read_loop,
                 Ok(0) => break 'read_loop,
                 Ok(_) => (),
                 Err(_e) => return Err(BrokenEndpoint),
+            }
+        }
         let mut monitored = MonitoredReader::from(&self.inbox[..]);
         match bincode::deserialize_from(&mut monitored) {
             Ok(msg) => {
                 let msg_size = monitored.bytes_read();
                 self.inbox.drain(0..(msg_size.try_into().unwrap()));
                 Ok(Some(msg))
+            }
@@ @@ -39,89 +47,117 @@ impl Endpoint { @@
                 _ => Err(MalformedMessage),
                 // println!("SERDE ERRKIND {:?}", e);
                 // Err(MalformedMessage)
             },
+        }
+    }
     pub fn send<T: serde::ser::Serialize>(&mut self, msg: &T) -> Result<(), EndpointError> {
         bincode::serialize_into(&mut self.stream, msg).map_err(|_| EndpointError::BrokenEndpoint)
+    }
+}
 impl EndpointManager {
     pub fn index_iter(&self) -> Range<usize> {
 ..self.num_endpoints()
+    }
     pub fn num_endpoints(&self) -> usize {
         self.endpoint_exts.len()
+    }
     pub fn send_to_setup(&mut self, index: usize, msg: &Msg) -> Result<(), ConnectError> {
         let endpoint = &mut self.endpoint_exts[index].endpoint;
         endpoint.send(msg).map_err(|err| {
             ConnectError::EndpointSetupError(endpoint.stream.local_addr().unwrap(), err)
         })
+    }
     pub fn send_to(&mut self, index: usize, msg: &Msg) -> Result<(), EndpointError> {
         self.endpoint_exts[index].endpoint.send(msg)
+    }
     pub fn try_recv_any_comms(
         &mut self,
         logger: &mut dyn Logger,
         deadline: Option<Instant>,
     ) -> Result<Option<(usize, Msg)>, SyncError> {
         use {SyncError as Se, TryRecyAnyError as Trae};
         match self.try_recv_any(deadline) {
+        match self.try_recv_any(logger, deadline) {
             Ok(tup) => Ok(Some(tup)),
             Err(Trae::Timeout) => Ok(None),
             Err(Trae::PollFailed) => Err(Se::PollFailed),
             Err(Trae::EndpointError { error, index }) => Err(Se::BrokenEndpoint(index)),
+            Err(Trae::EndpointError { error: _, index }) => Err(Se::BrokenEndpoint(index)),
+        }
+    }
     pub fn try_recv_any_setup(
         &mut self,
         logger: &mut dyn Logger,
         deadline: Option<Instant>,
     ) -> Result<(usize, Msg), ConnectError> {
         use {ConnectError as Ce, TryRecyAnyError as Trae};
         self.try_recv_any(deadline).map_err(|err| match err {
+        self.try_recv_any(logger, deadline).map_err(|err| match err {
             Trae::Timeout => Ce::Timeout,
             Trae::PollFailed => Ce::PollFailed,
             Trae::EndpointError { error, index } => Ce::EndpointSetupError(
                 self.endpoint_exts[index].endpoint.stream.local_addr().unwrap(),
                 error,
             ),
         })
+    }
     fn try_recv_any(&mut self, deadline: Option<Instant>) -> Result<(usize, Msg), TryRecyAnyError> {
         use TryRecyAnyError::*;
     fn try_recv_any(
         &mut self,
         logger: &mut dyn Logger,
         deadline: Option<Instant>,
     ) -> Result<(usize, Msg), TryRecyAnyError> {
         use TryRecyAnyError as Trea;
         // 1. try messages already buffered
         if let Some(x) = self.undelayed_messages.pop() {
             endptlog!(logger, "RECV undelayed_msg {:?}", &x);
             return Ok(x);
+        }
         loop {
             // 2. try read a message from an endpoint that raised an event with poll() but wasn't drained
             while let Some(index) = self.polled_undrained.pop() {
                 let endpoint = &mut self.endpoint_exts[index].endpoint;
                 if let Some(msg) =
                     endpoint.try_recv().map_err(|error| EndpointError { error, index })?
                 if let Some(msg) = endpoint
                     .try_recv(logger)
                     .map_err(|error| Trea::EndpointError { error, index })?
+                {
                     if !endpoint.inbox.is_empty() {
                     endptlog!(logger, "RECV polled_undrained {:?}", &msg);
                     // if !endpoint.inbox.is_empty() {
                     // there may be another message waiting!
                     self.polled_undrained.insert(index);
+                    }
+                    // }
                     return Ok((index, msg));
+                }
+            }
             // 3. No message yet. Do we have enough time to poll?
             let remaining = if let Some(deadline) = deadline {
                 Some(deadline.checked_duration_since(Instant::now()).ok_or(Timeout)?)
+                Some(deadline.checked_duration_since(Instant::now()).ok_or(Trea::Timeout)?)
             } else {
                 None
             };
             self.poll.poll(&mut self.events, remaining).map_err(|_| PollFailed)?;
+            self.poll.poll(&mut self.events, remaining).map_err(|_| Trea::PollFailed)?;
             for event in self.events.iter() {
                 let Token(index) = event.token();
                 self.polled_undrained.insert(index);
                 endptlog!(
                     logger,
                     "RECV poll event {:?} for endpoint index {:?}. undrained: {:?}",
                     &event,
                     index,
                     self.polled_undrained.iter()
                 );
                 if event.is_error() {
                     return Err(Trea::EndpointError {
                         error: EndpointError::BrokenEndpoint,
                         index,
                     });
+                }
+            }
             self.events.clear();
+        }
+    }
     pub fn undelay_all(&mut self) {
         if self.undelayed_messages.is_empty() {
             // fast path
             std::mem::swap(&mut self.delayed_messages, &mut self.undelayed_messages);
             return;
+        }
         // slow path
         self.undelayed_messages.extend(self.delayed_messages.drain(..));

src/runtime/error.rs

➞

Show inline comments

 use crate::common::*;
 #[derive(Debug, Clone)]
 pub enum EndpointError {
     MalformedMessage,
     BrokenEndpoint,
 #[derive(Debug)]
 pub enum ConnectError {
     BindFailed(SocketAddr),
     PollInitFailed,
     Timeout,
     PollFailed,
     AcceptFailed(SocketAddr),
     AlreadyConnected,
     PortPeerPolarityMismatch(PortId),
     EndpointSetupError(SocketAddr, EndpointError),
     SetupAlgMisbehavior,
+}
 ////////////////////////
 #[derive(Debug, Clone)]
 pub enum SyncError {
     Timeout,
     NotConnected,
     InconsistentProtoComponent(ProtoComponentId),
     IndistinguishableBatches([usize; 2]),
-    DistributedTimeout,
+    RoundFailure,
     PollFailed,
     BrokenEndpoint(usize),
+}
 #[derive(Debug, Clone)]
 pub enum EndpointError {
     MalformedMessage,
     BrokenEndpoint,
+}
 #[derive(Debug)]
 pub enum PortOpError {
     WrongPolarity,
     NotConnected,
     MultipleOpsOnPort,
     PortUnavailable,
+}
 #[derive(Debug, Eq, PartialEq)]
 pub enum GottenError {
     NoPreviousRound,
     PortDidntGet,
     PreviousSyncFailed,
+}
 #[derive(Debug, Eq, PartialEq)]
 pub enum NextBatchError {
     NotConnected,
+}
 #[derive(Debug)]
 pub enum ConnectError {
     BindFailed(SocketAddr),
     PollInitFailed,
     Timeout,
     PollFailed,
     AcceptFailed(SocketAddr),
     AlreadyConnected,
     PortPeerPolarityMismatch(PortId),
     EndpointSetupError(SocketAddr, EndpointError),
+}

src/runtime/logging.rs

➞

Show inline comments

@@ new file 100644 @@
 use super::*;
 impl FileLogger {
     pub fn new(connector_id: ConnectorId, file: std::fs::File) -> Self {
         Self(connector_id, file)
+    }
+}
 impl VecLogger {
     pub fn new(connector_id: ConnectorId) -> Self {
         Self(connector_id, Default::default())
+    }
+}
 /////////////////
 impl Logger for DummyLogger {
     fn line_writer(&mut self) -> &mut dyn std::io::Write {
         self
+    }
+}
 impl Logger for VecLogger {
     fn line_writer(&mut self) -> &mut dyn std::io::Write {
         let _ = write!(&mut self.1, "CID({}): ", self.0);
         self
+    }
+}
 impl Logger for FileLogger {
     fn line_writer(&mut self) -> &mut dyn std::io::Write {
         let _ = write!(&mut self.1, "CID({}): ", self.0);
         &mut self.1
+    }
+}
 ///////////////////
 impl Drop for VecLogger {
     fn drop(&mut self) {
         let stdout = std::io::stderr();
         let mut lock = stdout.lock();
         writeln!(lock, "--- DROP LOG DUMP ---").unwrap();
         let _ = std::io::Write::write(&mut lock, self.1.as_slice());
+    }
+}
 impl std::io::Write for VecLogger {
     fn flush(&mut self) -> Result<(), std::io::Error> {
         Ok(())
+    }
     fn write(&mut self, data: &[u8]) -> Result<usize, std::io::Error> {
         self.1.extend_from_slice(data);
         Ok(data.len())
+    }
+}
 impl std::io::Write for DummyLogger {
     fn flush(&mut self) -> Result<(), std::io::Error> {
         Ok(())
+    }
     fn write(&mut self, bytes: &[u8]) -> Result<usize, std::io::Error> {
         Ok(bytes.len())
+    }
+}

src/runtime/mod.rs

➞

Show inline comments

 mod communication;
 mod endpoints;
 pub mod error;
 mod logging;
 mod setup;
 #[cfg(test)]
 mod tests;
 use crate::common::*;
 use error::*;
 #[derive(Debug)]
 pub struct VecSet<T: std::cmp::Ord> {
     // invariant: ordered, deduplicated
     vec: Vec<T>,
+}
 #[derive(Debug, Clone, Copy, Eq, PartialEq, Hash)]
 pub enum LocalComponentId {
     Native,
     Proto(ProtoComponentId),
+}
 #[derive(Debug, Clone, Copy, Eq, PartialEq, Hash)]
 pub enum Route {
     LocalComponent(LocalComponentId),
     Endpoint { index: usize },
+}
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 pub struct MyPortInfo {
@@ @@ -28,26 +34,26 @@ pub struct MyPortInfo { @@
 pub enum Decision {
     Failure,
     Success(Predicate),
+}
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 pub enum Msg {
     SetupMsg(SetupMsg),
     CommMsg(CommMsg),
+}
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 pub enum SetupMsg {
     MyPortInfo(MyPortInfo),
     LeaderEcho { maybe_leader: ControllerId },
     LeaderAnnounce { leader: ControllerId },
     LeaderWave { wave_leader: ConnectorId },
     LeaderAnnounce { tree_leader: ConnectorId },
     YouAreMyParent,
+}
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 pub struct CommMsg {
     pub round_index: usize,
     pub contents: CommMsgContents,
+}
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 pub enum CommMsgContents {
     SendPayload(SendPayloadMsg),
     Suggest { suggestion: Decision }, // SINKWARD
     Announce { decision: Decision },  // SINKAWAYS
@@ @@ -69,52 +75,52 @@ pub struct Endpoint { @@
     inbox: Vec<u8>,
     stream: TcpStream,
+}
 #[derive(Debug, Clone)]
 pub struct ProtoComponent {
     state: ComponentState,
     ports: HashSet<PortId>,
+}
 pub trait Logger: Debug {
     fn line_writer(&mut self) -> &mut dyn std::io::Write;
+}
 #[derive(Debug)]
-pub struct VecLogger(ControllerId, Vec<u8>);
+pub struct VecLogger(ConnectorId, Vec<u8>);
 #[derive(Debug)]
 pub struct DummyLogger;
 #[derive(Debug)]
-pub struct FileLogger(ControllerId, std::fs::File);
+pub struct FileLogger(ConnectorId, std::fs::File);
 #[derive(Debug, Clone)]
 pub struct EndpointSetup {
     pub sock_addr: SocketAddr,
     pub is_active: bool,
+}
 #[derive(Debug)]
 pub struct EndpointExt {
     endpoint: Endpoint,
     getter_for_incoming: PortId,
+}
 #[derive(Debug)]
 pub struct Neighborhood {
     parent: Option<usize>,
-    children: Vec<usize>, // ordered, deduplicated
+    children: VecSet<usize>,
+}
 #[derive(Debug)]
 pub struct MemInMsg {
     inp: PortId,
     msg: Payload,
+}
 #[derive(Debug)]
 pub struct IdManager {
-    controller_id: ControllerId,
+    connector_id: ConnectorId,
     port_suffix_stream: U32Stream,
     proto_component_suffix_stream: U32Stream,
+}
 #[derive(Debug)]
 pub struct EndpointManager {
     // invariants:
     // 1. endpoint N is registered READ | WRITE with poller
     // 2. Events is empty
     poll: Poll,
     events: Events,
     polled_undrained: IndexSet<usize>,
     delayed_messages: Vec<(usize, Msg)>,
@@ @@ -168,46 +174,59 @@ pub struct NonsyncProtoContext<'a> { @@
     port_info: &'a mut PortInfo,
     id_manager: &'a mut IdManager,
     proto_component_ports: &'a mut HashSet<PortId>,
     unrun_components: &'a mut Vec<(ProtoComponentId, ProtoComponent)>,
+}
 pub struct SyncProtoContext<'a> {
     logger: &'a mut dyn Logger,
     predicate: &'a Predicate,
     port_info: &'a PortInfo,
     inbox: &'a HashMap<PortId, Payload>,
+}
 ////////////////
 impl<T: std::cmp::Ord> VecSet<T> {
     fn iter(&self) -> impl Iterator<Item = &T> {
         self.vec.iter()
+    }
     fn contains(&self, element: &T) -> bool {
         self.vec.binary_search(element).is_ok()
+    }
     fn new(mut vec: Vec<T>) -> Self {
         vec.sort();
         vec.dedup();
         Self { vec }
+    }
+}
 impl PortInfo {
     fn firing_var_for(&self, port: PortId) -> FiringVar {
         FiringVar(match self.polarities.get(&port).unwrap() {
             Getter => port,
             Putter => *self.peers.get(&port).unwrap(),
         })
+    }
+}
 impl IdManager {
-    fn new(controller_id: ControllerId) -> Self {
+    fn new(connector_id: ConnectorId) -> Self {
         Self {
-            controller_id,
+            connector_id,
             port_suffix_stream: Default::default(),
             proto_component_suffix_stream: Default::default(),
+        }
+    }
     fn new_port_id(&mut self) -> PortId {
-        Id { controller_id: self.controller_id, u32_suffix: self.port_suffix_stream.next() }.into()
+        Id { connector_id: self.connector_id, u32_suffix: self.port_suffix_stream.next() }.into()
+    }
     fn new_proto_component_id(&mut self) -> ProtoComponentId {
         Id {
-            controller_id: self.controller_id,
+            connector_id: self.connector_id,
             u32_suffix: self.proto_component_suffix_stream.next(),
+        }
         .into()
+    }
+}
 impl Drop for Connector {
     fn drop(&mut self) {
         log!(&mut *self.logger, "Connector dropping. Goodbye!");
+    }
+}
 impl Connector {
     pub fn swap_logger(&mut self, mut new_logger: Box<dyn Logger>) -> Box<dyn Logger> {
@@ @@ -263,76 +282,24 @@ impl Connector { @@
         self.native_ports.retain(|port| !ports.contains(port));
         // 4. add new component
         self.proto_components.insert(
             new_id,
             ProtoComponent {
                 state: self.proto_description.new_main_component(identifier, ports),
                 ports: ports.iter().copied().collect(),
             },
         );
         Ok(())
+    }
+}
 impl Logger for DummyLogger {
     fn line_writer(&mut self) -> &mut dyn std::io::Write {
         impl std::io::Write for DummyLogger {
             fn flush(&mut self) -> Result<(), std::io::Error> {
                 Ok(())
+            }
             fn write(&mut self, bytes: &[u8]) -> Result<usize, std::io::Error> {
                 Ok(bytes.len())
+            }
+        }
         self
+    }
+}
 impl VecLogger {
     pub fn new(controller_id: ControllerId) -> Self {
         Self(controller_id, Default::default())
+    }
+}
 impl Drop for VecLogger {
     fn drop(&mut self) {
         let stdout = std::io::stderr();
         let mut lock = stdout.lock();
         writeln!(lock, "--- DROP LOG DUMP ---").unwrap();
         let _ = std::io::Write::write(&mut lock, self.1.as_slice());
+    }
+}
 impl Logger for VecLogger {
     fn line_writer(&mut self) -> &mut dyn std::io::Write {
         let _ = write!(&mut self.1, "\nCID({}): ", self.0);
         self
+    }
+}
 impl FileLogger {
     pub fn new(controller_id: ControllerId, file: std::fs::File) -> Self {
         Self(controller_id, file)
+    }
+}
 impl Logger for FileLogger {
     fn line_writer(&mut self) -> &mut dyn std::io::Write {
         let _ = write!(&mut self.1, "\nCID({}): ", self.0);
         &mut self.1
+    }
+}
 impl std::io::Write for VecLogger {
     fn flush(&mut self) -> Result<(), std::io::Error> {
         Ok(())
+    }
     fn write(&mut self, data: &[u8]) -> Result<usize, std::io::Error> {
         self.1.extend_from_slice(data);
         Ok(data.len())
+    }
+}
 impl Predicate {
     #[inline]
     pub fn inserted(mut self, k: FiringVar, v: bool) -> Self {
         self.assigned.insert(k, v);
         self
+    }
     // returns true IFF self.unify would return Equivalent OR FormerNotLatter
     pub fn satisfies(&self, other: &Self) -> bool {
         let mut s_it = self.assigned.iter();
         let mut s = if let Some(s) = s_it.next() {
+            s
         } else {

src/runtime/setup.rs

➞

Show inline comments

 use crate::common::*;
 use crate::runtime::*;
 use std::io::ErrorKind::WouldBlock;
 impl Connector {
     pub fn new_simple(
         proto_description: Arc<ProtocolDescription>,
-        controller_id: ControllerId,
+        connector_id: ConnectorId,
     ) -> Self {
         let logger = Box::new(DummyLogger);
         // let logger = Box::new(DummyLogger);
         let surplus_sockets = 2;
-        Self::new(logger, proto_description, controller_id, surplus_sockets)
+        Self::new(logger, proto_description, connector_id, surplus_sockets)
+    }
     pub fn new(
         logger: Box<dyn Logger>,
+        mut logger: Box<dyn Logger>,
         proto_description: Arc<ProtocolDescription>,
-        controller_id: ControllerId,
+        connector_id: ConnectorId,
         surplus_sockets: u16,
     ) -> Self {
         log!(&mut *logger, "Created with connector_id {:?}", connector_id);
         Self {
             proto_description,
             proto_components: Default::default(),
             logger,
-            id_manager: IdManager::new(controller_id),
+            id_manager: IdManager::new(connector_id),
             native_ports: Default::default(),
             port_info: Default::default(),
             phased: ConnectorPhased::Setup { endpoint_setups: Default::default(), surplus_sockets },
+        }
+    }
     pub fn new_net_port(
         &mut self,
         polarity: Polarity,
         endpoint_setup: EndpointSetup,
     ) -> Result<PortId, ()> {
         match &mut self.phased {
             ConnectorPhased::Setup { endpoint_setups, .. } => {
@@ @@ -68,25 +69,25 @@ impl Connector { @@
                     &mut *self.logger,
                     endpoint_setups,
                     &mut self.port_info,
                     deadline,
                 )?;
                 log!(
                     self.logger,
                     "Successfully connected {} endpoints",
                     endpoint_manager.endpoint_exts.len()
                 );
                 // leader election and tree construction
                 let neighborhood = init_neighborhood(
-                    self.id_manager.controller_id,
+                    self.id_manager.connector_id,
                     &mut *self.logger,
                     &mut endpoint_manager,
                     deadline,
                 )?;
                 log!(self.logger, "Successfully created neighborhood {:?}", &neighborhood);
                 log!(self.logger, "connect() finished. setup phase complete");
                 // TODO session optimization goes here
                 self.phased = ConnectorPhased::Communication {
                     round_index: 0,
                     endpoint_manager,
                     neighborhood,
                     mem_inbox: Default::default(),
@@ @@ -206,25 +207,25 @@ fn new_endpoint_manager( @@
                             port: *local_port,
                         }));
                         endpoint
                             .send(&msg)
                             .map_err(|e| {
                                 EndpointSetupError(endpoint.stream.local_addr().unwrap(), e)
                             })
                             .unwrap();
                         log!(logger, "endpoint[{}] sent msg {:?}", index, &msg);
                         *sent_local_port = true;
+                    }
                     if event.is_readable() && recv_peer_port.is_none() {
                         let maybe_msg = endpoint.try_recv().map_err(|e| {
+                        let maybe_msg = endpoint.try_recv(logger).map_err(|e| {
                             EndpointSetupError(endpoint.stream.local_addr().unwrap(), e)
                         })?;
                         if maybe_msg.is_some() && !endpoint.inbox.is_empty() {
                             polled_undrained.insert(index);
+                        }
                         match maybe_msg {
                             None => {} // msg deserialization incomplete
                             Some(Msg::SetupMsg(SetupMsg::MyPortInfo(peer_info))) => {
                                 log!(logger, "endpoint[{}] got peer info {:?}", index, peer_info);
                                 if peer_info.polarity == local_polarity {
                                     return Err(ConnectError::PortPeerPolarityMismatch(
                                         *local_port,
@@ @@ -251,165 +252,226 @@ fn new_endpoint_manager( @@
                     if *sent_local_port && recv_peer_port.is_some() {
                         setup_incomplete.remove(&index);
                         log!(logger, "endpoint[{}] is finished!", index);
+                    }
+                }
                 Todo { todo_endpoint: TodoEndpoint::Listener(_), .. } => unreachable!(),
+            }
+        }
         events.clear();
+    }
     let endpoint_exts = todos
         .into_iter()
         .map(|Todo { todo_endpoint, local_port, .. }| EndpointExt {
         .enumerate()
         .map(|(index, Todo { todo_endpoint, local_port, .. })| EndpointExt {
             endpoint: match todo_endpoint {
                 TodoEndpoint::Endpoint(endpoint) => endpoint,
                 TodoEndpoint::Endpoint(mut endpoint) => {
                     poll.registry()
                         .reregister(&mut endpoint.stream, Token(index), Interest::READABLE)
                         .unwrap();
                     endpoint
+                }
                 TodoEndpoint::Listener(..) => unreachable!(),
             },
             getter_for_incoming: local_port,
         })
         .collect();
     Ok(EndpointManager {
         poll,
         events,
         polled_undrained,
         undelayed_messages: delayed_messages, // no longer delayed
         delayed_messages: Default::default(),
         endpoint_exts,
     })
+}
 fn init_neighborhood(
-    controller_id: ControllerId,
+    connector_id: ConnectorId,
     logger: &mut dyn Logger,
     em: &mut EndpointManager,
     deadline: Option<Instant>,
 ) -> Result<Neighborhood, ConnectError> {
     use {Msg::SetupMsg as S, SetupMsg::*};
     log!(logger, "beginning neighborhood construction");
     // 1. broadcast my ID as the first echo. await reply from all neighbors
     let echo = S(LeaderEcho { maybe_leader: controller_id });
     let mut awaiting = HashSet::with_capacity(em.endpoint_exts.len());
     for index in 0..em.endpoint_exts.len() {
         log!(logger, "{:?}'s initial echo to {:?}, {:?}", controller_id, index, &echo);
         em.send_to_setup(index, &echo)?;
     use {ConnectError::*, Msg::SetupMsg as S, SetupMsg::*};
     ////////////////////////////////
     #[derive(Debug)]
     struct WaveState {
         parent: Option<usize>,
         leader: ConnectorId,
+    }
     fn do_wave(
         em: &mut EndpointManager,
         awaiting: &mut HashSet<usize>,
         ws: &WaveState,
     ) -> Result<(), ConnectError> {
         awaiting.clear();
         let msg = S(LeaderWave { wave_leader: ws.leader });
         for index in em.index_iter() {
             if Some(index) != ws.parent {
                 em.send_to_setup(index, &msg)?;
                 awaiting.insert(index);
+            }
+        }
         Ok(())
+    }
     ///////////////////////
     /*
     Conceptually, we have two distinct disstributed algorithms back-to-back
 . Leader election using echo algorithm with extinction.
         - Each connector initiates a wave tagged with their ID
         - Connectors participate in waves of GREATER ID, abandoning previous waves
         - Only the wave of the connector with GREATEST ID completes, whereupon they are the leader
 . Tree construction
         - The leader broadcasts their leadership with msg A
         - Upon receiving their first announcement, connectors reply B, and send A to all peers
         - A controller exits once they have received A or B from each neighbor
     The actual implementation is muddier, because non-leaders aren't aware of termiantion of algorithm 1,
     so they rely on receipt of the leader's announcement to realize that algorithm 2 has begun.
     // 2. Receive incoming replies. whenever a higher-id echo arrives,
     //    adopt it as leader, sender as parent, and reset the await set.
     let mut parent: Option<usize> = None;
     let mut my_leader = controller_id;
     NOTE the distinction between PARENT and LEADER
     */
     log!(logger, "beginning neighborhood construction");
     if em.num_endpoints() == 0 {
         log!(logger, "Edge case of no neighbors! No parent an no children!");
         return Ok(Neighborhood { parent: None, children: VecSet::new(vec![]) });
+    }
     log!(logger, "Have {} endpoints. Must participate in distributed alg.", em.num_endpoints());
     let mut awaiting = HashSet::with_capacity(em.num_endpoints());
     // 1+ neighbors. Leader can only be learned by receiving messages
     // loop ends when I know my sink tree parent (implies leader was elected)
     let election_result: WaveState = {
         // initially: No parent, I'm the best leader.
         let mut best_wave = WaveState { parent: None, leader: connector_id };
         // start a wave for this initial state
         do_wave(em, &mut awaiting, &best_wave)?;
         // with 1+ neighbors, progress is only made in response to incoming messages
         em.undelay_all();
     'echo_loop: while !awaiting.is_empty() || parent.is_some() {
         let (index, msg) = em.try_recv_any_setup(deadline)?;
         log!(logger, "GOT from index {:?} msg {:?}", &index, &msg);
         'election: loop {
             log!(logger, "Election loop. awaiting {:?}...", awaiting.iter());
             let (recv_index, msg) = em.try_recv_any_setup(logger, deadline)?;
             log!(logger, "Received from index {:?} msg {:?}", &recv_index, &msg);
             match msg {
             S(LeaderAnnounce { leader }) => {
                 // someone else completed the echo and became leader first!
                 // the sender is my parent
                 parent = Some(index);
                 my_leader = leader;
                 awaiting.clear();
                 break 'echo_loop;
+            }
             S(LeaderEcho { maybe_leader }) => {
                 use Ordering::*;
                 match maybe_leader.cmp(&my_leader) {
                     Less => { /* ignore this wave */ }
                     Equal => {
                         awaiting.remove(&index);
                 S(LeaderAnnounce { tree_leader }) => {
                     let election_result =
                         WaveState { leader: tree_leader, parent: Some(recv_index) };
                     log!(logger, "Election lost! Result {:?}", &election_result);
                     assert!(election_result.leader >= best_wave.leader);
                     assert_ne!(election_result.leader, connector_id);
                     break 'election election_result;
+                }
                 S(LeaderWave { wave_leader }) => {
                     use Ordering as O;
                     match wave_leader.cmp(&best_wave.leader) {
                         O::Less => log!(
                             logger,
                             "Ignoring wave with Id {:?}<{:?}",
                             wave_leader,
                             best_wave.leader
                         ),
                         O::Greater => {
                             log!(
                                 logger,
                                 "Joining wave with Id {:?}>{:?}",
                                 wave_leader,
                                 best_wave.leader
                             );
                             best_wave = WaveState { leader: wave_leader, parent: Some(recv_index) };
                             log!(logger, "New wave state {:?}", &best_wave);
                             do_wave(em, &mut awaiting, &best_wave)?;
                             if awaiting.is_empty() {
                             if let Some(p) = parent {
                                 // return the echo to my parent
                                 em.send_to_setup(p, &S(LeaderEcho { maybe_leader }))?;
                             } else {
                                 // wave completed!
                                 break 'echo_loop;
+                            }
                                 log!(logger, "Special case! Only neighbor is parent. Replying to {:?} msg {:?}", recv_index, &msg);
                                 em.send_to_setup(recv_index, &msg)?;
+                            }
+                        }
                     Greater => {
                         // join new echo
                         log!(logger, "Setting leader to index {:?}", index);
                         parent = Some(index);
                         my_leader = maybe_leader;
                         let echo = S(LeaderEcho { maybe_leader: my_leader });
                         awaiting.clear();
                         if em.endpoint_exts.len() == 1 {
                             // immediately reply to parent
                             log!(logger, "replying echo to parent {:?} immediately", index);
                             em.send_to_setup(index, &echo)?;
                         O::Equal => {
                             assert!(awaiting.remove(&recv_index));
                             log!(
                                 logger,
                                 "Wave reply from index {:?} for leader {:?}. Now awaiting {} replies",
                                 recv_index,
                                 best_wave.leader,
                                 awaiting.len()
                             );
                             if awaiting.is_empty() {
                                 if let Some(parent) = best_wave.parent {
                                     log!(
                                         logger,
                                         "Sub-wave done! replying to parent {:?} msg {:?}",
                                         parent,
                                         &msg
                                     );
                                     em.send_to_setup(parent, &msg)?;
                                 } else {
                             for index2 in 0..em.endpoint_exts.len() {
                                 if index2 == index {
                                     // don't propagate echo to my parent
                                     continue;
                                     let election_result: WaveState = best_wave;
                                     log!(logger, "Election won! Result {:?}", &election_result);
                                     break 'election election_result;
+                                }
                                 log!(logger, "repeating echo {:?} to {:?}", &echo, index2);
                                 em.send_to_setup(index2, &echo)?;
                                 awaiting.insert(index2);
+                            }
+                        }
+                    }
+                }
                 S(YouAreMyParent) | S(MyPortInfo(_)) => unreachable!(),
                 comm_msg @ Msg::CommMsg { .. } => {
                     log!(logger, "delaying msg {:?} during election algorithm", comm_msg);
                     em.delayed_messages.push((recv_index, comm_msg));
+                }
             inappropriate_msg => {
                 log!(logger, "delaying msg {:?} during echo phase", inappropriate_msg);
                 em.delayed_messages.push((index, inappropriate_msg))
+            }
+        }
+    }
     match parent {
         None => assert_eq!(
             my_leader, controller_id,
             "I've got no parent, but I consider {:?} the leader?",
             my_leader
         ),
         Some(parent) => assert_ne!(
             my_leader, controller_id,
             "I have {:?} as parent, but I consider myself ({:?}) the leader?",
             parent, controller_id
         ),
+    }
     log!(logger, "DONE WITH ECHO! Leader has cid={:?}", my_leader);
     };
     // 3. broadcast leader announcement (except to parent: confirm they are your parent)
     //    in this loop, every node sends 1 message to each neighbor
     //    await 1 message from all non-parents.
     let msg_for_non_parents = S(LeaderAnnounce { leader: my_leader });
     for index in 0..em.endpoint_exts.len() {
         let msg = if Some(index) == parent {
             &S(YouAreMyParent)
     // starting algorithm 2. Send a message to every neighbor
     log!(logger, "Starting tree construction. Step 1: send one msg per neighbor");
     awaiting.clear();
     for index in em.index_iter() {
         if Some(index) == election_result.parent {
             em.send_to_setup(index, &S(YouAreMyParent))?;
         } else {
             awaiting.insert(index);
             &msg_for_non_parents
         };
         log!(logger, "ANNOUNCING to {:?} {:?}", index, msg);
         em.send_to_setup(index, msg)?;
             em.send_to_setup(index, &S(LeaderAnnounce { tree_leader: election_result.leader }))?;
+        }
+    }
-    let mut children = Vec::default();
+    let mut children = vec![];
     em.undelay_all();
     while !awaiting.is_empty() {
         log!(logger, "awaiting {:?}", &awaiting);
         let (index, msg) = em.try_recv_any_setup(deadline)?;
         log!(logger, "Tree construction_loop loop. awaiting {:?}...", awaiting.iter());
         let (recv_index, msg) = em.try_recv_any_setup(logger, deadline)?;
         log!(logger, "Received from index {:?} msg {:?}", &recv_index, &msg);
         match msg {
             S(LeaderWave { .. }) => { /* old message */ }
             S(LeaderAnnounce { .. }) => {
                 // not a child
                 log!(
                     logger,
                     "Got reply from non-child index {:?}. Children: {:?}",
                     recv_index,
                     children.iter()
                 );
                 if !awaiting.remove(&recv_index) {
                     return Err(SetupAlgMisbehavior);
+                }
+            }
             S(YouAreMyParent) => {
                 assert!(awaiting.remove(&index));
                 children.push(index);
                 if !awaiting.remove(&recv_index) {
                     log!(
                         logger,
                         "Got reply from child index {:?}. Children before... {:?}",
                         recv_index,
                         children.iter()
                     );
                     return Err(SetupAlgMisbehavior);
+                }
             S(LeaderAnnounce { leader }) => {
                 assert!(awaiting.remove(&index));
                 assert!(leader == my_leader);
                 assert!(Some(index) != parent);
                 // they wouldn't send me this if they considered me their parent
                 children.push(recv_index);
+            }
             inappropriate_msg => {
                 log!(logger, "delaying msg {:?} during echo-reply phase", inappropriate_msg);
                 em.delayed_messages.push((index, inappropriate_msg));
             S(MyPortInfo(_)) => unreachable!(),
             comm_msg @ Msg::CommMsg { .. } => {
                 log!(logger, "delaying msg {:?} during election algorithm", comm_msg);
                 em.delayed_messages.push((recv_index, comm_msg));
+            }
+        }
+    }
     children.sort();
     children.dedup();
     Ok(Neighborhood { parent, children })
     children.shrink_to_fit();
     let neighborhood =
         Neighborhood { parent: election_result.parent, children: VecSet::new(children) };
     log!(logger, "Neighborhood constructed {:?}", &neighborhood);
     Ok(neighborhood)
+}

src/runtime/tests.rs

➞

Show inline comments

 use crate as reowolf;
 use crossbeam_utils::thread::scope;
 use reowolf::{
     error::*,
     Polarity::{Getter, Putter},
     *,
 };
 use std::net::SocketAddr;
 use std::{sync::Arc, time::Duration};
 fn next_test_addr() -> SocketAddr {
     use std::{
         net::{Ipv4Addr, SocketAddrV4},
         sync::atomic::{AtomicU16, Ordering::SeqCst},
     };
     static TEST_PORT: AtomicU16 = AtomicU16::new(5_000);
@@ @@ -261,28 +262,28 @@ fn sync_sync() { @@
     */
     let mut c = Connector::new_simple(MINIMAL_PROTO.clone(), 0);
     let [p0, g0] = c.new_port_pair();
     let [p1, g1] = c.new_port_pair();
     c.add_component(b"sync", &[g0, p1]).unwrap();
     c.connect(Some(Duration::from_secs(1))).unwrap();
     c.put(p0, (b"hello" as &[_]).into()).unwrap();
     c.get(g1).unwrap();
     c.sync(Some(Duration::from_secs(1))).unwrap();
     c.gotten(g1).unwrap();
+}
-fn file_logged_connector(controller_id: ControllerId, path: &str) -> Connector {
+fn file_logged_connector(connector_id: ConnectorId, path: &str) -> Connector {
     let file = std::fs::File::create(path).unwrap();
     let file_logger = Box::new(FileLogger::new(controller_id, file));
     Connector::new(file_logger, MINIMAL_PROTO.clone(), controller_id, 8)
     let file_logger = Box::new(FileLogger::new(connector_id, file));
     Connector::new(file_logger, MINIMAL_PROTO.clone(), connector_id, 8)
+}
 #[test]
 fn double_net_connect() {
     let sock_addrs = [next_test_addr(), next_test_addr()];
     scope(|s| {
         s.spawn(|_| {
             let mut c = file_logged_connector(0, "./logs/double_net_a.txt");
             let [_p, _g] = [
                 c.new_net_port(Putter, EndpointSetup { sock_addr: sock_addrs[0], is_active: true })
                     .unwrap(),
                 c.new_net_port(Getter, EndpointSetup { sock_addr: sock_addrs[1], is_active: true })
@@ @@ -352,12 +353,114 @@ fn distributed_msg_bounce() { @@
+                )
                 .unwrap(),
             ];
             c.connect(Some(Duration::from_secs(1))).unwrap();
             c.put(p, (b"hello" as &[_]).into()).unwrap();
             c.get(g).unwrap();
             c.sync(Some(Duration::from_secs(1))).unwrap();
             c.gotten(g).unwrap();
         });
     })
     .unwrap();
+}
 #[test]
 fn local_timeout() {
     let mut c = file_logged_connector(0, "./logs/local_timeout.txt");
     let [_, g] = c.new_port_pair();
     c.connect(Some(Duration::from_secs(1))).unwrap();
     c.get(g).unwrap();
     match c.sync(Some(Duration::from_millis(200))) {
         Err(SyncError::RoundFailure) => {}
         res => panic!("expeted timeout. but got {:?}", res),
+    }
+}
 #[test]
 fn parent_timeout() {
     let sock_addr = next_test_addr();
     scope(|s| {
         s.spawn(|_| {
             // parent; times out
             let mut c = file_logged_connector(999, "./logs/parent_timeout_a.txt");
             let _ = c.new_net_port(Putter, EndpointSetup { sock_addr, is_active: true }).unwrap();
             c.connect(Some(Duration::from_secs(1))).unwrap();
             c.sync(Some(Duration::from_millis(300))).unwrap_err(); // timeout
         });
         s.spawn(|_| {
             // child
             let mut c = file_logged_connector(000, "./logs/parent_timeout_b.txt");
             let g = c.new_net_port(Getter, EndpointSetup { sock_addr, is_active: false }).unwrap();
             c.connect(Some(Duration::from_secs(1))).unwrap();
             c.get(g).unwrap(); // not matched by put
             c.sync(None).unwrap_err(); // no timeout
         });
     })
     .unwrap();
+}
 #[test]
 fn child_timeout() {
     let sock_addr = next_test_addr();
     scope(|s| {
         s.spawn(|_| {
             // child; times out
             let mut c = file_logged_connector(000, "./logs/child_timeout_a.txt");
             let _ = c.new_net_port(Putter, EndpointSetup { sock_addr, is_active: true }).unwrap();
             c.connect(Some(Duration::from_secs(1))).unwrap();
             c.sync(Some(Duration::from_millis(300))).unwrap_err(); // timeout
         });
         s.spawn(|_| {
             // parent
             let mut c = file_logged_connector(999, "./logs/child_timeout_b.txt");
             let g = c.new_net_port(Getter, EndpointSetup { sock_addr, is_active: false }).unwrap();
             c.connect(Some(Duration::from_secs(1))).unwrap();
             c.get(g).unwrap(); // not matched by put
             c.sync(None).unwrap_err(); // no timeout
         });
     })
     .unwrap();
+}
 #[test]
 fn chain_connect() {
     let sock_addrs = [next_test_addr(), next_test_addr(), next_test_addr(), next_test_addr()];
     scope(|s| {
         s.spawn(|_| {
             let mut c = file_logged_connector(0, "./logs/chain_connect_a.txt");
             c.new_net_port(Putter, EndpointSetup { sock_addr: sock_addrs[0], is_active: false })
                 .unwrap();
             c.connect(Some(Duration::from_secs(1))).unwrap();
         });
         s.spawn(|_| {
             let mut c = file_logged_connector(2, "./logs/chain_connect_b.txt");
             c.new_net_port(Getter, EndpointSetup { sock_addr: sock_addrs[0], is_active: true })
                 .unwrap();
             c.new_net_port(Putter, EndpointSetup { sock_addr: sock_addrs[1], is_active: false })
                 .unwrap();
             c.connect(Some(Duration::from_secs(1))).unwrap();
         });
         s.spawn(|_| {
             let mut c = file_logged_connector(1, "./logs/chain_connect_c.txt");
             c.new_net_port(Getter, EndpointSetup { sock_addr: sock_addrs[1], is_active: true })
                 .unwrap();
             // c.new_net_port(Putter, EndpointSetup { sock_addr: sock_addrs[2], is_active: false })
             //     .unwrap();
             c.connect(Some(Duration::from_secs(1))).unwrap();
         });
         // s.spawn(|_| {
         //     let mut c = file_logged_connector(3, "./logs/chain_connect_d.txt");
         //     c.new_net_port(Getter, EndpointSetup { sock_addr: sock_addrs[2], is_active: true })
         //         .unwrap();
         //     c.new_net_port(Putter, EndpointSetup { sock_addr: sock_addrs[3], is_active: false })
         //         .unwrap();
         //     c.connect(Some(Duration::from_secs(1))).unwrap();
         // });
         // s.spawn(|_| {
         //     let mut c = file_logged_connector(4, "./logs/chain_connect_e.txt");
         //     c.new_net_port(Getter, EndpointSetup { sock_addr: sock_addrs[3], is_active: true })
         //         .unwrap();
         //     c.connect(Some(Duration::from_secs(1))).unwrap();
         // });
     })
     .unwrap();
+}

0 comments (0 inline, 0 general)