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

@@ @@ -22,15 +22,15 @@ struct NativeBranch { @@
     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>,
+}
@@ @@ -217,20 +217,24 @@ impl Connector { @@
             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");
@@ @@ -282,13 +286,13 @@ impl Connector { @@
                     "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
@@ @@ -428,13 +432,17 @@ impl Connector { @@
                             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)?;
+                    }
@@ @@ -581,16 +589,17 @@ impl Connector { @@
                         // 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 => {
@@ @@ -674,16 +683,16 @@ impl BranchingNative { @@
                     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
@@ @@ -801,15 +810,16 @@ impl BranchingProtoComponent { @@
                     // 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);
+                }
@@ @@ -942,17 +952,17 @@ impl BranchingProtoComponent { @@
             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(),
@@ @@ -968,13 +978,13 @@ impl SolutionStorage { @@
     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());
@@ @@ -988,13 +998,13 @@ impl SolutionStorage { @@
             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();

src/runtime/mod.rs

➞

Show inline comments

@@ @@ -71,13 +71,19 @@ 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,
+}

src/runtime/setup.rs

➞

Show inline comments

@@ @@ -380,13 +380,13 @@ fn new_endpoint_manager( @@
                                             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

0 comments (0 inline, 0 general)