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

Changeset - 14c1a07d148a

Parent rev.

Child rev.

[Not reviewed]

3 17 1

Christopher Esterhuyse - 5 years ago 2020-07-22 15:32:26
christopher.esterhuyse@gmail.com

pseudo-socket ffi WIP. made better FFI socketaddr struct for use in platform independent C. new reowolf.h accordingly.

20 files changed with 152 insertions and 741 deletions:

examples/8_net_ports/amy.c

examples/9_net_self_putget/amy.c

examples/a_swap/amy.c

examples/a_swap/bob.c

examples/make.py

examples/pres_1/amy.c

examples/pres_1/bob.c

examples/pres_2/bob.c

examples/pres_3/amy.c

examples/pres_3/bob.c

examples/pres_4/bob.c

examples/pres_5/amy.c

examples/pres_5/bob.c

reowolf.h

465

src/ffi/mod.rs

src/ffi/pseudo_socket_api.rs

src/runtime/communication.rs

src/runtime/endpoints.rs

src/runtime/mod.rs

src/runtime/setup.rs

0 comments (0 inline, 0 general)

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, Polarity_Putter, EndpointPolarity_Active);
 	FfiSocketAddr addr = {{127,0,0,1}, 8000};
 	connector_add_net_port(c, &putter, addr, 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, Polarity_Getter, EndpointPolarity_Passive);
 	connector_add_net_port(c, &getter, addr, 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, Polarity_Putter, EndpointPolarity_Active);
 	FfiSocketAddr addr = {{127,0,0,1}, 8000};
 	connector_add_net_port(c, &putter, addr, 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, Polarity_Getter, EndpointPolarity_Passive);
 	connector_add_net_port(c, &getter, addr, 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

deleted file

examples/a_swap/bob.c

➞

Show inline comments

deleted file

examples/make.py

➞

Show inline comments

 import os, glob, subprocess, time
 script_path = os.path.dirname(os.path.realpath(__file__));
 for c_file in glob.glob(script_path + "/*/*.c", recursive=False):
   print("compiling", c_file)
   args = [
     "gcc",          # compiler
-    "-std=c11"      # C11 mode
+    "-std=c11",     # C11 mode
     "-L",           # lib path flag
     "./",           # where to look for libs
     "-lreowolf_rs", # add lib called "reowolf_rs"
     "-Wl,-R./",     # pass -R flag to linker: produce relocatable object
     c_file,         # input source file
     "-o",           # output flag
     c_file[:-2]     # output filename
   ];
   subprocess.run(args)
 input("Blocking until newline...");

examples/pres_1/amy.c

➞

Show inline comments

 #include "../../reowolf.h"
 #include "../utility.c"
 int main(int argc, char** argv) {
 	char msgbuf[64];
 	// ask user what message to send
 	size_t msglen = get_user_msg(msgbuf, sizeof(msgbuf));
 	// Create a connector, configured with our (trivial) protocol.
 	Arc_ProtocolDescription * pd = protocol_description_parse("", 0);
 	char logpath[] = "./pres_1_amy.txt";
 	Connector * c = connector_new_logging(pd, logpath, sizeof(logpath)-1);
 	rw_err_peek(c);
 	// ... with 1 outgoing network connection
 	PortId p0;
 	char addr_str[] = "127.0.0.1:8000";
 	connector_add_net_port(c, &p0, addr_str, sizeof(addr_str)-1,
 			Polarity_Putter, EndpointPolarity_Passive);
 	FfiSocketAddr addr = {{127,0,0,1}, 8000};
 	connector_add_net_port(c, &p0, addr, Polarity_Putter, EndpointPolarity_Passive);
 	rw_err_peek(c);
 	// Connect with peers (5000ms timeout).
 	connector_connect(c, 5000);
 	rw_err_peek(c);
 	// Prepare a message to send
 	connector_put_bytes(c, p0, msgbuf, msglen);
 	rw_err_peek(c);
 	// ... reach new consistent state within 1000ms deadline.
 	connector_sync(c, 1000);
 	rw_err_peek(c);
 	printf("Exiting\n");
 	protocol_description_destroy(pd);
 	connector_destroy(c);
 	sleep(1.0);
 	return 0;
+}
@@ \ No newline at end of file @@

examples/pres_1/bob.c

➞

Show inline comments

 #include "../../reowolf.h"
 #include "../utility.c"
 int main(int argc, char** argv) {
 	// Create a connector, configured with our (trivial) protocol.
 	Arc_ProtocolDescription * pd = protocol_description_parse("", 0);
 	char logpath[] = "./pres_1_bob.txt";
 	Connector * c = connector_new_logging(pd, logpath, sizeof(logpath)-1);
 	rw_err_peek(c);
 	// ... with 1 outgoing network connection
 	PortId p0;
 	char addr_str[] = "127.0.0.1:8000";
 	connector_add_net_port(c, &p0, addr_str, sizeof(addr_str)-1,
 			Polarity_Getter, EndpointPolarity_Active);
 	FfiSocketAddr addr = {{127,0,0,1}, 8000};
 	connector_add_net_port(c, &p0, addr, Polarity_Getter, EndpointPolarity_Active);
 	rw_err_peek(c);
 	// Connect with peers (5000ms timeout).
 	connector_connect(c, 5000);
 	rw_err_peek(c);
 	// Prepare to receive a message.
 	connector_get(c, p0);
 	rw_err_peek(c);
 	// ... reach new consistent state within 1000ms deadline.
 	connector_sync(c, 1000);
 	rw_err_peek(c);
 	// Read our received message
 	size_t msg_len;
 	const char * msg_ptr = connector_gotten_bytes(c, p0, &msg_len);
 	printf("Got msg `%.*s`\n", msg_len, msg_ptr);
 	printf("Exiting\n");
 	protocol_description_destroy(pd);
 	connector_destroy(c);
 	sleep(1.0);
 	return 0;
+}
@@ \ No newline at end of file @@

examples/pres_2/bob.c

➞

Show inline comments

 #include "../../reowolf.h"
 #include "../utility.c"
 int main(int argc, char** argv) {
 	// Create a connector, configured with a protocol defined in a file
 	char * pdl = buffer_pdl("./eg_protocols.pdl");
 	Arc_ProtocolDescription * pd = protocol_description_parse(pdl, strlen(pdl));
 	char logpath[] = "./pres_2_bob.txt";
 	Connector * c = connector_new_logging(pd, logpath, sizeof(logpath)-1);
 	rw_err_peek(c);
 	// ... with 1 outgoing network connection
 	PortId ports[3];
 	char addr_str[] = "127.0.0.1:8000";
 	connector_add_net_port(c, &ports[0], addr_str, sizeof(addr_str)-1,
 			Polarity_Getter, EndpointPolarity_Active);
 	FfiSocketAddr addr = {{127,0,0,1}, 8000};
 	connector_add_net_port(c, &ports[0], addr, Polarity_Getter, EndpointPolarity_Active);
 	connector_add_port_pair(c, &ports[1], &ports[2]);
 	connector_add_component(c, "pres_2", 6, ports, 2);
 	rw_err_peek(c);
 	// Connect with peers (5000ms timeout).
 	connector_connect(c, 5000);
 	rw_err_peek(c);
 	// Prepare to receive a message.
 	connector_get(c, ports[2]);
 	rw_err_peek(c);
 	// ... reach new consistent state within 1000ms deadline.
 	connector_sync(c, 1000);
 	rw_err_peek(c);
 	// Read our received message
 	size_t msg_len;
 	const char * msg_ptr = connector_gotten_bytes(c, ports[2], &msg_len);
 	printf("Got msg `%.*s`\n", msg_len, msg_ptr);
 	printf("Exiting\n");
 	protocol_description_destroy(pd);
 	connector_destroy(c);
 	free(pdl);
 	sleep(1.0);
 	return 0;
+}
@@ \ No newline at end of file @@

examples/pres_3/amy.c

➞

Show inline comments

 #include "../../reowolf.h"
 #include "../utility.c"
 int main(int argc, char** argv) {
 	// Create a connector, configured with our (trivial) protocol.
 	Arc_ProtocolDescription * pd = protocol_description_parse("", 0);
 	char logpath[] = "./pres_3_amy.txt";
 	Connector * c = connector_new_logging(pd, logpath, sizeof(logpath)-1);
 	rw_err_peek(c);
 	// ... with 2 outgoing network connections
 	PortId ports[2];
 	char * addr = "127.0.0.1:8000";
 	connector_add_net_port(c, &ports[0], addr, strlen(addr),
 			Polarity_Putter, EndpointPolarity_Passive);
 	FfiSocketAddr addr = {{127,0,0,1}, 8000};
 	connector_add_net_port(c, &ports[0], addr, Polarity_Putter, EndpointPolarity_Passive);
 	rw_err_peek(c);
 	addr = "127.0.0.1:8001";
 	connector_add_net_port(c, &ports[1], addr, strlen(addr),
 			Polarity_Putter, EndpointPolarity_Passive);
 	addr.port = 8001;
 	connector_add_net_port(c, &ports[1], addr, Polarity_Putter, EndpointPolarity_Passive);
 	rw_err_peek(c);
 	// Connect with peers (5000ms timeout).
 	connector_connect(c, 5000);
 	rw_err_peek(c);
 	printf("\nputting {A}...\n");
 	connector_put_bytes(c, ports[0], "A", 1);
 	connector_sync(c, 1000);
 	rw_err_peek(c);
 	printf("\nputting {B}...\n");
 	connector_put_bytes(c, ports[1], "B", 1);
 	connector_sync(c, 1000);
 	rw_err_peek(c);
 	printf("\nputting {A, B}...\n");
 	connector_put_bytes(c, ports[0], "A", 1);
 	connector_put_bytes(c, ports[1], "B", 1);
 	connector_sync(c, 1000);
 	rw_err_peek(c);
 	printf("\nExiting\n");
 	protocol_description_destroy(pd);
 	connector_destroy(c);
 	sleep(1.0);
 	return 0;
+}
@@ \ No newline at end of file @@

examples/pres_3/bob.c

➞

Show inline comments

 #include "../../reowolf.h"
 #include "../utility.c"
 int main(int argc, char** argv) {
 	// Create a connector, configured with our (trivial) protocol.
 	Arc_ProtocolDescription * pd = protocol_description_parse("", 0);
 	char logpath[] = "./pres_3_bob.txt";
 	Connector * c = connector_new_logging(pd, logpath, sizeof(logpath)-1);
 	rw_err_peek(c);
 	// ... with 2 outgoing network connections
 	PortId ports[2];
 	char * addr = "127.0.0.1:8000";
 	connector_add_net_port(c, &ports[0], addr, strlen(addr),
 			Polarity_Getter, EndpointPolarity_Active);
 	FfiSocketAddr addr = {{127,0,0,1}, 8000};
 	connector_add_net_port(c, &ports[0], addr, Polarity_Getter, EndpointPolarity_Active);
 	rw_err_peek(c);
 	addr = "127.0.0.1:8001";
 	connector_add_net_port(c, &ports[1], addr, strlen(addr),
 			Polarity_Getter, EndpointPolarity_Active);
 	addr.port = 8001;
 	connector_add_net_port(c, &ports[1], addr, Polarity_Getter, EndpointPolarity_Active);
 	rw_err_peek(c);
 	// Connect with peers (5000ms timeout).
 	connector_connect(c, 5000);
 	rw_err_peek(c);
 	for(int i=0; i<3; i++) {
 		printf("\nGetting from both...\n");
 		connector_get(c, ports[0]);
 		rw_err_peek(c);
 		connector_get(c, ports[1]);
 		rw_err_peek(c);
 		connector_sync(c, 1000);
 		rw_err_peek(c);
+	}
 	printf("Exiting\n");
 	protocol_description_destroy(pd);
 	connector_destroy(c);
 	sleep(1.0);
 	return 0;
+}
@@ \ No newline at end of file @@

examples/pres_4/bob.c

➞

Show inline comments

 #include "../../reowolf.h"
 #include "../utility.c"
 int main(int argc, char** argv) {
 	// Create a connector, configured with our (trivial) protocol.
 	Arc_ProtocolDescription * pd = protocol_description_parse("", 0);
 	char logpath[] = "./pres_3_bob.txt";
 	Connector * c = connector_new_logging(pd, logpath, sizeof(logpath)-1);
 	rw_err_peek(c);
 	// ... with 2 outgoing network connections
 	PortId ports[2];
 	char * addr = "127.0.0.1:8000";
 	connector_add_net_port(c, &ports[0], addr, strlen(addr),
 			Polarity_Getter, EndpointPolarity_Active);
 	FfiSocketAddr addr = {{127,0,0,1}, 8000};
 	connector_add_net_port(c, &ports[0], addr, Polarity_Getter, EndpointPolarity_Active);
 	rw_err_peek(c);
 	addr = "127.0.0.1:8001";
 	connector_add_net_port(c, &ports[1], addr, strlen(addr),
 			Polarity_Getter, EndpointPolarity_Active);
 	addr.port = 8001;
 	connector_add_net_port(c, &ports[1], addr, Polarity_Getter, EndpointPolarity_Active);
 	rw_err_peek(c);
 	// Connect with peers (5000ms timeout).
 	connector_connect(c, 5000);
 	rw_err_peek(c);
 	for(int i=0; i<3; i++) {
 		printf("\nNext round...\n");
 		printf("\nOption 0: Get {A}\n");
 		connector_get(c, ports[0]);
 		connector_next_batch(c);
 		rw_err_peek(c);
 		printf("\nOption 1: Get {B}\n");
 		connector_get(c, ports[1]);
 		connector_next_batch(c);
 		rw_err_peek(c);
 		printf("\nOption 2: Get {A, B}\n");
 		connector_get(c, ports[0]);
 		connector_get(c, ports[1]);
 		int code = connector_sync(c, 1000);
 		printf("Outcome: %d\n", code);
 		rw_err_peek(c);
+	}
 	printf("Exiting\n");
 	protocol_description_destroy(pd);
 	connector_destroy(c);
 	sleep(1.0);
 	return 0;
+}
@@ \ No newline at end of file @@

examples/pres_5/amy.c

➞

Show inline comments

 #include "../../reowolf.h"
 #include "../utility.c"
 int main(int argc, char** argv) {
 	// Create a connector, configured with our (trivial) protocol.
 	Arc_ProtocolDescription * pd = protocol_description_parse("", 0);
 	char logpath[] = "./pres_3_amy.txt";
 	Connector * c = connector_new_logging(pd, logpath, sizeof(logpath)-1);
 	rw_err_peek(c);
 	// ... with 2 outgoing network connections
 	PortId ports[2];
 	char * addr = "127.0.0.1:8000";
 	connector_add_net_port(c, &ports[0], addr, strlen(addr),
 			Polarity_Putter, EndpointPolarity_Passive);
 	FfiSocketAddr addr = {{127,0,0,1}, 8000};
 	connector_add_net_port(c, &ports[0], addr, Polarity_Putter, EndpointPolarity_Passive);
 	rw_err_peek(c);
 	addr = "127.0.0.1:8001";
 	connector_add_net_port(c, &ports[1], addr, strlen(addr),
 			Polarity_Putter, EndpointPolarity_Passive);
 	addr.port = 8001;
 	connector_add_net_port(c, &ports[1], addr, Polarity_Putter, EndpointPolarity_Passive);
 	rw_err_peek(c);
 	// Connect with peers (5000ms timeout).
 	connector_connect(c, 5000);
 	rw_err_peek(c);
 	printf("Round 0. Putting {ports[0]=\"r0p0\", ports[1]=\"r0p1\"}\n");
 	connector_put_bytes(c, ports[0], "r0p0", 4);
 	connector_put_bytes(c, ports[1], "r0p1", 4);
 	connector_sync(c, 1000);
 	rw_err_peek(c);
 	printf("Round 1. Putting {ports[1]=\"r1p1\"}\n");
 	connector_put_bytes(c, ports[1], "r1p1", 4);
 	connector_sync(c, 1000);
 	rw_err_peek(c);
 	printf("Round 2. Putting {ports[0]=\"r2p0\"}\n");
 	connector_put_bytes(c, ports[0], "r2p0", 4);
 	connector_sync(c, 1000);
 	rw_err_peek(c);
 	printf("\nExiting\n");
 	protocol_description_destroy(pd);
 	connector_destroy(c);
 	sleep(1.0);
 	return 0;
+}
@@ \ No newline at end of file @@

examples/pres_5/bob.c

➞

Show inline comments

 #include "../../reowolf.h"
 #include "../utility.c"
 int main(int argc, char** argv) {
 	// Create a connector, configured with a protocol defined in a file
 	char * pdl = buffer_pdl("./eg_protocols.pdl");
 	Arc_ProtocolDescription * pd = protocol_description_parse(pdl, strlen(pdl));
 	char logpath[] = "./pres_3_bob.txt";
 	Connector * c = connector_new_logging(pd, logpath, sizeof(logpath)-1);
 	rw_err_peek(c);
 	// ... with 2 outgoing network connections
 	PortId ports[4];
 	char * addr = "127.0.0.1:8000";
 	connector_add_net_port(c, &ports[0], addr, strlen(addr),
 			Polarity_Getter, EndpointPolarity_Active);
 	FfiSocketAddr addr = {{127,0,0,1}, 8000};
 	connector_add_net_port(c, &ports[0], addr, Polarity_Getter, EndpointPolarity_Active);
 	rw_err_peek(c);
 	addr = "127.0.0.1:8001";
 	connector_add_net_port(c, &ports[1], addr, strlen(addr),
 			Polarity_Getter, EndpointPolarity_Active);
 	addr.port = 8001;
 	connector_add_net_port(c, &ports[1], addr, Polarity_Getter, EndpointPolarity_Active);
 	connector_add_port_pair(c, &ports[2], &ports[3]);
 	connector_add_component(c, "alt_round_merger", 16, ports, 3);
 	rw_err_peek(c);
 	// Connect with peers (5000ms timeout).
 	connector_connect(c, 5000);
 	rw_err_peek(c);
 	for(int round=0; round<3; round++) {
 		printf("----------Round %d\n", round);
 		connector_get(c, ports[3]);
 		rw_err_peek(c);
 		connector_sync(c, 1000);
 		rw_err_peek(c);
 		size_t msg_len = 0;
 		const char * msg_ptr = connector_gotten_bytes(c, ports[3], &msg_len);
 		printf("Got msg `%.*s`\n", msg_len, msg_ptr);
+	}
 	printf("Exiting\n");
 	protocol_description_destroy(pd);
 	connector_destroy(c);
 	free(pdl);
 	sleep(1.0);
 	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 {
   EndpointPolarity_Active,
   EndpointPolarity_Passive,
 } EndpointPolarity;
 typedef enum {
   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);
 #define BAD_FD -5
-ErrorCode connector_get(Connector *connector, PortId port);
+#define CC_MAP_LOCK_POISONED -3
-const uint8_t *connector_gotten_bytes(Connector *connector, PortId port, uintptr_t *out_len);
+#define CLOSE_FAIL -4
 /**
  * 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);
 #define CONNECT_FAILED -6
 /**
  * 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);
 #define ERR_OK 0
 #endif /* REOWOLF_HEADER_DEFINED */
 /* CBindgen generated */
 #define ERR_REOWOLF -1
 #ifndef REOWOLF_HEADER_DEFINED
 #define REOWOLF_HEADER_DEFINED
 #define WOULD_BLOCK -7
 #include <stdarg.h>
 #include <stdbool.h>
 #include <stdint.h>
 #include <stdlib.h>
 #define WRONG_STATE -2
 typedef enum {
   EndpointPolarity_Active,
   EndpointPolarity_Passive,
 } EndpointPolarity;
 typedef enum {
   Polarity_Putter,
   Polarity_Getter,
 } Polarity;
 typedef struct Arc_ProtocolDescription Arc_ProtocolDescription;
 typedef struct Connector Connector;
 typedef int ErrorCode;
 typedef uint32_t ConnectorId;
-typedef uint32_t PortSuffix;
+typedef uint32_t U32Suffix;
 typedef struct {
   ConnectorId connector_id;
-  PortSuffix u32_suffix;
+  U32Suffix u32_suffix;
 } Id;
 typedef Id 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);
 /**
  * Given
  * - an initialized connector in setup or connecting state,
  * - a utf-8 encoded BIND socket addresses (i.e., "local"),
  * - a utf-8 encoded CONNECT socket addresses (i.e., "peer"),
  * returns [P, G] via out pointers [putter, getter],
  * - where P is a Putter port that sends messages into the socket
  * - where G is a Getter port that recvs messages from the socket
  */
 ErrorCode connector_add_udp_port(Connector *connector,
                                  PortId *putter[2],
                                  const uint8_t *local_addr_str_ptr,
                                  uintptr_t local_addr_str_len,
                                  const uint8_t *peer_addr_str_ptr,
                                  uintptr_t peer_addr_str_len);
 /**
  * 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 */
 /* CBindgen generated */
 #ifndef REOWOLF_HEADER_DEFINED
 #define REOWOLF_HEADER_DEFINED
 #include <stdarg.h>
 #include <stdbool.h>
 #include <stdint.h>
 #include <stdlib.h>
 typedef enum {
   EndpointPolarity_Active,
   EndpointPolarity_Passive,
 } EndpointPolarity;
 typedef enum {
   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;
 } Id;
 typedef Id 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);
 /**
  * Given
  * - an initialized connector in setup or connecting state,
  * - a utf-8 encoded BIND socket addresses (i.e., "local"),
  * - a utf-8 encoded CONNECT socket addresses (i.e., "peer"),
  * returns [P, G] via out pointers [putter, getter],
  * - where P is a Putter port that sends messages into the socket
  * - where G is a Getter port that recvs messages from the socket
  */
 ErrorCode connector_add_udp_port(Connector *connector,
                                  PortId (*putter)[2],
                                  const uint8_t *local_addr_str_ptr,
                                  uintptr_t local_addr_str_len,
                                  const uint8_t *peer_addr_str_ptr,
                                  uintptr_t peer_addr_str_len);
 /**
  * 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 */
 /* CBindgen generated */
 #ifndef REOWOLF_HEADER_DEFINED
 #define REOWOLF_HEADER_DEFINED
 #include <stdarg.h>
 #include <stdbool.h>
 #include <stdint.h>
 #include <stdlib.h>
 typedef enum {
   EndpointPolarity_Active,
   EndpointPolarity_Passive,
 } EndpointPolarity;
 typedef enum {
   Polarity_Putter,
   Polarity_Getter,
 } Polarity;
 typedef struct Arc_ProtocolDescription Arc_ProtocolDescription;
 typedef struct Connector Connector;
 typedef int ErrorCode;
 typedef uint32_t ConnectorId;
 typedef uint32_t PortSuffix;
 typedef struct {
   ConnectorId connector_id;
   PortSuffix u32_suffix;
 } Id;
 typedef Id PortId;
   uint8_t ipv4[4];
   uint16_t port;
 } FfiSocketAddr;
 /**
  * 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,
+int 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,
+int connector_add_net_port(Connector *connector,
                            PortId *port,
                                  const uint8_t *addr_str_ptr,
                                  uintptr_t addr_str_len,
                            FfiSocketAddr addr,
                            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);
 /**
  * Given
  * - an initialized connector in setup or connecting state,
  * - a utf-8 encoded BIND socket addresses (i.e., "local"),
  * - a utf-8 encoded CONNECT socket addresses (i.e., "peer"),
  * returns [P, G] via out pointers [putter, getter],
  * - where P is a Putter port that sends messages into the socket
  * - where G is a Getter port that recvs messages from the socket
  */
-ErrorCode connector_add_udp_port(Connector *connector,
+int connector_add_udp_port_pair(Connector *connector,
                                 PortId *putter,
                                 PortId *getter,
                                  const uint8_t *local_addr_str_ptr,
                                  uintptr_t local_addr_str_len,
                                  const uint8_t *peer_addr_str_ptr,
                                  uintptr_t peer_addr_str_len);
                                 FfiSocketAddr local_addr,
                                 FfiSocketAddr peer_addr);
 /**
  * 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);
+int 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);
+int 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,
+int 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/ffi/mod.rs

➞

Show inline comments

 use crate::{common::*, runtime::*};
 use core::{cell::RefCell, convert::TryFrom};
 use std::os::raw::c_int;
 use std::slice::from_raw_parts as slice_from_raw_parts;
 #[cfg(feature = "ffi_socket_api")]
 pub mod socket_api;
 // #[cfg(feature = "ffi_pseudo_socket_api")]
 // pub mod pseudo_socket_api;
 // Temporary simplfication: ignore ipv6. To revert, just refactor this structure and its usages
 #[repr(C)]
 pub struct FfiSocketAddr {
     pub ipv4: [u8; 4],
     pub port: u16,
+}
 impl Into<SocketAddr> for FfiSocketAddr {
     fn into(self) -> SocketAddr {
         (self.ipv4, self.port).into()
+    }
+}
 ///////////////////////////////////////////////
 #[derive(Default)]
 struct StoredError {
     // invariant: len is zero IFF its occupied
     // contents are 1+ bytes because we also store the NULL TERMINATOR
     buf: Vec<u8>,
+}
 impl StoredError {
     const NULL_TERMINATOR: u8 = 0;
     fn clear(&mut self) {
         // no null terminator either!
         self.buf.clear();
+    }
     fn debug_store<E: Debug>(&mut self, error: &E) {
         let _ = write!(&mut self.buf, "{:?}", error);
         self.buf.push(Self::NULL_TERMINATOR);
+    }
     fn tl_debug_store<E: Debug>(error: &E) {
         STORED_ERROR.with(|stored_error| {
             let mut stored_error = stored_error.borrow_mut();
             stored_error.clear();
             stored_error.debug_store(error);
         })
+    }
     fn bytes_store(&mut self, bytes: &[u8]) {
         let _ = self.buf.write_all(bytes);
         self.buf.push(Self::NULL_TERMINATOR);
+    }
     fn tl_bytes_store(bytes: &[u8]) {
         STORED_ERROR.with(|stored_error| {
             let mut stored_error = stored_error.borrow_mut();
             stored_error.clear();
             stored_error.bytes_store(bytes);
         })
+    }
     fn tl_clear() {
         STORED_ERROR.with(|stored_error| {
             let mut stored_error = stored_error.borrow_mut();
             stored_error.clear();
         })
+    }
     fn tl_bytes_peek() -> (*const u8, usize) {
         STORED_ERROR.with(|stored_error| {
             let stored_error = stored_error.borrow();
             match stored_error.buf.len() {
 => (core::ptr::null(), 0), // no error!
                 n => {
                     // stores an error of length n-1 AND a NULL TERMINATOR
                     (stored_error.buf.as_ptr(), n - 1)
+                }
+            }
         })
+    }
+}
 thread_local! {
     static STORED_ERROR: RefCell<StoredError> = RefCell::new(StoredError::default());
+}
 unsafe fn tl_socketaddr_from_raw(
     bytes_ptr: *const u8,
     bytes_len: usize,
 ) -> Result<SocketAddr, c_int> {
     std::str::from_utf8(&*slice_from_raw_parts(bytes_ptr, bytes_len))
         .map_err(|err| {
             StoredError::tl_debug_store(&err);
             ERR_REOWOLF
         })?
         .parse()
         .map_err(|err| {
             StoredError::tl_debug_store(&err);
             ERR_REOWOLF
         })
+}
 pub const ERR_OK: c_int = 0;
 pub const ERR_REOWOLF: c_int = -1;
 pub const WRONG_STATE: c_int = -2;
-pub const FD_LOCK_POISONED: c_int = -3;
+pub const CC_MAP_LOCK_POISONED: c_int = -3;
 pub const CLOSE_FAIL: c_int = -4;
 pub const BAD_FD: c_int = -5;
 pub const CONNECT_FAILED: c_int = -6;
 pub const WOULD_BLOCK: c_int = -7;
 ///////////////////// REOWOLF //////////////////////////
 /// 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.
 #[no_mangle]
 pub unsafe extern "C" fn reowolf_error_peek(len: *mut usize) -> *const u8 {
     let (err_ptr, err_len) = StoredError::tl_bytes_peek();
     if !len.is_null() {
         len.write(err_len);
+    }
     err_ptr
+}
 ///////////////////// PROTOCOL DESCRIPTION //////////////////////////
 /// 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
 #[no_mangle]
 pub unsafe extern "C" fn protocol_description_parse(
     pdl: *const u8,
     pdl_len: usize,
 ) -> *mut Arc<ProtocolDescription> {
     StoredError::tl_clear();
     match ProtocolDescription::parse(&*slice_from_raw_parts(pdl, pdl_len)) {
         Ok(new) => Box::into_raw(Box::new(Arc::new(new))),
         Err(err) => {
             StoredError::tl_bytes_store(err.as_bytes());
             std::ptr::null_mut()
+        }
+    }
+}
 /// Destroys the given initialized protocol description and frees its resources.
 #[no_mangle]
 pub unsafe extern "C" fn protocol_description_destroy(pd: *mut Arc<ProtocolDescription>) {
     drop(Box::from_raw(pd))
+}
 /// Given an initialized protocol description, initializes `out` with a clone which can be independently created or destroyed.
 #[no_mangle]
 pub unsafe extern "C" fn protocol_description_clone(
     pd: &Arc<ProtocolDescription>,
 ) -> *mut Arc<ProtocolDescription> {
     Box::into_raw(Box::new(pd.clone()))
+}
 ///////////////////// CONNECTOR //////////////////////////
 #[no_mangle]
 pub unsafe extern "C" fn connector_new_logging(
     pd: &Arc<ProtocolDescription>,
     path_ptr: *const u8,
     path_len: usize,
 ) -> *mut Connector {
     StoredError::tl_clear();
     let path_bytes = &*slice_from_raw_parts(path_ptr, path_len);
     let path_str = match std::str::from_utf8(path_bytes) {
         Ok(path_str) => path_str,
         Err(err) => {
             StoredError::tl_debug_store(&err);
             return std::ptr::null_mut();
+        }
     };
     match std::fs::File::create(path_str) {
         Ok(file) => {
             let connector_id = Connector::random_id();
             let file_logger = Box::new(FileLogger::new(connector_id, file));
             let c = Connector::new(file_logger, pd.clone(), connector_id);
             Box::into_raw(Box::new(c))
+        }
         Err(err) => {
             StoredError::tl_debug_store(&err);
             std::ptr::null_mut()
+        }
+    }
+}
 #[no_mangle]
 pub unsafe extern "C" fn connector_print_debug(connector: &mut Connector) {
     println!("Debug print dump {:#?}", connector);
+}
 /// Initializes `out` with a new connector using the given protocol description as its configuration.
 /// The connector uses the given (internal) connector ID.
 #[no_mangle]
 pub unsafe extern "C" fn connector_new(pd: &Arc<ProtocolDescription>) -> *mut Connector {
     let c = Connector::new(Box::new(DummyLogger), pd.clone(), Connector::random_id());
     Box::into_raw(Box::new(c))
+}
 /// Destroys the given a pointer to the connector on the heap, freeing its resources.
 /// Usable in {setup, communication} states.
 #[no_mangle]
 pub unsafe extern "C" fn connector_destroy(connector: *mut Connector) {
     drop(Box::from_raw(connector))
+}
 /// 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.
 #[no_mangle]
 pub unsafe extern "C" fn connector_add_port_pair(
     connector: &mut Connector,
     out_putter: *mut PortId,
     out_getter: *mut PortId,
 ) {
     let [o, i] = connector.new_port_pair();
     out_putter.write(o);
     out_getter.write(i);
+}
 /// 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.
 #[no_mangle]
 pub unsafe extern "C" fn connector_add_component(
     connector: &mut Connector,
     ident_ptr: *const u8,
     ident_len: usize,
     ports_ptr: *const PortId,
     ports_len: usize,
 ) -> c_int {
     StoredError::tl_clear();
     match connector.add_component(
         &*slice_from_raw_parts(ident_ptr, ident_len),
         &*slice_from_raw_parts(ports_ptr, ports_len),
     ) {
         Ok(()) => ERR_OK,
         Err(err) => {
             StoredError::tl_debug_store(&err);
             ERR_REOWOLF
+        }
+    }
+}
 /// 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.
 #[no_mangle]
 pub unsafe extern "C" fn connector_add_net_port(
     connector: &mut Connector,
     port: *mut PortId,
     addr_str_ptr: *const u8,
     addr_str_len: usize,
     addr: FfiSocketAddr,
     port_polarity: Polarity,
     endpoint_polarity: EndpointPolarity,
 ) -> c_int {
     StoredError::tl_clear();
     let addr = match tl_socketaddr_from_raw(addr_str_ptr, addr_str_len) {
         Ok(local) => local,
         Err(errcode) => return errcode,
     };
     match connector.new_net_port(port_polarity, addr, endpoint_polarity) {
     match connector.new_net_port(port_polarity, addr.into(), endpoint_polarity) {
         Ok(p) => {
             if !port.is_null() {
                 port.write(p);
+            }
             ERR_OK
+        }
         Err(err) => {
             StoredError::tl_debug_store(&err);
             ERR_REOWOLF
+        }
+    }
+}
 /// Given
 /// - an initialized connector in setup or connecting state,
 /// - a utf-8 encoded BIND socket addresses (i.e., "local"),
 /// - a utf-8 encoded CONNECT socket addresses (i.e., "peer"),
 /// returns [P, G] via out pointers [putter, getter],
 /// - where P is a Putter port that sends messages into the socket
 /// - where G is a Getter port that recvs messages from the socket
 #[no_mangle]
 pub unsafe extern "C" fn connector_add_udp_port_pair(
     connector: &mut Connector,
     putter: *mut PortId,
     getter: *mut PortId,
     local_addr_str_ptr: *const u8,
     local_addr_str_len: usize,
     peer_addr_str_ptr: *const u8,
     peer_addr_str_len: usize,
     local_addr: FfiSocketAddr,
     peer_addr: FfiSocketAddr,
 ) -> c_int {
     StoredError::tl_clear();
     let local = match tl_socketaddr_from_raw(local_addr_str_ptr, local_addr_str_len) {
         Ok(local) => local,
         Err(errcode) => return errcode,
     };
     let peer = match tl_socketaddr_from_raw(peer_addr_str_ptr, peer_addr_str_len) {
         Ok(local) => local,
         Err(errcode) => return errcode,
     };
     match connector.new_udp_mediator_component(local, peer) {
     match connector.new_udp_mediator_component(local_addr.into(), peer_addr.into()) {
         Ok([p, g]) => {
             if !putter.is_null() {
                 putter.write(p);
+            }
             if !getter.is_null() {
                 getter.write(g);
+            }
             ERR_OK
+        }
         Err(err) => {
             StoredError::tl_debug_store(&err);
             ERR_REOWOLF
+        }
+    }
+}
 /// Connects this connector to the distributed system of connectors reachable through endpoints,
 /// Usable in setup state, and changes the state to communication.
 #[no_mangle]
 pub unsafe extern "C" fn connector_connect(
     connector: &mut Connector,
     timeout_millis: i64,
 ) -> c_int {
     StoredError::tl_clear();
     let option_timeout_millis: Option<u64> = TryFrom::try_from(timeout_millis).ok();
     let timeout = option_timeout_millis.map(Duration::from_millis);
     match connector.connect(timeout) {
         Ok(()) => ERR_OK,
         Err(err) => {
             StoredError::tl_debug_store(&err);
             ERR_REOWOLF
+        }
+    }
+}
 // #[no_mangle]
 // pub unsafe extern "C" fn connector_put_payload(
 //     connector: &mut Connector,
 //     port: PortId,
 //     payload: *mut Payload,
 // ) -> c_int {
 //     match connector.put(port, payload.read()) {
 //         Ok(()) => 0,
 //         Err(err) => {
 //             StoredError::tl_debug_store(&err);
 //             -1
 //         }
 //     }
 // }
 // #[no_mangle]
 // pub unsafe extern "C" fn connector_put_payload_cloning(
 //     connector: &mut Connector,
 //     port: PortId,
 //     payload: &Payload,
 // ) -> c_int {
 //     match connector.put(port, payload.clone()) {
 //         Ok(()) => 0,
 //         Err(err) => {
 //             StoredError::tl_debug_store(&err);
 //             -1
 //         }
 //     }
 // }
 /// Convenience function combining the functionalities of
 /// "payload_new" with "connector_put_payload".
 #[no_mangle]
 pub unsafe extern "C" fn connector_put_bytes(
     connector: &mut Connector,
     port: PortId,
     bytes_ptr: *const u8,
     bytes_len: usize,
 ) -> c_int {
     StoredError::tl_clear();
     let bytes = &*slice_from_raw_parts(bytes_ptr, bytes_len);
     match connector.put(port, Payload::from(bytes)) {
         Ok(()) => ERR_OK,
         Err(err) => {
             StoredError::tl_debug_store(&err);
             ERR_REOWOLF
+        }
+    }
+}
 #[no_mangle]
 pub unsafe extern "C" fn connector_get(connector: &mut Connector, port: PortId) -> c_int {
     StoredError::tl_clear();
     match connector.get(port) {
         Ok(()) => ERR_OK,
         Err(err) => {
             StoredError::tl_debug_store(&err);
             ERR_REOWOLF
+        }
+    }
+}
 #[no_mangle]
 pub unsafe extern "C" fn connector_next_batch(connector: &mut Connector) -> isize {
     StoredError::tl_clear();
     match connector.next_batch() {
         Ok(n) => n as isize,
         Err(err) => {
             StoredError::tl_debug_store(&err);
             ERR_REOWOLF as isize
+        }
+    }
+}
 #[no_mangle]
 pub unsafe extern "C" fn connector_sync(connector: &mut Connector, timeout_millis: i64) -> isize {
     StoredError::tl_clear();
     let option_timeout_millis: Option<u64> = TryFrom::try_from(timeout_millis).ok();
     let timeout = option_timeout_millis.map(Duration::from_millis);
     match connector.sync(timeout) {
         Ok(n) => n as isize,
         Err(err) => {
             StoredError::tl_debug_store(&err);
             ERR_REOWOLF as isize
+        }
+    }
+}
 #[no_mangle]
 pub unsafe extern "C" fn connector_gotten_bytes(
     connector: &mut Connector,
     port: PortId,
     out_len: *mut usize,
 ) -> *const u8 {
     StoredError::tl_clear();
     match connector.gotten(port) {
         Ok(payload_borrow) => {
             let slice = payload_borrow.as_slice();
             if !out_len.is_null() {
                 out_len.write(slice.len());
+            }
             slice.as_ptr()
+        }
         Err(err) => {
             StoredError::tl_debug_store(&err);
             std::ptr::null()
+        }
+    }
+}
 // #[no_mangle]
 // unsafe extern "C" fn connector_gotten_payload(
 //     connector: &mut Connector,
 //     port: PortId,
 // ) -> *const Payload {
 //     StoredError::tl_clear();
 //     match connector.gotten(port) {
 //         Ok(payload_borrow) => payload_borrow,
 //         Err(err) => {
 //             StoredError::tl_debug_store(&err);
 //             std::ptr::null()
 //         }
 //     }
 // }
 ///////////////////// PAYLOAD //////////////////////////
 // #[no_mangle]
 // unsafe extern "C" fn payload_new(
 //     bytes_ptr: *const u8,
 //     bytes_len: usize,
 //     out_payload: *mut Payload,
 // ) {
 //     let bytes: &[u8] = &*slice_from_raw_parts(bytes_ptr, bytes_len);
 //     out_payload.write(Payload::from(bytes));
 // }
 // #[no_mangle]
 // unsafe extern "C" fn payload_destroy(payload: *mut Payload) {
 //     drop(Box::from_raw(payload))
 // }
 // #[no_mangle]
 // unsafe extern "C" fn payload_clone(payload: &Payload, out_payload: *mut Payload) {
 //     out_payload.write(payload.clone())
 // }
 // #[no_mangle]
 // unsafe extern "C" fn payload_peek_bytes(payload: &Payload, bytes_len: *mut usize) -> *const u8 {
 //     let slice = payload.as_slice();
 //     bytes_len.write(slice.len());
 //     slice.as_ptr()
 // }

src/ffi/pseudo_socket_api.rs

➞

Show inline comments

@@ file renamed from src/ffi/socket_api.rs to src/ffi/pseudo_socket_api.rs @@
 use super::*;
 use std::{
     collections::HashMap,
     ffi::c_void,
     net::{Ipv4Addr, SocketAddr, SocketAddrV4},
     os::raw::c_int,
     sync::RwLock,
 };
 ///////////////////////////////////////////////////////////////////
 struct FdAllocator {
     next: Option<c_int>,
     freed: Vec<c_int>,
+}
 struct ConnectorBound {
     connector: Connector,
     is_nonblocking: bool,
     putter: PortId,
     getter: PortId,
+}
-struct MaybeConnector {
+struct ConnectorComplex {
     // invariants:
     // 1. connector is a upd-socket singleton
     // 2. putter and getter are ports in the native interface with the appropriate polarities
     // 3. peer_addr always mirrors connector's single udp socket's connect addr. both are overwritten together.
     peer_addr: SocketAddr,
     connector_bound: Option<ConnectorBound>,
+}
 #[derive(Default)]
 struct FdcStorage {
     fd_to_c: HashMap<c_int, RwLock<MaybeConnector>>,
 struct CcMap {
     fd_to_cc: HashMap<c_int, RwLock<ConnectorComplex>>,
     fd_allocator: FdAllocator,
+}
 fn trivial_peer_addr() -> SocketAddr {
     // SocketAddrV4::new isn't a constant-time func
     SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::new(0, 0, 0, 0), 0))
+}
 ///////////////////////////////////////////////////////////////////
 impl Default for FdAllocator {
     fn default() -> Self {
         Self {
             next: Some(0), // positive values used only
             freed: vec![],
+        }
+    }
+}
 impl FdAllocator {
     fn alloc(&mut self) -> c_int {
         if let Some(fd) = self.freed.pop() {
             return fd;
+        }
         if let Some(fd) = self.next {
             self.next = fd.checked_add(1);
             return fd;
+        }
         panic!("No more Connector FDs to allocate!")
+    }
     fn free(&mut self, fd: c_int) {
         self.freed.push(fd);
+    }
+}
 lazy_static::lazy_static! {
-    static ref FDC_STORAGE: RwLock<FdcStorage> = Default::default();
+    static ref CC_MAP: RwLock<CcMap> = Default::default();
+}
-impl MaybeConnector {
+impl ConnectorComplex {
     fn connect(&mut self, peer_addr: SocketAddr) -> c_int {
         self.peer_addr = peer_addr;
         if let Some(ConnectorBound { connector, .. }) = &mut self.connector_bound {
             if connector.get_mut_udp_sock(0).unwrap().connect(peer_addr).is_err() {
                 return CONNECT_FAILED;
+            }
+        }
         ERR_OK
+    }
     unsafe fn send(&mut self, bytes_ptr: *const c_void, bytes_len: usize) -> isize {
         if let Some(ConnectorBound { connector, putter, .. }) = &mut self.connector_bound {
             match connector_put_bytes(connector, *putter, bytes_ptr as _, bytes_len) {
                 ERR_OK => connector_sync(connector, -1),
                 err => err as isize,
+            }
         } else {
             WRONG_STATE as isize // not bound!
+        }
+    }
     unsafe fn recv(&mut self, bytes_ptr: *const c_void, bytes_len: usize) -> isize {
         if let Some(ConnectorBound { connector, getter, .. }) = &mut self.connector_bound {
             connector_get(connector, *getter);
             match connector_sync(connector, -1) {
 => {
                     // batch index 0 means OK
                     let slice = connector.gotten(*getter).unwrap().as_slice();
                     let copied_bytes = slice.len().min(bytes_len);
                     std::ptr::copy_nonoverlapping(
                         slice.as_ptr(),
                         bytes_ptr as *mut u8,
                         copied_bytes,
                     );
                     copied_bytes as isize
+                }
                 err => return err as isize,
+            }
         } else {
             WRONG_STATE as isize // not bound!
+        }
+    }
+}
 ///////////////////////////////////////////////////////////////////
 #[no_mangle]
 pub extern "C" fn rw_socket(_domain: c_int, _type: c_int) -> c_int {
     // ignoring domain and type
-    let mut w = if let Ok(w) = FDC_STORAGE.write() { w } else { return FD_LOCK_POISONED };
+    let mut w = if let Ok(w) = CC_MAP.write() { w } else { return CC_MAP_LOCK_POISONED };
     let fd = w.fd_allocator.alloc();
     let mc = MaybeConnector { peer_addr: trivial_peer_addr(), connector_bound: None };
     w.fd_to_c.insert(fd, RwLock::new(mc));
     let cc = ConnectorComplex { peer_addr: trivial_peer_addr(), connector_bound: None };
     w.fd_to_cc.insert(fd, RwLock::new(cc));
     fd
+}
 #[no_mangle]
 pub extern "C" fn rw_close(fd: c_int, _how: c_int) -> c_int {
     // ignoring HOW
     let mut w = if let Ok(w) = FDC_STORAGE.write() { w } else { return FD_LOCK_POISONED };
     if w.fd_to_c.remove(&fd).is_some() {
     let mut w = if let Ok(w) = CC_MAP.write() { w } else { return CC_MAP_LOCK_POISONED };
     if w.fd_to_cc.remove(&fd).is_some() {
         w.fd_allocator.free(fd);
         ERR_OK
     } else {
         CLOSE_FAIL
+    }
+}
 #[no_mangle]
 pub unsafe extern "C" fn rw_bind(
     fd: c_int,
     local_addr: *const SocketAddr,
     _addr_len: usize,
 ) -> c_int {
     // assuming _domain is AF_INET and _type is SOCK_DGRAM
     let r = if let Ok(r) = FDC_STORAGE.read() { r } else { return FD_LOCK_POISONED };
     let mc = if let Some(mc) = r.fd_to_c.get(&fd) { mc } else { return BAD_FD };
     let mut mc = if let Ok(mc) = mc.write() { mc } else { return FD_LOCK_POISONED };
     let mc: &mut MaybeConnector = &mut mc;
     if mc.connector_bound.is_some() {
     let r = if let Ok(r) = CC_MAP.read() { r } else { return CC_MAP_LOCK_POISONED };
     let cc = if let Some(cc) = r.fd_to_cc.get(&fd) { cc } else { return BAD_FD };
     let mut cc = if let Ok(cc) = cc.write() { cc } else { return CC_MAP_LOCK_POISONED };
     let cc: &mut ConnectorComplex = &mut cc;
     if cc.connector_bound.is_some() {
         return WRONG_STATE;
+    }
-    mc.connector_bound = {
+    cc.connector_bound = {
         let mut connector = Connector::new(
             Box::new(crate::DummyLogger),
             crate::TRIVIAL_PD.clone(),
             Connector::random_id(),
         );
         let [putter, getter] =
             connector.new_udp_mediator_component(local_addr.read(), mc.peer_addr).unwrap();
         Some(ConnectorBound { connector, putter, getter })
             connector.new_udp_mediator_component(local_addr.read(), cc.peer_addr).unwrap();
         Some(ConnectorBound { connector, putter, getter, is_nonblocking: false })
     };
     ERR_OK
+}
 #[no_mangle]
 pub unsafe extern "C" fn rw_connect(
     fd: c_int,
     peer_addr: *const SocketAddr,
     _address_len: usize,
 ) -> c_int {
     // assuming _domain is AF_INET and _type is SOCK_DGRAM
     let r = if let Ok(r) = FDC_STORAGE.read() { r } else { return FD_LOCK_POISONED };
     let mc = if let Some(mc) = r.fd_to_c.get(&fd) { mc } else { return BAD_FD };
     let mut mc = if let Ok(mc) = mc.write() { mc } else { return FD_LOCK_POISONED };
     let mc: &mut MaybeConnector = &mut mc;
     mc.connect(peer_addr.read())
     let r = if let Ok(r) = CC_MAP.read() { r } else { return CC_MAP_LOCK_POISONED };
     let cc = if let Some(cc) = r.fd_to_cc.get(&fd) { cc } else { return BAD_FD };
     let mut cc = if let Ok(cc) = cc.write() { cc } else { return CC_MAP_LOCK_POISONED };
     let cc: &mut ConnectorComplex = &mut cc;
     cc.connect(peer_addr.read())
+}
 #[no_mangle]
 pub unsafe extern "C" fn rw_send(
     fd: c_int,
     bytes_ptr: *const c_void,
     bytes_len: usize,
     _flags: c_int,
 ) -> isize {
     // ignoring flags
     let r = if let Ok(r) = FDC_STORAGE.read() { r } else { return FD_LOCK_POISONED as isize };
     let mc = if let Some(mc) = r.fd_to_c.get(&fd) { mc } else { return BAD_FD as isize };
     let mut mc = if let Ok(mc) = mc.write() { mc } else { return FD_LOCK_POISONED as isize };
     let mc: &mut MaybeConnector = &mut mc;
     mc.send(bytes_ptr, bytes_len)
     let r = if let Ok(r) = CC_MAP.read() { r } else { return CC_MAP_LOCK_POISONED as isize };
     let cc = if let Some(cc) = r.fd_to_cc.get(&fd) { cc } else { return BAD_FD as isize };
     let mut cc = if let Ok(cc) = cc.write() { cc } else { return CC_MAP_LOCK_POISONED as isize };
     let cc: &mut ConnectorComplex = &mut cc;
     cc.send(bytes_ptr, bytes_len)
+}
 #[no_mangle]
 pub unsafe extern "C" fn rw_recv(
     fd: c_int,
     bytes_ptr: *mut c_void,
     bytes_len: usize,
     _flags: c_int,
 ) -> isize {
     // ignoring flags
     let r = if let Ok(r) = FDC_STORAGE.read() { r } else { return FD_LOCK_POISONED as isize };
     let mc = if let Some(mc) = r.fd_to_c.get(&fd) { mc } else { return BAD_FD as isize };
     let mut mc = if let Ok(mc) = mc.write() { mc } else { return FD_LOCK_POISONED as isize };
     let mc: &mut MaybeConnector = &mut mc;
     mc.recv(bytes_ptr, bytes_len)
     let r = if let Ok(r) = CC_MAP.read() { r } else { return CC_MAP_LOCK_POISONED as isize };
     let cc = if let Some(cc) = r.fd_to_cc.get(&fd) { cc } else { return BAD_FD as isize };
     let mut cc = if let Ok(cc) = cc.write() { cc } else { return CC_MAP_LOCK_POISONED as isize };
     let cc: &mut ConnectorComplex = &mut cc;
     cc.recv(bytes_ptr, bytes_len)
+}
 #[no_mangle]
 pub unsafe extern "C" fn rw_sendto(
     fd: c_int,
     bytes_ptr: *mut c_void,
     bytes_len: usize,
     _flags: c_int,
     peer_addr: *const SocketAddr,
     _addr_len: usize,
 ) -> isize {
     let r = if let Ok(r) = FDC_STORAGE.read() { r } else { return FD_LOCK_POISONED as isize };
     let mc = if let Some(mc) = r.fd_to_c.get(&fd) { mc } else { return BAD_FD as isize };
     let mut mc = if let Ok(mc) = mc.write() { mc } else { return FD_LOCK_POISONED as isize };
     let mc: &mut MaybeConnector = &mut mc;
     let r = if let Ok(r) = CC_MAP.read() { r } else { return CC_MAP_LOCK_POISONED as isize };
     let cc = if let Some(cc) = r.fd_to_cc.get(&fd) { cc } else { return BAD_FD as isize };
     let mut cc = if let Ok(cc) = cc.write() { cc } else { return CC_MAP_LOCK_POISONED as isize };
     let cc: &mut ConnectorComplex = &mut cc;
     // copy currently connected peer addr
-    let connected = mc.peer_addr;
+    let connected = cc.peer_addr;
     // connect to given peer_addr
-    match mc.connect(peer_addr.read()) {
+    match cc.connect(peer_addr.read()) {
         e if e != ERR_OK => return e as isize,
         _ => {}
+    }
     // send
-    let ret = mc.send(bytes_ptr, bytes_len);
+    let ret = cc.send(bytes_ptr, bytes_len);
     // restore connected peer addr
-    match mc.connect(connected) {
+    match cc.connect(connected) {
         e if e != ERR_OK => return e as isize,
         _ => {}
+    }
     ret
+}
 #[no_mangle]
 #[no_mangle]
 pub unsafe extern "C" fn rw_recvfrom(
     fd: c_int,
     bytes_ptr: *mut c_void,
     bytes_len: usize,
     _flags: c_int,
     peer_addr: *const SocketAddr,
     _addr_len: usize,
 ) -> isize {
     let r = if let Ok(r) = FDC_STORAGE.read() { r } else { return FD_LOCK_POISONED as isize };
     let mc = if let Some(mc) = r.fd_to_c.get(&fd) { mc } else { return BAD_FD as isize };
     let mut mc = if let Ok(mc) = mc.write() { mc } else { return FD_LOCK_POISONED as isize };
     let mc: &mut MaybeConnector = &mut mc;
     let r = if let Ok(r) = CC_MAP.read() { r } else { return CC_MAP_LOCK_POISONED as isize };
     let cc = if let Some(cc) = r.fd_to_cc.get(&fd) { cc } else { return BAD_FD as isize };
     let mut cc = if let Ok(cc) = cc.write() { cc } else { return CC_MAP_LOCK_POISONED as isize };
     let cc: &mut ConnectorComplex = &mut cc;
     // copy currently connected peer addr
-    let connected = mc.peer_addr;
+    let connected = cc.peer_addr;
     // connect to given peer_addr
-    match mc.connect(peer_addr.read()) {
+    match cc.connect(peer_addr.read()) {
         e if e != ERR_OK => return e as isize,
         _ => {}
+    }
     // send
-    let ret = mc.send(bytes_ptr, bytes_len);
+    let ret = cc.send(bytes_ptr, bytes_len);
     // restore connected peer addr
-    match mc.connect(connected) {
+    match cc.connect(connected) {
         e if e != ERR_OK => return e as isize,
         _ => {}
+    }
     ret
+}

src/runtime/communication.rs

➞

Show inline comments

 use super::*;
 use crate::common::*;
 use core::ops::{Deref, DerefMut};
 ////////////////
 // Guard protecting an incrementally unfoldable slice of MapTempGuard elements
 struct MapTempsGuard<'a, K, V>(&'a mut [HashMap<K, V>]);
 // Type protecting a temporary map; At the start and end of the Guard's lifetime, self.0.is_empty() must be true
 struct MapTempGuard<'a, K, V>(&'a mut HashMap<K, V>);
 #[derive(Default)]
 struct GetterBuffer {
     getters_and_sends: Vec<(PortId, SendPayloadMsg)>,
+}
 struct RoundCtx {
     solution_storage: SolutionStorage,
     spec_var_stream: SpecVarStream,
     getter_buffer: GetterBuffer,
     deadline: Option<Instant>,
+}
 struct BranchingNative {
     branches: HashMap<Predicate, NativeBranch>,
+}
 #[derive(Clone, Debug)]
 struct NativeBranch {
     index: usize,
     gotten: HashMap<PortId, Payload>,
     to_get: HashSet<PortId>,
+}
 #[derive(Debug)]
 struct SolutionStorage {
     old_local: HashSet<Predicate>,
     new_local: HashSet<Predicate>,
     // this pair acts as SubtreeId -> HashSet<Predicate> which is friendlier to iteration
     subtree_solutions: Vec<HashSet<Predicate>>,
     subtree_id_to_index: HashMap<SubtreeId, usize>,
+}
 #[derive(Debug)]
 struct BranchingProtoComponent {
     ports: HashSet<PortId>,
     branches: HashMap<Predicate, ProtoComponentBranch>,
+}
 #[derive(Debug, Clone)]
 struct ProtoComponentBranch {
     state: ComponentState,
     inner: ProtoComponentBranchInner,
     ended: bool,
+}
 struct CyclicDrainer<'a, K: Eq + Hash, V> {
     input: &'a mut HashMap<K, V>,
     inner: CyclicDrainInner<'a, K, V>,
+}
 struct CyclicDrainInner<'a, K: Eq + Hash, V> {
     swap: &'a mut HashMap<K, V>,
     output: &'a mut HashMap<K, V>,
+}
 trait ReplaceBoolTrue {
     fn replace_with_true(&mut self) -> bool;
+}
 impl ReplaceBoolTrue for bool {
     fn replace_with_true(&mut self) -> bool {
         let was = *self;
         *self = true;
         !was
+    }
+}
 ////////////////
 impl<'a, K, V> MapTempsGuard<'a, K, V> {
     fn reborrow(&mut self) -> MapTempsGuard<'_, K, V> {
         MapTempsGuard(self.0)
+    }
     fn split_first_mut(self) -> (MapTempGuard<'a, K, V>, MapTempsGuard<'a, K, V>) {
         let (head, tail) = self.0.split_first_mut().expect("Cache exhausted");
         (MapTempGuard::new(head), MapTempsGuard(tail))
+    }
+}
 impl<'a, K, V> MapTempGuard<'a, K, V> {
     fn new(map: &'a mut HashMap<K, V>) -> Self {
         assert!(map.is_empty()); // sanity check
         Self(map)
+    }
+}
 impl<'a, K, V> Drop for MapTempGuard<'a, K, V> {
     fn drop(&mut self) {
         assert!(self.0.is_empty()); // sanity check
+    }
+}
 impl<'a, K, V> Deref for MapTempGuard<'a, K, V> {
     type Target = HashMap<K, V>;
     fn deref(&self) -> &<Self as Deref>::Target {
         self.0
+    }
+}
 impl<'a, K, V> DerefMut for MapTempGuard<'a, K, V> {
     fn deref_mut(&mut self) -> &mut <Self as Deref>::Target {
         self.0
+    }
+}
 impl RoundCtxTrait for RoundCtx {
     fn get_deadline(&self) -> &Option<Instant> {
         &self.deadline
+    }
     fn getter_add(&mut self, getter: PortId, msg: SendPayloadMsg) {
         self.getter_buffer.getter_add(getter, msg)
+    }
+}
 impl Connector {
     fn get_comm_mut(&mut self) -> Option<&mut ConnectorCommunication> {
         if let ConnectorPhased::Communication(comm) = &mut self.phased {
             Some(comm)
         } else {
             None
+        }
+    }
     // #[cfg(ffi_socket_api)]
     pub(crate) fn get_mut_udp_sock(&mut self, index: usize) -> Option<&mut UdpSocket> {
         let sock = &mut self
             .get_comm_mut()?
             .endpoint_manager
             .udp_endpoint_store
             .endpoint_exts
             .get_mut(index)?
             .sock;
         Some(sock)
+    }
     pub fn gotten(&mut self, port: PortId) -> Result<&Payload, GottenError> {
         use GottenError as Ge;
         let comm = self.get_comm_mut().ok_or(Ge::NoPreviousRound)?;
         match &comm.round_result {
             Err(_) => Err(Ge::PreviousSyncFailed),
             Ok(None) => Err(Ge::NoPreviousRound),
             Ok(Some(round_ok)) => round_ok.gotten.get(&port).ok_or(Ge::PortDidntGet),
+        }
+    }
     pub fn next_batch(&mut self) -> Result<usize, WrongStateError> {
         // returns index of new batch
         let comm = self.get_comm_mut().ok_or(WrongStateError)?;
         comm.native_batches.push(Default::default());
         Ok(comm.native_batches.len() - 1)
+    }
     fn port_op_access(
         &mut self,
         port: PortId,
         expect_polarity: Polarity,
     ) -> Result<&mut NativeBatch, PortOpError> {
         use PortOpError as Poe;
         let Self { unphased: cu, phased } = self;
         if !cu.inner.native_ports.contains(&port) {
             return Err(Poe::PortUnavailable);
+        }
         match cu.inner.port_info.polarities.get(&port) {
             Some(p) if *p == expect_polarity => {}
             Some(_) => return Err(Poe::WrongPolarity),
             None => return Err(Poe::UnknownPolarity),
+        }
         match phased {
             ConnectorPhased::Setup { .. } => Err(Poe::NotConnected),
             ConnectorPhased::Communication(comm) => {
                 let batch = comm.native_batches.last_mut().unwrap(); // length >= 1 is invariant
                 Ok(batch)
+            }
+        }
+    }
     pub fn put(&mut self, port: PortId, payload: Payload) -> Result<(), PortOpError> {
         use PortOpError as Poe;
         let batch = self.port_op_access(port, Putter)?;
         if batch.to_put.contains_key(&port) {
             Err(Poe::MultipleOpsOnPort)
         } else {
             batch.to_put.insert(port, payload);
             Ok(())
+        }
+    }
     pub fn get(&mut self, port: PortId) -> Result<(), PortOpError> {
         use PortOpError as Poe;
         let batch = self.port_op_access(port, Getter)?;
         if batch.to_get.insert(port) {
             Ok(())
         } else {
             Err(Poe::MultipleOpsOnPort)
+        }
+    }
     // entrypoint for caller. overwrites round result enum, and returns what happened
     pub fn sync(&mut self, timeout: Option<Duration>) -> Result<usize, SyncError> {
         let Self { unphased: cu, phased } = self;
         match phased {
             ConnectorPhased::Setup { .. } => Err(SyncError::NotConnected),
             ConnectorPhased::Communication(comm) => {
                 match &comm.round_result {
                     Err(SyncError::Unrecoverable(e)) => {
                         log!(cu.inner.logger, "Attempted to start sync round, but previous error {:?} was unrecoverable!", e);
                         return Err(SyncError::Unrecoverable(e.clone()));
+                    }
                     _ => {}
+                }
                 comm.round_result = Self::connected_sync(cu, comm, timeout);
                 comm.round_index += 1;
                 match &comm.round_result {
                     Ok(None) => unreachable!(),
                     Ok(Some(ok_result)) => Ok(ok_result.batch_index),
                     Err(sync_error) => Err(sync_error.clone()),
+                }
+            }
+        }
+    }
     // private function. mutates state but returns with round
     // result ASAP (allows for convenient error return with ?)
     fn connected_sync(
         cu: &mut ConnectorUnphased,
         comm: &mut ConnectorCommunication,
         timeout: Option<Duration>,
     ) -> Result<Option<RoundOk>, SyncError> {
         //////////////////////////////////
         use SyncError as Se;
         //////////////////////////////////
         log!(
             cu.inner.logger,
             "~~~ SYNC called with timeout {:?}; starting round {}",
             &timeout,
             comm.round_index
         );
         // 1. run all proto components to Nonsync blockers
         // NOTE: original components are immutable until Decision::Success
         let mut branching_proto_components =
             HashMap::<ProtoComponentId, BranchingProtoComponent>::default();
         let mut unrun_components: Vec<(ProtoComponentId, ProtoComponent)> =
             cu.proto_components.iter().map(|(&k, v)| (k, v.clone())).collect();
         log!(cu.inner.logger, "Nonsync running {} proto components...", unrun_components.len());
         // drains unrun_components, and populates branching_proto_components.
         while let Some((proto_component_id, mut component)) = unrun_components.pop() {
             // TODO coalesce fields
             log!(
                 cu.inner.logger,
                 "Nonsync running proto component with ID {:?}. {} to go after this",
                 proto_component_id,
                 unrun_components.len()
             );
             let mut ctx = NonsyncProtoContext {
                 cu_inner: &mut cu.inner,
                 proto_component_id,
                 unrun_components: &mut unrun_components,
                 proto_component_ports: &mut cu
                     .proto_components
                     .get_mut(&proto_component_id)
                     .unwrap() // unrun_components' keys originate from proto_components
                     .ports,
             };
             let blocker = component.state.nonsync_run(&mut ctx, &cu.proto_description);
             log!(
                 cu.inner.logger,
                 "proto component {:?} ran to nonsync blocker {:?}",
                 proto_component_id,
                 &blocker
             );
             use NonsyncBlocker as B;
             match blocker {
                 B::ComponentExit => drop(component),
                 B::Inconsistent => return Err(Se::InconsistentProtoComponent(proto_component_id)),
                 B::SyncBlockStart => {
                     branching_proto_components
                         .insert(proto_component_id, BranchingProtoComponent::initial(component));
+                }
+            }
+        }
         log!(
             cu.inner.logger,
             "All {} proto components are now done with Nonsync phase",
             branching_proto_components.len(),
         );
         // Create temp structures needed for the synchronous phase of the round
         let mut rctx = RoundCtx {
             solution_storage: {
                 let n = std::iter::once(SubtreeId::LocalComponent(ComponentId::Native));
                 let c = cu
                     .proto_components
                     .keys()
                     .map(|&id| SubtreeId::LocalComponent(ComponentId::Proto(id)));
                 let e = comm
                     .neighborhood
                     .children
                     .iter()
                     .map(|&index| SubtreeId::NetEndpoint { index });
                 let subtree_id_iter = n.chain(c).chain(e);
                 log!(
                     cu.inner.logger,
                     "Children in subtree are: {:?}",
                     subtree_id_iter.clone().collect::<Vec<_>>()
                 );
                 SolutionStorage::new(subtree_id_iter)
             },
             spec_var_stream: cu.inner.id_manager.new_spec_var_stream(),
             getter_buffer: Default::default(),
             deadline: timeout.map(|to| Instant::now() + to),
         };
         log!(cu.inner.logger, "Round context structure initialized");
         // Explore all native branches eagerly. Find solutions, buffer messages, etc.
         log!(
             cu.inner.logger,
             "Translating {} native batches into branches...",
             comm.native_batches.len()
         );
         let native_spec_var = rctx.spec_var_stream.next();
         log!(cu.inner.logger, "Native branch spec var is {:?}", native_spec_var);
         let mut branching_native = BranchingNative { branches: Default::default() };
         'native_branches: for ((native_branch, index), branch_spec_val) in
             comm.native_batches.drain(..).zip(0..).zip(SpecVal::iter_domain())
+        {
             let NativeBatch { to_get, to_put } = native_branch;
             let predicate = {
                 let mut predicate = Predicate::default();
                 // assign trues for ports that fire
                 let firing_ports: HashSet<PortId> =
                     to_get.iter().chain(to_put.keys()).copied().collect();
                 for &port in to_get.iter().chain(to_put.keys()) {
                     let var = cu.inner.port_info.spec_var_for(port);
                     predicate.assigned.insert(var, SpecVal::FIRING);
+                }
                 // assign falses for all silent (not firing) ports
                 for &port in cu.inner.native_ports.difference(&firing_ports) {
                     let var = cu.inner.port_info.spec_var_for(port);
                     if let Some(SpecVal::FIRING) = predicate.assigned.insert(var, SpecVal::SILENT) {
                         log!(cu.inner.logger, "Native branch index={} contains internal inconsistency wrt. {:?}. Skipping", index, var);
                         continue 'native_branches;
+                    }
+                }
                 // this branch is consistent. distinguish it with a unique var:val mapping and proceed
                 predicate.inserted(native_spec_var, branch_spec_val)
             };
             log!(
                 cu.inner.logger,
                 "Native branch index={:?} has consistent {:?}",
                 index,
                 &predicate
             );
             // send all outgoing messages (by buffering them)
             for (putter, payload) in to_put {
                 let msg = SendPayloadMsg { predicate: predicate.clone(), payload };
                 log!(cu.inner.logger, "Native branch {} sending msg {:?}", index, &msg);
                 rctx.getter_buffer.putter_add(cu, putter, msg);
+            }
             let branch = NativeBranch { index, gotten: Default::default(), to_get };
             if branch.is_ended() {
                 log!(
                     cu.inner.logger,
                     "Native submitting solution for batch {} with {:?}",
                     index,
                     &predicate
                 );
                 rctx.solution_storage.submit_and_digest_subtree_solution(
                     &mut *cu.inner.logger,
                     SubtreeId::LocalComponent(ComponentId::Native),
                     predicate.clone(),
                 );
+            }
             if let Some(_) = branching_native.branches.insert(predicate, branch) {
                 // thanks to the native_spec_var, each batch has a distinct predicate
                 unreachable!()
+            }
+        }
         // restore the invariant: !native_batches.is_empty()
         comm.native_batches.push(Default::default());
         comm.endpoint_manager
             .udp_endpoints_round_start(&mut *cu.inner.logger, &mut rctx.spec_var_stream);
         comm.endpoint_manager.udp_endpoints_round_start(&mut *cu.inner.logger);
         // Call to another big method; keep running this round until a distributed decision is reached
         let decision = Self::sync_reach_decision(
             cu,
             comm,
             &mut branching_native,
             &mut branching_proto_components,
             &mut rctx,
         )?;
         log!(cu.inner.logger, "Committing to decision {:?}!", &decision);
         comm.endpoint_manager.udp_endpoints_round_end(&mut *cu.inner.logger, &decision)?;
         // propagate the decision to children
         let msg = Msg::CommMsg(CommMsg {
             round_index: comm.round_index,
             contents: CommMsgContents::CommCtrl(CommCtrlMsg::Announce {
                 decision: decision.clone(),
             }),
         });
         log!(
             cu.inner.logger,
             "Announcing decision {:?} through child endpoints {:?}",
             &msg,
             &comm.neighborhood.children
         );
         for &child in comm.neighborhood.children.iter() {
             comm.endpoint_manager.send_to_comms(child, &msg)?;
+        }
         let ret = match decision {
             Decision::Failure => {
                 // dropping {branching_proto_components, branching_native}
                 Err(Se::RoundFailure)
+            }
             Decision::Success(predicate) => {
                 // commit changes to component states
                 cu.proto_components.clear();
                 cu.proto_components.extend(
                     // consume branching proto components
                     branching_proto_components
                         .into_iter()
                         .map(|(id, bpc)| (id, bpc.collapse_with(&predicate))),
                 );
                 log!(
                     cu.inner.logger,
                     "End round with (updated) component states {:?}",
                     cu.proto_components.keys()
                 );
                 // consume native
                 Ok(Some(branching_native.collapse_with(&mut *cu.inner.logger, &predicate)))
+            }
         };
         log!(cu.inner.logger, "Sync round ending! Cleaning up");
         ret
+    }
     fn sync_reach_decision(
         cu: &mut ConnectorUnphased,
         comm: &mut ConnectorCommunication,
         branching_native: &mut BranchingNative,
         branching_proto_components: &mut HashMap<ProtoComponentId, BranchingProtoComponent>,
         rctx: &mut RoundCtx,
     ) -> Result<Decision, UnrecoverableSyncError> {
         let mut already_requested_failure = false;
         if branching_native.branches.is_empty() {
             log!(cu.inner.logger, "Native starts with no branches! Failure!");
             match comm.neighborhood.parent {
                 Some(parent) => {
                     if already_requested_failure.replace_with_true() {
                         Self::request_failure(cu, comm, parent)?
                     } else {
                         log!(cu.inner.logger, "Already requested failure");
+                    }
+                }
                 None => {
                     log!(cu.inner.logger, "No parent. Deciding on failure");
                     return Ok(Decision::Failure);
+                }
+            }
+        }
         log!(cu.inner.logger, "Done translating native batches into branches");
         let mut pcb_temps_owner = <[HashMap<Predicate, ProtoComponentBranch>; 3]>::default();
         let mut pcb_temps = MapTempsGuard(&mut pcb_temps_owner);
         let mut bn_temp_owner = <HashMap<Predicate, NativeBranch>>::default();
         // run all proto components to their sync blocker
         log!(
             cu.inner.logger,
             "Running all {} proto components to their sync blocker...",
             branching_proto_components.len()
         );
         for (&proto_component_id, proto_component) in branching_proto_components.iter_mut() {
             let BranchingProtoComponent { ports, branches } = proto_component;
             // must reborrow to constrain the lifetime of pcb_temps to inside the loop
             let (swap, pcb_temps) = pcb_temps.reborrow().split_first_mut();
             let (blocked, _pcb_temps) = pcb_temps.split_first_mut();
             // initially, no components have .ended==true
             // drain from branches --> blocked
             let cd = CyclicDrainer::new(branches, swap.0, blocked.0);
             BranchingProtoComponent::drain_branches_to_blocked(
                 cd,
                 cu,
                 rctx,
                 proto_component_id,
                 ports,
             )?;
             // swap the blocked branches back
             std::mem::swap(blocked.0, branches);
             if branches.is_empty() {
                 log!(cu.inner.logger, "{:?} has become inconsistent!", proto_component_id);
                 if let Some(parent) = comm.neighborhood.parent {
                     if already_requested_failure.replace_with_true() {
                         Self::request_failure(cu, comm, parent)?
                     } else {
                         log!(cu.inner.logger, "Already requested failure");
+                    }
                 } else {
                     log!(cu.inner.logger, "As the leader, deciding on timeout");
                     return Ok(Decision::Failure);
+                }
+            }
+        }
         log!(cu.inner.logger, "All proto components are blocked");
         log!(cu.inner.logger, "Entering decision loop...");
         comm.endpoint_manager.undelay_all();
         'undecided: loop {
             // drain payloads_to_get, sending them through endpoints / feeding them to components
             log!(
                 cu.inner.logger,
                 "Decision loop! have {} messages to recv",
                 rctx.getter_buffer.len()
             );
             while let Some((getter, send_payload_msg)) = rctx.getter_buffer.pop() {
                 assert!(cu.inner.port_info.polarities.get(&getter).copied() == Some(Getter));
                 let route = cu.inner.port_info.routes.get(&getter);
                 log!(
                     cu.inner.logger,
                     "Routing msg {:?} to {:?} via {:?}",
                     &send_payload_msg,
                     getter,
                     &route
                 );
                 match route {
                     None => log!(cu.inner.logger, "Delivery failed. Physical route unmapped!"),
                     Some(Route::UdpEndpoint { index }) => {
                         let udp_endpoint_ext =
                             &mut comm.endpoint_manager.udp_endpoint_store.endpoint_exts[*index];
                         let SendPayloadMsg { predicate, payload } = send_payload_msg;
                         log!(cu.inner.logger, "Delivering to udp endpoint index={}", index);
                         udp_endpoint_ext.outgoing_payloads.insert(predicate, payload);
+                    }
                     Some(Route::NetEndpoint { index }) => {
                         let msg = Msg::CommMsg(CommMsg {
                             round_index: comm.round_index,
                             contents: CommMsgContents::SendPayload(send_payload_msg),
                         });
                         comm.endpoint_manager.send_to_comms(*index, &msg)?;
+                    }
                     Some(Route::LocalComponent(ComponentId::Native)) => branching_native.feed_msg(
                         cu,
                         rctx,
                         getter,
                         &send_payload_msg,
                         MapTempGuard::new(&mut bn_temp_owner),
                     ),
                     Some(Route::LocalComponent(ComponentId::Proto(proto_component_id))) => {
                         if let Some(branching_component) =
                             branching_proto_components.get_mut(proto_component_id)
+                        {
                             let proto_component_id = *proto_component_id;
                             branching_component.feed_msg(
                                 cu,
                                 rctx,
                                 proto_component_id,
                                 getter,
                                 &send_payload_msg,
                                 pcb_temps.reborrow(),
                             )?;
                             if branching_component.branches.is_empty() {
                                 log!(
                                     cu.inner.logger,
                                     "{:?} has become inconsistent!",
                                     proto_component_id
                                 );
                                 if let Some(parent) = comm.neighborhood.parent {
                                     if already_requested_failure.replace_with_true() {
                                         Self::request_failure(cu, comm, parent)?
                                     } else {
                                         log!(cu.inner.logger, "Already requested failure");
+                                    }
                                 } else {
                                     log!(cu.inner.logger, "As the leader, deciding on timeout");
                                     return Ok(Decision::Failure);
+                                }
+                            }
                         } else {
                             log!(
                                 cu.inner.logger,
                                 "Delivery to getter {:?} msg {:?} failed because {:?} isn't here",
                                 getter,
                                 &send_payload_msg,
                                 proto_component_id
                             );
+                        }
+                    }
+                }
+            }
             // check if we have a solution yet
             log!(cu.inner.logger, "Check if we have any local decisions...");
             for solution in rctx.solution_storage.iter_new_local_make_old() {
                 log!(cu.inner.logger, "New local decision with solution {:?}...", &solution);
                 match comm.neighborhood.parent {
                     Some(parent) => {
                         log!(cu.inner.logger, "Forwarding to my parent {:?}", parent);
                         let suggestion = Decision::Success(solution);
                         let msg = Msg::CommMsg(CommMsg {
                             round_index: comm.round_index,
                             contents: CommMsgContents::CommCtrl(CommCtrlMsg::Suggest {
                                 suggestion,
                             }),
                         });
                         comm.endpoint_manager.send_to_comms(parent, &msg)?;
+                    }
                     None => {
                         log!(cu.inner.logger, "No parent. Deciding on solution {:?}", &solution);
                         return Ok(Decision::Success(solution));
+                    }
+                }
+            }
             // stuck! make progress by receiving a msg
             // try recv messages arriving through endpoints
             log!(cu.inner.logger, "No decision yet. Let's recv an endpoint msg...");
+            {
                 let (net_index, comm_ctrl_msg): (usize, CommCtrlMsg) = match comm
                     .endpoint_manager
                     .try_recv_any_comms(
                     &mut *cu.inner.logger,
                     &cu.inner.port_info,
                     rctx,
                     comm.round_index,
                 )? {
                     CommRecvOk::NewControlMsg { net_index, msg } => (net_index, msg),
                     CommRecvOk::NewPayloadMsgs => continue 'undecided,
                     CommRecvOk::TimeoutWithoutNew => {
                         log!(cu.inner.logger, "Reached user-defined deadling without decision...");
                         if let Some(parent) = comm.neighborhood.parent {
                             if already_requested_failure.replace_with_true() {
                                 Self::request_failure(cu, comm, parent)?
                             } else {
                                 log!(cu.inner.logger, "Already requested failure");
+                            }
                         } else {
                             log!(cu.inner.logger, "As the leader, deciding on timeout");
                             return Ok(Decision::Failure);
+                        }
                         rctx.deadline = None;
                         continue 'undecided;
+                    }
                 };
                 log!(
                     cu.inner.logger,
                     "Received from endpoint {} ctrl msg  {:?}",
                     net_index,
                     &comm_ctrl_msg
                 );
                 match comm_ctrl_msg {
                     CommCtrlMsg::Suggest { suggestion } => {
                         // only accept this control msg through a child endpoint
                         if comm.neighborhood.children.contains(&net_index) {
                             match suggestion {
                                 Decision::Success(predicate) => {
                                     // child solution contributes to local solution
                                     log!(
                                         cu.inner.logger,
                                         "Child provided solution {:?}",
                                         &predicate
                                     );
                                     let subtree_id = SubtreeId::NetEndpoint { index: net_index };
                                     rctx.solution_storage.submit_and_digest_subtree_solution(
                                         &mut *cu.inner.logger,
                                         subtree_id,
                                         predicate,
                                     );
+                                }
                                 Decision::Failure => {
                                     match comm.neighborhood.parent {
                                         None => {
                                             log!(cu.inner.logger, "I decide on my child's failure");
                                             break 'undecided Ok(Decision::Failure);
+                                        }
                                         Some(parent) => {
                                             log!(cu.inner.logger, "Forwarding failure through my parent endpoint {:?}", parent);
                                             if already_requested_failure.replace_with_true() {
                                                 Self::request_failure(cu, comm, parent)?
                                             } else {
                                                 log!(cu.inner.logger, "Already requested failure");
+                                            }
+                                        }
+                                    }
+                                }
+                            }
                         } else {
                             log!(
                                 cu.inner.logger,
                                 "Discarding suggestion {:?} from non-child endpoint idx {:?}",
                                 &suggestion,
                                 net_index
                             );
+                        }
+                    }
                     CommCtrlMsg::Announce { decision } => {
                         if Some(net_index) == comm.neighborhood.parent {
                             // adopt this decision
                             return Ok(decision);
                         } else {
                             log!(
                                 cu.inner.logger,
                                 "Discarding announcement {:?} from non-parent endpoint idx {:?}",
                                 &decision,
                                 net_index
                             );
+                        }
+                    }
+                }
+            }
             log!(cu.inner.logger, "Endpoint msg recv done");
+        }
+    }
     fn request_failure(
         cu: &mut ConnectorUnphased,
         comm: &mut ConnectorCommunication,
         parent: usize,
     ) -> Result<(), UnrecoverableSyncError> {
         log!(cu.inner.logger, "Forwarding to my parent {:?}", parent);
         let suggestion = Decision::Failure;
         let msg = Msg::CommMsg(CommMsg {
             round_index: comm.round_index,
             contents: CommMsgContents::CommCtrl(CommCtrlMsg::Suggest { suggestion }),
         });
         comm.endpoint_manager.send_to_comms(parent, &msg)
+    }
+}
 impl NativeBranch {
     fn is_ended(&self) -> bool {
         self.to_get.is_empty()
+    }
+}
 impl BranchingNative {
     fn feed_msg(
         &mut self,
         cu: &mut ConnectorUnphased,
         round_ctx: &mut RoundCtx,
         getter: PortId,
         send_payload_msg: &SendPayloadMsg,
         bn_temp: MapTempGuard<'_, Predicate, NativeBranch>,
     ) {
         log!(cu.inner.logger, "feeding native getter {:?} {:?}", getter, &send_payload_msg);
         assert!(cu.inner.port_info.polarities.get(&getter).copied() == Some(Getter));
         let mut draining = bn_temp;
         let finished = &mut self.branches;
         std::mem::swap(draining.0, finished);
         for (predicate, mut branch) in draining.drain() {
             log!(cu.inner.logger, "visiting native branch {:?} with {:?}", &branch, &predicate);
             // check if this branch expects to receive it
             let var = cu.inner.port_info.spec_var_for(getter);
             let mut feed_branch = |branch: &mut NativeBranch, predicate: &Predicate| {
                 branch.to_get.remove(&getter);
                 let was = branch.gotten.insert(getter, send_payload_msg.payload.clone());
                 assert!(was.is_none());
                 if branch.is_ended() {
                     log!(
                         cu.inner.logger,
                         "new native solution with {:?} is_ended() with gotten {:?}",
                         &predicate,
                         &branch.gotten
                     );
                     let subtree_id = SubtreeId::LocalComponent(ComponentId::Native);
                     round_ctx.solution_storage.submit_and_digest_subtree_solution(
                         &mut *cu.inner.logger,
                         subtree_id,
                         predicate.clone(),
                     );
                 } else {
                     log!(
                         cu.inner.logger,
                         "Fed native {:?} still has to_get {:?}",
                         &predicate,
                         &branch.to_get
                     );
+                }
             };
             if predicate.query(var) != Some(SpecVal::FIRING) {
                 // optimization. Don't bother trying this branch
                 log!(
                     cu.inner.logger,
                     "skipping branch with {:?} that doesn't want the message (fastpath)",
                     &predicate
                 );
                 Self::insert_branch_merging(finished, predicate, branch);
                 continue;
+            }
             use AssignmentUnionResult as Aur;
             match predicate.assignment_union(&send_payload_msg.predicate) {
                 Aur::Nonexistant => {
                     // this branch does not receive the message
                     log!(
                         cu.inner.logger,
                         "skipping branch with {:?} that doesn't want the message (slowpath)",
                         &predicate
                     );
                     Self::insert_branch_merging(finished, predicate, branch);
+                }
                 Aur::Equivalent | Aur::FormerNotLatter => {
                     // retain the existing predicate, but add this payload
                     feed_branch(&mut branch, &predicate);
                     log!(cu.inner.logger, "branch pred covers it! Accept the msg");
                     Self::insert_branch_merging(finished, predicate, branch);
+                }
                 Aur::LatterNotFormer => {
                     // fork branch, give fork the message and payload predicate. original branch untouched
                     let mut branch2 = branch.clone();
                     let predicate2 = send_payload_msg.predicate.clone();
                     feed_branch(&mut branch2, &predicate2);
                     log!(
                         cu.inner.logger,
                         "payload pred {:?} covers branch pred {:?}",
                         &predicate2,
                         &predicate
                     );
                     Self::insert_branch_merging(finished, predicate, branch);
                     Self::insert_branch_merging(finished, predicate2, branch2);
+                }
                 Aur::New(predicate2) => {
                     // fork branch, give fork the message and the new predicate. original branch untouched
                     let mut branch2 = branch.clone();
                     feed_branch(&mut branch2, &predicate2);
                     log!(
                         cu.inner.logger,
                         "new subsuming pred created {:?}. forking and feeding",
                         &predicate2
                     );
                     Self::insert_branch_merging(finished, predicate, branch);
                     Self::insert_branch_merging(finished, predicate2, branch2);
+                }
+            }
+        }
+    }
     fn insert_branch_merging(
         branches: &mut HashMap<Predicate, NativeBranch>,
         predicate: Predicate,
         mut branch: NativeBranch,
     ) {
         let e = branches.entry(predicate);
         use std::collections::hash_map::Entry;
         match e {
             Entry::Vacant(ev) => {
                 // no existing branch present. We insert it no problem. (The most common case)
                 ev.insert(branch);
+            }
             Entry::Occupied(mut eo) => {
                 // Oh dear, there is already a branch with this predicate.
                 // Rather than choosing either branch, we MERGE them.
                 // This means taking the UNION of their .gotten and the INTERSECTION of their .to_get
                 let old = eo.get_mut();
                 for (k, v) in branch.gotten.drain() {
                     if old.gotten.insert(k, v).is_none() {
                         // added a gotten element in `branch` not already in `old`
                         old.to_get.remove(&k);
+                    }
+                }
+            }
+        }
+    }
     fn collapse_with(self, logger: &mut dyn Logger, solution_predicate: &Predicate) -> RoundOk {
         log!(
             logger,
             "Collapsing native with {} branch preds {:?}",
             self.branches.len(),
             self.branches.keys()
         );
         for (branch_predicate, branch) in self.branches {
             log!(
                 logger,
                 "Considering native branch {:?} with to_get {:?} gotten {:?}",
                 &branch_predicate,
                 &branch.to_get,
                 &branch.gotten
             );
             if branch.is_ended() && branch_predicate.assigns_subset(solution_predicate) {
                 let NativeBranch { index, gotten, .. } = branch;
                 log!(logger, "Collapsed native has gotten {:?}", &gotten);
                 return RoundOk { batch_index: index, gotten };
+            }
+        }
         panic!("Native had no branches matching pred {:?}", solution_predicate);
+    }
+}
 impl BranchingProtoComponent {
     fn drain_branches_to_blocked(
         cd: CyclicDrainer<Predicate, ProtoComponentBranch>,
         cu: &mut ConnectorUnphased,
         rctx: &mut RoundCtx,
         proto_component_id: ProtoComponentId,
         ports: &HashSet<PortId>,
     ) -> Result<(), UnrecoverableSyncError> {
         cd.cyclic_drain(|mut predicate, mut branch, mut drainer| {
             let mut ctx = SyncProtoContext {
                 cu_inner: &mut cu.inner,
                 predicate: &predicate,
                 branch_inner: &mut branch.inner,
             };
             let blocker = branch.state.sync_run(&mut ctx, &cu.proto_description);
             log!(
                 cu.inner.logger,
                 "Proto component with id {:?} branch with pred {:?} hit blocker {:?}",
                 proto_component_id,
                 &predicate,
                 &blocker,
             );
             use SyncBlocker as B;
             match blocker {
                 B::Inconsistent => drop((predicate, branch)), // EXPLICIT inconsistency
                 B::NondetChoice { n } => {
                     let var = rctx.spec_var_stream.next();
                     for val in SpecVal::iter_domain().take(n as usize) {
                         let pred = predicate.clone().inserted(var, val);
                         let mut branch_n = branch.clone();
                         branch_n.inner.untaken_choice = Some(val.0);
                         drainer.add_input(pred, branch_n);
+                    }
+                }
                 B::CouldntReadMsg(port) => {
                     // move to "blocked"
                     assert!(!branch.inner.inbox.contains_key(&port));
                     drainer.add_output(predicate, branch);
+                }
                 B::CouldntCheckFiring(port) => {
                     // sanity check
                     let var = cu.inner.port_info.spec_var_for(port);
                     assert!(predicate.query(var).is_none());
                     // keep forks in "unblocked"
                     drainer.add_input(predicate.clone().inserted(var, SpecVal::SILENT), branch.clone());
                     drainer.add_input(predicate.inserted(var, SpecVal::FIRING), branch);
+                }
                 B::PutMsg(putter, payload) => {
                     // sanity check
                     assert_eq!(Some(&Putter), cu.inner.port_info.polarities.get(&putter));
                     // overwrite assignment
                     let var = cu.inner.port_info.spec_var_for(putter);
                     let was = predicate.assigned.insert(var, SpecVal::FIRING);
                     if was == Some(SpecVal::SILENT) {
                         log!(cu.inner.logger, "Proto component {:?} tried to PUT on port {:?} when pred said var {:?}==Some(false). inconsistent!", proto_component_id, putter, var);
                         // discard forever
                         drop((predicate, branch));
                     } else {
                         // keep in "unblocked"
                         branch.inner.did_put_or_get.insert(putter);
                         log!(cu.inner.logger, "Proto component {:?} putting payload {:?} on port {:?} (using var {:?})", proto_component_id, &payload, putter, var);
                         let msg = SendPayloadMsg { predicate: predicate.clone(), payload };
                         rctx.getter_buffer.putter_add(cu, putter, msg);
                         drainer.add_input(predicate, branch);
+                    }
+                }
                 B::SyncBlockEnd => {
                     // make concrete all variables
                     for port in ports.iter() {
                         let var = cu.inner.port_info.spec_var_for(*port);
                         let should_have_fired = branch.inner.did_put_or_get.contains(port);
                         let val = *predicate.assigned.entry(var).or_insert(SpecVal::SILENT);
                         let did_fire = val == SpecVal::FIRING;
                         if did_fire != should_have_fired {
                             log!(cu.inner.logger, "Inconsistent wrt. port {:?} var {:?} val {:?} did_fire={}, should_have_fired={}", port, var, val, did_fire, should_have_fired);
                             // IMPLICIT inconsistency
                             drop((predicate, branch));
                             return Ok(());
+                        }
+                    }
                     // submit solution for this component
                     let subtree_id = SubtreeId::LocalComponent(ComponentId::Proto(proto_component_id));
                     rctx.solution_storage.submit_and_digest_subtree_solution(
                         &mut *cu.inner.logger,
                         subtree_id,
                         predicate.clone(),
                     );
                     branch.ended = true;
                     // move to "blocked"
                     drainer.add_output(predicate, branch);
+                }
+            }
             Ok(())
         })
+    }
     // fn branch_merge_func(
     //     mut a: ProtoComponentBranch,
     //     b: &mut ProtoComponentBranch,
     // ) -> ProtoComponentBranch {
     //     if b.ended && !a.ended {
     //         a.ended = true;
     //         std::mem::swap(&mut a, b);
     //     }
     //     a
     // }
     fn feed_msg(
         &mut self,
         cu: &mut ConnectorUnphased,
         rctx: &mut RoundCtx,
         proto_component_id: ProtoComponentId,
         getter: PortId,
         send_payload_msg: &SendPayloadMsg,
         pcb_temps: MapTempsGuard<'_, Predicate, ProtoComponentBranch>,
     ) -> Result<(), UnrecoverableSyncError> {
         let logger = &mut *cu.inner.logger;
         log!(
             logger,
             "feeding proto component {:?} getter {:?} {:?}",
             proto_component_id,
             getter,
             &send_payload_msg
         );
         let BranchingProtoComponent { branches, ports } = self;
         let (mut unblocked, pcb_temps) = pcb_temps.split_first_mut();
         let (mut blocked, pcb_temps) = pcb_temps.split_first_mut();
         // partition drain from branches -> {unblocked, blocked}
         log!(logger, "visiting {} blocked branches...", branches.len());
         for (predicate, mut branch) in branches.drain() {
             if branch.ended {
                 log!(logger, "Skipping ended branch with {:?}", &predicate);
                 blocked.insert(predicate, branch);
                 continue;
+            }
             use AssignmentUnionResult as Aur;
             log!(logger, "visiting branch with pred {:?}", &predicate);
             match predicate.assignment_union(&send_payload_msg.predicate) {
                 Aur::Nonexistant => {
                     // this branch does not receive the message
                     log!(logger, "skipping branch");
                     blocked.insert(predicate, branch);
+                }
                 Aur::Equivalent | Aur::FormerNotLatter => {
                     // retain the existing predicate, but add this payload
                     log!(logger, "feeding this branch without altering its predicate");
                     branch.feed_msg(getter, send_payload_msg.payload.clone());
                     unblocked.insert(predicate, branch);
+                }
                 Aur::LatterNotFormer => {
                     // fork branch, give fork the message and payload predicate. original branch untouched
                     log!(logger, "Forking this branch, giving it the predicate of the msg");
                     let mut branch2 = branch.clone();
                     let predicate2 = send_payload_msg.predicate.clone();
                     branch2.feed_msg(getter, send_payload_msg.payload.clone());
                     blocked.insert(predicate, branch);
                     unblocked.insert(predicate2, branch2);
+                }
                 Aur::New(predicate2) => {
                     // fork branch, give fork the message and the new predicate. original branch untouched
                     log!(logger, "Forking this branch with new predicate {:?}", &predicate2);
                     let mut branch2 = branch.clone();
                     branch2.feed_msg(getter, send_payload_msg.payload.clone());
                     blocked.insert(predicate, branch);
                     unblocked.insert(predicate2, branch2);
+                }
+            }
+        }
         log!(logger, "blocked {:?} unblocked {:?}", blocked.len(), unblocked.len());
         // drain from unblocked --> blocked
         let (swap, _pcb_temps) = pcb_temps.split_first_mut();
         let cd = CyclicDrainer::new(unblocked.0, swap.0, blocked.0);
         BranchingProtoComponent::drain_branches_to_blocked(
             cd,
             cu,
             rctx,
             proto_component_id,
             ports,
         )?;
         // swap the blocked branches back
         std::mem::swap(blocked.0, branches);
         log!(cu.inner.logger, "component settles down with branches: {:?}", branches.keys());
         Ok(())
+    }
     fn collapse_with(self, solution_predicate: &Predicate) -> ProtoComponent {
         let BranchingProtoComponent { ports, branches } = self;
         for (branch_predicate, branch) in branches {
             if branch.ended && branch_predicate.assigns_subset(solution_predicate) {
                 let ProtoComponentBranch { state, .. } = branch;
                 return ProtoComponent { state, ports };
+            }
+        }
         panic!("ProtoComponent had no branches matching pred {:?}", solution_predicate);
+    }
     fn initial(ProtoComponent { state, ports }: ProtoComponent) -> Self {
         let branch = ProtoComponentBranch { state, inner: Default::default(), ended: false };
         Self { ports, branches: hashmap! { Predicate::default() => branch } }
+    }
+}
 impl SolutionStorage {
     fn new(subtree_ids: impl Iterator<Item = SubtreeId>) -> Self {
         let mut subtree_id_to_index: HashMap<SubtreeId, usize> = Default::default();
         let mut subtree_solutions = vec![];
         for id in subtree_ids {
             subtree_id_to_index.insert(id, subtree_solutions.len());
             subtree_solutions.push(Default::default())
+        }
         Self {
             subtree_solutions,
             subtree_id_to_index,
             old_local: Default::default(),
             new_local: Default::default(),
+        }
+    }
     // fn is_clear(&self) -> bool {
     //     self.subtree_id_to_index.is_empty()
     //         && self.subtree_solutions.is_empty()
     //         && self.old_local.is_empty()
     //         && self.new_local.is_empty()
     // }
     // fn clear(&mut self) {
     //     self.subtree_id_to_index.clear();
     //     self.subtree_solutions.clear();
     //     self.old_local.clear();
     //     self.new_local.clear();
     // }
     // fn reset(&mut self, subtree_ids: impl Iterator<Item = SubtreeId>) {
     //     self.subtree_id_to_index.clear();
     //     self.subtree_solutions.clear();
     //     self.old_local.clear();
     //     self.new_local.clear();
     //     for key in subtree_ids {
     //         self.subtree_id_to_index.insert(key, self.subtree_solutions.len());
     //         self.subtree_solutions.push(Default::default())
     //     }
     // }
     pub(crate) fn iter_new_local_make_old(&mut self) -> impl Iterator<Item = Predicate> + '_ {
         let Self { old_local, new_local, .. } = self;
         new_local.drain().map(move |local| {
             old_local.insert(local.clone());
             local
         })
+    }
     pub(crate) fn submit_and_digest_subtree_solution(
         &mut self,
         logger: &mut dyn Logger,
         subtree_id: SubtreeId,
         predicate: Predicate,
     ) {
         log!(logger, "++ new component solution {:?} {:?}", subtree_id, &predicate);
         let index = self.subtree_id_to_index[&subtree_id];
         let left = 0..index;
         let right = (index + 1)..self.subtree_solutions.len();
         let Self { subtree_solutions, new_local, old_local, .. } = self;
         let was_new = subtree_solutions[index].insert(predicate.clone());
         if was_new {
             let set_visitor = left.chain(right).map(|index| &subtree_solutions[index]);
             Self::elaborate_into_new_local_rec(
                 logger,
                 predicate,
                 set_visitor,
                 old_local,
                 new_local,
             );
+        }
+    }
     fn elaborate_into_new_local_rec<'a, 'b>(
         logger: &mut dyn Logger,
         partial: Predicate,
         mut set_visitor: impl Iterator<Item = &'b HashSet<Predicate>> + Clone,
         old_local: &'b HashSet<Predicate>,
         new_local: &'a mut HashSet<Predicate>,
     ) {
         if let Some(set) = set_visitor.next() {
             // incomplete solution. keep traversing
             for pred in set.iter() {
                 if let Some(elaborated) = pred.union_with(&partial) {
                     Self::elaborate_into_new_local_rec(
                         logger,
                         elaborated,
                         set_visitor.clone(),
                         old_local,
                         new_local,
+                    )
+                }
+            }
         } else {
             // recursive stop condition. `partial` is a local subtree solution
             if !old_local.contains(&partial) {
                 // ... and it hasn't been found before
                 log!(logger, "storing NEW LOCAL SOLUTION {:?}", &partial);
                 new_local.insert(partial);
+            }
+        }
+    }
+}
 impl GetterBuffer {
     fn len(&self) -> usize {
         self.getters_and_sends.len()
+    }
     fn pop(&mut self) -> Option<(PortId, SendPayloadMsg)> {
         self.getters_and_sends.pop()
+    }
     fn getter_add(&mut self, getter: PortId, msg: SendPayloadMsg) {
         self.getters_and_sends.push((getter, msg));
+    }
     fn putter_add(&mut self, cu: &mut ConnectorUnphased, putter: PortId, msg: SendPayloadMsg) {
         if let Some(&getter) = cu.inner.port_info.peers.get(&putter) {
             self.getter_add(getter, msg);
         } else {
             log!(cu.inner.logger, "Putter {:?} has no known peer!", putter);
             panic!("Putter {:?} has no known peer!");
+        }
+    }
+}
 impl SyncProtoContext<'_> {
     pub(crate) fn is_firing(&mut self, port: PortId) -> Option<bool> {
         let var = self.cu_inner.port_info.spec_var_for(port);
         self.predicate.query(var).map(SpecVal::is_firing)
+    }
     pub(crate) fn read_msg(&mut self, port: PortId) -> Option<&Payload> {
         self.branch_inner.did_put_or_get.insert(port);
         self.branch_inner.inbox.get(&port)
+    }
     pub(crate) fn take_choice(&mut self) -> Option<u16> {
         self.branch_inner.untaken_choice.take()
+    }
+}
 impl<'a, K: Eq + Hash, V> CyclicDrainInner<'a, K, V> {
     fn add_input(&mut self, k: K, v: V) {
         self.swap.insert(k, v);
+    }
     // fn merge_input_with<F: FnMut(V, &mut V) -> V>(&mut self, k: K, v: V, mut func: F) {
     //     use std::collections::hash_map::Entry;
     //     let e = self.swap.entry(k);
     //     match e {
     //         Entry::Vacant(ev) => {
     //             ev.insert(v);
     //         }
     //         Entry::Occupied(mut eo) => {
     //             let old = eo.get_mut();
     //             *old = func(v, old);
     //         }
     //     }
     // }
     fn add_output(&mut self, k: K, v: V) {
         self.output.insert(k, v);
+    }
+}
 impl NonsyncProtoContext<'_> {
     pub fn new_component(&mut self, moved_ports: HashSet<PortId>, state: ComponentState) {
         // called by a PROTO COMPONENT. moves its own ports.
         // 1. sanity check: this component owns these ports
         log!(
             self.cu_inner.logger,
             "Component {:?} added new component with state {:?}, moving ports {:?}",
             self.proto_component_id,
             &state,
             &moved_ports
         );
         assert!(self.proto_component_ports.is_subset(&moved_ports));
         // 2. remove ports from old component & update port->route
         let new_id = self.cu_inner.id_manager.new_proto_component_id();
         for port in moved_ports.iter() {
             self.proto_component_ports.remove(port);
             self.cu_inner
                 .port_info
                 .routes
                 .insert(*port, Route::LocalComponent(ComponentId::Proto(new_id)));
+        }
         // 3. create a new component
         self.unrun_components.push((new_id, ProtoComponent { state, ports: moved_ports }));
+    }
     pub fn new_port_pair(&mut self) -> [PortId; 2] {
         // adds two new associated ports, related to each other, and exposed to the proto component
         let [o, i] =
             [self.cu_inner.id_manager.new_port_id(), self.cu_inner.id_manager.new_port_id()];
         self.proto_component_ports.insert(o);
         self.proto_component_ports.insert(i);
         // {polarity, peer, route} known. {} unknown.
         self.cu_inner.port_info.polarities.insert(o, Putter);
         self.cu_inner.port_info.polarities.insert(i, Getter);
         self.cu_inner.port_info.peers.insert(o, i);
         self.cu_inner.port_info.peers.insert(i, o);
         let route = Route::LocalComponent(ComponentId::Proto(self.proto_component_id));
         self.cu_inner.port_info.routes.insert(o, route);
         self.cu_inner.port_info.routes.insert(i, route);
         log!(
             self.cu_inner.logger,
             "Component {:?} port pair (out->in) {:?} -> {:?}",
             self.proto_component_id,
             o,
+            i
         );
         [o, i]
+    }
+}
 impl ProtoComponentBranch {
     fn feed_msg(&mut self, getter: PortId, payload: Payload) {
         let was = self.inner.inbox.insert(getter, payload);
         assert!(was.is_none())
+    }
+}
 impl<'a, K: Eq + Hash + 'static, V: 'static> CyclicDrainer<'a, K, V> {
     fn new(
         input: &'a mut HashMap<K, V>,
         swap: &'a mut HashMap<K, V>,
         output: &'a mut HashMap<K, V>,
     ) -> Self {
         Self { input, inner: CyclicDrainInner { swap, output } }
+    }
     fn cyclic_drain<E>(
         self,
         mut func: impl FnMut(K, V, CyclicDrainInner<'_, K, V>) -> Result<(), E>,
     ) -> Result<(), E> {
         let Self { input, inner: CyclicDrainInner { swap, output } } = self;
         // assert!(swap.is_empty());
         while !input.is_empty() {
             for (k, v) in input.drain() {
                 func(k, v, CyclicDrainInner { swap, output })?
+            }
             std::mem::swap(input, swap);
+        }
         Ok(())
+    }
+}

src/runtime/endpoints.rs

➞

Show inline comments

 use super::*;
 struct MonitoredReader<R: Read> {
     bytes: usize,
     r: R,
+}
 enum PollAndPopulateError {
     PollFailed,
     Timeout,
+}
 struct TryRecvAnyNetError {
     error: NetEndpointError,
     index: usize,
+}
 /////////////////////
 impl NetEndpoint {
     fn bincode_opts() -> impl bincode::config::Options {
         bincode::config::DefaultOptions::default()
+    }
     pub(super) fn try_recv<T: serde::de::DeserializeOwned>(
         &mut self,
         logger: &mut dyn Logger,
     ) -> Result<Option<T>, NetEndpointError> {
         use NetEndpointError as Nee;
         // populate inbox as much as possible
         let before_len = self.inbox.len();
         'read_loop: loop {
             let res = self.stream.read_to_end(&mut self.inbox);
             match res {
                 Err(e) if would_block(&e) => break 'read_loop,
                 Ok(0) => break 'read_loop,
                 Ok(_) => (),
                 Err(_e) => return Err(Nee::BrokenNetEndpoint),
+            }
+        }
         endptlog!(
             logger,
             "Inbox bytes [{:x?}| {:x?}]",
             DenseDebugHex(&self.inbox[..before_len]),
             DenseDebugHex(&self.inbox[before_len..]),
         );
         let mut monitored = MonitoredReader::from(&self.inbox[..]);
         use bincode::config::Options;
         match Self::bincode_opts().deserialize_from(&mut monitored) {
             Ok(msg) => {
                 let msg_size = monitored.bytes_read();
                 self.inbox.drain(0..(msg_size.try_into().unwrap()));
                 endptlog!(
                     logger,
                     "Yielding msg. Inbox len {}-{}=={}: [{:?}]",
                     self.inbox.len() + msg_size,
                     msg_size,
                     self.inbox.len(),
                     DenseDebugHex(&self.inbox[..]),
                 );
                 Ok(Some(msg))
+            }
             Err(e) => match *e {
                 bincode::ErrorKind::Io(k) if k.kind() == std::io::ErrorKind::UnexpectedEof => {
                     Ok(None)
+                }
                 _ => Err(Nee::MalformedMessage),
             },
+        }
+    }
     pub(super) fn send<T: serde::ser::Serialize>(
         &mut self,
         msg: &T,
     ) -> Result<(), NetEndpointError> {
         use bincode::config::Options;
         use NetEndpointError as Nee;
         Self::bincode_opts()
             .serialize_into(&mut self.stream, msg)
             .map_err(|_| Nee::BrokenNetEndpoint)
+    }
+}
 impl EndpointManager {
     pub(super) fn index_iter(&self) -> Range<usize> {
 ..self.num_net_endpoints()
+    }
     pub(super) fn num_net_endpoints(&self) -> usize {
         self.net_endpoint_store.endpoint_exts.len()
+    }
     pub(super) fn send_to_comms(
         &mut self,
         index: usize,
         msg: &Msg,
     ) -> Result<(), UnrecoverableSyncError> {
         use UnrecoverableSyncError as Use;
         let net_endpoint = &mut self.net_endpoint_store.endpoint_exts[index].net_endpoint;
         net_endpoint.send(msg).map_err(|_| Use::BrokenNetEndpoint { index })
+    }
     pub(super) fn send_to_setup(&mut self, index: usize, msg: &Msg) -> Result<(), ConnectError> {
         let net_endpoint = &mut self.net_endpoint_store.endpoint_exts[index].net_endpoint;
         net_endpoint.send(msg).map_err(|err| {
             ConnectError::NetEndpointSetupError(net_endpoint.stream.local_addr().unwrap(), err)
         })
+    }
     /// Receive the first message of any kind at all.
     /// Why not return SetupMsg? Because often this message will be forwarded to several others,
     /// and by returning a Msg, it can be serialized in-place (NetEndpoints allow the sending of Msg types!)
     pub(super) fn try_recv_any_setup(
         &mut self,
         logger: &mut dyn Logger,
         deadline: &Option<Instant>,
     ) -> Result<(usize, Msg), ConnectError> {
         ///////////////////////////////////////////
         fn map_trane(
             trane: TryRecvAnyNetError,
             net_endpoint_store: &EndpointStore<NetEndpointExt>,
         ) -> ConnectError {
             ConnectError::NetEndpointSetupError(
                 net_endpoint_store.endpoint_exts[trane.index]
                     .net_endpoint
                     .stream
                     .local_addr()
                     .unwrap(), // stream must already be connected
                 trane.error,
+            )
+        }
         ///////////////////////////////////////////
         // try yield undelayed net message
         if let Some(tup) = self.undelayed_messages.pop() {
             endptlog!(logger, "RECV undelayed_msg {:?}", &tup);
             return Ok(tup);
+        }
         loop {
             // try recv from some polled undrained NET endpoint
             if let Some(tup) = self
                 .try_recv_undrained_net(logger)
                 .map_err(|trane| map_trane(trane, &self.net_endpoint_store))?
+            {
                 return Ok(tup);
+            }
             // poll if time remains
-            self.poll_and_polulate(logger, deadline)?;
+            self.poll_and_populate(logger, deadline)?;
+        }
+    }
     // drops all Setup messages,
     // buffers all future round messages,
     // drops all previous round messages,
     // enqueues all current round SendPayload messages using round_ctx.getter_add
     // returns the first comm_ctrl_msg encountered
     // only polls until SOME message is enqueued
     pub(super) fn try_recv_any_comms(
         &mut self,
         logger: &mut dyn Logger,
         port_info: &PortInfo,
         round_ctx: &mut impl RoundCtxTrait,
         round_index: usize,
     ) -> Result<CommRecvOk, UnrecoverableSyncError> {
         ///////////////////////////////////////////
         impl EndpointManager {
             fn handle_msg(
                 &mut self,
                 logger: &mut dyn Logger,
                 round_ctx: &mut impl RoundCtxTrait,
                 net_index: usize,
                 msg: Msg,
                 round_index: usize,
                 some_message_enqueued: &mut bool,
             ) -> Option<(usize, CommCtrlMsg)> {
                 let comm_msg_contents = match msg {
                     Msg::SetupMsg(..) => return None,
                     Msg::CommMsg(comm_msg) => match comm_msg.round_index.cmp(&round_index) {
                         Ordering::Equal => comm_msg.contents,
                         Ordering::Less => {
                             log!(
                                 logger,
                                 "We are in round {}, but msg is for round {}. Discard",
                                 comm_msg.round_index,
                                 round_index,
                             );
                             return None;
+                        }
                         Ordering::Greater => {
                             log!(
                                 logger,
                                 "We are in round {}, but msg is for round {}. Buffer",
                                 comm_msg.round_index,
                                 round_index,
                             );
                             self.delayed_messages.push((net_index, Msg::CommMsg(comm_msg)));
                             return None;
+                        }
                     },
                 };
                 match comm_msg_contents {
                     CommMsgContents::CommCtrl(comm_ctrl_msg) => Some((net_index, comm_ctrl_msg)),
                     CommMsgContents::SendPayload(send_payload_msg) => {
                         let getter =
                             self.net_endpoint_store.endpoint_exts[net_index].getter_for_incoming;
                         round_ctx.getter_add(getter, send_payload_msg);
                         *some_message_enqueued = true;
                         None
+                    }
+                }
+            }
+        }
         use {PollAndPopulateError as Pape, UnrecoverableSyncError as Use};
         ///////////////////////////////////////////
         let mut some_message_enqueued = false;
         // try yield undelayed net message
         while let Some((net_index, msg)) = self.undelayed_messages.pop() {
             if let Some((net_index, msg)) = self.handle_msg(
                 logger,
                 round_ctx,
                 net_index,
                 msg,
                 round_index,
                 &mut some_message_enqueued,
             ) {
                 return Ok(CommRecvOk::NewControlMsg { net_index, msg });
+            }
+        }
         loop {
             // try receive a net message
             while let Some((net_index, msg)) = self.try_recv_undrained_net(logger)? {
                 if let Some((net_index, msg)) = self.handle_msg(
                     logger,
                     round_ctx,
                     net_index,
                     msg,
                     round_index,
                     &mut some_message_enqueued,
                 ) {
                     return Ok(CommRecvOk::NewControlMsg { net_index, msg });
+                }
+            }
             // try receive a udp message
             let recv_buffer = self.udp_in_buffer.as_mut_slice();
             while let Some(index) = self.udp_endpoint_store.polled_undrained.pop() {
                 let ee = &mut self.udp_endpoint_store.endpoint_exts[index];
                 if let Some(bytes_written) = ee.sock.recv(recv_buffer).ok() {
                     // I received a payload!
                     self.udp_endpoint_store.polled_undrained.insert(index);
                     if !ee.received_this_round {
                         let payload = Payload::from(&recv_buffer[..bytes_written]);
                     let [branch_spec_var, port_spec_var] = [
                         ee.incoming_round_spec_var.unwrap(), // should not be NONE
                         port_info.spec_var_for(ee.getter_for_incoming),
                     ];
                     let branch_spec_val = SpecVal::nth_domain_element(ee.incoming_payloads.len());
                     ee.incoming_payloads.push(payload.clone());
                     let predicate = Predicate::default()
                         .inserted(branch_spec_var, branch_spec_val)
                         .inserted(port_spec_var, SpecVal::FIRING);
                     round_ctx
                         .getter_add(ee.getter_for_incoming, SendPayloadMsg { payload, predicate });
                         let port_spec_var = port_info.spec_var_for(ee.getter_for_incoming);
                         let predicate = Predicate::singleton(port_spec_var, SpecVal::FIRING);
                         round_ctx.getter_add(
                             ee.getter_for_incoming,
                             SendPayloadMsg { payload, predicate },
                         );
                         some_message_enqueued = true;
                         ee.received_this_round = true;
                     } else {
                         // lose the message!
+                    }
+                }
+            }
             if some_message_enqueued {
                 return Ok(CommRecvOk::NewPayloadMsgs);
+            }
             // poll if time remains
-            match self.poll_and_polulate(logger, round_ctx.get_deadline()) {
+            match self.poll_and_populate(logger, round_ctx.get_deadline()) {
                 Ok(()) => {} // continue looping
                 Err(Pape::Timeout) => return Ok(CommRecvOk::TimeoutWithoutNew),
                 Err(Pape::PollFailed) => return Err(Use::PollFailed),
+            }
+        }
+    }
     fn try_recv_undrained_net(
         &mut self,
         logger: &mut dyn Logger,
     ) -> Result<Option<(usize, Msg)>, TryRecvAnyNetError> {
         while let Some(index) = self.net_endpoint_store.polled_undrained.pop() {
             let net_endpoint = &mut self.net_endpoint_store.endpoint_exts[index].net_endpoint;
             if let Some(msg) = net_endpoint
                 .try_recv(logger)
                 .map_err(|error| TryRecvAnyNetError { error, index })?
+            {
                 endptlog!(logger, "RECV polled_undrained {:?}", &msg);
                 if !net_endpoint.inbox.is_empty() {
                     // there may be another message waiting!
                     self.net_endpoint_store.polled_undrained.insert(index);
+                }
                 return Ok(Some((index, msg)));
+            }
+        }
         Ok(None)
+    }
-    fn poll_and_polulate(
+    fn poll_and_populate(
         &mut self,
         logger: &mut dyn Logger,
         deadline: &Option<Instant>,
     ) -> Result<(), PollAndPopulateError> {
         use PollAndPopulateError as Pape;
         // No message yet. Do we have enough time to poll?
         let remaining = if let Some(deadline) = deadline {
             Some(deadline.checked_duration_since(Instant::now()).ok_or(Pape::Timeout)?)
         } else {
             None
         };
         // Yes we do! Poll with remaining time as poll deadline
         self.poll.poll(&mut self.events, remaining).map_err(|_| Pape::PollFailed)?;
         for event in self.events.iter() {
             match TokenTarget::from(event.token()) {
                 TokenTarget::Waker => {
                     // Can ignore. Residual event from endpoint manager setup procedure
+                }
                 TokenTarget::NetEndpoint { index } => {
                     self.net_endpoint_store.polled_undrained.insert(index);
                     endptlog!(
                         logger,
                         "RECV poll event {:?} for NET endpoint index {:?}. undrained: {:?}",
                         &event,
                         index,
                         self.net_endpoint_store.polled_undrained.iter()
                     );
+                }
                 TokenTarget::UdpEndpoint { index } => {
                     self.udp_endpoint_store.polled_undrained.insert(index);
                     endptlog!(
                         logger,
                         "RECV poll event {:?} for UDP endpoint index {:?}. undrained: {:?}",
                         &event,
                         index,
                         self.udp_endpoint_store.polled_undrained.iter()
                     );
+                }
+            }
+        }
         self.events.clear();
         Ok(())
+    }
     pub(super) fn undelay_all(&mut self) {
         if self.undelayed_messages.is_empty() {
             // fast path
             std::mem::swap(&mut self.delayed_messages, &mut self.undelayed_messages);
             return;
+        }
         // slow path
         self.undelayed_messages.extend(self.delayed_messages.drain(..));
+    }
     pub(super) fn udp_endpoints_round_start(
         &mut self,
         logger: &mut dyn Logger,
         spec_var_stream: &mut SpecVarStream,
     ) {
     pub(super) fn udp_endpoints_round_start(&mut self, logger: &mut dyn Logger) {
         log!(
             logger,
             "Starting round for {} udp endpoints",
             self.udp_endpoint_store.endpoint_exts.len()
         );
         for (index, ee) in self.udp_endpoint_store.endpoint_exts.iter_mut().enumerate() {
             let spec_var = spec_var_stream.next();
             log!(logger, "Udp endpoint given {} spec var {:?} for this round", index, spec_var);
             ee.incoming_round_spec_var = Some(spec_var);
         for ee in self.udp_endpoint_store.endpoint_exts.iter_mut() {
             ee.received_this_round = false;
+        }
+    }
     pub(super) fn udp_endpoints_round_end(
         &mut self,
         logger: &mut dyn Logger,
         decision: &Decision,
     ) -> Result<(), UnrecoverableSyncError> {
         log!(
             logger,
             "Ending round for {} udp endpoints",
             self.udp_endpoint_store.endpoint_exts.len()
         );
         use UnrecoverableSyncError as Use;
         if let Decision::Success(solution_predicate) = decision {
             'endpoint_loop: for (index, ee) in
                 self.udp_endpoint_store.endpoint_exts.iter_mut().enumerate()
+            {
                 ee.incoming_round_spec_var = None; // shouldn't be accessed before its overwritten next round; still adding for clarity.
                 for (payload_predicate, payload) in ee.outgoing_payloads.drain() {
                     if payload_predicate.assigns_subset(solution_predicate) {
                         ee.sock.send(payload.as_slice()).map_err(|e| {
                             println!("{:?}", e);
                             Use::BrokenUdpEndpoint { index }
                         })?;
                         log!(
                             logger,
                             "Sent payload {:?} with pred {:?} through Udp endpoint {}",
                             &payload,
                             &payload_predicate,
                             index
                         );
                         continue 'endpoint_loop; // send at most one payload per endpoint per round
+                    }
+                }
                 log!(logger, "Sent no message through Udp endpoint {}", index);
+            }
+        }
         Ok(())
+    }
+}
 // impl UdpEndpointExt {
 //     fn try_recv(
 //         &mut self,
 //         port_info: &PortInfo,
 //         udp_in_buffer: &mut UdpInBuffer,
 //     ) -> Option<SendPayloadMsg> {
 //         let recv_buffer = udp_in_buffer.as_mut_slice();
 //         let len = self.sock.recv(recv_buffer).ok()?;
 //         let payload = Payload::from(&recv_buffer[..len]);
 //         let branch_spec_var = self
 //             .incoming_round_spec_var
 //             .expect("Udp spec var should be Some(..) if recv() is called");
 //         let branch_spec_val = SpecVal::nth_domain_element(self.incoming_payloads.len());
 //         self.incoming_payloads.push(payload.clone());
 //         let predicate = Predicate::default()
 //             .inserted(branch_spec_var, branch_spec_val)
 //             .inserted(port_info.spec_var_for(self.getter_for_incoming), SpecVal::FIRING);
 //         Some(SendPayloadMsg { payload, predicate })
 //     }
 // }
 impl Debug for NetEndpoint {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         f.debug_struct("Endpoint").field("inbox", &self.inbox).finish()
+    }
+}
 impl<R: Read> From<R> for MonitoredReader<R> {
     fn from(r: R) -> Self {
         Self { r, bytes: 0 }
+    }
+}
 impl<R: Read> MonitoredReader<R> {
     pub(super) fn bytes_read(&self) -> usize {
         self.bytes
+    }
+}
 impl<R: Read> Read for MonitoredReader<R> {
     fn read(&mut self, buf: &mut [u8]) -> Result<usize, std::io::Error> {
         let n = self.r.read(buf)?;
         self.bytes += n;
         Ok(n)
+    }
+}
 impl Into<Msg> for SetupMsg {
     fn into(self) -> Msg {
         Msg::SetupMsg(self)
+    }
+}
 impl From<PollAndPopulateError> for ConnectError {
     fn from(pape: PollAndPopulateError) -> ConnectError {
         use {ConnectError as Ce, PollAndPopulateError as Pape};
         match pape {
             Pape::PollFailed => Ce::PollFailed,
             Pape::Timeout => Ce::Timeout,
+        }
+    }
+}
 impl From<TryRecvAnyNetError> for UnrecoverableSyncError {
     fn from(trane: TryRecvAnyNetError) -> UnrecoverableSyncError {
         let TryRecvAnyNetError { index, .. } = trane;
         UnrecoverableSyncError::BrokenNetEndpoint { index }
+    }
+}

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(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 + Send + Sync {
     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> {
     cu_inner: &'a mut ConnectorUnphasedInner, // persists between rounds
     proto_component_ports: &'a mut HashSet<PortId>, // sub-structure of component
     unrun_components: &'a mut Vec<(ProtoComponentId, ProtoComponent)>, // lives for Nonsync phase
     proto_component_id: ProtoComponentId,     // KEY in id->component map
+}
 pub(crate) struct SyncProtoContext<'a> {
     cu_inner: &'a mut ConnectorUnphasedInner, // persists between rounds
     branch_inner: &'a mut ProtoComponentBranchInner, // sub-structure of component branch
     predicate: &'a Predicate,                 // KEY in pred->branch map
+}
 #[derive(Default, Debug, Clone)]
 struct ProtoComponentBranchInner {
     untaken_choice: Option<u16>,
     did_put_or_get: HashSet<PortId>,
     inbox: 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 {
     getter_for_incoming: PortId,
     sock_addr: SocketAddr,
     endpoint_polarity: EndpointPolarity,
+}
 #[derive(Debug, Clone)]
 struct UdpEndpointSetup {
     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. 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
     received_this_round: bool,
     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>,
     inner: ConnectorUnphasedInner,
+}
 #[derive(Debug)]
 struct ConnectorUnphasedInner {
     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>,
+}
 #[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 {
         // Spec var stream starts where the current port_id stream ends, with gap of SKIP_N.
         // This gap is entirely unnecessary (i.e. 0 is fine)
         // It's purpose is only to make SpecVars easier to spot in logs.
         // E.g. spot the spec var: { v0_0, v1_2, v1_103 }
         const SKIP_N: u32 = 100;
         let port_suffix_stream = self.port_suffix_stream.clone().n_skipped(SKIP_N);
         SpecVarStream { connector_id: self.connector_id, port_suffix_stream }
+    }
     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.inner.logger, "Connector dropping. Goodbye!");
+    }
+}
 impl Connector {
     pub(crate) 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.inner.logger, &mut new_logger);
         new_logger
+    }
     pub fn get_logger(&mut self) -> &mut dyn Logger {
         &mut *self.unphased.inner.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.inner.id_manager.new_port_id(), cu.inner.id_manager.new_port_id()];
         cu.inner.native_ports.insert(o);
         cu.inner.native_ports.insert(i);
         // {polarity, peer, route} known. {} unknown.
         cu.inner.port_info.polarities.insert(o, Putter);
         cu.inner.port_info.polarities.insert(i, Getter);
         cu.inner.port_info.peers.insert(o, i);
         cu.inner.port_info.peers.insert(i, o);
         let route = Route::LocalComponent(ComponentId::Native);
         cu.inner.port_info.routes.insert(o, route);
         cu.inner.port_info.routes.insert(i, route);
         log!(cu.inner.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.inner.native_ports.contains(port) {
                 return Err(Ace::UnknownPort(*port));
+            }
             if expected_polarity != *cu.inner.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.inner.id_manager.new_proto_component_id();
         for port in ports.iter() {
             cu.inner
                 .port_info
                 .routes
                 .insert(*port, Route::LocalComponent(ComponentId::Proto(new_id)));
+        }
         cu.inner.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 singleton(k: SpecVar, v: SpecVal) -> Self {
         Self::default().inserted(k, v)
+    }
     #[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 {
         struct Assignment<'a>((&'a SpecVar, &'a SpecVal));
         impl Debug for Assignment<'_> {
             fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
                 write!(f, "{:?}={:?}", (self.0).0, (self.0).1)
+            }
+        }
         f.debug_set().entries(self.assigned.iter().map(Assignment)).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)))
+    }
+}
 impl IdParts for SpecVar {
     fn id_parts(self) -> (ConnectorId, U32Suffix) {
         self.0.id_parts()
+    }
+}
 impl Debug for SpecVar {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         let (a, b) = self.id_parts();
         write!(f, "v{}_{}", a, b)
+    }
+}
 impl SpecVal {
     const FIRING: Self = SpecVal(1);
     const SILENT: Self = SpecVal(0);
     fn is_firing(self) -> bool {
         self == Self::FIRING
         // all else treated as SILENT
+    }
     fn nth_domain_element(n: usize) -> Self {
         let n: u16 = n.try_into().unwrap();
         SpecVal(n)
+    }
     fn iter_domain() -> impl Iterator<Item = Self> {
         (0..).map(SpecVal)
+    }
+}
 impl Debug for SpecVal {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         self.0.fmt(f)
+    }
+}
 impl Default for UdpInBuffer {
     fn default() -> Self {
         let mut byte_vec = Vec::with_capacity(Self::CAPACITY);
         unsafe {
             // safe! this vector is guaranteed to have sufficient capacity
             byte_vec.set_len(Self::CAPACITY);
+        }
         Self { byte_vec }
+    }
+}
 impl UdpInBuffer {
     const CAPACITY: usize = u16::MAX as usize;
     fn as_mut_slice(&mut self) -> &mut [u8] {
         self.byte_vec.as_mut_slice()
+    }
+}

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,
     ) -> Self {
         log!(&mut *logger, "Created with connector_id {:?}", connector_id);
         Self {
             unphased: ConnectorUnphased {
                 proto_description,
                 proto_components: Default::default(),
                 inner: ConnectorUnphasedInner {
                     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(),
             })),
+        }
+    }
     /// Conceptually, this returning [p0, g1] is sugar for:
     /// 1. create port pair [p0, g0]
     /// 2. create port pair [p1, g1]
     /// 3. create udp component with interface of moved ports [p1, g0]
     /// 4. return [p0, g1]
     pub fn new_udp_mediator_component(
         &mut self,
         local_addr: SocketAddr,
         peer_addr: SocketAddr,
     ) -> 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 mut npid = || cu.inner.id_manager.new_port_id();
                 let [nin, nout, uin, uout] = [npid(), npid(), npid(), npid()];
                 cu.inner.native_ports.insert(nin);
                 cu.inner.native_ports.insert(nout);
                 cu.inner.port_info.polarities.insert(nin, Getter);
                 cu.inner.port_info.polarities.insert(nout, Putter);
                 cu.inner.port_info.polarities.insert(uin, Getter);
                 cu.inner.port_info.polarities.insert(uout, Putter);
                 cu.inner.port_info.peers.insert(nin, uout);
                 cu.inner.port_info.peers.insert(nout, uin);
                 cu.inner.port_info.peers.insert(uin, nout);
                 cu.inner.port_info.peers.insert(uout, nin);
                 cu.inner.port_info.routes.insert(nin, Route::LocalComponent(ComponentId::Native));
                 cu.inner.port_info.routes.insert(nout, Route::LocalComponent(ComponentId::Native));
                 cu.inner.port_info.routes.insert(uin, Route::UdpEndpoint { index: udp_index });
                 cu.inner.port_info.routes.insert(uout, Route::UdpEndpoint { index: udp_index });
                 setup.udp_endpoint_setups.push(UdpEndpointSetup {
                     local_addr,
                     peer_addr,
                     getter_for_incoming: nin,
                 });
                 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.inner.id_manager.new_port_id();
                 cu.inner.native_ports.insert(local_port);
                 // {polarity, route} known. {peer} unknown.
                 cu.inner.port_info.polarities.insert(local_port, polarity);
                 cu.inner
                     .port_info
                     .routes
                     .insert(local_port, Route::LocalComponent(ComponentId::Native));
                 log!(
                     cu.inner.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,
                     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.inner.logger, "Call to connecting in connected state");
                 Err(Ce::AlreadyConnected)
+            }
             ConnectorPhased::Setup(setup) => {
                 log!(cu.inner.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.inner.logger,
                     &setup.net_endpoint_setups,
                     &setup.udp_endpoint_setups,
                     &mut cu.inner.port_info,
                     &deadline,
                 )?;
                 log!(
                     cu.inner.logger,
                     "Successfully connected {} endpoints",
                     endpoint_manager.net_endpoint_store.endpoint_exts.len()
                 );
                 // leader election and tree construction
                 let neighborhood = init_neighborhood(
                     cu.inner.id_manager.connector_id,
                     &mut *cu.inner.logger,
                     &mut endpoint_manager,
                     &deadline,
                 )?;
                 log!(cu.inner.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.inner.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, 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 {
         // 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
     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() + 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 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();
             log!(logger, "Net endpoint {} beginning setup with {:?}", index, &endpoint_setup);
             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, getter_for_incoming: endpoint_setup.getter_for_incoming })
         })
         .collect::<Result<Vec<UdpTodo>, ConnectError>>()?;
     // 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);
             match token_target {
                 TokenTarget::Waker => {
                     log!(
                         logger,
                         "Notification from waker. connect_failed is {:?}",
                         net_connect_retry_later.iter()
                     );
                     assert!(waker_state.is_some());
                     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) => continue, // spurious wakeup
                             Err(_) => {
                                 log!(logger, "accept() failure on index {}", index);
                                 return Err(Ce::AcceptFailed(listener.local_addr().unwrap()));
+                            }
                             Ok((mut stream, peer_addr)) => {
                                 // successfully accepted the active peer
                                 // reusing the token, but now for the stream and not the listener
                                 poll.registry().deregister(listener).unwrap();
                                 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 net_connect_retry_later.insert(index) {
                                 log!(
                                     logger,
                                     "Connection failed for {:?}. List is {:?}",
                                     index,
                                     net_connect_retry_later.iter()
                                 );
                                 poll.registry().deregister(&mut net_endpoint.stream).unwrap();
                             } else {
                                 // 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 || moved_arc.waker_loop());
+                            }
                             continue;
+                        }
                         // event wasn't ERROR
                         if net_connect_retry_later.contains(&index) {
                             // spurious wakeup. already scheduled to retry connect later
                             continue;
+                        }
                         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.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.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.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.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(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.getter_for_incoming,
         })
         .collect();
     let udp_endpoint_exts = udp_todos
         .into_iter()
         .enumerate()
         .map(|(index, 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,
+                received_this_round: false,
                 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.inner.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.inner.logger,
             "Session gather loop. awaiting info from children {:?}...",
             awaiting.iter()
         );
         let (recv_index, msg) =
             comm.endpoint_manager.try_recv_any_setup(&mut *cu.inner.logger, deadline)?;
         log!(cu.inner.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.inner.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.inner.logger, "discarding old message {:?} during election", msg);
+            }
             msg @ S(Sm::SessionScatter { .. }) => {
                 log!(
                     cu.inner.logger,
                     "Endpoint {:?} sent unexpected scatter! {:?} I've not contributed yet!",
                     recv_index,
                     &msg
                 );
                 return Err(Ce::SetupAlgMisbehavior);
+            }
             msg @ Msg::CommMsg(..) => {
                 log!(cu.inner.logger, "delaying msg {:?} during session optimization", msg);
                 comm.endpoint_manager.delayed_messages.push((recv_index, msg));
+            }
+        }
+    }
     log!(
         cu.inner.logger,
         "Gathered all children's maps. ConnectorId set is... {:?}",
         unoptimized_map.keys()
     );
     let my_session_info = SessionInfo {
         port_info: cu.inner.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.inner.id_manager.connector_id, my_session_info);
     log!(
         cu.inner.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.inner.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.inner.logger,
                 "Session scatter recv loop. awaiting info from children {:?}...",
                 awaiting.iter()
             );
             let (recv_index, msg) =
                 comm.endpoint_manager.try_recv_any_setup(&mut *cu.inner.logger, deadline)?;
             log!(cu.inner.logger, "Received from index {:?} msg {:?}", &recv_index, &msg);
             match msg {
                 S(Sm::SessionScatter { optimized_map }) => {
                     if recv_index != parent {
                         log!(cu.inner.logger, "I expected the scatter from my parent only!");
                         return Err(Ce::SetupAlgMisbehavior);
+                    }
                     break 'scatter_loop optimized_map;
+                }
                 msg @ Msg::CommMsg { .. } => {
                     log!(cu.inner.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.inner.logger, "discarding old message {:?} during election", msg);
+                }
+            }
+        }
     } else {
         // by computing it myself
         log!(cu.inner.logger, "I am the leader! I will optimize this session");
         leader_session_map_optimize(&mut *cu.inner.logger, unoptimized_map)?
     };
     log!(
         cu.inner.logger,
         "Optimized info map is {:?}. Sending to children {:?}",
         &optimized_map,
         comm.neighborhood.children.iter()
     );
     log!(cu.inner.logger, "All session info dumped!: {:#?}", &optimized_map);
     let optimized_info =
         optimized_map.get(&cu.inner.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.inner.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.inner.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
         .iter_mut()
         .zip(endpoint_incoming_to_getter)
+    {
         ee.getter_for_incoming = getter;
+    }
     Ok(())
+}

0 comments (0 inline, 0 general)