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

Changeset - f11006c1092d

Parent rev.

Child rev.

[Not reviewed]

0 3 0

Christopher Esterhuyse - 5 years ago 2020-09-21 08:58:26
christopher.esterhuyse@gmail.com

sequencer 3 primitive and composite unit tests added

3 files changed with 67 insertions and 105 deletions:

src/runtime/communication.rs

src/runtime/mod.rs

src/runtime/tests.rs

0 comments (0 inline, 0 general)

src/runtime/communication.rs

➞

Show inline comments

@@ @@ -3,96 +3,99 @@ use crate::common::*; @@
 use core::ops::{Deref, DerefMut};
 ////////////////
 // Guard protecting an incrementally unfoldable slice of MapTempGuard elements
 struct MapTempsGuard<'a, K, V>(&'a mut [HashMap<K, V>]);
 // Type protecting a temporary map; At the start and end of the Guard's lifetime, self.0.is_empty() must be true
 struct MapTempGuard<'a, K, V>(&'a mut HashMap<K, V>);
 struct BranchingNative {
     branches: HashMap<Predicate, NativeBranch>,
+}
 #[derive(Clone, Debug)]
 struct NativeBranch {
     index: usize,
     gotten: HashMap<PortId, Payload>,
     to_get: HashSet<PortId>,
+}
 #[derive(Debug)]
 struct BranchingProtoComponent {
     branches: HashMap<Predicate, ProtoComponentBranch>,
+}
 #[derive(Debug, Clone)]
 struct ProtoComponentBranch {
     state: ComponentState,
     inner: ProtoComponentBranchInner,
     ended: bool,
+}
 struct CyclicDrainer<'a, K: Eq + Hash, V> {
     input: &'a mut HashMap<K, V>,
     inner: CyclicDrainInner<'a, K, V>,
+}
 struct CyclicDrainInner<'a, K: Eq + Hash, V> {
     swap: &'a mut HashMap<K, V>,
     output: &'a mut HashMap<K, V>,
+}
 trait ReplaceBoolTrue {
     fn replace_with_true(&mut self) -> bool;
+}
 impl ReplaceBoolTrue for bool {
     fn replace_with_true(&mut self) -> bool {
         let was = *self;
         *self = true;
         !was
+    }
+}
 // CuUndecided provides a mostly immutable view into the ConnectorUnphased structure,
 // making it harder to accidentally mutate its contents in a way that cannot be rolled back.
 impl CuUndecided for ConnectorUnphased {
     fn logger_and_protocol_description(&mut self) -> (&mut dyn Logger, &ProtocolDescription) {
         (&mut *self.inner.logger, &self.proto_description)
+    }
     fn logger(&mut self) -> &mut dyn Logger {
         &mut *self.inner.logger
+    }
     fn proto_description(&self) -> &ProtocolDescription {
         &self.proto_description
+    }
     fn native_component_id(&self) -> ComponentId {
         self.inner.native_component_id
+    }
+}
 ////////////////
 impl<'a, K, V> MapTempsGuard<'a, K, V> {
     fn reborrow(&mut self) -> MapTempsGuard<'_, K, V> {
         MapTempsGuard(self.0)
+    }
     fn split_first_mut(self) -> (MapTempGuard<'a, K, V>, MapTempsGuard<'a, K, V>) {
         let (head, tail) = self.0.split_first_mut().expect("Cache exhausted");
         (MapTempGuard::new(head), MapTempsGuard(tail))
+    }
+}
 impl<'a, K, V> MapTempGuard<'a, K, V> {
     fn new(map: &'a mut HashMap<K, V>) -> Self {
         assert!(map.is_empty()); // sanity check
         Self(map)
+    }
+}
 impl<'a, K, V> Drop for MapTempGuard<'a, K, V> {
     fn drop(&mut self) {
         assert!(self.0.is_empty()); // sanity check
+    }
+}
 impl<'a, K, V> Deref for MapTempGuard<'a, K, V> {
     type Target = HashMap<K, V>;
     fn deref(&self) -> &<Self as Deref>::Target {
         self.0
+    }
+}
 impl<'a, K, V> DerefMut for MapTempGuard<'a, K, V> {
     fn deref_mut(&mut self) -> &mut <Self as Deref>::Target {
         self.0
+    }
+}
 impl Connector {
     fn get_comm_mut(&mut self) -> Option<&mut ConnectorCommunication> {
         if let ConnectorPhased::Communication(comm) = &mut self.phased {
             Some(comm)
         } else {
@@ @@ -172,132 +175,134 @@ impl Connector { @@
                 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;
         //////////////////////////////////
         log!(@MARK, cu.logger(), "sync start {}", comm.round_index);
         log!(
             cu.logger(),
             "~~~ SYNC called with timeout {:?}; starting round {}",
             &timeout,
             comm.round_index
         );
         log!(@BENCH, cu.logger(), "");
         // 1. run all proto components to Nonsync blockers
         // iterate
         let mut current_state = cu.inner.current_state.clone();
         let mut branching_proto_components =
             HashMap::<ComponentId, BranchingProtoComponent>::default();
         let mut unrun_components: Vec<(ComponentId, ComponentState)> = cu
             .proto_components
             .iter()
             .map(|(&proto_id, proto)| (proto_id, proto.clone()))
             .collect();
         log!(cu.logger(), "Nonsync running {} proto components...", unrun_components.len());
         // drains unrun_components, and populates branching_proto_components.
         while let Some((proto_component_id, mut component)) = unrun_components.pop() {
             log!(
                 cu.logger(),
                 "Nonsync running proto component with ID {:?}. {} to go after this",
                 proto_component_id,
                 unrun_components.len()
             );
             let (logger, proto_description) = cu.logger_and_protocol_description();
             let mut ctx = NonsyncProtoContext {
                 current_state: &mut current_state,
-                logger: &mut *cu.inner.logger,
                 logger,
                 proto_component_id,
                 unrun_components: &mut unrun_components,
             };
-            let blocker = component.nonsync_run(&mut ctx, &cu.proto_description);
             let blocker = component.nonsync_run(&mut ctx, 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(),
         );
         log!(@BENCH, cu.logger(), "");
         // Create temp structures needed for the synchronous phase of the round
         let mut rctx = RoundCtx {
             current_state,
             solution_storage: {
                 let n = std::iter::once(SubtreeId::LocalComponent(cu.inner.native_component_id));
                 let c = cu.proto_components.keys().map(|&cid| SubtreeId::LocalComponent(cid));
                 let c =
                     branching_proto_components.keys().map(|&cid| SubtreeId::LocalComponent(cid));
                 let e = comm
                     .neighborhood
                     .children
                     .iter()
                     .map(|&index| SubtreeId::NetEndpoint { index });
                 let subtree_id_iter = n.chain(c).chain(e);
                 log!(
                     cu.inner.logger,
                     "Children in subtree are: {:?}",
                     subtree_id_iter.clone().collect::<Vec<_>>()
                 );
                 SolutionStorage::new(subtree_id_iter)
             },
             spec_var_stream: cu.inner.current_state.id_manager.new_spec_var_stream(),
             payload_inbox: Default::default(),
             deadline: timeout.map(|to| Instant::now() + to),
         };
         log!(cu.logger(), "Round context structure initialized");
         log!(@BENCH, cu.logger(), "");
         // Explore all native branches eagerly. Find solutions, buffer messages, etc.
         log!(
             cu.logger(),
             "Translating {} native batches into branches...",
             comm.native_batches.len()
         );
         let native_spec_var = rctx.spec_var_stream.next();
         log!(cu.logger(), "Native branch spec var is {:?}", native_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();
                 // all firing ports have SpecVal::FIRING
                 let firing_iter = to_get.iter().chain(to_put.keys()).copied();
                 log!(
                     cu.logger(),
                     "New native with firing ports {:?}",
                     firing_iter.clone().collect::<Vec<_>>()
                 );
                 let firing_ports: HashSet<PortId> = firing_iter.clone().collect();
                 for port in firing_iter {
                     let var = cu.inner.current_state.spec_var_for(port);
                     predicate.assigned.insert(var, SpecVal::FIRING);
+                }
@@ @@ -1125,169 +1130,125 @@ impl SolutionStorage { @@
                         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!(cu.logger(), "storing NEW LOCAL SOLUTION {:?}", &partial);
                 new_local.insert(partial);
+            }
+        }
+    }
+}
 impl SyncProtoContext<'_> {
     pub(crate) fn is_firing(&mut self, port: PortId) -> Option<bool> {
         let var = self.rctx.current_state.spec_var_for(port);
         self.predicate.query(var).map(SpecVal::is_firing)
+    }
     pub(crate) fn read_msg(&mut self, port: PortId) -> Option<&Payload> {
         self.branch_inner.did_put_or_get.insert(port);
         self.branch_inner.inbox.get(&port)
+    }
     pub(crate) fn take_choice(&mut self) -> Option<u16> {
         self.branch_inner.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
         );
         println!("MOVED PORTS {:#?}", &moved_ports);
         // sanity check
         for port in moved_ports.iter() {
             assert_eq!(
                 self.proto_component_id,
                 self.current_state.port_info.get(port).unwrap().owner
             );
+        }
         // 2. create new component
         let new_cid = self.current_state.id_manager.new_component_id();
         self.unrun_components.push((new_cid, state));
         // 3. update ownership of moved ports
         for port in moved_ports.iter() {
             self.current_state.port_info.get_mut(port).unwrap().owner = new_cid;
+        }
         // 3. create a new component
+    }
     pub fn new_port_pair(&mut self) -> [PortId; 2] {
         // adds two new associated ports, related to each other, and exposed to the proto component
         let mut new_cid_fn = || self.current_state.id_manager.new_port_id();
         let [o, i] = [new_cid_fn(), new_cid_fn()];
         self.current_state.port_info.insert(
             o,
             PortInfo {
                 route: Route::LocalComponent,
                 peer: Some(i),
                 polarity: Putter,
                 owner: self.proto_component_id,
             },
         );
         self.current_state.port_info.insert(
             i,
             PortInfo {
                 route: Route::LocalComponent,
                 peer: Some(o),
                 polarity: Getter,
                 owner: self.proto_component_id,
             },
         );
         log!(
             self.logger,
             "Component {:?} port pair (out->in) {:?} -> {:?}",
             self.proto_component_id,
             o,
+            i
         );
         [o, i]
+    }
+}
 impl ProtoComponentBranch {
     fn feed_msg(&mut self, getter: PortId, payload: Payload) {
         let was = self.inner.inbox.insert(getter, payload);
         assert!(was.is_none())
+    }
+}
 impl<'a, K: Eq + Hash + 'static, V: 'static> CyclicDrainer<'a, K, V> {
     fn new(
         input: &'a mut HashMap<K, V>,
         swap: &'a mut HashMap<K, V>,
         output: &'a mut HashMap<K, V>,
     ) -> Self {
         Self { input, inner: CyclicDrainInner { swap, output } }
+    }
     fn cyclic_drain<E>(
         self,
         mut func: impl FnMut(K, V, CyclicDrainInner<'_, K, V>) -> Result<(), E>,
     ) -> Result<(), E> {
         let Self { input, inner: CyclicDrainInner { swap, output } } = self;
         // assert!(swap.is_empty());
         while !input.is_empty() {
             for (k, v) in input.drain() {
                 func(k, v, CyclicDrainInner { swap, output })?
+            }
             std::mem::swap(input, swap);
+        }
         Ok(())
+    }
+}
 // struct ConnectorComm {
 //     logger: Box<dyn Logger>,
 //     pd: Arc<ProtocolDescription>,
 //     current_state: CurrentState,
 //     components: HashMap<ComponentId, ComponentState>,
 //     ports: HashMap<PortId, PortInfo>,
 //     endpoint_manager: EndpointManager,
 //     neighborhood: Neighborhood,
 //     native_batches: Vec<NativeBatch>,
 //     round_result: Result<Option<RoundOk>, SyncError>,
 // }
 // struct RoundTemp<'a> {
 //     deadline: Option<Instant>,
 //     msg_buf: Vec<(PortId, SendPayloadMsg)>,
 //     solution_storage: SolutionStorage,
 //     override_ports: HashMap<PortId, PortInfo>,
 //     spec_var_stream: SpecVarStream,
 //     branching_proto: HashMap<ComponentId, BranchingProtoComponent>,
 //     branching_native: BranchingNative,
 //     comm: &'a mut ConnectorComm,
 // }
 // impl ConnectorComm {
 //     fn sync(&mut self, deadline: Option<Duration>) -> Result<usize, ()> {
 //         RoundTemp {
 //             msg_buf: Default::default(),
 //             deadline: todo!(),
 //             solution_storage: todo!(),
 //             override_ports: Default::default(),
 //             spec_var_stream: todo!(),
 //             branching_proto: Default::default(),
 //             branching_native: todo!(),
 //             comm: self,
 //         }
 //         .sync()
 //     }
 // }
 // impl RoundTemp<'_> {
 //     fn sync(&mut self) -> Result<usize, ()> {
 //         todo!()
 //     }
 // }

src/runtime/mod.rs

➞

Show inline comments

@@ @@ -225,96 +225,97 @@ struct ConnectorCommunication { @@
     native_batches: Vec<NativeBatch>,
     round_result: Result<Option<RoundOk>, SyncError>,
+}
 #[derive(Debug)]
 struct ConnectorUnphased {
     proto_description: Arc<ProtocolDescription>,
     proto_components: HashMap<ComponentId, ComponentState>,
     inner: ConnectorUnphasedInner,
+}
 #[derive(Debug)]
 struct ConnectorUnphasedInner {
     logger: Box<dyn Logger>,
     current_state: CurrentState,
     native_component_id: ComponentId,
+}
 #[derive(Debug)]
 struct ConnectorSetup {
     net_endpoint_setups: Vec<NetEndpointSetup>,
     udp_endpoint_setups: Vec<UdpEndpointSetup>,
+}
 #[derive(Debug)]
 enum ConnectorPhased {
     Setup(Box<ConnectorSetup>),
     Communication(Box<ConnectorCommunication>),
+}
 #[derive(Default, Clone, Eq, PartialEq, Hash, serde::Serialize, serde::Deserialize)]
 struct Predicate {
     assigned: BTreeMap<SpecVar, SpecVal>,
+}
 #[derive(Debug)]
 struct SolutionStorage {
     old_local: HashSet<Predicate>,
     new_local: HashSet<Predicate>,
     // this pair acts as SubtreeId -> HashSet<Predicate> which is friendlier to iteration
     subtree_solutions: Vec<HashSet<Predicate>>,
     subtree_id_to_index: HashMap<SubtreeId, usize>,
+}
 struct RoundCtx {
     solution_storage: SolutionStorage,
     spec_var_stream: SpecVarStream,
     payload_inbox: Vec<(PortId, SendPayloadMsg)>,
     deadline: Option<Instant>,
     current_state: CurrentState,
+}
 trait CuUndecided {
     fn logger(&mut self) -> &mut dyn Logger;
     fn proto_description(&self) -> &ProtocolDescription;
     fn native_component_id(&self) -> ComponentId;
     fn logger_and_protocol_description(&mut self) -> (&mut dyn Logger, &ProtocolDescription);
+}
 #[derive(Debug, Default)]
 struct NativeBatch {
     // invariant: putters' and getters' polarities respected
     to_put: HashMap<PortId, Payload>,
     to_get: HashSet<PortId>,
+}
 #[derive(Debug, Copy, Clone, Eq, PartialEq, Hash)]
 enum TokenTarget {
     NetEndpoint { index: usize },
     UdpEndpoint { index: usize },
     Waker,
+}
 enum CommRecvOk {
     TimeoutWithoutNew,
     NewPayloadMsgs,
     NewControlMsg { net_index: usize, msg: CommCtrlMsg },
+}
 ////////////////
 fn would_block(err: &std::io::Error) -> bool {
     err.kind() == std::io::ErrorKind::WouldBlock
+}
 impl TokenTarget {
     const HALFWAY_INDEX: usize = usize::MAX / 2;
     const MAX_INDEX: usize = usize::MAX;
     const WAKER_TOKEN: usize = Self::MAX_INDEX;
+}
 impl From<Token> for TokenTarget {
     fn from(Token(index): Token) -> Self {
         if index == Self::WAKER_TOKEN {
             TokenTarget::Waker
         } else if let Some(shifted) = index.checked_sub(Self::HALFWAY_INDEX) {
             TokenTarget::UdpEndpoint { index: shifted }
         } else {
             TokenTarget::NetEndpoint { index }
+        }
+    }
+}
 impl Into<Token> for TokenTarget {
     fn into(self) -> Token {
         match self {
             TokenTarget::Waker => Token(Self::WAKER_TOKEN),
             TokenTarget::UdpEndpoint { index } => Token(index + Self::HALFWAY_INDEX),
             TokenTarget::NetEndpoint { index } => Token(index),
+        }
+    }
+}
 impl<T: std::cmp::Ord> VecSet<T> {

src/runtime/tests.rs

➞

Show inline comments

@@ @@ -1014,110 +1014,110 @@ fn pdl_msg_consensus() { @@
     c.put(p1, Payload::from(b"HELLO" as &[_])).unwrap();
     c.sync(SEC1).unwrap_err();
+}
 #[test]
 fn sequencer3_prim() {
     let test_log_path = Path::new("./logs/sequencer3_prim");
     let pdl = b"
     primitive seq3primitive(out a, out b, out c) {
         int i = 0;
         while(true) synchronous {
             out to = a;
             if     (i==1) to = b;
             else if(i==2) to = c;
             if(fires(to)) {
                 put(to, create(0));
                 i = (i + 1)%3;
+            }
+        }
+    }
     ";
     let pd = reowolf::ProtocolDescription::parse(pdl).unwrap();
     let mut c = file_logged_configured_connector(0, test_log_path, Arc::new(pd));
     // setup a session between (a) native, and (b) primitive sequencer3, connected by 3 ports.
     let [p0, g0] = c.new_port_pair();
     let [p1, g1] = c.new_port_pair();
     let [p2, g2] = c.new_port_pair();
     c.add_component(b"seq3primitive", &[p0, p1, p2]).unwrap();
     c.connect(None).unwrap();
     let mut which_of_three = move || {
         // setup three sync batches. sync. return which succeeded
         c.get(g0).unwrap();
         c.next_batch().unwrap();
         c.get(g1).unwrap();
         c.next_batch().unwrap();
         c.get(g2).unwrap();
         c.sync(None).unwrap()
     };
     const TEST_ROUNDS: usize = 50;
     // check that the batch index for rounds 0..TEST_ROUNDS are [0, 1, 2, 0, 1, 2, ...]
     for expected_batch_idx in (0..=2).cycle().take(TEST_ROUNDS) {
         assert_eq!(expected_batch_idx, which_of_three());
+    }
+}
 // #[test]
 // fn sequencer3_comp() {
 //     let test_log_path = Path::new("./logs/sequencer3_comp");
 //     let pdl = b"
 //     primitive fifo1_init(msg m, in a, out b) {
 //         while(true) synchronous {
 //             if(m != null && fires(b)) {
 //                 put(b, m);
 //                 m = null;
 //             } else if (m == null && fires(a)) {
 //                 m = get(a);
 //             }
 //         }
 //     }
 //     composite fifo1_full(in a, out b) {
 //         new fifo1_init(create(0), a, b);
 //     }
 //     composite fifo1(in a, out b) {
 //         new fifo1_init(null, a, b);
 //     }
 //     composite seq3composite(out a, out b, out c) {
 //         channel d -> e;
 //         channel f -> g;
 //         channel h -> i;
 //         channel j -> k;
 //         channel l -> m;
 //         channel n -> o;
 #[test]
 fn sequencer3_comp() {
     let test_log_path = Path::new("./logs/sequencer3_comp");
     let pdl = b"
     primitive fifo1_init(msg m, in a, out b) {
         while(true) synchronous {
             if(m != null && fires(b)) {
                 put(b, m);
                 m = null;
             } else if (m == null && fires(a)) {
                 m = get(a);
+            }
+        }
+    }
     composite fifo1_full(in a, out b) {
         new fifo1_init(create(0), a, b);
+    }
     composite fifo1(in a, out b) {
         new fifo1_init(null, a, b);
+    }
     composite seq3composite(out a, out b, out c) {
         channel d -> e;
         channel f -> g;
         channel h -> i;
         channel j -> k;
         channel l -> m;
         channel n -> o;
 //         new fifo1_full(o, d);
 //         new replicator2(e, f, a);
 //         new fifo1(g, h);
 //         new replicator2(i, j, b);
 //         new fifo1(k, l);
 //         new replicator2(m, n, c);
 //     }
 //     ";
 //     let pd = reowolf::ProtocolDescription::parse(pdl).unwrap();
 //     let mut c = file_logged_configured_connector(0, test_log_path, Arc::new(pd));
         new fifo1_full(o, d);
         new replicator2(e, f, a);
         new fifo1(g, h);
         new replicator2(i, j, b);
         new fifo1(k, l);
         new replicator2(m, n, c);
+    }
     ";
     let pd = reowolf::ProtocolDescription::parse(pdl).unwrap();
     let mut c = file_logged_configured_connector(0, test_log_path, Arc::new(pd));
 //     // setup a session between (a) native, and (b) composite sequencer3, connected by 3 ports.
 //     let [p0, g0] = c.new_port_pair();
 //     let [p1, g1] = c.new_port_pair();
 //     let [p2, g2] = c.new_port_pair();
 //     c.add_component(b"seq3composite", &[p0, p1, p2]).unwrap();
 //     c.connect(None).unwrap();
     // setup a session between (a) native, and (b) composite sequencer3, connected by 3 ports.
     let [p0, g0] = c.new_port_pair();
     let [p1, g1] = c.new_port_pair();
     let [p2, g2] = c.new_port_pair();
     c.add_component(b"seq3composite", &[p0, p1, p2]).unwrap();
     c.connect(None).unwrap();
 //     let mut which_of_three = move || {
 //         // setup three sync batches. sync. return which succeeded
 //         c.get(g0).unwrap();
 //         c.next_batch().unwrap();
 //         c.get(g1).unwrap();
 //         c.next_batch().unwrap();
 //         c.get(g2).unwrap();
 //         c.sync(None).unwrap()
 //     };
     let mut which_of_three = move || {
         // setup three sync batches. sync. return which succeeded
         c.get(g0).unwrap();
         c.next_batch().unwrap();
         c.get(g1).unwrap();
         c.next_batch().unwrap();
         c.get(g2).unwrap();
         c.sync(None).unwrap()
     };
 //     const TEST_ROUNDS: usize = 50;
 //     // check that the batch index for rounds 0..TEST_ROUNDS are [0, 1, 2, 0, 1, 2, ...]
 //     for expected_batch_idx in (0..=2).cycle().take(TEST_ROUNDS) {
 //         assert_eq!(expected_batch_idx, which_of_three());
 //     }
 // }
     const TEST_ROUNDS: usize = 50;
     // check that the batch index for rounds 0..TEST_ROUNDS are [0, 1, 2, 0, 1, 2, ...]
     for expected_batch_idx in (0..=2).cycle().take(TEST_ROUNDS) {
         assert_eq!(expected_batch_idx, which_of_three());
+    }
+}

0 comments (0 inline, 0 general)