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

Changeset - 0a71d0af9edf

Parent rev.

Child rev.

[Not reviewed]

0 7 0

Christopher Esterhuyse - 5 years ago 2020-03-03 15:04:11
christopheresterhuyse@gmail.com

histories

7 files changed with 221 insertions and 83 deletions:

src/common.rs

src/protocol/eval.rs

src/runtime/actors.rs

src/runtime/communication.rs

src/runtime/mod.rs

src/runtime/setup.rs

src/test/connector.rs

0 comments (0 inline, 0 general)

src/common.rs

➞

Show inline comments

@@ @@ -38,13 +38,13 @@ pub struct ChannelId { @@
 #[derive(Debug, Eq, PartialEq, Clone, Hash, Copy, Ord, PartialOrd)]
 pub enum Polarity {
     Putter, // output port (from the perspective of the component)
     Getter, // input port (from the perspective of the component)
+}
-#[derive(Eq, PartialEq, Ord, PartialOrd, Hash, Copy, Clone, Debug)]
 #[derive(Eq, PartialEq, Ord, PartialOrd, Hash, Copy, Clone)]
 #[repr(C)]
 pub struct Port(pub usize); // ports are COPY
 pub type Key = Port;
 #[derive(Eq, PartialEq, Copy, Clone, Debug)]
 pub enum MainComponentErr {
@@ @@ -108,12 +108,17 @@ pub trait PolyContext { @@
     fn is_firing(&mut self, ekey: Key) -> Option<bool>;
     fn read_msg(&mut self, ekey: Key) -> Option<&Payload>;
+}
 ///////////////////// IMPL /////////////////////
 impl Debug for Port {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         write!(f, "Port({})", self.0)
+    }
+}
 impl Key {
     pub fn from_raw(raw: usize) -> Self {
         Self(raw)
+    }
     pub fn to_raw(self) -> usize {
         self.0

src/protocol/eval.rs

➞

Show inline comments

@@ @@ -1454,13 +1454,13 @@ impl Store { @@
                 let subject;
                 match &h[indexing.subject] {
                     Expression::Variable(var) => {
                         let var = var.declaration.unwrap();
                         subject = self.map.get(&var).unwrap();
+                    }
-                    _ => unreachable!(),
+                    q => unreachable!("Reached {:?}", q),
+                }
                 match subject.get(&index) {
                     Some(value) => Ok(value),
                     None => Err(EvalContinuation::Inconsistent),
+                }
+            }
@@ @@ -1469,13 +1469,13 @@ impl Store { @@
                 let subject;
                 match &h[selecting.subject] {
                     Expression::Variable(var) => {
                         let var = var.declaration.unwrap();
                         subject = self.map.get(&var).unwrap();
+                    }
-                    _ => unreachable!(),
+                    q => unreachable!("Reached {:?}", q),
+                }
                 match subject.length() {
                     Some(value) => Ok(value),
                     None => Err(EvalContinuation::Inconsistent),
+                }
+            }

src/runtime/actors.rs

➞

Show inline comments

 use crate::common::*;
 use crate::runtime::{endpoint::*, *};
 #[derive(Debug)]
+#[derive(Debug, Clone)]
 pub(crate) struct MonoN {
     pub ekeys: HashSet<Key>,
     pub result: Option<(usize, HashMap<Key, Payload>)>,
+}
 #[derive(Debug)]
 pub(crate) struct PolyN {
@@ @@ -15,25 +15,26 @@ pub(crate) struct PolyN { @@
 pub(crate) struct BranchN {
     pub to_get: HashSet<Key>,
     pub gotten: HashMap<Key, Payload>,
     pub sync_batch_index: usize,
+}
 #[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct MonoP {
     pub state: ProtocolS,
     pub ekeys: HashSet<Key>,
+}
 #[derive(Debug)]
 pub(crate) struct PolyP {
     pub incomplete: HashMap<Predicate, BranchP>,
     pub complete: HashMap<Predicate, BranchP>,
     pub ekeys: HashSet<Key>,
+}
 #[derive(Debug, Clone)]
 pub(crate) struct BranchP {
     pub blocking_on: Option<Key>,
     pub outbox: HashMap<Key, Payload>,
     pub inbox: HashMap<Key, Payload>,
     pub state: ProtocolS,
+}
 //////////////////////////////////////////////////////////////////
@@ @@ -86,12 +87,13 @@ impl PolyP { @@
                         &branch.inbox,
                     );
                     if predicate.replace_assignment(channel_id, true) != Some(false) {
                         // don't rerun now. Rerun at next `sync_run`
                         log!(&mut m_ctx.inner.logger, "~ ... Delay {:?}", m_ctx.my_subtree_id,);
                         branch.blocking_on = Some(ekey);
                         self.incomplete.insert(predicate, branch);
                     } else {
                         log!(&mut m_ctx.inner.logger, "~ ... Drop {:?}", m_ctx.my_subtree_id,);
+                    }
                     // ELSE DROP
+                }
@@ @@ -211,18 +213,22 @@ impl PolyP { @@
             log!(
                 &mut m_ctx.inner.logger,
                 "... poly_recv_run matched running machine exactly! pred is {:?}",
                 &payload_predicate
             );
             branch.inbox.insert(ekey, payload);
             if branch.blocking_on == Some(ekey) {
                 branch.blocking_on = None;
                 vec![(payload_predicate, branch)]
             } else {
                 vec![]
+            }
         } else {
             log!(
                 &mut m_ctx.inner.logger,
                 "... poly_recv_run didn't have any exact matches... Let's try feed it to all branches",
             );
             let mut incomplete2 = HashMap::<_, _>::default();
             let to_run = self
                 .incomplete
                 .drain()
                 .filter_map(|(old_predicate, mut branch)| {
@@ @@ -238,13 +244,21 @@ impl PolyP { @@
                             if let Some(prev_payload) = branch.inbox.get(&ekey) {
                                 // Incorrect to receive two distinct messages in same branch!
                                 assert_eq!(prev_payload, &payload);
+                            }
                             branch.inbox.insert(ekey, payload.clone());
                             if branch.blocking_on == Some(ekey) {
                                 // run.
                                 branch.blocking_on = None;
                                 Some((old_predicate, branch))
                             } else {
                                 // don't bother running. its awaiting something else
                                 incomplete2.insert(old_predicate, branch);
                                 None
+                            }
+                        }
                         Csr::New(new) => {
                             log!(
                                 &mut m_ctx.inner.logger,
                                 "... poly_recv_run payloadpred {:?} and branchpred {:?} satisfied by new pred {:?}. FORKING",
                                 &payload_predicate,
@@ @@ -258,13 +272,21 @@ impl PolyP { @@
                                 assert_eq!(prev_payload, &payload);
+                            }
                             payload_branch.inbox.insert(ekey, payload.clone());
                             // put the original back untouched
                             incomplete2.insert(old_predicate, branch);
                             if payload_branch.blocking_on == Some(ekey) {
                                 // run the fork
                                 payload_branch.blocking_on = None;
                                 Some((new, payload_branch))
                             } else {
                                 // don't bother running. its awaiting something else
                                 incomplete2.insert(new, payload_branch);
                                 None
+                            }
+                        }
                         Csr::LatterNotFormer => {
                             log!(
                                 &mut m_ctx.inner.logger,
                                 "... poly_recv_run payloadpred {:?} subsumes branch pred {:?}. FORKING",
                                 &old_predicate,
@@ @@ -276,14 +298,22 @@ impl PolyP { @@
                                 // Incorrect to receive two distinct messages in same branch!
                                 assert_eq!(prev_payload, &payload);
+                            }
                             payload_branch.inbox.insert(ekey, payload.clone());
                             // put the original back untouched
                             incomplete2.insert(old_predicate, branch);
                             Some((payload_predicate.clone(), payload_branch))
                             incomplete2.insert(old_predicate.clone(), branch);
                             if payload_branch.blocking_on == Some(ekey) {
                                 // run the fork
                                 payload_branch.blocking_on = None;
                                 Some((old_predicate, payload_branch))
                             } else {
                                 // don't bother running. its awaiting something else
                                 incomplete2.insert(old_predicate, payload_branch);
                                 None
+                            }
+                        }
                         Csr::Nonexistant => {
                             log!(
                                 &mut m_ctx.inner.logger,
                                 "... poly_recv_run SKIPPING because branchpred={:?}. payloadpred={:?}",
                                 &old_predicate,
@@ @@ -304,27 +334,17 @@ impl PolyP { @@
             "... DONE FEEDING BRANCHES. {} branches to run!",
             to_run.len(),
         );
         self.poly_run_these_branches(m_ctx, protocol_description, to_run)
+    }
     pub(crate) fn become_mono(
         mut self,
         decision: &Predicate,
         table_row: &mut HashMap<Key, Payload>,
     ) -> MonoP {
         if let Some((_, branch)) = self.complete.drain().find(|(p, _)| decision.satisfies(p)) {
             let BranchP { inbox, state, outbox } = branch;
             for (key, payload) in inbox.into_iter().chain(outbox.into_iter()) {
                 table_row.insert(key, payload);
+            }
             self.incomplete.clear();
             MonoP { state, ekeys: self.ekeys }
         } else {
             panic!("No such solution!")
+        }
     pub(crate) fn choose_mono(&self, decision: &Predicate) -> Option<MonoP> {
         self.complete
             .iter()
             .find(|(p, _)| decision.satisfies(p))
             .map(|(_, branch)| MonoP { state: branch.state.clone(), ekeys: self.ekeys.clone() })
+    }
+}
 impl PolyN {
     pub fn sync_recv(
         &mut self,
@@ @@ -393,35 +413,16 @@ impl PolyN { @@
             branches2.len(),
             branches2.keys().collect::<Vec<_>>()
         );
         std::mem::swap(&mut branches2, &mut self.branches);
+    }
     pub fn become_mono(
         mut self,
         logger: &mut String,
         decision: &Predicate,
         table_row: &mut HashMap<Key, Payload>,
     ) -> MonoN {
         log!(
             logger,
             "decision {:?} with branch preds {:?}",
             decision,
             self.branches.iter().collect::<Vec<_>>()
         );
         if let Some((branch_pred, branch)) = self
             .branches
             .drain()
     pub fn choose_mono(&self, decision: &Predicate) -> Option<MonoN> {
         self.branches
             .iter()
             .find(|(p, branch)| branch.to_get.is_empty() && decision.satisfies(p))
+        {
             log!(logger, "decision {:?} mapped to branch {:?}", decision, branch_pred);
             let BranchN { gotten, sync_batch_index, .. } = branch;
             for (&key, payload) in gotten.iter() {
                 assert!(table_row.insert(key, payload.clone()).is_none());
+            }
             MonoN { ekeys: self.ekeys, result: Some((sync_batch_index, gotten)) }
         } else {
             log!(logger, "decision {:?} HAD NO SOLUTION!!?", decision);
             panic!("No such solution!")
+        }
             .map(|(_, branch)| {
                 let BranchN { gotten, sync_batch_index, .. } = branch.clone();
                 MonoN { ekeys: self.ekeys.clone(), result: Some((sync_batch_index, gotten)) }
             })
+    }
+}

src/runtime/communication.rs

➞

Show inline comments

 use crate::common::*;
 use crate::runtime::{actors::*, endpoint::*, errors::*, *};
 impl Controller {
     //usable BETWEEN rounds
     fn consistent(&self) -> bool {
         self.inner.mono_n.is_some()
+    }
     fn end_round_with_decision(&mut self, decision: Predicate) -> Result<(), SyncErr> {
         log!(&mut self.inner.logger, "ENDING ROUND WITH DECISION! {:?}", &decision);
         let mut table_row = HashMap::<Key, _>::default();
         // 1. become_mono for Poly actors
         self.inner.mono_n =
             self.ephemeral.poly_n.take().map(|poly_n| {
                 poly_n.become_mono(&mut self.inner.logger, &decision, &mut table_row)
             });
         self.inner.mono_ps.extend(
             self.ephemeral.poly_ps.drain(..).map(|m| m.become_mono(&decision, &mut table_row)),
         );
         // let mut table_row = HashMap::<Key, _>::default();
         // convert (Key=>Payload) map to (ChannelId=>Payload) map.
         let table_row: HashMap<_, _> = table_row
             .into_iter()
             .map(|(ekey, msg)| {
                 let channel_id = self.inner.endpoint_exts.get(ekey).unwrap().info.channel_id;
                 (channel_id, msg)
         let n_pair = {
             let poly_n = self.ephemeral.poly_n.take().unwrap();
             let mono_n = poly_n.choose_mono(&decision).unwrap();
             (mono_n, poly_n)
         };
         self.inner.mono_n = Some(n_pair.0.clone());
         let p_pairs = self
             .ephemeral
             .poly_ps
             .drain(..)
             .map(|poly_p| {
                 let mono_p = poly_p.choose_mono(&decision).unwrap();
                 (mono_p, poly_p)
             })
             .collect();
         // log all firing ports
         for (channel_id, payload) in table_row {
             log!(&mut self.inner.logger, "VALUE {:?} => Message({:?})", channel_id, payload);
+        }
         // log all silent ports
         for channel_id in decision.iter_matching(false) {
             log!(&mut self.inner.logger, "VALUE {:?} => *", channel_id);
+        }
             .collect::<Vec<_>>();
         self.inner.mono_ps.extend(p_pairs.iter().map(|p_pair| p_pair.0.clone()));
         self.round_histories.push(RoundHistory::Consistent(decision.clone(), n_pair, p_pairs));
         let announcement =
             CommMsgContents::Announce { oracle: decision }.into_msg(self.inner.round_index);
         for &child_ekey in self.inner.family.children_ekeys.iter() {
             log!(
                 &mut self.inner.logger,
                 "Forwarding {:?} to child with ekey {:?}",
@@ @@ -136,21 +135,50 @@ impl Controller { @@
+            }
             branches.insert(predicate, branch);
+        }
         Ok(PolyN { ekeys, branches })
+    }
     // Runs a synchronous round until all the actors are in decided state OR 1+ are inconsistent.
     // If a native requires setting up, arg `sync_batches` is Some, and those are used as the sync batches.
     pub fn sync_round(
         &mut self,
         deadline: Instant,
         sync_batches: Option<impl Iterator<Item = SyncBatch>>,
     ) -> Result<(), SyncErr> {
         // TODO! fuse handle_locals_return_decision and end_round_return_decision
         if !self.consistent() {
             // was previously inconsistent
             return Err(SyncErr::Inconsistent);
+        }
         match self.sync_round_inner(deadline, sync_batches) {
             Ok(()) => Ok(()),
             Err(e) => {
                 log!(
                     &mut self.inner.logger,
                     "/\\/\\/\\/\\/\\/ Sync round failed! Preparing for diagnosis...",
                 );
                 let h = RoundHistory::Inconsistent(
                     std::mem::take(&mut self.ephemeral.solution_storage),
                     self.ephemeral.poly_n.take().unwrap(),
                     std::mem::take(&mut self.ephemeral.poly_ps),
                 );
                 self.round_histories.push(h);
                 for (round_index, round) in self.round_histories.iter().enumerate() {
                     log!(&mut self.inner.logger, "round {}:{:#?}\n", round_index, round);
+                }
                 Err(e)
+            }
+        }
+    }
     // Runs a synchronous round until all the actors are in decided state OR 1+ are inconsistent.
     // If a native requires setting up, arg `sync_batches` is Some, and those are used as the sync batches.
     fn sync_round_inner(
         &mut self,
         deadline: Instant,
         sync_batches: Option<impl Iterator<Item = SyncBatch>>,
     ) -> Result<(), SyncErr> {
         assert!(self.ephemeral.is_clear());
         log!(
             &mut self.inner.logger,
             "~~~~~~~~ SYNC ROUND STARTS! ROUND={} ~~~~~~~~~",
             self.inner.round_index
@@ @@ -284,17 +312,13 @@ impl Controller { @@
         // 4. Receive incoming messages until the DECISION is made
         log!(&mut self.inner.logger, "`No decision yet`. Time to recv messages");
         self.undelay_all();
         'recv_loop: loop {
             log!(&mut self.inner.logger, "`POLLING`...");
             let received = self.recv(deadline)?.ok_or_else(|| {
                 log!(
                     &mut self.inner.logger,
                     ":( timing out. Solutions storage in state... {:#?}",
                     &self.ephemeral.solution_storage
                 );
                 log!(&mut self.inner.logger, ":( timing out");
                 SyncErr::Timeout
             })?;
             log!(&mut self.inner.logger, "::: message {:?}...", &received);
             let current_content = match received.msg {
                 Msg::SetupMsg(_) => {
                     // This occurs in the event the connector was malformed during connect()
@@ @@ -478,12 +502,13 @@ impl Into<PolyP> for MonoP { @@
             incomplete: hashmap! {
                 Predicate::new_trivial() =>
                 BranchP {
                     state: self.state,
                     inbox: Default::default(),
                     outbox: Default::default(),
                     blocking_on: None,
+                }
             },
             ekeys: self.ekeys,
+        }
+    }
+}

src/runtime/mod.rs

➞

Show inline comments

@@ @@ -3,13 +3,13 @@ pub mod ffi; @@
 mod actors;
 pub(crate) mod communication;
 pub(crate) mod connector;
 pub(crate) mod endpoint;
 pub mod errors;
 pub mod experimental;
+// pub mod experimental;
 mod serde;
 pub(crate) mod setup;
 pub(crate) type ProtocolD = crate::protocol::ProtocolDescriptionImpl;
 pub(crate) type ProtocolS = crate::protocol::ComponentStateImpl;
@@ @@ -92,17 +92,24 @@ struct MessengerState { @@
 #[derive(Debug)]
 struct ChannelIdStream {
     controller_id: ControllerId,
     next_channel_index: ChannelIndex,
+}
 #[derive(Debug)]
 enum RoundHistory {
     Consistent(Predicate, (MonoN, PolyN), Vec<(MonoP, PolyP)>),
     Inconsistent(SolutionStorage, PolyN, Vec<PolyP>),
+}
 #[derive(Debug)]
 struct Controller {
     protocol_description: Arc<ProtocolD>,
     inner: ControllerInner,
     ephemeral: ControllerEphemeral,
     round_histories: Vec<RoundHistory>,
+}
 #[derive(Debug)]
 struct ControllerInner {
     round_index: usize,
     channel_id_stream: ChannelIdStream,
     endpoint_exts: Arena<EndpointExt>,

src/runtime/setup.rs

➞

Show inline comments

@@ @@ -124,13 +124,18 @@ impl Controller { @@
             endpoint_exts,
             mono_ps: p_monos,
             mono_n: n_mono,
             round_index: 0,
             logger,
         };
         let controller = Self { protocol_description, inner, ephemeral: Default::default() };
         let controller = Self {
             protocol_description,
             inner,
             ephemeral: Default::default(),
             round_histories: vec![],
         };
         Ok((controller, native_interface))
+    }
     fn test_stream_connectivity(stream: &mut TcpStream) -> bool {
         use std::io::Write;
         stream.write(&[]).is_ok()

src/test/connector.rs

➞

Show inline comments

@@ @@ -76,12 +76,33 @@ primitive fifo_1(msg m, in i, out o) { @@
+        }
+    }
+}
 composite fifo_1_e(in i, out o) {
     new fifo_1(null, i, o);
+}
 primitive samelen(in a, in b, out c) {
     synchronous {
         msg m = get(a);
         msg n = get(b);
         assert(m.length == n.length);
         put(c, m);
+    }
+}
 primitive repl2(in a, out b, out c) {
     synchronous {
         msg m = get(a);
         put(b, m);
         put(c, m);
+    }
+}
 composite samelen_repl(in a, out b) {
     channel c -> d;
     channel e -> f;
     new samelen(a, f, c);
     new repl2(d, b, e);
+}
 ";
 #[test]
 fn connector_connects_ok() {
     // Test if we can connect natives using the given PDL
     /*
@@ @@ -677,6 +698,80 @@ fn connector_fifo_1_e() { @@
                 assert_eq!(Ok(0), x.sync(timeout));
                 assert_eq!(Ok(b"message~" as &[u8]), x.read_gotten(1));
+            }
         },
     ]));
+}
 #[test]
 #[should_panic]
 fn connector_causal_loop() {
     /*
         /-->\      /-->P|A-->\      /-->\
     Alice   exchange         exchange   Bob
         \<--/      \<--P|A<--/      \<--/
     */
     let timeout = Duration::from_millis(1_500);
     let addrs = [next_addr(), next_addr()];
     const N: usize = 1;
     assert!(run_connector_set(&[
         //
         &|x| {
             // Alice
             x.configure(PDL, b"exchange").unwrap();
             x.bind_port(0, Passive(addrs[0])).unwrap(); // peer out
             x.bind_port(1, Passive(addrs[1])).unwrap(); // peer in
             x.bind_port(2, Native).unwrap(); // native in
             x.bind_port(3, Native).unwrap(); // native out
             x.connect(timeout).unwrap();
             for _ in 0..N {
                 assert_eq!(Ok(()), x.put(0, b"A->B".to_vec()));
                 assert_eq!(Ok(()), x.get(1));
                 assert_eq!(Ok(0), x.sync(timeout));
                 assert_eq!(Ok(b"B->A" as &[u8]), x.read_gotten(1));
+            }
         },
         &|x| {
             // Bob
             x.configure(PDL, b"exchange").unwrap();
             x.bind_port(0, Active(addrs[1])).unwrap(); // peer out
             x.bind_port(1, Active(addrs[0])).unwrap(); // peer in
             x.bind_port(2, Native).unwrap(); // native in
             x.bind_port(3, Native).unwrap(); // native out
             x.connect(timeout).unwrap();
             for _ in 0..N {
                 assert_eq!(Ok(()), x.put(0, b"B->A".to_vec()));
                 assert_eq!(Ok(()), x.get(1));
                 assert_eq!(Ok(0), x.sync(timeout));
                 assert_eq!(Ok(b"A->B" as &[u8]), x.read_gotten(1));
+            }
         },
     ]));
+}
 #[test]
 #[should_panic]
 fn connector_causal_loop2() {
     /*
         /-->\     /<---\
     Alice   samelen-->repl
         \<-------------/
     */
     let timeout = Duration::from_millis(1_500);
     // let addrs = [next_addr(), next_addr()];
     const N: usize = 1;
     assert!(run_connector_set(&[
         //
         &|x| {
             // Alice
             x.configure(PDL, b"samelen_repl").unwrap();
             x.bind_port(0, Native).unwrap();
             x.bind_port(1, Native).unwrap();
             x.connect(timeout).unwrap();
             for _ in 0..N {
                 assert_eq!(Ok(()), x.put(0, b"foo".to_vec()));
                 assert_eq!(Ok(()), x.get(1));
                 assert_eq!(Ok(0), x.sync(timeout));
+            }
         },
     ]));
+}

0 comments (0 inline, 0 general)