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

 use super::*;
 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 {
             None
+        }
+    }
     pub fn gotten(&mut self, port: PortId) -> Result<&Payload, GottenError> {
         use GottenError as Ge;
         let comm = self.get_comm_mut().ok_or(Ge::NoPreviousRound)?;
         match &comm.round_result {
             Err(_) => Err(Ge::PreviousSyncFailed),
             Ok(None) => Err(Ge::NoPreviousRound),
             Ok(Some(round_ok)) => round_ok.gotten.get(&port).ok_or(Ge::PortDidntGet),
+        }
+    }
     pub fn next_batch(&mut self) -> Result<usize, WrongStateError> {
         // returns index of new batch
         let comm = self.get_comm_mut().ok_or(WrongStateError)?;
         comm.native_batches.push(Default::default());
         Ok(comm.native_batches.len() - 1)
+    }
     fn port_op_access(
         &mut self,
         port: PortId,
         expect_polarity: Polarity,
     ) -> Result<&mut NativeBatch, PortOpError> {
         use PortOpError as Poe;
         let Self { unphased: cu, phased } = self;
         let info = cu.inner.current_state.port_info.get(&port).ok_or(Poe::UnknownPolarity)?;
         if info.owner != cu.inner.native_component_id {
             return Err(Poe::PortUnavailable);
+        }
         if info.polarity != expect_polarity {
             return Err(Poe::WrongPolarity);
+        }
         match phased {
             ConnectorPhased::Setup { .. } => Err(Poe::NotConnected),
             ConnectorPhased::Communication(comm) => {
                 let batch = comm.native_batches.last_mut().unwrap(); // length >= 1 is invariant
                 Ok(batch)
+            }
+        }
+    }
     pub fn put(&mut self, port: PortId, payload: Payload) -> Result<(), PortOpError> {
         use PortOpError as Poe;
         let batch = self.port_op_access(port, Putter)?;
         if batch.to_put.contains_key(&port) {
             Err(Poe::MultipleOpsOnPort)
         } else {
             batch.to_put.insert(port, payload);
             Ok(())
+        }
+    }
     pub fn get(&mut self, port: PortId) -> Result<(), PortOpError> {
         use PortOpError as Poe;
         let batch = self.port_op_access(port, Getter)?;
         if batch.to_get.insert(port) {
             Ok(())
         } else {
             Err(Poe::MultipleOpsOnPort)
+        }
+    }
     // entrypoint for caller. overwrites round result enum, and returns what happened
     pub fn sync(&mut self, timeout: Option<Duration>) -> Result<usize, SyncError> {
         let Self { unphased: cu, phased } = self;
         match phased {
             ConnectorPhased::Setup { .. } => Err(SyncError::NotConnected),
             ConnectorPhased::Communication(comm) => {
                 match &comm.round_result {
                     Err(SyncError::Unrecoverable(e)) => {
                         log!(cu.logger(), "Attempted to start sync round, but previous error {:?} was unrecoverable!", e);
                         return Err(SyncError::Unrecoverable(e.clone()));
+                    }
                     _ => {}
+                }
                 comm.round_result = Self::connected_sync(cu, comm, timeout);
                 comm.round_index += 1;
                 match &comm.round_result {
                     Ok(None) => unreachable!(),
                     Ok(Some(ok_result)) => Ok(ok_result.batch_index),
                     Err(sync_error) => Err(sync_error.clone()),
+                }
+            }
+        }
+    }
     // private function. mutates state but returns with round
     // result ASAP (allows for convenient error return with ?)
     fn connected_sync(
         cu: &mut ConnectorUnphased,
         comm: &mut ConnectorCommunication,
         timeout: Option<Duration>,
     ) -> Result<Option<RoundOk>, SyncError> {
         //////////////////////////////////
         use SyncError as Se;
         //////////////////////////////////
         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);
+                }
                 // all silent ports have SpecVal::SILENT
                 for (port, port_info) in cu.inner.current_state.port_info.iter() {
                     if port_info.owner != cu.inner.native_component_id {
                         // not my port
                         continue;
+                    }
                     if firing_ports.contains(port) {
                         // this one is FIRING
                         continue;
+                    }
                     let var = cu.inner.current_state.spec_var_for(*port);
                     if let Some(SpecVal::FIRING) = predicate.assigned.insert(var, SpecVal::SILENT) {
                         log!(cu.logger(), "Native branch index={} contains internal inconsistency wrt. {:?}. Skipping", index, var);
                         continue 'native_branches;
+                    }
+                }
                 // this branch is consistent. distinguish it with a unique var:val mapping and proceed
                 predicate.inserted(native_spec_var, branch_spec_val)
             };
             log!(cu.logger(), "Native branch index={:?} has consistent {:?}", index, &predicate);
             // send all outgoing messages (by buffering them)
             for (putter, payload) in to_put {
                 let msg = SendPayloadMsg { predicate: predicate.clone(), payload };
                 log!(
                     cu.logger(),
                     "Native branch {} sending msg {:?} with putter {:?}",
                     index,
                     &msg,
                     putter
                 );
                 // sanity check
                 assert_eq!(Putter, cu.inner.current_state.port_info.get(&putter).unwrap().polarity);
                 rctx.putter_push(cu, putter, msg);
+            }
             let branch = NativeBranch { index, gotten: Default::default(), to_get };
             if branch.is_ended() {
                 log!(
                     cu.logger(),
                     "Native submitting solution for batch {} with {:?}",
                     index,
                     &predicate
                 );
                 rctx.solution_storage.submit_and_digest_subtree_solution(
                     cu,
                     SubtreeId::LocalComponent(cu.inner.native_component_id),
                     predicate.clone(),
                 );
+            }
@@ @@ -1077,217 +1082,173 @@ impl SolutionStorage { @@
     //     self.subtree_id_to_index.clear();
     //     self.subtree_solutions.clear();
     //     self.old_local.clear();
     //     self.new_local.clear();
     //     for key in subtree_ids {
     //         self.subtree_id_to_index.insert(key, self.subtree_solutions.len());
     //         self.subtree_solutions.push(Default::default())
     //     }
     // }
     pub(crate) fn iter_new_local_make_old(&mut self) -> impl Iterator<Item = Predicate> + '_ {
         let Self { old_local, new_local, .. } = self;
         new_local.drain().map(move |local| {
             old_local.insert(local.clone());
             local
         })
+    }
     pub(crate) fn submit_and_digest_subtree_solution(
         &mut self,
         cu: &mut impl CuUndecided,
         subtree_id: SubtreeId,
         predicate: Predicate,
     ) {
         log!(cu.logger(), "++ new component solution {:?} {:?}", subtree_id, &predicate);
         let index = self.subtree_id_to_index[&subtree_id];
         let left = 0..index;
         let right = (index + 1)..self.subtree_solutions.len();
         let Self { subtree_solutions, new_local, old_local, .. } = self;
         let was_new = subtree_solutions[index].insert(predicate.clone());
         if was_new {
             let set_visitor = left.chain(right).map(|index| &subtree_solutions[index]);
             Self::elaborate_into_new_local_rec(cu, predicate, set_visitor, old_local, new_local);
+        }
+    }
     fn elaborate_into_new_local_rec<'a, 'b>(
         cu: &mut impl CuUndecided,
         partial: Predicate,
         mut set_visitor: impl Iterator<Item = &'b HashSet<Predicate>> + Clone,
         old_local: &'b HashSet<Predicate>,
         new_local: &'a mut HashSet<Predicate>,
     ) {
         if let Some(set) = set_visitor.next() {
             // incomplete solution. keep traversing
             for pred in set.iter() {
                 if let Some(elaborated) = pred.union_with(&partial) {
                     Self::elaborate_into_new_local_rec(
                         cu,
                         elaborated,
                         set_visitor.clone(),
                         old_local,
                         new_local,
+                    )
+                }
+            }
         } else {
             // recursive stop condition. `partial` is a local subtree solution
             if !old_local.contains(&partial) {
                 // ... and it hasn't been found before
                 log!(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

@@ @@ -177,192 +177,193 @@ struct Neighborhood { @@
     parent: Option<usize>,
     children: VecSet<usize>,
+}
 #[derive(Debug, Clone)]
 struct IdManager {
     connector_id: ConnectorId,
     port_suffix_stream: U32Stream,
     component_suffix_stream: U32Stream,
+}
 struct UdpInBuffer {
     byte_vec: Vec<u8>,
+}
 #[derive(Debug)]
 struct SpecVarStream {
     connector_id: ConnectorId,
     port_suffix_stream: U32Stream,
+}
 #[derive(Debug)]
 struct EndpointManager {
     // invariants:
     // 1. net and udp endpoints are registered with poll. Poll token computed with TargetToken::into
     // 2. Events is empty
     poll: Poll,
     events: Events,
     delayed_messages: Vec<(usize, Msg)>,
     undelayed_messages: Vec<(usize, Msg)>,
     net_endpoint_store: EndpointStore<NetEndpointExt>,
     udp_endpoint_store: EndpointStore<UdpEndpointExt>,
     udp_in_buffer: UdpInBuffer,
+}
 #[derive(Debug)]
 struct EndpointStore<T> {
     endpoint_exts: Vec<T>,
     polled_undrained: VecSet<usize>,
+}
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 struct PortInfo {
     owner: ComponentId,
     peer: Option<PortId>,
     polarity: Polarity,
     route: Route,
+}
 #[derive(Debug)]
 struct ConnectorCommunication {
     round_index: usize,
     endpoint_manager: EndpointManager,
     neighborhood: Neighborhood,
     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> {
     fn new(mut vec: Vec<T>) -> Self {
         vec.sort();
         vec.dedup();
         Self { vec }
+    }
     fn contains(&self, element: &T) -> bool {
         self.vec.binary_search(element).is_ok()
+    }
     fn insert(&mut self, element: T) -> bool {
         match self.vec.binary_search(&element) {
             Ok(_) => false,
             Err(index) => {
                 self.vec.insert(index, element);
                 true
+            }
+        }
+    }
     fn iter(&self) -> std::slice::Iter<T> {
         self.vec.iter()
+    }
     fn pop(&mut self) -> Option<T> {
         self.vec.pop()
+    }
+}
 impl CurrentState {
     fn spec_var_for(&self, port: PortId) -> SpecVar {
         let info = self.port_info.get(&port).unwrap();
         SpecVar(match info.polarity {
             Getter => port,
             Putter => info.peer.unwrap(),
         })
+    }
+}
 impl SpecVarStream {
     fn next(&mut self) -> SpecVar {
         let phantom_port: PortId =
             Id { connector_id: self.connector_id, u32_suffix: self.port_suffix_stream.next() }
                 .into();
         SpecVar(phantom_port)
+    }
+}
 impl IdManager {
     fn new(connector_id: ConnectorId) -> Self {
         Self {
             connector_id,
             port_suffix_stream: Default::default(),
             component_suffix_stream: Default::default(),
+        }

src/runtime/tests.rs

➞

Show inline comments

@@ @@ -966,158 +966,158 @@ fn pdl_reo_fifo1full() { @@
     let test_log_path = Path::new("./logs/pdl_reo_fifo1full");
     let pdl = b"
     primitive fifo1full(in a, out b) {
         msg m = create(0);
         while(true) synchronous {
             if(m == null) {
                 if(fires(a)) m=get(a);
             } else {
                 if(fires(b)) put(b, m);
                 m = null;
+            }
+        }
+    }
     ";
     let pd = reowolf::ProtocolDescription::parse(pdl).unwrap();
     let mut c = file_logged_configured_connector(0, test_log_path, Arc::new(pd));
     let [_p0, g0] = c.new_port_pair();
     let [p1, g1] = c.new_port_pair();
     c.add_component(b"fifo1full", &[g0, p1]).unwrap();
     c.connect(None).unwrap();
     c.get(g1).unwrap();
     c.sync(None).unwrap();
     assert_eq!(0, c.gotten(g1).unwrap().len());
+}
 #[test]
 fn pdl_msg_consensus() {
     let test_log_path = Path::new("./logs/pdl_msg_consensus");
     let pdl = b"
     primitive msgconsensus(in a, in b) {
         while(true) synchronous {
             msg x = get(a);
             msg y = get(b);
             assert(x == y);
+        }
+    }
     ";
     let pd = reowolf::ProtocolDescription::parse(pdl).unwrap();
     let mut c = file_logged_configured_connector(0, test_log_path, Arc::new(pd));
     let [p0, g0] = c.new_port_pair();
     let [p1, g1] = c.new_port_pair();
     c.add_component(b"msgconsensus", &[g0, g1]).unwrap();
     c.connect(None).unwrap();
     c.put(p0, Payload::from(b"HELLO" as &[_])).unwrap();
     c.put(p1, Payload::from(b"HELLO" as &[_])).unwrap();
     c.sync(SEC1).unwrap();
     c.put(p0, Payload::from(b"HEY" as &[_])).unwrap();
     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)