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

Changeset - aa88e65c3d92

Parent rev.

Child rev.

[Not reviewed]

0 4 0

Christopher Esterhuyse - 5 years ago 2020-02-10 15:15:28
christopheresterhuyse@gmail.com

exchange

4 files changed with 44 insertions and 35 deletions:

src/runtime/actors.rs

src/runtime/communication.rs

src/runtime/mod.rs

src/test/connector.rs

0 comments (0 inline, 0 general)

src/runtime/actors.rs

➞

Show inline comments

@@ @@ -192,210 +192,219 @@ impl PolyP { @@
         let to_run = if self.complete.contains_key(&payload_predicate) {
             // exact match with stopped machine
             log!(
                 &mut m_ctx.inner.logger,
                 "... poly_recv_run matched stopped machine exactly! nothing to do here",
             );
             vec![]
         } else if let Some(mut branch) = self.incomplete.remove(&payload_predicate) {
             // exact match with running machine
             log!(
                 &mut m_ctx.inner.logger,
                 "... poly_recv_run matched running machine exactly! pred is {:?}",
                 &payload_predicate
             );
             branch.inbox.insert(ekey, payload);
             vec![(payload_predicate, branch)]
         } else {
             log!(
                 &mut m_ctx.inner.logger,
                 "... poly_recv_run didn't have any exact matches... Let's try feed it to all branches",
             );
             let mut incomplete2 = HashMap::<_, _>::default();
             let to_run = self
                 .incomplete
                 .drain()
                 .filter_map(|(old_predicate, mut branch)| {
                     use CommonSatResult as Csr;
                     match old_predicate.common_satisfier(&payload_predicate) {
                         Csr::FormerNotLatter | Csr::Equivalent => {
                             log!(
                                 &mut m_ctx.inner.logger,
                                 "... poly_recv_run This branch is compatible unaltered! branch pred: {:?}",
                                 &old_predicate
                             );
                             // old_predicate COVERS the assumptions of payload_predicate
                             let was = branch.inbox.insert(ekey, payload.clone());
                             assert!(was.is_none()); // INBOX MUST BE EMPTY!
                             Some((old_predicate, branch))
+                        }
                         Csr::New(new) => {
                             log!(
                                 &mut m_ctx.inner.logger,
                                 "... poly_recv_run payloadpred {:?} and branchpred {:?} satisfied by new pred {:?}. FORKING",
                                 &payload_predicate,
                                 &old_predicate,
                                 &new,
                             );
                             // payload_predicate has new assumptions. FORK!
                             let mut payload_branch = branch.clone();
                             let was = payload_branch.inbox.insert(ekey, payload.clone());
                             assert!(was.is_none()); // INBOX MUST BE EMPTY!
                             // put the original back untouched
                             incomplete2.insert(old_predicate, branch);
                             Some((new, payload_branch))
+                        }
                         Csr::LatterNotFormer => {
                             log!(
                                 &mut m_ctx.inner.logger,
                                 "... poly_recv_run payloadpred {:?} subsumes branch pred {:?}. FORKING",
                                 &old_predicate,
                                 &payload_predicate,
                             );
                             // payload_predicate has new assumptions. FORK!
                             let mut payload_branch = branch.clone();
                             let was = payload_branch.inbox.insert(ekey, payload.clone());
                             assert!(was.is_none()); // INBOX MUST BE EMPTY!
                             // put the original back untouched
                             incomplete2.insert(old_predicate, branch);
                             Some((payload_predicate.clone(), payload_branch))
+                        }
                         Csr::Nonexistant => {
                             log!(
                                 &mut m_ctx.inner.logger,
                                 "... poly_recv_run SKIPPING because branchpred={:?}. payloadpred={:?}",
                                 &old_predicate,
                                 &payload_predicate,
                             );
                             // predicates contradict
                             incomplete2.insert(old_predicate, branch);
                             None
+                        }
+                    }
                 })
                 .collect();
             std::mem::swap(&mut self.incomplete, &mut incomplete2);
             to_run
         };
         log!(
             &mut m_ctx.inner.logger,
             "... DONE FEEDING BRANCHES. {} branches to run!",
             to_run.len(),
         );
         self.poly_run_these_branches(m_ctx, protocol_description, to_run)
+    }
     pub(crate) fn become_mono(
         mut self,
         decision: &Predicate,
         table_row: &mut HashMap<Key, Payload>,
     ) -> MonoP {
         if let Some((_, branch)) = self.complete.drain().find(|(p, _)| decision.satisfies(p)) {
             let BranchP { inbox, state, outbox } = branch;
             for (key, payload) in inbox.into_iter().chain(outbox.into_iter()) {
                 table_row.insert(key, payload);
+            }
             self.incomplete.clear();
             MonoP { state, ekeys: self.ekeys }
         } else {
             panic!("No such solution!")
+        }
+    }
+}
 impl PolyN {
     pub fn sync_recv(
         &mut self,
         ekey: Key,
         logger: &mut String,
         payload: Payload,
         payload_predicate: Predicate,
         solution_storage: &mut SolutionStorage,
     ) {
         let mut branches2: HashMap<_, _> = Default::default();
         for (old_predicate, mut branch) in self.branches.drain() {
             use CommonSatResult as Csr;
             let case = old_predicate.common_satisfier(&payload_predicate);
             let mut report_if_solution =
                 |branch: &BranchN, pred: &Predicate, logger: &mut String| {
                     if branch.to_get.is_empty() {
                         solution_storage.submit_and_digest_subtree_solution(
                             logger,
                             SubtreeId::PolyN,
                             pred.clone(),
                         );
+                    }
                 };
             log!(
                 logger,
                 "Feeding msg {:?} {:?} to native branch with pred {:?}. Predicate case {:?}",
                 &payload_predicate,
                 &payload,
                 &old_predicate,
                 &case
             );
             match case {
                 Csr::Nonexistant => { /* skip branch */ }
                 Csr::FormerNotLatter | Csr::Equivalent => {
                     // Feed the message to this branch in-place. no need to modify pred.
                     if branch.to_get.remove(&ekey) {
                         branch.gotten.insert(ekey, payload.clone());
                         report_if_solution(&branch, &old_predicate, logger);
+                    }
+                }
                 Csr::LatterNotFormer => {
                     // create a new branch with the payload_predicate.
                     let mut forked = branch.clone();
                     if forked.to_get.remove(&ekey) {
                         forked.gotten.insert(ekey, payload.clone());
                         report_if_solution(&forked, &payload_predicate, logger);
                         branches2.insert(payload_predicate.clone(), forked);
+                    }
+                }
                 Csr::New(new) => {
                     // create a new branch with the newly-created predicate
                     let mut forked = branch.clone();
                     if forked.to_get.remove(&ekey) {
                         forked.gotten.insert(ekey, payload.clone());
                         report_if_solution(&forked, &new, logger);
                         branches2.insert(new.clone(), forked);
+                    }
+                }
+            }
             // unlike PolyP machines, Native branches do not become inconsistent
             branches2.insert(old_predicate, branch);
+        }
         log!(
             logger,
             "Native now has {} branches with predicates: {:?}",
             branches2.len(),
             branches2.keys().collect::<Vec<_>>()
         );
         std::mem::swap(&mut branches2, &mut self.branches);
+    }
     pub fn become_mono(
         mut self,
         logger: &mut String,
         decision: &Predicate,
         table_row: &mut HashMap<Key, Payload>,
     ) -> MonoN {
         if let Some((_, branch)) = self
         log!(
             logger,
             "decision {:?} with branch preds {:?}",
             decision,
             self.branches.iter().collect::<Vec<_>>()
         );
         if let Some((branch_pred, branch)) = self
             .branches
             .drain()
             .find(|(p, branch)| branch.to_get.is_empty() && decision.satisfies(p))
+        {
             log!(logger, "decision {:?} mapped to branch {:?}", decision, branch_pred);
             let BranchN { gotten, sync_batch_index, .. } = branch;
             for (&key, payload) in gotten.iter() {
                 assert!(table_row.insert(key, payload.clone()).is_none());
+            }
             MonoN { ekeys: self.ekeys, result: Some((sync_batch_index, gotten)) }
         } else {
             log!(logger, "decision {:?} HAD NO SOLUTION!!?", decision);
             panic!("No such solution!")
+        }
+    }
+}

src/runtime/communication.rs

➞

Show inline comments

 use crate::common::*;
 use crate::runtime::{actors::*, endpoint::*, errors::*, *};
 impl Controller {
     fn end_round_with_decision(&mut self, decision: Predicate) -> Result<(), SyncErr> {
         log!(&mut self.inner.logger, "ENDING ROUND WITH DECISION! {:?}", &decision);
         let mut table_row = HashMap::<Key, _>::default();
         // 1. become_mono for Poly actors
         self.inner.mono_n = self
             .ephemeral
             .poly_n
             .take()
             .map(|poly_n| poly_n.become_mono(&decision, &mut table_row));
         self.inner.mono_n =
             self.ephemeral.poly_n.take().map(|poly_n| {
                 poly_n.become_mono(&mut self.inner.logger, &decision, &mut table_row)
             });
         self.inner.mono_ps.extend(
             self.ephemeral.poly_ps.drain(..).map(|m| m.become_mono(&decision, &mut table_row)),
         );
         // convert (Key=>Payload) map to (ChannelId=>Payload) map.
         let table_row: HashMap<_, _> = table_row
             .into_iter()
             .map(|(ekey, msg)| {
                 let channel_id = self.inner.endpoint_exts.get(ekey).unwrap().info.channel_id;
                 (channel_id, msg)
             })
             .collect();
         // log all firing ports
         for (channel_id, payload) in table_row {
             log!(&mut self.inner.logger, "VALUE {:?} => Message({:?})", channel_id, payload);
+        }
         // log all silent ports
         for channel_id in decision.iter_matching(false) {
             log!(&mut self.inner.logger, "VALUE {:?} => *", channel_id);
+        }
         let announcement =
             CommMsgContents::Announce { oracle: decision }.into_msg(self.inner.round_index);
         for &child_ekey in self.inner.family.children_ekeys.iter() {
             log!(
                 &mut self.inner.logger,
                 "Forwarding {:?} to child with ekey {:?}",
                 &announcement,
                 child_ekey
             );
             self.inner
                 .endpoint_exts
                 .get_mut(child_ekey)
                 .expect("eefef")
                 .endpoint
                 .send(announcement.clone())?;
+        }
         self.inner.round_index += 1;
         self.ephemeral.clear();
         Ok(())
+    }
     // Drain self.ephemeral.solution_storage and handle the new locals. Return decision if one is found
     fn handle_locals_maybe_decide(&mut self) -> Result<bool, SyncErr> {
         if let Some(parent_ekey) = self.inner.family.parent_ekey {
             // I have a parent -> I'm not the leader
             let parent_endpoint =
                 &mut self.inner.endpoint_exts.get_mut(parent_ekey).expect("huu").endpoint;
             for partial_oracle in self.ephemeral.solution_storage.iter_new_local_make_old() {
                 let msg =
                     CommMsgContents::Elaborate { partial_oracle }.into_msg(self.inner.round_index);
                 log!(&mut self.inner.logger, "Sending {:?} to parent {:?}", &msg, parent_ekey);
                 parent_endpoint.send(msg)?;
+            }
             Ok(false)
         } else {
             // I have no parent -> I'm the leader
             assert!(self.inner.family.parent_ekey.is_none());
             let maybe_decision = self.ephemeral.solution_storage.iter_new_local_make_old().next();
             Ok(if let Some(decision) = maybe_decision {
                 log!(&mut self.inner.logger, "DECIDE ON {:?} AS LEADER!", &decision);
                 self.end_round_with_decision(decision)?;
                 true
             } else {
                 false
             })
+        }
+    }
     fn kick_off_native(
         &mut self,
         sync_batches: impl Iterator<Item = SyncBatch>,
     ) -> Result<PolyN, EndpointErr> {
         let MonoN { ekeys, .. } = self.inner.mono_n.take().unwrap();
         let Self { inner: ControllerInner { endpoint_exts, round_index, .. }, .. } = self;
         let mut branches = HashMap::<_, _>::default();
         for (sync_batch_index, SyncBatch { puts, gets }) in sync_batches.enumerate() {
             let ekey_to_channel_id = |ekey| endpoint_exts.get(ekey).unwrap().info.channel_id;
             let all_ekeys = ekeys.iter().copied();
             let all_channel_ids = all_ekeys.map(ekey_to_channel_id);
             let mut predicate = Predicate::new_trivial();
             // assign TRUE for puts and gets
             let true_ekeys = puts.keys().chain(gets.iter()).copied();
             let true_channel_ids = true_ekeys.clone().map(ekey_to_channel_id);
             predicate.batch_assign_nones(true_channel_ids, true);
             // assign FALSE for all in interface not assigned true
             predicate.batch_assign_nones(all_channel_ids.clone(), false);
             if branches.contains_key(&predicate) {
                 // TODO what do I do with redundant predicates?
                 unimplemented!(
                     "Having multiple batches with the same
                     predicate requires the support of oracle boolean variables"
+                )
+            }
             let branch = BranchN { to_get: gets, gotten: Default::default(), sync_batch_index };
             for (ekey, payload) in puts {
                 log!(
                     &mut self.inner.logger,
                     "... ... Initial native put msg {:?} pred {:?} batch {:?}",
                     &payload,
                     &predicate,
                     sync_batch_index,
                 );
                 let msg =
                     CommMsgContents::SendPayload { payload_predicate: predicate.clone(), payload }
                         .into_msg(*round_index);
                 endpoint_exts.get_mut(ekey).unwrap().endpoint.send(msg)?;
+            }
             log!(
                 &mut self.inner.logger,
-                "... Initial native branch (batch index={} with pred {:?}",
                 "... Initial native branch batch index={} with pred {:?}",
                 sync_batch_index,
                 &predicate
             );
             if branch.to_get.is_empty() {
                 self.ephemeral.solution_storage.submit_and_digest_subtree_solution(
                     &mut self.inner.logger,
                     SubtreeId::PolyN,
                     predicate.clone(),
                 );
+            }
             branches.insert(predicate, branch);
+        }
         Ok(PolyN { ekeys, branches })
+    }
     // Runs a synchronous round until all the actors are in decided state OR 1+ are inconsistent.
     // If a native requires setting up, arg `sync_batches` is Some, and those are used as the sync batches.
     pub fn sync_round(
         &mut self,
         deadline: Instant,
         sync_batches: Option<impl Iterator<Item = SyncBatch>>,
     ) -> Result<(), SyncErr> {
         // TODO! fuse handle_locals_return_decision and end_round_return_decision
         assert!(self.ephemeral.is_clear());
         log!(
             &mut self.inner.logger,
             "~~~~~~~~ SYNC ROUND STARTS! ROUND={} ~~~~~~~~~",
             self.inner.round_index
         );
         // 1. Run the Mono for each Mono actor (stored in `self.mono_ps`).
         //    Some actors are dropped. some new actors are created.
         //    Ultimately, we have 0 Mono actors and a list of unnamed sync_actors
         log!(&mut self.inner.logger, "Got {} MonoP's to run!", self.inner.mono_ps.len());
         self.ephemeral.poly_ps.clear();
         // let mut poly_ps: Vec<PolyP> = vec![];
         while let Some(mut mono_p) = self.inner.mono_ps.pop() {
             let mut m_ctx = MonoPContext {
                 ekeys: &mut mono_p.ekeys,
                 inner: &mut self.inner,
                 // endpoint_exts: &mut self.endpoint_exts,
                 // mono_ps: &mut self.mono_ps,
                 // channel_id_stream: &mut self.channel_id_stream,
             };
             // cross boundary into crate::protocol
             let blocker = mono_p.state.pre_sync_run(&mut m_ctx, &self.protocol_description);
             log!(&mut self.inner.logger, "... MonoP's pre_sync_run got blocker {:?}", &blocker);
             match blocker {
                 MonoBlocker::Inconsistent => return Err(SyncErr::Inconsistent),
                 MonoBlocker::ComponentExit => drop(mono_p),
                 MonoBlocker::SyncBlockStart => self.ephemeral.poly_ps.push(mono_p.into()),
+            }
+        }
         log!(
             &mut self.inner.logger,
             "Finished running all MonoPs! Have {} PolyPs waiting",
             self.ephemeral.poly_ps.len()
         );
         // 3. define the mapping from ekey -> actor
         //    this is needed during the event loop to determine which actor
         //    should receive the incoming message.
         //    TODO: store and update this mapping rather than rebuilding it each round.
         let ekey_to_holder: HashMap<Key, PolyId> = {
             use PolyId::*;
             let n = self.inner.mono_n.iter().flat_map(|m| m.ekeys.iter().map(move |&e| (e, N)));
             let p = self
                 .ephemeral
                 .poly_ps
                 .iter()
                 .enumerate()
                 .flat_map(|(index, m)| m.ekeys.iter().map(move |&e| (e, P { index })));
             n.chain(p).collect()
         };
         log!(
             &mut self.inner.logger,
             "SET OF PolyPs and MonoPs final! ekey lookup map is {:?}",
             &ekey_to_holder
         );
         // 4. Create the solution storage. it tracks the solutions of "subtrees"
         //    of the controller in the overlay tree.
         self.ephemeral.solution_storage.reset({
             let n = self.inner.mono_n.iter().map(|_| SubtreeId::PolyN);
             let m = (0..self.ephemeral.poly_ps.len()).map(|index| SubtreeId::PolyP { index });
             let c = self
                 .inner
                 .family
                 .children_ekeys
                 .iter()
                 .map(|&ekey| SubtreeId::ChildController { ekey });
             let subtree_id_iter = n.chain(m).chain(c);
             log!(
                 &mut self.inner.logger,
                 "Solution Storage has subtree Ids: {:?}",
                 &subtree_id_iter.clone().collect::<Vec<_>>()
             );
             subtree_id_iter
         });
         // 5. kick off the synchronous round of the native actor if it exists
         log!(&mut self.inner.logger, "Kicking off native's synchronous round...");
         assert_eq!(sync_batches.is_some(), self.inner.mono_n.is_some()); // TODO better err
         self.ephemeral.poly_n = if let Some(sync_batches) = sync_batches {
             // using if let because of nested ? operator
             // TODO check that there are 1+ branches or NO SOLUTION
             let poly_n = self.kick_off_native(sync_batches)?;
             log!(
                 &mut self.inner.logger,
                 "PolyN kicked off, and has branches with predicates... {:?}",
                 poly_n.branches.keys().collect::<Vec<_>>()
             );
             Some(poly_n)
         } else {
             log!(&mut self.inner.logger, "NO NATIVE COMPONENT");
             None
         };
         // 6. Kick off the synchronous round of each protocol actor
         //    If just one actor becomes inconsistent now, there can be no solution!
         //    TODO distinguish between completed and not completed poly_p's?
         log!(&mut self.inner.logger, "Kicking off {} PolyP's.", self.ephemeral.poly_ps.len());
         for (index, poly_p) in self.ephemeral.poly_ps.iter_mut().enumerate() {
             let my_subtree_id = SubtreeId::PolyP { index };
             let m_ctx = PolyPContext {
                 my_subtree_id,
                 inner: &mut self.inner,
                 solution_storage: &mut self.ephemeral.solution_storage,
             };
             use SyncRunResult as Srr;
             let blocker = poly_p.poly_run(m_ctx, &self.protocol_description)?;
             log!(&mut self.inner.logger, "... PolyP's poly_run got blocker {:?}", &blocker);
             match blocker {
                 Srr::NoBranches => return Err(SyncErr::Inconsistent),
                 Srr::AllBranchesComplete | Srr::BlockingForRecv => (),
+            }
+        }
         log!(&mut self.inner.logger, "All Poly machines have been kicked off!");
         // 7. `solution_storage` may have new solutions for this controller
         //    handle their discovery. LEADER => announce, otherwise => send to parent
+        {
             let peeked = self.ephemeral.solution_storage.peek_new_locals().collect::<Vec<_>>();
             log!(
                 &mut self.inner.logger,
                 "Got {} controller-local solutions before a single RECV: {:?}",
                 peeked.len(),
                 peeked
             );
+        }
         if self.handle_locals_maybe_decide()? {
             return Ok(());
+        }
         // 4. Receive incoming messages until the DECISION is made
         log!(&mut self.inner.logger, "No decision yet. Time to recv messages");
         self.undelay_all();
         'recv_loop: loop {
             let received = self.recv(deadline)?.ok_or_else(|| {
                 log!(
                     &mut self.inner.logger,
                     ":( timing out. Solutions storage in state... {:#?}",
                     &self.ephemeral.solution_storage
                 );
                 SyncErr::Timeout
             })?;
             let current_content = match received.msg {
                 Msg::SetupMsg(_) => {
                     log!(&mut self.inner.logger, "recvd message {:?} and its SETUP :(", &received);
                     // This occurs in the event the connector was malformed during connect()
                     return Err(SyncErr::UnexpectedSetupMsg);
+                }
                 Msg::CommMsg(CommMsg { round_index, .. })
                     if round_index < self.inner.round_index =>
+                {
                     // Old message! Can safely discard
                     log!(&mut self.inner.logger, "recvd message {:?} and its OLD! :(", &received);
                     drop(received);
                     continue 'recv_loop;
+                }
                 Msg::CommMsg(CommMsg { round_index, .. })
                     if round_index > self.inner.round_index =>
+                {
                     // Message from a next round. Keep for later!
                     log!(
                         &mut self.inner.logger,
                         "ecvd message {:?} and its for later. DELAY! :(",
                         &received
                     );

src/runtime/mod.rs

➞

Show inline comments

@@ @@ -197,334 +197,334 @@ trait Messengerlike { @@
         self.get_endpoint_mut(to).send(msg)
+    }
     // attempt to receive a message from one of the endpoints before the deadline
     fn recv(&mut self, deadline: Instant) -> Result<Option<ReceivedMsg>, MessengerRecvErr> {
         // try get something buffered
         if let Some(x) = self.get_state_mut().undelayed.pop() {
             return Ok(Some(x));
+        }
         loop {
             // polled_undrained may not be empty
             while let Some(eekey) = self.get_state_mut().polled_undrained.pop() {
                 if let Some(msg) = self.get_endpoint_mut(eekey).recv()? {
                     // this endpoint MAY still have messages! check again in future
                     self.get_state_mut().polled_undrained.insert(eekey);
                     return Ok(Some(ReceivedMsg { recipient: eekey, msg }));
+                }
+            }
             let state = self.get_state_mut();
             match state.poll_events(deadline) {
                 Ok(()) => {
                     for e in state.events.iter() {
                         state.polled_undrained.insert(Key::from_token(e.token()));
+                    }
+                }
                 Err(PollDeadlineErr::PollingFailed) => return Err(MessengerRecvErr::PollingFailed),
                 Err(PollDeadlineErr::Timeout) => return Ok(None),
+            }
+        }
+    }
+}
 /////////////////////////////////
 impl Debug for SolutionStorage {
     fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
         f.pad("Solutions: [")?;
         for (subtree_id, &index) in self.subtree_id_to_index.iter() {
             let sols = &self.subtree_solutions[index];
             f.write_fmt(format_args!("{:?} => {:?}, ", subtree_id, sols))?;
+        }
         f.pad("]")
+    }
+}
 impl From<EvalErr> for SyncErr {
     fn from(e: EvalErr) -> SyncErr {
         SyncErr::EvalErr(e)
+    }
+}
 impl From<MessengerRecvErr> for SyncErr {
     fn from(e: MessengerRecvErr) -> SyncErr {
         SyncErr::MessengerRecvErr(e)
+    }
+}
 impl From<MessengerRecvErr> for ConnectErr {
     fn from(e: MessengerRecvErr) -> ConnectErr {
         ConnectErr::MessengerRecvErr(e)
+    }
+}
 impl From<EndpointErr> for MessengerRecvErr {
     fn from(e: EndpointErr) -> MessengerRecvErr {
         MessengerRecvErr::EndpointErr(e)
+    }
+}
 impl<T> Default for Arena<T> {
     fn default() -> Self {
         Self { storage: vec![] }
+    }
+}
 impl<T> Arena<T> {
     pub fn alloc(&mut self, t: T) -> Key {
         self.storage.push(t);
         Key::from_raw(self.storage.len() as u64 - 1)
+    }
     pub fn get(&self, key: Key) -> Option<&T> {
         self.storage.get(key.to_raw() as usize)
+    }
     pub fn get_mut(&mut self, key: Key) -> Option<&mut T> {
         self.storage.get_mut(key.to_raw() as usize)
+    }
     pub fn type_convert<X>(self, f: impl FnMut((Key, T)) -> X) -> Arena<X> {
         Arena { storage: self.keyspace().zip(self.storage.into_iter()).map(f).collect() }
+    }
     pub fn iter(&self) -> impl Iterator<Item = (Key, &T)> {
         self.keyspace().zip(self.storage.iter())
+    }
     pub fn len(&self) -> usize {
         self.storage.len()
+    }
     pub fn keyspace(&self) -> impl Iterator<Item = Key> {
         (0..(self.storage.len() as u64)).map(Key::from_raw)
+    }
+}
 impl ChannelIdStream {
     fn new(controller_id: ControllerId) -> Self {
         Self { controller_id, next_channel_index: 0 }
+    }
     fn next(&mut self) -> ChannelId {
         self.next_channel_index += 1;
         ChannelId { controller_id: self.controller_id, channel_index: self.next_channel_index - 1 }
+    }
+}
 impl MessengerState {
     // does NOT guarantee that events is non-empty
     fn poll_events(&mut self, deadline: Instant) -> Result<(), PollDeadlineErr> {
         use PollDeadlineErr::*;
         self.events.clear();
         let poll_timeout = deadline.checked_duration_since(Instant::now()).ok_or(Timeout)?;
         self.poll.poll(&mut self.events, Some(poll_timeout)).map_err(|_| PollingFailed)?;
         Ok(())
+    }
+}
 impl From<PollDeadlineErr> for ConnectErr {
     fn from(e: PollDeadlineErr) -> ConnectErr {
         match e {
             PollDeadlineErr::Timeout => ConnectErr::Timeout,
             PollDeadlineErr::PollingFailed => ConnectErr::PollingFailed,
+        }
+    }
+}
 impl std::ops::Not for Polarity {
     type Output = Self;
     fn not(self) -> Self::Output {
         use Polarity::*;
         match self {
             Putter => Getter,
             Getter => Putter,
+        }
+    }
+}
 impl Predicate {
     // returns true IFF self.unify would return Equivalent OR FormerNotLatter
     pub fn satisfies(&self, other: &Self) -> bool {
         let mut s_it = self.assigned.iter();
         let mut s = if let Some(s) = s_it.next() {
+            s
         } else {
             return other.assigned.is_empty();
         };
         for (oid, ob) in other.assigned.iter() {
             while s.0 < oid {
                 s = if let Some(s) = s_it.next() {
+                    s
                 } else {
                     return false;
                 };
+            }
             if s.0 > oid || s.1 != ob {
                 return false;
+            }
+        }
         true
+    }
     /// Given self and other, two predicates, return the most general Predicate possible, N
     /// such that n.satisfies(self) && n.satisfies(other).
     /// If none exists Nonexistant is returned.
     /// If the resulting predicate is equivlanet to self, other, or both,
     /// FormerNotLatter, LatterNotFormer and Equivalent are returned respectively.
     /// otherwise New(N) is returned.
     pub fn common_satisfier(&self, other: &Self) -> CommonSatResult {
         use CommonSatResult::*;
         // iterators over assignments of both predicates. Rely on SORTED ordering of BTreeMap's keys.
         let [mut s_it, mut o_it] = [self.assigned.iter(), other.assigned.iter()];
         let [mut s, mut o] = [s_it.next(), o_it.next()];
         // lists of assignments in self but not other and vice versa.
         let [mut s_not_o, mut o_not_s] = [vec![], vec![]];
         loop {
             match [s, o] {
                 [None, None] => break,
                 [None, Some(x)] => {
                     o_not_s.push(x);
                     o_not_s.extend(o_it);
                     break;
+                }
                 [Some(x), None] => {
                     s_not_o.push(x);
                     s_not_o.extend(s_it);
                     break;
+                }
                 [Some((sid, sb)), Some((oid, ob))] => {
                     if sid < oid {
                         // o is missing this element
                         s_not_o.push((sid, sb));
                         s = s_it.next();
                     } else if sid > oid {
                         // s is missing this element
-                        o_not_s.push((sid, sb));
+                        o_not_s.push((oid, ob));
                         o = o_it.next();
                     } else if sb != ob {
                         assert_eq!(sid, oid);
                         // both predicates assign the variable but differ on the value
                         return Nonexistant;
                     } else {
                         // both predicates assign the variable to the same value
                         s = s_it.next();
                         o = o_it.next();
+                    }
+                }
+            }
+        }
         // Observed zero inconsistencies. A unified predicate exists...
         match [s_not_o.is_empty(), o_not_s.is_empty()] {
             [true, true] => Equivalent,       // ... equivalent to both.
             [false, true] => FormerNotLatter, // ... equivalent to self.
             [true, false] => LatterNotFormer, // ... equivalent to other.
             [false, false] => {
                 // ... which is the union of the predicates' assignments but
                 //     is equivalent to neither self nor other.
-                let mut predicate = self.clone();
+                let mut new = self.clone();
                 for (&id, &b) in o_not_s {
-                    predicate.assigned.insert(id, b);
+                    new.assigned.insert(id, b);
+                }
-                New(predicate)
+                New(new)
+            }
+        }
+    }
     pub fn iter_matching(&self, value: bool) -> impl Iterator<Item = ChannelId> + '_ {
         self.assigned
             .iter()
             .filter_map(move |(&channel_id, &b)| if b == value { Some(channel_id) } else { None })
+    }
     pub fn batch_assign_nones(
         &mut self,
         channel_ids: impl Iterator<Item = ChannelId>,
         value: bool,
     ) {
         for channel_id in channel_ids {
             self.assigned.entry(channel_id).or_insert(value);
+        }
+    }
     pub fn replace_assignment(&mut self, channel_id: ChannelId, value: bool) -> Option<bool> {
         self.assigned.insert(channel_id, value)
+    }
     pub fn union_with(&self, other: &Self) -> Option<Self> {
         let mut res = self.clone();
         for (&channel_id, &assignment_1) in other.assigned.iter() {
             match res.assigned.insert(channel_id, assignment_1) {
                 Some(assignment_2) if assignment_1 != assignment_2 => return None,
                 _ => {}
+            }
+        }
         Some(res)
+    }
     pub fn query(&self, x: ChannelId) -> Option<bool> {
         self.assigned.get(&x).copied()
+    }
     pub fn new_trivial() -> Self {
         Self { assigned: Default::default() }
+    }
+}
 impl Debug for Predicate {
     fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
         f.pad("{")?;
         for (ChannelId { controller_id, channel_index }, &v) in self.assigned.iter() {
             f.write_fmt(format_args!(
                 "({:?},{:?})=>{}, ",
                 controller_id,
                 channel_index,
                 if v { 'T' } else { 'F' }
             ))?
+        }
         f.pad("}")
+    }
+}
 #[test]
 fn pred_sat() {
     use maplit::btreemap;
     let mut c = ChannelIdStream::new(0);
     let ch = std::iter::repeat_with(move || c.next()).take(5).collect::<Vec<_>>();
     let p = Predicate::new_trivial();
     let p_0t = Predicate { assigned: btreemap! { ch[0] => true } };
     let p_0f = Predicate { assigned: btreemap! { ch[0] => false } };
     let p_0f_3f = Predicate { assigned: btreemap! { ch[0] => false, ch[3] => false } };
     let p_0f_3t = Predicate { assigned: btreemap! { ch[0] => false, ch[3] => true } };
     assert!(p.satisfies(&p));
     assert!(p_0t.satisfies(&p_0t));
     assert!(p_0f.satisfies(&p_0f));
     assert!(p_0f_3f.satisfies(&p_0f_3f));
     assert!(p_0f_3t.satisfies(&p_0f_3t));
     assert!(p_0t.satisfies(&p));
     assert!(p_0f.satisfies(&p));
     assert!(p_0f_3f.satisfies(&p_0f));
     assert!(p_0f_3t.satisfies(&p_0f));
     assert!(!p.satisfies(&p_0t));
     assert!(!p.satisfies(&p_0f));
     assert!(!p_0f.satisfies(&p_0t));
     assert!(!p_0t.satisfies(&p_0f));
     assert!(!p_0f_3f.satisfies(&p_0f_3t));
     assert!(!p_0f_3t.satisfies(&p_0f_3f));
     assert!(!p_0t.satisfies(&p_0f_3f));
     assert!(!p_0f.satisfies(&p_0f_3f));
     assert!(!p_0t.satisfies(&p_0f_3t));
     assert!(!p_0f.satisfies(&p_0f_3t));
+}
 #[test]
 fn pred_common_sat() {
     use maplit::btreemap;
     use CommonSatResult::*;
     let mut c = ChannelIdStream::new(0);
     let ch = std::iter::repeat_with(move || c.next()).take(5).collect::<Vec<_>>();
     let p = Predicate::new_trivial();
     let p_0t = Predicate { assigned: btreemap! { ch[0] => true } };
     let p_0f = Predicate { assigned: btreemap! { ch[0] => false } };
     let p_3f = Predicate { assigned: btreemap! { ch[3] => false } };
     let p_0f_3f = Predicate { assigned: btreemap! { ch[0] => false, ch[3] => false } };
     let p_0f_3t = Predicate { assigned: btreemap! { ch[0] => false, ch[3] => true } };
     assert_eq![p.common_satisfier(&p), Equivalent];
     assert_eq![p_0t.common_satisfier(&p_0t), Equivalent];
     assert_eq![p.common_satisfier(&p_0t), LatterNotFormer];
     assert_eq![p_0t.common_satisfier(&p), FormerNotLatter];
     assert_eq![p_0t.common_satisfier(&p_0f), Nonexistant];
     assert_eq![p_0f_3t.common_satisfier(&p_0f_3f), Nonexistant];
     assert_eq![p_0f_3t.common_satisfier(&p_3f), Nonexistant];
     assert_eq![p_3f.common_satisfier(&p_0f_3t), Nonexistant];
     assert_eq![p_0f.common_satisfier(&p_3f), New(p_0f_3f)];
+}

src/test/connector.rs

➞

Show inline comments

 extern crate test_generator;
 use super::*;
 use crate::common::*;
 use crate::runtime::{errors::*, PortBinding::*};
 static PDL: &[u8] = b"
 primitive blocked(in i, out o) {
     while(true) synchronous {}
+}
 primitive forward(in i, out o) {
     while(true) synchronous {
         put(o, get(i));
+    }
+}
 primitive sync(in i, out o) {
     while(true) synchronous {
         if (fires(i)) put(o, get(i));
+    }
+}
 primitive fifo_1(in i, out o) {
     msg holding = null;
     while(true) synchronous {
         if (holding == null && fires(i)) {
             holding = get(i);
         } else if (holding != null && fires(o)) {
             put(o, holding);
             holding = null;
+        }
+    }
+}
 primitive alternator_2(in i, out a, out b) {
     while(true) {
         synchronous { put(a, get(i)); }
         synchronous { put(b, get(i)); }
+    }
+}
 composite sync_2(in i, out o) {
     channel x -> y;
     new sync(i, x);
     new sync(y, o);
+}
 composite forward_pair(in ia, out oa, in ib, out ob) {
     new forward(ia, oa);
     new forward(ib, ob);
 primitive exchange(in ai, out ao, in bi, out bo) {
     // Note the implicit causal relationship
     while(true) synchronous {
         if(fires(ai)) {
             put(bo, get(ai));
             put(ao, get(bi));
+        }
+    }
+}
 primitive forward_nonzero(in i, out o) {
     while(true) synchronous {
         msg m = get(i);
         assert(m[0] != 0);
         put(o, m);
+    }
+}
 primitive token_spout(out o) {
     while(true) synchronous {
         put(o, create(0));
+    }
+}
 primitive wait_n(int to_wait, out o) {
     while(to_wait > 0) synchronous() to_wait -= 1;
     synchronous { put(o, create(0)); }
+}
 composite wait_10(out o) {
     new wait_n(10, o);
+}
 ";
 #[test]
 fn connects_ok() {
     // Test if we can connect natives using the given PDL
     /*
     Alice -->silence--P|A-->silence--> Bob
     */
     let timeout = Duration::from_millis(1_500);
     let addrs = [next_addr()];
     assert!(run_connector_set(&[
         &|x| {
             // Alice
             x.configure(PDL, b"blocked").unwrap();
             x.bind_port(0, Native).unwrap();
             x.bind_port(1, Passive(addrs[0])).unwrap();
             x.connect(timeout).unwrap();
         },
         &|x| {
             // Bob
             x.configure(PDL, b"blocked").unwrap();
             x.bind_port(0, Active(addrs[0])).unwrap();
             x.bind_port(1, Native).unwrap();
             x.connect(timeout).unwrap();
         },
     ]));
+}
 #[test]
 fn connected_but_silent_natives() {
     // Test if we can connect natives and have a trivial sync round
     /*
     Alice -->silence--P|A-->silence--> Bob
     */
     let timeout = Duration::from_millis(1_500);
     let addrs = [next_addr()];
     assert!(run_connector_set(&[
         &|x| {
             // Alice
             x.configure(PDL, b"blocked").unwrap();
             x.bind_port(0, Native).unwrap();
             x.bind_port(1, Passive(addrs[0])).unwrap();
             x.connect(timeout).unwrap();
             assert_eq!(Ok(0), x.sync(timeout));
         },
         &|x| {
             // Bob
             x.configure(PDL, b"blocked").unwrap();
             x.bind_port(0, Active(addrs[0])).unwrap();
             x.bind_port(1, Native).unwrap();
             x.connect(timeout).unwrap();
             assert_eq!(Ok(0), x.sync(timeout));
         },
     ]));
+}
 #[test]
 fn self_forward_ok() {
     // Test a deterministic system
     // where a native has no network bindings
     // and sends messages to itself
     /*
         /-->\
     Alice   forward
         \<--/
     */
     let timeout = Duration::from_millis(1_500);
     const N: usize = 5;
     static MSG: &[u8] = b"Echo!";
     assert!(run_connector_set(&[
         //
         &|x| {
             // Alice
             x.configure(PDL, b"forward").unwrap();
             x.bind_port(0, Native).unwrap();
             x.bind_port(1, Native).unwrap();
             x.connect(timeout).unwrap();
             for _ in 0..N {
                 x.put(0, MSG.to_vec()).unwrap();
                 x.get(1).unwrap();
                 assert_eq!(Ok(0), x.sync(timeout));
                 assert_eq!(Ok(MSG), x.read_gotten(1));
+            }
         },
     ]));
+}
 #[test]
 fn token_spout_ok() {
     // Test a deterministic system where the proto
     // creates token messages
     /*
     Alice<--token_spout
     */
     let timeout = Duration::from_millis(1_500);
     const N: usize = 5;
     assert!(run_connector_set(&[
         //
         &|x| {
             // Alice
             x.configure(PDL, b"token_spout").unwrap();
             x.bind_port(0, Native).unwrap();
             x.connect(timeout).unwrap();
             for _ in 0..N {
                 x.get(0).unwrap();
                 assert_eq!(Ok(0), x.sync(timeout));
                 assert_eq!(Ok(&[] as &[u8]), x.read_gotten(0));
+            }
         },
     ]));
+}
 #[test]
 fn waiter_ok() {
     // Test a stateful proto that blocks port 0 for 10 rounds
     // and then sends a single token on the 11th
     /*
     Alice<--token_spout
     */
     let timeout = Duration::from_millis(1_500);
     assert!(run_connector_set(&[
         //
         &|x| {
             // Alice
             x.configure(PDL, b"wait_10").unwrap();
             x.bind_port(0, Native).unwrap();
             x.connect(timeout).unwrap();
             for _ in 0..10 {
                 assert_eq!(Ok(0), x.sync(timeout));
                 assert_eq!(Err(ReadGottenErr::DidNotGet), x.read_gotten(0));
+            }
             x.get(0).unwrap();
             assert_eq!(Ok(0), x.sync(timeout));
             assert_eq!(Ok(&[] as &[u8]), x.read_gotten(0));
         },
     ]));
+}
 #[test]
 fn self_forward_timeout() {
     // Test a deterministic system
     // where a native has no network bindings
     // and sends messages to itself
     /*
         /-->\
     Alice   forward
         \<--/
     */
     let timeout = Duration::from_millis(500);
     static MSG: &[u8] = b"Echo!";
     assert!(run_connector_set(&[
         //
         &|x| {
             // Sender
             x.configure(PDL, b"forward").unwrap();
             x.bind_port(0, Native).unwrap();
             x.bind_port(1, Native).unwrap();
             x.connect(timeout).unwrap();
             x.put(0, MSG.to_vec()).unwrap();
             // native and forward components cannot find a solution
             assert_eq!(Err(SyncErr::Timeout), x.sync(timeout));
         },
     ]));
+}
 #[test]
 fn forward_det() {
     // Test if a deterministic protocol and natives can pass one message
     /*
     Alice -->forward--P|A-->forward--> Bob
     */
     let timeout = Duration::from_millis(1_500);
     let addrs = [next_addr()];
@@ @@ -329,382 +334,378 @@ fn putter_determines() { @@
             // Bob
             x.configure(PDL, b"sync").unwrap();
             x.bind_port(0, Passive(addrs[0])).unwrap();
             x.bind_port(1, Native).unwrap();
             x.connect(timeout).unwrap();
             for _i in 0..N {
                 // batches [{0=>*}, {0=>?}]
                 x.get(0).unwrap();
                 x.next_batch().unwrap();
                 assert_eq!(Ok(0), x.sync(timeout));
                 assert_eq!(Ok(MSG), x.read_gotten(0));
+            }
         },
     ]));
+}
 #[test]
 fn getter_determines() {
     // putter and getter
     /*
     Alice -->sync--A|P-->sync--> Bob
     */
     let timeout = Duration::from_millis(1_500);
     let addrs = [next_addr()];
     const N: usize = 5;
     static MSG: &[u8] = b"Hidey ho!";
     assert!(run_connector_set(&[
         //
         &|x| {
             // Alice
             x.configure(PDL, b"sync").unwrap();
             x.bind_port(0, Native).unwrap();
             x.bind_port(1, Active(addrs[0])).unwrap();
             x.connect(timeout).unwrap();
             for _i in 0..N {
                 // batches [{0=>?}, {0=>*}]
                 x.put(0, MSG.to_vec()).unwrap();
                 x.next_batch().unwrap();
                 assert_eq!(Ok(0), x.sync(timeout));
+            }
         },
         &|x| {
             // Bob
             x.configure(PDL, b"sync").unwrap();
             x.bind_port(0, Passive(addrs[0])).unwrap();
             x.bind_port(1, Native).unwrap();
             x.connect(timeout).unwrap();
             for _i in 0..N {
                 x.get(0).unwrap();
                 assert_eq!(Ok(0), x.sync(timeout));
                 assert_eq!(Ok(MSG), x.read_gotten(0));
+            }
         },
     ]));
+}
 #[test]
 fn fifo_2() {
     // Test a deterministic system which
     // alternates sending Sender's messages to A or B
     /*                    /--|-->A
     Sender -->alternator_2
                           \--|-->B
     */
     let timeout = Duration::from_millis(1_500);
     let addrs = [next_addr(), next_addr()];
     const N: usize = 5;
     static MSG: &[u8] = b"message";
     assert!(run_connector_set(&[
         //
         &|x| {
             // Sender
             x.configure(PDL, b"alternator_2").unwrap();
             x.bind_port(0, Native).unwrap();
             x.bind_port(1, Passive(addrs[0])).unwrap();
             x.bind_port(2, Passive(addrs[1])).unwrap();
             x.connect(timeout).unwrap();
             for _ in 0..N {
                 for _ in 0..2 {
                     x.put(0, MSG.to_vec()).unwrap();
                     assert_eq!(0, x.sync(timeout).unwrap());
+                }
+            }
         },
         &|x| {
             // A
             x.configure(PDL, b"sync").unwrap();
             x.bind_port(0, Active(addrs[0])).unwrap();
             x.bind_port(1, Native).unwrap();
             x.connect(timeout).unwrap();
             for _ in 0..N {
                 // get msg round
                 x.get(0).unwrap();
                 assert_eq!(Ok(0), x.sync(timeout)); // GET ONE
                 assert_eq!(Ok(MSG), x.read_gotten(0));
                 // silent round
                 assert_eq!(Ok(0), x.sync(timeout)); // MISS ONE
                 assert_eq!(Err(ReadGottenErr::DidNotGet), x.read_gotten(0));
+            }
         },
         &|x| {
             // B
             x.configure(PDL, b"sync").unwrap();
             x.bind_port(0, Active(addrs[1])).unwrap();
             x.bind_port(1, Native).unwrap();
             x.connect(timeout).unwrap();
             for _ in 0..N {
                 // silent round
                 assert_eq!(Ok(0), x.sync(timeout)); // MISS ONE
                 assert_eq!(Err(ReadGottenErr::DidNotGet), x.read_gotten(0));
                 // get msg round
                 x.get(0).unwrap();
                 assert_eq!(Ok(0), x.sync(timeout)); // GET ONE
                 assert_eq!(Ok(MSG), x.read_gotten(0));
+            }
         },
     ]));
+}
 #[test]
 fn alternator_2() {
     // Test a deterministic system which
     // alternates sending Sender's messages to A or B
     /*                    /--|-->A
     Sender -->alternator_2
                           \--|-->B
     */
     let timeout = Duration::from_millis(1_500);
     let addrs = [next_addr(), next_addr()];
     const N: usize = 5;
     static MSG: &[u8] = b"message";
     assert!(run_connector_set(&[
         //
         &|x| {
             // Sender
             x.configure(PDL, b"alternator_2").unwrap();
             x.bind_port(0, Native).unwrap();
             x.bind_port(1, Passive(addrs[0])).unwrap();
             x.bind_port(2, Passive(addrs[1])).unwrap();
             x.connect(timeout).unwrap();
             for _ in 0..N {
                 for _ in 0..2 {
                     x.put(0, MSG.to_vec()).unwrap();
                     assert_eq!(0, x.sync(timeout).unwrap());
+                }
+            }
         },
         &|x| {
             // A
             x.configure(PDL, b"sync").unwrap();
             x.bind_port(0, Active(addrs[0])).unwrap();
             x.bind_port(1, Native).unwrap();
             x.connect(timeout).unwrap();
             for _ in 0..N {
                 // get msg round
                 x.get(0).unwrap();
                 assert_eq!(Ok(0), x.sync(timeout)); // GET ONE
                 assert_eq!(Ok(MSG), x.read_gotten(0));
                 // silent round
                 assert_eq!(Ok(0), x.sync(timeout)); // MISS ONE
                 assert_eq!(Err(ReadGottenErr::DidNotGet), x.read_gotten(0));
+            }
         },
         &|x| {
             // B
             x.configure(PDL, b"sync").unwrap();
             x.bind_port(0, Active(addrs[1])).unwrap();
             x.bind_port(1, Native).unwrap();
             x.connect(timeout).unwrap();
             for _ in 0..N {
                 // silent round
                 assert_eq!(Ok(0), x.sync(timeout)); // MISS ONE
                 assert_eq!(Err(ReadGottenErr::DidNotGet), x.read_gotten(0));
                 // get msg round
                 x.get(0).unwrap();
                 assert_eq!(Ok(0), x.sync(timeout)); // GET ONE
                 assert_eq!(Ok(MSG), x.read_gotten(0));
+            }
         },
     ]));
+}
 #[test]
 // PANIC TODO: eval::1536
 fn composite_chain_a() {
     // Check if composition works. Forward messages through long chains
     /*
     Alice -->sync-->sync-->A|P-->sync--> Bob
     */
     let timeout = Duration::from_millis(1_500);
     let addrs = [next_addr(), next_addr()];
     const N: usize = 1;
     static MSG: &[u8] = b"SSS";
     assert!(run_connector_set(&[
         //
         &|x| {
             // Alice
             x.configure(PDL, b"sync_2").unwrap();
             x.bind_port(0, Native).unwrap();
             x.bind_port(1, Active(addrs[0])).unwrap();
             x.connect(timeout).unwrap();
             for _ in 0..N {
                 x.put(0, MSG.to_vec()).unwrap();
                 assert_eq!(0, x.sync(timeout).unwrap());
+            }
         },
         &|x| {
             // Bob
             x.configure(PDL, b"forward").unwrap();
             x.bind_port(0, Passive(addrs[0])).unwrap();
             x.bind_port(1, Native).unwrap();
             x.connect(timeout).unwrap();
             for _ in 0..N {
                 // get msg round
                 x.get(0).unwrap();
                 assert_eq!(Ok(0), x.sync(timeout));
                 assert_eq!(Ok(MSG), x.read_gotten(0));
+            }
         },
     ]));
+}
 #[test]
 // PANIC TODO: eval::1536
 fn composite_chain_b() {
     // Check if composition works. Forward messages through long chains
     /*
     Alice -->sync-->sync-->A|P-->sync-->sync--> Bob
     */
     let timeout = Duration::from_millis(1_500);
     let addrs = [next_addr(), next_addr()];
     const N: usize = 1;
     static MSG: &[u8] = b"SSS";
     assert!(run_connector_set(&[
         //
         &|x| {
             // Alice
             x.configure(PDL, b"sync_2").unwrap();
             x.bind_port(0, Native).unwrap();
             x.bind_port(1, Active(addrs[0])).unwrap();
             x.connect(timeout).unwrap();
             for _ in 0..N {
                 x.put(0, MSG.to_vec()).unwrap();
                 assert_eq!(0, x.sync(timeout).unwrap());
+            }
         },
         &|x| {
             // Bob
             x.configure(PDL, b"sync_2").unwrap();
             x.bind_port(0, Passive(addrs[0])).unwrap();
             x.bind_port(1, Native).unwrap();
             x.connect(timeout).unwrap();
             for _ in 0..N {
                 // get msg round
                 x.get(0).unwrap();
                 assert_eq!(Ok(0), x.sync(timeout));
                 assert_eq!(Ok(MSG), x.read_gotten(0));
+            }
         },
     ]));
+}
 #[test]
 // PANIC TODO: eval::1605
 fn exchange() {
     /*
         /-->forward-->P|A-->forward-->\
     Alice                             Bob
         \<--forward<--P|A<--forward<--/
         /-->\      /-->P|A-->\      /-->\
     Alice   exchange         exchange   Bob
         \<--/      \<--P|A<--/      \<--/
     */
     let timeout = Duration::from_millis(1_500);
     let addrs = [next_addr(), next_addr()];
     const N: usize = 1;
     assert!(run_connector_set(&[
         //
         &|x| {
             // Alice
-            x.configure(PDL, b"forward_pair").unwrap();
+            x.configure(PDL, b"exchange").unwrap();
             x.bind_port(0, Native).unwrap(); // native in
             x.bind_port(1, Passive(addrs[0])).unwrap(); // peer out
             x.bind_port(2, Passive(addrs[1])).unwrap(); // peer in
             x.bind_port(3, Native).unwrap(); // native out
             x.bind_port(1, Native).unwrap(); // native out
             x.bind_port(2, Passive(addrs[0])).unwrap(); // peer out
             x.bind_port(3, Passive(addrs[1])).unwrap(); // peer in
             x.connect(timeout).unwrap();
             for _ in 0..N {
                 x.put(0, b"A->B".to_vec()).unwrap();
                 x.get(1).unwrap();
                 assert_eq!(Ok(()), x.put(0, b"A->B".to_vec()));
                 assert_eq!(Ok(()), x.get(1));
                 assert_eq!(Ok(0), x.sync(timeout));
-                assert_eq!(Ok(b"B->A" as &[u8]), x.read_gotten(0));
+                assert_eq!(Ok(b"B->A" as &[u8]), x.read_gotten(1));
+            }
         },
         &|x| {
             // Bob
-            x.configure(PDL, b"forward_pair").unwrap();
+            x.configure(PDL, b"exchange").unwrap();
             x.bind_port(0, Native).unwrap(); // native in
             x.bind_port(1, Active(addrs[1])).unwrap(); // peer out
             x.bind_port(2, Active(addrs[0])).unwrap(); // peer in
             x.bind_port(3, Native).unwrap(); // native out
             x.bind_port(1, Native).unwrap(); // native out
             x.bind_port(2, Active(addrs[1])).unwrap(); // peer out
             x.bind_port(3, Active(addrs[0])).unwrap(); // peer in
             x.connect(timeout).unwrap();
             for _ in 0..N {
                 x.put(0, b"B->A".to_vec()).unwrap();
                 x.get(1).unwrap();
                 assert_eq!(Ok(()), x.put(0, b"B->A".to_vec()));
                 assert_eq!(Ok(()), x.get(1));
                 assert_eq!(Ok(0), x.sync(timeout));
-                assert_eq!(Ok(b"A->B" as &[u8]), x.read_gotten(0));
+                assert_eq!(Ok(b"A->B" as &[u8]), x.read_gotten(1));
+            }
         },
     ]));
+}
 #[test]
 // THIS DOES NOT YET WORK. TODOS are hit
 fn filter_messages() {
     // Make a protocol whose behavior depends on the contents of messages
     // Getter prohibits the receipt of messages of the form [0, ...].
     // those messages are silent
     /*
     Sender -->forward-->P|A-->forward_nonzero--> Receiver
     */
     let timeout = Duration::from_millis(1_500);
     let addrs = [next_addr()];
     const N: usize = 1;
     assert!(run_connector_set(&[
         //
         &|x| {
             // Sender
             x.configure(PDL, b"forward").unwrap();
             x.bind_port(0, Native).unwrap();
             x.bind_port(1, Passive(addrs[0])).unwrap();
             x.connect(timeout).unwrap();
             for i in (0..3).cycle().take(N) {
                 let msg = vec![i as u8]; // messages [0], [1], [2], [0], [1] ...
                                          // batches: [{0=>*}, {0=>?}]
                 x.next_batch().unwrap();
                 x.put(0, msg.clone()).unwrap();
                 assert_eq!(0, x.sync(timeout).unwrap());
                 match x.sync(timeout).unwrap() {
 => {
                         // not sent
                         assert_eq!(&msg, &[0u8]);
+                    }
 => {
                         // sent
                         assert_ne!(&msg, &[0u8]);
+                    }
                     _ => unreachable!(),
+                }
+            }
         },
         &|x| {
             // Receiver
             x.configure(PDL, b"forward_nonzero").unwrap();
             x.bind_port(0, Active(addrs[0])).unwrap();
             x.bind_port(1, Native).unwrap();
             x.connect(timeout).unwrap();
             for _ in 0..N {
                 // round _i batches:[0=>*, 0=>?]
                 x.next_batch().unwrap();
                 x.get(0).unwrap();
                 match x.sync(timeout).unwrap() {
 => {
                         // nothing received
                         assert_eq!(Err(ReadGottenErr::DidNotGet), x.read_gotten(0));
+                    }
 => {
                         // msg received
                         assert_ne!(&[0u8], x.read_gotten(0).unwrap());
+                    }
                     _ => unreachable!(),
+                }
+            }
         },
     ]));
+}

0 comments (0 inline, 0 general)