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

Changeset - 700221108e9f

Parent rev.

Child rev.

[Not reviewed]

0 8 0

Christopher Esterhuyse - 5 years ago 2020-07-20 09:55:39
christopher.esterhuyse@gmail.com

socket api relaxed. bind and connect in any order. bind once and connect any number of times

8 files changed with 143 insertions and 136 deletions:

Cargo.toml

src/common.rs

src/ffi/mod.rs

src/ffi/socket_api.rs

src/runtime/communication.rs

src/runtime/mod.rs

src/runtime/setup.rs

src/runtime/tests.rs

0 comments (0 inline, 0 general)

Cargo.toml

➞

Show inline comments

@@ @@ -10,35 +10,36 @@ edition = "2018" @@
 [dependencies]
 # convenience macros
 maplit = "1.0.2"
 derive_more = "0.99.2"
 # runtime
 bincode = "1.3.1"
 serde = { version = "1.0.114", features = ["derive"] }
 getrandom = "0.1.14" # tiny crate. used to guess controller-id
 # network
 mio = { version = "0.7.0", package = "mio", features = ["udp", "tcp", "os-poll"] }
 socket2 = { version = "0.3.12", optional = true }
 # protocol
 backtrace = "0.3"
 lazy_static = "1.4.0"
 # socket ffi
 atomic_refcell = { version = "0.1.6", optional = true }
 [dev-dependencies]
 # test-generator = "0.3.0"
 crossbeam-utils = "0.7.2"
 lazy_static = "1.4.0"
 [lib]
 # compile target: dynamically linked library using C ABI
 crate-type = ["cdylib"]
 [features]
 default = ["ffi", "ffi_socket_api"] # // "session_optimization",
 ffi = [] # see src/ffi.rs
 ffi_socket_api = ["ffi", "atomic_refcell"]
+ffi_socket_api = ["ffi", "atomic_refcell", "socket2"]
 endpoint_logging = [] # see src/macros.rs
 session_optimization = [] # see src/runtime/setup.rs
@@ \ No newline at end of file @@

src/common.rs

➞

Show inline comments

@@ @@ -103,24 +103,28 @@ 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
+    }
     pub(crate) fn n_skipped(mut self, n: u32) -> Self {
         self.next = self.next.saturating_add(n);
         self
+    }
+}
 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 {

src/ffi/mod.rs

➞

Show inline comments

@@ @@ -75,24 +75,25 @@ unsafe fn tl_socketaddr_from_raw( @@
         .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 CLOSE_FAIL: c_int = -4;
 pub const BAD_FD: c_int = -5;
 pub const CONNECT_FAILED: c_int = -6;
 ///////////////////// 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() {
@@ @@ -147,44 +148,44 @@ pub unsafe extern "C" fn connector_new_logging( @@
     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, 8);
             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(), 8);
     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,

src/ffi/socket_api.rs

➞

Show inline comments

 use super::*;
 use atomic_refcell::AtomicRefCell;
 use std::{collections::HashMap, ffi::c_void, net::SocketAddr, os::raw::c_int, sync::RwLock};
 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>,
+}
 enum MaybeConnector {
     New,
     Bound { local_addr: SocketAddr },
     Connected { connector: Connector, putter: PortId, getter: PortId },
 struct ConnectorBound {
     connector: Connector,
     putter: PortId,
     getter: PortId,
+}
 struct MaybeConnector {
     // 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 ConnectorStorage {
     fd_to_connector: HashMap<c_int, AtomicRefCell<MaybeConnector>>,
 struct CspStorage {
     fd_to_mc: HashMap<c_int, AtomicRefCell<MaybeConnector>>,
     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 {
@@ @@ -35,136 +53,130 @@ impl FdAllocator { @@
+        }
         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 CONNECTOR_STORAGE: RwLock<ConnectorStorage> = Default::default();
+    static ref CSP_STORAGE: RwLock<CspStorage> = Default::default();
+}
 ///////////////////////////////////////////////////////////////////
 #[no_mangle]
 pub extern "C" fn rw_socket(_domain: c_int, _type: c_int) -> c_int {
     // assuming _domain is AF_INET and _type is SOCK_DGRAM
     let mut w = if let Ok(w) = CONNECTOR_STORAGE.write() { w } else { return FD_LOCK_POISONED };
     // ignoring domain and type
     let mut w = if let Ok(w) = CSP_STORAGE.write() { w } else { return FD_LOCK_POISONED };
     let fd = w.fd_allocator.alloc();
     w.fd_to_connector.insert(fd, AtomicRefCell::new(MaybeConnector::New));
     let mc = MaybeConnector { peer_addr: trivial_peer_addr(), connector_bound: None };
     w.fd_to_mc.insert(fd, AtomicRefCell::new(mc));
     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) = CONNECTOR_STORAGE.write() { w } else { return FD_LOCK_POISONED };
+    let mut w = if let Ok(w) = CSP_STORAGE.write() { w } else { return FD_LOCK_POISONED };
     w.fd_allocator.free(fd);
-    if w.fd_to_connector.remove(&fd).is_some() {
+    if w.fd_to_mc.remove(&fd).is_some() {
         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 {
     use MaybeConnector as Mc;
     // assuming _domain is AF_INET and _type is SOCK_DGRAM
     let r = if let Ok(r) = CONNECTOR_STORAGE.read() { r } else { return FD_LOCK_POISONED };
     let mc = if let Some(mc) = r.fd_to_connector.get(&fd) { mc } else { return BAD_FD };
     let mc: &mut Mc = &mut mc.borrow_mut();
     let _ = if let Mc::New = mc { () } else { return WRONG_STATE };
     *mc = Mc::Bound { local_addr: local_addr.read() };
     let r = if let Ok(r) = CSP_STORAGE.read() { r } else { return FD_LOCK_POISONED };
     let mc = if let Some(mc) = r.fd_to_mc.get(&fd) { mc } else { return BAD_FD };
     let mc: &mut MaybeConnector = &mut mc.borrow_mut();
     if mc.connector_bound.is_some() {
         return WRONG_STATE;
+    }
     mc.connector_bound = {
         let mut connector = Connector::new(
             Box::new(crate::DummyLogger),
             crate::TRIVIAL_PD.clone(),
             Connector::random_id(),
         );
         let [putter, getter] = connector.new_udp_port(local_addr.read(), mc.peer_addr).unwrap();
         Some(ConnectorBound { connector, putter, getter })
     };
     ERR_OK
+}
 #[no_mangle]
 pub unsafe extern "C" fn rw_connect(
     fd: c_int,
     peer_addr: *const SocketAddr,
     _address_len: usize,
 ) -> c_int {
     use MaybeConnector as Mc;
     // assuming _domain is AF_INET and _type is SOCK_DGRAM
     let r = if let Ok(r) = CONNECTOR_STORAGE.read() { r } else { return FD_LOCK_POISONED };
     let mc = if let Some(mc) = r.fd_to_connector.get(&fd) { mc } else { return BAD_FD };
     let mc: &mut Mc = &mut mc.borrow_mut();
     let local_addr =
         if let Mc::Bound { local_addr } = mc { local_addr } else { return WRONG_STATE };
     let peer_addr = peer_addr.read();
     let (connector, [putter, getter]) = {
         let mut c = Connector::new(
             Box::new(DummyLogger),
             crate::TRIVIAL_PD.clone(),
             Connector::random_id(),
 ,
         );
         let [putter, getter] = c.new_udp_port(*local_addr, peer_addr).unwrap();
         (c, [putter, getter])
     };
     *mc = Mc::Connected { connector, putter, getter };
     let r = if let Ok(r) = CSP_STORAGE.read() { r } else { return FD_LOCK_POISONED };
     let mc = if let Some(mc) = r.fd_to_mc.get(&fd) { mc } else { return BAD_FD };
     let mc: &mut MaybeConnector = &mut mc.borrow_mut();
     mc.peer_addr = peer_addr.read();
     if let Some(ConnectorBound { connector, .. }) = &mut mc.connector_bound {
         if connector.get_mut_udp_sock(0).unwrap().connect(mc.peer_addr).is_err() {
             return CONNECT_FAILED;
+        }
+    }
     ERR_OK
+}
 #[no_mangle]
 pub unsafe extern "C" fn rw_send(
     fd: c_int,
     bytes_ptr: *const c_void,
     bytes_len: usize,
     _flags: c_int,
 ) -> isize {
     use MaybeConnector as Mc;
     // ignoring flags
     let r =
         if let Ok(r) = CONNECTOR_STORAGE.read() { r } else { return FD_LOCK_POISONED as isize };
     let mc = if let Some(mc) = r.fd_to_connector.get(&fd) { mc } else { return BAD_FD as isize };
     let mc: &mut Mc = &mut mc.borrow_mut();
     let (connector, putter) = if let Mc::Connected { connector, putter, .. } = mc {
         (connector, *putter)
     let r = if let Ok(r) = CSP_STORAGE.read() { r } else { return FD_LOCK_POISONED as isize };
     let mc = if let Some(mc) = r.fd_to_mc.get(&fd) { mc } else { return BAD_FD as isize };
     let mc: &mut MaybeConnector = &mut mc.borrow_mut();
     if let Some(ConnectorBound { connector, putter, .. }) = &mut mc.connector_bound {
         match connector_put_bytes(connector, *putter, bytes_ptr as _, bytes_len) {
             ERR_OK => connector_sync(connector, -1),
             err => err as isize,
+        }
     } else {
         return WRONG_STATE as isize;
     };
     match connector_put_bytes(connector, putter, bytes_ptr as _, bytes_len) {
         ERR_OK => {}
         err => return err as isize,
         WRONG_STATE as isize // not bound!
+    }
     connector_sync(connector, -1)
+}
 #[no_mangle]
 pub unsafe extern "C" fn rw_recv(
     fd: c_int,
     bytes_ptr: *mut c_void,
     bytes_len: usize,
     _flags: c_int,
 ) -> isize {
     use MaybeConnector as Mc;
     // ignoring flags
     let r =
         if let Ok(r) = CONNECTOR_STORAGE.read() { r } else { return FD_LOCK_POISONED as isize };
     let mc = if let Some(mc) = r.fd_to_connector.get(&fd) { mc } else { return BAD_FD as isize };
     let mc: &mut Mc = &mut mc.borrow_mut();
     let (connector, getter) = if let Mc::Connected { connector, getter, .. } = mc {
         (connector, *getter)
     let r = if let Ok(r) = CSP_STORAGE.read() { r } else { return FD_LOCK_POISONED as isize };
     let mc = if let Some(mc) = r.fd_to_mc.get(&fd) { mc } else { return BAD_FD as isize };
     let mc: &mut MaybeConnector = &mut mc.borrow_mut();
     if let Some(ConnectorBound { connector, getter, .. }) = &mut mc.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 {
         return WRONG_STATE as isize;
     };
     match connector_get(connector, getter) {
         ERR_OK => {}
         err => return err as isize,
         WRONG_STATE as isize // not bound!
+    }
     match connector_sync(connector, -1) {
 => {} // singleton batch index
         err => return err as isize,
     };
     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
+}

src/runtime/communication.rs

➞

Show inline comments

@@ @@ -104,34 +104,35 @@ impl RoundCtxTrait for RoundCtx { @@
     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
+        }
+    }
     // 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)
     // }
     // #[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)?;
@@ @@ -229,27 +230,24 @@ impl Connector { @@
         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,
                 // logger: &mut *cu.inner.logger,
                 // port_info: &mut cu.inner.port_info,
                 // id_manager: &mut cu.inner.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.inner.logger,
                 "proto component {:?} ran to nonsync blocker {:?}",
@@ @@ -410,25 +408,24 @@ impl Connector { @@
                         .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");
         // dropping {solution_storage, payloads_to_get}
         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() {
@@ @@ -883,62 +880,34 @@ impl BranchingProtoComponent { @@
                 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::Inconsistent => {
                     // EXPLICIT inconsistency
                     drop((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);
+                }
                 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);
@@ @@ -953,24 +922,49 @@ impl BranchingProtoComponent { @@
                         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);
     //     }

src/runtime/mod.rs

➞

Show inline comments

@@ @@ -21,42 +21,40 @@ pub struct Connector { @@
     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,
     proto_component_ports: &'a mut HashSet<PortId>,
     unrun_components: &'a mut Vec<(ProtoComponentId, ProtoComponent)>,
     proto_component_id: ProtoComponentId,
     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,
     branch_inner: &'a mut ProtoComponentBranchInner,
     predicate: &'a Predicate,
     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>,
@@ @@ -238,25 +236,24 @@ struct ConnectorUnphased { @@
+}
 #[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>,
     surplus_sockets: u16,
+}
 #[derive(Debug)]
 enum ConnectorPhased {
     Setup(Box<ConnectorSetup>),
     Communication(Box<ConnectorCommunication>),
+}
 #[derive(Default, Clone, Eq, PartialEq, Hash, serde::Serialize, serde::Deserialize)]
 struct Predicate {
     assigned: BTreeMap<SpecVar, SpecVal>,
+}
 #[derive(Debug, Default)]
 struct NativeBatch {
@@ @@ -349,29 +346,30 @@ impl SpecVarStream { @@
         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 {
         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
+        }
         // 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()
+    }
@@ @@ -481,25 +479,24 @@ impl Predicate { @@
     //     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);

src/runtime/setup.rs

➞

Show inline comments

 use crate::common::*;
 use crate::runtime::*;
 impl Connector {
     pub fn new(
         mut logger: Box<dyn Logger>,
         proto_description: Arc<ProtocolDescription>,
         connector_id: ConnectorId,
         surplus_sockets: u16,
     ) -> Self {
         log!(&mut *logger, "Created with connector_id {:?}", connector_id);
         Self {
             unphased: ConnectorUnphased {
                 proto_description,
                 proto_components: Default::default(),
                 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(),
                 surplus_sockets,
             })),
+        }
+    }
     pub fn new_udp_port(
         &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) => {

src/runtime/tests.rs

➞

Show inline comments

@@ @@ -25,25 +25,25 @@ fn next_test_addr() -> SocketAddr { @@
 fn file_logged_connector(connector_id: ConnectorId, dir_path: &Path) -> Connector {
     file_logged_configured_connector(connector_id, dir_path, MINIMAL_PROTO.clone())
+}
 fn file_logged_configured_connector(
     connector_id: ConnectorId,
     dir_path: &Path,
     pd: Arc<ProtocolDescription>,
 ) -> Connector {
     let _ = std::fs::create_dir(dir_path); // we will check failure soon
     let path = dir_path.join(format!("cid_{:?}.txt", connector_id));
     let file = File::create(path).unwrap();
     let file_logger = Box::new(FileLogger::new(connector_id, file));
-    Connector::new(file_logger, pd, connector_id, 8)
     Connector::new(file_logger, pd, connector_id)
+}
 static MINIMAL_PDL: &'static [u8] = b"
 primitive together(in ia, in ib, out oa, out ob){
   while(true) synchronous() {
     if(fires(ia)) {
       put(oa, get(ia));
       put(ob, get(ib));
+    }
+  }
+}
 ";
 lazy_static::lazy_static! {
@@ @@ -58,25 +58,25 @@ lazy_static::lazy_static! { @@
     };
+}
 fn new_u8_buffer(cap: usize) -> Vec<u8> {
     let mut v = Vec::with_capacity(cap);
     // Safe! len will cover owned bytes in valid state
     unsafe { v.set_len(cap) }
+    v
+}
 //////////////////////////////////////////
 #[test]
 fn basic_connector() {
-    Connector::new(Box::new(DummyLogger), MINIMAL_PROTO.clone(), 0, 0);
     Connector::new(Box::new(DummyLogger), MINIMAL_PROTO.clone(), 0);
+}
 #[test]
 fn basic_logged_connector() {
     let test_log_path = Path::new("./logs/basic_logged_connector");
     file_logged_connector(0, test_log_path);
+}
 #[test]
 fn new_port_pair() {
     let test_log_path = Path::new("./logs/new_port_pair");
     let mut c = file_logged_connector(0, test_log_path);

0 comments (0 inline, 0 general)