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

Changeset - 20e64ced68b1

Parent rev.

Child rev.

[Not reviewed]

0 1 0

Christopher Esterhuyse - 5 years ago 2020-07-21 12:18:17
christopher.esterhuyse@gmail.com

reversed a predicate simplification; turns out to break the mutual inconsistency property of native branches in some corner cases where 3+ branches have colliding predicates

1 file changed with 19 insertions and 41 deletions:

src/runtime/communication.rs

0 comments (0 inline, 0 general)

src/runtime/communication.rs

➞

Show inline comments

@@ @@ -14,25 +14,25 @@ struct GetterBuffer { @@
+}
 struct RoundCtx {
     solution_storage: SolutionStorage,
     spec_var_stream: SpecVarStream,
     getter_buffer: GetterBuffer,
     deadline: Option<Instant>,
+}
 struct BranchingNative {
     branches: HashMap<Predicate, NativeBranch>,
+}
 #[derive(Clone, Debug)]
 struct NativeBranch {
-    batch_index: usize,
+    index: usize,
     gotten: HashMap<PortId, Payload>,
     to_get: HashSet<PortId>,
+}
 #[derive(Debug)]
 struct SolutionStorage {
     old_local: HashSet<Predicate>,
     new_local: HashSet<Predicate>,
     // this pair acts as SubtreeId -> HashSet<Predicate> which is friendlier to iteration
     subtree_solutions: Vec<HashSet<Predicate>>,
     subtree_id_to_index: HashMap<SubtreeId, usize>,
+}
 #[derive(Debug)]
@@ @@ -294,103 +294,81 @@ impl Connector { @@
             spec_var_stream: cu.inner.id_manager.new_spec_var_stream(),
             getter_buffer: Default::default(),
             deadline: timeout.map(|to| Instant::now() + to),
         };
         log!(cu.inner.logger, "Round context structure initialized");
         // Explore all native branches eagerly. Find solutions, buffer messages, etc.
         log!(
             cu.inner.logger,
             "Translating {} native batches into branches...",
             comm.native_batches.len()
         );
         let mut native_spec_assign_stream = None;
         let native_spec_var = rctx.spec_var_stream.next();
         log!(cu.inner.logger, "Native branch spec var is {:?}", native_spec_var);
         let mut branching_native = BranchingNative { branches: Default::default() };
-        'native_branches: for ((native_branch, batch_index), branch_spec_val) in
+        'native_branches: for ((native_branch, index), branch_spec_val) in
             comm.native_batches.drain(..).zip(0..).zip(SpecVal::iter_domain())
+        {
             let NativeBatch { to_get, to_put } = native_branch;
             let predicate = {
                 let mut predicate = Predicate::default();
                 // assign trues for ports that fire
                 let firing_ports: HashSet<PortId> =
                     to_get.iter().chain(to_put.keys()).copied().collect();
                 for &port in to_get.iter().chain(to_put.keys()) {
                     let var = cu.inner.port_info.spec_var_for(port);
                     predicate.assigned.insert(var, SpecVal::FIRING);
+                }
                 // assign falses for all silent (not firing) ports
                 for &port in cu.inner.native_ports.difference(&firing_ports) {
                     let var = cu.inner.port_info.spec_var_for(port);
                     if let Some(SpecVal::FIRING) = predicate.assigned.insert(var, SpecVal::SILENT) {
-                        log!(cu.inner.logger, "Native branch index={} contains internal inconsistency wrt. {:?}. Skipping", batch_index, var);
+                        log!(cu.inner.logger, "Native branch index={} contains internal inconsistency wrt. {:?}. Skipping", index, var);
                         continue 'native_branches;
+                    }
+                }
                 predicate
                 // this branch is consistent. distinguish it with a unique var:val mapping and proceed
                 predicate.inserted(native_spec_var, branch_spec_val)
             };
             log!(
                 cu.inner.logger,
                 "Native batch_index={:?} has consistent {:?}",
                 batch_index,
                 "Native branch index={:?} has consistent {:?}",
                 index,
                 &predicate
             );
             // send all outgoing messages (by buffering them)
             for (putter, payload) in to_put {
                 let msg = SendPayloadMsg { predicate: predicate.clone(), payload };
-                log!(cu.inner.logger, "Native branch {} sending msg {:?}", batch_index, &msg);
+                log!(cu.inner.logger, "Native branch {} sending msg {:?}", index, &msg);
                 rctx.getter_buffer.putter_add(cu, putter, msg);
+            }
             let branch = NativeBranch { batch_index, gotten: Default::default(), to_get };
             if let Some(old_branch) = branching_native.branches.remove(&predicate) {
                 let (var, vals): (SpecVar, [SpecVal; 2]) = {
                     let (var, val_iter) = native_spec_assign_stream.get_or_insert_with(|| {
                         (rctx.spec_var_stream.next(), SpecVal::iter_domain())
                     });
                     let vals = [
                         val_iter.next().expect("Exhausted specval space!"),
                         val_iter.next().expect("Exhausted specval space!"),
                     ];
                     (*var, vals)
                 };
                 log!(
                     cu.inner.logger,
                     "Branch collision on {:?}. Going to distinguish them with a spec var {:?} ands vals {:?}",
                     &predicate,
                     var,
                     vals
                 );
                 branching_native
                     .branches
                     .insert(predicate.clone().inserted(var, vals[0]), old_branch);
                 branching_native.branches.insert(predicate.inserted(var, vals[1]), branch);
             } else {
                 // no branch collision
                 branching_native.branches.insert(predicate, branch);
+            }
+        }
         for (predicate, branch) in branching_native.branches.iter() {
             let branch = NativeBranch { index, gotten: Default::default(), to_get };
             if branch.is_ended() {
                 log!(
                     cu.inner.logger,
                     "Native submitting solution for branch {} with {:?}",
                     branch.batch_index,
                     "Native submitting solution for batch {} with {:?}",
                     index,
                     &predicate
                 );
                 rctx.solution_storage.submit_and_digest_subtree_solution(
                     &mut *cu.inner.logger,
                     SubtreeId::LocalComponent(ComponentId::Native),
                     predicate.clone(),
                 );
+            }
             if let Some(_) = branching_native.branches.insert(predicate, branch) {
                 // thanks to the native_spec_var, each batch has a distinct predicate
                 unreachable!()
+            }
+        }
         // restore the invariant: !native_batches.is_empty()
         comm.native_batches.push(Default::default());
         comm.endpoint_manager
             .udp_endpoints_round_start(&mut *cu.inner.logger, &mut rctx.spec_var_stream);
         // Call to another big method; keep running this round until a distributed decision is reached
         let decision = Self::sync_reach_decision(
             cu,
             comm,
             &mut branching_native,
             &mut branching_proto_components,
@@ @@ -870,27 +848,27 @@ impl BranchingNative { @@
             self.branches.len(),
             self.branches.keys()
         );
         for (branch_predicate, branch) in self.branches {
             log!(
                 logger,
                 "Considering native branch {:?} with to_get {:?} gotten {:?}",
                 &branch_predicate,
                 &branch.to_get,
                 &branch.gotten
             );
             if branch.is_ended() && branch_predicate.assigns_subset(solution_predicate) {
-                let NativeBranch { batch_index, gotten, .. } = branch;
+                let NativeBranch { index, gotten, .. } = branch;
                 log!(logger, "Collapsed native has gotten {:?}", &gotten);
                 return RoundOk { batch_index, gotten };
+                return RoundOk { batch_index: index, gotten };
+            }
+        }
         panic!("Native had no branches matching pred {:?}", solution_predicate);
+    }
+}
 impl BranchingProtoComponent {
     fn drain_branches_to_blocked(
         cd: CyclicDrainer<Predicate, ProtoComponentBranch>,
         cu: &mut ConnectorUnphased,
         rctx: &mut RoundCtx,
         proto_component_id: ProtoComponentId,
         ports: &HashSet<PortId>,

0 comments (0 inline, 0 general)