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

Changeset - e7b7d53e6952

Parent rev.

Child rev.

[Not reviewed]

1 4 2

Christopher Esterhuyse - 5 years ago 2020-06-23 17:52:58
christopher.esterhuyse@gmail.com

more tests

6 files changed with 345 insertions and 308 deletions:

src/common.rs

src/runtime/communication.rs

108

100

src/runtime/endpoints.rs

115

src/runtime/mod.rs

187

src/runtime/setup.rs

src/runtime/tests.rs

0 comments (0 inline, 0 general)

src/common.rs

➞

Show inline comments

@@ @@ -43,24 +43,28 @@ pub struct Id { @@
 pub struct U32Stream {
     next: u32,
+}
 // globally unique
 #[derive(
     Copy, Clone, Eq, PartialEq, Ord, Hash, PartialOrd, serde::Serialize, serde::Deserialize,
 )]
 pub struct PortId(Id);
 #[derive(
     Copy, Clone, Eq, PartialEq, Ord, Hash, PartialOrd, serde::Serialize, serde::Deserialize,
 )]
 pub struct FiringVar(pub(crate) PortId);
 #[derive(
     Copy, Clone, Eq, PartialEq, Ord, Hash, PartialOrd, serde::Serialize, serde::Deserialize,
 )]
 pub struct ProtoComponentId(Id);
 #[derive(Debug, Clone, Eq, PartialEq, Ord, PartialOrd)]
 pub struct Payload(Arc<Vec<u8>>);
 #[derive(
     Debug, Eq, PartialEq, Clone, Hash, Copy, Ord, PartialOrd, serde::Serialize, serde::Deserialize,
 )]
 pub enum Polarity {
     Putter, // output port (from the perspective of the component)
     Getter, // input port (from the perspective of the component)
+}
@@ @@ -166,25 +170,30 @@ impl<'de> serde::Deserialize<'de> for Payload { @@
 impl std::iter::FromIterator<u8> for Payload {
     fn from_iter<I: IntoIterator<Item = u8>>(it: I) -> Self {
         Self(Arc::new(it.into_iter().collect()))
+    }
+}
 impl From<Vec<u8>> for Payload {
     fn from(s: Vec<u8>) -> Self {
         Self(s.into())
+    }
+}
 impl Debug for PortId {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         write!(f, "PortId({},{})", self.0.controller_id, self.0.u32_suffix)
         write!(f, "PID<{},{}>", self.0.controller_id, self.0.u32_suffix)
+    }
+}
 impl Debug for FiringVar {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         write!(f, "VID<{},{}>", (self.0).0.controller_id, (self.0).0.u32_suffix)
+    }
+}
 impl Debug for ProtoComponentId {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         write!(f, "ProtoComponentId({},{})", self.0.controller_id, self.0.u32_suffix)
+    }
+}
 impl std::ops::Not for Polarity {
     type Output = Self;
     fn not(self) -> Self::Output {
         use Polarity::*;
         match self {

src/runtime/communication.rs

➞

Show inline comments

@@ @@ -30,80 +30,24 @@ struct ProtoComponentBranch { @@
     state: ComponentState,
+}
 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>,
+}
 ////////////////
 impl NonsyncProtoContext<'_> {
     pub fn new_component(&mut self, moved_ports: HashSet<PortId>, state: ComponentState) {
         // called by a PROTO COMPONENT. moves its own ports.
         // 1. sanity check: this component owns these ports
         log!(
             self.logger,
             "Component {:?} added new component with state {:?}, moving ports {:?}",
             self.proto_component_id,
             &state,
             &moved_ports
         );
         assert!(self.proto_component_ports.is_subset(&moved_ports));
         // 2. remove ports from old component & update port->route
         let new_id = self.id_manager.new_proto_component_id();
         for port in moved_ports.iter() {
             self.proto_component_ports.remove(port);
             self.port_info
                 .routes
                 .insert(*port, Route::LocalComponent(LocalComponentId::Proto(new_id)));
+        }
         // 3. create a new component
         self.unrun_components.push((new_id, ProtoComponent { state, ports: moved_ports }));
+    }
     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 [o, i] = [self.id_manager.new_port_id(), self.id_manager.new_port_id()];
         self.proto_component_ports.insert(o);
         self.proto_component_ports.insert(i);
         // {polarity, peer, route} known. {} unknown.
         self.port_info.polarities.insert(o, Putter);
         self.port_info.polarities.insert(i, Getter);
         self.port_info.peers.insert(o, i);
         self.port_info.peers.insert(i, o);
         let route = Route::LocalComponent(LocalComponentId::Proto(self.proto_component_id));
         self.port_info.routes.insert(o, route);
         self.port_info.routes.insert(i, route);
         log!(
             self.logger,
             "Component {:?} port pair (out->in) {:?} -> {:?}",
             self.proto_component_id,
             o,
+            i
         );
         [o, i]
+    }
+}
 impl SyncProtoContext<'_> {
     pub fn is_firing(&mut self, port: PortId) -> Option<bool> {
         let var = self.port_info.firing_var_for(port);
         self.predicate.query(var)
+    }
     pub fn read_msg(&mut self, port: PortId) -> Option<&Payload> {
         self.inbox.get(&port)
+    }
+}
 impl Connector {
     pub fn gotten(&mut self, port: PortId) -> Result<&Payload, GottenError> {
         use GottenError::*;
         match &mut self.phased {
             ConnectorPhased::Setup { .. } => Err(NoPreviousRound),
             ConnectorPhased::Communication { round_result, .. } => match round_result {
                 Err(_) => Err(PreviousSyncFailed),
                 Ok(None) => Err(NoPreviousRound),
                 Ok(Some((_index, gotten))) => gotten.get(&port).ok_or(PortDidntGet),
             },
+        }
+    }
@@ @@ -625,84 +569,49 @@ impl BranchingNative { @@
                         logger,
                         "new subsuming pred created {:?}. forking and feeding",
                         &predicate2
                     );
                     finished.insert(predicate, branch);
                     finished.insert(predicate2, branch2);
+                }
+            }
+        }
+    }
     fn collapse_with(self, solution_predicate: &Predicate) -> (usize, HashMap<PortId, Payload>) {
         for (branch_predicate, branch) in self.branches {
-            if branch_predicate.satisfies(solution_predicate) {
+            if solution_predicate.satisfies(&branch_predicate) {
                 let NativeBranch { index, gotten, .. } = branch;
                 return (index, gotten);
+            }
+        }
         panic!("Native had no branches matching pred {:?}", solution_predicate);
+    }
+}
 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 cylic_drain(self, mut func: impl FnMut(K, V, CyclicDrainInner<'_, K, V>)) {
         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 })
+            }
+        }
+    }
+}
 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 ProtoComponentBranch {
     fn feed_msg(&mut self, getter: PortId, payload: Payload) {
         let was = self.inbox.insert(getter, payload);
         assert!(was.is_none())
+    }
+}
 impl BranchingProtoComponent {
     fn drain_branches_to_blocked(
         cd: CyclicDrainer<Predicate, ProtoComponentBranch>,
         //
         logger: &mut dyn Logger,
         port_info: &PortInfo,
         proto_description: &ProtocolDescription,
         solution_storage: &mut SolutionStorage,
         mut outbox_unqueue: impl FnMut(PortId, SendPayloadMsg),
         proto_component_id: ProtoComponentId,
         ports: &HashSet<PortId>,
     ) {
         cd.cylic_drain(|mut predicate, mut branch, mut drainer| {
             let mut ctx = SyncProtoContext {
                     logger,
                     predicate: &predicate,
                     port_info,
                     proto_component_id,
                     inbox: &branch.inbox,
                 };
                 let blocker = branch.state.sync_run(&mut ctx, proto_description);
                 log!(
                     logger,
                     "Proto component with id {:?} branch with pred {:?} hit blocker {:?}",
                     proto_component_id,
                     &predicate,
                     &blocker,
                 );
                 use SyncBlocker as B;
                 match blocker {
@@ @@ -734,25 +643,24 @@ impl BranchingProtoComponent { @@
                         // sanity check
                         let var = port_info.firing_var_for(port);
                         assert!(predicate.query(var).is_none());
                         // keep forks in "unblocked"
                         drainer.add_input(predicate.clone().inserted(var, false), branch.clone());
                         drainer.add_input(predicate.inserted(var, true), branch);
+                    }
                     B::PutMsg(putter, payload) => {
                         // sanity check
                         assert_eq!(Some(&Putter), port_info.polarities.get(&putter));
                         // overwrite assignment
                         let var = port_info.firing_var_for(putter);
                         let was = predicate.assigned.insert(var, true);
                         if was == Some(false) {
                             log!(logger, "Proto component {:?} tried to PUT on port {:?} when pred said var {:?}==Some(false). inconsistent!", proto_component_id, putter, var);
                             // discard forever
                             drop((predicate, branch));
                         } else {
                             // keep in "unblocked"
                             log!(logger, "Proto component {:?} putting payload {:?} on port {:?} (using var {:?})", proto_component_id, &payload, putter, var);
                             outbox_unqueue(
                                 putter,
                                 SendPayloadMsg { predicate: predicate.clone(), payload },
                             );
@@ @@ -764,72 +672,87 @@ impl BranchingProtoComponent { @@
+    }
     fn feed_msg(
         &mut self,
         logger: &mut dyn Logger,
         port_info: &PortInfo,
         proto_description: &ProtocolDescription,
         solution_storage: &mut SolutionStorage,
         proto_component_id: ProtoComponentId,
         outbox_unqueue: impl FnMut(PortId, SendPayloadMsg),
         getter: PortId,
         send_payload_msg: SendPayloadMsg,
     ) {
         log!(
             logger,
             "feeding proto component {:?} getter {:?} {:?}",
             proto_component_id,
             getter,
             &send_payload_msg
         );
         let BranchingProtoComponent { branches, ports } = self;
         let mut unblocked = HashMap::default();
         let mut blocked = HashMap::default();
         // partition drain from branches -> {unblocked, blocked}
         log!(logger, "visiting {} blocked branches...", branches.len());
         for (predicate, mut branch) in branches.drain() {
             use CommonSatResult as Csr;
             log!(logger, "visiting branch with pred {:?}", &predicate);
             match predicate.common_satisfier(&send_payload_msg.predicate) {
                 Csr::Nonexistant => {
                     // this branch does not receive the message
                     log!(logger, "skipping branch");
                     blocked.insert(predicate, branch);
+                }
                 Csr::Equivalent | Csr::FormerNotLatter => {
                     // retain the existing predicate, but add this payload
                     log!(logger, "feeding this branch without altering its predicate");
                     branch.feed_msg(getter, send_payload_msg.payload.clone());
                     unblocked.insert(predicate, branch);
+                }
                 Csr::LatterNotFormer => {
                     // fork branch, give fork the message and payload predicate. original branch untouched
                     log!(logger, "Forking this branch, giving it the predicate of the msg");
                     let mut branch2 = branch.clone();
                     let predicate2 = send_payload_msg.predicate.clone();
                     branch2.feed_msg(getter, send_payload_msg.payload.clone());
                     blocked.insert(predicate, branch);
                     unblocked.insert(predicate2, branch2);
+                }
                 Csr::New(predicate2) => {
                     // fork branch, give fork the message and the new predicate. original branch untouched
                     log!(logger, "Forking this branch with new predicate {:?}", &predicate2);
                     let mut branch2 = branch.clone();
                     branch2.feed_msg(getter, send_payload_msg.payload.clone());
                     blocked.insert(predicate, branch);
                     unblocked.insert(predicate2, branch2);
+                }
+            }
+        }
         log!(logger, "blocked {:?} unblocked {:?}", blocked.len(), unblocked.len());
         // drain from unblocked --> blocked
         let mut swap = HashMap::default();
         let cd = CyclicDrainer::new(&mut unblocked, &mut swap, &mut blocked);
         BranchingProtoComponent::drain_branches_to_blocked(
             cd,
             logger,
             port_info,
             proto_description,
             solution_storage,
             outbox_unqueue,
             proto_component_id,
             ports,
         );
         // swap the blocked branches back
         std::mem::swap(&mut blocked, branches);
         log!(logger, "component settles down with branches: {:?}", branches.keys());
+    }
     fn collapse_with(self, solution_predicate: &Predicate) -> ProtoComponent {
         let BranchingProtoComponent { ports, branches } = self;
         for (branch_predicate, branch) in branches {
             if branch_predicate.satisfies(solution_predicate) {
                 let ProtoComponentBranch { state, .. } = branch;
                 return ProtoComponent { state, ports };
+            }
+        }
         panic!("ProtoComponent had no branches matching pred {:?}", solution_predicate);
+    }
     fn initial(ProtoComponent { state, ports }: ProtoComponent) -> Self {
@@ @@ -865,62 +788,58 @@ impl SolutionStorage { @@
         self.new_local.clear();
+    }
     pub(crate) fn reset(&mut self, subtree_ids: impl Iterator<Item = Route>) {
         self.subtree_id_to_index.clear();
         self.subtree_solutions.clear();
         self.old_local.clear();
         self.new_local.clear();
         for key in subtree_ids {
             self.subtree_id_to_index.insert(key, self.subtree_solutions.len());
             self.subtree_solutions.push(Default::default())
+        }
+    }
     pub(crate) fn peek_new_locals(&self) -> impl Iterator<Item = &Predicate> + '_ {
         self.new_local.iter()
+    }
     // pub(crate) fn peek_new_locals(&self) -> impl Iterator<Item = &Predicate> + '_ {
     //     self.new_local.iter()
     // }
     pub(crate) fn iter_new_local_make_old(&mut self) -> impl Iterator<Item = Predicate> + '_ {
         let Self { old_local, new_local, .. } = self;
         new_local.drain().map(move |local| {
             old_local.insert(local.clone());
             local
         })
+    }
     pub(crate) fn submit_and_digest_subtree_solution(
         &mut self,
         logger: &mut dyn Logger,
         subtree_id: Route,
         predicate: Predicate,
     ) {
         log!(logger, "NEW COMPONENT SOLUTION {:?} {:?}", subtree_id, &predicate);
         let index = self.subtree_id_to_index[&subtree_id];
         let left = 0..index;
         let right = (index + 1)..self.subtree_solutions.len();
         let Self { subtree_solutions, new_local, old_local, .. } = self;
         let was_new = subtree_solutions[index].insert(predicate.clone());
         if was_new {
             let set_visitor = left.chain(right).map(|index| &subtree_solutions[index]);
             Self::elaborate_into_new_local_rec(
                 logger,
                 predicate,
                 set_visitor,
                 old_local,
                 new_local,
             );
+        }
+    }
     fn elaborate_into_new_local_rec<'a, 'b>(
         logger: &mut dyn Logger,
         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(
@@ @@ -933,12 +852,101 @@ impl SolutionStorage { @@
+                }
+            }
         } else {
             // recursive stop condition. `partial` is a local subtree solution
             if !old_local.contains(&partial) {
                 // ... and it hasn't been found before
                 log!(logger, "storing NEW LOCAL SOLUTION {:?}", &partial);
                 new_local.insert(partial);
+            }
+        }
+    }
+}
 impl SyncProtoContext<'_> {
     pub fn is_firing(&mut self, port: PortId) -> Option<bool> {
         let var = self.port_info.firing_var_for(port);
         self.predicate.query(var)
+    }
     pub fn read_msg(&mut self, port: PortId) -> Option<&Payload> {
         self.inbox.get(&port)
+    }
+}
 impl<'a, K: Eq + Hash, V> CyclicDrainInner<'a, K, V> {
     fn add_input(&mut self, k: K, v: V) {
         self.swap.insert(k, v);
+    }
     fn add_output(&mut self, k: K, v: V) {
         self.output.insert(k, v);
+    }
+}
 impl NonsyncProtoContext<'_> {
     pub fn new_component(&mut self, moved_ports: HashSet<PortId>, state: ComponentState) {
         // called by a PROTO COMPONENT. moves its own ports.
         // 1. sanity check: this component owns these ports
         log!(
             self.logger,
             "Component {:?} added new component with state {:?}, moving ports {:?}",
             self.proto_component_id,
             &state,
             &moved_ports
         );
         assert!(self.proto_component_ports.is_subset(&moved_ports));
         // 2. remove ports from old component & update port->route
         let new_id = self.id_manager.new_proto_component_id();
         for port in moved_ports.iter() {
             self.proto_component_ports.remove(port);
             self.port_info
                 .routes
                 .insert(*port, Route::LocalComponent(LocalComponentId::Proto(new_id)));
+        }
         // 3. create a new component
         self.unrun_components.push((new_id, ProtoComponent { state, ports: moved_ports }));
+    }
     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 [o, i] = [self.id_manager.new_port_id(), self.id_manager.new_port_id()];
         self.proto_component_ports.insert(o);
         self.proto_component_ports.insert(i);
         // {polarity, peer, route} known. {} unknown.
         self.port_info.polarities.insert(o, Putter);
         self.port_info.polarities.insert(i, Getter);
         self.port_info.peers.insert(o, i);
         self.port_info.peers.insert(i, o);
         let route = Route::LocalComponent(LocalComponentId::Proto(self.proto_component_id));
         self.port_info.routes.insert(o, route);
         self.port_info.routes.insert(i, route);
         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.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 cylic_drain(self, mut func: impl FnMut(K, V, CyclicDrainInner<'_, K, V>)) {
         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);
+        }
+    }
+}

src/runtime/endpoints.rs

➞

Show inline comments

@@ new file 100644 @@
 use super::*;
 struct MonitoredReader<R: Read> {
     bytes: usize,
     r: R,
+}
 /////////////////////
 impl Endpoint {
     pub fn try_recv<T: serde::de::DeserializeOwned>(&mut self) -> Result<Option<T>, EndpointError> {
         use EndpointError::*;
         // populate inbox as much as possible
         'read_loop: loop {
             match self.stream.read_to_end(&mut self.inbox) {
                 Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => break 'read_loop,
                 Ok(0) => break 'read_loop,
                 Ok(_) => (),
                 Err(_e) => return Err(BrokenEndpoint),
+            }
+        }
         let mut monitored = MonitoredReader::from(&self.inbox[..]);
         match bincode::deserialize_from(&mut monitored) {
             Ok(msg) => {
                 let msg_size = monitored.bytes_read();
                 self.inbox.drain(0..(msg_size.try_into().unwrap()));
                 Ok(Some(msg))
+            }
             Err(e) => match *e {
                 bincode::ErrorKind::Io(k) if k.kind() == std::io::ErrorKind::UnexpectedEof => {
                     Ok(None)
+                }
                 _ => Err(MalformedMessage),
                 // println!("SERDE ERRKIND {:?}", e);
                 // Err(MalformedMessage)
             },
+        }
+    }
     pub fn send<T: serde::ser::Serialize>(&mut self, msg: &T) -> Result<(), ()> {
         bincode::serialize_into(&mut self.stream, msg).map_err(drop)
+    }
+}
 impl EndpointManager {
     pub fn send_to(&mut self, index: usize, msg: &Msg) -> Result<(), ()> {
         self.endpoint_exts[index].endpoint.send(msg)
+    }
     pub fn try_recv_any(&mut self, deadline: Instant) -> Result<(usize, Msg), TryRecyAnyError> {
         use TryRecyAnyError::*;
         // 1. try messages already buffered
         if let Some(x) = self.undelayed_messages.pop() {
             return Ok(x);
+        }
         loop {
             // 2. try read a message from an endpoint that raised an event with poll() but wasn't drained
             while let Some(index) = self.polled_undrained.pop() {
                 let endpoint = &mut self.endpoint_exts[index].endpoint;
                 if let Some(msg) =
                     endpoint.try_recv().map_err(|error| EndpointError { error, index })?
+                {
                     if !endpoint.inbox.is_empty() {
                         // there may be another message waiting!
                         self.polled_undrained.insert(index);
+                    }
                     return Ok((index, msg));
+                }
+            }
             // 3. No message yet. Do we have enough time to poll?
             let remaining = deadline.checked_duration_since(Instant::now()).ok_or(Timeout)?;
             self.poll.poll(&mut self.events, Some(remaining)).map_err(|_| PollFailed)?;
             for event in self.events.iter() {
                 let Token(index) = event.token();
                 self.polled_undrained.insert(index);
+            }
             self.events.clear();
+        }
+    }
     pub fn undelay_all(&mut self) {
         if self.undelayed_messages.is_empty() {
             // fast path
             std::mem::swap(&mut self.delayed_messages, &mut self.undelayed_messages);
             return;
+        }
         // slow path
         self.undelayed_messages.extend(self.delayed_messages.drain(..));
+    }
+}
 impl Debug for Endpoint {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         f.debug_struct("Endpoint").field("inbox", &self.inbox).finish()
+    }
+}
 impl<R: Read> From<R> for MonitoredReader<R> {
     fn from(r: R) -> Self {
         Self { r, bytes: 0 }
+    }
+}
 impl<R: Read> MonitoredReader<R> {
     pub fn bytes_read(&self) -> usize {
         self.bytes
+    }
+}
 impl<R: Read> Read for MonitoredReader<R> {
     fn read(&mut self, buf: &mut [u8]) -> Result<usize, std::io::Error> {
         let n = self.r.read(buf)?;
         self.bytes += n;
         Ok(n)
+    }
+}
 impl Into<Msg> for SetupMsg {
     fn into(self) -> Msg {
         Msg::SetupMsg(self)
+    }
+}

src/runtime/mod.rs

➞

Show inline comments

 mod communication;
 mod endpoints;
 pub mod error;
-mod setup2;
+mod setup;
 #[cfg(test)]
 mod tests;
 use crate::common::*;
 use error::*;
 #[derive(
     Debug, Copy, Clone, Eq, PartialEq, Ord, Hash, PartialOrd, serde::Serialize, serde::Deserialize,
 )]
 pub struct FiringVar(PortId);
 #[derive(Debug, Clone, Copy, Eq, PartialEq, Hash)]
 pub enum LocalComponentId {
     Native,
     Proto(ProtoComponentId),
+}
 #[derive(Debug, Clone, Copy, Eq, PartialEq, Hash)]
 pub enum Route {
     LocalComponent(LocalComponentId),
     Endpoint { index: usize },
+}
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 pub struct MyPortInfo {
@@ @@ -72,24 +69,28 @@ pub struct Endpoint { @@
     inbox: Vec<u8>,
     stream: TcpStream,
+}
 #[derive(Debug, Clone)]
 pub struct ProtoComponent {
     state: ComponentState,
     ports: HashSet<PortId>,
+}
 pub trait Logger: Debug {
     fn line_writer(&mut self) -> &mut dyn std::fmt::Write;
     fn dump_log(&self, w: &mut dyn std::io::Write);
+}
 #[derive(Debug)]
 pub struct StringLogger(ControllerId, String);
 #[derive(Debug)]
 pub struct DummyLogger;
 #[derive(Debug, Clone)]
 pub struct EndpointSetup {
     pub sock_addr: SocketAddr,
     pub is_active: bool,
+}
 #[derive(Debug)]
 pub struct EndpointExt {
     endpoint: Endpoint,
     getter_for_incoming: PortId,
+}
 #[derive(Debug)]
 pub struct Neighborhood {
@@ @@ -141,104 +142,48 @@ pub enum ConnectorPhased { @@
         endpoint_setups: Vec<(PortId, EndpointSetup)>,
         surplus_sockets: u16,
     },
     Communication {
         round_index: usize,
         endpoint_manager: EndpointManager,
         neighborhood: Neighborhood,
         mem_inbox: Vec<MemInMsg>,
         native_batches: Vec<NativeBatch>,
         round_result: Result<Option<(usize, HashMap<PortId, Payload>)>, SyncError>,
     },
+}
 #[derive(Debug)]
 pub struct StringLogger(ControllerId, String);
 #[derive(Default, Clone, Eq, PartialEq, Hash, serde::Serialize, serde::Deserialize)]
 pub struct Predicate {
     pub assigned: BTreeMap<FiringVar, bool>,
+}
 #[derive(Debug, Default)]
 pub struct NativeBatch {
     // invariant: putters' and getters' polarities respected
     to_put: HashMap<PortId, Payload>,
     to_get: HashSet<PortId>,
+}
 pub struct MonitoredReader<R: Read> {
     bytes: usize,
     r: R,
+}
 pub struct NonsyncProtoContext<'a> {
     logger: &'a mut dyn Logger,
     proto_component_id: ProtoComponentId,
     port_info: &'a mut PortInfo,
     id_manager: &'a mut IdManager,
     proto_component_ports: &'a mut HashSet<PortId>,
     unrun_components: &'a mut Vec<(ProtoComponentId, ProtoComponent)>,
+}
 pub struct SyncProtoContext<'a> {
     logger: &'a mut dyn Logger,
     predicate: &'a Predicate,
     proto_component_id: ProtoComponentId,
     port_info: &'a PortInfo,
     inbox: &'a HashMap<PortId, Payload>,
+}
 // pub struct MonoPContext<'a> {
 //     inner: &'a mut ControllerInner,
 //     ports: &'a mut HashSet<PortId>,
 //     mono_ps: &'a mut Vec<MonoP>,
 // }
 // pub struct PolyPContext<'a> {
 //     my_subtree_id: SubtreeId,
 //     inner: &'a mut Connector,
 //     solution_storage: &'a mut SolutionStorage,
 // }
 // impl PolyPContext<'_> {
 //     #[inline(always)]
 //     fn reborrow<'a>(&'a mut self) -> PolyPContext<'a> {
 //         let Self { solution_storage, my_subtree_id, inner } = self;
 //         PolyPContext { solution_storage, my_subtree_id: *my_subtree_id, inner }
 //     }
 // }
 // struct BranchPContext<'m, 'r> {
 //     m_ctx: PolyPContext<'m>,
 //     ports: &'r HashSet<PortId>,
 //     predicate: &'r Predicate,
 //     inbox: &'r HashMap<PortId, Payload>,
 // }
 // #[derive(Debug)]
 // pub enum SyncRunResult {
 //     BlockingForRecv,
 //     AllBranchesComplete,
 //     NoBranches,
 // }
 // #[derive(Debug, Copy, Clone, Eq, PartialEq, Ord, PartialOrd, Hash)]
 // pub enum PolyId {
 //     N,
 //     P { index: usize },
 // }
 // #[derive(Debug, Copy, Clone, Eq, PartialEq, Ord, PartialOrd, Hash)]
 // pub enum SubtreeId {
 //     PolyN,
 //     PolyP { index: usize },
 //     ChildController { port: PortId },
 // }
 // #[derive(Debug)]
 // pub struct NativeBranch {
 //     gotten: HashMap<PortId, Payload>,
 //     to_get: HashSet<PortId>,
 // }
 ////////////////
 impl PortInfo {
     fn firing_var_for(&self, port: PortId) -> FiringVar {
         FiringVar(match self.polarities.get(&port).unwrap() {
             Getter => port,
             Putter => *self.peers.get(&port).unwrap(),
         })
+    }
+}
 impl IdManager {
     fn new(controller_id: ControllerId) -> Self {
         Self {
@@ @@ -305,183 +250,82 @@ impl Connector { @@
         self.native_ports.retain(|port| !ports.contains(port));
         // 4. add new component
         self.proto_components.insert(
             new_id,
             ProtoComponent {
                 state: self.proto_description.new_main_component(identifier, ports),
                 ports: ports.iter().copied().collect(),
             },
         );
         Ok(())
+    }
+}
 impl EndpointManager {
     fn send_to(&mut self, index: usize, msg: &Msg) -> Result<(), ()> {
         self.endpoint_exts[index].endpoint.send(msg)
+    }
     fn try_recv_any(&mut self, deadline: Instant) -> Result<(usize, Msg), TryRecyAnyError> {
         use TryRecyAnyError::*;
         // 1. try messages already buffered
         if let Some(x) = self.undelayed_messages.pop() {
             return Ok(x);
+        }
         loop {
             // 2. try read a message from an endpoint that raised an event with poll() but wasn't drained
             while let Some(index) = self.polled_undrained.pop() {
                 let endpoint = &mut self.endpoint_exts[index].endpoint;
                 if let Some(msg) =
                     endpoint.try_recv().map_err(|error| EndpointError { error, index })?
+                {
                     if !endpoint.inbox.is_empty() {
                         // there may be another message waiting!
                         self.polled_undrained.insert(index);
+                    }
                     return Ok((index, msg));
+                }
+            }
             // 3. No message yet. Do we have enough time to poll?
             let remaining = deadline.checked_duration_since(Instant::now()).ok_or(Timeout)?;
             self.poll.poll(&mut self.events, Some(remaining)).map_err(|_| PollFailed)?;
             for event in self.events.iter() {
                 let Token(index) = event.token();
                 self.polled_undrained.insert(index);
+            }
             self.events.clear();
+        }
+    }
     fn undelay_all(&mut self) {
         if self.undelayed_messages.is_empty() {
             // fast path
             std::mem::swap(&mut self.delayed_messages, &mut self.undelayed_messages);
             return;
+        }
         // slow path
         self.undelayed_messages.extend(self.delayed_messages.drain(..));
+    }
+}
 impl Debug for Endpoint {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         f.debug_struct("Endpoint").field("inbox", &self.inbox).finish()
+    }
+}
 impl<R: Read> From<R> for MonitoredReader<R> {
     fn from(r: R) -> Self {
         Self { r, bytes: 0 }
+    }
+}
 impl<R: Read> MonitoredReader<R> {
     pub fn bytes_read(&self) -> usize {
         self.bytes
+    }
+}
 impl<R: Read> Read for MonitoredReader<R> {
     fn read(&mut self, buf: &mut [u8]) -> Result<usize, std::io::Error> {
         let n = self.r.read(buf)?;
         self.bytes += n;
         Ok(n)
 impl Logger for DummyLogger {
     fn line_writer(&mut self) -> &mut dyn std::fmt::Write {
         impl std::fmt::Write for DummyLogger {
             fn write_str(&mut self, _: &str) -> Result<(), std::fmt::Error> {
                 Ok(())
+            }
+        }
 impl Into<Msg> for SetupMsg {
     fn into(self) -> Msg {
         Msg::SetupMsg(self)
         self
+    }
     fn dump_log(&self, _: &mut dyn std::io::Write) {}
+}
 impl StringLogger {
     pub fn new(controller_id: ControllerId) -> Self {
         Self(controller_id, String::default())
+    }
+}
 impl Drop for StringLogger {
     fn drop(&mut self) {
-        let stdout = std::io::stdout();
+        let stdout = std::io::stderr();
         let mut lock = stdout.lock();
         writeln!(lock, "--- DROP LOG DUMP ---").unwrap();
         self.dump_log(&mut lock);
         // lock.flush().unwrap();
         // std::thread::sleep(Duration::from_millis(50));
+    }
+}
 impl Logger for StringLogger {
     fn line_writer(&mut self) -> &mut dyn std::fmt::Write {
         use std::fmt::Write;
         let _ = write!(&mut self.1, "\nCID({}): ", self.0);
         self
+    }
     fn dump_log(&self, w: &mut dyn std::io::Write) {
         let _ = w.write(self.1.as_bytes());
+    }
+}
 impl std::fmt::Write for StringLogger {
     fn write_str(&mut self, s: &str) -> Result<(), std::fmt::Error> {
         self.1.write_str(s)
+    }
+}
 impl Endpoint {
     fn try_recv<T: serde::de::DeserializeOwned>(&mut self) -> Result<Option<T>, EndpointError> {
         use EndpointError::*;
         // populate inbox as much as possible
         'read_loop: loop {
             match self.stream.read_to_end(&mut self.inbox) {
                 Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => break 'read_loop,
                 Ok(0) => break 'read_loop,
                 Ok(_) => (),
                 Err(_e) => return Err(BrokenEndpoint),
+            }
+        }
         let mut monitored = MonitoredReader::from(&self.inbox[..]);
         match bincode::deserialize_from(&mut monitored) {
             Ok(msg) => {
                 let msg_size = monitored.bytes_read();
                 self.inbox.drain(0..(msg_size.try_into().unwrap()));
                 Ok(Some(msg))
+            }
             Err(e) => match *e {
                 bincode::ErrorKind::Io(k) if k.kind() == std::io::ErrorKind::UnexpectedEof => {
                     Ok(None)
+                }
                 _ => Err(MalformedMessage),
                 // println!("SERDE ERRKIND {:?}", e);
                 // Err(MalformedMessage)
             },
+        }
+    }
     fn send<T: serde::ser::Serialize>(&mut self, msg: &T) -> Result<(), ()> {
         bincode::serialize_into(&mut self.stream, msg).map_err(drop)
+    }
+}
 impl Connector {
     pub fn get_logger(&self) -> &dyn Logger {
         &*self.logger
+    }
     pub fn print_state(&self) {
         let stdout = std::io::stdout();
         let mut lock = stdout.lock();
         writeln!(
             lock,
             "--- Connector with ControllerId={:?}.\n::LOG_DUMP:\n",
             self.id_manager.controller_id
+        )
         .unwrap();
         self.get_logger().dump_log(&mut lock);
         writeln!(lock, "\n\nDEBUG_PRINT:\n{:#?}\n", self).unwrap();
+    }
+}
 // impl Debug for SolutionStorage {
 //     fn fmt(&self, f: &mut 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 Predicate {
     #[inline]
     pub fn inserted(mut self, k: FiringVar, v: bool) -> Self {
         self.assigned.insert(k, v);
         self
+    }
     // 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 {
@@ @@ -556,39 +400,24 @@ impl Predicate { @@
             [true, false] => Csr::LatterNotFormer, // ... equivalent to other.
             [false, false] => {
                 // ... which is the union of the predicates' assignments but
                 //     is equivalent to neither self nor other.
                 let mut new = self.clone();
                 for (&id, &b) in o_not_s {
                     new.assigned.insert(id, b);
+                }
                 Csr::New(new)
+            }
+        }
+    }
     // pub fn iter_matching(&self, value: bool) -> impl Iterator<Item = FiringVar> + '_ {
     //     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 = PortId>, value: bool) {
     //     for channel_id in channel_ids {
     //         self.assigned.entry(channel_id).or_insert(value);
     //     }
     // }
     // pub fn replace_assignment(&mut self, channel_id: PortId, 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, var: FiringVar) -> Option<bool> {
         self.assigned.get(&var).copied()

src/runtime/setup.rs

➞

Show inline comments

@@ file renamed from src/runtime/setup2.rs to src/runtime/setup.rs @@
 use crate::common::*;
 use crate::runtime::*;
 impl Connector {
     pub fn new_simple(
         proto_description: Arc<ProtocolDescription>,
         controller_id: ControllerId,
     ) -> Self {
         let logger = Box::new(StringLogger::new(controller_id));
         // let logger = Box::new(DummyLogger);
         let surplus_sockets = 8;
         Self::new(logger, proto_description, controller_id, surplus_sockets)
+    }
     pub fn new(
         logger: Box<dyn Logger>,
         proto_description: Arc<ProtocolDescription>,
         controller_id: ControllerId,
         surplus_sockets: u16,
     ) -> Self {
         Self {
             proto_description,
             proto_components: Default::default(),

src/runtime/tests.rs

➞

Show inline comments

@@ @@ -12,87 +12,77 @@ fn next_test_addr() -> SocketAddr { @@
     static TEST_PORT: AtomicU16 = AtomicU16::new(5_000);
     let port = TEST_PORT.fetch_add(1, SeqCst);
     SocketAddrV4::new(Ipv4Addr::LOCALHOST, port).into()
+}
 lazy_static::lazy_static! {
     static ref MINIMAL_PROTO: Arc<ProtocolDescription> =
         { Arc::new(reowolf::ProtocolDescription::parse(b"").unwrap()) };
+}
 #[test]
 fn simple_connector() {
     let c = Connector::new_simple(MINIMAL_PROTO.clone(), 0);
     println!("{:#?}", c);
     Connector::new_simple(MINIMAL_PROTO.clone(), 0);
+}
 #[test]
 fn new_port_pair() {
     let mut c = Connector::new_simple(MINIMAL_PROTO.clone(), 0);
     let [_, _] = c.new_port_pair();
     let [_, _] = c.new_port_pair();
     println!("{:#?}", c);
+}
 #[test]
 fn new_sync() {
     let mut c = Connector::new_simple(MINIMAL_PROTO.clone(), 0);
     let [o, i] = c.new_port_pair();
     c.add_component(b"sync", &[i, o]).unwrap();
     println!("{:#?}", c);
+}
 #[test]
 fn new_net_port() {
     let mut c = Connector::new_simple(MINIMAL_PROTO.clone(), 0);
     let sock_addr = next_test_addr();
     let _ = c.new_net_port(Getter, EndpointSetup { sock_addr, is_active: false }).unwrap();
     let _ = c.new_net_port(Putter, EndpointSetup { sock_addr, is_active: true }).unwrap();
     println!("{:#?}", c);
+}
 #[test]
 fn trivial_connect() {
     let mut c = Connector::new_simple(MINIMAL_PROTO.clone(), 0);
     c.connect(Duration::from_secs(1)).unwrap();
     println!("{:#?}", c);
+}
 #[test]
 fn single_node_connect() {
     let sock_addr = next_test_addr();
     let mut c = Connector::new_simple(MINIMAL_PROTO.clone(), 0);
     let _ = c.new_net_port(Getter, EndpointSetup { sock_addr, is_active: false }).unwrap();
     let _ = c.new_net_port(Putter, EndpointSetup { sock_addr, is_active: true }).unwrap();
     let res = c.connect(Duration::from_secs(1));
     println!("{:#?}", c);
     c.get_logger().dump_log(&mut std::io::stdout().lock());
     res.unwrap();
     c.connect(Duration::from_secs(1)).unwrap();
+}
 #[test]
 fn multithreaded_connect() {
     let sock_addr = next_test_addr();
     scope(|s| {
         s.spawn(|_| {
             let mut c = Connector::new_simple(MINIMAL_PROTO.clone(), 0);
             let _ = c.new_net_port(Getter, EndpointSetup { sock_addr, is_active: true }).unwrap();
             c.connect(Duration::from_secs(1)).unwrap();
             c.print_state();
         });
         s.spawn(|_| {
             let mut c = Connector::new_simple(MINIMAL_PROTO.clone(), 1);
             let _ = c.new_net_port(Putter, EndpointSetup { sock_addr, is_active: false }).unwrap();
             c.connect(Duration::from_secs(1)).unwrap();
             c.print_state();
         });
     })
     .unwrap();
+}
 #[test]
 fn put_no_sync() {
     let mut c = Connector::new_simple(MINIMAL_PROTO.clone(), 0);
     let [o, _] = c.new_port_pair();
     c.connect(Duration::from_secs(1)).unwrap();
     c.put(o, (b"hi" as &[_]).into()).unwrap();
+}
@@ @@ -110,25 +100,24 @@ fn dup_put_bad() { @@
     let mut c = Connector::new_simple(MINIMAL_PROTO.clone(), 0);
     let [o, _] = c.new_port_pair();
     c.connect(Duration::from_secs(1)).unwrap();
     c.put(o, (b"hi" as &[_]).into()).unwrap();
     c.put(o, (b"hi" as &[_]).into()).unwrap_err();
+}
 #[test]
 fn trivial_sync() {
     let mut c = Connector::new_simple(MINIMAL_PROTO.clone(), 0);
     c.connect(Duration::from_secs(1)).unwrap();
     c.sync(Duration::from_secs(1)).unwrap();
     c.print_state();
+}
 #[test]
 fn unconnected_gotten_err() {
     let mut c = Connector::new_simple(MINIMAL_PROTO.clone(), 0);
     let [_, i] = c.new_port_pair();
     assert_eq!(reowolf::error::GottenError::NoPreviousRound, c.gotten(i).unwrap_err());
+}
 #[test]
 fn connected_gotten_err_no_round() {
     let mut c = Connector::new_simple(MINIMAL_PROTO.clone(), 0);
@@ @@ -241,39 +230,125 @@ fn connector_pair_nondet() { @@
             let mut c = Connector::new_simple(MINIMAL_PROTO.clone(), 1);
             let p = c.new_net_port(Putter, EndpointSetup { sock_addr, is_active: false }).unwrap();
             c.connect(Duration::from_secs(1)).unwrap();
             c.put(p, (b"hello" as &[_]).into()).unwrap();
             c.sync(Duration::from_secs(1)).unwrap();
         });
     })
     .unwrap();
+}
 #[test]
 fn cannot_use_moved_ports() {
     let mut c = Connector::new_simple(MINIMAL_PROTO.clone(), 1);
     let [p, g] = c.new_port_pair();
     c.add_component(b"sync", &[g, p]).unwrap();
     /*
     native p|-->|g sync
     */
     let mut c = Connector::new_simple(MINIMAL_PROTO.clone(), 1);
     let [p, g] = c.new_port_pair();
     c.add_component(b"sync", &[g, p]).unwrap();
     c.connect(Duration::from_secs(1)).unwrap();
     c.put(p, (b"hello" as &[_]).into()).unwrap_err();
     c.get(g).unwrap_err();
+}
 #[test]
 fn sync_sync() {
     let mut c = Connector::new_simple(MINIMAL_PROTO.clone(), 1);
     let [p0, g0] = c.new_port_pair();
     let [p1, g1] = c.new_port_pair();
     c.add_component(b"sync", &[g0, p1]).unwrap();
     /*
     native p0|-->|g0 sync
            g1|<--|p1
     */
     let mut c = Connector::new_simple(MINIMAL_PROTO.clone(), 0);
     let [p0, g0] = c.new_port_pair();
     let [p1, g1] = c.new_port_pair();
     c.add_component(b"sync", &[g0, p1]).unwrap();
     c.connect(Duration::from_secs(1)).unwrap();
     c.put(p0, (b"hello" as &[_]).into()).unwrap();
     c.get(g1).unwrap();
     c.sync(Duration::from_secs(1)).unwrap();
     c.gotten(g1).unwrap();
+}
 #[test]
 fn double_net_connect() {
     let sock_addrs = [next_test_addr(), next_test_addr()];
     scope(|s| {
         s.spawn(|_| {
             let mut c = Connector::new_simple(MINIMAL_PROTO.clone(), 2);
             let [_p, _g] = [
                 c.new_net_port(Putter, EndpointSetup { sock_addr: sock_addrs[0], is_active: true })
                     .unwrap(),
                 c.new_net_port(Getter, EndpointSetup { sock_addr: sock_addrs[1], is_active: true })
                     .unwrap(),
             ];
             c.connect(Duration::from_secs(1)).unwrap();
         });
         s.spawn(|_| {
             let mut c = Connector::new_simple(MINIMAL_PROTO.clone(), 3);
             let [_g, _p] = [
                 c.new_net_port(
                     Getter,
                     EndpointSetup { sock_addr: sock_addrs[0], is_active: false },
+                )
                 .unwrap(),
                 c.new_net_port(
                     Putter,
                     EndpointSetup { sock_addr: sock_addrs[1], is_active: false },
+                )
                 .unwrap(),
             ];
             c.connect(Duration::from_secs(1)).unwrap();
         });
     })
     .unwrap();
+}
 #[test]
 fn distributed_msg_bounce() {
     /*
     native[0] | sync 0.p|-->|1.p native[1]
 .g|<--|1.g
     */
     let sock_addrs = [next_test_addr(), next_test_addr()];
     scope(|s| {
         s.spawn(|_| {
             /*
             native | sync p|-->
                    |      g|<--
             */
             let mut c = Connector::new_simple(MINIMAL_PROTO.clone(), 4);
             let [p, g] = [
                 c.new_net_port(Putter, EndpointSetup { sock_addr: sock_addrs[0], is_active: true })
                     .unwrap(),
                 c.new_net_port(Getter, EndpointSetup { sock_addr: sock_addrs[1], is_active: true })
                     .unwrap(),
             ];
             c.add_component(b"sync", &[g, p]).unwrap();
             c.connect(Duration::from_secs(1)).unwrap();
             c.sync(Duration::from_secs(1)).unwrap();
         });
         s.spawn(|_| {
             /*
             native p|-->
                    g|<--
             */
             let mut c = Connector::new_simple(MINIMAL_PROTO.clone(), 5);
             let [g, p] = [
                 c.new_net_port(
                     Getter,
                     EndpointSetup { sock_addr: sock_addrs[0], is_active: false },
+                )
                 .unwrap(),
                 c.new_net_port(
                     Putter,
                     EndpointSetup { sock_addr: sock_addrs[1], is_active: false },
+                )
                 .unwrap(),
             ];
             c.connect(Duration::from_secs(1)).unwrap();
             c.put(p, (b"hello" as &[_]).into()).unwrap();
             c.get(g).unwrap();
             c.sync(Duration::from_secs(1)).unwrap();
             c.gotten(g).unwrap();
         });
     })
     .unwrap();
+}

0 comments (0 inline, 0 general)