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

Changeset - d23010006486

Parent rev.

Child rev.

[Not reviewed]

0 8 0

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

bugfix: increment round index on recovery to avoid mixing messages once we switch away from tcp. refactoring communication internals to simplify bookkeeping structures. more consts in tests to make them terser

8 files changed with 227 insertions and 129 deletions:

src/common.rs

src/protocol/library.rs

src/protocol/mod.rs

src/protocol/parser.rs

src/runtime/communication.rs

src/runtime/error.rs

src/runtime/mod.rs

src/runtime/tests.rs

133

0 comments (0 inline, 0 general)

src/common.rs

➞

Show inline comments

@@ @@ -60,48 +60,49 @@ 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 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 {

src/protocol/library.rs

➞

Show inline comments

 use crate::protocol::ast::*;
 use crate::protocol::inputsource::*;
 // TABBED OUT FOR NOW
 pub fn get_declarations(h: &mut Heap, i: ImportId) -> Result<Vec<DeclarationId>, ParseError> {
     if h[i].value == b"std.reo" {
         let mut vec = Vec::new();
         vec.push(cd(h, i, b"sync", &[Type::INPUT, Type::OUTPUT]));
         vec.push(cd(h, i, b"syncdrain", &[Type::INPUT, Type::INPUT]));
         vec.push(cd(h, i, b"syncspout", &[Type::OUTPUT, Type::OUTPUT]));
         vec.push(cd(h, i, b"asyncdrain", &[Type::INPUT, Type::INPUT]));
         vec.push(cd(h, i, b"asyncspout", &[Type::OUTPUT, Type::OUTPUT]));
         vec.push(cd(h, i, b"merger", &[Type::INPUT_ARRAY, Type::OUTPUT]));
         vec.push(cd(h, i, b"router", &[Type::INPUT, Type::OUTPUT_ARRAY]));
         vec.push(cd(h, i, b"consensus", &[Type::INPUT_ARRAY, Type::OUTPUT]));
         vec.push(cd(h, i, b"replicator", &[Type::INPUT, Type::OUTPUT_ARRAY]));
         vec.push(cd(h, i, b"alternator", &[Type::INPUT_ARRAY, Type::OUTPUT]));
         vec.push(cd(h, i, b"roundrobin", &[Type::INPUT, Type::OUTPUT_ARRAY]));
         vec.push(cd(h, i, b"node", &[Type::INPUT_ARRAY, Type::OUTPUT_ARRAY]));
         vec.push(cd(h, i, b"fifo", &[Type::INPUT, Type::OUTPUT]));
         vec.push(cd(h, i, b"xfifo", &[Type::INPUT, Type::OUTPUT, Type::MESSAGE]));
         vec.push(cd(h, i, b"nfifo", &[Type::INPUT, Type::OUTPUT, Type::INT]));
         vec.push(cd(h, i, b"ufifo", &[Type::INPUT, Type::OUTPUT]));
         Ok(vec)
     } else if h[i].value == b"std.buf" {
         let mut vec = Vec::new();
         vec.push(fd(h, i, b"writeByte", Type::BYTE, &[Type::MESSAGE, Type::INT, Type::BYTE]));
         vec.push(fd(h, i, b"writeShort", Type::SHORT, &[Type::MESSAGE, Type::INT, Type::SHORT]));

src/protocol/mod.rs

➞

Show inline comments

 mod arena;
 mod ast;
 mod eval;
 pub(crate) mod inputsource;
 mod lexer;
 mod library;
+// mod library;
 mod parser;
 use crate::common::*;
 use crate::protocol::ast::*;
 use crate::protocol::eval::*;
 use crate::protocol::inputsource::*;
 use crate::protocol::parser::*;
 #[derive(serde::Serialize, serde::Deserialize)]
 #[repr(C)]
 pub struct ProtocolDescription {
     heap: Heap,
     source: InputSource,
     root: RootId,
+}
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 pub(crate) struct ComponentState {
     prompt: Prompt,
+}
 pub(crate) enum EvalContext<'a> {
     Nonsync(&'a mut NonsyncProtoContext<'a>),
     Sync(&'a mut SyncProtoContext<'a>),
     // None,
+}
 //////////////////////////////////////////////
 impl std::fmt::Debug for ProtocolDescription {
     fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
-        write!(f, "(A big honkin' protocol description)")
+        write!(f, "(An opaque protocol description)")
+    }
+}
 impl ProtocolDescription {
     pub fn parse(buffer: &[u8]) -> Result<Self, String> {
         let mut heap = Heap::new();
         let mut source = InputSource::from_buffer(buffer).unwrap();
         let mut parser = Parser::new(&mut source);
         match parser.parse(&mut heap) {
             Ok(root) => {
                 return Ok(ProtocolDescription { heap, source, root });
+            }
             Err(err) => {
                 let mut vec: Vec<u8> = Vec::new();
                 err.write(&source, &mut vec).unwrap();
                 Err(String::from_utf8_lossy(&vec).to_string())
+            }
+        }
+    }
     pub(crate) fn component_polarities(
         &self,
         identifier: &[u8],
     ) -> Result<Vec<Polarity>, AddComponentError> {
         use AddComponentError::*;
         let h = &self.heap;

src/protocol/parser.rs

➞

Show inline comments

 use crate::protocol::ast::*;
 use crate::protocol::inputsource::*;
 use crate::protocol::lexer::*;
 use crate::protocol::library;
+// use crate::protocol::library;
 // The following indirection is needed due to a bug in the cbindgen tool.
 type Unit = ();
 type VisitorResult = Result<Unit, ParseError>;
 trait Visitor: Sized {
     fn visit_protocol_description(&mut self, h: &mut Heap, pd: RootId) -> VisitorResult {
         recursive_protocol_description(self, h, pd)
+    }
     fn visit_pragma(&mut self, _h: &mut Heap, _pragma: PragmaId) -> VisitorResult {
         Ok(())
+    }
     fn visit_import(&mut self, _h: &mut Heap, _import: ImportId) -> VisitorResult {
         Ok(())
+    }
     fn visit_symbol_definition(&mut self, h: &mut Heap, def: DefinitionId) -> VisitorResult {
         recursive_symbol_definition(self, h, def)
+    }
     fn visit_component_definition(&mut self, h: &mut Heap, def: ComponentId) -> VisitorResult {
         recursive_component_definition(self, h, def)
+    }
     fn visit_composite_definition(&mut self, h: &mut Heap, def: CompositeId) -> VisitorResult {
         recursive_composite_definition(self, h, def)
@@ @@ -776,54 +776,55 @@ impl BuildSymbolDeclarations { @@
                     Declaration::Defined(defined) => Err(ParseError::new(
                         h[defined.definition].position(),
                         format!("Defined symbol clash: {}", h[id]),
                     )),
                     Declaration::Imported(imported) => Err(ParseError::new(
                         h[imported.import].position(),
                         format!("Imported symbol clash: {}", h[id]),
                     )),
                 };
+            }
+        }
         self.declarations.push(decl);
         Ok(())
+    }
+}
 impl Visitor for BuildSymbolDeclarations {
     fn visit_protocol_description(&mut self, h: &mut Heap, pd: RootId) -> VisitorResult {
         recursive_protocol_description(self, h, pd)?;
         // Move all collected declarations to the protocol description
         h[pd].declarations.append(&mut self.declarations);
         Ok(())
+    }
     fn visit_import(&mut self, h: &mut Heap, import: ImportId) -> VisitorResult {
         let vec = library::get_declarations(h, import)?;
         // Destructively iterate over the vector
         for decl in vec {
             self.checked_add(h, decl)?;
+        }
         Ok(())
         todo!()
         // let vec = library::get_declarations(h, import)?;
         // // Destructively iterate over the vector
         // for decl in vec {
         //     self.checked_add(h, decl)?;
         // }
         // Ok(())
+    }
     fn visit_symbol_definition(&mut self, h: &mut Heap, definition: DefinitionId) -> VisitorResult {
         let signature = Signature::from_definition(h, definition);
         let decl = h
             .alloc_defined_declaration(|this| DefinedDeclaration { this, definition, signature })
             .upcast();
         self.checked_add(h, decl)?;
         Ok(())
+    }
+}
 struct LinkCallExpressions {
     pd: Option<RootId>,
     composite: bool,
     new_statement: bool,
+}
 impl LinkCallExpressions {
     fn new() -> Self {
         LinkCallExpressions { pd: None, composite: false, new_statement: false }
+    }
     fn get_declaration(
         &self,
         h: &Heap,

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()
+}
 #[derive(Debug)]
 struct SolutionStorage {
     old_local: HashSet<Predicate>,
     new_local: HashSet<Predicate>,
     // this pair acts as Route -> HashSet<Predicate> which is friendlier to iteration
     subtree_solutions: Vec<HashSet<Predicate>>,
     subtree_id_to_index: HashMap<Route, 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 PayloadMsgSender {
     fn putter_send(
         &mut self,
         cu: &mut ConnectorUnphased,
         putter: PortId,
         msg: SendPayloadMsg,
     ) -> Result<(), SyncError>;
+}
 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 Connector {
     pub fn gotten(&mut self, port: PortId) -> Result<&Payload, GottenError> {
         use GottenError as Ge;
         let Self { phased, .. } = self;
         match phased {
             ConnectorPhased::Setup { .. } => Err(Ge::NoPreviousRound),
             ConnectorPhased::Communication(comm) => 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),
             },
+        }
@@ @@ -111,74 +114,73 @@ impl Connector { @@
         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, phased } = self;
         match phased {
             ConnectorPhased::Setup { .. } => Err(SyncError::NotConnected),
             ConnectorPhased::Communication(comm) => {
                 comm.round_result = Self::connected_sync(unphased, 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;
         let deadline = timeout.map(|to| Instant::now() + to);
+        // let deadline = timeout.map(|to| Instant::now() + to);
         log!(
             cu.logger,
             "~~~ SYNC called with timeout {:?}; starting round {}",
             &timeout,
             comm.round_index
         );
         let mut spec_var_stream = cu.id_manager.new_spec_var_stream();
         // 1. run all proto components to Nonsync blockers
         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.logger, "Nonsync running {} proto components...", unrun_components.len());
         while let Some((proto_component_id, mut component)) = unrun_components.pop() {
             // TODO coalesce fields
             log!(
                 cu.logger,
                 "Nonsync running proto component with ID {:?}. {} to go after this",
                 proto_component_id,
                 unrun_components.len()
             );
             let mut ctx = NonsyncProtoContext {
                 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
@@ @@ -186,131 +188,132 @@ impl Connector { @@
             };
             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(),
         );
         // NOTE: all msgs in outbox are of form (Getter, Payload)
         let mut payloads_to_get: Vec<(PortId, SendPayloadMsg)> = vec![];
         // create the solution storage
         let mut solution_storage = {
         let mut rctx = RoundCtx {
             solution_storage: {
                 let n = std::iter::once(Route::LocalComponent(ComponentId::Native));
             let c =
                 cu.proto_components.keys().map(|&id| Route::LocalComponent(ComponentId::Proto(id)));
                 let c = cu
                     .proto_components
                     .keys()
                     .map(|&id| Route::LocalComponent(ComponentId::Proto(id)));
                 let e = comm.neighborhood.children.iter().map(|&index| Route::Endpoint { index });
                 SolutionStorage::new(n.chain(c).chain(e))
             },
             spec_var_stream: cu.id_manager.new_spec_var_stream(),
             getter_buffer: Default::default(),
             deadline: timeout.map(|to| Instant::now() + to),
         };
-        log!(cu.logger, "Solution storage initialized");
+        log!(cu.logger, "Round context structure initialized");
         // 2. kick off the native
         log!(
             cu.logger,
             "Translating {} native batches into branches...",
             comm.native_batches.len()
         );
         let native_branch_spec_var = spec_var_stream.next();
+        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().inserted(native_branch_spec_var, branch_spec_val);
                 // 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 silent 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;
+                    }
+                }
                 predicate
             };
             log!(cu.logger, "Native branch index={:?} has consistent {:?}", index, &predicate);
             // put all messages
             for (putter, payload) in to_put {
                 let msg = SendPayloadMsg { predicate: predicate.clone(), payload };
                 log!(cu.logger, "Native branch {} sending msg {:?}", index, &msg);
-                payloads_to_get.putter_send(cu, putter, msg)?;
+                rctx.getter_buffer.putter_add(cu, putter, msg)?;
+            }
             if to_get.is_empty() {
                 log!(
                     cu.logger,
                     "Native submitting solution for batch {} with {:?}",
                     index,
                     &predicate
                 );
                 solution_storage.submit_and_digest_subtree_solution(
+                rctx.solution_storage.submit_and_digest_subtree_solution(
                     &mut *cu.logger,
                     Route::LocalComponent(ComponentId::Native),
                     predicate.clone(),
                 );
+            }
             let branch = NativeBranch { index, gotten: Default::default(), to_get };
             if let Some(_existing) = branching_native.branches.insert(predicate, branch) {
             if let Some(_) = branching_native.branches.insert(predicate, branch) {
                 // thanks to the native_branch_spec_var, each batch has a distinct predicate
                 unreachable!()
                 // return Err(Se::IndistinguishableBatches([index, existing.index]));
+            }
+        }
         // restore the invariant
         comm.native_batches.push(Default::default());
         let decision = Self::sync_reach_decision(
             cu,
             comm,
             &mut branching_native,
             &mut branching_proto_components,
             solution_storage,
             payloads_to_get,
             deadline,
             &mut rctx,
         )?;
         log!(cu.logger, "Committing to decision {:?}!", &decision);
         // propagate the decision to children
         let msg = Msg::CommMsg(CommMsg {
             round_index: comm.round_index,
             contents: CommMsgContents::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
@@ @@ -319,281 +322,280 @@ impl Connector { @@
                     // consume branching proto components
                     branching_proto_components
                         .into_iter()
                         .map(|(id, bpc)| (id, bpc.collapse_with(&predicate))),
                 );
                 log!(
                     cu.logger,
                     "End round with (updated) component states {:?}",
                     cu.proto_components.keys()
                 );
                 // consume native
                 Ok(Some(branching_native.collapse_with(&mut *cu.logger, &predicate)))
+            }
         };
         log!(cu.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>,
         mut solution_storage: SolutionStorage,
         mut payloads_to_get: Vec<(PortId, SendPayloadMsg)>,
         mut deadline: Option<Instant>,
         rctx: &mut RoundCtx,
     ) -> Result<Decision, SyncError> {
         let mut already_requested_failure = false;
         if branching_native.branches.is_empty() {
             log!(cu.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.logger, "Already requested failure");
+                    }
+                }
                 None => {
                     log!(cu.logger, "No parent. Deciding on failure");
                     return Ok(Decision::Failure);
+                }
+            }
+        }
         log!(cu.logger, "Done translating native batches into branches");
         // run all proto components to their sync blocker
         log!(
             cu.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;
             let mut swap = HashMap::default();
             // initially, no components have .ended==true
             let mut blocked = HashMap::default();
             // drain from branches --> blocked
             let cd = CyclicDrainer::new(branches, &mut swap, &mut blocked);
             BranchingProtoComponent::drain_branches_to_blocked(
                 cd,
                 cu,
                 &mut solution_storage,
                 &mut payloads_to_get,
                 rctx,
                 proto_component_id,
                 ports,
             )?;
             // swap the blocked branches back
             std::mem::swap(&mut blocked, branches);
             if branches.is_empty() {
                 log!(cu.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.logger, "Already requested failure");
+                    }
                 } else {
                     log!(cu.logger, "As the leader, deciding on timeout");
                     return Ok(Decision::Failure);
+                }
+            }
+        }
         log!(cu.logger, "All proto components are blocked");
         log!(cu.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.logger, "Decision loop! have {} messages to recv", payloads_to_get.len());
             while let Some((getter, send_payload_msg)) = payloads_to_get.pop() {
             log!(cu.logger, "Decision loop! have {} messages to recv", rctx.getter_buffer.len());
             while let Some((getter, send_payload_msg)) = rctx.getter_buffer.pop() {
                 assert!(cu.port_info.polarities.get(&getter).copied() == Some(Getter));
                 let route = cu.port_info.routes.get(&getter);
                 log!(cu.logger, "Routing msg {:?} to {:?}", &send_payload_msg, &route);
                 match route {
                     None => {
                         log!(
                             cu.logger,
                             "Delivery to getter {:?} msg {:?} failed. Physical route unmapped!",
                             getter,
                             &send_payload_msg
                         );
+                    }
                     Some(Route::Endpoint { 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,
                         &mut solution_storage,
+                        &mut rctx.solution_storage,
                         getter,
                         &send_payload_msg,
                     ),
                     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,
-                                &mut solution_storage,
+                                rctx,
                                 proto_component_id,
                                 &mut payloads_to_get,
                                 getter,
                                 &send_payload_msg,
                             )?;
                             if branching_component.branches.is_empty() {
                                 log!(
                                     cu.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.logger, "Already requested failure");
+                                    }
                                 } else {
                                     log!(cu.logger, "As the leader, deciding on timeout");
                                     return Ok(Decision::Failure);
+                                }
+                            }
                         } else {
                             log!(
                                 cu.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.logger, "Check if we have any local decisions...");
             for solution in solution_storage.iter_new_local_make_old() {
+            for solution in rctx.solution_storage.iter_new_local_make_old() {
                 log!(cu.logger, "New local decision with solution {:?}...", &solution);
                 match comm.neighborhood.parent {
                     Some(parent) => {
                         log!(cu.logger, "Forwarding to my parent {:?}", parent);
                         let suggestion = Decision::Success(solution);
                         let msg = Msg::CommMsg(CommMsg {
                             round_index: comm.round_index,
                             contents: CommMsgContents::Suggest { suggestion },
                         });
                         comm.endpoint_manager.send_to_comms(parent, &msg)?;
+                    }
                     None => {
                         log!(cu.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.logger, "No decision yet. Let's recv an endpoint msg...");
+            {
                 let (endpoint_index, msg) = loop {
                     match comm.endpoint_manager.try_recv_any_comms(&mut *cu.logger, deadline)? {
                     match comm
                         .endpoint_manager
                         .try_recv_any_comms(&mut *cu.logger, rctx.deadline)?
+                    {
                         None => {
                             log!(cu.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.logger, "Already requested failure");
+                                }
                             } else {
                                 log!(cu.logger, "As the leader, deciding on timeout");
                                 return Ok(Decision::Failure);
+                            }
                             deadline = None;
+                            rctx.deadline = None;
+                        }
                         Some((endpoint_index, msg)) => break (endpoint_index, msg),
+                    }
                 };
                 log!(cu.logger, "Received from endpoint {} msg {:?}", endpoint_index, &msg);
                 let comm_msg_contents = match msg {
                     Msg::SetupMsg(..) => {
                         log!(cu.logger, "Discarding setup message; that phase is over");
                         continue 'undecided;
+                    }
                     Msg::CommMsg(comm_msg) => match comm_msg.round_index.cmp(&comm.round_index) {
                         Ordering::Equal => comm_msg.contents,
                         Ordering::Less => {
                             log!(
                                 cu.logger,
                                 "We are in round {}, but msg is for round {}. Discard",
                                 comm_msg.round_index,
                                 comm.round_index,
                             );
                             drop(comm_msg);
                             continue 'undecided;
+                        }
                         Ordering::Greater => {
                             log!(
                                 cu.logger,
                                 "We are in round {}, but msg is for round {}. Buffer",
                                 comm_msg.round_index,
                                 comm.round_index,
                             );
                             comm.endpoint_manager
                                 .delayed_messages
                                 .push((endpoint_index, Msg::CommMsg(comm_msg)));
                             continue 'undecided;
+                        }
                     },
                 };
                 match comm_msg_contents {
                     CommMsgContents::SendPayload(send_payload_msg) => {
                         let getter =
                             comm.endpoint_manager.endpoint_exts[endpoint_index].getter_for_incoming;
                         assert!(cu.port_info.polarities.get(&getter) == Some(&Getter));
                         log!(
                             cu.logger,
                             "Msg routed to getter port {:?}. Buffer for recv loop",
                             getter,
                         );
-                        payloads_to_get.push((getter, send_payload_msg));
+                        rctx.getter_buffer.getter_add(getter, send_payload_msg);
+                    }
                     CommMsgContents::Suggest { suggestion } => {
                         // only accept this control msg through a child endpoint
                         if comm.neighborhood.children.contains(&endpoint_index) {
                             match suggestion {
                                 Decision::Success(predicate) => {
                                     // child solution contributes to local solution
                                     log!(cu.logger, "Child provided solution {:?}", &predicate);
                                     let route = Route::Endpoint { index: endpoint_index };
                                     solution_storage.submit_and_digest_subtree_solution(
+                                    rctx.solution_storage.submit_and_digest_subtree_solution(
                                         &mut *cu.logger,
                                         route,
                                         predicate,
                                     );
+                                }
                                 Decision::Failure => {
                                     match comm.neighborhood.parent {
                                         None => {
                                             log!(cu.logger, "I decide on my child's failure");
                                             break 'undecided Ok(Decision::Failure);
+                                        }
                                         Some(parent) => {
                                             log!(cu.logger, "Forwarding failure through my parent endpoint {:?}", parent);
                                             if already_requested_failure.replace_with_true() {
                                                 Self::request_failure(cu, comm, parent)?
                                             } else {
                                                 log!(cu.logger, "Already requested failure");
+                                            }
+                                        }
+                                    }
+                                }
+                            }
                         } else {
                             log!(
@@ @@ -732,131 +734,139 @@ impl BranchingNative { @@
         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() && solution_predicate.consistent_with(&branch_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);
+    }
+}
 // |putter, m| {
 //     let getter = *cu.port_info.peers.get(&putter).unwrap();
 //     payloads_to_get.push((getter, m));
 // },
 impl BranchingProtoComponent {
     fn drain_branches_to_blocked(
         cd: CyclicDrainer<Predicate, ProtoComponentBranch>,
         cu: &mut ConnectorUnphased,
         solution_storage: &mut SolutionStorage,
         payload_msg_sender: &mut impl PayloadMsgSender,
         rctx: &mut RoundCtx,
         proto_component_id: ProtoComponentId,
         ports: &HashSet<PortId>,
     ) -> Result<(), SyncError> {
         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 => {
                     // make concrete all variables
                     for &port in ports.iter() {
                         let var = cu.port_info.spec_var_for(port);
                         predicate.assigned.entry(var).or_insert(SpecVal::SILENT);
+                    }
                     // submit solution for this component
                     solution_storage.submit_and_digest_subtree_solution(
+                    rctx.solution_storage.submit_and_digest_subtree_solution(
                         &mut *cu.logger,
                         Route::LocalComponent(ComponentId::Proto(proto_component_id)),
                         predicate.clone(),
                     );
                     branch.ended = true;
                     // move to "blocked"
                     drainer.add_output(predicate, branch);
+                }
                 B::CouldntReadMsg(port) => {
                     // move to "blocked"
                     assert!(!branch.inbox.contains_key(&port));
                     drainer.add_output(predicate, branch);
+                }
                 B::CouldntCheckFiring(port) => {
                     // sanity check
                     let var = cu.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.port_info.polarities.get(&putter));
                     // overwrite assignment
                     let var = cu.port_info.spec_var_for(putter);
                     let was = predicate.assigned.insert(var, SpecVal::FIRING);
                     if was == Some(SpecVal::SILENT) {
                         log!(cu.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"
                         log!(cu.logger, "Proto component {:?} putting payload {:?} on port {:?} (using var {:?})", proto_component_id, &payload, putter, var);
                         let msg = SendPayloadMsg { predicate: predicate.clone(), payload };
-                        payload_msg_sender.putter_send(cu, putter, msg)?;
+                        rctx.getter_buffer.putter_add(cu, putter, msg)?;
                         drainer.add_input(predicate, branch);
+                    }
+                }
+            }
             Ok(())
         })
+    }
     fn feed_msg(
         &mut self,
         cu: &mut ConnectorUnphased,
-        solution_storage: &mut SolutionStorage,
+        rctx: &mut RoundCtx,
         proto_component_id: ProtoComponentId,
         payload_msg_sender: &mut impl PayloadMsgSender,
         getter: PortId,
         send_payload_msg: &SendPayloadMsg,
     ) -> Result<(), SyncError> {
         let logger = &mut *cu.logger;
         log!(
             logger,
             "feeding proto component {:?} getter {:?} {:?}",
             proto_component_id,
             getter,
             &send_payload_msg
         );
         let BranchingProtoComponent { branches, ports } = self;
         let mut unblocked = HashMap::default();
         let mut blocked = HashMap::default();
         // 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 AllMapperResult as Amr;
             log!(logger, "visiting branch with pred {:?}", &predicate);
@@ @@ -877,113 +887,114 @@ impl BranchingProtoComponent { @@
                     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);
+                }
                 Amr::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 mut swap = HashMap::default();
         let cd = CyclicDrainer::new(&mut unblocked, &mut swap, &mut blocked);
         BranchingProtoComponent::drain_branches_to_blocked(
             cd,
             cu,
             solution_storage,
             payload_msg_sender,
             rctx,
             proto_component_id,
             ports,
         )?;
         // swap the blocked branches back
         std::mem::swap(&mut blocked, branches);
         log!(cu.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 && solution_predicate.consistent_with(&branch_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 { inbox: Default::default(), state, ended: false };
         let branch = ProtoComponentBranch {
             inbox: Default::default(),
             state,
             ended: false,
             untaken_choice: None,
         };
         Self { ports, branches: hashmap! { Predicate::default() => branch  } }
+    }
+}
 impl SolutionStorage {
     fn new(routes: impl Iterator<Item = Route>) -> Self {
         let mut subtree_id_to_index: HashMap<Route, usize> = Default::default();
         let mut subtree_solutions = vec![];
         for key in routes {
             subtree_id_to_index.insert(key, 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 = Route>) {
         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 peek_new_locals(&self) -> impl Iterator<Item = &Predicate> + '_ {
     //     self.new_local.iter()
     // }
     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: Route,
         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,
@@ @@ -1003,71 +1014,83 @@ impl SolutionStorage { @@
     ) {
         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 PayloadMsgSender for Vec<(PortId, SendPayloadMsg)> {
     fn putter_send(
 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,
     ) -> Result<(), SyncError> {
         if let Some(&getter) = cu.port_info.peers.get(&putter) {
-            self.push((getter, msg));
+            self.getter_add(getter, msg);
             Ok(())
         } else {
             Err(SyncError::MalformedStateError(MalformedStateError::GetterUnknownFor { putter }))
+        }
+    }
+}
 impl SyncProtoContext<'_> {
     pub(crate) fn is_firing(&mut self, port: PortId) -> Option<bool> {
         let var = self.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.inbox.get(&port)
+    }
     pub(crate) fn take_choice(&mut self) -> Option<u16> {
         self.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 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.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.id_manager.new_proto_component_id();
         for port in moved_ports.iter() {

src/runtime/error.rs

➞

Show inline comments

@@ @@ -6,49 +6,48 @@ pub enum ConnectError { @@
     PollInitFailed,
     Timeout,
     PollFailed,
     AcceptFailed(SocketAddr),
     AlreadyConnected,
     PortPeerPolarityMismatch(PortId),
     EndpointSetupError(SocketAddr, EndpointError),
     SetupAlgMisbehavior,
+}
 #[derive(Eq, PartialEq, Copy, Clone, Debug)]
 pub enum AddComponentError {
     NoSuchComponent,
     NonPortTypeParameters,
     CannotMovePort(PortId),
     WrongNumberOfParamaters { expected: usize },
     UnknownPort(PortId),
     WrongPortPolarity { port: PortId, expected_polarity: Polarity },
     DuplicateMovedPort(PortId),
+}
 ////////////////////////
 #[derive(Debug, Clone)]
 pub enum SyncError {
     NotConnected,
     InconsistentProtoComponent(ProtoComponentId),
     IndistinguishableBatches([usize; 2]),
     RoundFailure,
     PollFailed,
     BrokenEndpoint(usize),
     MalformedStateError(MalformedStateError),
+}
 #[derive(Debug, Clone)]
 pub enum MalformedStateError {
     PortCannotPut(PortId),
     GetterUnknownFor { putter: PortId },
+}
 #[derive(Debug, Clone)]
 pub enum EndpointError {
     MalformedMessage,
     BrokenEndpoint,
+}
 #[derive(Debug)]
 pub enum PortOpError {
     WrongPolarity,
     UnknownPolarity,
     NotConnected,
     MultipleOpsOnPort,
     PortUnavailable,
+}
 #[derive(Debug, Eq, PartialEq)]

src/runtime/mod.rs

➞

Show inline comments

@@ @@ -20,48 +20,49 @@ use error::*; @@
 #[derive(Debug)]
 pub struct Connector {
     unphased: ConnectorUnphased,
     phased: ConnectorPhased,
+}
 pub trait Logger: Debug {
     fn line_writer(&mut self) -> &mut dyn std::io::Write;
+}
 #[derive(Debug)]
 pub struct VecLogger(ConnectorId, Vec<u8>);
 #[derive(Debug)]
 pub struct DummyLogger;
 #[derive(Debug)]
 pub struct FileLogger(ConnectorId, std::fs::File);
 pub(crate) struct NonsyncProtoContext<'a> {
     logger: &'a mut dyn Logger,
     proto_component_id: ProtoComponentId,
     port_info: &'a mut PortInfo,
     id_manager: &'a mut IdManager,
     proto_component_ports: &'a mut HashSet<PortId>,
     unrun_components: &'a mut Vec<(ProtoComponentId, ProtoComponent)>,
+}
 pub(crate) struct SyncProtoContext<'a> {
     logger: &'a mut dyn Logger,
     untaken_choice: &'a mut Option<u16>,
     predicate: &'a Predicate,
     port_info: &'a PortInfo,
     inbox: &'a HashMap<PortId, Payload>,
+}
 #[derive(
     Copy, Clone, Eq, PartialEq, Ord, Hash, PartialOrd, serde::Serialize, serde::Deserialize,
 )]
 struct SpecVar(PortId);
 #[derive(
     Copy, Clone, Eq, PartialEq, Ord, Hash, PartialOrd, serde::Serialize, serde::Deserialize,
 )]
 struct SpecVal(u16);
 #[derive(Debug)]
 struct RoundOk {
     batch_index: usize,
     gotten: HashMap<PortId, Payload>,
+}
 #[derive(Default)]
 struct VecSet<T: std::cmp::Ord> {
     // invariant: ordered, deduplicated
     vec: Vec<T>,
+}
 #[derive(Debug, Clone, Copy, Eq, PartialEq, Hash, serde::Serialize, serde::Deserialize)]

src/runtime/tests.rs

➞

Show inline comments

 use crate as reowolf;
 use crossbeam_utils::thread::scope;
 use reowolf::{
     error::*,
     EndpointPolarity::{Active, Passive},
     Polarity::{Getter, Putter},
     *,
 };
 use std::{fs::File, net::SocketAddr, path::Path, sync::Arc, time::Duration};
 //////////////////////////////////////////
 const SEC1: Option<Duration> = Some(Duration::from_secs(1));
 const SEC5: Option<Duration> = Some(Duration::from_secs(5));
 const SEC15: Option<Duration> = Some(Duration::from_secs(15));
 const MS300: Option<Duration> = Some(Duration::from_millis(300));
 fn next_test_addr() -> SocketAddr {
     use std::{
         net::{Ipv4Addr, SocketAddrV4},
         sync::atomic::{AtomicU16, Ordering::SeqCst},
     };
     static TEST_PORT: AtomicU16 = AtomicU16::new(5_000);
     let port = TEST_PORT.fetch_add(1, SeqCst);
     SocketAddrV4::new(Ipv4Addr::LOCALHOST, port).into()
+}
 fn file_logged_connector(connector_id: ConnectorId, dir_path: &Path) -> Connector {
     let _ = std::fs::create_dir(dir_path); // we will check failure soon
     let path = dir_path.join(format!("cid_{:?}.txt", connector_id));
     let file = File::create(path).unwrap();
     let file_logger = Box::new(FileLogger::new(connector_id, file));
     Connector::new(file_logger, MINIMAL_PROTO.clone(), connector_id, 8)
+}
 static MINIMAL_PDL: &'static [u8] = b"
 primitive together(in ia, in ib, out oa, out ob){
   while(true) synchronous() {
     if(fires(ia)) {
       put(oa, get(ia));
       put(ob, get(ib));
+    }
+  }
@@ @@ -66,523 +70,590 @@ fn new_port_pair() { @@
     let [_, _] = c.new_port_pair();
+}
 #[test]
 fn new_sync() {
     let test_log_path = Path::new("./logs/new_sync");
     let mut c = file_logged_connector(0, test_log_path);
     let [o, i] = c.new_port_pair();
     c.add_component(b"sync", &[i, o]).unwrap();
+}
 #[test]
 fn new_net_port() {
     let test_log_path = Path::new("./logs/new_net_port");
     let mut c = file_logged_connector(0, test_log_path);
     let sock_addr = next_test_addr();
     let _ = c.new_net_port(Getter, sock_addr, Passive).unwrap();
     let _ = c.new_net_port(Putter, sock_addr, Active).unwrap();
+}
 #[test]
 fn trivial_connect() {
     let test_log_path = Path::new("./logs/trivial_connect");
     let mut c = file_logged_connector(0, test_log_path);
-    c.connect(Some(Duration::from_secs(1))).unwrap();
+    c.connect(SEC1).unwrap();
+}
 #[test]
 fn single_node_connect() {
     let sock_addr = next_test_addr();
     let test_log_path = Path::new("./logs/single_node_connect");
     let mut c = file_logged_connector(0, test_log_path);
     let _ = c.new_net_port(Getter, sock_addr, Passive).unwrap();
     let _ = c.new_net_port(Putter, sock_addr, Active).unwrap();
-    c.connect(Some(Duration::from_secs(1))).unwrap();
+    c.connect(SEC1).unwrap();
+}
 #[test]
 fn minimal_net_connect() {
     let sock_addr = next_test_addr();
     let test_log_path = Path::new("./logs/minimal_net_connect");
     scope(|s| {
         s.spawn(|_| {
             let mut c = file_logged_connector(0, test_log_path);
             let _ = c.new_net_port(Getter, sock_addr, Active).unwrap();
-            c.connect(Some(Duration::from_secs(1))).unwrap();
+            c.connect(SEC1).unwrap();
         });
         s.spawn(|_| {
             let mut c = file_logged_connector(1, test_log_path);
             let _ = c.new_net_port(Putter, sock_addr, Passive).unwrap();
-            c.connect(Some(Duration::from_secs(1))).unwrap();
+            c.connect(SEC1).unwrap();
         });
     })
     .unwrap();
+}
 #[test]
 fn put_no_sync() {
     let test_log_path = Path::new("./logs/put_no_sync");
     let mut c = file_logged_connector(0, test_log_path);
     let [o, _] = c.new_port_pair();
-    c.connect(Some(Duration::from_secs(1))).unwrap();
+    c.connect(SEC1).unwrap();
     c.put(o, TEST_MSG.clone()).unwrap();
+}
 #[test]
 fn wrong_polarity_bad() {
     let test_log_path = Path::new("./logs/wrong_polarity_bad");
     let mut c = file_logged_connector(0, test_log_path);
     let [_, i] = c.new_port_pair();
-    c.connect(Some(Duration::from_secs(1))).unwrap();
+    c.connect(SEC1).unwrap();
     c.put(i, TEST_MSG.clone()).unwrap_err();
+}
 #[test]
 fn dup_put_bad() {
     let test_log_path = Path::new("./logs/dup_put_bad");
     let mut c = file_logged_connector(0, test_log_path);
     let [o, _] = c.new_port_pair();
-    c.connect(Some(Duration::from_secs(1))).unwrap();
+    c.connect(SEC1).unwrap();
     c.put(o, TEST_MSG.clone()).unwrap();
     c.put(o, TEST_MSG.clone()).unwrap_err();
+}
 #[test]
 fn trivial_sync() {
     let test_log_path = Path::new("./logs/trivial_sync");
     let mut c = file_logged_connector(0, test_log_path);
     c.connect(Some(Duration::from_secs(1))).unwrap();
     c.sync(Some(Duration::from_secs(1))).unwrap();
     c.connect(SEC1).unwrap();
     c.sync(SEC1).unwrap();
+}
 #[test]
 fn unconnected_gotten_err() {
     let test_log_path = Path::new("./logs/unconnected_gotten_err");
     let mut c = file_logged_connector(0, test_log_path);
     let [_, i] = c.new_port_pair();
     assert_eq!(reowolf::error::GottenError::NoPreviousRound, c.gotten(i).unwrap_err());
+}
 #[test]
 fn connected_gotten_err_no_round() {
     let test_log_path = Path::new("./logs/connected_gotten_err_no_round");
     let mut c = file_logged_connector(0, test_log_path);
     let [_, i] = c.new_port_pair();
-    c.connect(Some(Duration::from_secs(1))).unwrap();
+    c.connect(SEC1).unwrap();
     assert_eq!(reowolf::error::GottenError::NoPreviousRound, c.gotten(i).unwrap_err());
+}
 #[test]
 fn connected_gotten_err_ungotten() {
     let test_log_path = Path::new("./logs/connected_gotten_err_ungotten");
     let mut c = file_logged_connector(0, test_log_path);
     let [_, i] = c.new_port_pair();
     c.connect(Some(Duration::from_secs(1))).unwrap();
     c.sync(Some(Duration::from_secs(1))).unwrap();
     c.connect(SEC1).unwrap();
     c.sync(SEC1).unwrap();
     assert_eq!(reowolf::error::GottenError::PortDidntGet, c.gotten(i).unwrap_err());
+}
 #[test]
 fn native_polarity_checks() {
     let test_log_path = Path::new("./logs/native_polarity_checks");
     let mut c = file_logged_connector(0, test_log_path);
     let [o, i] = c.new_port_pair();
-    c.connect(Some(Duration::from_secs(1))).unwrap();
+    c.connect(SEC1).unwrap();
     // fail...
     c.get(o).unwrap_err();
     c.put(i, TEST_MSG.clone()).unwrap_err();
     // succeed..
     c.get(i).unwrap();
     c.put(o, TEST_MSG.clone()).unwrap();
+}
 #[test]
 fn native_multiple_gets() {
     let test_log_path = Path::new("./logs/native_multiple_gets");
     let mut c = file_logged_connector(0, test_log_path);
     let [_, i] = c.new_port_pair();
-    c.connect(Some(Duration::from_secs(1))).unwrap();
+    c.connect(SEC1).unwrap();
     c.get(i).unwrap();
     c.get(i).unwrap_err();
+}
 #[test]
 fn next_batch() {
     let test_log_path = Path::new("./logs/next_batch");
     let mut c = file_logged_connector(0, test_log_path);
     c.next_batch().unwrap_err();
-    c.connect(Some(Duration::from_secs(1))).unwrap();
+    c.connect(SEC1).unwrap();
     c.next_batch().unwrap();
     c.next_batch().unwrap();
     c.next_batch().unwrap();
+}
 #[test]
 fn native_self_msg() {
     let test_log_path = Path::new("./logs/native_self_msg");
     let mut c = file_logged_connector(0, test_log_path);
     let [o, i] = c.new_port_pair();
-    c.connect(Some(Duration::from_secs(1))).unwrap();
+    c.connect(SEC1).unwrap();
     c.get(i).unwrap();
     c.put(o, TEST_MSG.clone()).unwrap();
-    c.sync(Some(Duration::from_secs(1))).unwrap();
+    c.sync(SEC1).unwrap();
+}
 #[test]
 fn two_natives_msg() {
     let test_log_path = Path::new("./logs/two_natives_msg");
     let sock_addr = next_test_addr();
     scope(|s| {
         s.spawn(|_| {
             let mut c = file_logged_connector(0, test_log_path);
             let g = c.new_net_port(Getter, sock_addr, Active).unwrap();
-            c.connect(Some(Duration::from_secs(1))).unwrap();
+            c.connect(SEC1).unwrap();
             c.get(g).unwrap();
-            c.sync(Some(Duration::from_secs(1))).unwrap();
+            c.sync(SEC1).unwrap();
             c.gotten(g).unwrap();
         });
         s.spawn(|_| {
             let mut c = file_logged_connector(1, test_log_path);
             let p = c.new_net_port(Putter, sock_addr, Passive).unwrap();
-            c.connect(Some(Duration::from_secs(1))).unwrap();
+            c.connect(SEC1).unwrap();
             c.put(p, TEST_MSG.clone()).unwrap();
-            c.sync(Some(Duration::from_secs(1))).unwrap();
+            c.sync(SEC1).unwrap();
         });
     })
     .unwrap();
+}
 #[test]
 fn trivial_nondet() {
     let test_log_path = Path::new("./logs/trivial_nondet");
     let mut c = file_logged_connector(0, test_log_path);
     let [_, i] = c.new_port_pair();
-    c.connect(Some(Duration::from_secs(1))).unwrap();
+    c.connect(SEC1).unwrap();
     c.get(i).unwrap();
     // getting 0 batch
     c.next_batch().unwrap();
     // silent 1 batch
-    assert_eq!(1, c.sync(Some(Duration::from_secs(1))).unwrap());
+    assert_eq!(1, c.sync(SEC1).unwrap());
     c.gotten(i).unwrap_err();
+}
 #[test]
 fn connector_pair_nondet() {
     let test_log_path = Path::new("./logs/connector_pair_nondet");
     let sock_addr = next_test_addr();
     scope(|s| {
         s.spawn(|_| {
             let mut c = file_logged_connector(0, test_log_path);
             let g = c.new_net_port(Getter, sock_addr, Active).unwrap();
-            c.connect(Some(Duration::from_secs(1))).unwrap();
+            c.connect(SEC1).unwrap();
             c.next_batch().unwrap();
             c.get(g).unwrap();
-            assert_eq!(1, c.sync(Some(Duration::from_secs(1))).unwrap());
+            assert_eq!(1, c.sync(SEC1).unwrap());
             c.gotten(g).unwrap();
         });
         s.spawn(|_| {
             let mut c = file_logged_connector(1, test_log_path);
             let p = c.new_net_port(Putter, sock_addr, Passive).unwrap();
-            c.connect(Some(Duration::from_secs(1))).unwrap();
+            c.connect(SEC1).unwrap();
             c.put(p, TEST_MSG.clone()).unwrap();
-            c.sync(Some(Duration::from_secs(1))).unwrap();
+            c.sync(SEC1).unwrap();
         });
     })
     .unwrap();
+}
 #[test]
 fn native_immediately_inconsistent() {
     let test_log_path = Path::new("./logs/native_immediately_inconsistent");
     let mut c = file_logged_connector(0, test_log_path);
     let [_, g] = c.new_port_pair();
-    c.connect(Some(Duration::from_secs(1))).unwrap();
+    c.connect(SEC1).unwrap();
     c.get(g).unwrap();
-    c.sync(Some(Duration::from_secs(30))).unwrap_err();
+    c.sync(SEC15).unwrap_err();
+}
 #[test]
 fn native_recovers() {
     let test_log_path = Path::new("./logs/native_recovers");
     let mut c = file_logged_connector(0, test_log_path);
     let [p, g] = c.new_port_pair();
-    c.connect(Some(Duration::from_secs(1))).unwrap();
+    c.connect(SEC1).unwrap();
     c.get(g).unwrap();
-    c.sync(Some(Duration::from_secs(30))).unwrap_err();
+    c.sync(SEC15).unwrap_err();
     c.put(p, TEST_MSG.clone()).unwrap();
     c.get(g).unwrap();
-    c.sync(Some(Duration::from_secs(30))).unwrap();
+    c.sync(SEC15).unwrap();
+}
 #[test]
 fn cannot_use_moved_ports() {
     /*
     native p|-->|g sync
     */
     let test_log_path = Path::new("./logs/cannot_use_moved_ports");
     let mut c = file_logged_connector(0, test_log_path);
     let [p, g] = c.new_port_pair();
     c.add_component(b"sync", &[g, p]).unwrap();
-    c.connect(Some(Duration::from_secs(1))).unwrap();
+    c.connect(SEC1).unwrap();
     c.put(p, TEST_MSG.clone()).unwrap_err();
     c.get(g).unwrap_err();
+}
 #[test]
 fn sync_sync() {
     /*
     native p0|-->|g0 sync
            g1|<--|p1
     */
     let test_log_path = Path::new("./logs/sync_sync");
     let mut c = file_logged_connector(0, test_log_path);
     let [p0, g0] = c.new_port_pair();
     let [p1, g1] = c.new_port_pair();
     c.add_component(b"sync", &[g0, p1]).unwrap();
-    c.connect(Some(Duration::from_secs(1))).unwrap();
+    c.connect(SEC1).unwrap();
     c.put(p0, TEST_MSG.clone()).unwrap();
     c.get(g1).unwrap();
-    c.sync(Some(Duration::from_secs(1))).unwrap();
+    c.sync(SEC1).unwrap();
     c.gotten(g1).unwrap();
+}
 #[test]
 fn double_net_connect() {
     let test_log_path = Path::new("./logs/double_net_connect");
     let sock_addrs = [next_test_addr(), next_test_addr()];
     scope(|s| {
         s.spawn(|_| {
             let mut c = file_logged_connector(0, test_log_path);
             let [_p, _g] = [
                 c.new_net_port(Putter, sock_addrs[0], Active).unwrap(),
                 c.new_net_port(Getter, sock_addrs[1], Active).unwrap(),
             ];
-            c.connect(Some(Duration::from_secs(1))).unwrap();
+            c.connect(SEC1).unwrap();
         });
         s.spawn(|_| {
             let mut c = file_logged_connector(1, test_log_path);
             let [_g, _p] = [
                 c.new_net_port(Getter, sock_addrs[0], Passive).unwrap(),
                 c.new_net_port(Putter, sock_addrs[1], Passive).unwrap(),
             ];
-            c.connect(Some(Duration::from_secs(1))).unwrap();
+            c.connect(SEC1).unwrap();
         });
     })
     .unwrap();
+}
 #[test]
 fn distributed_msg_bounce() {
     /*
     native[0] | sync 0.p|-->|1.p native[1]
 .g|<--|1.g
     */
     let test_log_path = Path::new("./logs/distributed_msg_bounce");
     let sock_addrs = [next_test_addr(), next_test_addr()];
     scope(|s| {
         s.spawn(|_| {
             /*
             native | sync p|-->
                    |      g|<--
             */
             let mut c = file_logged_connector(0, test_log_path);
             let [p, g] = [
                 c.new_net_port(Putter, sock_addrs[0], Active).unwrap(),
                 c.new_net_port(Getter, sock_addrs[1], Active).unwrap(),
             ];
             c.add_component(b"sync", &[g, p]).unwrap();
             c.connect(Some(Duration::from_secs(1))).unwrap();
             c.sync(Some(Duration::from_secs(1))).unwrap();
             c.connect(SEC1).unwrap();
             c.sync(SEC1).unwrap();
         });
         s.spawn(|_| {
             /*
             native p|-->
                    g|<--
             */
             let mut c = file_logged_connector(1, test_log_path);
             let [g, p] = [
                 c.new_net_port(Getter, sock_addrs[0], Passive).unwrap(),
                 c.new_net_port(Putter, sock_addrs[1], Passive).unwrap(),
             ];
-            c.connect(Some(Duration::from_secs(1))).unwrap();
+            c.connect(SEC1).unwrap();
             c.put(p, TEST_MSG.clone()).unwrap();
             c.get(g).unwrap();
-            c.sync(Some(Duration::from_secs(1))).unwrap();
+            c.sync(SEC1).unwrap();
             c.gotten(g).unwrap();
         });
     })
     .unwrap();
+}
 #[test]
 fn local_timeout() {
     let test_log_path = Path::new("./logs/local_timeout");
     let mut c = file_logged_connector(0, test_log_path);
     let [_, g] = c.new_port_pair();
-    c.connect(Some(Duration::from_secs(1))).unwrap();
+    c.connect(SEC1).unwrap();
     c.get(g).unwrap();
-    match c.sync(Some(Duration::from_millis(200))) {
+    match c.sync(MS300) {
         Err(SyncError::RoundFailure) => {}
         res => panic!("expeted timeout. but got {:?}", res),
+    }
+}
 #[test]
 fn parent_timeout() {
     let test_log_path = Path::new("./logs/parent_timeout");
     let sock_addr = next_test_addr();
     scope(|s| {
         s.spawn(|_| {
             // parent; times out
             let mut c = file_logged_connector(999, test_log_path);
             let _ = c.new_net_port(Putter, sock_addr, Active).unwrap();
             c.connect(Some(Duration::from_secs(1))).unwrap();
             c.sync(Some(Duration::from_millis(300))).unwrap_err(); // timeout
             c.connect(SEC1).unwrap();
             c.sync(MS300).unwrap_err(); // timeout
         });
         s.spawn(|_| {
             // child
             let mut c = file_logged_connector(000, test_log_path);
             let g = c.new_net_port(Getter, sock_addr, Passive).unwrap();
-            c.connect(Some(Duration::from_secs(1))).unwrap();
+            c.connect(SEC1).unwrap();
             c.get(g).unwrap(); // not matched by put
             c.sync(None).unwrap_err(); // no timeout
         });
     })
     .unwrap();
+}
 #[test]
 fn child_timeout() {
     let test_log_path = Path::new("./logs/child_timeout");
     let sock_addr = next_test_addr();
     scope(|s| {
         s.spawn(|_| {
             // child; times out
             let mut c = file_logged_connector(000, test_log_path);
             let _ = c.new_net_port(Putter, sock_addr, Active).unwrap();
             c.connect(Some(Duration::from_secs(1))).unwrap();
             c.sync(Some(Duration::from_millis(300))).unwrap_err(); // timeout
             c.connect(SEC1).unwrap();
             c.sync(MS300).unwrap_err(); // timeout
         });
         s.spawn(|_| {
             // parent
             let mut c = file_logged_connector(999, test_log_path);
             let g = c.new_net_port(Getter, sock_addr, Passive).unwrap();
-            c.connect(Some(Duration::from_secs(1))).unwrap();
+            c.connect(SEC1).unwrap();
             c.get(g).unwrap(); // not matched by put
             c.sync(None).unwrap_err(); // no timeout
         });
     })
     .unwrap();
+}
 #[test]
 fn chain_connect() {
     let test_log_path = Path::new("./logs/chain_connect");
     let sock_addrs = [next_test_addr(), next_test_addr(), next_test_addr(), next_test_addr()];
     scope(|s| {
         s.spawn(|_| {
             let mut c = file_logged_connector(0, test_log_path);
             c.new_net_port(Putter, sock_addrs[0], Passive).unwrap();
-            c.connect(Some(Duration::from_secs(2))).unwrap();
+            c.connect(SEC5).unwrap();
         });
         s.spawn(|_| {
             let mut c = file_logged_connector(10, test_log_path);
             c.new_net_port(Getter, sock_addrs[0], Active).unwrap();
             c.new_net_port(Putter, sock_addrs[1], Passive).unwrap();
-            c.connect(Some(Duration::from_secs(2))).unwrap();
+            c.connect(SEC5).unwrap();
         });
         s.spawn(|_| {
             // LEADER
             let mut c = file_logged_connector(7, test_log_path);
             c.new_net_port(Getter, sock_addrs[1], Active).unwrap();
             c.new_net_port(Putter, sock_addrs[2], Passive).unwrap();
-            c.connect(Some(Duration::from_secs(2))).unwrap();
+            c.connect(SEC5).unwrap();
         });
         s.spawn(|_| {
             let mut c = file_logged_connector(4, test_log_path);
             c.new_net_port(Getter, sock_addrs[2], Active).unwrap();
             c.new_net_port(Putter, sock_addrs[3], Passive).unwrap();
-            c.connect(Some(Duration::from_secs(2))).unwrap();
+            c.connect(SEC5).unwrap();
         });
         s.spawn(|_| {
             let mut c = file_logged_connector(1, test_log_path);
             c.new_net_port(Getter, sock_addrs[3], Active).unwrap();
-            c.connect(Some(Duration::from_secs(2))).unwrap();
+            c.connect(SEC5).unwrap();
         });
     })
     .unwrap();
+}
 #[test]
 fn net_self_loop() {
     let test_log_path = Path::new("./logs/net_self_loop");
     let sock_addr = next_test_addr();
     let mut c = file_logged_connector(0, test_log_path);
     let p = c.new_net_port(Putter, sock_addr, Active).unwrap();
     let g = c.new_net_port(Getter, sock_addr, Passive).unwrap();
-    c.connect(Some(Duration::from_secs(1))).unwrap();
+    c.connect(SEC1).unwrap();
     c.put(p, TEST_MSG.clone()).unwrap();
     c.get(g).unwrap();
-    c.sync(Some(Duration::from_millis(500))).unwrap();
+    c.sync(MS300).unwrap();
+}
 #[test]
 fn nobody_connects_active() {
     let test_log_path = Path::new("./logs/nobody_connects_active");
     let sock_addr = next_test_addr();
     let mut c = file_logged_connector(0, test_log_path);
     let _g = c.new_net_port(Getter, sock_addr, Active).unwrap();
     c.connect(Some(Duration::from_secs(5))).unwrap_err();
+}
 #[test]
 fn nobody_connects_passive() {
     let test_log_path = Path::new("./logs/nobody_connects_passive");
     let sock_addr = next_test_addr();
     let mut c = file_logged_connector(0, test_log_path);
     let _g = c.new_net_port(Getter, sock_addr, Passive).unwrap();
     c.connect(Some(Duration::from_secs(5))).unwrap_err();
+}
 #[test]
 fn together() {
     let test_log_path = Path::new("./logs/together");
     let sock_addrs = [next_test_addr(), next_test_addr()];
     scope(|s| {
         s.spawn(|_| {
             let mut c = file_logged_connector(0, test_log_path);
             let [p0, p1] = c.new_port_pair();
             let p2 = c.new_net_port(Getter, sock_addrs[0], Passive).unwrap();
             let p3 = c.new_net_port(Putter, sock_addrs[1], Active).unwrap();
             let [p4, p5] = c.new_port_pair();
             c.add_component(b"together", &[p1, p2, p3, p4]).unwrap();
-            c.connect(Some(Duration::from_secs(1))).unwrap();
+            c.connect(SEC1).unwrap();
             c.put(p0, TEST_MSG.clone()).unwrap();
             c.get(p5).unwrap();
-            c.sync(Some(Duration::from_millis(500))).unwrap();
+            c.sync(MS300).unwrap();
             c.gotten(p5).unwrap();
         });
         s.spawn(|_| {
             let mut c = file_logged_connector(1, test_log_path);
             let [p0, p1] = c.new_port_pair();
             let p2 = c.new_net_port(Getter, sock_addrs[1], Passive).unwrap();
             let p3 = c.new_net_port(Putter, sock_addrs[0], Active).unwrap();
             let [p4, p5] = c.new_port_pair();
             c.add_component(b"together", &[p1, p2, p3, p4]).unwrap();
-            c.connect(Some(Duration::from_secs(1))).unwrap();
+            c.connect(SEC1).unwrap();
             c.put(p0, TEST_MSG.clone()).unwrap();
             c.get(p5).unwrap();
-            c.sync(Some(Duration::from_millis(500))).unwrap();
+            c.sync(MS300).unwrap();
             c.gotten(p5).unwrap();
         });
     })
     .unwrap();
+}
 #[test]
 fn native_batch_distinguish() {
     let test_log_path = Path::new("./logs/native_batch_distinguish");
     let mut c = file_logged_connector(0, test_log_path);
-    c.connect(Some(Duration::from_secs(1))).unwrap();
+    c.connect(SEC1).unwrap();
     c.next_batch().unwrap();
     c.sync(Some(Duration::from_secs(3))).unwrap();
     c.sync(SEC1).unwrap();
+}
 #[test]
 fn multirounds() {
     let test_log_path = Path::new("./logs/multirounds");
     let sock_addrs = [next_test_addr(), next_test_addr()];
     scope(|s| {
         s.spawn(|_| {
             let mut c = file_logged_connector(0, test_log_path);
             let p0 = c.new_net_port(Putter, sock_addrs[0], Active).unwrap();
             let p1 = c.new_net_port(Getter, sock_addrs[1], Passive).unwrap();
             c.connect(SEC1).unwrap();
             for _ in 0..10 {
                 c.put(p0, TEST_MSG.clone()).unwrap();
                 c.get(p1).unwrap();
                 c.sync(SEC1).unwrap();
+            }
         });
         s.spawn(|_| {
             let mut c = file_logged_connector(1, test_log_path);
             let p0 = c.new_net_port(Getter, sock_addrs[0], Passive).unwrap();
             let p1 = c.new_net_port(Putter, sock_addrs[1], Active).unwrap();
             c.connect(SEC1).unwrap();
             for _ in 0..10 {
                 c.get(p0).unwrap();
                 c.put(p1, TEST_MSG.clone()).unwrap();
                 c.sync(SEC1).unwrap();
+            }
         });
     })
     .unwrap();
+}
 #[test]
 fn multi_recover() {
     let test_log_path = Path::new("./logs/multi_recover");
     let sock_addrs = [next_test_addr(), next_test_addr()];
     let success_iter = [true, false].iter().copied().cycle().take(10);
     scope(|s| {
         s.spawn(|_| {
             let mut c = file_logged_connector(0, test_log_path);
             let p0 = c.new_net_port(Putter, sock_addrs[0], Active).unwrap();
             let p1 = c.new_net_port(Getter, sock_addrs[1], Passive).unwrap();
             c.connect(SEC1).unwrap();
             for succeeds in success_iter.clone() {
                 c.put(p0, TEST_MSG.clone()).unwrap();
                 if succeeds {
                     c.get(p1).unwrap();
+                }
                 let res = c.sync(MS300);
                 assert_eq!(res.is_ok(), succeeds);
+            }
         });
         s.spawn(|_| {
             let mut c = file_logged_connector(1, test_log_path);
             let p0 = c.new_net_port(Getter, sock_addrs[0], Passive).unwrap();
             let p1 = c.new_net_port(Putter, sock_addrs[1], Active).unwrap();
             c.connect(SEC1).unwrap();
             for succeeds in success_iter.clone() {
                 c.get(p0).unwrap();
                 c.put(p1, TEST_MSG.clone()).unwrap();
                 let res = c.sync(MS300);
                 assert_eq!(res.is_ok(), succeeds);
+            }
         });
     })
     .unwrap();
+}

0 comments (0 inline, 0 general)