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

Changeset - db17da820a3b

Parent rev.

Child rev.

[Not reviewed]

0 11 0

Christopher Esterhuyse - 5 years ago 2020-07-01 08:41:15
christopher.esterhuyse@gmail.com

pruned dependencies. cleaned up visibility

11 files changed with 68 insertions and 75 deletions:

Cargo.toml

cbindgen.toml

examples/8_net_ports/amy.c

examples/9_net_self_putget/amy.c

examples/a_swap/amy.c

examples/a_swap/bob.c

reowolf.h

src/common.rs

src/runtime/error.rs

src/runtime/mod.rs

src/runtime/setup.rs

0 comments (0 inline, 0 general)

Cargo.toml

➞

Show inline comments

 [package]
 name = "reowolf_rs"
 version = "0.1.4"
 authors = [
 	"Christopher Esterhuyse <esterhuy@cwi.nl, christopher.esterhuyse@gmail.com>",
 	"Hans-Dieter Hiep <hdh@cwi.nl>"
+]
 edition = "2018"
 [dependencies]
 # convenience macros
 maplit = "1.0.2"
 derive_more = "0.99.2"
 # runtime
 bincode = "1.2.1"
 serde = { version = "1.0.114", features = ["derive"] }
 getrandom = "0.1.14" # tiny crate. used to guess controller-id
 # take_mut = "0.2.2"
 indexmap = "1.4.0" # hashsets/hashmaps with efficient arbitrary element removal
 # network
 # integer-encoding = "1.1.5"
 # byteorder = "1.3.4"
 mio = { version = "0.7.0", package = "mio", features = ["tcp", "os-poll"] }
 # protocol
 backtrace = "0.3"
 [dev-dependencies]
 # test-generator = "0.3.0"
 crossbeam-utils = "0.7.2"
 lazy_static = "1.4.0"
 [lib]
 # compile target: dynamically linked library using C ABI
 crate-type = ["cdylib"]
 [features]
 default = ["ffi"]
 ffi = [] # no feature dependencies
@@ \ No newline at end of file @@

cbindgen.toml

➞

Show inline comments

 language = "C"
 header = "/* CBindgen generated */"
 include_guard = "REOWOLF_HEADER_DEFINED"
 [parse]
 parse_deps = false
@@ \ No newline at end of file @@
 parse_deps = false
 [enum]
 prefix_with_name = true
@@ \ No newline at end of file @@

examples/8_net_ports/amy.c

➞

Show inline comments

 #include <stdio.h>
 #include <string.h>
 #include "../../reowolf.h"
 #include "../utility.c"
 int main(int argc, char** argv) {
 	Arc_ProtocolDescription * pd = protocol_description_parse("", 0);
 	char logpath[] = "./8_amy_log.txt";
 	Connector * c = connector_new_logging(pd, logpath, sizeof(logpath)-1);
 	printf("Error str `%s`\n", reowolf_error_peek(NULL));
 	PortId putter, getter;
 	char addr_str[] = "127.0.0.1:8000";
 	connector_add_net_port(
-		c, &putter, addr_str, sizeof(addr_str)-1, Putter, Active);
+		c, &putter, addr_str, sizeof(addr_str)-1, Polarity_Putter, EndpointPolarity_Active);
 	printf("Error str `%s`\n", reowolf_error_peek(NULL));
 	connector_add_net_port(
-		c, &getter, addr_str, sizeof(addr_str)-1, Getter, Passive);
+		c, &getter, addr_str, sizeof(addr_str)-1, Polarity_Getter, EndpointPolarity_Passive);
 	printf("Error str `%s`\n", reowolf_error_peek(NULL));
 	connector_connect(c, 4000);
 	printf("Error str `%s`\n", reowolf_error_peek(NULL));
 	protocol_description_destroy(pd);
 	connector_destroy(c);
 	return 0;
+}
@@ \ No newline at end of file @@

examples/9_net_self_putget/amy.c

➞

Show inline comments

 #include <stdio.h>
 #include <string.h>
 #include "../../reowolf.h"
 #include "../utility.c"
 int main(int argc, char** argv) {
 	Arc_ProtocolDescription * pd = protocol_description_parse("", 0);
 	char logpath[] = "./9_amy_log.txt";
 	Connector * c = connector_new_logging(pd, logpath, sizeof(logpath)-1);
 	printf("Error str `%s`\n", reowolf_error_peek(NULL));
 	PortId putter, getter;
 	char addr_str[] = "127.0.0.1:8000";
 	connector_add_net_port(
-		c, &putter, addr_str, sizeof(addr_str)-1, Putter, Active);
+		c, &putter, addr_str, sizeof(addr_str)-1, Polarity_Putter, EndpointPolarity_Active);
 	printf("Error str `%s`\n", reowolf_error_peek(NULL));
 	connector_add_net_port(
-		c, &getter, addr_str, sizeof(addr_str)-1, Getter, Passive);
+		c, &getter, addr_str, sizeof(addr_str)-1, Polarity_Getter, EndpointPolarity_Passive);
 	printf("Error str `%s`\n", reowolf_error_peek(NULL));
 	connector_connect(c, 4000);
 	printf("Error str `%s`\n", reowolf_error_peek(NULL));
 	connector_put_bytes(c, putter, "hi", 2);
 	printf("Error str `%s`\n", reowolf_error_peek(NULL));
 	connector_get(c, getter);
 	printf("Error str `%s`\n", reowolf_error_peek(NULL));
 	connector_sync(c, 4000);
 	printf("Error str `%s`\n", reowolf_error_peek(NULL));
 	protocol_description_destroy(pd);
 	connector_destroy(c);
 	return 0;
+}
@@ \ No newline at end of file @@

examples/a_swap/amy.c

➞

Show inline comments

 #include <stdio.h>
 #include <string.h>
 #include "../../reowolf.h"
 #include "../utility.c"
 int main(int argc, char** argv) {
     if(argc != 3) {
         printf("Expected arg[1] and arg[2] for use as addr str\n");
         exit(1);
+    }
     char * pdl_ptr = buffer_pdl("eg_protocols.pdl");
     size_t pdl_len = strlen(pdl_ptr);
     Arc_ProtocolDescription * pd = protocol_description_parse(pdl_ptr, pdl_len);
     char logpath[] = "./a_amy_log.txt";
     Connector * c = connector_new_logging(pd, logpath, sizeof(logpath)-1);
     printf("Error str `%s`\n", reowolf_error_peek(NULL));
     PortId ports[6];
     connector_add_port_pair(c, &ports[0], &ports[1]);
     printf("Error str `%s`\n", reowolf_error_peek(NULL));
-    connector_add_net_port(c, &ports[2], argv[1], strlen(argv[1]), Getter, Passive);
+    connector_add_net_port(c, &ports[2], argv[1], strlen(argv[1]), Polarity_Getter, EndpointPolarity_Passive);
     printf("Error str `%s`\n", reowolf_error_peek(NULL));
-    connector_add_net_port(c, &ports[3], argv[2], strlen(argv[2]), Putter, Active);
+    connector_add_net_port(c, &ports[3], argv[2], strlen(argv[2]), Polarity_Putter, EndpointPolarity_Active);
     printf("Error str `%s`\n", reowolf_error_peek(NULL));
     connector_add_port_pair(c, &ports[4], &ports[5]);
     printf("Error str `%s`\n", reowolf_error_peek(NULL));
     // native {0,1,2,3,4,5}
     connector_add_component(c, "together", 8, &ports[1], 4);
     printf("Error str `%s`\n", reowolf_error_peek(NULL));
     // native {0,5} together {1,2,3,4}
     connector_connect(c, 4000);
     printf("Error str `%s`\n", reowolf_error_peek(NULL));
 	connector_put_bytes(c, ports[0], "hi", 2);
     printf("Error str `%s`\n", reowolf_error_peek(NULL));
 	connector_get(c, ports[5]);
     printf("Error str `%s`\n", reowolf_error_peek(NULL));
     connector_sync(c, 1000);
     printf("Error str `%s`\n", reowolf_error_peek(NULL));
 	size_t msg_len;
 	const char * msg_ptr = connector_gotten_bytes(c, ports[5], &msg_len);
     printf("Error str `%s`\n", reowolf_error_peek(NULL));
 	printf("Got msg `%.*s`\n", msg_len, msg_ptr);
     protocol_description_destroy(pd);
     connector_destroy(c);
     return 0;
+}
@@ \ No newline at end of file @@

examples/a_swap/bob.c

➞

Show inline comments

 #include <stdio.h>
 #include <string.h>
 #include "../../reowolf.h"
 #include "../utility.c"
 int main(int argc, char** argv) {
     if(argc != 3) {
         printf("Expected arg[1] and arg[2] for use as addr str\n");
         exit(1);
+    }
     char * pdl_ptr = buffer_pdl("eg_protocols.pdl");
     size_t pdl_len = strlen(pdl_ptr);
     Arc_ProtocolDescription * pd = protocol_description_parse(pdl_ptr, pdl_len);
     char logpath[] = "./a_bob_log.txt";
     Connector * c = connector_new_logging(pd, logpath, sizeof(logpath)-1);
     printf("Error str `%s`\n", reowolf_error_peek(NULL));
     PortId ports[6];
     connector_add_port_pair(c, &ports[0], &ports[1]);
     printf("Error str `%s`\n", reowolf_error_peek(NULL));
-    connector_add_net_port(c, &ports[2], argv[1], strlen(argv[1]), Getter, Passive);
+    connector_add_net_port(c, &ports[2], argv[1], strlen(argv[1]), Polarity_Getter, EndpointPolarity_Passive);
     printf("Error str `%s`\n", reowolf_error_peek(NULL));
-    connector_add_net_port(c, &ports[3], argv[2], strlen(argv[2]), Putter, Active);
+    connector_add_net_port(c, &ports[3], argv[2], strlen(argv[2]), Polarity_Putter, EndpointPolarity_Active);
     printf("Error str `%s`\n", reowolf_error_peek(NULL));
     connector_add_port_pair(c, &ports[4], &ports[5]);
     printf("Error str `%s`\n", reowolf_error_peek(NULL));
     // native {0,1,2,3,4,5}
     connector_add_component(c, "together", 8, &ports[1], 4);
     printf("Error str `%s`\n", reowolf_error_peek(NULL));
     // native {0,5} together {1,2,3,4}
     connector_connect(c, 4000);
     printf("Error str `%s`\n", reowolf_error_peek(NULL));
 	connector_put_bytes(c, ports[0], "hi", 2);
     printf("Error str `%s`\n", reowolf_error_peek(NULL));
 	connector_get(c, ports[5]);
     printf("Error str `%s`\n", reowolf_error_peek(NULL));
     connector_sync(c, 1000);
     printf("Error str `%s`\n", reowolf_error_peek(NULL));
 	size_t msg_len;
 	const char * msg_ptr = connector_gotten_bytes(c, ports[5], &msg_len);
     printf("Error str `%s`\n", reowolf_error_peek(NULL));
 	printf("Got msg `%.*s`\n", msg_len, msg_ptr);
     protocol_description_destroy(pd);
     connector_destroy(c);
     return 0;
+}
@@ \ No newline at end of file @@

reowolf.h

➞

Show inline comments

 /* CBindgen generated */
 #ifndef REOWOLF_HEADER_DEFINED
 #define REOWOLF_HEADER_DEFINED
 #include <stdarg.h>
 #include <stdbool.h>
 #include <stdint.h>
 #include <stdlib.h>
 typedef enum {
   Active,
   Passive,
   EndpointPolarity_Active,
   EndpointPolarity_Passive,
 } EndpointPolarity;
 typedef enum {
   Putter,
   Getter,
   Polarity_Putter,
   Polarity_Getter,
 } Polarity;
 typedef struct Arc_ProtocolDescription Arc_ProtocolDescription;
 typedef struct Connector Connector;
 typedef int32_t ErrorCode;
 typedef uint32_t ConnectorId;
 typedef uint32_t PortSuffix;
 typedef struct {
   ConnectorId connector_id;
   PortSuffix u32_suffix;
 } PortId;
 /**
  * Given
  * - an initialized connector in setup or connecting state,
  * - a string slice for the component's identifier in the connector's configured protocol description,
  * - a set of ports (represented as a slice; duplicates are ignored) in the native component's interface,
  * the connector creates a new (internal) protocol component C, such that the set of native ports are moved to C.
  * Usable in {setup, communication} states.
  */
 ErrorCode connector_add_component(Connector *connector,
                                   const uint8_t *ident_ptr,
                                   uintptr_t ident_len,
                                   const PortId *ports_ptr,
                                   uintptr_t ports_len);
 /**
  * Given
  * - an initialized connector in setup or connecting state,
  * - a utf-8 encoded socket address,
  * - the logical polarity of P,
  * - the "physical" polarity in {Active, Passive} of the endpoint through which P's peer will be discovered,
  * returns P, a port newly added to the native interface.
  */
 ErrorCode connector_add_net_port(Connector *connector,
                                  PortId *port,
                                  const uint8_t *addr_str_ptr,
                                  uintptr_t addr_str_len,
                                  Polarity port_polarity,
                                  EndpointPolarity endpoint_polarity);
 /**
  * Given an initialized connector in setup or connecting state,
  * - Creates a new directed port pair with logical channel putter->getter,
  * - adds the ports to the native component's interface,
  * - and returns them using the given out pointers.
  * Usable in {setup, communication} states.
  */
 void connector_add_port_pair(Connector *connector, PortId *out_putter, PortId *out_getter);
 /**
  * Connects this connector to the distributed system of connectors reachable through endpoints,
  * Usable in setup state, and changes the state to communication.
  */
 ErrorCode connector_connect(Connector *connector, int64_t timeout_millis);
 /**
  * Destroys the given a pointer to the connector on the heap, freeing its resources.
  * Usable in {setup, communication} states.
  */
 void connector_destroy(Connector *connector);
 ErrorCode connector_get(Connector *connector, PortId port);
 const uint8_t *connector_gotten_bytes(Connector *connector, PortId port, uintptr_t *out_len);
 /**
  * Initializes `out` with a new connector using the given protocol description as its configuration.
  * The connector uses the given (internal) connector ID.
  */
 Connector *connector_new(const Arc_ProtocolDescription *pd);
 Connector *connector_new_logging(const Arc_ProtocolDescription *pd,
                                  const uint8_t *path_ptr,
                                  uintptr_t path_len);
 intptr_t connector_next_batch(Connector *connector);
 void connector_print_debug(Connector *connector);
 /**
  * Convenience function combining the functionalities of
  * "payload_new" with "connector_put_payload".
  */
 ErrorCode connector_put_bytes(Connector *connector,
                               PortId port,
                               const uint8_t *bytes_ptr,
                               uintptr_t bytes_len);
 intptr_t connector_sync(Connector *connector, int64_t timeout_millis);
 /**
  * Given an initialized protocol description, initializes `out` with a clone which can be independently created or destroyed.
  */
 Arc_ProtocolDescription *protocol_description_clone(const Arc_ProtocolDescription *pd);
 /**
  * Destroys the given initialized protocol description and frees its resources.
  */
 void protocol_description_destroy(Arc_ProtocolDescription *pd);
 /**
  * Parses the utf8-encoded string slice to initialize a new protocol description object.
  * - On success, initializes `out` and returns 0
  * - On failure, stores an error string (see `reowolf_error_peek`) and returns -1
  */
 Arc_ProtocolDescription *protocol_description_parse(const uint8_t *pdl, uintptr_t pdl_len);
 /**
  * Returns length (via out pointer) and pointer (via return value) of the last Reowolf error.
  * - pointer is NULL iff there was no last error
  * - data at pointer is null-delimited
  * - len does NOT include the length of the null-delimiter
  * If len is NULL, it will not written to.
  */
 const uint8_t *reowolf_error_peek(uintptr_t *len);
 #endif /* REOWOLF_HEADER_DEFINED */

src/common.rs

➞

Show inline comments

 ///////////////////// PRELUDE /////////////////////
 pub(crate) use crate::protocol::{ComponentState, ProtocolDescription};
 pub(crate) use crate::runtime::{NonsyncProtoContext, SyncProtoContext};
+pub(crate) use crate::runtime::{error::AddComponentError, NonsyncProtoContext, SyncProtoContext};
 pub use core::{
+pub(crate) use core::{
     cmp::Ordering,
     fmt::{Debug, Formatter},
     hash::{Hash, Hasher},
     ops::{Range, RangeFrom},
     hash::Hash,
     ops::Range,
     time::Duration,
 };
 pub use indexmap::{IndexMap, IndexSet};
 pub use maplit::{hashmap, hashset};
 pub use mio::{
 // pub(crate) use indexmap::IndexSet;
 pub(crate) use maplit::hashmap;
 pub(crate) use mio::{
     net::{TcpListener, TcpStream},
     Events, Interest, Poll, Token,
 };
 pub use std::{
     collections::{hash_map::Entry, BTreeMap, HashMap, HashSet},
 pub(crate) use std::{
     collections::{BTreeMap, HashMap, HashSet},
     convert::TryInto,
     io::{Read, Write},
     net::SocketAddr,
     sync::Arc,
     time::Instant,
 };
 pub use Polarity::*;
 ///////////////////// DEFS /////////////////////
 pub(crate) use Polarity::*;
 pub type ConnectorId = u32;
 pub type PortSuffix = u32;
 #[derive(
     Copy, Clone, Eq, PartialEq, Ord, Hash, PartialOrd, serde::Serialize, serde::Deserialize,
 )]
 // acquired via error in the Rust API
 pub struct ProtoComponentId(Id);
 #[derive(
     Copy, Clone, Eq, PartialEq, Ord, Hash, PartialOrd, serde::Serialize, serde::Deserialize,
 )]
 #[repr(C)]
 pub struct Id {
     pub(crate) connector_id: ConnectorId,
     pub(crate) u32_suffix: PortSuffix,
+}
 #[derive(Debug, Default)]
 pub struct U32Stream {
     next: u32,
+}
 // globally unique
 #[derive(
     Copy, Clone, Eq, PartialEq, Ord, Hash, PartialOrd, serde::Serialize, serde::Deserialize,
 )]
 #[repr(transparent)]
 pub struct PortId(Id);
 #[derive(
     Copy, Clone, Eq, PartialEq, Ord, Hash, PartialOrd, serde::Serialize, serde::Deserialize,
 )]
 pub struct FiringVar(pub(crate) PortId);
 #[derive(
     Copy, Clone, Eq, PartialEq, Ord, Hash, PartialOrd, serde::Serialize, serde::Deserialize,
 )]
 pub struct ProtoComponentId(Id);
 #[derive(Debug, Clone, Eq, PartialEq, Ord, PartialOrd)]
 // #[repr(C)]
 #[derive(Default, Debug, Clone, Eq, PartialEq, Ord, PartialOrd)]
 pub struct Payload(Arc<Vec<u8>>);
 #[derive(
     Debug, Eq, PartialEq, Clone, Hash, Copy, Ord, PartialOrd, serde::Serialize, serde::Deserialize,
 )]
 #[repr(C)]
 pub enum Polarity {
     Putter, // output port (from the perspective of the component)
     Getter, // input port (from the perspective of the component)
+}
 #[derive(
     Debug, Eq, PartialEq, Clone, Hash, Copy, Ord, PartialOrd, serde::Serialize, serde::Deserialize,
 )]
 #[repr(C)]
 pub enum EndpointPolarity {
     Active,  // calls connect()
     Passive, // calls bind() listen() accept()
+}
 #[derive(Eq, PartialEq, Copy, Clone, Debug)]
 pub enum AddComponentError {
     NoSuchComponent,
     NonPortTypeParameters,
     CannotMovePort(PortId),
     WrongNumberOfParamaters { expected: usize },
     UnknownPort(PortId),
     WrongPortPolarity { port: PortId, expected_polarity: Polarity },
     DuplicateMovedPort(PortId),
+}
 #[derive(
     Copy, Clone, Eq, PartialEq, Ord, Hash, PartialOrd, serde::Serialize, serde::Deserialize,
 )]
 pub(crate) struct FiringVar(pub(crate) PortId);
 #[derive(Debug, Clone)]
 pub enum NonsyncBlocker {
+pub(crate) enum NonsyncBlocker {
     Inconsistent,
     ComponentExit,
     SyncBlockStart,
+}
 #[derive(Debug, Clone)]
 pub enum SyncBlocker {
+pub(crate) enum SyncBlocker {
     Inconsistent,
     SyncBlockEnd,
     CouldntReadMsg(PortId),
     CouldntCheckFiring(PortId),
     PutMsg(PortId, Payload),
+}
 ///////////////////// IMPL /////////////////////
 impl U32Stream {
     pub fn next(&mut self) -> u32 {
+    pub(crate) fn next(&mut self) -> u32 {
         if self.next == u32::MAX {
             panic!("NO NEXT!")
+        }
         self.next += 1;
         self.next - 1
+    }
+}
 impl From<Id> for PortId {
     fn from(id: Id) -> PortId {
         Self(id)
+    }
+}
 impl From<Id> for ProtoComponentId {
     fn from(id: Id) -> ProtoComponentId {
         Self(id)
+    }
+}
 impl From<&[u8]> for Payload {
     fn from(s: &[u8]) -> Payload {
         Payload(Arc::new(s.to_vec()))
+    }
+}
 impl Payload {
     pub fn new(len: usize) -> Payload {
         let mut v = Vec::with_capacity(len);
         unsafe {
             v.set_len(len);
+        }
         Payload(Arc::new(v))
+    }
     pub fn len(&self) -> usize {
         self.0.len()
+    }
     pub fn as_slice(&self) -> &[u8] {
         &self.0
+    }
     pub fn as_mut_slice(&mut self) -> &mut [u8] {
         Arc::make_mut(&mut self.0) as _
+    }
-    pub fn concat_with(&mut self, other: &Self) {
+    pub fn concatenate_with(&mut self, other: &Self) {
         let bytes = other.as_slice().iter().copied();
         let me = Arc::make_mut(&mut self.0);
         me.extend(bytes);
+    }
+}
 impl serde::Serialize for Payload {
     fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
     where
         S: serde::Serializer,
+    {
         let inner: &Vec<u8> = &self.0;
         inner.serialize(serializer)
+    }
+}
 impl<'de> serde::Deserialize<'de> for Payload {
     fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
     where
         D: serde::Deserializer<'de>,
+    {
         let inner: Vec<u8> = Vec::deserialize(deserializer)?;
         Ok(Self(Arc::new(inner)))
+    }
+}
 impl std::iter::FromIterator<u8> for Payload {
     fn from_iter<I: IntoIterator<Item = u8>>(it: I) -> Self {
         Self(Arc::new(it.into_iter().collect()))
+    }
+}
 impl From<Vec<u8>> for Payload {
     fn from(s: Vec<u8>) -> Self {
         Self(s.into())
+    }
+}
 impl Debug for PortId {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         write!(f, "ptID({}'{})", self.0.connector_id, self.0.u32_suffix)
+    }
+}
 impl Debug for FiringVar {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         write!(f, "fvID({}'{})", (self.0).0.connector_id, (self.0).0.u32_suffix)
+    }
+}
 impl Debug for ProtoComponentId {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         write!(f, "pcID({}'{})", self.0.connector_id, self.0.u32_suffix)
+    }
+}
 impl std::ops::Not for Polarity {
     type Output = Self;
     fn not(self) -> Self::Output {
         use Polarity::*;
         match self {
             Putter => Getter,
             Getter => Putter,
+        }
+    }
+}

src/runtime/error.rs

➞

Show inline comments

 use crate::common::*;
 #[derive(Debug)]
 pub enum ConnectError {
     BindFailed(SocketAddr),
     PollInitFailed,
     Timeout,
     PollFailed,
     AcceptFailed(SocketAddr),
     AlreadyConnected,
     PortPeerPolarityMismatch(PortId),
     EndpointSetupError(SocketAddr, EndpointError),
     SetupAlgMisbehavior,
+}
 #[derive(Eq, PartialEq, Copy, Clone, Debug)]
 pub enum AddComponentError {
     NoSuchComponent,
     NonPortTypeParameters,
     CannotMovePort(PortId),
     WrongNumberOfParamaters { expected: usize },
     UnknownPort(PortId),
     WrongPortPolarity { port: PortId, expected_polarity: Polarity },
     DuplicateMovedPort(PortId),
+}
 ////////////////////////
 #[derive(Debug, Clone)]
 pub enum SyncError {
     NotConnected,
     InconsistentProtoComponent(ProtoComponentId),
     IndistinguishableBatches([usize; 2]),
     RoundFailure,
     PollFailed,
     BrokenEndpoint(usize),
     MalformedStateError(MalformedStateError),
+}
 #[derive(Debug, Clone)]
 pub enum MalformedStateError {
     PortCannotPut(PortId),
     GetterUnknownFor { putter: PortId },
+}
 #[derive(Debug, Clone)]
 pub enum EndpointError {
     MalformedMessage,
     BrokenEndpoint,
+}
 #[derive(Debug)]
 pub enum PortOpError {
     WrongPolarity,
     UnknownPolarity,
     NotConnected,
     MultipleOpsOnPort,
     PortUnavailable,
+}
 #[derive(Debug, Eq, PartialEq)]
 pub enum GottenError {
     NoPreviousRound,
     PortDidntGet,
     PreviousSyncFailed,
+}
 #[derive(Debug, Eq, PartialEq)]
 pub enum NextBatchError {
     NotConnected,
+}
 #[derive(Debug, Eq, PartialEq)]
 pub enum NewNetPortError {
     AlreadyConnected,
+}

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::*;
 #[derive(Debug)]
 pub struct Connector {
     unphased: ConnectorUnphased,
     phased: ConnectorPhased,
+}
 pub trait Logger: Debug {
     fn line_writer(&mut self) -> &mut dyn std::io::Write;
+}
 #[derive(Debug)]
 pub struct VecLogger(ConnectorId, Vec<u8>);
 #[derive(Debug)]
 pub struct DummyLogger;
 #[derive(Debug)]
 pub struct FileLogger(ConnectorId, std::fs::File);
 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,
     predicate: &'a Predicate,
     port_info: &'a PortInfo,
     inbox: &'a HashMap<PortId, Payload>,
+}
 #[derive(Debug)]
 struct RoundOk {
     batch_index: usize,
     gotten: HashMap<PortId, Payload>,
+}
 #[derive(Default)]
 struct VecSet<T: std::cmp::Ord> {
     // invariant: ordered, deduplicated
     vec: Vec<T>,
+}
 #[derive(Debug, Clone, Copy, Eq, PartialEq, Hash, serde::Serialize, serde::Deserialize)]
 enum ComponentId {
     Native,
     Proto(ProtoComponentId),
+}
 #[derive(Debug, Clone, Copy, Eq, PartialEq, Hash, serde::Serialize, serde::Deserialize)]
 enum Route {
     LocalComponent(ComponentId),
     Endpoint { index: usize },
+}
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 struct MyPortInfo {
     polarity: Polarity,
     port: PortId,
+}
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 enum Decision {
     Failure,
     Success(Predicate),
+}
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 enum Msg {
     SetupMsg(SetupMsg),
     CommMsg(CommMsg),
+}
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 enum SetupMsg {
     MyPortInfo(MyPortInfo),
     LeaderWave { wave_leader: ConnectorId },
     LeaderAnnounce { tree_leader: ConnectorId },
     YouAreMyParent,
     SessionGather { unoptimized_map: HashMap<ConnectorId, SessionInfo> },
     SessionScatter { optimized_map: HashMap<ConnectorId, SessionInfo> },
+}
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 struct SessionInfo {
     serde_proto_description: SerdeProtocolDescription,
     port_info: PortInfo,
     proto_components: HashMap<ProtoComponentId, ProtoComponent>,
+}
 #[derive(Debug, Clone)]
 struct SerdeProtocolDescription(Arc<ProtocolDescription>);
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 struct CommMsg {
     round_index: usize,
     contents: CommMsgContents,
+}
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 enum CommMsgContents {
     SendPayload(SendPayloadMsg),
     Suggest { suggestion: Decision }, // SINKWARD
     Announce { decision: Decision },  // SINKAWAYS
+}
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 struct SendPayloadMsg {
     predicate: Predicate,
     payload: Payload,
+}
 #[derive(Debug, PartialEq)]
 enum CommonSatResult {
     FormerNotLatter,
     LatterNotFormer,
     Equivalent,
     New(Predicate),
     Nonexistant,
+}
 struct Endpoint {
     inbox: Vec<u8>,
     stream: TcpStream,
+}
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 struct ProtoComponent {
     state: ComponentState,
     ports: HashSet<PortId>,
+}
 #[derive(Debug, Clone)]
 struct EndpointSetup {
     sock_addr: SocketAddr,
     endpoint_polarity: EndpointPolarity,
+}
 #[derive(Debug)]
 struct EndpointExt {
     endpoint: Endpoint,
     getter_for_incoming: PortId,
+}
 #[derive(Debug)]
 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 EndpointManager {
     // invariants:
     // 1. endpoint N is registered READ | WRITE with poller
     // 2. Events is empty
     poll: Poll,
     events: Events,
-    polled_undrained: IndexSet<usize>,
+    polled_undrained: VecSet<usize>,
     delayed_messages: Vec<(usize, Msg)>,
     undelayed_messages: Vec<(usize, Msg)>,
     endpoint_exts: Vec<EndpointExt>,
+}
 #[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)]
 enum ConnectorPhased {
     Setup { endpoint_setups: Vec<(PortId, EndpointSetup)>, surplus_sockets: u16 },
     Communication(Box<ConnectorCommunication>),
+}
 #[derive(Default, Clone, Eq, PartialEq, Hash, serde::Serialize, serde::Deserialize)]
 struct Predicate {
     assigned: BTreeMap<FiringVar, bool>,
+}
 #[derive(Debug, Default)]
 struct NativeBatch {
     // invariant: putters' and getters' polarities respected
     to_put: HashMap<PortId, Payload>,
     to_get: HashSet<PortId>,
+}
 ////////////////
 pub fn would_block(err: &std::io::Error) -> bool {
     err.kind() == std::io::ErrorKind::WouldBlock
+}
 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 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 firing_var_for(&self, port: PortId) -> FiringVar {
         FiringVar(match self.polarities.get(&port).unwrap() {
             Getter => port,
             Putter => *self.peers.get(&port).unwrap(),
         })
+    }
+}
 impl IdManager {
     fn new(connector_id: ConnectorId) -> Self {
         Self {
             connector_id,
             port_suffix_stream: Default::default(),
             proto_component_suffix_stream: Default::default(),
+        }
+    }
     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::*;
         // 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(WrongNumberOfParamaters { expected: polarities.len() });
+        }
         for (&expected_polarity, port) in polarities.iter().zip(ports.iter()) {
             if !cu.native_ports.contains(port) {
                 return Err(UnknownPort(*port));
+            }
             if expected_polarity != *cu.port_info.polarities.get(port).unwrap() {
                 return Err(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: FiringVar, v: bool) -> Self {
         self.assigned.insert(k, v);
         self
+    }
     // returns true IFF self.unify would return Equivalent OR FormerNotLatter
     pub fn satisfies(&self, other: &Self) -> bool {
         let mut s_it = self.assigned.iter();
         let mut s = if let Some(s) = s_it.next() {
+            s
         } else {
             return other.assigned.is_empty();
         };
         for (oid, ob) in other.assigned.iter() {
             while s.0 < oid {
                 s = if let Some(s) = s_it.next() {
+                    s
                 } else {
                     return false;
                 };
+            }
             if s.0 > oid || s.1 != ob {
                 return false;
+            }
+        }
         true
+    }
     /// Given self and other, two predicates, return the most general Predicate possible, N
     /// such that n.satisfies(self) && n.satisfies(other).
     /// If none exists Nonexistant is returned.
     /// If the resulting predicate is equivlanet to self, other, or both,
     /// FormerNotLatter, LatterNotFormer and Equivalent are returned respectively.
     /// otherwise New(N) is returned.
     fn common_satisfier(&self, other: &Self) -> CommonSatResult {
         use CommonSatResult as Csr;
         // 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 Csr::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] => Csr::Equivalent,       // ... equivalent to both.
             [false, true] => Csr::FormerNotLatter, // ... equivalent to self.
             [true, false] => Csr::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);
+                }
                 Csr::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: FiringVar) -> Option<bool> {
         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 {
         struct MySet<'a>(&'a Predicate, bool);
         impl Debug for MySet<'_> {
             fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
                 let iter = self.0.assigned.iter().filter_map(|(port, &firing)| {
                     if firing == self.1 {
                         Some(port)
                     } else {
                         None
+                    }
                 });
                 f.debug_set().entries(iter).finish()
+            }
+        }
         f.debug_struct("Predicate")
             .field("Trues", &MySet(self, true))
             .field("Falses", &MySet(self, false))
             .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)?;
         Ok(Self(Arc::new(inner)))
+    }
+}

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 { endpoint_setups: Default::default(), surplus_sockets },
+        }
+    }
     pub fn new_net_port(
         &mut self,
         polarity: Polarity,
         sock_addr: SocketAddr,
         endpoint_polarity: EndpointPolarity,
     ) -> Result<PortId, NewNetPortError> {
         let Self { unphased: up, phased } = self;
         match phased {
             ConnectorPhased::Communication { .. } => Err(NewNetPortError::AlreadyConnected),
             ConnectorPhased::Setup { endpoint_setups, .. } => {
                 let endpoint_setup = EndpointSetup { sock_addr, endpoint_polarity };
                 let p = up.id_manager.new_port_id();
                 up.native_ports.insert(p);
                 // {polarity, route} known. {peer} unknown.
                 up.port_info.polarities.insert(p, polarity);
                 up.port_info.routes.insert(p, Route::LocalComponent(ComponentId::Native));
                 log!(
                     up.logger,
                     "Added net port {:?} with polarity {:?} and endpoint setup {:?} ",
                     p,
                     polarity,
                     &endpoint_setup
                 );
                 endpoint_setups.push((p, endpoint_setup));
                 Ok(p)
+            }
+        }
+    }
     pub fn connect(&mut self, timeout: Option<Duration>) -> Result<(), ConnectError> {
         use ConnectError::*;
         let Self { unphased: cu, phased } = self;
         match phased {
             ConnectorPhased::Communication { .. } => {
                 log!(cu.logger, "Call to connecting in connected state");
                 Err(AlreadyConnected)
+            }
             ConnectorPhased::Setup { endpoint_setups, .. } => {
                 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,
                     endpoint_setups,
                     &mut cu.port_info,
                     deadline,
                 )?;
                 log!(
                     cu.logger,
                     "Successfully connected {} endpoints",
                     endpoint_manager.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),
                 };
                 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,
     endpoint_setups: &[(PortId, EndpointSetup)],
     port_info: &mut PortInfo,
     deadline: Option<Instant>,
 ) -> Result<EndpointManager, ConnectError> {
     ////////////////////////////////////////////
     use std::sync::atomic::AtomicBool;
     use ConnectError::*;
     const BOTH: Interest = Interest::READABLE.add(Interest::WRITABLE);
     struct Todo {
         todo_endpoint: TodoEndpoint,
         endpoint_setup: EndpointSetup,
         local_port: PortId,
         sent_local_port: bool,          // true <-> I've sent my local port
         recv_peer_port: Option<PortId>, // Some(..) <-> I've received my peer's port
+    }
     enum TodoEndpoint {
         Accepting(TcpListener),
         Endpoint(Endpoint),
+    }
     fn init_todo(
         token: Token,
         local_port: PortId,
         endpoint_setup: &EndpointSetup,
         poll: &mut Poll,
     ) -> Result<Todo, ConnectError> {
         let todo_endpoint = if let EndpointPolarity::Active = endpoint_setup.endpoint_polarity {
             let mut stream = TcpStream::connect(endpoint_setup.sock_addr)
                 .expect("mio::TcpStream connect should not fail!");
             poll.registry().register(&mut stream, token, BOTH).unwrap();
             TodoEndpoint::Endpoint(Endpoint { stream, inbox: vec![] })
         } else {
             let mut listener = TcpListener::bind(endpoint_setup.sock_addr)
                 .map_err(|_| BindFailed(endpoint_setup.sock_addr))?;
             poll.registry().register(&mut listener, token, BOTH).unwrap();
             TodoEndpoint::Accepting(listener)
         };
         Ok(Todo {
             todo_endpoint,
             local_port,
             sent_local_port: false,
             recv_peer_port: None,
             endpoint_setup: endpoint_setup.clone(),
         })
+    }
     ////////////////////////////////////////////
     // 1. Start to construct EndpointManager
     const WAKER_TOKEN: Token = Token(usize::MAX);
     const WAKER_PERIOD: Duration = Duration::from_millis(300);
     assert!(endpoint_setups.len() < WAKER_TOKEN.0); // using MAX usize as waker token
     let mut waker_continue_signal: Option<Arc<AtomicBool>> = None;
     let mut poll = Poll::new().map_err(|_| PollInitFailed)?;
     let mut events = Events::with_capacity(endpoint_setups.len() * 2 + 4);
-    let mut polled_undrained = IndexSet::default();
+    let mut polled_undrained = VecSet::default();
     let mut delayed_messages = vec![];
     // 2. create a registered (TcpListener/Endpoint) for passive / active respectively
     let mut todos = endpoint_setups
         .iter()
         .enumerate()
         .map(|(index, (local_port, endpoint_setup))| {
             init_todo(Token(index), *local_port, endpoint_setup, &mut poll)
         })
         .collect::<Result<Vec<Todo>, ConnectError>>()?;
     // 3. Using poll to drive progress:
     //    - accept an incoming connection for each TcpListener (turning them into endpoints too)
     //    - for each endpoint, send the local PortId
     //    - for each endpoint, recv the peer's PortId, and
     // all in connect_failed are NOT registered with Poll
     let mut connect_failed: HashSet<usize> = Default::default();
     let mut setup_incomplete: HashSet<usize> = (0..todos.len()).collect();
     while !setup_incomplete.is_empty() {
         let remaining = if let Some(deadline) = deadline {
             Some(deadline.checked_duration_since(Instant::now()).ok_or(Timeout)?)
         } else {
             None
         };
         poll.poll(&mut events, remaining).map_err(|_| PollFailed)?;
         for event in events.iter() {
             let token = event.token();
             let Token(index) = token;
             if token == WAKER_TOKEN {
                 log!(
                     logger,
                     "Notification from waker. connect_failed is {:?}",
                     connect_failed.iter()
                 );
                 assert!(waker_continue_signal.is_some());
                 for index in connect_failed.drain() {
                     let todo: &mut Todo = &mut todos[index];
                     log!(
                         logger,
                         "Restarting connection with endpoint {:?} {:?}",
                         index,
                         todo.endpoint_setup.sock_addr
                     );
                     match &mut todo.todo_endpoint {
                         TodoEndpoint::Endpoint(endpoint) => {
                             let mut new_stream = TcpStream::connect(todo.endpoint_setup.sock_addr)
                                 .expect("mio::TcpStream connect should not fail!");
                             std::mem::swap(&mut endpoint.stream, &mut new_stream);
                             poll.registry()
                                 .register(&mut endpoint.stream, Token(index), BOTH)
                                 .unwrap();
+                        }
                         _ => unreachable!(),
+                    }
+                }
             } else {
                 let todo: &mut Todo = &mut todos[index];
                 // FIRST try convert this into an endpoint
                 if let TodoEndpoint::Accepting(listener) = &mut todo.todo_endpoint {
                     match listener.accept() {
                         Ok((mut stream, peer_addr)) => {
                             poll.registry().deregister(listener).unwrap();
                             poll.registry().register(&mut stream, token, BOTH).unwrap();
                             log!(
                                 logger,
                                 "Endpoint[{}] accepted a connection from {:?}",
                                 index,
                                 peer_addr
                             );
                             let endpoint = Endpoint { stream, inbox: vec![] };
                             todo.todo_endpoint = TodoEndpoint::Endpoint(endpoint);
+                        }
                         Err(e) if would_block(&e) => {
                             log!(logger, "Spurious wakeup on listener {:?}", index)
+                        }
                         Err(_) => {
                             log!(logger, "accept() failure on index {}", index);
                             return Err(AcceptFailed(listener.local_addr().unwrap()));
+                        }
+                    }
+                }
                 if let TodoEndpoint::Endpoint(endpoint) = &mut todo.todo_endpoint {
                     if event.is_error() {
                         if todo.endpoint_setup.endpoint_polarity == EndpointPolarity::Passive {
                             // right now you cannot retry an acceptor.
                             return Err(AcceptFailed(endpoint.stream.local_addr().unwrap()));
+                        }
                         if connect_failed.insert(index) {
                             log!(
                                 logger,
                                 "Connection failed for {:?}. List is {:?}",
                                 index,
                                 connect_failed.iter()
                             );
                             poll.registry().deregister(&mut endpoint.stream).unwrap();
                         } else {
                             // spurious wakeup
                             continue;
+                        }
                         if waker_continue_signal.is_none() {
                             log!(logger, "First connect failure. Starting waker thread");
                             let waker =
                                 Arc::new(mio::Waker::new(poll.registry(), WAKER_TOKEN).unwrap());
                             let wcs = Arc::new(AtomicBool::from(true));
                             let wcs2 = wcs.clone();
                             std::thread::spawn(move || {
                                 while wcs2.load(std::sync::atomic::Ordering::SeqCst) {
                                     std::thread::sleep(WAKER_PERIOD);
                                     let _ = waker.wake();
+                                }
                             });
                             waker_continue_signal = Some(wcs);
+                        }
                         continue;
+                    }
                     if connect_failed.contains(&index) {
                         // spurious wakeup
                         continue;
+                    }
                     if !setup_incomplete.contains(&index) {
                         // spurious wakeup
                         continue;
+                    }
                     let local_polarity = *port_info.polarities.get(&todo.local_port).unwrap();
                     if event.is_writable() && !todo.sent_local_port {
                         let msg = Msg::SetupMsg(SetupMsg::MyPortInfo(MyPortInfo {
                             polarity: local_polarity,
                             port: todo.local_port,
                         }));
                         endpoint
                             .send(&msg)
                             .map_err(|e| {
                                 EndpointSetupError(endpoint.stream.local_addr().unwrap(), e)
                             })
                             .unwrap();
                         log!(logger, "endpoint[{}] sent msg {:?}", index, &msg);
                         todo.sent_local_port = true;
+                    }
                     if event.is_readable() && todo.recv_peer_port.is_none() {
                         let maybe_msg = endpoint.try_recv(logger).map_err(|e| {
                             EndpointSetupError(endpoint.stream.local_addr().unwrap(), e)
                         })?;
                         if maybe_msg.is_some() && !endpoint.inbox.is_empty() {
                             polled_undrained.insert(index);
+                        }
                         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(
                                         todo.local_port,
                                     ));
+                                }
                                 todo.recv_peer_port = Some(peer_info.port);
                                 // 1. finally learned the peer of this port!
                                 port_info.peers.insert(todo.local_port, 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, todo.local_port);
                                 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::Endpoint { index });
+                            }
                             Some(inappropriate_msg) => {
                                 log!(
                                     logger,
                                     "delaying msg {:?} during channel setup phase",
                                     inappropriate_msg
                                 );
                                 delayed_messages.push((index, inappropriate_msg));
+                            }
+                        }
+                    }
                     if todo.sent_local_port && todo.recv_peer_port.is_some() {
                         setup_incomplete.remove(&index);
                         log!(logger, "endpoint[{}] is finished!", index);
+                    }
+                }
+            }
+        }
         events.clear();
+    }
     let endpoint_exts = todos
         .into_iter()
         .enumerate()
         .map(|(index, Todo { todo_endpoint, local_port, .. })| EndpointExt {
             endpoint: match todo_endpoint {
                 TodoEndpoint::Endpoint(mut endpoint) => {
                     poll.registry()
                         .reregister(&mut endpoint.stream, Token(index), Interest::READABLE)
                         .unwrap();
                     endpoint
+                }
                 _ => unreachable!(),
             },
             getter_for_incoming: local_port,
         })
         .collect();
     if let Some(wcs) = waker_continue_signal {
         log!(logger, "Sending waker the stop signal");
         wcs.store(false, std::sync::atomic::Ordering::SeqCst);
+    }
     Ok(EndpointManager {
         poll,
         events,
         polled_undrained,
         undelayed_messages: delayed_messages, // no longer delayed
         delayed_messages: Default::default(),
         endpoint_exts,
     })
+}
 fn init_neighborhood(
     connector_id: ConnectorId,
     logger: &mut dyn Logger,
     em: &mut EndpointManager,
     deadline: Option<Instant>,
 ) -> Result<Neighborhood, ConnectError> {
     ////////////////////////////////
     use {ConnectError::*, Msg::SetupMsg as S, SetupMsg::*};
     #[derive(Debug)]
     struct WaveState {
         parent: Option<usize>,
         leader: ConnectorId,
+    }
     fn do_wave(
         em: &mut EndpointManager,
         awaiting: &mut HashSet<usize>,
         ws: &WaveState,
     ) -> Result<(), ConnectError> {
         awaiting.clear();
         let msg = S(LeaderWave { wave_leader: ws.leader });
         for index in em.index_iter() {
             if Some(index) != ws.parent {
                 em.send_to_setup(index, &msg)?;
                 awaiting.insert(index);
+            }
+        }
         Ok(())
+    }
     ///////////////////////
     /*
     Conceptually, we have two distinct disstributed algorithms back-to-back
 . Leader election using echo algorithm with extinction.
         - Each connector initiates a wave tagged with their ID
         - Connectors participate in waves of GREATER ID, abandoning previous waves
         - Only the wave of the connector with GREATEST ID completes, whereupon they are the leader
 . Tree construction
         - The leader broadcasts their leadership with msg A
         - Upon receiving their first announcement, connectors reply B, and send A to all peers
         - A controller exits once they have received A or B from each neighbor
     The actual implementation is muddier, because non-leaders aren't aware of termiantion of algorithm 1,
     so they rely on receipt of the leader's announcement to realize that algorithm 2 has begun.
     NOTE the distinction between PARENT and LEADER
     */
     log!(logger, "beginning neighborhood construction");
     if em.num_endpoints() == 0 {
         log!(logger, "Edge case of no neighbors! No parent an no children!");
         return Ok(Neighborhood { parent: None, children: VecSet::new(vec![]) });
+    }
     log!(logger, "Have {} endpoints. Must participate in distributed alg.", em.num_endpoints());
     let mut awaiting = HashSet::with_capacity(em.num_endpoints());
     // 1+ neighbors. Leader can only be learned by receiving messages
     // loop ends when I know my sink tree parent (implies leader was elected)
     let election_result: WaveState = {
         // initially: No parent, I'm the best leader.
         let mut best_wave = WaveState { parent: None, leader: connector_id };
         // start a wave for this initial state
         do_wave(em, &mut awaiting, &best_wave)?;
         // with 1+ neighbors, progress is only made in response to incoming messages
         em.undelay_all();
         'election: loop {
             log!(logger, "Election loop. awaiting {:?}...", awaiting.iter());
             let (recv_index, msg) = em.try_recv_any_setup(logger, deadline)?;
             log!(logger, "Received from index {:?} msg {:?}", &recv_index, &msg);
             match msg {
                 S(LeaderAnnounce { tree_leader }) => {
                     let election_result =
                         WaveState { leader: tree_leader, parent: Some(recv_index) };
                     log!(logger, "Election lost! Result {:?}", &election_result);
                     assert!(election_result.leader >= best_wave.leader);
                     assert_ne!(election_result.leader, connector_id);
                     break 'election election_result;
+                }
                 S(LeaderWave { wave_leader }) => {
                     use Ordering as O;
                     match wave_leader.cmp(&best_wave.leader) {
                         O::Less => log!(
                             logger,
                             "Ignoring wave with Id {:?}<{:?}",
                             wave_leader,
                             best_wave.leader
                         ),
                         O::Greater => {
                             log!(
                                 logger,
                                 "Joining wave with Id {:?}>{:?}",
                                 wave_leader,
                                 best_wave.leader
                             );
                             best_wave = WaveState { leader: wave_leader, parent: Some(recv_index) };
                             log!(logger, "New wave state {:?}", &best_wave);
                             do_wave(em, &mut awaiting, &best_wave)?;
                             if awaiting.is_empty() {
                                 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(YouAreMyParent) | msg @ S(MyPortInfo(_)) => {
                     log!(logger, "Endpont {:?} sent unexpected msg! {:?}", recv_index, &msg);
                     return Err(SetupAlgMisbehavior);
+                }
                 msg @ S(SessionScatter { .. })
                 | msg @ S(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(YouAreMyParent))?;
         } else {
             awaiting.insert(index);
             em.send_to_setup(index, &S(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(LeaderAnnounce { .. }) => {
                 // not a child
                 log!(
                     logger,
                     "Got reply from non-child index {:?}. Children: {:?}",
                     recv_index,
                     children.iter()
                 );
                 if !awaiting.remove(&recv_index) {
                     return Err(SetupAlgMisbehavior);
+                }
+            }
             S(YouAreMyParent) => {
                 if !awaiting.remove(&recv_index) {
                     log!(
                         logger,
                         "Got reply from child index {:?}. Children before... {:?}",
                         recv_index,
                         children.iter()
                     );
                     return Err(SetupAlgMisbehavior);
+                }
                 children.push(recv_index);
+            }
             msg @ S(MyPortInfo(_)) | msg @ S(LeaderWave { .. }) => {
                 log!(logger, "discarding old message {:?} during election", msg);
+            }
             msg @ S(SessionScatter { .. })
             | msg @ S(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::*, Msg::SetupMsg as S, SetupMsg::*};
     ////////////////////////////////////////
     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(SessionGather { unoptimized_map: child_unoptimized_map }) => {
                 if !awaiting.remove(&recv_index) {
                     log!(
                         cu.logger,
                         "Wasn't expecting session info from {:?}. Got {:?}",
                         recv_index,
                         &child_unoptimized_map
                     );
                     return Err(SetupAlgMisbehavior);
+                }
                 unoptimized_map.extend(child_unoptimized_map.into_iter());
+            }
             msg @ S(YouAreMyParent)
             | msg @ S(MyPortInfo(..))
             | msg @ S(LeaderAnnounce { .. })
             | msg @ S(LeaderWave { .. }) => {
                 log!(cu.logger, "discarding old message {:?} during election", msg);
+            }
             msg @ S(SessionScatter { .. }) => {
                 log!(
                     cu.logger,
                     "Endpoint {:?} sent unexpected scatter! {:?} I've not contributed yet!",
                     recv_index,
                     &msg
                 );
                 return Err(SetupAlgMisbehavior);
+            }
             msg @ Msg::CommMsg(..) => {
                 log!(cu.logger, "delaying msg {:?} during session optimization", msg);
                 comm.endpoint_manager.delayed_messages.push((recv_index, msg));
+            }
+        }
+    }
     log!(
         cu.logger,
         "Gathered all children's maps. ConnectorId set is... {:?}",
         unoptimized_map.keys()
     );
     let my_session_info = SessionInfo {
         port_info: cu.port_info.clone(),
         proto_components: cu.proto_components.clone(),
         serde_proto_description: SerdeProtocolDescription(cu.proto_description.clone()),
     };
     unoptimized_map.insert(cu.id_manager.connector_id, my_session_info);
     log!(cu.logger, "Inserting my own info. Unoptimized subtree map is {:?}", &unoptimized_map);
     // acquire the optimized info...
     let optimized_map = if let Some(parent) = comm.neighborhood.parent {
         // ... as a message from my parent
         log!(cu.logger, "Forwarding gathered info to parent {:?}", parent);
         let msg = S(SessionGather { unoptimized_map });
         comm.endpoint_manager.send_to_setup(parent, &msg)?;
         'scatter_loop: loop {
             log!(
                 cu.logger,
                 "Session scatter recv loop. awaiting info from children {:?}...",
                 awaiting.iter()
             );
             let (recv_index, msg) =
                 comm.endpoint_manager.try_recv_any_setup(&mut *cu.logger, deadline)?;
             log!(cu.logger, "Received from index {:?} msg {:?}", &recv_index, &msg);
             match msg {
                 S(SessionScatter { optimized_map }) => {
                     if recv_index != parent {
                         log!(cu.logger, "I expected the scatter from my parent only!");
                         return Err(SetupAlgMisbehavior);
+                    }
                     break 'scatter_loop optimized_map;
+                }
                 msg @ Msg::CommMsg { .. } => {
                     log!(cu.logger, "delaying msg {:?} during scatter recv", msg);
                     comm.endpoint_manager.delayed_messages.push((recv_index, msg));
+                }
                 msg @ S(SessionGather { .. })
                 | msg @ S(YouAreMyParent)
                 | msg @ S(MyPortInfo(..))
                 | msg @ S(LeaderAnnounce { .. })
                 | msg @ S(LeaderWave { .. }) => {
                     log!(cu.logger, "discarding old message {:?} during election", msg);
+                }
+            }
+        }
     } else {
         // by computing it myself
         log!(cu.logger, "I am the leader! I will optimize this session");
         leader_session_map_optimize(&mut *cu.logger, unoptimized_map)?
     };
     log!(
         cu.logger,
         "Optimized info map is {:?}. Sending to children {:?}",
         &optimized_map,
         comm.neighborhood.children.iter()
     );
     log!(cu.logger, "All session info dumped!: {:#?}", &optimized_map);
     let optimized_info =
         optimized_map.get(&cu.id_manager.connector_id).expect("HEY NO INFO FOR ME?").clone();
     let msg = S(SessionScatter { optimized_map });
     for &child in comm.neighborhood.children.iter() {
         comm.endpoint_manager.send_to_setup(child, &msg)?;
+    }
     apply_optimizations(cu, comm, optimized_info)?;
     log!(cu.logger, "Session optimizations applied");
     Ok(())
+}
 fn leader_session_map_optimize(
     logger: &mut dyn Logger,
     unoptimized_map: HashMap<ConnectorId, SessionInfo>,
 ) -> Result<HashMap<ConnectorId, SessionInfo>, ConnectError> {
     log!(logger, "Session map optimize START");
     log!(logger, "Session map optimize END");
     Ok(unoptimized_map)
+}
 fn apply_optimizations(
     cu: &mut ConnectorUnphased,
     _comm: &mut ConnectorCommunication,
     session_info: SessionInfo,
 ) -> Result<(), ConnectError> {
     let SessionInfo { proto_components, port_info, serde_proto_description } = session_info;
     cu.port_info = port_info;
     cu.proto_components = proto_components;
     cu.proto_description = serde_proto_description.0;
     Ok(())
+}

0 comments (0 inline, 0 general)