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

Changeset - a3c92705eeee

Parent rev.

Child rev.

[Not reviewed]

0 10 4

Christopher Esterhuyse - 5 years ago 2020-07-14 11:39:53
christopher.esterhuyse@gmail.com

bugfix: native component branch forks that clash are MERGED rather than overwritten. Avoids race condition where (1) branch x ends and submits a solution, (2) branch y is created, has same predicate as x and overwrites it, but has a subset of its messages, (3) round ends in success but branch x is gone, so no suitable native branch is found

14 files changed with 405 insertions and 56 deletions:

examples/eg_protocols.pdl

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/utility.c

src/common.rs

src/ffi/mod.rs

src/runtime/communication.rs

src/runtime/logging.rs

src/runtime/mod.rs

src/runtime/tests.rs

0 comments (0 inline, 0 general)

examples/eg_protocols.pdl

➞

Show inline comments

 primitive pres_2(in i, out o) {
   synchronous {
     //put(o, get(i));
+  }
+}
 primitive together(in ia, in ib, out oa, out ob){
-  while(true) synchronous() {
   while(true) synchronous {
     if(fires(ia)) {
       put(oa, get(ia));
       put(ob, get(ib));
+    }
+  }
+}

examples/pres_1/amy.c

➞

Show inline comments

 #include "../../reowolf.h"
 #include "../utility.c"
 int main(int argc, char** argv) {
 	// Create a connector...
 	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);
-	printf("Error str `%s`\n", reowolf_error_peek(NULL));
+	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);
 	printf("Error str `%s`\n", reowolf_error_peek(NULL));
 	connector_add_net_port(c, &p0, addr_str, sizeof(addr_str)-1,
 			Polarity_Putter, EndpointPolarity_Passive);
 	rw_err_peek(c);
-	// Connect! Begin communication. 5000ms timeout
+	// Connect with peers (5000ms timeout).
 	connector_connect(c, 5000);
-	printf("Error str `%s`\n", reowolf_error_peek(NULL));
+	rw_err_peek(c);
 	// Ask to receive a message...
 	connector_get(c, p0);
 	printf("Error str `%s`\n", reowolf_error_peek(NULL));
 	// Prepare a message to send
 	connector_put_bytes(c, p0, msgbuf, msglen);
 	rw_err_peek(c);
-	// ... or timeout within 1000ms.
+	// ... reach new consistent state within 1000ms deadline.
 	connector_sync(c, 1000);
 	printf("Error str `%s`\n", reowolf_error_peek(NULL));
 	// Print the message we received
 	size_t msg_len;
 	const char * msg_ptr = connector_gotten_bytes(c, p0, &msg_len);
 	printf("Got msg `%.*s`\n", msg_len, msg_ptr);
 	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...
+	// 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);
-	printf("Error str `%s`\n", reowolf_error_peek(NULL));
+	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);
 	printf("Error str `%s`\n", reowolf_error_peek(NULL));
 	connector_add_net_port(c, &p0, addr_str, sizeof(addr_str)-1,
 			Polarity_Getter, EndpointPolarity_Active);
 	rw_err_peek(c);
-	// Connect (5000ms timeout). Begin communication.
+	// Connect with peers (5000ms timeout).
 	connector_connect(c, 5000);
-	printf("Error str `%s`\n", reowolf_error_peek(NULL));
+	rw_err_peek(c);
 	// Send a single 2-byte message...
 	connector_put_bytes(c, p0, "hi", 2);
 	printf("Error str `%s`\n", reowolf_error_peek(NULL));
 	// Prepare to receive a message.
 	connector_get(c, p0);
 	rw_err_peek(c);
-	// ... and acknowledge receipt within 1000ms.
+	// ... reach new consistent state within 1000ms deadline.
 	connector_sync(c, 1000);
 	printf("Error str `%s`\n", reowolf_error_peek(NULL));
 	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

@@ new file 100644 @@
 #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);
 	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

@@ new file 100644 @@
 #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);
 	rw_err_peek(c);
 	addr = "127.0.0.1:8001";
 	connector_add_net_port(c, &ports[1], addr, strlen(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

@@ new file 100644 @@
 #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);
 	rw_err_peek(c);
 	addr = "127.0.0.1:8001";
 	connector_add_net_port(c, &ports[1], addr, strlen(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

@@ new file 100644 @@
 #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);
 	rw_err_peek(c);
 	addr = "127.0.0.1:8001";
 	connector_add_net_port(c, &ports[1], addr, strlen(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/utility.c

➞

Show inline comments

 #include <stdio.h>
 #include <stdlib.h>
 #include <errno.h>
 #include <string.h>
 #include <unistd.h>
 #include "../reowolf.h"
 size_t get_user_msg(char * buf, size_t cap) {
 	memset(buf, 0, cap);
 	printf("Insert a msg of max len %d: ", cap);
 	fgets(buf, cap, stdin);
 	for(size_t len = 0; len<cap; len++)
 		if(buf[len]==0 || buf[len]=='\n')
 			return len;
 	return cap;
+}
 void rw_err_peek(Connector * c) {
 	printf("Error str `%s`\n", reowolf_error_peek(NULL));
+}
 // allocates a buffer!
 char * buffer_pdl(char * filename) {
 	FILE *f = fopen(filename, "rb");
 	if (f == NULL) {
 		printf("Opening pdl file returned errno %d!\n", errno);
 		exit(1);
+	}
 	fseek(f, 0, SEEK_END);
 	long fsize = ftell(f);
 	fseek(f, 0, SEEK_SET);
 	char *pdl = malloc(fsize + 1);
 	fread(pdl, 1, fsize, f);
 	fclose(f);
 	pdl[fsize] = 0;
 	return pdl;
+}

src/common.rs

➞

Show inline comments

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

src/ffi/mod.rs

➞

Show inline comments

@@ @@ -332,107 +332,109 @@ pub unsafe extern "C" fn connector_connect( @@
 //     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,

src/runtime/communication.rs

➞

Show inline comments

 use super::*;
 use crate::common::*;
 ////////////////
 #[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>, // native branch is ended iff to_get.is_empty()
     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 {
     inbox: HashMap<PortId, Payload>,
     state: ComponentState,
     untaken_choice: Option<u16>,
     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 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)
+    }
@@ @@ -203,159 +203,163 @@ impl Connector { @@
                 logger: &mut *cu.logger,
                 port_info: &mut cu.port_info,
                 id_manager: &mut cu.id_manager,
                 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.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.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.logger,
                     "Children in subtree are: {:?}",
                     subtree_id_iter.clone().collect::<Vec<_>>()
                 );
                 SolutionStorage::new(subtree_id_iter)
             },
             spec_var_stream: cu.id_manager.new_spec_var_stream(),
             getter_buffer: Default::default(),
             deadline: timeout.map(|to| Instant::now() + to),
         };
         log!(cu.logger, "Round context structure initialized");
         // Explore all native branches eagerly. Find solutions, buffer messages, etc.
         log!(
             cu.logger,
             "Translating {} native batches into branches...",
             comm.native_batches.len()
         );
         let native_branch_spec_var = rctx.spec_var_stream.next();
         log!(cu.logger, "Native branch spec var is {:?}", native_branch_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.port_info.spec_var_for(port);
                     predicate.assigned.insert(var, SpecVal::FIRING);
+                }
                 // assign falses for all silent (not firing) ports
                 for &port in cu.native_ports.difference(&firing_ports) {
                     let var = cu.port_info.spec_var_for(port);
                     if let Some(SpecVal::FIRING) = predicate.assigned.insert(var, SpecVal::SILENT) {
                         log!(cu.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_branch_spec_var, branch_spec_val)
             };
             log!(cu.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.logger, "Native branch {} sending msg {:?}", index, &msg);
                 rctx.getter_buffer.putter_add(cu, putter, msg);
+            }
             if to_get.is_empty() {
                 // this branch is immediately ready to be part of a solution
             let branch = NativeBranch { index, gotten: Default::default(), to_get };
             if branch.is_ended() {
                 log!(
                     cu.logger,
                     "Native submitting solution for batch {} with {:?}",
                     index,
                     &predicate
                 );
                 rctx.solution_storage.submit_and_digest_subtree_solution(
                     &mut *cu.logger,
                     SubtreeId::LocalComponent(ComponentId::Native),
                     predicate.clone(),
                 );
+            }
             let branch = NativeBranch { index, gotten: Default::default(), to_get };
             if let Some(_) = branching_native.branches.insert(predicate, branch) {
                 // thanks to the native_branch_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.logger, &mut rctx.spec_var_stream);
         // 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.logger, "Committing to decision {:?}!", &decision);
         comm.endpoint_manager.udp_endpoints_round_end(&mut *cu.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.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()
@@ @@ -604,197 +608,238 @@ impl Connector { @@
                                             } else {
                                                 log!(cu.logger, "Already requested failure");
+                                            }
+                                        }
+                                    }
+                                }
+                            }
                         } else {
                             log!(
                                 cu.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.logger,
                                 "Discarding announcement {:?} from non-parent endpoint idx {:?}",
                                 &decision,
                                 net_index
                             );
+                        }
+                    }
+                }
+            }
             log!(cu.logger, "Endpoint msg recv done");
+        }
+    }
     fn request_failure(
         cu: &mut ConnectorUnphased,
         comm: &mut ConnectorCommunication,
         parent: usize,
     ) -> Result<(), UnrecoverableSyncError> {
         log!(cu.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,
         solution_storage: &mut SolutionStorage,
         getter: PortId,
         send_payload_msg: &SendPayloadMsg,
     ) {
         log!(cu.logger, "feeding native getter {:?} {:?}", getter, &send_payload_msg);
         assert!(cu.port_info.polarities.get(&getter).copied() == Some(Getter));
         let mut draining = HashMap::default();
         let finished = &mut self.branches;
         std::mem::swap(&mut draining, finished);
         for (predicate, mut branch) in draining.drain() {
             log!(cu.logger, "visiting native branch {:?} with {:?}", &branch, &predicate);
             // check if this branch expects to receive it
             let var = cu.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());
                 branch.to_get.remove(&getter);
                 if branch.to_get.is_empty() {
                 if branch.is_ended() {
                     log!(
                         cu.logger,
-                        "new native solution with {:?} (to_get.is_empty()) with gotten {:?}",
+                        "new native solution with {:?} is_ended() with gotten {:?}",
                         &predicate,
                         &branch.gotten
                     );
                     let subtree_id = SubtreeId::LocalComponent(ComponentId::Native);
                     solution_storage.submit_and_digest_subtree_solution(
                         &mut *cu.logger,
                         subtree_id,
                         predicate.clone(),
                     );
                 } else {
                     log!(
                         cu.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.logger,
                     "skipping branch with {:?} that doesn't want the message (fastpath)",
                     &predicate
                 );
-                finished.insert(predicate, branch);
+                Self::fold_into(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.logger,
                         "skipping branch with {:?} that doesn't want the message (slowpath)",
                         &predicate
                     );
-                    finished.insert(predicate, branch);
+                    Self::fold_into(finished, predicate, branch);
+                }
                 Aur::Equivalent | Aur::FormerNotLatter => {
                     // retain the existing predicate, but add this payload
                     feed_branch(&mut branch, &predicate);
                     log!(cu.logger, "branch pred covers it! Accept the msg");
-                    finished.insert(predicate, branch);
+                    Self::fold_into(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.logger,
                         "payload pred {:?} covers branch pred {:?}",
                         &predicate2,
                         &predicate
                     );
                     finished.insert(predicate, branch);
                     finished.insert(predicate2, branch2);
                     Self::fold_into(finished, predicate, branch);
                     Self::fold_into(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.logger,
                         "new subsuming pred created {:?}. forking and feeding",
                         &predicate2
                     );
                     finished.insert(predicate, branch);
                     finished.insert(predicate2, branch2);
                     Self::fold_into(finished, predicate, branch);
                     Self::fold_into(finished, predicate2, branch2);
+                }
+            }
+        }
+    }
     fn fold_into(
         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) => {
                 ev.insert(branch);
+            }
             Entry::Occupied(mut eo) => {
                 let b = eo.get_mut();
                 for (k, v) in branch.gotten.drain() {
                     if b.gotten.insert(k, v).is_none() {
                         b.to_get.remove(&k);
+                    }
+                }
+            }
+        }
         // if let Some(prev) = branches.insert(predicate, branch)
+    }
     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 {
             if branch.to_get.is_empty() && branch_predicate.assigns_subset(solution_predicate) {
             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.cylic_drain(|mut predicate, mut branch, mut drainer| {
             let mut ctx = SyncProtoContext {
                 untaken_choice: &mut branch.untaken_choice,
                 logger: &mut *cu.logger,
                 predicate: &predicate,
                 port_info: &cu.port_info,
                 inbox: &branch.inbox,
             };
             let blocker = branch.state.sync_run(&mut ctx, &cu.proto_description);
             log!(
                 cu.logger,
                 "Proto component with id {:?} branch with pred {:?} hit blocker {:?}",
                 proto_component_id,
                 &predicate,
                 &blocker,
             );
             use SyncBlocker as B;
             match blocker {
                 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.untaken_choice = Some(val.0);
                         drainer.add_input(pred, branch_n);
+                    }
+                }
                 B::Inconsistent => {
                     // branch is inconsistent. throw it away
                     drop((predicate, branch));
+                }
                 B::SyncBlockEnd => {

src/runtime/logging.rs

➞

Show inline comments

 use super::*;
 fn secs_since_unix_epoch() -> f64 {
     std::time::SystemTime::now()
         .duration_since(std::time::UNIX_EPOCH)
         .map(|dur| dur.as_secs_f64())
         .unwrap_or(0.)
+}
 impl FileLogger {
     pub fn new(connector_id: ConnectorId, file: std::fs::File) -> Self {
         Self(connector_id, file)
+    }
+}
 impl VecLogger {
     pub fn new(connector_id: ConnectorId) -> Self {
         Self(connector_id, Default::default())
+    }
+}
 /////////////////
 impl Logger for DummyLogger {
     fn line_writer(&mut self) -> Option<&mut dyn std::io::Write> {
         None
+    }
+}
 impl Logger for VecLogger {
     fn line_writer(&mut self) -> Option<&mut dyn std::io::Write> {
-        let _ = write!(&mut self.1, "CID({}) at {:?} ", self.0, Instant::now());
+        let _ = write!(&mut self.1, "CID({}) at {:.6} ", self.0, secs_since_unix_epoch());
         Some(self)
+    }
+}
 impl Logger for FileLogger {
     fn line_writer(&mut self) -> Option<&mut dyn std::io::Write> {
-        let _ = write!(&mut self.1, "CID({}) at {:?} ", self.0, Instant::now());
+        let _ = write!(&mut self.1, "CID({}) at {:.6} ", self.0, secs_since_unix_epoch());
         Some(&mut self.1)
+    }
+}
 ///////////////////
 impl Drop for VecLogger {
     fn drop(&mut self) {
         let stdout = std::io::stderr();
         let mut lock = stdout.lock();
         writeln!(lock, "--- DROP LOG DUMP ---").unwrap();
         let _ = std::io::Write::write(&mut lock, self.1.as_slice());
+    }
+}
 impl std::io::Write for VecLogger {
     fn flush(&mut self) -> Result<(), std::io::Error> {
         Ok(())
+    }
     fn write(&mut self, data: &[u8]) -> Result<usize, std::io::Error> {
         self.1.extend_from_slice(data);
         Ok(data.len())
+    }
+}

src/runtime/mod.rs

➞

Show inline comments

@@ @@ -306,100 +306,102 @@ impl<T: std::cmp::Ord> VecSet<T> { @@
         vec.sort();
         vec.dedup();
         Self { vec }
+    }
     fn contains(&self, element: &T) -> bool {
         self.vec.binary_search(element).is_ok()
+    }
     fn insert(&mut self, element: T) -> bool {
         match self.vec.binary_search(&element) {
             Ok(_) => false,
             Err(index) => {
                 self.vec.insert(index, element);
                 true
+            }
+        }
+    }
     fn iter(&self) -> std::slice::Iter<T> {
         self.vec.iter()
+    }
     fn pop(&mut self) -> Option<T> {
         self.vec.pop()
+    }
+}
 impl PortInfo {
     fn spec_var_for(&self, port: PortId) -> SpecVar {
         SpecVar(match self.polarities.get(&port).unwrap() {
             Getter => port,
             Putter => *self.peers.get(&port).unwrap(),
         })
+    }
+}
 impl SpecVarStream {
     fn next(&mut self) -> SpecVar {
         let phantom_port: PortId =
             Id { connector_id: self.connector_id, u32_suffix: self.port_suffix_stream.next() }
                 .into();
         SpecVar(phantom_port)
+    }
+}
 impl IdManager {
     fn new(connector_id: ConnectorId) -> Self {
         Self {
             connector_id,
             port_suffix_stream: Default::default(),
             proto_component_suffix_stream: Default::default(),
+        }
+    }
     fn new_spec_var_stream(&self) -> SpecVarStream {
         SpecVarStream {
             connector_id: self.connector_id,
             port_suffix_stream: self.port_suffix_stream.clone(),
         let mut port_suffix_stream = self.port_suffix_stream.clone();
         const JUMP_OVER: usize = 100; // Jumping is entirely unnecessary. It's only used to make spec vars easier to spot in logs
         for _ in 0..JUMP_OVER {
             port_suffix_stream.next(); // throw away an ID
+        }
         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.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.logger, &mut new_logger);
         new_logger
+    }
     pub fn get_logger(&mut self) -> &mut dyn Logger {
         &mut *self.unphased.logger
+    }
     pub fn new_port_pair(&mut self) -> [PortId; 2] {
         let cu = &mut self.unphased;
         // adds two new associated ports, related to each other, and exposed to the native
         let [o, i] = [cu.id_manager.new_port_id(), cu.id_manager.new_port_id()];
         cu.native_ports.insert(o);
         cu.native_ports.insert(i);
         // {polarity, peer, route} known. {} unknown.
         cu.port_info.polarities.insert(o, Putter);
         cu.port_info.polarities.insert(i, Getter);
         cu.port_info.peers.insert(o, i);
         cu.port_info.peers.insert(i, o);
         let route = Route::LocalComponent(ComponentId::Native);
         cu.port_info.routes.insert(o, route);
@@ @@ -512,110 +514,122 @@ impl Predicate { @@
                         assert_eq!(sid, oid);
                         // both predicates assign the variable but differ on the value
                         return Aur::Nonexistant;
                     } else {
                         // both predicates assign the variable to the same value
                         s = s_it.next();
                         o = o_it.next();
+                    }
+                }
+            }
+        }
         // Observed zero inconsistencies. A unified predicate exists...
         match [s_not_o.is_empty(), o_not_s.is_empty()] {
             [true, true] => Aur::Equivalent,       // ... equivalent to both.
             [false, true] => Aur::FormerNotLatter, // ... equivalent to self.
             [true, false] => Aur::LatterNotFormer, // ... equivalent to other.
             [false, false] => {
                 // ... which is the union of the predicates' assignments but
                 //     is equivalent to neither self nor other.
                 let mut new = self.clone();
                 for (&id, &b) in o_not_s {
                     new.assigned.insert(id, b);
+                }
                 Aur::New(new)
+            }
+        }
+    }
     pub fn union_with(&self, other: &Self) -> Option<Self> {
         let mut res = self.clone();
         for (&channel_id, &assignment_1) in other.assigned.iter() {
             match res.assigned.insert(channel_id, assignment_1) {
                 Some(assignment_2) if assignment_1 != assignment_2 => return None,
                 _ => {}
+            }
+        }
         Some(res)
+    }
     pub fn query(&self, var: SpecVar) -> Option<SpecVal> {
         self.assigned.get(&var).copied()
+    }
+}
 impl<T: Debug + std::cmp::Ord> Debug for VecSet<T> {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         f.debug_set().entries(self.vec.iter()).finish()
+    }
+}
 impl Debug for Predicate {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         f.debug_tuple("Predicate").field(&self.assigned).finish()
         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 {
         f.debug_tuple("vrID").field(&self.0).finish()
         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/tests.rs

➞

Show inline comments

@@ @@ -746,48 +746,91 @@ fn udp_to_reowolf() { @@
         });
         s.spawn(|_| {
             barrier.wait();
             // udp thread
             let udp = std::net::UdpSocket::bind(sock_addrs[1]).unwrap();
             udp.connect(sock_addrs[0]).unwrap();
             for _ in 0..15 {
                 udp.send(TEST_MSG_BYTES).unwrap();
                 std::thread::sleep(MS100.unwrap());
+            }
             barrier.wait();
         });
     })
     .unwrap();
+}
 #[test]
 fn udp_reowolf_swap() {
     let test_log_path = Path::new("./logs/udp_reowolf_swap");
     let sock_addrs = [next_test_addr(), next_test_addr()];
     let barrier = std::sync::Barrier::new(2);
     scope(|s| {
         s.spawn(|_| {
             barrier.wait();
             // reowolf thread
             let mut c = file_logged_connector(0, test_log_path);
             let [p0, p1] = c.new_udp_port(sock_addrs[0], sock_addrs[1]).unwrap();
             c.connect(SEC1).unwrap();
             c.put(p0, TEST_MSG.clone()).unwrap();
             c.get(p1).unwrap();
             c.sync(SEC5).unwrap();
             assert_eq!(c.gotten(p1).unwrap().as_slice(), TEST_MSG_BYTES);
             barrier.wait();
         });
         s.spawn(|_| {
             barrier.wait();
             // udp thread
             let udp = std::net::UdpSocket::bind(sock_addrs[1]).unwrap();
             udp.connect(sock_addrs[0]).unwrap();
             let mut buf = new_u8_buffer(256);
             udp.send(TEST_MSG_BYTES).unwrap();
             let len = udp.recv(&mut buf).unwrap();
             assert_eq!(TEST_MSG_BYTES, &buf[0..len]);
             barrier.wait();
         });
     })
     .unwrap();
+}
 #[test]
 fn pres_3() {
     let test_log_path = Path::new("./logs/pres_3");
     let sock_addrs = [next_test_addr(), next_test_addr()];
     scope(|s| {
         s.spawn(|_| {
             // "amy"
             let mut c = file_logged_connector(0, test_log_path);
             let p0 = c.new_net_port(Putter, sock_addrs[0], Active).unwrap();
             let p1 = c.new_net_port(Putter, sock_addrs[1], Active).unwrap();
             c.connect(SEC1).unwrap();
             // put {A} and FAIL
             c.put(p0, TEST_MSG.clone()).unwrap();
             c.sync(SEC1).unwrap_err();
             // put {B} and FAIL
             c.put(p1, TEST_MSG.clone()).unwrap();
             c.sync(SEC1).unwrap_err();
             // put {A, B} and SUCCEED
             c.put(p0, TEST_MSG.clone()).unwrap();
             c.put(p1, TEST_MSG.clone()).unwrap();
             c.sync(SEC1).unwrap();
         });
         s.spawn(|_| {
             // "bob"
             let mut c = file_logged_connector(1, test_log_path);
             let p0 = c.new_net_port(Getter, sock_addrs[0], Passive).unwrap();
             let p1 = c.new_net_port(Getter, sock_addrs[1], Passive).unwrap();
             c.connect(SEC1).unwrap();
             for _ in 0..2 {
                 // get {A, B} and FAIL
                 c.get(p0).unwrap();
                 c.get(p1).unwrap();
                 c.sync(SEC1).unwrap_err();
+            }
             // get {A, B} and SUCCEED
             c.get(p0).unwrap();
             c.get(p1).unwrap();
             c.sync(SEC1).unwrap();
         });
     })
     .unwrap();
+}

0 comments (0 inline, 0 general)