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

Changeset - 702ecfb0e5e9

Parent rev.

Child rev.

[Not reviewed]

0 3 0

Christopher Esterhuyse - 5 years ago 2020-07-07 11:13:20
christopher.esterhuyse@gmail.com

distinguished subtree_id (identifying a connector's child in the consensus tree) from Route, as with the addition of udp endpoints, they no longer coincide

3 files changed with 38 insertions and 22 deletions:

src/runtime/communication.rs

src/runtime/mod.rs

src/runtime/setup.rs

0 comments (0 inline, 0 general)

src/runtime/communication.rs

➞

Show inline comments

@@ @@ -16,27 +16,27 @@ struct BranchingNative { @@
     branches: HashMap<Predicate, NativeBranch>,
+}
 #[derive(Clone, Debug)]
 struct NativeBranch {
     index: usize,
     gotten: HashMap<PortId, Payload>,
     to_get: HashSet<PortId>, // native branch is ended iff to_get.is_empty()
+}
 #[derive(Debug)]
 struct SolutionStorage {
     old_local: HashSet<Predicate>,
     new_local: HashSet<Predicate>,
-    // this pair acts as Route -> HashSet<Predicate> which is friendlier to iteration
+    // this pair acts as SubtreeId -> HashSet<Predicate> which is friendlier to iteration
     subtree_solutions: Vec<HashSet<Predicate>>,
-    subtree_id_to_index: HashMap<Route, usize>,
+    subtree_id_to_index: HashMap<SubtreeId, usize>,
+}
 #[derive(Debug)]
 struct BranchingProtoComponent {
     ports: HashSet<PortId>,
     branches: HashMap<Predicate, ProtoComponentBranch>,
+}
 #[derive(Debug, Clone)]
 struct ProtoComponentBranch {
     inbox: HashMap<PortId, Payload>,
     state: ComponentState,
     untaken_choice: Option<u16>,
     ended: bool,
@@ @@ -211,32 +211,36 @@ impl Connector { @@
+                }
+            }
+        }
         log!(
             cu.logger,
             "All {} proto components are now done with Nonsync phase",
             branching_proto_components.len(),
         );
         // Create temp structures needed for the synchronous phase of the round
         let mut rctx = RoundCtx {
             solution_storage: {
-                let n = std::iter::once(Route::LocalComponent(ComponentId::Native));
+                let n = std::iter::once(SubtreeId::LocalComponent(ComponentId::Native));
                 let c = cu
                     .proto_components
                     .keys()
                     .map(|&id| Route::LocalComponent(ComponentId::Proto(id)));
                 let e = comm.neighborhood.children.iter().map(|&index| Route::Endpoint { index });
                 let route_iter = n.chain(c).chain(e);
                 SolutionStorage::new(route_iter)
                     .map(|&id| SubtreeId::LocalComponent(ComponentId::Proto(id)));
                 let e = comm
                     .neighborhood
                     .children
                     .iter()
                     .map(|&index| SubtreeId::NetEndpoint { index });
                 let subtree_id_iter = n.chain(c).chain(e);
                 SolutionStorage::new(subtree_id_iter)
             },
             spec_var_stream: cu.id_manager.new_spec_var_stream(),
             getter_buffer: Default::default(),
             deadline: timeout.map(|to| Instant::now() + to),
         };
         log!(cu.logger, "Round context structure initialized");
         // Explore all native branches eagerly. Find solutions, buffer messages, etc.
         log!(
             cu.logger,
             "Translating {} native batches into branches...",
             comm.native_batches.len()
@@ @@ -276,25 +280,25 @@ impl Connector { @@
                 rctx.getter_buffer.putter_add(cu, putter, msg);
+            }
             if to_get.is_empty() {
                 // this branch is immediately ready to be part of a solution
                 log!(
                     cu.logger,
                     "Native submitting solution for batch {} with {:?}",
                     index,
                     &predicate
                 );
                 rctx.solution_storage.submit_and_digest_subtree_solution(
                     &mut *cu.logger,
-                    Route::LocalComponent(ComponentId::Native),
+                    SubtreeId::LocalComponent(ComponentId::Native),
                     predicate.clone(),
                 );
+            }
             let branch = NativeBranch { index, gotten: Default::default(), to_get };
             if let Some(_) = branching_native.branches.insert(predicate, branch) {
                 // thanks to the native_branch_spec_var, each batch has a distinct predicate
                 unreachable!()
+            }
+        }
         // restore the invariant: !native_batches.is_empty()
         comm.native_batches.push(Default::default());
         // Call to another big method; keep running this round until a distributed decision is reached
@@ @@ -422,25 +426,29 @@ impl Connector { @@
                 assert!(cu.port_info.polarities.get(&getter).copied() == Some(Getter));
                 let route = cu.port_info.routes.get(&getter);
                 log!(cu.logger, "Routing msg {:?} to {:?}", &send_payload_msg, &route);
                 match route {
                     None => {
                         log!(
                             cu.logger,
                             "Delivery to getter {:?} msg {:?} failed. Physical route unmapped!",
                             getter,
                             &send_payload_msg
                         );
+                    }
                     Some(Route::Endpoint { index }) => {
                     Some(Route::UdpEndpoint { index }) => {
                         // TODO UDP RECV
                         todo!()
+                    }
                     Some(Route::NetEndpoint { index }) => {
                         let msg = Msg::CommMsg(CommMsg {
                             round_index: comm.round_index,
                             contents: CommMsgContents::SendPayload(send_payload_msg),
                         });
                         comm.endpoint_manager.send_to_comms(*index, &msg)?;
+                    }
                     Some(Route::LocalComponent(ComponentId::Native)) => branching_native.feed_msg(
                         cu,
                         &mut rctx.solution_storage,
                         getter,
                         &send_payload_msg,
                     ),
@@ @@ -575,28 +583,29 @@ impl Connector { @@
                             "Msg routed to getter port {:?}. Buffer for recv loop",
                             getter,
                         );
                         rctx.getter_buffer.getter_add(getter, send_payload_msg);
+                    }
                     CommMsgContents::Suggest { suggestion } => {
                         // only accept this control msg through a child endpoint
                         if comm.neighborhood.children.contains(&endpoint_index) {
                             match suggestion {
                                 Decision::Success(predicate) => {
                                     // child solution contributes to local solution
                                     log!(cu.logger, "Child provided solution {:?}", &predicate);
                                     let route = Route::Endpoint { index: endpoint_index };
                                     let subtree_id =
                                         SubtreeId::NetEndpoint { index: endpoint_index };
                                     rctx.solution_storage.submit_and_digest_subtree_solution(
                                         &mut *cu.logger,
-                                        route,
+                                        subtree_id,
                                         predicate,
                                     );
+                                }
                                 Decision::Failure => {
                                     match comm.neighborhood.parent {
                                         None => {
                                             log!(cu.logger, "I decide on my child's failure");
                                             break 'undecided Ok(Decision::Failure);
+                                        }
                                         Some(parent) => {
                                             log!(cu.logger, "Forwarding failure through my parent endpoint {:?}", parent);
                                             if already_requested_failure.replace_with_true() {
@@ @@ -668,28 +677,28 @@ impl BranchingNative { @@
             let var = cu.port_info.spec_var_for(getter);
             let mut feed_branch = |branch: &mut NativeBranch, predicate: &Predicate| {
                 let was = branch.gotten.insert(getter, send_payload_msg.payload.clone());
                 assert!(was.is_none());
                 branch.to_get.remove(&getter);
                 if branch.to_get.is_empty() {
                     log!(
                         cu.logger,
                         "new native solution with {:?} (to_get.is_empty()) with gotten {:?}",
                         &predicate,
                         &branch.gotten
                     );
-                    let route = Route::LocalComponent(ComponentId::Native);
+                    let subtree_id = SubtreeId::LocalComponent(ComponentId::Native);
                     solution_storage.submit_and_digest_subtree_solution(
                         &mut *cu.logger,
-                        route,
+                        subtree_id,
                         predicate.clone(),
                     );
+                }
             };
             if predicate.query(var) != Some(SpecVal::FIRING) {
                 // optimization. Don't bother trying this branch
                 log!(
                     cu.logger,
                     "skipping branch with {:?} that doesn't want the message (fastpath)",
                     &predicate
                 );
                 finished.insert(predicate, branch);
@@ @@ -795,27 +804,28 @@ impl BranchingProtoComponent { @@
+                }
                 B::Inconsistent => {
                     // branch is inconsistent. throw it away
                     drop((predicate, branch));
+                }
                 B::SyncBlockEnd => {
                     // make concrete all variables
                     for &port in ports.iter() {
                         let var = cu.port_info.spec_var_for(port);
                         predicate.assigned.entry(var).or_insert(SpecVal::SILENT);
+                    }
                     // submit solution for this component
                     let subtree_id = SubtreeId::LocalComponent(ComponentId::Proto(proto_component_id));
                     rctx.solution_storage.submit_and_digest_subtree_solution(
                         &mut *cu.logger,
-                        Route::LocalComponent(ComponentId::Proto(proto_component_id)),
+                        subtree_id,
                         predicate.clone(),
                     );
                     branch.ended = true;
                     // move to "blocked"
                     drainer.add_output(predicate, branch);
+                }
                 B::CouldntReadMsg(port) => {
                     // move to "blocked"
                     assert!(!branch.inbox.contains_key(&port));
                     drainer.add_output(predicate, branch);
+                }
                 B::CouldntCheckFiring(port) => {
@@ @@ -936,71 +946,71 @@ impl BranchingProtoComponent { @@
+    }
     fn initial(ProtoComponent { state, ports }: ProtoComponent) -> Self {
         let branch = ProtoComponentBranch {
             inbox: Default::default(),
             state,
             ended: false,
             untaken_choice: None,
         };
         Self { ports, branches: hashmap! { Predicate::default() => branch  } }
+    }
+}
 impl SolutionStorage {
     fn new(routes: impl Iterator<Item = Route>) -> Self {
         let mut subtree_id_to_index: HashMap<Route, usize> = Default::default();
     fn new(subtree_ids: impl Iterator<Item = SubtreeId>) -> Self {
         let mut subtree_id_to_index: HashMap<SubtreeId, usize> = Default::default();
         let mut subtree_solutions = vec![];
         for key in routes {
             subtree_id_to_index.insert(key, subtree_solutions.len());
         for id in subtree_ids {
             subtree_id_to_index.insert(id, subtree_solutions.len());
             subtree_solutions.push(Default::default())
+        }
         Self {
             subtree_solutions,
             subtree_id_to_index,
             old_local: Default::default(),
             new_local: Default::default(),
+        }
+    }
     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();
+    }
-    fn reset(&mut self, subtree_ids: impl Iterator<Item = Route>) {
+    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 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,
+        subtree_id: SubtreeId,
         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(

src/runtime/mod.rs

➞

Show inline comments

@@ @@ -65,25 +65,31 @@ struct RoundOk { @@
 struct VecSet<T: std::cmp::Ord> {
     // invariant: ordered, deduplicated
     vec: Vec<T>,
+}
 #[derive(Debug, Clone, Copy, Eq, PartialEq, Hash, serde::Serialize, serde::Deserialize)]
 enum ComponentId {
     Native,
     Proto(ProtoComponentId),
+}
 #[derive(Debug, Clone, Copy, Eq, PartialEq, Hash, serde::Serialize, serde::Deserialize)]
 enum Route {
     LocalComponent(ComponentId),
     Endpoint { index: usize },
     NetEndpoint { index: usize },
     UdpEndpoint { index: usize },
+}
 #[derive(Debug, Clone, Copy, Eq, PartialEq, Hash, serde::Serialize, serde::Deserialize)]
 enum SubtreeId {
     LocalComponent(ComponentId),
     NetEndpoint { index: usize },
+}
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 struct MyPortInfo {
     polarity: Polarity,
     port: PortId,
+}
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 enum Decision {
     Failure,
     Success(Predicate),
+}
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]

src/runtime/setup.rs

➞

Show inline comments

@@ @@ -374,25 +374,25 @@ fn new_endpoint_manager( @@
                                     if let Some(route) = port_info.routes.get(&peer_info.port) {
                                         // check just for logging purposes
                                         log!(
                                             logger,
                                             "Special case! Route to peer {:?} already known to be {:?}. Leave untouched",
                                             peer_info.port,
                                             route
                                         );
+                                    }
                                     port_info
                                         .routes
                                         .entry(peer_info.port)
-                                        .or_insert(Route::Endpoint { index });
+                                        .or_insert(Route::NetEndpoint { index });
+                                }
                                 Some(inappropriate_msg) => {
                                     log!(
                                         logger,
                                         "delaying msg {:?} during channel setup phase",
                                         inappropriate_msg
                                     );
                                     delayed_messages.push((index, inappropriate_msg));
+                                }
+                            }
+                        }
                         // is the setup for this net_endpoint now complete?

0 comments (0 inline, 0 general)