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

Changeset - fd68eb8aabd0

Parent rev.

Child rev.

[Not reviewed]

0 5 0

Christopher Esterhuyse - 5 years ago 2020-07-09 15:37:30
christopher.esterhuyse@gmail.com

bidirectional udp endpoints in

5 files changed with 156 insertions and 107 deletions:

src/macros.rs

src/runtime/logging.rs

src/runtime/mod.rs

src/runtime/setup.rs

src/runtime/tests.rs

0 comments (0 inline, 0 general)

src/macros.rs

➞

Show inline comments

 macro_rules! endptlog {
     ($logger:expr, $($arg:tt)*) => {{
     	if cfg!(feature = "endpoint_logging") {
 	        let w = $logger.line_writer();
 	        let _ = writeln!(w, $($arg)*);
 	        if let Some(w) = $logger.line_writer() {
 	        	let _ = writeln!(w, $($arg)*);
+	        }
+	    }
     }};
+}
 macro_rules! log {
     ($logger:expr, $($arg:tt)*) => {{
         let _ = writeln!($logger.line_writer(), $($arg)*);
     	if let Some(w) = $logger.line_writer() {
         	let _ = writeln!(w, $($arg)*);
+    	}
     }};
+}

src/runtime/logging.rs

➞

Show inline comments

 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
     fn line_writer(&mut self) -> Option<&mut dyn std::io::Write> {
         None
+    }
+}
 impl Logger for VecLogger {
     fn line_writer(&mut self) -> &mut dyn std::io::Write {
+    fn line_writer(&mut self) -> Option<&mut dyn std::io::Write> {
         let _ = write!(&mut self.1, "CID({}) at {:?} ", self.0, Instant::now());
         self
+        Some(self)
+    }
+}
 impl Logger for FileLogger {
     fn line_writer(&mut self) -> &mut dyn std::io::Write {
+    fn line_writer(&mut self) -> Option<&mut dyn std::io::Write> {
         let _ = write!(&mut self.1, "CID({}) at {:?} ", self.0, Instant::now());
         &mut self.1
+        Some(&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

 /// cbindgen:ignore
 mod communication;
 /// cbindgen:ignore
 mod endpoints;
 pub mod error;
 /// cbindgen:ignore
 mod logging;
 /// cbindgen:ignore
 mod setup;
 #[cfg(feature = "ffi")]
 pub mod ffi;
 #[cfg(test)]
 mod tests;
 use crate::common::*;
 use error::*;
 use mio::net::UdpSocket;
 #[derive(Debug)]
 pub struct Connector {
     unphased: ConnectorUnphased,
     phased: ConnectorPhased,
+}
 pub trait Logger: Debug {
     fn line_writer(&mut self) -> &mut dyn std::io::Write;
+    fn line_writer(&mut self) -> Option<&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);
 pub(crate) 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(crate) struct SyncProtoContext<'a> {
     logger: &'a mut dyn Logger,
     untaken_choice: &'a mut Option<u16>,
     predicate: &'a Predicate,
     port_info: &'a PortInfo,
     inbox: &'a HashMap<PortId, Payload>,
+}
 #[derive(
     Copy, Clone, Eq, PartialEq, Ord, Hash, PartialOrd, serde::Serialize, serde::Deserialize,
 )]
 struct SpecVar(PortId);
 #[derive(
     Copy, Clone, Eq, PartialEq, Ord, Hash, PartialOrd, serde::Serialize, serde::Deserialize,
 )]
 struct SpecVal(u16);
 #[derive(Debug)]
 struct RoundOk {
     batch_index: usize,
     gotten: HashMap<PortId, Payload>,
+}
 #[derive(Default)]
 struct VecSet<T: std::cmp::Ord> {
     // invariant: ordered, deduplicated
     vec: Vec<T>,
+}
 #[derive(Debug, Clone, Copy, Eq, PartialEq, Hash, serde::Serialize, serde::Deserialize)]
 enum ComponentId {
     Native,
     Proto(ProtoComponentId),
+}
 #[derive(Debug, Clone, Copy, Eq, PartialEq, Hash, serde::Serialize, serde::Deserialize)]
 enum Route {
     LocalComponent(ComponentId),
     NetEndpoint { index: usize },
     UdpEndpoint { index: usize },
+}
 #[derive(Debug, Clone, Copy, Eq, PartialEq, Hash, serde::Serialize, serde::Deserialize)]
 enum SubtreeId {
     LocalComponent(ComponentId),
     NetEndpoint { index: usize },
+}
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 struct MyPortInfo {
     polarity: Polarity,
     port: PortId,
+}
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 enum Decision {
     Failure,
     Success(Predicate),
+}
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 enum Msg {
     SetupMsg(SetupMsg),
     CommMsg(CommMsg),
+}
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 enum SetupMsg {
     MyPortInfo(MyPortInfo),
     LeaderWave { wave_leader: ConnectorId },
     LeaderAnnounce { tree_leader: ConnectorId },
     YouAreMyParent,
     SessionGather { unoptimized_map: HashMap<ConnectorId, SessionInfo> },
     SessionScatter { optimized_map: HashMap<ConnectorId, SessionInfo> },
+}
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 struct SessionInfo {
     serde_proto_description: SerdeProtocolDescription,
     port_info: PortInfo,
     endpoint_incoming_to_getter: Vec<PortId>,
     proto_components: HashMap<ProtoComponentId, ProtoComponent>,
+}
 #[derive(Debug, Clone)]
 struct SerdeProtocolDescription(Arc<ProtocolDescription>);
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 struct CommMsg {
     round_index: usize,
     contents: CommMsgContents,
+}
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 enum CommMsgContents {
     SendPayload(SendPayloadMsg),
     CommCtrl(CommCtrlMsg),
+}
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 enum CommCtrlMsg {
     Suggest { suggestion: Decision }, // SINKWARD
     Announce { decision: Decision },  // SINKAWAYS
+}
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 struct SendPayloadMsg {
     predicate: Predicate,
     payload: Payload,
+}
 #[derive(Debug, PartialEq)]
 enum AssignmentUnionResult {
     FormerNotLatter,
     LatterNotFormer,
     Equivalent,
     New(Predicate),
     Nonexistant,
+}
 struct NetEndpoint {
     inbox: Vec<u8>,
     stream: TcpStream,
+}
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 struct ProtoComponent {
     state: ComponentState,
     ports: HashSet<PortId>,
+}
 #[derive(Debug, Clone)]
 struct NetEndpointSetup {
-    local_port: PortId,
+    getter_for_incoming: PortId,
     sock_addr: SocketAddr,
     endpoint_polarity: EndpointPolarity,
+}
 #[derive(Debug, Clone)]
 struct UdpEndpointSetup {
-    local_port: PortId,
+    getter_for_incoming: PortId,
     local_addr: SocketAddr,
     peer_addr: SocketAddr,
+}
 #[derive(Debug)]
 struct NetEndpointExt {
     net_endpoint: NetEndpoint,
     getter_for_incoming: PortId,
+}
 #[derive(Debug)]
 struct Neighborhood {
     parent: Option<usize>,
     children: VecSet<usize>,
+}
 #[derive(Debug)]
 struct IdManager {
     connector_id: ConnectorId,
     port_suffix_stream: U32Stream,
     proto_component_suffix_stream: U32Stream,
+}
 #[derive(Debug)]
 struct UdpInBuffer {
     byte_vec: Vec<u8>,
+}
 #[derive(Debug)]
 struct SpecVarStream {
     connector_id: ConnectorId,
     port_suffix_stream: U32Stream,
+}
 #[derive(Debug)]
 struct EndpointManager {
     // invariants:
-    // 1. endpoint N is registered READ | WRITE with poller
+    // 1. net and udp endpoints are registered with poll. Poll token computed with TargetToken::into
     // 2. Events is empty
     poll: Poll,
     events: Events,
     delayed_messages: Vec<(usize, Msg)>,
     undelayed_messages: Vec<(usize, Msg)>,
     net_endpoint_store: EndpointStore<NetEndpointExt>,
     udp_endpoint_store: EndpointStore<UdpEndpointExt>,
     udp_in_buffer: UdpInBuffer,
+}
 #[derive(Debug)]
 struct EndpointStore<T> {
     endpoint_exts: Vec<T>,
     polled_undrained: VecSet<usize>,
+}
 #[derive(Debug)]
 struct UdpEndpointExt {
     sock: UdpSocket, // already bound and connected
     outgoing_payloads: HashMap<Predicate, Payload>,
     incoming_round_spec_var: Option<SpecVar>,
     getter_for_incoming: PortId,
     incoming_payloads: Vec<Payload>,
+}
 #[derive(Clone, Debug, Default, serde::Serialize, serde::Deserialize)]
 struct PortInfo {
     polarities: HashMap<PortId, Polarity>,
     peers: HashMap<PortId, PortId>,
     routes: HashMap<PortId, Route>,
+}
 #[derive(Debug)]
 struct ConnectorCommunication {
     round_index: usize,
     endpoint_manager: EndpointManager,
     neighborhood: Neighborhood,
     native_batches: Vec<NativeBatch>,
     round_result: Result<Option<RoundOk>, SyncError>,
+}
 #[derive(Debug)]
 struct ConnectorUnphased {
     proto_description: Arc<ProtocolDescription>,
     proto_components: HashMap<ProtoComponentId, ProtoComponent>,
     logger: Box<dyn Logger>,
     id_manager: IdManager,
     native_ports: HashSet<PortId>,
     port_info: PortInfo,
+}
 #[derive(Debug)]
 struct ConnectorSetup {
     net_endpoint_setups: Vec<NetEndpointSetup>,
     udp_endpoint_setups: Vec<UdpEndpointSetup>,
     surplus_sockets: u16,
+}
 #[derive(Debug)]
 enum ConnectorPhased {
     Setup(Box<ConnectorSetup>),
     Communication(Box<ConnectorCommunication>),
+}
 #[derive(Default, Clone, Eq, PartialEq, Hash, serde::Serialize, serde::Deserialize)]
 struct Predicate {
     assigned: BTreeMap<SpecVar, SpecVal>,
+}
 #[derive(Debug, Default)]
 struct NativeBatch {
     // invariant: putters' and getters' polarities respected
     to_put: HashMap<PortId, Payload>,
     to_get: HashSet<PortId>,
+}
 #[derive(Debug, Copy, Clone, Eq, PartialEq, Hash)]
 enum TokenTarget {
     NetEndpoint { index: usize },
     UdpEndpoint { index: usize },
     Waker,
+}
 trait RoundCtxTrait {
     fn get_deadline(&self) -> &Option<Instant>;
     fn getter_add(&mut self, getter: PortId, msg: SendPayloadMsg);
+}
 enum CommRecvOk {
     TimeoutWithoutNew,
     NewPayloadMsgs,
     NewControlMsg { net_index: usize, msg: CommCtrlMsg },
+}
 ////////////////
 fn would_block(err: &std::io::Error) -> bool {
     err.kind() == std::io::ErrorKind::WouldBlock
+}
 impl TokenTarget {
     const HALFWAY_INDEX: usize = usize::MAX / 2;
     const MAX_INDEX: usize = usize::MAX;
     const WAKER_TOKEN: usize = Self::MAX_INDEX;
+}
 impl From<Token> for TokenTarget {
     fn from(Token(index): Token) -> Self {
         if index == Self::WAKER_TOKEN {
             TokenTarget::Waker
         } else if let Some(shifted) = index.checked_sub(Self::HALFWAY_INDEX) {
             TokenTarget::UdpEndpoint { index: shifted }
         } else {
             TokenTarget::NetEndpoint { index }
+        }
+    }
+}
 impl Into<Token> for TokenTarget {
     fn into(self) -> Token {
         match self {
             TokenTarget::Waker => Token(Self::WAKER_TOKEN),
             TokenTarget::UdpEndpoint { index } => Token(index + Self::HALFWAY_INDEX),
             TokenTarget::NetEndpoint { index } => Token(index),
+        }
+    }
+}
 impl<T: std::cmp::Ord> VecSet<T> {
     fn new(mut vec: Vec<T>) -> Self {
         vec.sort();
         vec.dedup();
         Self { vec }
+    }
     fn contains(&self, element: &T) -> bool {
         self.vec.binary_search(element).is_ok()
+    }
     fn insert(&mut self, element: T) -> bool {
         match self.vec.binary_search(&element) {
             Ok(_) => false,
             Err(index) => {
                 self.vec.insert(index, element);
                 true
+            }
+        }
+    }
     fn iter(&self) -> std::slice::Iter<T> {
         self.vec.iter()
+    }
     fn pop(&mut self) -> Option<T> {
         self.vec.pop()
+    }
+}
 impl PortInfo {
     fn spec_var_for(&self, port: PortId) -> SpecVar {
         SpecVar(match self.polarities.get(&port).unwrap() {
             Getter => port,
             Putter => *self.peers.get(&port).unwrap(),
         })
+    }
+}
 impl SpecVarStream {
     fn next(&mut self) -> SpecVar {
         let phantom_port: PortId =
             Id { connector_id: self.connector_id, u32_suffix: self.port_suffix_stream.next() }
                 .into();
         SpecVar(phantom_port)
+    }
+}
 impl IdManager {
     fn new(connector_id: ConnectorId) -> Self {
         Self {
             connector_id,
             port_suffix_stream: Default::default(),
             proto_component_suffix_stream: Default::default(),
+        }
+    }
     fn new_spec_var_stream(&self) -> SpecVarStream {
         SpecVarStream {
             connector_id: self.connector_id,
             port_suffix_stream: self.port_suffix_stream.clone(),
+        }
+    }
     fn new_port_id(&mut self) -> PortId {
         Id { connector_id: self.connector_id, u32_suffix: self.port_suffix_stream.next() }.into()
+    }
     fn new_proto_component_id(&mut self) -> ProtoComponentId {
         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.unphased.logger, "Connector dropping. Goodbye!");
+    }
+}
 impl Connector {
     fn random_id() -> ConnectorId {
         type Bytes8 = [u8; std::mem::size_of::<ConnectorId>()];
         unsafe {
             let mut bytes = std::mem::MaybeUninit::<Bytes8>::uninit();
             // getrandom is the canonical crate for a small, secure rng
             getrandom::getrandom(&mut *bytes.as_mut_ptr()).unwrap();
             // safe! representations of all valid Byte8 values are valid ConnectorId values
             std::mem::transmute::<_, _>(bytes.assume_init())
+        }
+    }
     pub fn swap_logger(&mut self, mut new_logger: Box<dyn Logger>) -> Box<dyn Logger> {
         std::mem::swap(&mut self.unphased.logger, &mut new_logger);
         new_logger
+    }
     pub fn get_logger(&mut self) -> &mut dyn Logger {
         &mut *self.unphased.logger
+    }
     pub fn new_port_pair(&mut self) -> [PortId; 2] {
         let cu = &mut self.unphased;
         // adds two new associated ports, related to each other, and exposed to the native
         let [o, i] = [cu.id_manager.new_port_id(), cu.id_manager.new_port_id()];
         cu.native_ports.insert(o);
         cu.native_ports.insert(i);
         // {polarity, peer, route} known. {} unknown.
         cu.port_info.polarities.insert(o, Putter);
         cu.port_info.polarities.insert(i, Getter);
         cu.port_info.peers.insert(o, i);
         cu.port_info.peers.insert(i, o);
         let route = Route::LocalComponent(ComponentId::Native);
         cu.port_info.routes.insert(o, route);
         cu.port_info.routes.insert(i, route);
         log!(cu.logger, "Added port pair (out->in) {:?} -> {:?}", o, i);
         [o, i]
+    }
     pub fn add_component(
         &mut self,
         identifier: &[u8],
         ports: &[PortId],
     ) -> Result<(), AddComponentError> {
         // called by the USER. moves ports owned by the NATIVE
         use AddComponentError as Ace;
         // 1. check if this is OK
         let cu = &mut self.unphased;
         let polarities = cu.proto_description.component_polarities(identifier)?;
         if polarities.len() != ports.len() {
             return Err(Ace::WrongNumberOfParamaters { expected: polarities.len() });
+        }
         for (&expected_polarity, port) in polarities.iter().zip(ports.iter()) {
             if !cu.native_ports.contains(port) {
                 return Err(Ace::UnknownPort(*port));
+            }
             if expected_polarity != *cu.port_info.polarities.get(port).unwrap() {
                 return Err(Ace::WrongPortPolarity { port: *port, expected_polarity });
+            }
+        }
         // 3. remove ports from old component & update port->route
         let new_id = cu.id_manager.new_proto_component_id();
         for port in ports.iter() {
             cu.port_info.routes.insert(*port, Route::LocalComponent(ComponentId::Proto(new_id)));
+        }
         cu.native_ports.retain(|port| !ports.contains(port));
         // 4. add new component
         cu.proto_components.insert(
             new_id,
             ProtoComponent {
                 state: cu.proto_description.new_main_component(identifier, ports),
                 ports: ports.iter().copied().collect(),
             },
         );
         Ok(())
+    }
+}
 impl Predicate {
     #[inline]
     pub fn inserted(mut self, k: SpecVar, v: SpecVal) -> Self {
         self.assigned.insert(k, v);
         self
+    }
     pub fn assigns_subset(&self, maybe_superset: &Self) -> bool {
         for (var, val) in self.assigned.iter() {
             match maybe_superset.assigned.get(var) {
                 Some(val2) if val2 == val => {}
                 _ => return false, // var unmapped, or mapped differently
+            }
+        }
         true
+    }
     // returns true IFF self.unify would return Equivalent OR FormerNotLatter
     // pub fn consistent_with(&self, other: &Self) -> bool {
     //     let [larger, smaller] =
     //         if self.assigned.len() > other.assigned.len() { [self, other] } else { [other, self] };
     //     for (var, val) in smaller.assigned.iter() {
     //         match larger.assigned.get(var) {
     //             Some(val2) if val2 != val => return false,
     //             _ => {}
     //         }
     //     }
     //     true
     // }
     /// Given self and other, two predicates, return the predicate whose
     /// assignments are the union of those of self and other.
     ///
     fn assignment_union(&self, other: &Self) -> AssignmentUnionResult {
         use AssignmentUnionResult as Aur;
         // iterators over assignments of both predicates. Rely on SORTED ordering of BTreeMap's keys.
         let [mut s_it, mut o_it] = [self.assigned.iter(), other.assigned.iter()];
         let [mut s, mut o] = [s_it.next(), o_it.next()];
         // lists of assignments in self but not other and vice versa.
         let [mut s_not_o, mut o_not_s] = [vec![], vec![]];
         loop {
             match [s, o] {
                 [None, None] => break,
                 [None, Some(x)] => {
                     o_not_s.push(x);
                     o_not_s.extend(o_it);
                     break;
+                }
                 [Some(x), None] => {
                     s_not_o.push(x);
                     s_not_o.extend(s_it);
                     break;
+                }
                 [Some((sid, sb)), Some((oid, ob))] => {
                     if sid < oid {
                         // o is missing this element
                         s_not_o.push((sid, sb));
                         s = s_it.next();
                     } else if sid > oid {
                         // s is missing this element
                         o_not_s.push((oid, ob));
                         o = o_it.next();
                     } else if sb != ob {
                         assert_eq!(sid, oid);
                         // both predicates assign the variable but differ on the value
                         return Aur::Nonexistant;
                     } else {
                         // both predicates assign the variable to the same value
                         s = s_it.next();
                         o = o_it.next();
+                    }
+                }
+            }
+        }
         // Observed zero inconsistencies. A unified predicate exists...
         match [s_not_o.is_empty(), o_not_s.is_empty()] {
             [true, true] => Aur::Equivalent,       // ... equivalent to both.
             [false, true] => Aur::FormerNotLatter, // ... equivalent to self.
             [true, false] => Aur::LatterNotFormer, // ... equivalent to other.
             [false, false] => {
                 // ... which is the union of the predicates' assignments but
                 //     is equivalent to neither self nor other.
                 let mut new = self.clone();
                 for (&id, &b) in o_not_s {
                     new.assigned.insert(id, b);
+                }
                 Aur::New(new)
+            }
+        }
+    }
     pub fn union_with(&self, other: &Self) -> Option<Self> {
         let mut res = self.clone();
         for (&channel_id, &assignment_1) in other.assigned.iter() {
             match res.assigned.insert(channel_id, assignment_1) {
                 Some(assignment_2) if assignment_1 != assignment_2 => return None,
                 _ => {}
+            }
+        }
         Some(res)
+    }
     pub fn query(&self, var: SpecVar) -> Option<SpecVal> {
         self.assigned.get(&var).copied()
+    }
+}
 impl<T: Debug + std::cmp::Ord> Debug for VecSet<T> {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         f.debug_set().entries(self.vec.iter()).finish()
+    }
+}
 impl Debug for Predicate {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         f.debug_tuple("Predicate").field(&self.assigned).finish()
+    }
+}
 impl serde::Serialize for SerdeProtocolDescription {
     fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
     where
         S: serde::Serializer,
+    {
         let inner: &ProtocolDescription = &self.0;
         inner.serialize(serializer)
+    }
+}
 impl<'de> serde::Deserialize<'de> for SerdeProtocolDescription {
     fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
     where
         D: serde::Deserializer<'de>,
+    {
         let inner: ProtocolDescription = ProtocolDescription::deserialize(deserializer)?;

src/runtime/setup.rs

➞

Show inline comments

 use crate::common::*;
 use crate::runtime::*;
 impl Connector {
     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 {
             unphased: ConnectorUnphased {
                 proto_description,
                 proto_components: Default::default(),
                 logger,
                 id_manager: IdManager::new(connector_id),
                 native_ports: Default::default(),
                 port_info: Default::default(),
             },
             phased: ConnectorPhased::Setup(Box::new(ConnectorSetup {
                 net_endpoint_setups: Default::default(),
                 udp_endpoint_setups: Default::default(),
                 surplus_sockets,
             })),
+        }
+    }
     pub fn new_udp_port(
         &mut self,
         polarity: Polarity,
         local_addr: SocketAddr,
         peer_addr: SocketAddr,
     ) -> Result<PortId, WrongStateError> {
+    ) -> Result<[PortId; 2], WrongStateError> {
         let Self { unphased: cu, phased } = self;
         match phased {
             ConnectorPhased::Communication(..) => Err(WrongStateError),
             ConnectorPhased::Setup(setup) => {
                 let udp_index = setup.udp_endpoint_setups.len();
                 let [port_nat, port_udp] =
                     [cu.id_manager.new_port_id(), cu.id_manager.new_port_id()];
                 cu.native_ports.insert(port_nat);
                 cu.port_info.peers.insert(port_nat, port_udp);
                 cu.port_info.peers.insert(port_udp, port_nat);
                 cu.port_info.routes.insert(port_nat, Route::LocalComponent(ComponentId::Native));
                 cu.port_info.routes.insert(port_udp, Route::UdpEndpoint { index: udp_index });
                 cu.port_info.polarities.insert(port_nat, polarity);
                 cu.port_info.polarities.insert(port_udp, !polarity);
                 let mut npid = || cu.id_manager.new_port_id();
                 let [nin, nout, uin, uout] = [npid(), npid(), npid(), npid()];
                 cu.native_ports.insert(nin);
                 cu.native_ports.insert(nout);
                 cu.port_info.polarities.insert(nin, Getter);
                 cu.port_info.polarities.insert(nout, Putter);
                 cu.port_info.polarities.insert(uin, Getter);
                 cu.port_info.polarities.insert(uout, Putter);
                 cu.port_info.peers.insert(nin, uout);
                 cu.port_info.peers.insert(nout, uin);
                 cu.port_info.peers.insert(uin, nout);
                 cu.port_info.peers.insert(uout, nin);
                 cu.port_info.routes.insert(nin, Route::LocalComponent(ComponentId::Native));
                 cu.port_info.routes.insert(nout, Route::LocalComponent(ComponentId::Native));
                 cu.port_info.routes.insert(uin, Route::UdpEndpoint { index: udp_index });
                 cu.port_info.routes.insert(uout, Route::UdpEndpoint { index: udp_index });
                 setup.udp_endpoint_setups.push(UdpEndpointSetup {
                     local_addr,
                     peer_addr,
-                    local_port: port_nat,
+                    getter_for_incoming: nin,
                 });
-                Ok(port_nat)
+                Ok([nout, nin])
+            }
+        }
+    }
     pub fn new_net_port(
         &mut self,
         polarity: Polarity,
         sock_addr: SocketAddr,
         endpoint_polarity: EndpointPolarity,
     ) -> Result<PortId, WrongStateError> {
         let Self { unphased: cu, phased } = self;
         match phased {
             ConnectorPhased::Communication(..) => Err(WrongStateError),
             ConnectorPhased::Setup(setup) => {
                 let local_port = cu.id_manager.new_port_id();
                 cu.native_ports.insert(local_port);
                 // {polarity, route} known. {peer} unknown.
                 cu.port_info.polarities.insert(local_port, polarity);
                 cu.port_info.routes.insert(local_port, Route::LocalComponent(ComponentId::Native));
                 log!(
                     cu.logger,
                     "Added net port {:?} with polarity {:?} addr {:?} endpoint_polarity {:?}",
                     local_port,
                     polarity,
                     &sock_addr,
                     endpoint_polarity
                 );
                 setup.net_endpoint_setups.push(NetEndpointSetup {
                     sock_addr,
                     endpoint_polarity,
                     local_port,
+                    getter_for_incoming: local_port,
                 });
                 Ok(local_port)
+            }
+        }
+    }
     pub fn connect(&mut self, timeout: Option<Duration>) -> Result<(), ConnectError> {
         use ConnectError as Ce;
         let Self { unphased: cu, phased } = self;
         match &phased {
             ConnectorPhased::Communication { .. } => {
                 log!(cu.logger, "Call to connecting in connected state");
                 Err(Ce::AlreadyConnected)
+            }
             ConnectorPhased::Setup(setup) => {
                 log!(cu.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 *cu.logger,
                     &setup.net_endpoint_setups,
                     &setup.udp_endpoint_setups,
                     &mut cu.port_info,
                     &deadline,
                 )?;
                 log!(
                     cu.logger,
                     "Successfully connected {} endpoints",
                     endpoint_manager.net_endpoint_store.endpoint_exts.len()
                 );
                 // leader election and tree construction
                 let neighborhood = init_neighborhood(
                     cu.id_manager.connector_id,
                     &mut *cu.logger,
                     &mut endpoint_manager,
                     &deadline,
                 )?;
                 log!(cu.logger, "Successfully created neighborhood {:?}", &neighborhood);
                 let mut comm = ConnectorCommunication {
                     round_index: 0,
                     endpoint_manager,
                     neighborhood,
                     native_batches: vec![Default::default()],
                     round_result: Ok(None),
                 };
                 if cfg!(feature = "session_optimization") {
                     session_optimize(cu, &mut comm, &deadline)?;
+                }
                 log!(cu.logger, "connect() finished. setup phase complete");
                 self.phased = ConnectorPhased::Communication(Box::new(comm));
                 Ok(())
+            }
+        }
+    }
+}
 fn new_endpoint_manager(
     logger: &mut dyn Logger,
     net_endpoint_setups: &[NetEndpointSetup],
     udp_endpoint_setups: &[UdpEndpointSetup],
     port_info: &mut PortInfo,
     deadline: &Option<Instant>,
 ) -> Result<EndpointManager, ConnectError> {
     ////////////////////////////////////////////
     use std::sync::atomic::AtomicBool;
+    use std::sync::atomic::{AtomicBool, Ordering::SeqCst};
     use ConnectError as Ce;
     const BOTH: Interest = Interest::READABLE.add(Interest::WRITABLE);
     const WAKER_PERIOD: Duration = Duration::from_millis(300);
     struct WakerState {
         continue_signal: AtomicBool,
         waker: mio::Waker,
+    }
     impl WakerState {
         fn waker_loop(&self) {
             while self.continue_signal.load(SeqCst) {
                 std::thread::sleep(WAKER_PERIOD);
                 let _ = self.waker.wake();
+            }
+        }
         fn waker_stop(&self) {
             self.continue_signal.store(false, SeqCst);
             // TODO keep waker registered?
+        }
+    }
     struct Todo {
         // becomes completed once sent_local_port && recv_peer_port.is_some()
         // we send local port if we haven't already and we receive a writable event
         // we recv peer port if we haven't already and we receive a readbale event
         todo_endpoint: TodoEndpoint,
         endpoint_setup: NetEndpointSetup,
         sent_local_port: bool,          // true <-> I've sent my local port
         recv_peer_port: Option<PortId>, // Some(..) <-> I've received my peer's port
+    }
     struct UdpTodo {
         local_port: PortId,
         // becomes completed once we receive our first writable event
         getter_for_incoming: PortId,
         sock: UdpSocket,
+    }
     enum TodoEndpoint {
         Accepting(TcpListener),
         NetEndpoint(NetEndpoint),
+    }
     ////////////////////////////////////////////
     // 1. Start to construct EndpointManager
     const WAKER_PERIOD: Duration = Duration::from_millis(300);
     struct WakerState {
         continue_signal: AtomicBool,
         waker: mio::Waker,
+    }
     let mut waker_state: Option<Arc<WakerState>> = None;
     let mut poll = Poll::new().map_err(|_| Ce::PollInitFailed)?;
     let mut events = Events::with_capacity(net_endpoint_setups.len() * 2 + 4);
     let mut events =
         Events::with_capacity((net_endpoint_setups.len() + udp_endpoint_setups.len()) * 2 + 4);
     let [mut net_polled_undrained, udp_polled_undrained] = [VecSet::default(), VecSet::default()];
     let mut delayed_messages = vec![];
-    // 2. create a registered (TcpListener/Endpoint) for passive / active respectively
+    // 2. Create net/udp TODOs, each already registered with poll
     let mut net_todos = net_endpoint_setups
         .iter()
         .enumerate()
         .map(|(index, endpoint_setup)| {
             let token = TokenTarget::NetEndpoint { index }.into();
             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::NetEndpoint(NetEndpoint { stream, inbox: vec![] })
             } else {
                 let mut listener = TcpListener::bind(endpoint_setup.sock_addr)
                     .map_err(|_| Ce::BindFailed(endpoint_setup.sock_addr))?;
                 poll.registry().register(&mut listener, token, BOTH).unwrap();
                 TodoEndpoint::Accepting(listener)
             };
             Ok(Todo {
                 todo_endpoint,
                 sent_local_port: false,
                 recv_peer_port: None,
                 endpoint_setup: endpoint_setup.clone(),
             })
         })
         .collect::<Result<Vec<Todo>, ConnectError>>()?;
     let udp_todos = udp_endpoint_setups
         .iter()
         .enumerate()
         .map(|(index, endpoint_setup)| {
             let mut sock = UdpSocket::bind(endpoint_setup.local_addr)
                 .map_err(|_| Ce::BindFailed(endpoint_setup.local_addr))?;
             sock.connect(endpoint_setup.peer_addr)
                 .map_err(|_| Ce::UdpConnectFailed(endpoint_setup.peer_addr))?;
             poll.registry()
                 .register(&mut sock, TokenTarget::UdpEndpoint { index }.into(), Interest::WRITABLE)
                 .unwrap();
-            Ok(UdpTodo { sock, local_port: endpoint_setup.local_port })
+            Ok(UdpTodo { sock, getter_for_incoming: endpoint_setup.getter_for_incoming })
         })
         .collect::<Result<Vec<UdpTodo>, 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
     // all in connect_failed are NOT registered with Poll
     let mut connect_failed: HashSet<usize> = Default::default();
     // TODO register udps, and all them to incomplete list
     // Initially, (1) no net connections have failed, and (2) all udp and net endpoint setups are incomplete
     let mut net_connect_retry_later: HashSet<usize> = Default::default();
     let mut setup_incomplete: HashSet<TokenTarget> = {
         let net_todo_targets_iter =
             (0..net_todos.len()).map(|index| TokenTarget::NetEndpoint { index });
         let udp_todo_targets_iter =
             (0..udp_todos.len()).map(|index| TokenTarget::UdpEndpoint { index });
         net_todo_targets_iter.chain(udp_todo_targets_iter).collect()
     };
     // progress by reacting to poll events. continue until every endpoint is set up
     while !setup_incomplete.is_empty() {
         let remaining = if let Some(deadline) = deadline {
             Some(deadline.checked_duration_since(Instant::now()).ok_or(Ce::Timeout)?)
         } else {
             None
         };
         poll.poll(&mut events, remaining).map_err(|_| Ce::PollFailed)?;
         for event in events.iter() {
             let token = event.token();
             let token_target = TokenTarget::from(token);
             if !setup_incomplete.contains(&token_target) {
                 // spurious wakeup
                 continue;
+            }
             match token_target {
                 TokenTarget::Waker => {
                     log!(
                         logger,
                         "Notification from waker. connect_failed is {:?}",
-                        connect_failed.iter()
+                        net_connect_retry_later.iter()
                     );
                     assert!(waker_state.is_some());
-                    for net_index in connect_failed.drain() {
+                    for net_index in net_connect_retry_later.drain() {
                         let net_todo = &mut net_todos[net_index];
                         log!(
                             logger,
                             "Restarting connection with endpoint {:?} {:?}",
                             net_index,
                             net_todo.endpoint_setup.sock_addr
                         );
                         match &mut net_todo.todo_endpoint {
                             TodoEndpoint::NetEndpoint(endpoint) => {
                                 let mut new_stream =
                                     TcpStream::connect(net_todo.endpoint_setup.sock_addr)
                                         .expect("mio::TcpStream connect should not fail!");
                                 std::mem::swap(&mut endpoint.stream, &mut new_stream);
                                 let token = TokenTarget::NetEndpoint { index: net_index }.into();
                                 poll.registry()
                                     .register(&mut endpoint.stream, token, BOTH)
                                     .unwrap();
+                            }
                             _ => unreachable!(),
+                        }
+                    }
+                }
                 TokenTarget::UdpEndpoint { index } => {
                     if !setup_incomplete.contains(&token_target) {
                         // spurious wakeup. this endpoint has already been set up!
                         continue;
+                    }
                     let udp_todo: &UdpTodo = &udp_todos[index];
                     if event.is_error() {
                         return Err(Ce::BindFailed(udp_todo.sock.local_addr().unwrap()));
+                    }
                     setup_incomplete.remove(&token_target);
+                }
                 TokenTarget::NetEndpoint { index } => {
                     let net_todo = &mut net_todos[index];
                     if let TodoEndpoint::Accepting(listener) = &mut net_todo.todo_endpoint {
                         // FIRST try complete this connection
                         match listener.accept() {
                             Err(e) if would_block(&e) => {
                                 log!(logger, "Spurious wakeup on listener {:?}", index)
+                            }
                             Err(e) if would_block(&e) => continue, // spurious wakeup
                             Err(_) => {
                                 log!(logger, "accept() failure on index {}", index);
                                 return Err(Ce::AcceptFailed(listener.local_addr().unwrap()));
+                            }
                             Ok((mut stream, peer_addr)) => {
                                 // success!
                                 // successfully accepted the active peer
                                 // reusing the token, but now for the stream and not the listener
                                 poll.registry().deregister(listener).unwrap();
                                 // reusing original token as-is
                                 poll.registry().register(&mut stream, token, BOTH).unwrap();
                                 log!(
                                     logger,
                                     "Endpoint[{}] accepted a connection from {:?}",
                                     index,
                                     peer_addr
                                 );
                                 let net_endpoint = NetEndpoint { stream, inbox: vec![] };
                                 net_todo.todo_endpoint = TodoEndpoint::NetEndpoint(net_endpoint);
+                            }
+                        }
+                    }
                     if let TodoEndpoint::NetEndpoint(net_endpoint) = &mut net_todo.todo_endpoint {
                         if event.is_error() {
                             if net_todo.endpoint_setup.endpoint_polarity
                                 == EndpointPolarity::Passive
+                            {
                                 // right now you cannot retry an acceptor. return failure
                                 return Err(Ce::AcceptFailed(
                                     net_endpoint.stream.local_addr().unwrap(),
                                 ));
+                            }
                             // this actively-connecting endpoint failed to connect!
-                            if connect_failed.insert(index) {
+                            if net_connect_retry_later.insert(index) {
                                 log!(
                                     logger,
                                     "Connection failed for {:?}. List is {:?}",
                                     index,
-                                    connect_failed.iter()
+                                    net_connect_retry_later.iter()
                                 );
                                 poll.registry().deregister(&mut net_endpoint.stream).unwrap();
                             } else {
                                 // spurious wakeup.
+                                // spurious wakeup. already scheduled to retry connect later
                                 continue;
+                            }
                             if waker_state.is_none() {
                                 log!(logger, "First connect failure. Starting waker thread");
                                 let arc = Arc::new(WakerState {
                                     waker: mio::Waker::new(
                                         poll.registry(),
                                         TokenTarget::Waker.into(),
+                                    )
                                     .unwrap(),
                                     continue_signal: true.into(),
                                 });
                                 let moved_arc = arc.clone();
                                 waker_state = Some(arc);
                                 std::thread::spawn(move || {
                                     while moved_arc
                                         .continue_signal
                                         .load(std::sync::atomic::Ordering::SeqCst)
+                                    {
                                         std::thread::sleep(WAKER_PERIOD);
                                         let _ = moved_arc.waker.wake();
+                                    }
                                 });
                                 std::thread::spawn(move || moved_arc.waker_loop());
+                            }
                             continue;
+                        }
                         // event wasn't ERROR
                         if connect_failed.contains(&index) {
                             // spurious wakeup
                         if net_connect_retry_later.contains(&index) {
                             // spurious wakeup. already scheduled to retry connect later
                             continue;
+                        }
                         let local_polarity =
                             *port_info.polarities.get(&net_todo.endpoint_setup.local_port).unwrap();
                         if !setup_incomplete.contains(&token_target) {
                             // spurious wakeup. this endpoint has already been completed!
                             if event.is_readable() {
                                 net_polled_undrained.insert(index);
+                            }
                             continue;
+                        }
                         let local_polarity = *port_info
                             .polarities
                             .get(&net_todo.endpoint_setup.getter_for_incoming)
                             .unwrap();
                         if event.is_writable() && !net_todo.sent_local_port {
                             // can write and didn't send setup msg yet? Do so!
                             let msg = Msg::SetupMsg(SetupMsg::MyPortInfo(MyPortInfo {
                                 polarity: local_polarity,
-                                port: net_todo.endpoint_setup.local_port,
+                                port: net_todo.endpoint_setup.getter_for_incoming,
                             }));
                             net_endpoint
                                 .send(&msg)
                                 .map_err(|e| {
                                     Ce::NetEndpointSetupError(
                                         net_endpoint.stream.local_addr().unwrap(),
                                         e,
+                                    )
                                 })
                                 .unwrap();
                             log!(logger, "endpoint[{}] sent msg {:?}", index, &msg);
                             net_todo.sent_local_port = true;
+                        }
                         if event.is_readable() && net_todo.recv_peer_port.is_none() {
                             // can read and didn't recv setup msg yet? Do so!
                             let maybe_msg = net_endpoint.try_recv(logger).map_err(|e| {
                                 Ce::NetEndpointSetupError(
                                     net_endpoint.stream.local_addr().unwrap(),
                                     e,
+                                )
                             })?;
                             if maybe_msg.is_some() && !net_endpoint.inbox.is_empty() {
                                 net_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(
-                                            net_todo.endpoint_setup.local_port,
+                                            net_todo.endpoint_setup.getter_for_incoming,
                                         ));
+                                    }
                                     net_todo.recv_peer_port = Some(peer_info.port);
                                     // 1. finally learned the peer of this port!
                                     port_info
                                         .peers
                                         .insert(net_todo.endpoint_setup.local_port, peer_info.port);
                                     port_info.peers.insert(
                                         net_todo.endpoint_setup.getter_for_incoming,
                                         peer_info.port,
                                     );
                                     // 2. learned the info of this peer port
                                     port_info.polarities.insert(peer_info.port, peer_info.polarity);
                                     port_info
                                         .peers
                                         .insert(peer_info.port, net_todo.endpoint_setup.local_port);
                                     port_info.peers.insert(
                                         peer_info.port,
                                         net_todo.endpoint_setup.getter_for_incoming,
                                     );
                                     if let Some(route) = port_info.routes.get(&peer_info.port) {
                                         // check just for logging purposes
                                         log!(
                                             logger,
                                             "Special case! Route to peer {:?} already known to be {:?}. Leave untouched",
                                             peer_info.port,
                                             route
                                         );
+                                    }
                                     port_info
                                         .routes
                                         .entry(peer_info.port)
                                         .or_insert(Route::NetEndpoint { index });
+                                }
                                 Some(inappropriate_msg) => {
                                     log!(
                                         logger,
                                         "delaying msg {:?} during channel setup phase",
                                         inappropriate_msg
                                     );
                                     delayed_messages.push((index, inappropriate_msg));
+                                }
+                            }
+                        }
                         // is the setup for this net_endpoint now complete?
                         if net_todo.sent_local_port && net_todo.recv_peer_port.is_some() {
                             // yes! connected, sent my info and received peer's info
                             setup_incomplete.remove(&token_target);
                             log!(logger, "endpoint[{}] is finished!", index);
+                        }
+                    }
+                }
+            }
+        }
         events.clear();
+    }
     log!(logger, "Endpoint setup complete! Cleaning up and building structures");
     if let Some(arc) = waker_state {
         log!(logger, "Sending waker the stop signal");
         arc.continue_signal.store(false, std::sync::atomic::Ordering::SeqCst);
         // TODO leave the waker registered?
     if let Some(ws) = waker_state.take() {
         ws.waker_stop();
+    }
     let net_endpoint_exts = net_todos
         .into_iter()
         .enumerate()
         .map(|(index, Todo { todo_endpoint, endpoint_setup, .. })| NetEndpointExt {
             net_endpoint: match todo_endpoint {
                 TodoEndpoint::NetEndpoint(mut net_endpoint) => {
                     let token = TokenTarget::NetEndpoint { index }.into();
                     poll.registry()
                         .reregister(&mut net_endpoint.stream, token, Interest::READABLE)
                         .unwrap();
                     net_endpoint
+                }
                 _ => unreachable!(),
             },
-            getter_for_incoming: endpoint_setup.local_port,
+            getter_for_incoming: endpoint_setup.getter_for_incoming,
         })
         .collect();
     let udp_endpoint_exts = udp_todos
         .into_iter()
         .enumerate()
         .map(|(index, udp_todo)| {
-            let UdpTodo { mut sock, local_port } = udp_todo;
+            let UdpTodo { mut sock, getter_for_incoming } = udp_todo;
             let token = TokenTarget::UdpEndpoint { index }.into();
             poll.registry().reregister(&mut sock, token, Interest::READABLE).unwrap();
             UdpEndpointExt {
                 sock,
                 outgoing_payloads: Default::default(),
                 incoming_round_spec_var: None,
-                getter_for_incoming: local_port,
                 getter_for_incoming,
                 incoming_payloads: Default::default(),
+            }
         })
         .collect();
     Ok(EndpointManager {
         poll,
         events,
         undelayed_messages: delayed_messages, // no longer delayed
         delayed_messages: Default::default(),
         net_endpoint_store: EndpointStore {
             endpoint_exts: net_endpoint_exts,
             polled_undrained: net_polled_undrained,
         },
         udp_endpoint_store: EndpointStore {
             endpoint_exts: udp_endpoint_exts,
             polled_undrained: udp_polled_undrained,
         },
         udp_in_buffer: Default::default(),
     })
+}
 fn init_neighborhood(
     connector_id: ConnectorId,
     logger: &mut dyn Logger,
     em: &mut EndpointManager,
     deadline: &Option<Instant>,
 ) -> Result<Neighborhood, ConnectError> {
     ////////////////////////////////
     use {ConnectError as Ce, Msg::SetupMsg as S, SetupMsg as Sm};
     #[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(Sm::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.
     NOTE the distinction between PARENT and LEADER
     */
     log!(logger, "beginning neighborhood construction");
     if em.num_net_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_net_endpoints());
     let mut awaiting = HashSet::with_capacity(em.num_net_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();
         '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(Sm::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(Sm::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() {
                                 log!(logger, "Special case! Only neighbor is parent. Replying to {:?} msg {:?}", recv_index, &msg);
                                 em.send_to_setup(recv_index, &msg)?;
+                            }
+                        }
                         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 {
                                     let election_result: WaveState = best_wave;
                                     log!(logger, "Election won! Result {:?}", &election_result);
                                     break 'election election_result;
+                                }
+                            }
+                        }
+                    }
+                }
                 msg @ S(Sm::YouAreMyParent) | msg @ S(Sm::MyPortInfo(_)) => {
                     log!(logger, "Endpont {:?} sent unexpected msg! {:?}", recv_index, &msg);
                     return Err(Ce::SetupAlgMisbehavior);
+                }
                 msg @ S(Sm::SessionScatter { .. })
                 | msg @ S(Sm::SessionGather { .. })
                 | msg @ Msg::CommMsg { .. } => {
                     log!(logger, "delaying msg {:?} during election algorithm", msg);
                     em.delayed_messages.push((recv_index, msg));
+                }
+            }
+        }
     };
     // 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(Sm::YouAreMyParent))?;
         } else {
             awaiting.insert(index);
             em.send_to_setup(
                 index,
                 &S(Sm::LeaderAnnounce { tree_leader: election_result.leader }),
             )?;
+        }
+    }
     let mut children = vec![];
     em.undelay_all();
     while !awaiting.is_empty() {
         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(Sm::LeaderAnnounce { .. }) => {
                 // not a child
                 log!(
                     logger,
                     "Got reply from non-child index {:?}. Children: {:?}",
                     recv_index,
                     children.iter()
                 );
                 if !awaiting.remove(&recv_index) {
                     return Err(Ce::SetupAlgMisbehavior);
+                }
+            }
             S(Sm::YouAreMyParent) => {
                 if !awaiting.remove(&recv_index) {
                     log!(
                         logger,
                         "Got reply from child index {:?}. Children before... {:?}",
                         recv_index,
                         children.iter()
                     );
                     return Err(Ce::SetupAlgMisbehavior);
+                }
                 children.push(recv_index);
+            }
             msg @ S(Sm::MyPortInfo(_)) | msg @ S(Sm::LeaderWave { .. }) => {
                 log!(logger, "discarding old message {:?} during election", msg);
+            }
             msg @ S(Sm::SessionScatter { .. })
             | msg @ S(Sm::SessionGather { .. })
             | msg @ Msg::CommMsg { .. } => {
                 log!(logger, "delaying msg {:?} during election", msg);
                 em.delayed_messages.push((recv_index, msg));
+            }
+        }
+    }
     children.shrink_to_fit();
     let neighborhood =
         Neighborhood { parent: election_result.parent, children: VecSet::new(children) };
     log!(logger, "Neighborhood constructed {:?}", &neighborhood);
     Ok(neighborhood)
+}
 fn session_optimize(
     cu: &mut ConnectorUnphased,
     comm: &mut ConnectorCommunication,
     deadline: &Option<Instant>,
 ) -> Result<(), ConnectError> {
     ////////////////////////////////////////
     use {ConnectError as Ce, Msg::SetupMsg as S, SetupMsg as Sm};
     ////////////////////////////////////////
     log!(cu.logger, "Beginning session optimization");
     // populate session_info_map from a message per child
     let mut unoptimized_map: HashMap<ConnectorId, SessionInfo> = Default::default();
     let mut awaiting: HashSet<usize> = comm.neighborhood.children.iter().copied().collect();
     comm.endpoint_manager.undelay_all();
     while !awaiting.is_empty() {
         log!(
             cu.logger,
             "Session gather loop. awaiting info from children {:?}...",
             awaiting.iter()
         );
         let (recv_index, msg) =
             comm.endpoint_manager.try_recv_any_setup(&mut *cu.logger, deadline)?;
         log!(cu.logger, "Received from index {:?} msg {:?}", &recv_index, &msg);
         match msg {
             S(Sm::SessionGather { unoptimized_map: child_unoptimized_map }) => {
                 if !awaiting.remove(&recv_index) {
                     log!(
                         cu.logger,
                         "Wasn't expecting session info from {:?}. Got {:?}",
                         recv_index,
                         &child_unoptimized_map
                     );
                     return Err(Ce::SetupAlgMisbehavior);
+                }
                 unoptimized_map.extend(child_unoptimized_map.into_iter());
+            }
             msg @ S(Sm::YouAreMyParent)
             | msg @ S(Sm::MyPortInfo(..))
             | msg @ S(Sm::LeaderAnnounce { .. })
             | msg @ S(Sm::LeaderWave { .. }) => {
                 log!(cu.logger, "discarding old message {:?} during election", msg);
+            }
             msg @ S(Sm::SessionScatter { .. }) => {
                 log!(
                     cu.logger,
                     "Endpoint {:?} sent unexpected scatter! {:?} I've not contributed yet!",
                     recv_index,
                     &msg
                 );
                 return Err(Ce::SetupAlgMisbehavior);
+            }
             msg @ Msg::CommMsg(..) => {
                 log!(cu.logger, "delaying msg {:?} during session optimization", msg);
                 comm.endpoint_manager.delayed_messages.push((recv_index, msg));
+            }
+        }
+    }
     log!(
         cu.logger,
         "Gathered all children's maps. ConnectorId set is... {:?}",
         unoptimized_map.keys()
     );
     let my_session_info = SessionInfo {
         port_info: cu.port_info.clone(),
         proto_components: cu.proto_components.clone(),
         serde_proto_description: SerdeProtocolDescription(cu.proto_description.clone()),
         endpoint_incoming_to_getter: comm
             .endpoint_manager
             .net_endpoint_store
             .endpoint_exts
             .iter()
             .map(|ee| ee.getter_for_incoming)
             .collect(),
     };
     unoptimized_map.insert(cu.id_manager.connector_id, my_session_info);
     log!(cu.logger, "Inserting my own info. Unoptimized subtree map is {:?}", &unoptimized_map);
     // acquire the optimized info...
     let optimized_map = if let Some(parent) = comm.neighborhood.parent {
         // ... as a message from my parent
         log!(cu.logger, "Forwarding gathered info to parent {:?}", parent);
         let msg = S(Sm::SessionGather { unoptimized_map });
         comm.endpoint_manager.send_to_setup(parent, &msg)?;
         'scatter_loop: loop {
             log!(
                 cu.logger,
                 "Session scatter recv loop. awaiting info from children {:?}...",
                 awaiting.iter()
             );
             let (recv_index, msg) =
                 comm.endpoint_manager.try_recv_any_setup(&mut *cu.logger, deadline)?;
             log!(cu.logger, "Received from index {:?} msg {:?}", &recv_index, &msg);
             match msg {
                 S(Sm::SessionScatter { optimized_map }) => {
                     if recv_index != parent {
                         log!(cu.logger, "I expected the scatter from my parent only!");
                         return Err(Ce::SetupAlgMisbehavior);
+                    }
                     break 'scatter_loop optimized_map;
+                }
                 msg @ Msg::CommMsg { .. } => {
                     log!(cu.logger, "delaying msg {:?} during scatter recv", msg);
                     comm.endpoint_manager.delayed_messages.push((recv_index, msg));
+                }
                 msg @ S(Sm::SessionGather { .. })
                 | msg @ S(Sm::YouAreMyParent)
                 | msg @ S(Sm::MyPortInfo(..))
                 | msg @ S(Sm::LeaderAnnounce { .. })
                 | msg @ S(Sm::LeaderWave { .. }) => {
                     log!(cu.logger, "discarding old message {:?} during election", msg);
+                }
+            }
+        }
     } else {
         // by computing it myself
         log!(cu.logger, "I am the leader! I will optimize this session");
         leader_session_map_optimize(&mut *cu.logger, unoptimized_map)?
     };
     log!(
         cu.logger,
         "Optimized info map is {:?}. Sending to children {:?}",
         &optimized_map,
         comm.neighborhood.children.iter()
     );
     log!(cu.logger, "All session info dumped!: {:#?}", &optimized_map);
     let optimized_info =
         optimized_map.get(&cu.id_manager.connector_id).expect("HEY NO INFO FOR ME?").clone();
     let msg = S(Sm::SessionScatter { optimized_map });
     for &child in comm.neighborhood.children.iter() {
         comm.endpoint_manager.send_to_setup(child, &msg)?;
+    }
     apply_optimizations(cu, comm, optimized_info)?;
     log!(cu.logger, "Session optimizations applied");
     Ok(())
+}
 fn leader_session_map_optimize(
     logger: &mut dyn Logger,
     unoptimized_map: HashMap<ConnectorId, SessionInfo>,
 ) -> Result<HashMap<ConnectorId, SessionInfo>, ConnectError> {
     log!(logger, "Session map optimize START");
     log!(logger, "Session map optimize END");
     Ok(unoptimized_map)
+}
 fn apply_optimizations(
     cu: &mut ConnectorUnphased,
     comm: &mut ConnectorCommunication,
     session_info: SessionInfo,
 ) -> Result<(), ConnectError> {
     let SessionInfo {
         proto_components,
         port_info,
         serde_proto_description,
         endpoint_incoming_to_getter,
     } = session_info;
     // TODO some info which should be read-only can be mutated with the current scheme
     cu.port_info = port_info;
     cu.proto_components = proto_components;
     cu.proto_description = serde_proto_description.0;
     for (ee, getter) in comm
         .endpoint_manager
         .net_endpoint_store
         .endpoint_exts

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::{fs::File, net::SocketAddr, path::Path, sync::Arc, time::Duration};
 //////////////////////////////////////////
 const MS100: Option<Duration> = Some(Duration::from_millis(100));
 const MS300: Option<Duration> = Some(Duration::from_millis(300));
 const SEC1: Option<Duration> = Some(Duration::from_secs(1));
 const SEC5: Option<Duration> = Some(Duration::from_secs(5));
 const SEC15: Option<Duration> = Some(Duration::from_secs(15));
 const MS300: Option<Duration> = Some(Duration::from_millis(300));
 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()
+}
 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)
+}
 static MINIMAL_PDL: &'static [u8] = b"
 primitive together(in ia, in ib, out oa, out ob){
   while(true) synchronous() {
     if(fires(ia)) {
       put(oa, get(ia));
       put(ob, get(ib));
+    }
+  }
+}
 ";
 lazy_static::lazy_static! {
     static ref MINIMAL_PROTO: Arc<ProtocolDescription> = {
         Arc::new(reowolf::ProtocolDescription::parse(MINIMAL_PDL).unwrap())
     };
+}
 static TEST_MSG_BYTES: &'static [u8] = b"hello";
 lazy_static::lazy_static! {
     static ref TEST_MSG: Payload = {
         Payload::from(TEST_MSG_BYTES)
     };
+}
 //////////////////////////////////////////
 #[test]
 fn basic_connector() {
     Connector::new(Box::new(DummyLogger), MINIMAL_PROTO.clone(), 0, 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 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 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 test_log_path = Path::new("./logs/new_net_port");
     let mut c = file_logged_connector(0, test_log_path);
     let sock_addrs = [next_test_addr()];
     let _ = c.new_net_port(Getter, sock_addrs[0], Passive).unwrap();
     let _ = c.new_net_port(Putter, sock_addrs[0], Active).unwrap();
+}
 #[test]
 fn trivial_connect() {
     let test_log_path = Path::new("./logs/trivial_connect");
     let mut c = file_logged_connector(0, test_log_path);
     c.connect(SEC1).unwrap();
+}
 #[test]
 fn single_node_connect() {
     let test_log_path = Path::new("./logs/single_node_connect");
     let sock_addrs = [next_test_addr()];
     let mut c = file_logged_connector(0, test_log_path);
     let _ = c.new_net_port(Getter, sock_addrs[0], Passive).unwrap();
     let _ = c.new_net_port(Putter, sock_addrs[0], Active).unwrap();
     c.connect(SEC1).unwrap();
+}
 #[test]
 fn minimal_net_connect() {
     let test_log_path = Path::new("./logs/minimal_net_connect");
     let sock_addrs = [next_test_addr()];
     scope(|s| {
         s.spawn(|_| {
             let mut c = file_logged_connector(0, test_log_path);
             let _ = c.new_net_port(Getter, sock_addrs[0], Active).unwrap();
             c.connect(SEC1).unwrap();
         });
         s.spawn(|_| {
             let mut c = file_logged_connector(1, test_log_path);
             let _ = c.new_net_port(Putter, sock_addrs[0], Passive).unwrap();
             c.connect(SEC1).unwrap();
         });
     })
     .unwrap();
+}
 #[test]
 fn put_no_sync() {
     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(SEC1).unwrap();
     c.put(o, TEST_MSG.clone()).unwrap();
+}
 #[test]
 fn wrong_polarity_bad() {
     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(SEC1).unwrap();
     c.put(i, TEST_MSG.clone()).unwrap_err();
+}
 #[test]
 fn dup_put_bad() {
     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(SEC1).unwrap();
     c.put(o, TEST_MSG.clone()).unwrap();
     c.put(o, TEST_MSG.clone()).unwrap_err();
+}
 #[test]
 fn trivial_sync() {
     let test_log_path = Path::new("./logs/trivial_sync");
     let mut c = file_logged_connector(0, test_log_path);
     c.connect(SEC1).unwrap();
     c.sync(SEC1).unwrap();
+}
 #[test]
 fn unconnected_gotten_err() {
     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 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(SEC1).unwrap();
     assert_eq!(reowolf::error::GottenError::NoPreviousRound, c.gotten(i).unwrap_err());
+}
 #[test]
 fn connected_gotten_err_ungotten() {
     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(SEC1).unwrap();
     c.sync(SEC1).unwrap();
     assert_eq!(reowolf::error::GottenError::PortDidntGet, c.gotten(i).unwrap_err());
+}
 #[test]
 fn native_polarity_checks() {
     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(SEC1).unwrap();
     // fail...
     c.get(o).unwrap_err();
     c.put(i, TEST_MSG.clone()).unwrap_err();
     // succeed..
     c.get(i).unwrap();
     c.put(o, TEST_MSG.clone()).unwrap();
+}
 #[test]
 fn native_multiple_gets() {
     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(SEC1).unwrap();
     c.get(i).unwrap();
     c.get(i).unwrap_err();
+}
 #[test]
 fn next_batch() {
     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(SEC1).unwrap();
     c.next_batch().unwrap();
     c.next_batch().unwrap();
     c.next_batch().unwrap();
+}
 #[test]
 fn native_self_msg() {
     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(SEC1).unwrap();
     c.get(i).unwrap();
     c.put(o, TEST_MSG.clone()).unwrap();
     c.sync(SEC1).unwrap();
+}
 #[test]
 fn two_natives_msg() {
     let test_log_path = Path::new("./logs/two_natives_msg");
     let sock_addrs = [next_test_addr()];
     scope(|s| {
         s.spawn(|_| {
             let mut c = file_logged_connector(0, test_log_path);
             let g = c.new_net_port(Getter, sock_addrs[0], Active).unwrap();
             c.connect(SEC1).unwrap();
             c.get(g).unwrap();
             c.sync(SEC1).unwrap();
             c.gotten(g).unwrap();
         });
         s.spawn(|_| {
             let mut c = file_logged_connector(1, test_log_path);
             let p = c.new_net_port(Putter, sock_addrs[0], Passive).unwrap();
             c.connect(SEC1).unwrap();
             c.put(p, TEST_MSG.clone()).unwrap();
             c.sync(SEC1).unwrap();
         });
     })
     .unwrap();
+}
 #[test]
 fn trivial_nondet() {
     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(SEC1).unwrap();
     c.get(i).unwrap();
     // getting 0 batch
     c.next_batch().unwrap();
     // silent 1 batch
     assert_eq!(1, c.sync(SEC1).unwrap());
     c.gotten(i).unwrap_err();
+}
 #[test]
 fn connector_pair_nondet() {
     let test_log_path = Path::new("./logs/connector_pair_nondet");
     let sock_addrs = [next_test_addr()];
     scope(|s| {
         s.spawn(|_| {
             let mut c = file_logged_connector(0, test_log_path);
             let g = c.new_net_port(Getter, sock_addrs[0], Active).unwrap();
             c.connect(SEC1).unwrap();
             c.next_batch().unwrap();
             c.get(g).unwrap();
             assert_eq!(1, c.sync(SEC1).unwrap());
             c.gotten(g).unwrap();
         });
         s.spawn(|_| {
             let mut c = file_logged_connector(1, test_log_path);
             let p = c.new_net_port(Putter, sock_addrs[0], Passive).unwrap();
             c.connect(SEC1).unwrap();
             c.put(p, TEST_MSG.clone()).unwrap();
             c.sync(SEC1).unwrap();
         });
     })
     .unwrap();
+}
 #[test]
 fn native_immediately_inconsistent() {
     let test_log_path = Path::new("./logs/native_immediately_inconsistent");
     let mut c = file_logged_connector(0, test_log_path);
     let [_, g] = c.new_port_pair();
     c.connect(SEC1).unwrap();
     c.get(g).unwrap();
     c.sync(SEC15).unwrap_err();
+}
 #[test]
 fn native_recovers() {
     let test_log_path = Path::new("./logs/native_recovers");
     let mut c = file_logged_connector(0, test_log_path);
     let [p, g] = c.new_port_pair();
     c.connect(SEC1).unwrap();
     c.get(g).unwrap();
     c.sync(SEC15).unwrap_err();
     c.put(p, TEST_MSG.clone()).unwrap();
     c.get(g).unwrap();
     c.sync(SEC15).unwrap();
+}
 #[test]
 fn cannot_use_moved_ports() {
     /*
     native p|-->|g sync
     */
     let test_log_path = Path::new("./logs/cannot_use_moved_ports");
     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(SEC1).unwrap();
     c.put(p, TEST_MSG.clone()).unwrap_err();
     c.get(g).unwrap_err();
+}
 #[test]
 fn sync_sync() {
     /*
     native p0|-->|g0 sync
            g1|<--|p1
     */
     let test_log_path = Path::new("./logs/sync_sync");
     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(SEC1).unwrap();
     c.put(p0, TEST_MSG.clone()).unwrap();
     c.get(g1).unwrap();
     c.sync(SEC1).unwrap();
     c.gotten(g1).unwrap();
+}
 #[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, test_log_path);
             let [_p, _g] = [
                 c.new_net_port(Putter, sock_addrs[0], Active).unwrap(),
                 c.new_net_port(Getter, sock_addrs[1], Active).unwrap(),
             ];
             c.connect(SEC1).unwrap();
         });
         s.spawn(|_| {
             let mut c = file_logged_connector(1, test_log_path);
             let [_g, _p] = [
                 c.new_net_port(Getter, sock_addrs[0], Passive).unwrap(),
                 c.new_net_port(Putter, sock_addrs[1], Passive).unwrap(),
             ];
             c.connect(SEC1).unwrap();
         });
     })
     .unwrap();
+}
 #[test]
 fn distributed_msg_bounce() {
     /*
     native[0] | sync 0.p|-->|1.p native[1]
 .g|<--|1.g
     */
     let test_log_path = Path::new("./logs/distributed_msg_bounce");
     let sock_addrs = [next_test_addr(), next_test_addr()];
     scope(|s| {
         s.spawn(|_| {
             /*
             native | sync p|-->
                    |      g|<--
             */
             let mut c = file_logged_connector(0, test_log_path);
             let [p, g] = [
                 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(SEC1).unwrap();
             c.sync(SEC1).unwrap();
         });
         s.spawn(|_| {
             /*
             native p|-->
                    g|<--
             */
             let mut c = file_logged_connector(1, test_log_path);
             let [g, p] = [
                 c.new_net_port(Getter, sock_addrs[0], Passive).unwrap(),
                 c.new_net_port(Putter, sock_addrs[1], Passive).unwrap(),
             ];
             c.connect(SEC1).unwrap();
             c.put(p, TEST_MSG.clone()).unwrap();
             c.get(g).unwrap();
             c.sync(SEC1).unwrap();
             c.gotten(g).unwrap();
         });
     })
     .unwrap();
+}
 #[test]
 fn local_timeout() {
     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(SEC1).unwrap();
     c.get(g).unwrap();
     match c.sync(MS300) {
         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_addrs = [next_test_addr()];
     scope(|s| {
         s.spawn(|_| {
             // parent; times out
             let mut c = file_logged_connector(999, test_log_path);
             let _ = c.new_net_port(Putter, sock_addrs[0], Active).unwrap();
             c.connect(SEC1).unwrap();
             c.sync(MS300).unwrap_err(); // timeout
         });
         s.spawn(|_| {
             // child
             let mut c = file_logged_connector(000, test_log_path);
             let g = c.new_net_port(Getter, sock_addrs[0], Passive).unwrap();
             c.connect(SEC1).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_addrs = [next_test_addr()];
     scope(|s| {
         s.spawn(|_| {
             // child; times out
             let mut c = file_logged_connector(000, test_log_path);
             let _ = c.new_net_port(Putter, sock_addrs[0], Active).unwrap();
             c.connect(SEC1).unwrap();
             c.sync(MS300).unwrap_err(); // timeout
         });
         s.spawn(|_| {
             // parent
             let mut c = file_logged_connector(999, test_log_path);
             let g = c.new_net_port(Getter, sock_addrs[0], Passive).unwrap();
             c.connect(SEC1).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, test_log_path);
             c.new_net_port(Putter, sock_addrs[0], Passive).unwrap();
             c.connect(SEC5).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(SEC5).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(SEC5).unwrap();
         });
         s.spawn(|_| {
             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(SEC5).unwrap();
         });
         s.spawn(|_| {
             let mut c = file_logged_connector(1, test_log_path);
             c.new_net_port(Getter, sock_addrs[3], Active).unwrap();
             c.connect(SEC5).unwrap();
         });
     })
     .unwrap();
+}
 #[test]
 fn net_self_loop() {
     let test_log_path = Path::new("./logs/net_self_loop");
     let sock_addrs = [next_test_addr()];
     let mut c = file_logged_connector(0, test_log_path);
     let p = c.new_net_port(Putter, sock_addrs[0], Active).unwrap();
     let g = c.new_net_port(Getter, sock_addrs[0], Passive).unwrap();
     c.connect(SEC1).unwrap();
     c.put(p, TEST_MSG.clone()).unwrap();
     c.get(g).unwrap();
     c.sync(MS300).unwrap();
+}
 #[test]
 fn nobody_connects_active() {
     let test_log_path = Path::new("./logs/nobody_connects_active");
     let sock_addrs = [next_test_addr()];
     let mut c = file_logged_connector(0, test_log_path);
     let _g = c.new_net_port(Getter, sock_addrs[0], Active).unwrap();
     c.connect(Some(Duration::from_secs(5))).unwrap_err();
+}
 #[test]
 fn nobody_connects_passive() {
     let test_log_path = Path::new("./logs/nobody_connects_passive");
     let sock_addrs = [next_test_addr()];
     let mut c = file_logged_connector(0, test_log_path);
     let _g = c.new_net_port(Getter, sock_addrs[0], Passive).unwrap();
     c.connect(Some(Duration::from_secs(5))).unwrap_err();
+}
 #[test]
 fn together() {
     let test_log_path = Path::new("./logs/together");
     let sock_addrs = [next_test_addr(), next_test_addr()];
     scope(|s| {
         s.spawn(|_| {
             let mut c = file_logged_connector(0, test_log_path);
             let [p0, p1] = c.new_port_pair();
             let p2 = c.new_net_port(Getter, sock_addrs[0], Passive).unwrap();
             let p3 = c.new_net_port(Putter, sock_addrs[1], Active).unwrap();
             let [p4, p5] = c.new_port_pair();
             c.add_component(b"together", &[p1, p2, p3, p4]).unwrap();
             c.connect(SEC1).unwrap();
             c.put(p0, TEST_MSG.clone()).unwrap();
             c.get(p5).unwrap();
             c.sync(MS300).unwrap();
             c.gotten(p5).unwrap();
         });
         s.spawn(|_| {
             let mut c = file_logged_connector(1, test_log_path);
             let [p0, p1] = c.new_port_pair();
             let p2 = c.new_net_port(Getter, sock_addrs[1], Passive).unwrap();
             let p3 = c.new_net_port(Putter, sock_addrs[0], Active).unwrap();
             let [p4, p5] = c.new_port_pair();
             c.add_component(b"together", &[p1, p2, p3, p4]).unwrap();
             c.connect(SEC1).unwrap();
             c.put(p0, TEST_MSG.clone()).unwrap();
             c.get(p5).unwrap();
             c.sync(MS300).unwrap();
             c.gotten(p5).unwrap();
         });
     })
     .unwrap();
+}
 #[test]
 fn native_batch_distinguish() {
     let test_log_path = Path::new("./logs/native_batch_distinguish");
     let mut c = file_logged_connector(0, test_log_path);
     c.connect(SEC1).unwrap();
     c.next_batch().unwrap();
     c.sync(SEC1).unwrap();
+}
 #[test]
 fn multirounds() {
     let test_log_path = Path::new("./logs/multirounds");
     let sock_addrs = [next_test_addr(), next_test_addr()];
     scope(|s| {
         s.spawn(|_| {
             let mut c = file_logged_connector(0, test_log_path);
             let p0 = c.new_net_port(Putter, sock_addrs[0], Active).unwrap();
             let p1 = c.new_net_port(Getter, sock_addrs[1], Passive).unwrap();
             c.connect(SEC1).unwrap();
             for _ in 0..10 {
                 c.put(p0, TEST_MSG.clone()).unwrap();
                 c.get(p1).unwrap();
                 c.sync(SEC1).unwrap();
+            }
         });
         s.spawn(|_| {
             let mut c = file_logged_connector(1, test_log_path);
             let p0 = c.new_net_port(Getter, sock_addrs[0], Passive).unwrap();
             let p1 = c.new_net_port(Putter, sock_addrs[1], Active).unwrap();
             c.connect(SEC1).unwrap();
             for _ in 0..10 {
                 c.get(p0).unwrap();
                 c.put(p1, TEST_MSG.clone()).unwrap();
                 c.sync(SEC1).unwrap();
+            }
         });
     })
     .unwrap();
+}
 #[test]
 fn multi_recover() {
     let test_log_path = Path::new("./logs/multi_recover");
     let sock_addrs = [next_test_addr(), next_test_addr()];
     let success_iter = [true, false].iter().copied().cycle().take(10);
     scope(|s| {
         s.spawn(|_| {
             let mut c = file_logged_connector(0, test_log_path);
             let p0 = c.new_net_port(Putter, sock_addrs[0], Active).unwrap();
             let p1 = c.new_net_port(Getter, sock_addrs[1], Passive).unwrap();
             c.connect(SEC1).unwrap();
             for succeeds in success_iter.clone() {
                 c.put(p0, TEST_MSG.clone()).unwrap();
                 if succeeds {
                     c.get(p1).unwrap();
+                }
                 let res = c.sync(MS300);
                 assert_eq!(res.is_ok(), succeeds);
+            }
         });
         s.spawn(|_| {
             let mut c = file_logged_connector(1, test_log_path);
             let p0 = c.new_net_port(Getter, sock_addrs[0], Passive).unwrap();
             let p1 = c.new_net_port(Putter, sock_addrs[1], Active).unwrap();
             c.connect(SEC1).unwrap();
             for succeeds in success_iter.clone() {
                 c.get(p0).unwrap();
                 c.put(p1, TEST_MSG.clone()).unwrap();
                 let res = c.sync(MS300);
                 assert_eq!(res.is_ok(), succeeds);
+            }
         });
     })
     .unwrap();
+}
 #[test]
 fn udp_self_connect() {
     let test_log_path = Path::new("./logs/udp_self_connect");
     let sock_addrs = [next_test_addr(), next_test_addr()];
     let mut c = file_logged_connector(0, test_log_path);
     c.new_udp_port(Putter, sock_addrs[0], sock_addrs[1]).unwrap();
     c.new_udp_port(Getter, sock_addrs[1], sock_addrs[0]).unwrap();
     c.new_udp_port(sock_addrs[0], sock_addrs[1]).unwrap();
     c.new_udp_port(sock_addrs[1], sock_addrs[0]).unwrap();
     c.connect(SEC1).unwrap();
+}
 #[test]
 fn solo_udp_put_success() {
     let test_log_path = Path::new("./logs/solo_udp_put_success");
     let sock_addrs = [next_test_addr(), next_test_addr()];
     let mut c = file_logged_connector(0, test_log_path);
-    let p0 = c.new_udp_port(Putter, sock_addrs[0], sock_addrs[1]).unwrap();
+    let [p0, _] = c.new_udp_port(sock_addrs[0], sock_addrs[1]).unwrap();
     c.connect(SEC1).unwrap();
     c.put(p0, TEST_MSG.clone()).unwrap();
     c.sync(MS300).unwrap();
+}
 #[test]
 fn solo_udp_get_fail() {
     let test_log_path = Path::new("./logs/solo_udp_get_fail");
     let sock_addrs = [next_test_addr(), next_test_addr()];
     let mut c = file_logged_connector(0, test_log_path);
-    let p0 = c.new_udp_port(Getter, sock_addrs[0], sock_addrs[1]).unwrap();
+    let [_, p0] = c.new_udp_port(sock_addrs[0], sock_addrs[1]).unwrap();
     c.connect(SEC1).unwrap();
     c.get(p0).unwrap();
     c.sync(MS300).unwrap_err();
+}
 #[test]
 fn reowolf_to_udp() {
     let test_log_path = Path::new("./logs/reowolf_to_udp");
     let sock_addrs = [next_test_addr(), next_test_addr()];
     let barrier = std::sync::Barrier::new(2);
     scope(|s| {
         s.spawn(|_| {
             barrier.wait();
             // reowolf thread
             let mut c = file_logged_connector(0, test_log_path);
-            let p0 = c.new_udp_port(Putter, sock_addrs[0], sock_addrs[1]).unwrap();
+            let [p0, _] = c.new_udp_port(sock_addrs[0], sock_addrs[1]).unwrap();
             c.connect(SEC1).unwrap();
             c.put(p0, TEST_MSG.clone()).unwrap();
             c.sync(MS300).unwrap();
             barrier.wait();
         });
         s.spawn(|_| {
             barrier.wait();
             // udp thread
             let udp = std::net::UdpSocket::bind(sock_addrs[1]).unwrap();
             udp.connect(sock_addrs[0]).unwrap();
             let mut buf = unsafe {
                 // canonical way to create uninitalized byte buffer
                 let mut v = Vec::with_capacity(256);
                 v.set_len(256);
+                v
             };
             let len = udp.recv(&mut buf).unwrap();
             assert_eq!(TEST_MSG_BYTES, &buf[0..len]);
             barrier.wait();
         });
     })
     .unwrap();
+}
 #[test]
 fn udp_to_reowolf() {
     let test_log_path = Path::new("./logs/udp_to_reowolf");
     let sock_addrs = [next_test_addr(), next_test_addr()];
     let barrier = std::sync::Barrier::new(2);
     scope(|s| {
         s.spawn(|_| {
             barrier.wait();
             // reowolf thread
             let mut c = file_logged_connector(0, test_log_path);
-            let p0 = c.new_udp_port(Getter, sock_addrs[0], sock_addrs[1]).unwrap();
+            let [_, p0] = c.new_udp_port(sock_addrs[0], sock_addrs[1]).unwrap();
             c.connect(SEC1).unwrap();
             c.get(p0).unwrap();
-            c.sync(SEC1).unwrap();
+            c.sync(SEC5).unwrap();
             assert_eq!(c.gotten(p0).unwrap().as_slice(), TEST_MSG_BYTES);
             barrier.wait();
         });
         s.spawn(|_| {
             barrier.wait();
             // udp thread
             let udp = std::net::UdpSocket::bind(sock_addrs[1]).unwrap();
             udp.connect(sock_addrs[0]).unwrap();
-            for _ in 0..5 {
+            for _ in 0..15 {
                 udp.send(TEST_MSG_BYTES).unwrap();
                 std::thread::sleep(MS100.unwrap());
+            }
             barrier.wait();
         });
     })
     .unwrap();
+}
 #[test]
 fn udp_reowolf_swap() {
     let test_log_path = Path::new("./logs/udp_reowolf_swap");
     let sock_addrs = [next_test_addr(), next_test_addr()];
     let barrier = std::sync::Barrier::new(2);
     scope(|s| {
         s.spawn(|_| {
             barrier.wait();
             // reowolf thread
             let mut c = file_logged_connector(0, test_log_path);
             let [p0, p1] = c.new_udp_port(sock_addrs[0], sock_addrs[1]).unwrap();
             c.connect(SEC1).unwrap();
             c.put(p0, TEST_MSG.clone()).unwrap();
             c.get(p1).unwrap();
             c.sync(SEC5).unwrap();
             assert_eq!(c.gotten(p1).unwrap().as_slice(), TEST_MSG_BYTES);
             barrier.wait();
         });
         s.spawn(|_| {
             barrier.wait();
             // udp thread
             let udp = std::net::UdpSocket::bind(sock_addrs[1]).unwrap();
             udp.connect(sock_addrs[0]).unwrap();
             let mut buf = unsafe {
                 // canonical way to create uninitalized byte buffer
                 let mut v = Vec::with_capacity(256);
                 v.set_len(256);
+                v
             };
             udp.send(TEST_MSG_BYTES).unwrap();
             let len = udp.recv(&mut buf).unwrap();
             assert_eq!(TEST_MSG_BYTES, &buf[0..len]);
             barrier.wait();
         });
     })
     .unwrap();
+}

0 comments (0 inline, 0 general)