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

@@ @@ -336,66 +336,75 @@ impl PolyN { @@
                 "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() {
@@ @@ -79,97 +78,97 @@ impl Controller { @@
+        }
+    }
     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);

src/runtime/mod.rs

➞

Show inline comments

@@ @@ -341,123 +341,123 @@ impl Predicate { @@
         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' }

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();
         },
     ]));
+}
@@ @@ -473,223 +478,219 @@ fn alternator_2() { @@
             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();

0 comments (0 inline, 0 general)