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

Changeset - 39625667a9b6

Parent rev.

Child rev.

[Not reviewed]

0 2 0

Christopher Esterhuyse - 5 years ago 2020-02-04 15:07:29
christopheresterhuyse@gmail.com

more debug prints on branch predicates

2 files changed with 42 insertions and 3 deletions:

src/runtime/actors.rs

src/runtime/communication.rs

0 comments (0 inline, 0 general)

src/runtime/actors.rs

➞

Show inline comments

@@ @@ -9,105 +9,114 @@ pub(crate) struct MonoN { @@
 #[derive(Debug)]
 pub(crate) struct PolyN {
     pub ekeys: HashSet<Key>,
     pub branches: HashMap<Predicate, BranchN>,
+}
 #[derive(Debug, Clone)]
 pub(crate) struct BranchN {
     pub to_get: HashSet<Key>,
     pub gotten: HashMap<Key, Payload>,
     pub sync_batch_index: usize,
+}
 #[derive(Debug)]
 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 inbox: HashMap<Key, Payload>,
     pub state: ProtocolS,
+}
 //////////////////////////////////////////////////////////////////
 impl PolyP {
     pub(crate) fn poly_run(
         &mut self,
         m_ctx: PolyPContext,
         protocol_description: &ProtocolD,
     ) -> Result<SyncRunResult, EndpointErr> {
         let to_run: Vec<_> = self.incomplete.drain().collect();
         self.poly_run_these_branches(m_ctx, protocol_description, to_run)
+    }
     pub(crate) fn poly_run_these_branches(
         &mut self,
         mut m_ctx: PolyPContext,
         protocol_description: &ProtocolD,
         mut to_run: Vec<(Predicate, BranchP)>,
     ) -> Result<SyncRunResult, EndpointErr> {
         use SyncRunResult as Srr;
         let cid = m_ctx.inner.channel_id_stream.controller_id;
         lockprintln!("{:?}: ~ Running branches for PolyP {:?}!", cid, m_ctx.my_subtree_id,);
         while let Some((mut predicate, mut branch)) = to_run.pop() {
             let mut r_ctx = BranchPContext {
                 m_ctx: m_ctx.reborrow(),
                 ekeys: &self.ekeys,
                 predicate: &predicate,
                 inbox: &branch.inbox,
             };
             use PolyBlocker as Sb;
             let blocker = branch.state.sync_run(&mut r_ctx, protocol_description);
             lockprintln!(
                 "{:?}: ~ ... ran PolyP {:?} with branch pred {:?} to blocker {:?}",
                 cid,
                 r_ctx.m_ctx.my_subtree_id,
                 &predicate,
                 &blocker
             );
             match blocker {
                 Sb::Inconsistent => {} // DROP
                 Sb::CouldntReadMsg(ekey) => {
                     assert!(self.ekeys.contains(&ekey));
                     let channel_id =
                         r_ctx.m_ctx.inner.endpoint_exts.get(ekey).unwrap().info.channel_id;
                     if predicate.replace_assignment(channel_id, true) != Some(false) {
                         // don't rerun now. Rerun at next `sync_run`
                         self.incomplete.insert(predicate, branch);
+                    }
                     // ELSE DROP
+                }
                 Sb::CouldntCheckFiring(ekey) => {
                     assert!(self.ekeys.contains(&ekey));
                     let channel_id =
                         r_ctx.m_ctx.inner.endpoint_exts.get(ekey).unwrap().info.channel_id;
                     // split the branch!
                     let branch_f = branch.clone();
                     let mut predicate_f = predicate.clone();
                     if predicate_f.replace_assignment(channel_id, false).is_some() {
                         panic!("OI HANS QUERY FIRST!");
+                    }
                     assert!(predicate.replace_assignment(channel_id, true).is_none());
                     to_run.push((predicate, branch));
                     to_run.push((predicate_f, branch_f));
+                }
                 Sb::SyncBlockEnd => {
                     // come up with the predicate for this local solution
                     let ekeys_channel_id_iter = self
                         .ekeys
                         .iter()
                         .map(|&ekey| m_ctx.inner.endpoint_exts.get(ekey).unwrap().info.channel_id);
                     predicate.batch_assign_nones(ekeys_channel_id_iter, false);
                     // report the local solution
                     m_ctx
                         .solution_storage
                         .submit_and_digest_subtree_solution(m_ctx.my_subtree_id, predicate.clone());
                     // store the solution for recovering later
                     self.complete.insert(predicate, branch);
+                }
                 Sb::PutMsg(ekey, payload) => {
                     assert!(self.ekeys.contains(&ekey));
                     let EndpointExt { info, endpoint } =
                         m_ctx.inner.endpoint_exts.get_mut(ekey).unwrap();
                     if predicate.replace_assignment(info.channel_id, true) != Some(false) {
                         let msg = CommMsgContents::SendPayload {
                             payload_predicate: predicate.clone(),
                             payload,

src/runtime/communication.rs

➞

Show inline comments

@@ @@ -406,169 +406,199 @@ impl Controller { @@
                     if self.handle_locals_maybe_decide()? {
                         return Ok(());
+                    }
+                }
+            }
+        }
+    }
+}
 impl ControllerEphemeral {
     fn is_clear(&self) -> bool {
         self.solution_storage.is_clear()
             && self.poly_n.is_none()
             && self.poly_ps.is_empty()
             && self.ekey_to_holder.is_empty()
+    }
     fn clear(&mut self) {
         self.solution_storage.clear();
         self.poly_n.take();
         self.poly_ps.clear();
         self.ekey_to_holder.clear();
+    }
+}
 impl Into<PolyP> for MonoP {
     fn into(self) -> PolyP {
         PolyP {
             complete: Default::default(),
             incomplete: hashmap! {
                 Predicate::new_trivial() =>
                 BranchP {
                     state: self.state,
                     inbox: Default::default(),
+                }
             },
             ekeys: self.ekeys,
+        }
+    }
+}
 impl From<EndpointErr> for SyncErr {
     fn from(e: EndpointErr) -> SyncErr {
         SyncErr::EndpointErr(e)
+    }
+}
 impl MonoContext for MonoPContext<'_> {
     type D = ProtocolD;
     type S = ProtocolS;
     fn new_component(&mut self, moved_ekeys: HashSet<Key>, init_state: Self::S) {
         lockprintln!(
             "{:?}: !! MonoContext callback to new_component with ekeys {:?}!",
             self.inner.channel_id_stream.controller_id,
             &moved_ekeys,
         );
         if moved_ekeys.is_subset(self.ekeys) {
             self.ekeys.retain(|x| !moved_ekeys.contains(x));
             self.inner.mono_ps.push(MonoP { state: init_state, ekeys: moved_ekeys });
         } else {
             panic!("MachineP attempting to move alien ekey!");
+        }
+    }
     fn new_channel(&mut self) -> [Key; 2] {
         let [a, b] = Endpoint::new_memory_pair();
         let channel_id = self.inner.channel_id_stream.next();
         let kp = self.inner.endpoint_exts.alloc(EndpointExt {
             info: EndpointInfo { polarity: Putter, channel_id },
             endpoint: a,
         });
         let kg = self.inner.endpoint_exts.alloc(EndpointExt {
             info: EndpointInfo { polarity: Putter, channel_id },
             endpoint: b,
         });
         lockprintln!(
             "{:?}: !! MonoContext callback to new_channel. returning ekeys {:?}!",
             self.inner.channel_id_stream.controller_id,
             [kp, kg],
         );
         [kp, kg]
+    }
     fn new_random(&self) -> u64 {
         type Bytes8 = [u8; std::mem::size_of::<u64>()];
         let mut bytes = Bytes8::default();
         getrandom::getrandom(&mut bytes).unwrap();
         unsafe { std::mem::transmute::<Bytes8, _>(bytes) }
         let val = unsafe { std::mem::transmute::<Bytes8, _>(bytes) };
         lockprintln!(
             "{:?}: !! MonoContext callback to new_random. returning val {:?}!",
             self.inner.channel_id_stream.controller_id,
             val,
         );
         val
+    }
+}
 impl SolutionStorage {
     fn is_clear(&self) -> bool {
         self.subtree_id_to_index.is_empty()
             && self.subtree_solutions.is_empty()
             && self.old_local.is_empty()
             && self.new_local.is_empty()
+    }
     fn clear(&mut self) {
         self.subtree_id_to_index.clear();
         self.subtree_solutions.clear();
         self.old_local.clear();
         self.new_local.clear();
+    }
     pub(crate) fn reset(&mut self, subtree_ids: impl Iterator<Item = SubtreeId>) {
         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 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,
         subtree_id: SubtreeId,
         predicate: 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(predicate, set_visitor, old_local, new_local);
+        }
+    }
     fn elaborate_into_new_local_rec<'a, 'b>(
         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(
                         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
                 new_local.insert(partial);
+            }
+        }
+    }
+}
 impl PolyContext for BranchPContext<'_, '_> {
     type D = ProtocolD;
     fn is_firing(&self, ekey: Key) -> Option<bool> {
         assert!(self.ekeys.contains(&ekey));
         let channel_id = self.m_ctx.inner.endpoint_exts.get(ekey).unwrap().info.channel_id;
         self.predicate.query(channel_id)
         let val = self.predicate.query(channel_id);
         lockprintln!(
             "{:?}: !! PolyContext callback to is_firing by {:?}! returning {:?}",
             self.m_ctx.inner.channel_id_stream.controller_id,
             self.m_ctx.my_subtree_id,
             val,
         );
         val
+    }
     fn read_msg(&self, ekey: Key) -> Option<&Payload> {
         assert!(self.ekeys.contains(&ekey));
         self.inbox.get(&ekey)
         let val = self.inbox.get(&ekey);
         lockprintln!(
             "{:?}: !! PolyContext callback to read_msg by {:?}! returning {:?}",
             self.m_ctx.inner.channel_id_stream.controller_id,
             self.m_ctx.my_subtree_id,
             val,
         );
         val
+    }
+}

0 comments (0 inline, 0 general)