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

Changeset - 41bf21c39c5b

Parent rev.

Child rev.

[Not reviewed]

0 5 0

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

better testing and logging

5 files changed with 142 insertions and 125 deletions:

src/common.rs

src/lib.rs

src/runtime/mod.rs

src/runtime/setup.rs

src/runtime/tests.rs

128

120

0 comments (0 inline, 0 general)

src/common.rs

➞

Show inline comments

 ///////////////////// PRELUDE /////////////////////
 pub use crate::protocol::{ComponentState, ProtocolDescription};
 pub use crate::runtime::{NonsyncProtoContext, SyncProtoContext};
 pub use core::{
     cmp::Ordering,
     fmt::{Debug, Formatter},
     hash::{Hash, Hasher},
     ops::{Range, RangeFrom},
     time::Duration,
 };
 pub use indexmap::{IndexMap, IndexSet};
 pub use maplit::{hashmap, hashset};
 pub use mio::{
     net::{TcpListener, TcpStream},
     Events, Interest, Poll, Token,
 };
 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 ConnectorId = u32;
 pub type PortSuffix = u32;
 #[derive(
     Copy, Clone, Eq, PartialEq, Ord, Hash, PartialOrd, serde::Serialize, serde::Deserialize,
 )]
 pub struct Id {
     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,
 )]
 pub struct PortId(Id);
 #[derive(
     Copy, Clone, Eq, PartialEq, Ord, Hash, PartialOrd, serde::Serialize, serde::Deserialize,
 )]
 pub struct FiringVar(pub(crate) PortId);
 #[derive(
     Copy, Clone, Eq, PartialEq, Ord, Hash, PartialOrd, serde::Serialize, serde::Deserialize,
 )]
 pub struct ProtoComponentId(Id);
 #[derive(Debug, Clone, Eq, PartialEq, Ord, PartialOrd)]
 pub struct Payload(Arc<Vec<u8>>);
 #[derive(
     Debug, Eq, PartialEq, Clone, Hash, Copy, Ord, PartialOrd, serde::Serialize, serde::Deserialize,
 )]
 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,
 )]
 pub enum EndpointPolarity {
     Active,  // calls connect()
     Passive, // calls bind() listen() accept()
+}
 #[derive(Eq, PartialEq, Copy, Clone, Debug)]
 pub enum AddComponentError {
     NoSuchComponent,
     NonPortTypeParameters,
     CannotMovePort(PortId),
     WrongNumberOfParamaters { expected: usize },
     UnknownPort(PortId),
     WrongPortPolarity { port: PortId, expected_polarity: Polarity },
     DuplicateMovedPort(PortId),
+}
 #[derive(Debug, Clone)]
 pub enum NonsyncBlocker {
     Inconsistent,
     ComponentExit,
     SyncBlockStart,
+}
 #[derive(Debug, Clone)]
 pub enum SyncBlocker {
     Inconsistent,
     SyncBlockEnd,
     CouldntReadMsg(PortId),
     CouldntCheckFiring(PortId),
     PutMsg(PortId, Payload),
+}
 ///////////////////// IMPL /////////////////////
 impl U32Stream {
     pub fn next(&mut self) -> u32 {
         if self.next == u32::MAX {
             panic!("NO NEXT!")
+        }
         self.next += 1;
         self.next - 1
+    }
+}
 impl From<Id> for PortId {
     fn from(id: Id) -> PortId {
         Self(id)
+    }
+}
 impl From<Id> for ProtoComponentId {
     fn from(id: Id) -> ProtoComponentId {
         Self(id)
+    }
+}
 impl From<&[u8]> for Payload {
     fn from(s: &[u8]) -> Payload {
         Payload(Arc::new(s.to_vec()))
+    }
+}
 impl Payload {
     pub fn new(len: usize) -> Payload {
         let mut v = Vec::with_capacity(len);
         unsafe {
             v.set_len(len);
+        }
         Payload(Arc::new(v))
+    }
     pub fn len(&self) -> usize {
         self.0.len()
+    }
     pub fn as_slice(&self) -> &[u8] {
         &self.0
+    }
     pub fn as_mut_slice(&mut self) -> &mut [u8] {
         Arc::make_mut(&mut self.0) as _
+    }
     pub fn concat_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,
     ) -> std::result::Result<<S as serde::Serializer>::Ok, <S as serde::Serializer>::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,
     ) -> std::result::Result<Self, <D as serde::Deserializer<'de>>::Error>
     where
         D: serde::Deserializer<'de>,
+    {
         let inner: Vec<u8> = Vec::deserialize(deserializer)?;

src/lib.rs

➞

Show inline comments

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

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 {
     polarity: Polarity,
     port: PortId,
+}
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 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),
     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
+}
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 pub struct SendPayloadMsg {
     predicate: Predicate,
     payload: Payload,
+}
 #[derive(Debug, PartialEq)]
 pub enum CommonSatResult {
     FormerNotLatter,
     LatterNotFormer,
     Equivalent,
     New(Predicate),
     Nonexistant,
+}
 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(ConnectorId, Vec<u8>);
 #[derive(Debug)]
 pub struct DummyLogger;
 #[derive(Debug)]
 pub struct FileLogger(ConnectorId, std::fs::File);
 #[derive(Debug, Clone)]
 pub struct EndpointSetup {
     pub sock_addr: SocketAddr,
-    pub is_active: bool,
+    pub endpoint_polarity: EndpointPolarity,
+}
 #[derive(Debug)]
 pub struct EndpointExt {
     endpoint: Endpoint,
     getter_for_incoming: PortId,
+}
 #[derive(Debug)]
 pub struct Neighborhood {
     parent: Option<usize>,
     children: VecSet<usize>,
+}
 #[derive(Debug)]
 pub struct MemInMsg {
     inp: PortId,
     msg: Payload,
+}
 #[derive(Debug)]
 pub struct IdManager {
     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)>,
     undelayed_messages: Vec<(usize, Msg)>,
     endpoint_exts: Vec<EndpointExt>,
+}
 #[derive(Debug, Default)]
 pub struct PortInfo {
     polarities: HashMap<PortId, Polarity>,
     peers: HashMap<PortId, PortId>,
     routes: HashMap<PortId, Route>,
+}
 #[derive(Debug)]
 pub struct Connector {
     proto_description: Arc<ProtocolDescription>,
     proto_components: HashMap<ProtoComponentId, ProtoComponent>,
     logger: Box<dyn Logger>,
     id_manager: IdManager,
     native_ports: HashSet<PortId>,
     port_info: PortInfo,
     phased: ConnectorPhased,
+}
 #[derive(Debug)]
 pub enum ConnectorPhased {
     Setup {
         endpoint_setups: Vec<(PortId, EndpointSetup)>,
         surplus_sockets: u16,
     },
     Communication {
         round_index: usize,
         endpoint_manager: EndpointManager,
         neighborhood: Neighborhood,
         mem_inbox: Vec<MemInMsg>,
         native_batches: Vec<NativeBatch>,
         round_result: Result<Option<(usize, HashMap<PortId, Payload>)>, SyncError>,
     },
+}
 #[derive(Default, Clone, Eq, PartialEq, Hash, serde::Serialize, serde::Deserialize)]
 pub struct Predicate {
     pub assigned: BTreeMap<FiringVar, bool>,
+}
 #[derive(Debug, Default)]
 pub struct NativeBatch {
     // invariant: putters' and getters' polarities respected
     to_put: HashMap<PortId, Payload>,
     to_get: HashSet<PortId>,
+}
 pub struct NonsyncProtoContext<'a> {
     logger: &'a mut dyn Logger,
     proto_component_id: ProtoComponentId,
     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()

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>,
         connector_id: ConnectorId,
     ) -> Self {
         let logger = Box::new(DummyLogger);
         // let logger = Box::new(DummyLogger);
         let surplus_sockets = 2;
         Self::new(logger, proto_description, connector_id, surplus_sockets)
+    }
     pub fn new(
         mut logger: Box<dyn Logger>,
         proto_description: Arc<ProtocolDescription>,
         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(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,
         sock_addr: SocketAddr,
         endpoint_polarity: EndpointPolarity,
     ) -> Result<PortId, ()> {
         match &mut self.phased {
             ConnectorPhased::Setup { endpoint_setups, .. } => {
                 let endpoint_setup = EndpointSetup { sock_addr, endpoint_polarity };
                 let p = self.id_manager.new_port_id();
                 self.native_ports.insert(p);
                 // {polarity, route} known. {peer} unknown.
                 self.port_info.polarities.insert(p, polarity);
                 self.port_info.routes.insert(p, Route::LocalComponent(LocalComponentId::Native));
                 log!(
                     self.logger,
                     "Added net port {:?} with polarity {:?} and endpoint setup {:?} ",
                     p,
                     polarity,
                     &endpoint_setup
                 );
                 endpoint_setups.push((p, endpoint_setup));
                 Ok(p)
+            }
             ConnectorPhased::Communication { .. } => Err(()),
+        }
+    }
     pub fn connect(&mut self, timeout: Option<Duration>) -> Result<(), ConnectError> {
         use ConnectError::*;
         match &mut self.phased {
             ConnectorPhased::Communication { .. } => {
                 log!(self.logger, "Call to connecting in connected state");
                 Err(AlreadyConnected)
+            }
             ConnectorPhased::Setup { endpoint_setups, .. } => {
                 log!(self.logger, "~~~ CONNECT called timeout {:?}", timeout);
                 let deadline = timeout.map(|to| Instant::now() + to);
                 // connect all endpoints in parallel; send and receive peer ids through ports
                 let mut endpoint_manager = new_endpoint_manager(
                     &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.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(),
                     native_batches: vec![Default::default()],
                     round_result: Ok(None),
                 };
                 Ok(())
+            }
+        }
+    }
+}
 fn new_endpoint_manager(
     logger: &mut dyn Logger,
     endpoint_setups: &[(PortId, EndpointSetup)],
     port_info: &mut PortInfo,
     deadline: Option<Instant>,
 ) -> Result<EndpointManager, ConnectError> {
     ////////////////////////////////////////////
     use ConnectError::*;
     const BOTH: Interest = Interest::READABLE.add(Interest::WRITABLE);
     struct Todo {
         todo_endpoint: TodoEndpoint,
         local_port: PortId,
         sent_local_port: bool,          // true <-> I've sent my local port
         recv_peer_port: Option<PortId>, // Some(..) <-> I've received my peer's port
+    }
     enum TodoEndpoint {
         Listener(TcpListener),
         Endpoint(Endpoint),
+    }
     fn init_todo(
         token: Token,
         local_port: PortId,
         endpoint_setup: &EndpointSetup,
         poll: &mut Poll,
     ) -> Result<Todo, ConnectError> {
-        let todo_endpoint = if endpoint_setup.is_active {
+        let todo_endpoint = if let EndpointPolarity::Active = endpoint_setup.endpoint_polarity {
             let mut stream = TcpStream::connect(endpoint_setup.sock_addr)
                 .expect("mio::TcpStream connect should not fail!");
             poll.registry().register(&mut stream, token, BOTH).unwrap();
             TodoEndpoint::Endpoint(Endpoint { stream, inbox: vec![] })
         } else {
             let mut listener = TcpListener::bind(endpoint_setup.sock_addr)
                 .map_err(|_| BindFailed(endpoint_setup.sock_addr))?;
             poll.registry().register(&mut listener, token, BOTH).unwrap();
             TodoEndpoint::Listener(listener)
         };
         Ok(Todo { todo_endpoint, local_port, sent_local_port: false, recv_peer_port: None })
     };
     ////////////////////////////////////////////
     // 1. Start to construct EndpointManager
     let mut poll = Poll::new().map_err(|_| PollInitFailed)?;
     let mut events = Events::with_capacity(64);
     let mut polled_undrained = IndexSet::<usize>::default();
     let mut delayed_messages = vec![];
     // 2. create a registered (TcpListener/Endpoint) for passive / active respectively
     let mut todos = endpoint_setups
         .iter()
         .enumerate()
         .map(|(index, (local_port, endpoint_setup))| {
             init_todo(Token(index), *local_port, endpoint_setup, &mut poll)
         })
         .collect::<Result<Vec<Todo>, ConnectError>>()?;
     // 3. Using poll to drive progress:
     //    - accept an incoming connection for each TcpListener (turning them into endpoints too)
     //    - for each endpoint, send the local PortId
     //    - for each endpoint, recv the peer's PortId, and
     let mut setup_incomplete: HashSet<usize> = (0..todos.len()).collect();
     while !setup_incomplete.is_empty() {
         let remaining = if let Some(deadline) = deadline {
             Some(deadline.checked_duration_since(Instant::now()).ok_or(Timeout)?)
         } else {
             None
         };
         poll.poll(&mut events, remaining).map_err(|_| PollFailed)?;
         for event in events.iter() {
             let token = event.token();
             let Token(index) = token;
             let todo: &mut Todo = &mut todos[index];
             if let TodoEndpoint::Listener(listener) = &mut todo.todo_endpoint {
                 match listener.accept() {
                     Ok((mut stream, peer_addr)) => {
                         poll.registry().deregister(listener).unwrap();
                         poll.registry().register(&mut stream, token, BOTH).unwrap();
                         log!(
                             logger,
                             "Endpoint[{}] accepted a connection from {:?}",
                             index,
                             peer_addr
                         );
                         let endpoint = Endpoint { stream, inbox: vec![] };
                         todo.todo_endpoint = TodoEndpoint::Endpoint(endpoint);
+                    }
                     Err(e) if e.kind() == WouldBlock => {}
                     Err(_) => return Err(AcceptFailed(listener.local_addr().unwrap())),
+                }
+            }
             match todo {
                 Todo {
                     todo_endpoint: TodoEndpoint::Endpoint(endpoint),
                     local_port,
                     sent_local_port,
                     recv_peer_port,
                     ..
                 } => {
                     if !setup_incomplete.contains(&index) {
                         continue;
+                    }
                     let local_polarity = *port_info.polarities.get(local_port).unwrap();
                     if event.is_writable() && !*sent_local_port {
                         let msg = Msg::SetupMsg(SetupMsg::MyPortInfo(MyPortInfo {
                             polarity: local_polarity,
                             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(logger).map_err(|e| {
                             EndpointSetupError(endpoint.stream.local_addr().unwrap(), e)
                         })?;
                         if maybe_msg.is_some() && !endpoint.inbox.is_empty() {
                             polled_undrained.insert(index);
+                        }

src/runtime/tests.rs

➞

Show inline comments

 use crate as reowolf;
 use crossbeam_utils::thread::scope;
 use reowolf::{
     error::*,
     EndpointPolarity::{Active, Passive},
     Polarity::{Getter, Putter},
     *,
 };
 use std::net::SocketAddr;
 use std::{sync::Arc, time::Duration};
 use std::{fs::File, net::SocketAddr, path::Path, 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);
     let port = TEST_PORT.fetch_add(1, SeqCst);
     SocketAddrV4::new(Ipv4Addr::LOCALHOST, port).into()
+}
 lazy_static::lazy_static! {
     static ref MINIMAL_PROTO: Arc<ProtocolDescription> =
         { Arc::new(reowolf::ProtocolDescription::parse(b"").unwrap()) };
+}
 fn file_logged_connector(connector_id: ConnectorId, dir_path: &Path) -> Connector {
     let _ = std::fs::create_dir(dir_path); // we will check failure soon
     let path = dir_path.join(format!("cid_{:?}.txt", connector_id));
     let file = File::create(path).unwrap();
     let file_logger = Box::new(FileLogger::new(connector_id, file));
     Connector::new(file_logger, MINIMAL_PROTO.clone(), connector_id, 8)
+}
 //////////////////////////////////////////
 #[test]
-fn simple_connector() {
+fn basic_connector() {
     Connector::new_simple(MINIMAL_PROTO.clone(), 0);
+}
 #[test]
 fn basic_logged_connector() {
     let test_log_path = Path::new("./logs/basic_logged_connector");
     file_logged_connector(0, test_log_path);
+}
 #[test]
 fn new_port_pair() {
     let mut c = Connector::new_simple(MINIMAL_PROTO.clone(), 0);
     let test_log_path = Path::new("./logs/new_port_pair");
     let mut c = file_logged_connector(0, test_log_path);
     let [_, _] = c.new_port_pair();
     let [_, _] = c.new_port_pair();
+}
 #[test]
 fn new_sync() {
     let mut c = Connector::new_simple(MINIMAL_PROTO.clone(), 0);
     let test_log_path = Path::new("./logs/new_sync");
     let mut c = file_logged_connector(0, test_log_path);
     let [o, i] = c.new_port_pair();
     c.add_component(b"sync", &[i, o]).unwrap();
+}
 #[test]
 fn new_net_port() {
     let mut c = Connector::new_simple(MINIMAL_PROTO.clone(), 0);
     let test_log_path = Path::new("./logs/new_net_port");
     let mut c = file_logged_connector(0, test_log_path);
     let sock_addr = next_test_addr();
     let _ = c.new_net_port(Getter, EndpointSetup { sock_addr, is_active: false }).unwrap();
     let _ = c.new_net_port(Putter, EndpointSetup { sock_addr, is_active: true }).unwrap();
     let _ = c.new_net_port(Getter, sock_addr, Passive).unwrap();
     let _ = c.new_net_port(Putter, sock_addr, Active).unwrap();
+}
 #[test]
 fn trivial_connect() {
     let mut c = Connector::new_simple(MINIMAL_PROTO.clone(), 0);
     let test_log_path = Path::new("./logs/trivial_connect");
     let mut c = file_logged_connector(0, test_log_path);
     c.connect(Some(Duration::from_secs(1))).unwrap();
+}
 #[test]
 fn single_node_connect() {
     let sock_addr = next_test_addr();
     let mut c = Connector::new_simple(MINIMAL_PROTO.clone(), 0);
     let _ = c.new_net_port(Getter, EndpointSetup { sock_addr, is_active: false }).unwrap();
     let _ = c.new_net_port(Putter, EndpointSetup { sock_addr, is_active: true }).unwrap();
     let test_log_path = Path::new("./logs/single_node_connect");
     let mut c = file_logged_connector(0, test_log_path);
     let _ = c.new_net_port(Getter, sock_addr, Passive).unwrap();
     let _ = c.new_net_port(Putter, sock_addr, Active).unwrap();
     c.connect(Some(Duration::from_secs(1))).unwrap();
+}
 #[test]
 fn multithreaded_connect() {
     let sock_addr = next_test_addr();
     let test_log_path = Path::new("./logs/multithreaded_connect");
     scope(|s| {
         s.spawn(|_| {
             let mut c = Connector::new_simple(MINIMAL_PROTO.clone(), 0);
             let _ = c.new_net_port(Getter, EndpointSetup { sock_addr, is_active: true }).unwrap();
             let mut c = file_logged_connector(0, test_log_path);
             let _ = c.new_net_port(Getter, sock_addr, Active).unwrap();
             c.connect(Some(Duration::from_secs(1))).unwrap();
         });
         s.spawn(|_| {
             let mut c = Connector::new_simple(MINIMAL_PROTO.clone(), 1);
             let _ = c.new_net_port(Putter, EndpointSetup { sock_addr, is_active: false }).unwrap();
             let mut c = file_logged_connector(1, test_log_path);
             let _ = c.new_net_port(Putter, sock_addr, Passive).unwrap();
             c.connect(Some(Duration::from_secs(1))).unwrap();
         });
     })
     .unwrap();
+}
 #[test]
 fn put_no_sync() {
     let mut c = Connector::new_simple(MINIMAL_PROTO.clone(), 0);
     let test_log_path = Path::new("./logs/put_no_sync");
     let mut c = file_logged_connector(0, test_log_path);
     let [o, _] = c.new_port_pair();
     c.connect(Some(Duration::from_secs(1))).unwrap();
     c.put(o, (b"hi" as &[_]).into()).unwrap();
+}
 #[test]
 fn wrong_polarity_bad() {
     let mut c = Connector::new_simple(MINIMAL_PROTO.clone(), 0);
     let test_log_path = Path::new("./logs/wrong_polarity_bad");
     let mut c = file_logged_connector(0, test_log_path);
     let [_, i] = c.new_port_pair();
     c.connect(Some(Duration::from_secs(1))).unwrap();
     c.put(i, (b"hi" as &[_]).into()).unwrap_err();
+}
 #[test]
 fn dup_put_bad() {
     let mut c = Connector::new_simple(MINIMAL_PROTO.clone(), 0);
     let test_log_path = Path::new("./logs/dup_put_bad");
     let mut c = file_logged_connector(0, test_log_path);
     let [o, _] = c.new_port_pair();
     c.connect(Some(Duration::from_secs(1))).unwrap();
     c.put(o, (b"hi" as &[_]).into()).unwrap();
     c.put(o, (b"hi" as &[_]).into()).unwrap_err();
+}
 #[test]
 fn trivial_sync() {
     let mut c = Connector::new_simple(MINIMAL_PROTO.clone(), 0);
     let test_log_path = Path::new("./logs/trivial_sync");
     let mut c = file_logged_connector(0, test_log_path);
     c.connect(Some(Duration::from_secs(1))).unwrap();
     c.sync(Some(Duration::from_secs(1))).unwrap();
+}
 #[test]
 fn unconnected_gotten_err() {
     let mut c = Connector::new_simple(MINIMAL_PROTO.clone(), 0);
     let test_log_path = Path::new("./logs/unconnected_gotten_err");
     let mut c = file_logged_connector(0, test_log_path);
     let [_, i] = c.new_port_pair();
     assert_eq!(reowolf::error::GottenError::NoPreviousRound, c.gotten(i).unwrap_err());
+}
 #[test]
 fn connected_gotten_err_no_round() {
     let mut c = Connector::new_simple(MINIMAL_PROTO.clone(), 0);
     let test_log_path = Path::new("./logs/connected_gotten_err_no_round");
     let mut c = file_logged_connector(0, test_log_path);
     let [_, i] = c.new_port_pair();
     c.connect(Some(Duration::from_secs(1))).unwrap();
     assert_eq!(reowolf::error::GottenError::NoPreviousRound, c.gotten(i).unwrap_err());
+}
 #[test]
 fn connected_gotten_err_ungotten() {
     let mut c = Connector::new_simple(MINIMAL_PROTO.clone(), 0);
     let test_log_path = Path::new("./logs/connected_gotten_err_ungotten");
     let mut c = file_logged_connector(0, test_log_path);
     let [_, i] = c.new_port_pair();
     c.connect(Some(Duration::from_secs(1))).unwrap();
     c.sync(Some(Duration::from_secs(1))).unwrap();
     assert_eq!(reowolf::error::GottenError::PortDidntGet, c.gotten(i).unwrap_err());
+}
 #[test]
 fn native_polarity_checks() {
     let mut c = Connector::new_simple(MINIMAL_PROTO.clone(), 0);
     let test_log_path = Path::new("./logs/native_polarity_checks");
     let mut c = file_logged_connector(0, test_log_path);
     let [o, i] = c.new_port_pair();
     c.connect(Some(Duration::from_secs(1))).unwrap();
     // fail...
     c.get(o).unwrap_err();
     c.put(i, (b"hi" as &[_]).into()).unwrap_err();
     // succeed..
     c.get(i).unwrap();
     c.put(o, (b"hi" as &[_]).into()).unwrap();
+}
 #[test]
 fn native_multiple_gets() {
     let mut c = Connector::new_simple(MINIMAL_PROTO.clone(), 0);
     let test_log_path = Path::new("./logs/native_multiple_gets");
     let mut c = file_logged_connector(0, test_log_path);
     let [_, i] = c.new_port_pair();
     c.connect(Some(Duration::from_secs(1))).unwrap();
     c.get(i).unwrap();
     c.get(i).unwrap_err();
+}
 #[test]
 fn next_batch() {
     let mut c = Connector::new_simple(MINIMAL_PROTO.clone(), 0);
     let test_log_path = Path::new("./logs/next_batch");
     let mut c = file_logged_connector(0, test_log_path);
     c.next_batch().unwrap_err();
     c.connect(Some(Duration::from_secs(1))).unwrap();
     c.next_batch().unwrap();
     c.next_batch().unwrap();
     c.next_batch().unwrap();
+}
 #[test]
 fn native_self_msg() {
     let mut c = Connector::new_simple(MINIMAL_PROTO.clone(), 0);
     let test_log_path = Path::new("./logs/native_self_msg");
     let mut c = file_logged_connector(0, test_log_path);
     let [o, i] = c.new_port_pair();
     c.connect(Some(Duration::from_secs(1))).unwrap();
     c.get(i).unwrap();
     c.put(o, (b"hi" as &[_]).into()).unwrap();
     c.sync(Some(Duration::from_secs(1))).unwrap();
+}
 #[test]
 fn two_natives_msg() {
     let test_log_path = Path::new("./logs/two_natives_msg");
     let sock_addr = next_test_addr();
     scope(|s| {
         s.spawn(|_| {
             let mut c = Connector::new_simple(MINIMAL_PROTO.clone(), 0);
             let g = c.new_net_port(Getter, EndpointSetup { sock_addr, is_active: true }).unwrap();
             let mut c = file_logged_connector(0, test_log_path);
             let g = c.new_net_port(Getter, sock_addr, Active).unwrap();
             c.connect(Some(Duration::from_secs(1))).unwrap();
             c.get(g).unwrap();
             c.sync(Some(Duration::from_secs(1))).unwrap();
             c.gotten(g).unwrap();
         });
         s.spawn(|_| {
             let mut c = Connector::new_simple(MINIMAL_PROTO.clone(), 1);
             let p = c.new_net_port(Putter, EndpointSetup { sock_addr, is_active: false }).unwrap();
             let mut c = file_logged_connector(1, test_log_path);
             let p = c.new_net_port(Putter, sock_addr, Passive).unwrap();
             c.connect(Some(Duration::from_secs(1))).unwrap();
             c.put(p, (b"hello" as &[_]).into()).unwrap();
             c.sync(Some(Duration::from_secs(1))).unwrap();
         });
     })
     .unwrap();
+}
 #[test]
 fn trivial_nondet() {
     let mut c = Connector::new_simple(MINIMAL_PROTO.clone(), 0);
     let test_log_path = Path::new("./logs/trivial_nondet");
     let mut c = file_logged_connector(0, test_log_path);
     let [_, i] = c.new_port_pair();
     c.connect(Some(Duration::from_secs(1))).unwrap();
     c.get(i).unwrap();
     // getting 0 batch
     c.next_batch().unwrap();
     // silent 1 batch
     assert_eq!(1, c.sync(Some(Duration::from_secs(1))).unwrap());
     c.gotten(i).unwrap_err();
+}
 #[test]
 fn connector_pair_nondet() {
     let test_log_path = Path::new("./logs/connector_pair_nondet");
     let sock_addr = next_test_addr();
     scope(|s| {
         s.spawn(|_| {
             let mut c = Connector::new_simple(MINIMAL_PROTO.clone(), 0);
             let g = c.new_net_port(Getter, EndpointSetup { sock_addr, is_active: true }).unwrap();
             let mut c = file_logged_connector(0, test_log_path);
             let g = c.new_net_port(Getter, sock_addr, Active).unwrap();
             c.connect(Some(Duration::from_secs(1))).unwrap();
             c.next_batch().unwrap();
             c.get(g).unwrap();
             assert_eq!(1, c.sync(Some(Duration::from_secs(1))).unwrap());
             c.gotten(g).unwrap();
         });
         s.spawn(|_| {
             let mut c = Connector::new_simple(MINIMAL_PROTO.clone(), 1);
             let p = c.new_net_port(Putter, EndpointSetup { sock_addr, is_active: false }).unwrap();
             let mut c = file_logged_connector(1, test_log_path);
             let p = c.new_net_port(Putter, sock_addr, Passive).unwrap();
             c.connect(Some(Duration::from_secs(1))).unwrap();
             c.put(p, (b"hello" as &[_]).into()).unwrap();
             c.sync(Some(Duration::from_secs(1))).unwrap();
         });
     })
     .unwrap();
+}
 #[test]
 fn cannot_use_moved_ports() {
     /*
     native p|-->|g sync
     */
     let mut c = Connector::new_simple(MINIMAL_PROTO.clone(), 1);
     let test_log_path = Path::new("./logs/cannot_use_moved_ports"); /*
                                                                     native p|-->|g sync
                                                                     */
     let mut c = file_logged_connector(0, test_log_path);
     let [p, g] = c.new_port_pair();
     c.add_component(b"sync", &[g, p]).unwrap();
     c.connect(Some(Duration::from_secs(1))).unwrap();
     c.put(p, (b"hello" as &[_]).into()).unwrap_err();
     c.get(g).unwrap_err();
+}
 #[test]
 fn sync_sync() {
     /*
     native p0|-->|g0 sync
            g1|<--|p1
     */
     let mut c = Connector::new_simple(MINIMAL_PROTO.clone(), 0);
     let test_log_path = Path::new("./logs/sync_sync"); /*
                                                        native p0|-->|g0 sync
                                                               g1|<--|p1
                                                        */
     let mut c = file_logged_connector(0, test_log_path);
     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(connector_id: ConnectorId, path: &str) -> Connector {
     let file = std::fs::File::create(path).unwrap();
     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 test_log_path = Path::new("./logs/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 mut c = file_logged_connector(0, test_log_path);
             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 })
                     .unwrap(),
                 c.new_net_port(Putter, sock_addrs[0], Active).unwrap(),
                 c.new_net_port(Getter, sock_addrs[1], Active).unwrap(),
             ];
             c.connect(Some(Duration::from_secs(1))).unwrap();
         });
         s.spawn(|_| {
-            let mut c = file_logged_connector(1, "./logs/double_net_b.txt");
+            let mut c = file_logged_connector(1, test_log_path);
             let [_g, _p] = [
                 c.new_net_port(
                     Getter,
                     EndpointSetup { sock_addr: sock_addrs[0], is_active: false },
+                )
                 .unwrap(),
                 c.new_net_port(
                     Putter,
                     EndpointSetup { sock_addr: sock_addrs[1], is_active: false },
+                )
                 .unwrap(),
                 c.new_net_port(Getter, sock_addrs[0], Passive).unwrap(),
                 c.new_net_port(Putter, sock_addrs[1], Passive).unwrap(),
             ];
             c.connect(Some(Duration::from_secs(1))).unwrap();
         });
     })
     .unwrap();
+}
 #[test]
 fn distributed_msg_bounce() {
     let test_log_path = Path::new("./logs/distributed_msg_bounce");
     /*
     native[0] | sync 0.p|-->|1.p native[1]
 .g|<--|1.g
     */
     let sock_addrs = [next_test_addr(), next_test_addr()];
     scope(|s| {
         s.spawn(|_| {
             /*
             native | sync p|-->
                    |      g|<--
             */
-            let mut c = file_logged_connector(0, "./logs/distributed_msg_bounce_a.txt");
+            let mut c = file_logged_connector(0, test_log_path);
             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 })
                     .unwrap(),
                 c.new_net_port(Putter, sock_addrs[0], Active).unwrap(),
                 c.new_net_port(Getter, sock_addrs[1], Active).unwrap(),
             ];
             c.add_component(b"sync", &[g, p]).unwrap();
             c.connect(Some(Duration::from_secs(1))).unwrap();
             c.sync(Some(Duration::from_secs(1))).unwrap();
         });
         s.spawn(|_| {
             /*
             native p|-->
                    g|<--
             */
-            let mut c = file_logged_connector(1, "./logs/distributed_msg_bounce_b.txt");
+            let mut c = file_logged_connector(1, test_log_path);
             let [g, p] = [
                 c.new_net_port(
                     Getter,
                     EndpointSetup { sock_addr: sock_addrs[0], is_active: false },
+                )
                 .unwrap(),
                 c.new_net_port(
                     Putter,
                     EndpointSetup { sock_addr: sock_addrs[1], is_active: false },
+                )
                 .unwrap(),
                 c.new_net_port(Getter, sock_addrs[0], Passive).unwrap(),
                 c.new_net_port(Putter, sock_addrs[1], Passive).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 test_log_path = Path::new("./logs/local_timeout");
     let mut c = file_logged_connector(0, test_log_path);
     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 test_log_path = Path::new("./logs/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();
             let mut c = file_logged_connector(999, test_log_path);
             let _ = c.new_net_port(Putter, sock_addr, Active).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();
             let mut c = file_logged_connector(000, test_log_path);
             let g = c.new_net_port(Getter, sock_addr, Passive).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 test_log_path = Path::new("./logs/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();
             let mut c = file_logged_connector(000, test_log_path);
             let _ = c.new_net_port(Putter, sock_addr, Active).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();
             let mut c = file_logged_connector(999, test_log_path);
             let g = c.new_net_port(Getter, sock_addr, Passive).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 test_log_path = Path::new("./logs/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();
             let mut c = file_logged_connector(0, test_log_path);
             c.new_net_port(Putter, sock_addrs[0], Passive).unwrap();
             c.connect(Some(Duration::from_secs(1))).unwrap();
         });
         s.spawn(|_| {
             let mut c = file_logged_connector(10, test_log_path);
             c.new_net_port(Getter, sock_addrs[0], Active).unwrap();
             c.new_net_port(Putter, sock_addrs[1], Passive).unwrap();
             c.connect(Some(Duration::from_secs(1))).unwrap();
         });
         s.spawn(|_| {
             // LEADER
             let mut c = file_logged_connector(7, test_log_path);
             c.new_net_port(Getter, sock_addrs[1], Active).unwrap();
             c.new_net_port(Putter, sock_addrs[2], Passive).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();
             let mut c = file_logged_connector(4, test_log_path);
             c.new_net_port(Getter, sock_addrs[2], Active).unwrap();
             c.new_net_port(Putter, sock_addrs[3], Passive).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();
             let mut c = file_logged_connector(1, test_log_path);
             c.new_net_port(Getter, sock_addrs[3], Active).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)