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

Changeset - 8ab15200d9a4

Parent rev.

Child rev.

[Not reviewed]

0 4 0

Christopher Esterhuyse - 5 years ago 2020-06-24 18:52:35
christopher.esterhuyse@gmail.com

misc refactoring: (1) more thorough error handling, (2) more modular functions

4 files changed with 167 insertions and 123 deletions:

src/runtime/communication.rs

155

121

src/runtime/endpoints.rs

src/runtime/error.rs

src/runtime/setup.rs

0 comments (0 inline, 0 general)

src/runtime/communication.rs

➞

Show inline comments

@@ @@ -17,120 +17,126 @@ struct SolutionStorage { @@
     new_local: HashSet<Predicate>,
     // this pair acts as Route -> HashSet<Predicate> which is friendlier to iteration
     subtree_solutions: Vec<HashSet<Predicate>>,
     subtree_id_to_index: HashMap<Route, usize>,
+}
 #[derive(Debug)]
 struct BranchingProtoComponent {
     ports: HashSet<PortId>,
     branches: HashMap<Predicate, ProtoComponentBranch>,
+}
 #[derive(Debug, Clone)]
 struct ProtoComponentBranch {
     inbox: HashMap<PortId, Payload>,
     state: ComponentState,
+}
 struct CyclicDrainer<'a, K: Eq + Hash, V> {
     input: &'a mut HashMap<K, V>,
     inner: CyclicDrainInner<'a, K, V>,
+}
 struct CyclicDrainInner<'a, K: Eq + Hash, V> {
     swap: &'a mut HashMap<K, V>,
     output: &'a mut HashMap<K, V>,
+}
 trait PayloadMsgSender {
     fn send(&mut self, port_info: &PortInfo, putter: &PortId, msg: SendPayloadMsg);
     fn putter_send(
         &mut self,
         cu: &mut ConnectorUnphased,
         putter: PortId,
         msg: SendPayloadMsg,
     ) -> Result<(), SyncError>;
+}
 ////////////////
 impl Connector {
     pub fn gotten(&mut self, port: PortId) -> Result<&Payload, GottenError> {
         use GottenError::*;
         let Self { phased, .. } = self;
         match phased {
             ConnectorPhased::Setup { .. } => Err(NoPreviousRound),
             ConnectorPhased::Communication(ConnectorCommunication { round_result, .. }) => {
                 match round_result {
                     Err(_) => Err(PreviousSyncFailed),
                     Ok(None) => Err(NoPreviousRound),
                     Ok(Some(round_ok)) => round_ok.gotten.get(&port).ok_or(PortDidntGet),
+                }
+            }
+        }
+    }
     pub fn put(&mut self, port: PortId, payload: Payload) -> Result<(), PortOpError> {
         use PortOpError::*;
         let Self { unphased, phased } = self;
         if !unphased.native_ports.contains(&port) {
             return Err(PortUnavailable);
+        }
         if Putter != *unphased.port_info.polarities.get(&port).unwrap() {
             return Err(WrongPolarity);
+        }
         match phased {
             ConnectorPhased::Setup { .. } => Err(NotConnected),
             ConnectorPhased::Communication(ConnectorCommunication { native_batches, .. }) => {
                 let batch = native_batches.last_mut().unwrap();
                 if batch.to_put.contains_key(&port) {
                     return Err(MultipleOpsOnPort);
+                }
                 batch.to_put.insert(port, payload);
                 Ok(())
+            }
+        }
+    }
     pub fn next_batch(&mut self) -> Result<usize, NextBatchError> {
         // returns index of new batch
         use NextBatchError::*;
         let Self { phased, .. } = self;
         match phased {
             ConnectorPhased::Setup { .. } => Err(NotConnected),
             ConnectorPhased::Communication(ConnectorCommunication { native_batches, .. }) => {
                 native_batches.push(Default::default());
                 Ok(native_batches.len() - 1)
+            }
+        }
+    }
     pub fn get(&mut self, port: PortId) -> Result<(), PortOpError> {
     fn port_op_access(
         &mut self,
         port: PortId,
         expect_polarity: Polarity,
     ) -> Result<&mut NativeBatch, PortOpError> {
         use PortOpError::*;
         let Self { unphased, phased } = self;
         if !unphased.native_ports.contains(&port) {
             return Err(PortUnavailable);
+        }
         if Getter != *unphased.port_info.polarities.get(&port).unwrap() {
             return Err(WrongPolarity);
         match unphased.port_info.polarities.get(&port) {
             Some(p) if *p == expect_polarity => {}
             Some(_) => return Err(WrongPolarity),
             None => return Err(UnknownPolarity),
+        }
         match phased {
             ConnectorPhased::Setup { .. } => Err(NotConnected),
             ConnectorPhased::Communication(ConnectorCommunication { native_batches, .. }) => {
                 let batch = native_batches.last_mut().unwrap();
                 if !batch.to_get.insert(port) {
                     return Err(MultipleOpsOnPort);
+                }
                 Ok(())
                 let batch = native_batches.last_mut().unwrap(); // length >= invariant
                 Ok(batch)
+            }
+        }
+    }
     pub fn put(&mut self, port: PortId, payload: Payload) -> Result<(), PortOpError> {
         use PortOpError::*;
         let batch = self.port_op_access(port, Putter)?;
         if batch.to_put.contains_key(&port) {
             Err(MultipleOpsOnPort)
         } else {
             batch.to_put.insert(port, payload);
             Ok(())
+        }
+    }
     pub fn get(&mut self, port: PortId) -> Result<(), PortOpError> {
         use PortOpError::*;
         let batch = self.port_op_access(port, Getter)?;
         if batch.to_get.insert(port) {
             Ok(())
         } else {
             Err(MultipleOpsOnPort)
+        }
+    }
     // entrypoint for caller. overwrites round result enum, and returns what happened
     pub fn sync(&mut self, timeout: Option<Duration>) -> Result<usize, SyncError> {
         let Self { unphased, phased } = self;
         match phased {
             ConnectorPhased::Setup { .. } => Err(SyncError::NotConnected),
             ConnectorPhased::Communication(comm) => {
                 comm.round_result = Self::connected_sync(unphased, comm, timeout);
                 match &comm.round_result {
                     Ok(None) => unreachable!(),
                     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 ?)
     fn connected_sync(
         cu: &mut ConnectorUnphased,
         comm: &mut ConnectorCommunication,
         timeout: Option<Duration>,
     ) -> Result<Option<RoundOk>, SyncError> {
         use SyncError as Se;
         let mut deadline = timeout.map(|to| Instant::now() + to);
@@ @@ -143,49 +149,49 @@ impl Connector { @@
         // 1. run all proto components to Nonsync blockers
         let mut branching_proto_components =
             HashMap::<ProtoComponentId, BranchingProtoComponent>::default();
         let mut unrun_components: Vec<(ProtoComponentId, ProtoComponent)> =
             cu.proto_components.iter().map(|(&k, v)| (k, v.clone())).collect();
         log!(cu.logger, "Nonsync running {} proto components...", unrun_components.len());
         while let Some((proto_component_id, mut component)) = unrun_components.pop() {
             // TODO coalesce fields
             log!(
                 cu.logger,
                 "Nonsync running proto component with ID {:?}. {} to go after this",
                 proto_component_id,
                 unrun_components.len()
             );
             let mut ctx = NonsyncProtoContext {
                 logger: &mut *cu.logger,
                 port_info: &mut cu.port_info,
                 id_manager: &mut cu.id_manager,
                 proto_component_id,
                 unrun_components: &mut unrun_components,
                 proto_component_ports: &mut cu
                     .proto_components
                     .get_mut(&proto_component_id)
                     .unwrap()
+                    .unwrap() // unrun_components' keys originate from proto_components
                     .ports,
             };
             let blocker = component.state.nonsync_run(&mut ctx, &cu.proto_description);
             log!(
                 cu.logger,
                 "proto component {:?} ran to nonsync blocker {:?}",
                 proto_component_id,
                 &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));
+                }
+            }
+        }
         log!(
             cu.logger,
             "All {} proto components are now done with Nonsync phase",
             branching_proto_components.len(),
         );
@@ @@ -210,181 +216,195 @@ impl Connector { @@
             comm.native_batches.len()
         );
         let mut branching_native = BranchingNative { branches: Default::default() };
         for (index, NativeBatch { to_get, to_put }) in comm.native_batches.drain(..).enumerate() {
             let predicate = {
                 let mut predicate = Predicate::default();
                 // assign trues
                 for &port in to_get.iter().chain(to_put.keys()) {
                     let var = cu.port_info.firing_var_for(port);
                     predicate.assigned.insert(var, true);
+                }
                 // assign falses
                 for &port in cu.native_ports.iter() {
                     let var = cu.port_info.firing_var_for(port);
                     predicate.assigned.entry(var).or_insert(false);
+                }
                 predicate
             };
             log!(cu.logger, "Native branch {} has pred {:?}", index, &predicate);
             // put all messages
             for (putter, payload) in to_put {
                 let msg = SendPayloadMsg { predicate: predicate.clone(), payload };
                 log!(cu.logger, "Native branch {} sending msg {:?}", index, &msg);
                 // rely on invariant: sync batches respect port polarity
                 let getter = *cu.port_info.peers.get(&putter).unwrap();
                 payloads_to_get.push((getter, msg));
                 payloads_to_get.putter_send(cu, putter, msg)?;
+            }
             if to_get.is_empty() {
                 log!(
                     cu.logger,
                     "Native submitting solution for batch {} with {:?}",
                     index,
                     &predicate
                 );
                 solution_storage.submit_and_digest_subtree_solution(
                     &mut *cu.logger,
                     Route::LocalComponent(ComponentId::Native),
                     predicate.clone(),
                 );
+            }
             let branch = NativeBranch { index, gotten: Default::default(), to_get };
             if let Some(existing) = branching_native.branches.insert(predicate, branch) {
                 // TODO
                 return Err(Se::IndistinguishableBatches([index, existing.index]));
+            }
+        }
         log!(cu.logger, "Done translating native batches into branches");
         comm.native_batches.push(Default::default());
         // run all proto components to their sync blocker
         log!(
             cu.logger,
             "Running all {} proto components to their sync blocker...",
             branching_proto_components.len()
         );
         for (&proto_component_id, proto_component) in branching_proto_components.iter_mut() {
             let BranchingProtoComponent { ports, branches } = proto_component;
             let mut swap = HashMap::default();
             let mut blocked = HashMap::default();
             // drain from branches --> blocked
             let cd = CyclicDrainer::new(branches, &mut swap, &mut blocked);
             BranchingProtoComponent::drain_branches_to_blocked(
                 cd,
                 cu,
                 &mut solution_storage,
                 &mut payloads_to_get,
                 proto_component_id,
                 ports,
             );
+            )?;
             // swap the blocked branches back
             std::mem::swap(&mut blocked, branches);
+        }
         log!(cu.logger, "All proto components are blocked");
         log!(cu.logger, "Entering decision loop...");
         comm.endpoint_manager.undelay_all();
         let decision = 'undecided: loop {
             // drain payloads_to_get, sending them through endpoints / feeding them to components
             while let Some((getter, send_payload_msg)) = payloads_to_get.pop() {
                 assert!(cu.port_info.polarities.get(&getter).copied() == Some(Getter));
                 match cu.port_info.routes.get(&getter).unwrap() {
                     Route::Endpoint { index } => {
                 match cu.port_info.routes.get(&getter) {
                     None => {
                         log!(
                             cu.logger,
                             "Delivery to getter {:?} msg {:?} failed. Physical route unmapped!",
                             getter,
                             &send_payload_msg
                         );
+                    }
                     Some(Route::Endpoint { index }) => {
                         let msg = Msg::CommMsg(CommMsg {
                             round_index: comm.round_index,
                             contents: CommMsgContents::SendPayload(send_payload_msg),
                         });
-                        comm.endpoint_manager.send_to(*index, &msg).unwrap();
+                        comm.endpoint_manager.send_to_comms(*index, &msg)?;
+                    }
                     Route::LocalComponent(ComponentId::Native) => branching_native.feed_msg(
+                    Some(Route::LocalComponent(ComponentId::Native)) => branching_native.feed_msg(
                         cu,
                         &mut solution_storage,
                         // &mut Pay
                         getter,
                         send_payload_msg,
+                        &send_payload_msg,
                     ),
                     Route::LocalComponent(ComponentId::Proto(proto_component_id)) => {
+                    Some(Route::LocalComponent(ComponentId::Proto(proto_component_id))) => {
                         if let Some(branching_component) =
                             branching_proto_components.get_mut(proto_component_id)
+                        {
                             let proto_component_id = *proto_component_id;
                             // let ConnectorUnphased { port_info, proto_description, .. } = cu;
                             branching_component.feed_msg(
                                 cu,
                                 &mut solution_storage,
                                 proto_component_id,
                                 &mut payloads_to_get,
                                 getter,
                                 send_payload_msg,
+                            )
                                 &send_payload_msg,
                             )?;
                         } else {
                             log!(
                                 cu.logger,
                                 "Delivery to getter {:?} msg {:?} failed because {:?} isn't here",
                                 getter,
                                 &send_payload_msg,
                                 proto_component_id
                             );
+                        }
+                    }
+                }
+            }
             // check if we have a solution yet
             log!(cu.logger, "Check if we have any local decisions...");
             for solution in solution_storage.iter_new_local_make_old() {
                 log!(cu.logger, "New local decision with solution {:?}...", &solution);
                 match comm.neighborhood.parent {
                     Some(parent) => {
                         log!(cu.logger, "Forwarding to my parent {:?}", parent);
                         let suggestion = Decision::Success(solution);
                         let msg = Msg::CommMsg(CommMsg {
                             round_index: comm.round_index,
                             contents: CommMsgContents::Suggest { suggestion },
                         });
-                        comm.endpoint_manager.send_to(parent, &msg).unwrap();
+                        comm.endpoint_manager.send_to_comms(parent, &msg)?;
+                    }
                     None => {
                         log!(cu.logger, "No parent. Deciding on solution {:?}", &solution);
                         break 'undecided Decision::Success(solution);
+                    }
+                }
+            }
             // stuck! make progress by receiving a msg
             // try recv messages arriving through endpoints
             log!(cu.logger, "No decision yet. Let's recv an endpoint msg...");
+            {
                 let (endpoint_index, msg) = loop {
                     match comm.endpoint_manager.try_recv_any_comms(&mut *cu.logger, deadline)? {
                         None => {
                             log!(cu.logger, "Reached user-defined deadling without decision...");
                             if let Some(parent) = comm.neighborhood.parent {
                                 log!(
                                     cu.logger,
                                     "Sending failure request to parent index {}",
                                     parent
                                 );
                                 let msg = Msg::CommMsg(CommMsg {
                                     round_index: comm.round_index,
                                     contents: CommMsgContents::Suggest {
                                         suggestion: Decision::Failure,
                                     },
                                 });
-                                comm.endpoint_manager.send_to(parent, &msg).unwrap();
+                                comm.endpoint_manager.send_to_comms(parent, &msg)?;
                             } else {
                                 log!(cu.logger, "As the leader, deciding on timeout");
                                 break 'undecided Decision::Failure;
+                            }
                             deadline = None;
+                        }
                         Some((endpoint_index, msg)) => break (endpoint_index, msg),
+                    }
                 };
                 log!(cu.logger, "Received from endpoint {} msg {:?}", endpoint_index, &msg);
                 let comm_msg_contents = match msg {
                     Msg::SetupMsg(..) => {
                         log!(cu.logger, "Discarding setup message; that phase is over");
                         continue 'undecided;
+                    }
                     Msg::CommMsg(comm_msg) => match comm_msg.round_index.cmp(&comm.round_index) {
                         Ordering::Equal => comm_msg.contents,
                         Ordering::Less => {
                             log!(
                                 cu.logger,
                                 "We are in round {}, but msg is for round {}. Discard",
                                 comm_msg.round_index,
                                 comm.round_index,
                             );
@@ @@ -427,118 +447,117 @@ impl Connector { @@
                                     let route = Route::Endpoint { index: endpoint_index };
                                     solution_storage.submit_and_digest_subtree_solution(
                                         &mut *cu.logger,
                                         route,
                                         predicate,
                                     );
+                                }
                                 Decision::Failure => {
                                     match comm.neighborhood.parent {
                                         None => {
                                             log!(
                                                 cu.logger,
                                                 "As sink, I decide on my child's failure"
                                             );
                                             // I am the sink. Decide on failed
                                             break 'undecided Decision::Failure;
+                                        }
                                         Some(parent) => {
                                             log!(cu.logger, "Forwarding failure through my parent endpoint {:?}", parent);
                                             // I've got a parent. Forward the failure suggestion.
                                             let msg = Msg::CommMsg(CommMsg {
                                                 round_index: comm.round_index,
                                                 contents: CommMsgContents::Suggest { suggestion },
                                             });
-                                            comm.endpoint_manager.send_to(parent, &msg).unwrap();
+                                            comm.endpoint_manager.send_to_comms(parent, &msg)?;
+                                        }
+                                    }
+                                }
+                            }
                         } else {
                             log!(
                                 cu.logger,
                                 "Discarding suggestion {:?} from non-child endpoint idx {:?}",
                                 &suggestion,
                                 endpoint_index
                             );
+                        }
+                    }
                     CommMsgContents::Announce { decision } => {
                         if Some(endpoint_index) == comm.neighborhood.parent {
                             // adopt this decision
                             break 'undecided decision;
                         } else {
                             log!(
                                 cu.logger,
                                 "Discarding announcement {:?} from non-parent endpoint idx {:?}",
                                 &decision,
                                 endpoint_index
                             );
+                        }
+                    }
+                }
+            }
             log!(cu.logger, "Endpoint msg recv done");
         };
         log!(cu.logger, "Committing to decision {:?}!", &decision);
         // propagate the decision to children
         let msg = Msg::CommMsg(CommMsg {
             round_index: comm.round_index,
             contents: CommMsgContents::Announce { decision: decision.clone() },
         });
         log!(
             cu.logger,
             "Announcing decision {:?} through child endpoints {:?}",
             &msg,
             &comm.neighborhood.children
         );
         for &child in comm.neighborhood.children.iter() {
-            comm.endpoint_manager.send_to(child, &msg).unwrap();
+            comm.endpoint_manager.send_to_comms(child, &msg)?;
+        }
         match decision {
             Decision::Failure => Err(Se::RoundFailure),
             Decision::Success(predicate) => {
                 // commit changes to component states
                 cu.proto_components.clear();
                 cu.proto_components.extend(
                     branching_proto_components
                         .into_iter()
                         .map(|(id, bpc)| (id, bpc.collapse_with(&predicate))),
                 );
                 Ok(Some(branching_native.collapse_with(&predicate)))
+            }
+        }
+    }
+}
 impl BranchingNative {
     fn feed_msg(
         &mut self,
         cu: &mut ConnectorUnphased,
         solution_storage: &mut SolutionStorage,
         getter: PortId,
         send_payload_msg: SendPayloadMsg,
+        send_payload_msg: &SendPayloadMsg,
     ) {
         log!(cu.logger, "feeding native getter {:?} {:?}", getter, &send_payload_msg);
         assert!(cu.port_info.polarities.get(&getter).copied() == Some(Getter));
         let mut draining = HashMap::default();
         let finished = &mut self.branches;
         std::mem::swap(&mut draining, finished);
         for (predicate, mut branch) in draining.drain() {
             log!(cu.logger, "visiting native branch {:?} with {:?}", &branch, &predicate);
             // check if this branch expects to receive it
             let var = cu.port_info.firing_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() {
                     let route = Route::LocalComponent(ComponentId::Native);
                     solution_storage.submit_and_digest_subtree_solution(
                         &mut *cu.logger,
                         route,
                         predicate.clone(),
                     );
+                }
             };
             if predicate.query(var) != Some(true) {
@@ @@ -599,128 +618,129 @@ impl BranchingNative { @@
+    }
     fn collapse_with(self, solution_predicate: &Predicate) -> RoundOk {
         for (branch_predicate, branch) in self.branches {
             if solution_predicate.satisfies(&branch_predicate) {
                 let NativeBranch { index, gotten, .. } = branch;
                 return RoundOk { batch_index: index, gotten };
+            }
+        }
         panic!("Native had no branches matching pred {:?}", solution_predicate);
+    }
+}
 // |putter, m| {
 //     let getter = *cu.port_info.peers.get(&putter).unwrap();
 //     payloads_to_get.push((getter, m));
 // },
 impl BranchingProtoComponent {
     fn drain_branches_to_blocked(
         cd: CyclicDrainer<Predicate, ProtoComponentBranch>,
         cu: &mut ConnectorUnphased,
         solution_storage: &mut SolutionStorage,
         payload_msg_sender: &mut impl PayloadMsgSender,
         proto_component_id: ProtoComponentId,
         ports: &HashSet<PortId>,
     ) {
+    ) -> Result<(), SyncError> {
         cd.cylic_drain(|mut predicate, mut branch, mut drainer| {
             let mut ctx = SyncProtoContext {
                     logger: &mut *cu.logger,
                     predicate: &predicate,
                     port_info: &cu.port_info,
                     inbox: &branch.inbox,
                 };
                 let blocker = branch.state.sync_run(&mut ctx, &cu.proto_description);
                 log!(
                     cu.logger,
                     "Proto component with id {:?} branch with pred {:?} hit blocker {:?}",
                     proto_component_id,
                     &predicate,
                     &blocker,
                 );
                 use SyncBlocker as B;
                 match blocker {
                     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.firing_var_for(port);
                             predicate.assigned.entry(var).or_insert(false);
+                        }
                         // submit solution for this component
                         solution_storage.submit_and_digest_subtree_solution(
                             &mut *cu.logger,
                             Route::LocalComponent(ComponentId::Proto(proto_component_id)),
                             predicate.clone(),
                         );
                         // 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) => {
                         // sanity check
                 logger: &mut *cu.logger,
                 predicate: &predicate,
                 port_info: &cu.port_info,
                 inbox: &branch.inbox,
             };
             let blocker = branch.state.sync_run(&mut ctx, &cu.proto_description);
             log!(
                 cu.logger,
                 "Proto component with id {:?} branch with pred {:?} hit blocker {:?}",
                 proto_component_id,
                 &predicate,
                 &blocker,
             );
             use SyncBlocker as B;
             match blocker {
                 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.firing_var_for(port);
                         assert!(predicate.query(var).is_none());
                         // keep forks in "unblocked"
                         drainer.add_input(predicate.clone().inserted(var, false), branch.clone());
                         drainer.add_input(predicate.inserted(var, true), branch);
                         predicate.assigned.entry(var).or_insert(false);
+                    }
                     B::PutMsg(putter, payload) => {
                         // sanity check
                         assert_eq!(Some(&Putter), cu.port_info.polarities.get(&putter));
                         // overwrite assignment
                         let var = cu.port_info.firing_var_for(putter);
                         let was = predicate.assigned.insert(var, true);
                         if was == Some(false) {
                             log!(cu.logger, "Proto component {:?} tried to PUT on port {:?} when pred said var {:?}==Some(false). inconsistent!", proto_component_id, putter, var);
                             // discard forever
                             drop((predicate, branch));
                         } else {
                             // keep in "unblocked"
                             log!(cu.logger, "Proto component {:?} putting payload {:?} on port {:?} (using var {:?})", proto_component_id, &payload, putter, var);
                             let msg = SendPayloadMsg { predicate: predicate.clone(), payload };
                             payload_msg_sender.send(&cu.port_info, &putter, msg);
                             drainer.add_input(predicate, branch);
+                        }
                     // submit solution for this component
                     solution_storage.submit_and_digest_subtree_solution(
                         &mut *cu.logger,
                         Route::LocalComponent(ComponentId::Proto(proto_component_id)),
                         predicate.clone(),
                     );
                     // 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) => {
                     // sanity check
                     let var = cu.port_info.firing_var_for(port);
                     assert!(predicate.query(var).is_none());
                     // keep forks in "unblocked"
                     drainer.add_input(predicate.clone().inserted(var, false), branch.clone());
                     drainer.add_input(predicate.inserted(var, true), branch);
+                }
                 B::PutMsg(putter, payload) => {
                     // sanity check
                     assert_eq!(Some(&Putter), cu.port_info.polarities.get(&putter));
                     // overwrite assignment
                     let var = cu.port_info.firing_var_for(putter);
                     let was = predicate.assigned.insert(var, true);
                     if was == Some(false) {
                         log!(cu.logger, "Proto component {:?} tried to PUT on port {:?} when pred said var {:?}==Some(false). inconsistent!", proto_component_id, putter, var);
                         // discard forever
                         drop((predicate, branch));
                     } else {
                         // keep in "unblocked"
                         log!(cu.logger, "Proto component {:?} putting payload {:?} on port {:?} (using var {:?})", proto_component_id, &payload, putter, var);
                         let msg = SendPayloadMsg { predicate: predicate.clone(), payload };
                         payload_msg_sender.putter_send(cu, putter, msg)?;
                         drainer.add_input(predicate, branch);
+                    }
+                }
         });
+            }
             Ok(())
         })
+    }
     fn feed_msg(
         &mut self,
         cu: &mut ConnectorUnphased,
         solution_storage: &mut SolutionStorage,
         proto_component_id: ProtoComponentId,
         payload_msg_sender: &mut impl PayloadMsgSender,
         getter: PortId,
         send_payload_msg: SendPayloadMsg,
     ) {
         send_payload_msg: &SendPayloadMsg,
     ) -> Result<(), SyncError> {
         let logger = &mut *cu.logger;
         log!(
             logger,
             "feeding proto component {:?} getter {:?} {:?}",
             proto_component_id,
             getter,
             &send_payload_msg
         );
         let BranchingProtoComponent { branches, ports } = self;
         let mut unblocked = HashMap::default();
         let mut blocked = HashMap::default();
         // partition drain from branches -> {unblocked, blocked}
         log!(logger, "visiting {} blocked branches...", branches.len());
         for (predicate, mut branch) in branches.drain() {
             use CommonSatResult as Csr;
             log!(logger, "visiting branch with pred {:?}", &predicate);
             match predicate.common_satisfier(&send_payload_msg.predicate) {
                 Csr::Nonexistant => {
                     // this branch does not receive the message
                     log!(logger, "skipping branch");
                     blocked.insert(predicate, branch);
+                }
                 Csr::Equivalent | Csr::FormerNotLatter => {
                     // retain the existing predicate, but add this payload
@@ @@ -737,52 +757,53 @@ impl BranchingProtoComponent { @@
                     blocked.insert(predicate, branch);
                     unblocked.insert(predicate2, branch2);
+                }
                 Csr::New(predicate2) => {
                     // fork branch, give fork the message and the new predicate. original branch untouched
                     log!(logger, "Forking this branch with new predicate {:?}", &predicate2);
                     let mut branch2 = branch.clone();
                     branch2.feed_msg(getter, send_payload_msg.payload.clone());
                     blocked.insert(predicate, branch);
                     unblocked.insert(predicate2, branch2);
+                }
+            }
+        }
         log!(logger, "blocked {:?} unblocked {:?}", blocked.len(), unblocked.len());
         // drain from unblocked --> blocked
         let mut swap = HashMap::default();
         let cd = CyclicDrainer::new(&mut unblocked, &mut swap, &mut blocked);
         BranchingProtoComponent::drain_branches_to_blocked(
             cd,
             cu,
             solution_storage,
             payload_msg_sender,
             proto_component_id,
             ports,
         );
+        )?;
         // swap the blocked branches back
         std::mem::swap(&mut blocked, branches);
         log!(cu.logger, "component settles down with branches: {:?}", branches.keys());
         Ok(())
+    }
     fn collapse_with(self, solution_predicate: &Predicate) -> ProtoComponent {
         let BranchingProtoComponent { ports, branches } = self;
         for (branch_predicate, branch) in branches {
             if branch_predicate.satisfies(solution_predicate) {
                 let ProtoComponentBranch { state, .. } = branch;
                 return ProtoComponent { state, ports };
+            }
+        }
         panic!("ProtoComponent had no branches matching pred {:?}", solution_predicate);
+    }
     fn initial(ProtoComponent { state, ports }: ProtoComponent) -> Self {
         let branch = ProtoComponentBranch { inbox: Default::default(), state };
         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();
         let mut subtree_solutions = vec![];
         for key in routes {
             subtree_id_to_index.insert(key, subtree_solutions.len());
             subtree_solutions.push(Default::default())
+        }
@@ @@ -859,51 +880,60 @@ impl SolutionStorage { @@
         if let Some(set) = set_visitor.next() {
             // incomplete solution. keep traversing
             for pred in set.iter() {
                 if let Some(elaborated) = pred.union_with(&partial) {
                     Self::elaborate_into_new_local_rec(
                         logger,
                         elaborated,
                         set_visitor.clone(),
                         old_local,
                         new_local,
+                    )
+                }
+            }
         } else {
             // recursive stop condition. `partial` is a local subtree solution
             if !old_local.contains(&partial) {
                 // ... and it hasn't been found before
                 log!(logger, "storing NEW LOCAL SOLUTION {:?}", &partial);
                 new_local.insert(partial);
+            }
+        }
+    }
+}
 impl PayloadMsgSender for Vec<(PortId, SendPayloadMsg)> {
     fn send(&mut self, port_info: &PortInfo, putter: &PortId, msg: SendPayloadMsg) {
         let getter = *port_info.peers.get(putter).unwrap();
         self.push((getter, msg));
     fn putter_send(
         &mut self,
         cu: &mut ConnectorUnphased,
         putter: PortId,
         msg: SendPayloadMsg,
     ) -> Result<(), SyncError> {
         if let Some(&getter) = cu.port_info.peers.get(&putter) {
             self.push((getter, msg));
             Ok(())
         } else {
             Err(SyncError::MalformedStateError(MalformedStateError::GetterUnknownFor { putter }))
+        }
+    }
+}
 impl SyncProtoContext<'_> {
     pub(crate) fn is_firing(&mut self, port: PortId) -> Option<bool> {
         let var = self.port_info.firing_var_for(port);
         self.predicate.query(var)
+    }
     pub(crate) fn read_msg(&mut self, port: PortId) -> Option<&Payload> {
         self.inbox.get(&port)
+    }
+}
 impl<'a, K: Eq + Hash, V> CyclicDrainInner<'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);
+    }
+}
 impl NonsyncProtoContext<'_> {
     pub fn new_component(&mut self, moved_ports: HashSet<PortId>, state: ComponentState) {
         // called by a PROTO COMPONENT. moves its own ports.
         // 1. sanity check: this component owns these ports
         log!(
@@ @@ -939,35 +969,39 @@ impl NonsyncProtoContext<'_> { @@
         log!(
             self.logger,
             "Component {:?} port pair (out->in) {:?} -> {:?}",
             self.proto_component_id,
             o,
+            i
         );
         [o, i]
+    }
+}
 impl ProtoComponentBranch {
     fn feed_msg(&mut self, getter: PortId, payload: Payload) {
         let was = self.inbox.insert(getter, payload);
         assert!(was.is_none())
+    }
+}
 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 } }
+    }
     fn cylic_drain(self, mut func: impl FnMut(K, V, CyclicDrainInner<'_, K, V>)) {
     fn cylic_drain<E>(
         self,
         mut func: impl FnMut(K, V, CyclicDrainInner<'_, K, V>) -> Result<(), E>,
     ) -> Result<(), E> {
         let Self { input, inner: CyclicDrainInner { 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, CyclicDrainInner { swap, output })?
+            }
             std::mem::swap(input, swap);
+        }
         Ok(())
+    }
+}

src/runtime/endpoints.rs

➞

Show inline comments

@@ @@ -37,48 +37,52 @@ impl Endpoint { @@
                 Ok(Some(msg))
+            }
             Err(e) => match *e {
                 bincode::ErrorKind::Io(k) if k.kind() == std::io::ErrorKind::UnexpectedEof => {
                     Ok(None)
+                }
                 _ => Err(MalformedMessage),
                 // println!("SERDE ERRKIND {:?}", e);
                 // Err(MalformedMessage)
             },
+        }
+    }
     pub(super) fn send<T: serde::ser::Serialize>(&mut self, msg: &T) -> Result<(), EndpointError> {
         bincode::serialize_into(&mut self.stream, msg).map_err(|_| EndpointError::BrokenEndpoint)
+    }
+}
 impl EndpointManager {
     pub(super) fn index_iter(&self) -> Range<usize> {
 ..self.num_endpoints()
+    }
     pub(super) fn num_endpoints(&self) -> usize {
         self.endpoint_exts.len()
+    }
     pub(super) fn send_to_comms(&mut self, index: usize, msg: &Msg) -> Result<(), SyncError> {
         let endpoint = &mut self.endpoint_exts[index].endpoint;
         endpoint.send(msg).map_err(|_| SyncError::BrokenEndpoint(index))
+    }
     pub(super) fn send_to_setup(&mut self, index: usize, msg: &Msg) -> Result<(), ConnectError> {
         let endpoint = &mut self.endpoint_exts[index].endpoint;
         endpoint.send(msg).map_err(|err| {
             ConnectError::EndpointSetupError(endpoint.stream.local_addr().unwrap(), err)
         })
+    }
     pub(super) fn send_to(&mut self, index: usize, msg: &Msg) -> Result<(), EndpointError> {
         self.endpoint_exts[index].endpoint.send(msg)
+    }
     pub(super) fn try_recv_any_comms(
         &mut self,
         logger: &mut dyn Logger,
         deadline: Option<Instant>,
     ) -> Result<Option<(usize, Msg)>, SyncError> {
         use {SyncError as Se, TryRecyAnyError as Trae};
         match self.try_recv_any(logger, deadline) {
             Ok(tup) => Ok(Some(tup)),
             Err(Trae::Timeout) => Ok(None),
             Err(Trae::PollFailed) => Err(Se::PollFailed),
             Err(Trae::EndpointError { error: _, index }) => Err(Se::BrokenEndpoint(index)),
+        }
+    }
     pub(super) fn try_recv_any_setup(
         &mut self,

src/runtime/error.rs

➞

Show inline comments

 use crate::common::*;
 #[derive(Debug)]
 pub enum ConnectError {
     BindFailed(SocketAddr),
     PollInitFailed,
     Timeout,
     PollFailed,
     AcceptFailed(SocketAddr),
     AlreadyConnected,
     PortPeerPolarityMismatch(PortId),
     EndpointSetupError(SocketAddr, EndpointError),
     SetupAlgMisbehavior,
+}
 ////////////////////////
 #[derive(Debug, Clone)]
 pub enum SyncError {
     NotConnected,
     InconsistentProtoComponent(ProtoComponentId),
     IndistinguishableBatches([usize; 2]),
     RoundFailure,
     PollFailed,
     BrokenEndpoint(usize),
     MalformedStateError(MalformedStateError),
+}
 #[derive(Debug, Clone)]
 pub enum MalformedStateError {
     PortCannotPut(PortId),
     GetterUnknownFor { putter: PortId },
+}
 #[derive(Debug, Clone)]
 pub enum EndpointError {
     MalformedMessage,
     BrokenEndpoint,
+}
 #[derive(Debug)]
 pub enum PortOpError {
     WrongPolarity,
     UnknownPolarity,
     NotConnected,
     MultipleOpsOnPort,
     PortUnavailable,
+}
 #[derive(Debug, Eq, PartialEq)]
 pub enum GottenError {
     NoPreviousRound,
     PortDidntGet,
     PreviousSyncFailed,
+}
 #[derive(Debug, Eq, PartialEq)]
 pub enum NextBatchError {
     NotConnected,
+}

src/runtime/setup.rs

➞

Show inline comments

 use crate::common::*;
 use crate::runtime::*;
 use std::io::ErrorKind::WouldBlock;
 impl Connector {
     pub fn new(
         mut logger: Box<dyn Logger>,
         proto_description: Arc<ProtocolDescription>,
         connector_id: ConnectorId,
         surplus_sockets: u16,
     ) -> Self {
         log!(&mut *logger, "Created with connector_id {:?}", connector_id);
         Self {
             unphased: ConnectorUnphased {
                 proto_description,
                 proto_components: Default::default(),
                 logger,
                 id_manager: IdManager::new(connector_id),
                 native_ports: Default::default(),
                 port_info: Default::default(),
             },
             phased: ConnectorPhased::Setup { endpoint_setups: Default::default(), surplus_sockets },
+        }
+    }
     pub fn new_net_port(
         &mut self,
         polarity: Polarity,
@@ @@ -226,49 +225,49 @@ fn new_endpoint_manager( @@
                             log!(logger, "Spurious wakeup on listener {:?}", index)
+                        }
                         Err(_) => {
                             log!(logger, "accept() failure on index {}", index);
                             return Err(AcceptFailed(listener.local_addr().unwrap()));
+                        }
+                    }
+                }
                 if let TodoEndpoint::Endpoint(endpoint) = &mut todo.todo_endpoint {
                     if event.is_error() {
                         if todo.endpoint_setup.endpoint_polarity == EndpointPolarity::Passive {
                             // right now you cannot retry an acceptor.
                             return Err(AcceptFailed(endpoint.stream.local_addr().unwrap()));
+                        }
                         connect_failed.insert(index);
                         if waker_continue_signal.is_none() {
                             log!(logger, "First connect failure. Starting waker thread");
                             let waker =
                                 Arc::new(mio::Waker::new(poll.registry(), WAKER_TOKEN).unwrap());
                             let wcs = Arc::new(AtomicBool::from(true));
                             let wcs2 = wcs.clone();
                             std::thread::spawn(move || {
                                 while wcs2.load(std::sync::atomic::Ordering::SeqCst) {
                                     std::thread::sleep(WAKER_PERIOD);
-                                    waker.wake().expect("unable to wake");
+                                    let _ = waker.wake();
+                                }
                             });
                             waker_continue_signal = Some(wcs);
+                        }
                         continue;
+                    }
                     if connect_failed.contains(&index) {
                         // spurious wakeup
                         continue;
+                    }
                     if !setup_incomplete.contains(&index) {
                         // spurious wakeup
                         continue;
+                    }
                     let local_polarity = *port_info.polarities.get(&todo.local_port).unwrap();
                     if event.is_writable() && !todo.sent_local_port {
                         let msg = Msg::SetupMsg(SetupMsg::MyPortInfo(MyPortInfo {
                             polarity: local_polarity,
                             port: todo.local_port,
                         }));
                         endpoint
                             .send(&msg)
                             .map_err(|e| {
                                 EndpointSetupError(endpoint.stream.local_addr().unwrap(), e)

0 comments (0 inline, 0 general)