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

Changeset - f91ffb2597cc

Parent rev.

Child rev.

[Not reviewed]

0 2 0

Christopher Esterhuyse - 5 years ago 2020-04-20 13:28:53
christopher.esterhuyse@gmail.com

had my predicates backwards. fixed bug in Native's recv_react proc

2 files changed with 28 insertions and 24 deletions:

src/runtime/actors.rs

src/runtime/communication.rs

0 comments (0 inline, 0 general)

src/runtime/actors.rs

➞

Show inline comments

@@ @@ -230,217 +230,213 @@ 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)| {
                     use CommonSatResult as Csr;
                     match old_predicate.common_satisfier(&payload_predicate) {
                         Csr::FormerNotLatter | Csr::Equivalent => {
                             log!(
                                 &mut m_ctx.inner.logger,
                                 "... poly_recv_run This branch is compatible unaltered! branch pred: {:?}",
                                 &old_predicate
                             );
                             // old_predicate COVERS the assumptions of payload_predicate
                             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,
                                 &old_predicate,
                                 &new,
                             );
                             // payload_predicate has new assumptions. FORK!
                             let mut payload_branch = branch.clone();
                             if let Some(prev_payload) = payload_branch.inbox.get(&ekey) {
                                 // 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);
                             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,
                                 &payload_predicate,
                             );
                             // payload_predicate has new assumptions. FORK!
                             let mut payload_branch = branch.clone();
                             if let Some(prev_payload) = payload_branch.inbox.get(&ekey) {
                                 // 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.clone(), branch);
                             if payload_branch.blocking_on == Some(ekey) {
                                 // run the fork
                                 payload_branch.blocking_on = None;
-                                Some((old_predicate, payload_branch))
+                                Some((payload_predicate.clone(), payload_branch))
                             } else {
                                 // don't bother running. its awaiting something else
-                                incomplete2.insert(old_predicate, payload_branch);
+                                incomplete2.insert(payload_predicate.clone(), payload_branch);
                                 None
+                            }
+                        }
                         Csr::Nonexistant => {
                             log!(
                                 &mut m_ctx.inner.logger,
                                 "... poly_recv_run SKIPPING because branchpred={:?}. payloadpred={:?}",
                                 &old_predicate,
                                 &payload_predicate,
                             );
                             // predicates contradict
                             incomplete2.insert(old_predicate, branch);
                             None
+                        }
+                    }
                 })
                 .collect();
             std::mem::swap(&mut self.incomplete, &mut incomplete2);
             to_run
         };
         log!(
             &mut m_ctx.inner.logger,
             "... DONE FEEDING BRANCHES. {} branches to run!",
             to_run.len(),
         );
         self.poly_run_these_branches(m_ctx, protocol_description, to_run)
+    }
     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,
         ekey: Key,
         logger: &mut String,
         payload: Payload,
         payload_predicate: Predicate,
         solution_storage: &mut SolutionStorage,
     ) {
         let mut branches2: HashMap<_, _> = Default::default();
         for (old_predicate, mut branch) in self.branches.drain() {
             use CommonSatResult as Csr;
             let case = old_predicate.common_satisfier(&payload_predicate);
             let mut report_if_solution =
                 |branch: &BranchN, pred: &Predicate, logger: &mut String| {
                     if branch.to_get.is_empty() {
                         log!(logger, "Native reporting solution with inbox {:#?}", &branch.gotten);
                         solution_storage.submit_and_digest_subtree_solution(
                             logger,
                             SubtreeId::PolyN,
                             pred.clone(),
                         );
+                    }
                 };
             log!(
                 logger,
                 "Feeding msg {:?} {:?} to native branch with pred {:?}. Predicate case {:?}",
                 &payload_predicate,
                 &payload,
                 &old_predicate,
                 &case
             );
             match case {
                 Csr::Nonexistant => { /* skip branch */ }
-                Csr::FormerNotLatter | Csr::Equivalent => {
+                Csr::LatterNotFormer | Csr::Equivalent => {
                     // Feed the message to this branch in-place. no need to modify pred.
                     if branch.to_get.remove(&ekey) {
                         branch.gotten.insert(ekey, payload.clone());
                         report_if_solution(&branch, &old_predicate, logger);
+                    }
+                }
-                Csr::LatterNotFormer => {
+                Csr::FormerNotLatter => {
                     // create a new branch with the payload_predicate.
                     let mut forked = branch.clone();
                     if forked.to_get.remove(&ekey) {
                         forked.gotten.insert(ekey, payload.clone());
                         report_if_solution(&forked, &payload_predicate, logger);
                         branches2.insert(payload_predicate.clone(), forked);
+                    }
+                }
                 Csr::New(new) => {
                     // create a new branch with the newly-created predicate
                     let mut forked = branch.clone();
                     if forked.to_get.remove(&ekey) {
                         forked.gotten.insert(ekey, payload.clone());
                         report_if_solution(&forked, &new, logger);
                         branches2.insert(new.clone(), forked);
+                    }
+                }
+            }
-            // unlike PolyP machines, Native branches do not become inconsistent
+            // unlike PolyP machines, Native branches do NOT become inconsistent
             branches2.insert(old_predicate, branch);
+        }
         log!(
             logger,
             "Native now has {} branches with predicates: {:?}",
             branches2.len(),
             branches2.keys().collect::<Vec<_>>()
         );
         log!(logger, "Native now has branches {:#?}", &branches2);
         std::mem::swap(&mut branches2, &mut self.branches);
+    }
     pub fn choose_mono(&self, decision: &Predicate) -> Option<MonoN> {
         self.branches
             .iter()
             .find(|(p, branch)| branch.to_get.is_empty() && decision.satisfies(p))
             .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 {
     fn end_round_with_decision(&mut self, decision: Decision) -> Result<(), SyncErr> {
         log!(&mut self.inner.logger, "ENDING ROUND WITH DECISION! {:?}", &decision);
         let ret = match &decision {
             Decision::Success(predicate) => {
                 // overwrite MonoN/P
                 self.inner.mono_n =
                     self.ephemeral.poly_n.take().unwrap().choose_mono(predicate).unwrap();
                 self.inner.mono_n = {
                     let poly_n = self.ephemeral.poly_n.take().unwrap();
                     poly_n.choose_mono(predicate).unwrap_or_else(|| {
                         panic!(
                             "Ending round with decision pred {:#?} but poly_n has branches {:#?}. My log is... {}",
                             &predicate, &poly_n.branches, &self.inner.logger
                         );
                     })
                 };
                 self.inner.mono_ps.clear();
                 self.inner.mono_ps.extend(
                     self.ephemeral
                         .poly_ps
                         .drain(..)
                         .map(|poly_p| poly_p.choose_mono(predicate).unwrap()),
                 );
                 Ok(())
+            }
             Decision::Failure => Err(SyncErr::Timeout),
         };
         let announcement = CommMsgContents::Announce { 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 {:?}",
                 &announcement,
                 child_ekey
             );
             self.inner
                 .endpoint_exts
                 .get_mut(child_ekey)
                 .expect("eefef")
                 .endpoint
                 .send(announcement.clone())?;
+        }
         self.inner.round_index += 1;
         self.ephemeral.clear();
         ret
+    }
     // Drain self.ephemeral.solution_storage and handle the new locals. Return decision if one is found
     fn handle_locals_maybe_decide(&mut self) -> Result<bool, SyncErr> {
         if let Some(parent_ekey) = self.inner.family.parent_ekey {
             // I have a parent -> I'm not the leader
             let parent_endpoint =
                 &mut self.inner.endpoint_exts.get_mut(parent_ekey).expect("huu").endpoint;
             for partial_oracle in self.ephemeral.solution_storage.iter_new_local_make_old() {
                 let msg =
                     CommMsgContents::Elaborate { partial_oracle }.into_msg(self.inner.round_index);
                 log!(&mut self.inner.logger, "Sending {:?} to parent {:?}", &msg, parent_ekey);
                 parent_endpoint.send(msg)?;
+            }
             Ok(false)
         } else {
             // I have no parent -> I'm the leader
             assert!(self.inner.family.parent_ekey.is_none());
             let maybe_predicate = self.ephemeral.solution_storage.iter_new_local_make_old().next();
             Ok(if let Some(predicate) = maybe_predicate {
                 let decision = Decision::Success(predicate);
                 log!(&mut self.inner.logger, "DECIDE ON {:?} AS LEADER!", &decision);
                 self.end_round_with_decision(decision)?;
                 true
             } else {
                 false
             })
+        }
+    }
     fn kick_off_native(
         &mut self,
         sync_batches: impl Iterator<Item = SyncBatch>,
     ) -> Result<PolyN, EndpointErr> {
         let MonoN { ekeys, .. } = self.inner.mono_n.clone();
         let Self { inner: ControllerInner { endpoint_exts, round_index, .. }, .. } = self;
         let mut branches = HashMap::<_, _>::default();
         for (sync_batch_index, SyncBatch { puts, gets }) in sync_batches.enumerate() {
             let ekey_to_channel_id = |ekey| endpoint_exts.get(ekey).unwrap().info.channel_id;
             let all_ekeys = ekeys.iter().copied();
             let all_channel_ids = all_ekeys.map(ekey_to_channel_id);
             let mut predicate = Predicate::new_trivial();
             // assign TRUE for puts and gets
             let true_ekeys = puts.keys().chain(gets.iter()).copied();
             let true_channel_ids = true_ekeys.clone().map(ekey_to_channel_id);
             predicate.batch_assign_nones(true_channel_ids, true);
             // assign FALSE for all in interface not assigned true
             predicate.batch_assign_nones(all_channel_ids.clone(), false);
             if branches.contains_key(&predicate) {
                 // TODO what do I do with redundant predicates?
                 unimplemented!(
                     "Having multiple batches with the same
                     predicate requires the support of oracle boolean variables"
+                )
+            }
             let branch = BranchN { to_get: gets, gotten: Default::default(), sync_batch_index };
             for (ekey, payload) in puts {
                 log!(
                     &mut self.inner.logger,
                     "... ... Initial native put msg {:?} pred {:?} batch {:?}",
                     &payload,
                     &predicate,
                     sync_batch_index,
@@ @@ -326,239 +333,240 @@ impl Controller { @@
                                     .send(announcement.clone())?;
                                 continue; // poll some more
+                            }
+                        }
+                    }
                 },
             };
             log!(&mut self.inner.logger, "::: message {:?}...", &received);
             let current_content = match received.msg {
                 Msg::SetupMsg(s) => {
                     // This occurs in the event the connector was malformed during connect()
                     println!("WASNT EXPECTING {:?}", s);
                     return Err(SyncErr::UnexpectedSetupMsg);
+                }
                 Msg::CommMsg(CommMsg { round_index, .. })
                     if round_index < self.inner.round_index =>
+                {
                     // Old message! Can safely discard
                     log!(&mut self.inner.logger, "...and its OLD! :(");
                     drop(received);
                     continue 'recv_loop;
+                }
                 Msg::CommMsg(CommMsg { round_index, .. })
                     if round_index > self.inner.round_index =>
+                {
                     // Message from a next round. Keep for later!
                     log!(&mut self.inner.logger, "... DELAY! :(");
                     self.delay(received);
                     continue 'recv_loop;
+                }
                 Msg::CommMsg(CommMsg { contents, round_index }) => {
                     log!(
                         &mut self.inner.logger,
                         "... its a round-appropriate CommMsg with key {:?}",
                         received.recipient
                     );
                     assert_eq!(round_index, self.inner.round_index);
                     contents
+                }
             };
             match current_content {
                 CommMsgContents::Failure => match self.inner.family.parent_ekey {
                     Some(parent_ekey) => {
                         let announcement = Msg::CommMsg(CommMsg {
                             round_index: self.inner.round_index,
                             contents: CommMsgContents::Failure,
                         });
                         log!(
                             &mut self.inner.logger,
                             "Forwarding {:?} to parent with ekey {:?}",
                             &announcement,
                             parent_ekey
                         );
                         self.inner
                             .endpoint_exts
                             .get_mut(parent_ekey)
                             .expect("ss")
                             .endpoint
                             .send(announcement.clone())?;
+                    }
                     None => return self.end_round_with_decision(Decision::Failure),
                 },
                 CommMsgContents::Elaborate { partial_oracle } => {
                     // Child controller submitted a subtree solution.
                     if !self.inner.family.children_ekeys.contains(&received.recipient) {
                         return Err(SyncErr::ElaborateFromNonChild);
+                    }
                     let subtree_id = SubtreeId::ChildController { ekey: received.recipient };
                     log!(
                         &mut self.inner.logger,
                         "Received elaboration from child for subtree {:?}: {:?}",
                         subtree_id,
                         &partial_oracle
                     );
                     self.ephemeral.solution_storage.submit_and_digest_subtree_solution(
                         &mut self.inner.logger,
                         subtree_id,
                         partial_oracle,
                     );
                     if self.handle_locals_maybe_decide()? {
                         return Ok(());
+                    }
+                }
                 CommMsgContents::Announce { decision } => {
                     if self.inner.family.parent_ekey != Some(received.recipient) {
                         return Err(SyncErr::AnnounceFromNonParent);
+                    }
                     log!(
                         &mut self.inner.logger,
                         "Received ANNOUNCEMENT from from parent {:?}: {:?}",
                         received.recipient,
                         &decision
                     );
                     return self.end_round_with_decision(decision);
+                }
                 CommMsgContents::SendPayload { payload_predicate, payload } => {
                     // check that we expect to be able to receive payloads from this sender
                     assert_eq!(
                         Getter,
                         self.inner.endpoint_exts.get(received.recipient).unwrap().info.polarity
                     );
                     // message for some actor. Feed it to the appropriate actor
                     // and then give them another chance to run.
                     let subtree_id = ekey_to_holder.get(&received.recipient);
                     log!(
                         &mut self.inner.logger,
                         "Received SendPayload for subtree {:?} with pred {:?} and payload {:?}",
                         subtree_id,
                         &payload_predicate,
                         &payload
                     );
                     let channel_id = self
                         .inner
                         .endpoint_exts
                         .get(received.recipient)
                         .expect("UEHFU")
                         .info
                         .channel_id;
                     if payload_predicate.query(channel_id) != Some(true) {
                         // sender didn't preserve the invariant
                         return Err(SyncErr::PayloadPremiseExcludesTheChannel(channel_id));
+                    }
                     match subtree_id {
                         None => {
                             // this happens when a message is sent to a component that has exited.
                             // It's safe to drop this message;
                             // The sender branch will certainly not be part of the solution
+                        }
                         Some(PolyId::N) => {
                             // Message for NativeMachine
                             self.ephemeral.poly_n.as_mut().unwrap().sync_recv(
                                 received.recipient,
                                 &mut self.inner.logger,
                                 payload,
                                 payload_predicate,
                                 &mut self.ephemeral.solution_storage,
                             );
                             if self.handle_locals_maybe_decide()? {
                                 return Ok(());
+                            }
+                        }
                         Some(PolyId::P { index }) => {
                             // Message for protocol actor
                             let channel_id = self
                                 .inner
                                 .endpoint_exts
                                 .get(received.recipient)
                                 .expect("UEHFU")
                                 .info
                                 .channel_id;
                             if payload_predicate.query(channel_id) != Some(true) {
                                 // sender didn't preserve the invariant
                                 return Err(SyncErr::PayloadPremiseExcludesTheChannel(channel_id));
+                            }
                             let poly_p = &mut self.ephemeral.poly_ps[*index];
                             let m_ctx = PolyPContext {
                                 my_subtree_id: SubtreeId::PolyP { index: *index },
                                 inner: &mut self.inner,
                                 solution_storage: &mut self.ephemeral.solution_storage,
                             };
                             use SyncRunResult as Srr;
                             let blocker = poly_p.poly_recv_run(
                                 m_ctx,
                                 &self.protocol_description,
                                 received.recipient,
                                 payload_predicate,
                                 payload,
                             )?;
                             log!(
                                 &mut self.inner.logger,
                                 "... Fed the msg to PolyP {:?} and ran it to blocker {:?}",
                                 subtree_id,
                                 blocker
                             );
                             match blocker {
                                 Srr::NoBranches => return Err(SyncErr::Inconsistent),
                                 Srr::BlockingForRecv | Srr::AllBranchesComplete => {
+                                    {
                                         let peeked = self
                                             .ephemeral
                                             .solution_storage
                                             .peek_new_locals()
                                             .collect::<Vec<_>>();
                                         log!(
                                             &mut self.inner.logger,
                                             "Got {} new controller-local solutions from RECV: {:?}",
                                             peeked.len(),
                                             peeked
                                         );
+                                    }
                                     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.mono_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(),
                     outbox: Default::default(),
                     blocking_on: None,
+                }
             },
             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) {
         log!(
             &mut self.inner.logger,
             "!! MonoContext callback to new_component with ekeys {:?}!",
             &moved_ekeys,

0 comments (0 inline, 0 general)