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

Changeset - 66513c4f112d

Parent rev.

Child rev.

[Not reviewed]

0 2 0

Christopher Esterhuyse - 5 years ago 2020-09-22 17:28:43
christopher.esterhuyse@gmail.com

more comments and simplification

2 files changed with 91 insertions and 34 deletions:

src/common.rs

src/runtime/communication.rs

0 comments (0 inline, 0 general)

src/common.rs

➞

Show inline comments

@@ @@ -97,14 +97,14 @@ pub(crate) enum SyncBlocker { @@
     SyncBlockEnd,
     CouldntReadMsg(PortId),
     CouldntCheckFiring(PortId),
     PutMsg(PortId, Payload),
     NondetChoice { n: u16 },
+}
 pub(crate) struct DenseDebugHex<'a>(pub &'a [u8]);
 struct DenseDebugHex<'a>(pub &'a [u8]);
 struct DebuggableIter<I: Iterator<Item = T> + Clone, T: Debug>(pub(crate) I);
 ///////////////////// IMPL /////////////////////
 impl IdParts for Id {
     fn id_parts(self) -> (ConnectorId, U32Suffix) {
         (self.connector_id, self.u32_suffix)
+    }
+}
@@ @@ -233,6 +233,12 @@ impl Debug for DenseDebugHex<'_> { @@
         for b in self.0 {
             write!(f, "{:02X?}", b)?;
+        }
         Ok(())
+    }
+}
 impl<I: Iterator<Item = T> + Clone, T: Debug> Debug for DebuggableIter<I, T> {
     fn fmt(&self, f: &mut Formatter<'_>) -> Result<(), std::fmt::Error> {
         f.debug_list().entries(self.0.clone()).finish()
+    }
+}

src/runtime/communication.rs

➞

Show inline comments

 use super::*;
 use crate::common::*;
 use core::ops::{Deref, DerefMut};
 ////////////////
 // Guard protecting an incrementally unfoldable slice of MapTempGuard elements
 struct MapTempsGuard<'a, K, V>(&'a mut [HashMap<K, V>]);
 // Type protecting a temporary map; At the start and end of the Guard's lifetime, self.0.is_empty() must be true
 struct MapTempGuard<'a, K, V>(&'a mut HashMap<K, V>);
 // Once the synchronous round has begun, this structure manages the
 // native component's speculative branches, one per synchronous batch.
 struct BranchingNative {
     branches: HashMap<Predicate, NativeBranch>,
+}
 // Corresponds to one of the native's synchronous batches during the synchronous round.
 // ports marked for message receipt correspond to entries of
 // (a) `gotten` if they have not received yet,
 // (b) `to_get` if they have already received, with the given payload.
 // The branch corresponds to a component solution IFF to_get is empty.
 #[derive(Clone, Debug)]
 struct NativeBranch {
     index: usize,
     gotten: HashMap<PortId, Payload>,
     to_get: HashSet<PortId>,
+}
 // Manages a protocol component's speculative branches for the duration
 // of the synchronous round.
 #[derive(Debug)]
 struct BranchingProtoComponent {
     branches: HashMap<Predicate, ProtoComponentBranch>,
+}
 // One specualtive branch of a protocol component.
 // `ended` IFF this branch has reached SyncBlocker::SyncBlockEnd before.
 #[derive(Debug, Clone)]
 struct ProtoComponentBranch {
     state: ComponentState,
     inner: ProtoComponentBranchInner,
     ended: bool,
+}
 // A structure wrapping a set of three pointers, making it impossible
 // to miss that they are being setup for `cyclic_drain`.
 struct CyclicDrainer<'a, K: Eq + Hash, V> {
     input: &'a mut HashMap<K, V>,
-    inner: CyclicDrainInner<'a, K, V>,
+    inner: CyclicDrainerInner<'a, K, V>,
+}
 struct CyclicDrainInner<'a, K: Eq + Hash, V> {
 // Inner substructure of the Cyclic drainer to be passed through a callback function.
 // See `CyclicDrainer::cyclic_drain`.
 struct CyclicDrainerInner<'a, K: Eq + Hash, V> {
     swap: &'a mut HashMap<K, V>,
     output: &'a mut HashMap<K, V>,
+}
 // Small convenience trait for extending the stdlib's bool type with
 // an optionlike replace method for increasing brevity.
 trait ReplaceBoolTrue {
     fn replace_with_true(&mut self) -> bool;
+}
 impl ReplaceBoolTrue for bool {
     fn replace_with_true(&mut self) -> bool {
         let was = *self;
@@ @@ -182,12 +205,15 @@ impl Connector { @@
     /// states irreversably.
     /// Note that the (b) case necessitates the success of a distributed rollback procedure,
     /// which this component may initiate, but cannot guarantee will succeed in time or at all.
     /// consequently, the given timeout duration represents a duration in which the connector
     /// will make a best effort to fail the round and return control flow to the caller.
     pub fn sync(&mut self, timeout: Option<Duration>) -> Result<usize, SyncError> {
         // This method first destructures the connector, and checks for obvious
         // failure cases. The bulk of the behavior continues in `connected_sync`,
         // to minimize indentation, and enable convient ?-style short circuit syntax.
         let Self { unphased: cu, phased } = self;
         match phased {
             ConnectorPhased::Setup { .. } => Err(SyncError::NotConnected),
             ConnectorPhased::Communication(comm) => {
                 match &comm.round_result {
                     Err(SyncError::Unrecoverable(e)) => {
@@ @@ -203,14 +229,17 @@ impl Connector { @@
                     Ok(Some(ok_result)) => Ok(ok_result.batch_index),
                     Err(sync_error) => Err(sync_error.clone()),
+                }
+            }
+        }
+    }
     // private function. mutates state but returns with round
     // result ASAP (allows for convenient error return with ?)
     // Attempts to complete the synchronous round for the given
     // communication-phased connector structure.
     // Modifies components and ports in `cu` IFF the round succeeds.
     #[inline]
     fn connected_sync(
         cu: &mut ConnectorUnphased,
         comm: &mut ConnectorCommunication,
         timeout: Option<Duration>,
     ) -> Result<Option<RoundOk>, SyncError> {
         //////////////////////////////////
@@ @@ -223,24 +252,28 @@ impl Connector { @@
             "~~~ SYNC called with timeout {:?}; starting round {}",
             &timeout,
             comm.round_index
         );
         log!(@BENCH, cu.logger(), "");
         // 1. run all proto components to Nonsync blockers
         // iterate
         // Create separate storages for ports and components stored in `cu`,
         // while kicking off the branching of components until the set of
         // components entering their synchronous block is finalized in `branching_proto_components`.
         // This is the last time cu's components and ports are accessed until the round is decided.
         let mut current_state = cu.current_state.clone();
         let mut branching_proto_components =
             HashMap::<ComponentId, BranchingProtoComponent>::default();
         let mut unrun_components: Vec<(ComponentId, ComponentState)> = cu
             .proto_components
             .iter()
             .map(|(&proto_id, proto)| (proto_id, proto.clone()))
             .collect();
         log!(cu.logger(), "Nonsync running {} proto components...", unrun_components.len());
         // drains unrun_components, and populates branching_proto_components.
         // initially, the set of components to run is the set of components stored by `cu`,
         // but they are eventually drained into `branching_proto_components`.
         // Some components exit first, and others are created and put into `unrun_components`.
         while let Some((proto_component_id, mut component)) = unrun_components.pop() {
             log!(
                 cu.logger(),
                 "Nonsync running proto component with ID {:?}. {} to go after this",
                 proto_component_id,
                 unrun_components.len()
@@ @@ -260,70 +293,83 @@ impl Connector { @@
                 &blocker
             );
             use NonsyncBlocker as B;
             match blocker {
                 B::ComponentExit => drop(component),
                 B::Inconsistent => return Err(Se::InconsistentProtoComponent(proto_component_id)),
                 B::SyncBlockStart => {
                     branching_proto_components
                         .insert(proto_component_id, BranchingProtoComponent::initial(component));
+                }
                 B::SyncBlockStart => assert!(branching_proto_components
                     .insert(proto_component_id, BranchingProtoComponent::initial(component))
                     .is_none()), // Some(_) returned IFF some component identifier key is overwritten (BAD!)
+            }
+        }
         log!(
             cu.logger(),
             "All {} proto components are now done with Nonsync phase",
             branching_proto_components.len(),
         );
         log!(@BENCH, cu.logger(), "");
-        // Create temp structures needed for the synchronous phase of the round
+        // Create temporary structures needed for the synchronous phase of the round
         let mut rctx = RoundCtx {
             current_state,
+            current_state, // already used previously, now moved into RoundCtx
             solution_storage: {
                 let n = std::iter::once(SubtreeId::LocalComponent(cu.native_component_id));
                 let c =
                     branching_proto_components.keys().map(|&cid| SubtreeId::LocalComponent(cid));
                 let e = comm
                     .neighborhood
                     .children
                     .iter()
                     .map(|&index| SubtreeId::NetEndpoint { index });
                 let subtree_id_iter = n.chain(c).chain(e);
                 let subtree_id_iter = {
                     // Create an iterator over the identifiers of this
                     // connector's childen in the _solution tree_.
                     // Namely, the native, all locally-managed components,
                     // and all this connector's children in the _consensus tree_ (other connectors).
                     let n = std::iter::once(SubtreeId::LocalComponent(cu.native_component_id));
                     let c = branching_proto_components
                         .keys()
                         .map(|&cid| SubtreeId::LocalComponent(cid));
                     let e = comm
                         .neighborhood
                         .children
                         .iter()
                         .map(|&index| SubtreeId::NetEndpoint { index });
                     n.chain(c).chain(e)
                 };
                 log!(
                     cu.logger,
                     "Children in subtree are: {:?}",
-                    subtree_id_iter.clone().collect::<Vec<_>>()
+                    DebuggableIter(subtree_id_iter.clone())
                 );
                 SolutionStorage::new(subtree_id_iter)
             },
             spec_var_stream: cu.current_state.id_manager.new_spec_var_stream(),
             payload_inbox: Default::default(),
+            payload_inbox: Default::default(), // buffer for in-memory payloads to be handled
             deadline: timeout.map(|to| Instant::now() + to),
         };
         log!(cu.logger(), "Round context structure initialized");
         log!(@BENCH, cu.logger(), "");
         // Explore all native branches eagerly. Find solutions, buffer messages, etc.
         // Prepare the branching native component, involving the conversion
         // of its synchronous batches (user provided) into speculative branches eagerly.
         // As a side effect, send all PUTs with the appropriate predicates.
         // Afterwards, each native component's speculative branch finds a local
         // solution the moment it's received all the messages it's awaiting.
         log!(
             cu.logger(),
             "Translating {} native batches into branches...",
             comm.native_batches.len()
         );
         // Allocate a single speculative variable to distinguish each native branch.
         // This enables native components to have distinct branches with identical
         // FIRING variables.
         let native_spec_var = rctx.spec_var_stream.next();
         log!(cu.logger(), "Native branch spec var is {:?}", native_spec_var);
         let mut branching_native = BranchingNative { branches: Default::default() };
         '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;
             // compute the solution predicate to associate with this branch.
             let predicate = {
                 let mut predicate = Predicate::default();
                 // all firing ports have SpecVal::FIRING
                 let firing_iter = to_get.iter().chain(to_put.keys()).copied();
                 log!(
                     cu.logger(),
                     "New native with firing ports {:?}",
                     firing_iter.clone().collect::<Vec<_>>()
                 );
                 let firing_ports: HashSet<PortId> = firing_iter.clone().collect();
@@ @@ -364,12 +410,13 @@ impl Connector { @@
                 // sanity check
                 assert_eq!(Putter, cu.current_state.port_info.get(&putter).unwrap().polarity);
                 rctx.putter_push(cu, putter, msg);
+            }
             let branch = NativeBranch { index, gotten: Default::default(), to_get };
             if branch.is_ended() {
                 // empty to_get set => already corresponds with a component solution
                 log!(
                     cu.logger(),
                     "Native submitting solution for batch {} with {:?}",
                     index,
                     &predicate
                 );
@@ @@ -456,12 +503,13 @@ impl Connector { @@
         branching_proto_components: &mut HashMap<ComponentId, BranchingProtoComponent>,
         rctx: &mut RoundCtx,
     ) -> Result<Decision, UnrecoverableSyncError> {
         log!(@MARK, cu.logger(), "decide start");
         let mut already_requested_failure = false;
         if branching_native.branches.is_empty() {
             // An unsatisfiable native is the easiest way to detect failure
             log!(cu.logger(), "Native starts with no branches! Failure!");
             match comm.neighborhood.parent {
                 Some(parent) => {
                     if already_requested_failure.replace_with_true() {
                         Self::request_failure(cu, comm, parent)?
                     } else {
@@ @@ -471,12 +519,15 @@ impl Connector { @@
                 None => {
                     log!(cu.logger(), "No parent. Deciding on failure");
                     return Ok(Decision::Failure);
+                }
+            }
+        }
         // Create a small set of "workspace" hashmaps, to be passed by-reference into various calls.
         // This is an optimization, avoiding repeated allocation.
         let mut pcb_temps_owner = <[HashMap<Predicate, ProtoComponentBranch>; 3]>::default();
         let mut pcb_temps = MapTempsGuard(&mut pcb_temps_owner);
         let mut bn_temp_owner = <HashMap<Predicate, NativeBranch>>::default();
         // run all proto components to their sync blocker
         log!(
@@ @@ -1159,13 +1210,13 @@ impl SyncProtoContext<'_> { @@
         self.branch_inner.inbox.get(&port)
+    }
     pub(crate) fn take_choice(&mut self) -> Option<u16> {
         self.branch_inner.untaken_choice.take()
+    }
+}
-impl<'a, K: Eq + Hash, V> CyclicDrainInner<'a, K, V> {
+impl<'a, K: Eq + Hash, V> CyclicDrainerInner<'a, K, V> {
     fn add_input(&mut self, k: K, v: V) {
         self.swap.insert(k, v);
+    }
     fn add_output(&mut self, k: K, v: V) {
         self.output.insert(k, v);
+    }
@@ @@ -1249,23 +1300,23 @@ impl ProtoComponentBranch { @@
 impl<'a, K: Eq + Hash + 'static, V: 'static> CyclicDrainer<'a, K, V> {
     fn new(
         input: &'a mut HashMap<K, V>,
         swap: &'a mut HashMap<K, V>,
         output: &'a mut HashMap<K, V>,
     ) -> Self {
-        Self { input, inner: CyclicDrainInner { swap, output } }
+        Self { input, inner: CyclicDrainerInner { swap, output } }
+    }
     fn cyclic_drain<E>(
         self,
-        mut func: impl FnMut(K, V, CyclicDrainInner<'_, K, V>) -> Result<(), E>,
+        mut func: impl FnMut(K, V, CyclicDrainerInner<'_, K, V>) -> Result<(), E>,
     ) -> Result<(), E> {
-        let Self { input, inner: CyclicDrainInner { swap, output } } = self;
+        let Self { input, inner: CyclicDrainerInner { swap, output } } = self;
         // assert!(swap.is_empty());
         while !input.is_empty() {
             for (k, v) in input.drain() {
-                func(k, v, CyclicDrainInner { swap, output })?
+                func(k, v, CyclicDrainerInner { swap, output })?
+            }
             std::mem::swap(input, swap);
+        }
         Ok(())
+    }
+}

0 comments (0 inline, 0 general)