CSY/reowolf Changeset - 4ef99b52a75d · Centrum Wiskunde & Informatica (CWI)

Changeset - 4ef99b52a75d

Parent rev.

Child rev.

[Not reviewed]

0 5 0

Christopher Esterhuyse - 5 years ago 2020-07-23 15:37:37
christopher.esterhuyse@gmail.com

tightened visibility. cleaned up and simplified UDP mediator components

5 files changed with 11 insertions and 24 deletions:

Cargo.toml

src/runtime/communication.rs

src/runtime/endpoints.rs

src/runtime/mod.rs

src/runtime/setup.rs

0 comments (0 inline, 0 general)

Cargo.toml

➞

Show inline comments

 [package]
 name = "reowolf_rs"
 version = "0.1.4"
 authors = [
 	"Christopher Esterhuyse <esterhuy@cwi.nl, christopher.esterhuyse@gmail.com>",
 	"Hans-Dieter Hiep <hdh@cwi.nl>"
+]
 edition = "2018"
 [dependencies]
 # convenience macros
 maplit = "1.0.2"
 derive_more = "0.99.2"
 # runtime
 bincode = "1.3.1"
 serde = { version = "1.0.114", features = ["derive"] }
 getrandom = "0.1.14" # tiny crate. used to guess controller-id
 # network
 mio = { version = "0.7.0", package = "mio", features = ["udp", "tcp", "os-poll"] }
 socket2 = { version = "0.3.12", optional = true }
 # protocol
 backtrace = "0.3"
 lazy_static = "1.4.0"
 # ffi
 # socket ffi
 libc = { version = "^0.2", optional = true }
 os_socketaddr = { version = "0.1.0", optional = true }
 [dev-dependencies]
 # test-generator = "0.3.0"
 crossbeam-utils = "0.7.2"
 lazy_static = "1.4.0"
 [lib]
 # compile target: dynamically linked library using C ABI
 crate-type = ["cdylib"]
 [features]
-default = ["ffi", "session_optimization", "ffi_pseudo_socket_api"]
 default = ["ffi", "session_optimization"]
 ffi = [] # see src/ffi/mod.rs
 ffi_pseudo_socket_api = ["ffi", "libc", "os_socketaddr"]# see src/ffi/pseudo_socket_api.rs
 endpoint_logging = [] # see src/macros.rs
 session_optimization = [] # see src/runtime/setup.rs
@@ \ No newline at end of file @@

src/runtime/communication.rs

➞

Show inline comments

@@ @@ -163,386 +163,384 @@ impl Connector { @@
+    }
     pub fn get(&mut self, port: PortId) -> Result<(), PortOpError> {
         use PortOpError as Poe;
         let batch = self.port_op_access(port, Getter)?;
         if batch.to_get.insert(port) {
             Ok(())
         } else {
             Err(Poe::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: cu, phased } = self;
         match phased {
             ConnectorPhased::Setup { .. } => Err(SyncError::NotConnected),
             ConnectorPhased::Communication(comm) => {
                 match &comm.round_result {
                     Err(SyncError::Unrecoverable(e)) => {
                         log!(cu.inner.logger, "Attempted to start sync round, but previous error {:?} was unrecoverable!", e);
                         return Err(SyncError::Unrecoverable(e.clone()));
+                    }
                     _ => {}
+                }
                 comm.round_result = Self::connected_sync(cu, comm, timeout);
                 comm.round_index += 1;
                 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;
         //////////////////////////////////
         log!(
             cu.inner.logger,
             "~~~ SYNC called with timeout {:?}; starting round {}",
             &timeout,
             comm.round_index
         );
         // 1. run all proto components to Nonsync blockers
         // NOTE: original components are immutable until Decision::Success
         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.inner.logger, "Nonsync running {} proto components...", unrun_components.len());
         // drains unrun_components, and populates branching_proto_components.
         while let Some((proto_component_id, mut component)) = unrun_components.pop() {
             // TODO coalesce fields
             log!(
                 cu.inner.logger,
                 "Nonsync running proto component with ID {:?}. {} to go after this",
                 proto_component_id,
                 unrun_components.len()
             );
             let mut ctx = NonsyncProtoContext {
                 cu_inner: &mut cu.inner,
                 proto_component_id,
                 unrun_components: &mut unrun_components,
                 proto_component_ports: &mut cu
                     .proto_components
                     .get_mut(&proto_component_id)
                     .unwrap() // unrun_components' keys originate from proto_components
                     .ports,
             };
             let blocker = component.state.nonsync_run(&mut ctx, &cu.proto_description);
             log!(
                 cu.inner.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.inner.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(SubtreeId::LocalComponent(ComponentId::Native));
                 let c = cu
                     .proto_components
                     .keys()
                     .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);
                 log!(
                     cu.inner.logger,
                     "Children in subtree are: {:?}",
                     subtree_id_iter.clone().collect::<Vec<_>>()
                 );
                 SolutionStorage::new(subtree_id_iter)
             },
             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 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, 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", index, var);
                         continue 'native_branches;
+                    }
+                }
                 // 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 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 {:?}", index, &msg);
                 rctx.getter_buffer.putter_add(cu, putter, msg);
+            }
             let branch = NativeBranch { index, gotten: Default::default(), to_get };
             if branch.is_ended() {
                 log!(
                     cu.inner.logger,
                     "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);
         // 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,
             &mut rctx,
         )?;
         log!(cu.inner.logger, "Committing to decision {:?}!", &decision);
         comm.endpoint_manager.udp_endpoints_round_end(&mut *cu.inner.logger, &decision)?;
         // propagate the decision to children
         let msg = Msg::CommMsg(CommMsg {
             round_index: comm.round_index,
             contents: CommMsgContents::CommCtrl(CommCtrlMsg::Announce {
                 decision: decision.clone(),
             }),
         });
         log!(
             cu.inner.logger,
             "Announcing decision {:?} through child endpoints {:?}",
             &msg,
             &comm.neighborhood.children
         );
         for &child in comm.neighborhood.children.iter() {
             comm.endpoint_manager.send_to_comms(child, &msg)?;
+        }
         let ret = match decision {
             Decision::Failure => {
                 // dropping {branching_proto_components, branching_native}
                 Err(Se::RoundFailure)
+            }
             Decision::Success(predicate) => {
                 // commit changes to component states
                 cu.proto_components.clear();
                 cu.proto_components.extend(
                     // consume branching proto components
                     branching_proto_components
                         .into_iter()
                         .map(|(id, bpc)| (id, bpc.collapse_with(&predicate))),
                 );
                 log!(
                     cu.inner.logger,
                     "End round with (updated) component states {:?}",
                     cu.proto_components.keys()
                 );
                 // consume native
                 Ok(Some(branching_native.collapse_with(&mut *cu.inner.logger, &predicate)))
+            }
         };
         log!(cu.inner.logger, "Sync round ending! Cleaning up");
         ret
+    }
     fn sync_reach_decision(
         cu: &mut ConnectorUnphased,
         comm: &mut ConnectorCommunication,
         branching_native: &mut BranchingNative,
         branching_proto_components: &mut HashMap<ProtoComponentId, BranchingProtoComponent>,
         rctx: &mut RoundCtx,
     ) -> Result<Decision, UnrecoverableSyncError> {
         let mut already_requested_failure = false;
         if branching_native.branches.is_empty() {
             log!(cu.inner.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 {
                         log!(cu.inner.logger, "Already requested failure");
+                    }
+                }
                 None => {
                     log!(cu.inner.logger, "No parent. Deciding on failure");
                     return Ok(Decision::Failure);
+                }
+            }
+        }
         log!(cu.inner.logger, "Done translating native batches into branches");
         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!(
             cu.inner.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;
             // must reborrow to constrain the lifetime of pcb_temps to inside the loop
             let (swap, pcb_temps) = pcb_temps.reborrow().split_first_mut();
             let (blocked, _pcb_temps) = pcb_temps.split_first_mut();
             // initially, no components have .ended==true
             // drain from branches --> blocked
             let cd = CyclicDrainer::new(branches, swap.0, blocked.0);
             BranchingProtoComponent::drain_branches_to_blocked(
                 cd,
                 cu,
                 rctx,
                 proto_component_id,
                 ports,
             )?;
             // swap the blocked branches back
             std::mem::swap(blocked.0, branches);
             if branches.is_empty() {
                 log!(cu.inner.logger, "{:?} has become inconsistent!", proto_component_id);
                 if let Some(parent) = comm.neighborhood.parent {
                     if already_requested_failure.replace_with_true() {
                         Self::request_failure(cu, comm, parent)?
                     } else {
                         log!(cu.inner.logger, "Already requested failure");
+                    }
                 } else {
                     log!(cu.inner.logger, "As the leader, deciding on timeout");
                     return Ok(Decision::Failure);
+                }
+            }
+        }
         log!(cu.inner.logger, "All proto components are blocked");
         log!(cu.inner.logger, "Entering decision loop...");
         comm.endpoint_manager.undelay_all();
         'undecided: loop {
             // drain payloads_to_get, sending them through endpoints / feeding them to components
             log!(
                 cu.inner.logger,
                 "Decision loop! have {} messages to recv",
                 rctx.getter_buffer.len()
             );
             while let Some((getter, send_payload_msg)) = rctx.getter_buffer.pop() {
                 assert!(cu.inner.port_info.polarities.get(&getter).copied() == Some(Getter));
                 let route = cu.inner.port_info.routes.get(&getter);
                 log!(
                     cu.inner.logger,
                     "Routing msg {:?} to {:?} via {:?}",
                     &send_payload_msg,
                     getter,
                     &route
                 );
                 match route {
                     None => log!(cu.inner.logger, "Delivery failed. Physical route unmapped!"),
                     Some(Route::UdpEndpoint { index }) => {
                         let udp_endpoint_ext =
                             &mut comm.endpoint_manager.udp_endpoint_store.endpoint_exts[*index];
                         let SendPayloadMsg { predicate, payload } = send_payload_msg;
                         log!(cu.inner.logger, "Delivering to udp endpoint index={}", index);
                         udp_endpoint_ext.outgoing_payloads.insert(predicate, payload);
+                    }
                     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,
                         rctx,
                         getter,
                         &send_payload_msg,
                         MapTempGuard::new(&mut bn_temp_owner),
                     ),
                     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;
                             branching_component.feed_msg(
                                 cu,
                                 rctx,
                                 proto_component_id,
                                 getter,
                                 &send_payload_msg,
                                 pcb_temps.reborrow(),
                             )?;
                             if branching_component.branches.is_empty() {
                                 log!(
                                     cu.inner.logger,
                                     "{:?} has become inconsistent!",
                                     proto_component_id
                                 );
                                 if let Some(parent) = comm.neighborhood.parent {
                                     if already_requested_failure.replace_with_true() {
                                         Self::request_failure(cu, comm, parent)?
                                     } else {
                                         log!(cu.inner.logger, "Already requested failure");
+                                    }
                                 } else {
                                     log!(cu.inner.logger, "As the leader, deciding on timeout");

src/runtime/endpoints.rs

➞

Show inline comments

@@ @@ -45,384 +45,373 @@ impl NetEndpoint { @@
             Ok(msg) => {
                 let msg_size = monitored.bytes_read();
                 self.inbox.drain(0..(msg_size.try_into().unwrap()));
                 endptlog!(
                     logger,
                     "Yielding msg. Inbox len {}-{}=={}: [{:?}]",
                     self.inbox.len() + msg_size,
                     msg_size,
                     self.inbox.len(),
                     DenseDebugHex(&self.inbox[..]),
                 );
                 Ok(Some(msg))
+            }
             Err(e) => match *e {
                 bincode::ErrorKind::Io(k) if k.kind() == std::io::ErrorKind::UnexpectedEof => {
                     Ok(None)
+                }
                 _ => Err(Nee::MalformedMessage),
             },
+        }
+    }
     pub(super) fn send<T: serde::ser::Serialize>(
         &mut self,
         msg: &T,
     ) -> Result<(), NetEndpointError> {
         use bincode::config::Options;
         use NetEndpointError as Nee;
         Self::bincode_opts()
             .serialize_into(&mut self.stream, msg)
             .map_err(|_| Nee::BrokenNetEndpoint)
+    }
+}
 impl EndpointManager {
     pub(super) fn index_iter(&self) -> Range<usize> {
 ..self.num_net_endpoints()
+    }
     pub(super) fn num_net_endpoints(&self) -> usize {
         self.net_endpoint_store.endpoint_exts.len()
+    }
     pub(super) fn send_to_comms(
         &mut self,
         index: usize,
         msg: &Msg,
     ) -> Result<(), UnrecoverableSyncError> {
         use UnrecoverableSyncError as Use;
         let net_endpoint = &mut self.net_endpoint_store.endpoint_exts[index].net_endpoint;
         net_endpoint.send(msg).map_err(|_| Use::BrokenNetEndpoint { index })
+    }
     pub(super) fn send_to_setup(&mut self, index: usize, msg: &Msg) -> Result<(), ConnectError> {
         let net_endpoint = &mut self.net_endpoint_store.endpoint_exts[index].net_endpoint;
         net_endpoint.send(msg).map_err(|err| {
             ConnectError::NetEndpointSetupError(net_endpoint.stream.local_addr().unwrap(), err)
         })
+    }
     /// Receive the first message of any kind at all.
     /// Why not return SetupMsg? Because often this message will be forwarded to several others,
     /// and by returning a Msg, it can be serialized in-place (NetEndpoints allow the sending of Msg types!)
     pub(super) fn try_recv_any_setup(
         &mut self,
         logger: &mut dyn Logger,
         deadline: &Option<Instant>,
     ) -> Result<(usize, Msg), ConnectError> {
         ///////////////////////////////////////////
         fn map_trane(
             trane: TryRecvAnyNetError,
             net_endpoint_store: &EndpointStore<NetEndpointExt>,
         ) -> ConnectError {
             ConnectError::NetEndpointSetupError(
                 net_endpoint_store.endpoint_exts[trane.index]
                     .net_endpoint
                     .stream
                     .local_addr()
                     .unwrap(), // stream must already be connected
                 trane.error,
+            )
+        }
         ///////////////////////////////////////////
         // try yield undelayed net message
         if let Some(tup) = self.undelayed_messages.pop() {
             endptlog!(logger, "RECV undelayed_msg {:?}", &tup);
             return Ok(tup);
+        }
         loop {
             // try recv from some polled undrained NET endpoint
             if let Some(tup) = self
                 .try_recv_undrained_net(logger)
                 .map_err(|trane| map_trane(trane, &self.net_endpoint_store))?
+            {
                 return Ok(tup);
+            }
             // poll if time remains
             self.poll_and_populate(logger, deadline)?;
+        }
+    }
     // drops all Setup messages,
     // buffers all future round messages,
     // drops all previous round messages,
     // enqueues all current round SendPayload messages using round_ctx.getter_add
     // returns the first comm_ctrl_msg encountered
     // only polls until SOME message is enqueued
     pub(super) fn try_recv_any_comms(
         &mut self,
         logger: &mut dyn Logger,
         port_info: &PortInfo,
         round_ctx: &mut impl RoundCtxTrait,
         round_index: usize,
     ) -> Result<CommRecvOk, UnrecoverableSyncError> {
         ///////////////////////////////////////////
         impl EndpointManager {
             fn handle_msg(
                 &mut self,
                 logger: &mut dyn Logger,
                 round_ctx: &mut impl RoundCtxTrait,
                 net_index: usize,
                 msg: Msg,
                 round_index: usize,
                 some_message_enqueued: &mut bool,
             ) -> Option<(usize, CommCtrlMsg)> {
                 let comm_msg_contents = match msg {
                     Msg::SetupMsg(..) => return None,
                     Msg::CommMsg(comm_msg) => match comm_msg.round_index.cmp(&round_index) {
                         Ordering::Equal => comm_msg.contents,
                         Ordering::Less => {
                             log!(
                                 logger,
                                 "We are in round {}, but msg is for round {}. Discard",
                                 comm_msg.round_index,
                                 round_index,
                             );
                             return None;
+                        }
                         Ordering::Greater => {
                             log!(
                                 logger,
                                 "We are in round {}, but msg is for round {}. Buffer",
                                 comm_msg.round_index,
                                 round_index,
                             );
                             self.delayed_messages.push((net_index, Msg::CommMsg(comm_msg)));
                             return None;
+                        }
                     },
                 };
                 match comm_msg_contents {
                     CommMsgContents::CommCtrl(comm_ctrl_msg) => Some((net_index, comm_ctrl_msg)),
                     CommMsgContents::SendPayload(send_payload_msg) => {
                         let getter =
                             self.net_endpoint_store.endpoint_exts[net_index].getter_for_incoming;
                         round_ctx.getter_add(getter, send_payload_msg);
                         *some_message_enqueued = true;
                         None
+                    }
+                }
+            }
+        }
         use {PollAndPopulateError as Pape, UnrecoverableSyncError as Use};
         ///////////////////////////////////////////
         let mut some_message_enqueued = false;
         // try yield undelayed net message
         while let Some((net_index, msg)) = self.undelayed_messages.pop() {
             if let Some((net_index, msg)) = self.handle_msg(
                 logger,
                 round_ctx,
                 net_index,
                 msg,
                 round_index,
                 &mut some_message_enqueued,
             ) {
                 return Ok(CommRecvOk::NewControlMsg { net_index, msg });
+            }
+        }
         loop {
             // try receive a net message
             while let Some((net_index, msg)) = self.try_recv_undrained_net(logger)? {
                 if let Some((net_index, msg)) = self.handle_msg(
                     logger,
                     round_ctx,
                     net_index,
                     msg,
                     round_index,
                     &mut some_message_enqueued,
                 ) {
                     return Ok(CommRecvOk::NewControlMsg { net_index, msg });
+                }
+            }
             // try receive a udp message
             let recv_buffer = self.udp_in_buffer.as_mut_slice();
             while let Some(index) = self.udp_endpoint_store.polled_undrained.pop() {
                 let ee = &mut self.udp_endpoint_store.endpoint_exts[index];
-                if let Some((bytes_written, from)) = ee.sock.recv_from(recv_buffer).ok() {
+                if let Some(bytes_written) = ee.sock.recv(recv_buffer).ok() {
                     // I received a payload!
                     self.udp_endpoint_store.polled_undrained.insert(index);
-                    if !ee.received_from_this_round.is_none() {
+                    if !ee.received_this_round {
                         let payload = Payload::from(&recv_buffer[..bytes_written]);
                         let port_spec_var = port_info.spec_var_for(ee.getter_for_incoming);
                         let predicate = Predicate::singleton(port_spec_var, SpecVal::FIRING);
                         round_ctx.getter_add(
                             ee.getter_for_incoming,
                             SendPayloadMsg { payload, predicate },
                         );
                         some_message_enqueued = true;
-                        ee.received_from_this_round = Some(from);
+                        ee.received_this_round = true;
                     } else {
                         // lose the message!
+                    }
+                }
+            }
             if some_message_enqueued {
                 return Ok(CommRecvOk::NewPayloadMsgs);
+            }
             // poll if time remains
             match self.poll_and_populate(logger, round_ctx.get_deadline()) {
                 Ok(()) => {} // continue looping
                 Err(Pape::Timeout) => return Ok(CommRecvOk::TimeoutWithoutNew),
                 Err(Pape::PollFailed) => return Err(Use::PollFailed),
+            }
+        }
+    }
     fn try_recv_undrained_net(
         &mut self,
         logger: &mut dyn Logger,
     ) -> Result<Option<(usize, Msg)>, TryRecvAnyNetError> {
         while let Some(index) = self.net_endpoint_store.polled_undrained.pop() {
             let net_endpoint = &mut self.net_endpoint_store.endpoint_exts[index].net_endpoint;
             if let Some(msg) = net_endpoint
                 .try_recv(logger)
                 .map_err(|error| TryRecvAnyNetError { error, index })?
+            {
                 endptlog!(logger, "RECV polled_undrained {:?}", &msg);
                 if !net_endpoint.inbox.is_empty() {
                     // there may be another message waiting!
                     self.net_endpoint_store.polled_undrained.insert(index);
+                }
                 return Ok(Some((index, msg)));
+            }
+        }
         Ok(None)
+    }
     fn poll_and_populate(
         &mut self,
         logger: &mut dyn Logger,
         deadline: &Option<Instant>,
     ) -> Result<(), PollAndPopulateError> {
         use PollAndPopulateError as Pape;
         // No message yet. Do we have enough time to poll?
         let remaining = if let Some(deadline) = deadline {
             Some(deadline.checked_duration_since(Instant::now()).ok_or(Pape::Timeout)?)
         } else {
             None
         };
         // Yes we do! Poll with remaining time as poll deadline
         self.poll.poll(&mut self.events, remaining).map_err(|_| Pape::PollFailed)?;
         for event in self.events.iter() {
             match TokenTarget::from(event.token()) {
                 TokenTarget::Waker => {
                     // Can ignore. Residual event from endpoint manager setup procedure
+                }
                 TokenTarget::NetEndpoint { index } => {
                     self.net_endpoint_store.polled_undrained.insert(index);
                     endptlog!(
                         logger,
                         "RECV poll event {:?} for NET endpoint index {:?}. undrained: {:?}",
                         &event,
                         index,
                         self.net_endpoint_store.polled_undrained.iter()
                     );
+                }
                 TokenTarget::UdpEndpoint { index } => {
                     self.udp_endpoint_store.polled_undrained.insert(index);
                     endptlog!(
                         logger,
                         "RECV poll event {:?} for UDP endpoint index {:?}. undrained: {:?}",
                         &event,
                         index,
                         self.udp_endpoint_store.polled_undrained.iter()
                     );
+                }
+            }
+        }
         self.events.clear();
         Ok(())
+    }
     pub(super) fn undelay_all(&mut self) {
         if self.undelayed_messages.is_empty() {
             // fast path
             std::mem::swap(&mut self.delayed_messages, &mut self.undelayed_messages);
             return;
+        }
         // slow path
         self.undelayed_messages.extend(self.delayed_messages.drain(..));
+    }
     pub(super) fn udp_endpoints_round_start(&mut self, logger: &mut dyn Logger) {
         log!(
             logger,
             "Starting round for {} udp endpoints",
             self.udp_endpoint_store.endpoint_exts.len()
         );
         for ee in self.udp_endpoint_store.endpoint_exts.iter_mut() {
             ee.received_from_this_round = None;
+        }
+    }
     pub(super) fn udp_endpoints_round_end(
         &mut self,
         logger: &mut dyn Logger,
         decision: &Decision,
     ) -> Result<(), UnrecoverableSyncError> {
         // retain received_from_this_round for use in pseudo_socket_api::recv_from
         log!(
             logger,
             "Ending round for {} udp endpoints",
             self.udp_endpoint_store.endpoint_exts.len()
         );
         use UnrecoverableSyncError as Use;
         if let Decision::Success(solution_predicate) = decision {
             'endpoint_loop: for (index, ee) in
                 self.udp_endpoint_store.endpoint_exts.iter_mut().enumerate()
+            {
                 for (payload_predicate, payload) in ee.outgoing_payloads.drain() {
             for (index, ee) in self.udp_endpoint_store.endpoint_exts.iter_mut().enumerate() {
                 'outgoing_loop: for (payload_predicate, payload) in ee.outgoing_payloads.drain() {
                     if payload_predicate.assigns_subset(solution_predicate) {
                         ee.sock.send(payload.as_slice()).map_err(|e| {
                             println!("{:?}", e);
                             Use::BrokenUdpEndpoint { index }
                         })?;
                         log!(
                             logger,
                             "Sent payload {:?} with pred {:?} through Udp endpoint {}",
                             &payload,
                             &payload_predicate,
                             index
                         );
                         continue 'endpoint_loop; // send at most one payload per endpoint per round
                         // send at most one payload per endpoint per round
                         break 'outgoing_loop;
+                    }
+                }
-                log!(logger, "Sent no message through Udp endpoint {}", index);
+                ee.received_this_round = false;
+            }
+        }
         Ok(())
+    }
+}
 impl Debug for NetEndpoint {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         f.debug_struct("Endpoint").field("inbox", &self.inbox).finish()
+    }
+}
 impl<R: Read> From<R> for MonitoredReader<R> {
     fn from(r: R) -> Self {
         Self { r, bytes: 0 }
+    }
+}
 impl<R: Read> MonitoredReader<R> {
     pub(super) fn bytes_read(&self) -> usize {
         self.bytes
+    }
+}
 impl<R: Read> Read for MonitoredReader<R> {
     fn read(&mut self, buf: &mut [u8]) -> Result<usize, std::io::Error> {
         let n = self.r.read(buf)?;
         self.bytes += n;
         Ok(n)
+    }
+}
 impl Into<Msg> for SetupMsg {
     fn into(self) -> Msg {
         Msg::SetupMsg(self)
+    }
+}
 impl From<PollAndPopulateError> for ConnectError {
     fn from(pape: PollAndPopulateError) -> ConnectError {
         use {ConnectError as Ce, PollAndPopulateError as Pape};
         match pape {
             Pape::PollFailed => Ce::PollFailed,
             Pape::Timeout => Ce::Timeout,
+        }
+    }
+}
 impl From<TryRecvAnyNetError> for UnrecoverableSyncError {
     fn from(trane: TryRecvAnyNetError) -> UnrecoverableSyncError {
         let TryRecvAnyNetError { index, .. } = trane;
         UnrecoverableSyncError::BrokenNetEndpoint { index }
+    }
+}

src/runtime/mod.rs

➞

Show inline comments

 /// cbindgen:ignore
 mod communication;
 /// cbindgen:ignore
 mod endpoints;
 pub mod error;
 /// cbindgen:ignore
 mod logging;
 /// cbindgen:ignore
 mod setup;
 #[cfg(test)]
 mod tests;
 use crate::common::*;
 use error::*;
 use mio::net::UdpSocket;
 #[derive(Debug)]
 pub struct Connector {
     unphased: ConnectorUnphased,
     phased: ConnectorPhased,
+}
 pub trait Logger: Debug + Send + Sync {
     fn line_writer(&mut self) -> Option<&mut dyn std::io::Write>;
+}
 #[derive(Debug)]
 pub struct VecLogger(ConnectorId, Vec<u8>);
 #[derive(Debug)]
 pub struct DummyLogger;
 #[derive(Debug)]
 pub struct FileLogger(ConnectorId, std::fs::File);
 pub(crate) struct NonsyncProtoContext<'a> {
     cu_inner: &'a mut ConnectorUnphasedInner, // persists between rounds
     proto_component_ports: &'a mut HashSet<PortId>, // sub-structure of component
     unrun_components: &'a mut Vec<(ProtoComponentId, ProtoComponent)>, // lives for Nonsync phase
     proto_component_id: ProtoComponentId,     // KEY in id->component map
+}
 pub(crate) struct SyncProtoContext<'a> {
     cu_inner: &'a mut ConnectorUnphasedInner, // persists between rounds
     branch_inner: &'a mut ProtoComponentBranchInner, // sub-structure of component branch
     predicate: &'a Predicate,                 // KEY in pred->branch map
+}
 #[derive(Default, Debug, Clone)]
 struct ProtoComponentBranchInner {
     untaken_choice: Option<u16>,
     did_put_or_get: HashSet<PortId>,
     inbox: HashMap<PortId, Payload>,
+}
 #[derive(
     Copy, Clone, Eq, PartialEq, Ord, Hash, PartialOrd, serde::Serialize, serde::Deserialize,
 )]
 struct SpecVar(PortId);
 #[derive(
     Copy, Clone, Eq, PartialEq, Ord, Hash, PartialOrd, serde::Serialize, serde::Deserialize,
 )]
 struct SpecVal(u16);
 #[derive(Debug)]
 struct RoundOk {
     batch_index: usize,
     gotten: HashMap<PortId, Payload>,
+}
 #[derive(Default)]
 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),
     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)]
 enum Msg {
     SetupMsg(SetupMsg),
     CommMsg(CommMsg),
+}
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 enum SetupMsg {
     MyPortInfo(MyPortInfo),
     LeaderWave { wave_leader: ConnectorId },
     LeaderAnnounce { tree_leader: ConnectorId },
     YouAreMyParent,
     SessionGather { unoptimized_map: HashMap<ConnectorId, SessionInfo> },
     SessionScatter { optimized_map: HashMap<ConnectorId, SessionInfo> },
+}
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 struct SessionInfo {
     serde_proto_description: SerdeProtocolDescription,
     port_info: PortInfo,
     endpoint_incoming_to_getter: Vec<PortId>,
     proto_components: HashMap<ProtoComponentId, ProtoComponent>,
+}
 #[derive(Debug, Clone)]
 struct SerdeProtocolDescription(Arc<ProtocolDescription>);
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 struct CommMsg {
     round_index: usize,
     contents: CommMsgContents,
+}
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 enum CommMsgContents {
     SendPayload(SendPayloadMsg),
     CommCtrl(CommCtrlMsg),
+}
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 enum CommCtrlMsg {
     Suggest { suggestion: Decision }, // SINKWARD
     Announce { decision: Decision },  // SINKAWAYS
+}
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 struct SendPayloadMsg {
     predicate: Predicate,
     payload: Payload,
+}
 #[derive(Debug, PartialEq)]
 enum AssignmentUnionResult {
     FormerNotLatter,
     LatterNotFormer,
     Equivalent,
     New(Predicate),
     Nonexistant,
+}
 struct NetEndpoint {
     inbox: Vec<u8>,
     stream: TcpStream,
+}
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 struct ProtoComponent {
     state: ComponentState,
     ports: HashSet<PortId>,
+}
 #[derive(Debug, Clone)]
 struct NetEndpointSetup {
     getter_for_incoming: PortId,
     sock_addr: SocketAddr,
     endpoint_polarity: EndpointPolarity,
+}
 #[derive(Debug, Clone)]
 struct UdpEndpointSetup {
     getter_for_incoming: PortId,
     local_addr: SocketAddr,
     peer_addr: SocketAddr,
+}
 #[derive(Debug)]
 struct NetEndpointExt {
     net_endpoint: NetEndpoint,
     getter_for_incoming: PortId,
+}
 #[derive(Debug)]
 struct UdpEndpointExt {
     sock: UdpSocket, // already bound and connected
-    received_from_this_round: Option<SocketAddr>,
+    received_this_round: bool,
     outgoing_payloads: HashMap<Predicate, Payload>,
     getter_for_incoming: PortId,
+}
 #[derive(Debug)]
 struct Neighborhood {
     parent: Option<usize>,
     children: VecSet<usize>,
+}
 #[derive(Debug)]
 struct IdManager {
     connector_id: ConnectorId,
     port_suffix_stream: U32Stream,
     proto_component_suffix_stream: U32Stream,
+}
 #[derive(Debug)]
 struct UdpInBuffer {
     byte_vec: Vec<u8>,
+}
 #[derive(Debug)]
 struct SpecVarStream {
     connector_id: ConnectorId,
     port_suffix_stream: U32Stream,
+}
 #[derive(Debug)]
 struct EndpointManager {
     // invariants:
     // 1. net and udp endpoints are registered with poll. Poll token computed with TargetToken::into
     // 2. Events is empty
     poll: Poll,
     events: Events,
     delayed_messages: Vec<(usize, Msg)>,
     undelayed_messages: Vec<(usize, Msg)>,
     net_endpoint_store: EndpointStore<NetEndpointExt>,
     udp_endpoint_store: EndpointStore<UdpEndpointExt>,
     udp_in_buffer: UdpInBuffer,
+}
 #[derive(Debug)]
 struct EndpointStore<T> {
     endpoint_exts: Vec<T>,
     polled_undrained: VecSet<usize>,
+}
 #[derive(Clone, Debug, Default, serde::Serialize, serde::Deserialize)]
 struct PortInfo {
     polarities: HashMap<PortId, Polarity>,
     peers: HashMap<PortId, PortId>,
     routes: HashMap<PortId, Route>,
+}
 #[derive(Debug)]
 struct ConnectorCommunication {
     round_index: usize,
     endpoint_manager: EndpointManager,
     neighborhood: Neighborhood,
     native_batches: Vec<NativeBatch>,
     round_result: Result<Option<RoundOk>, SyncError>,
+}
 #[derive(Debug)]
 struct ConnectorUnphased {
     proto_description: Arc<ProtocolDescription>,
     proto_components: HashMap<ProtoComponentId, ProtoComponent>,
     inner: ConnectorUnphasedInner,
+}
 #[derive(Debug)]
 struct ConnectorUnphasedInner {
     logger: Box<dyn Logger>,
     id_manager: IdManager,
     native_ports: HashSet<PortId>,
     port_info: PortInfo,
+}
 #[derive(Debug)]
 struct ConnectorSetup {
     net_endpoint_setups: Vec<NetEndpointSetup>,
     udp_endpoint_setups: Vec<UdpEndpointSetup>,
+}
 #[derive(Debug)]
 enum ConnectorPhased {
     Setup(Box<ConnectorSetup>),
     Communication(Box<ConnectorCommunication>),
+}
 #[derive(Default, Clone, Eq, PartialEq, Hash, serde::Serialize, serde::Deserialize)]
 struct Predicate {
     assigned: BTreeMap<SpecVar, SpecVal>,
+}
 #[derive(Debug, Default)]
 struct NativeBatch {
     // invariant: putters' and getters' polarities respected
     to_put: HashMap<PortId, Payload>,
     to_get: HashSet<PortId>,
+}
 #[derive(Debug, Copy, Clone, Eq, PartialEq, Hash)]
 enum TokenTarget {
     NetEndpoint { index: usize },
     UdpEndpoint { index: usize },
     Waker,
+}
 trait RoundCtxTrait {
     fn get_deadline(&self) -> &Option<Instant>;
     fn getter_add(&mut self, getter: PortId, msg: SendPayloadMsg);
+}
 enum CommRecvOk {
     TimeoutWithoutNew,
     NewPayloadMsgs,
     NewControlMsg { net_index: usize, msg: CommCtrlMsg },
+}
 ////////////////
 fn would_block(err: &std::io::Error) -> bool {
     err.kind() == std::io::ErrorKind::WouldBlock
+}
 impl TokenTarget {
     const HALFWAY_INDEX: usize = usize::MAX / 2;
     const MAX_INDEX: usize = usize::MAX;
     const WAKER_TOKEN: usize = Self::MAX_INDEX;
+}
 impl From<Token> for TokenTarget {
     fn from(Token(index): Token) -> Self {
         if index == Self::WAKER_TOKEN {
             TokenTarget::Waker
         } else if let Some(shifted) = index.checked_sub(Self::HALFWAY_INDEX) {
             TokenTarget::UdpEndpoint { index: shifted }
         } else {
             TokenTarget::NetEndpoint { index }
+        }
+    }
+}
 impl Into<Token> for TokenTarget {
     fn into(self) -> Token {
         match self {
             TokenTarget::Waker => Token(Self::WAKER_TOKEN),
             TokenTarget::UdpEndpoint { index } => Token(index + Self::HALFWAY_INDEX),
             TokenTarget::NetEndpoint { index } => Token(index),
+        }
+    }
+}
 impl<T: std::cmp::Ord> VecSet<T> {
     fn new(mut vec: Vec<T>) -> Self {
         vec.sort();
         vec.dedup();
         Self { vec }
+    }
     fn contains(&self, element: &T) -> bool {
         self.vec.binary_search(element).is_ok()
+    }
     fn insert(&mut self, element: T) -> bool {
         match self.vec.binary_search(&element) {
             Ok(_) => false,
             Err(index) => {
                 self.vec.insert(index, element);
                 true
+            }
+        }
+    }
     fn iter(&self) -> std::slice::Iter<T> {
         self.vec.iter()
+    }
     fn pop(&mut self) -> Option<T> {
         self.vec.pop()
+    }
+}
 impl PortInfo {
     fn spec_var_for(&self, port: PortId) -> SpecVar {
         SpecVar(match self.polarities.get(&port).unwrap() {
             Getter => port,
             Putter => *self.peers.get(&port).unwrap(),
         })
+    }
+}
 impl SpecVarStream {
     fn next(&mut self) -> SpecVar {
         let phantom_port: PortId =
             Id { connector_id: self.connector_id, u32_suffix: self.port_suffix_stream.next() }
                 .into();
         SpecVar(phantom_port)
+    }
+}
 impl IdManager {
     fn new(connector_id: ConnectorId) -> Self {
         Self {
             connector_id,
             port_suffix_stream: Default::default(),
             proto_component_suffix_stream: Default::default(),
+        }
+    }
     fn new_spec_var_stream(&self) -> SpecVarStream {
         // Spec var stream starts where the current port_id stream ends, with gap of SKIP_N.
         // This gap is entirely unnecessary (i.e. 0 is fine)
         // It's purpose is only to make SpecVars easier to spot in logs.
         // E.g. spot the spec var: { v0_0, v1_2, v1_103 }
         const SKIP_N: u32 = 100;
         let port_suffix_stream = self.port_suffix_stream.clone().n_skipped(SKIP_N);
         SpecVarStream { connector_id: self.connector_id, port_suffix_stream }
+    }
     fn new_port_id(&mut self) -> PortId {
         Id { connector_id: self.connector_id, u32_suffix: self.port_suffix_stream.next() }.into()

src/runtime/setup.rs

➞

Show inline comments

@@ @@ -322,385 +322,385 @@ fn new_endpoint_manager( @@
                                 // successfully accepted the active peer
                                 // reusing the token, but now for the stream and not the listener
                                 poll.registry().deregister(listener).unwrap();
                                 poll.registry().register(&mut stream, token, BOTH).unwrap();
                                 log!(
                                     logger,
                                     "Endpoint[{}] accepted a connection from {:?}",
                                     index,
                                     peer_addr
                                 );
                                 let net_endpoint = NetEndpoint { stream, inbox: vec![] };
                                 net_todo.todo_endpoint = TodoEndpoint::NetEndpoint(net_endpoint);
+                            }
+                        }
+                    }
                     if let TodoEndpoint::NetEndpoint(net_endpoint) = &mut net_todo.todo_endpoint {
                         if event.is_error() {
                             if net_todo.endpoint_setup.endpoint_polarity
                                 == EndpointPolarity::Passive
+                            {
                                 // right now you cannot retry an acceptor. return failure
                                 return Err(Ce::AcceptFailed(
                                     net_endpoint.stream.local_addr().unwrap(),
                                 ));
+                            }
                             // this actively-connecting endpoint failed to connect!
                             if net_connect_retry_later.insert(index) {
                                 log!(
                                     logger,
                                     "Connection failed for {:?}. List is {:?}",
                                     index,
                                     net_connect_retry_later.iter()
                                 );
                                 poll.registry().deregister(&mut net_endpoint.stream).unwrap();
                             } else {
                                 // spurious wakeup. already scheduled to retry connect later
                                 continue;
+                            }
                             if waker_state.is_none() {
                                 log!(logger, "First connect failure. Starting waker thread");
                                 let arc = Arc::new(WakerState {
                                     waker: mio::Waker::new(
                                         poll.registry(),
                                         TokenTarget::Waker.into(),
+                                    )
                                     .unwrap(),
                                     continue_signal: true.into(),
                                 });
                                 let moved_arc = arc.clone();
                                 waker_state = Some(arc);
                                 std::thread::spawn(move || moved_arc.waker_loop());
+                            }
                             continue;
+                        }
                         // event wasn't ERROR
                         if net_connect_retry_later.contains(&index) {
                             // spurious wakeup. already scheduled to retry connect later
                             continue;
+                        }
                         if !setup_incomplete.contains(&token_target) {
                             // spurious wakeup. this endpoint has already been completed!
                             if event.is_readable() {
                                 net_polled_undrained.insert(index);
+                            }
                             continue;
+                        }
                         let local_polarity = *port_info
                             .polarities
                             .get(&net_todo.endpoint_setup.getter_for_incoming)
                             .unwrap();
                         if event.is_writable() && !net_todo.sent_local_port {
                             // can write and didn't send setup msg yet? Do so!
                             let msg = Msg::SetupMsg(SetupMsg::MyPortInfo(MyPortInfo {
                                 polarity: local_polarity,
                                 port: net_todo.endpoint_setup.getter_for_incoming,
                             }));
                             net_endpoint
                                 .send(&msg)
                                 .map_err(|e| {
                                     Ce::NetEndpointSetupError(
                                         net_endpoint.stream.local_addr().unwrap(),
                                         e,
+                                    )
                                 })
                                 .unwrap();
                             log!(logger, "endpoint[{}] sent msg {:?}", index, &msg);
                             net_todo.sent_local_port = true;
+                        }
                         if event.is_readable() && net_todo.recv_peer_port.is_none() {
                             // can read and didn't recv setup msg yet? Do so!
                             let maybe_msg = net_endpoint.try_recv(logger).map_err(|e| {
                                 Ce::NetEndpointSetupError(
                                     net_endpoint.stream.local_addr().unwrap(),
                                     e,
+                                )
                             })?;
                             if maybe_msg.is_some() && !net_endpoint.inbox.is_empty() {
                                 net_polled_undrained.insert(index);
+                            }
                             match maybe_msg {
                                 None => {} // msg deserialization incomplete
                                 Some(Msg::SetupMsg(SetupMsg::MyPortInfo(peer_info))) => {
                                     log!(
                                         logger,
                                         "endpoint[{}] got peer info {:?}",
                                         index,
                                         peer_info
                                     );
                                     if peer_info.polarity == local_polarity {
                                         return Err(ConnectError::PortPeerPolarityMismatch(
                                             net_todo.endpoint_setup.getter_for_incoming,
                                         ));
+                                    }
                                     net_todo.recv_peer_port = Some(peer_info.port);
                                     // 1. finally learned the peer of this port!
                                     port_info.peers.insert(
                                         net_todo.endpoint_setup.getter_for_incoming,
                                         peer_info.port,
                                     );
                                     // 2. learned the info of this peer port
                                     port_info.polarities.insert(peer_info.port, peer_info.polarity);
                                     port_info.peers.insert(
                                         peer_info.port,
                                         net_todo.endpoint_setup.getter_for_incoming,
                                     );
                                     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::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?
                         if net_todo.sent_local_port && net_todo.recv_peer_port.is_some() {
                             // yes! connected, sent my info and received peer's info
                             setup_incomplete.remove(&token_target);
                             log!(logger, "endpoint[{}] is finished!", index);
+                        }
+                    }
+                }
+            }
+        }
         events.clear();
+    }
     log!(logger, "Endpoint setup complete! Cleaning up and building structures");
     if let Some(ws) = waker_state.take() {
         ws.waker_stop();
+    }
     let net_endpoint_exts = net_todos
         .into_iter()
         .enumerate()
         .map(|(index, Todo { todo_endpoint, endpoint_setup, .. })| NetEndpointExt {
             net_endpoint: match todo_endpoint {
                 TodoEndpoint::NetEndpoint(mut net_endpoint) => {
                     let token = TokenTarget::NetEndpoint { index }.into();
                     poll.registry()
                         .reregister(&mut net_endpoint.stream, token, Interest::READABLE)
                         .unwrap();
                     net_endpoint
+                }
                 _ => unreachable!(),
             },
             getter_for_incoming: endpoint_setup.getter_for_incoming,
         })
         .collect();
     let udp_endpoint_exts = udp_todos
         .into_iter()
         .enumerate()
         .map(|(index, udp_todo)| {
             let UdpTodo { mut sock, getter_for_incoming } = udp_todo;
             let token = TokenTarget::UdpEndpoint { index }.into();
             poll.registry().reregister(&mut sock, token, Interest::READABLE).unwrap();
             UdpEndpointExt {
                 sock,
                 outgoing_payloads: Default::default(),
-                received_from_this_round: None,
+                received_this_round: false,
                 getter_for_incoming,
+            }
         })
         .collect();
     Ok(EndpointManager {
         poll,
         events,
         undelayed_messages: delayed_messages, // no longer delayed
         delayed_messages: Default::default(),
         net_endpoint_store: EndpointStore {
             endpoint_exts: net_endpoint_exts,
             polled_undrained: net_polled_undrained,
         },
         udp_endpoint_store: EndpointStore {
             endpoint_exts: udp_endpoint_exts,
             polled_undrained: udp_polled_undrained,
         },
         udp_in_buffer: Default::default(),
     })
+}
 fn init_neighborhood(
     connector_id: ConnectorId,
     logger: &mut dyn Logger,
     em: &mut EndpointManager,
     deadline: &Option<Instant>,
 ) -> Result<Neighborhood, ConnectError> {
     ////////////////////////////////
     use {ConnectError as Ce, Msg::SetupMsg as S, SetupMsg as Sm};
     #[derive(Debug)]
     struct WaveState {
         parent: Option<usize>,
         leader: ConnectorId,
+    }
     fn do_wave(
         em: &mut EndpointManager,
         awaiting: &mut HashSet<usize>,
         ws: &WaveState,
     ) -> Result<(), ConnectError> {
         awaiting.clear();
         let msg = S(Sm::LeaderWave { wave_leader: ws.leader });
         for index in em.index_iter() {
             if Some(index) != ws.parent {
                 em.send_to_setup(index, &msg)?;
                 awaiting.insert(index);
+            }
+        }
         Ok(())
+    }
     ///////////////////////
     /*
     Conceptually, we have two distinct disstributed algorithms back-to-back
 . Leader election using echo algorithm with extinction.
         - Each connector initiates a wave tagged with their ID
         - Connectors participate in waves of GREATER ID, abandoning previous waves
         - Only the wave of the connector with GREATEST ID completes, whereupon they are the leader
 . Tree construction
         - The leader broadcasts their leadership with msg A
         - Upon receiving their first announcement, connectors reply B, and send A to all peers
         - A controller exits once they have received A or B from each neighbor
     The actual implementation is muddier, because non-leaders aren't aware of termiantion of algorithm 1,
     so they rely on receipt of the leader's announcement to realize that algorithm 2 has begun.
     NOTE the distinction between PARENT and LEADER
     */
     log!(logger, "beginning neighborhood construction");
     if em.num_net_endpoints() == 0 {
         log!(logger, "Edge case of no neighbors! No parent an no children!");
         return Ok(Neighborhood { parent: None, children: VecSet::new(vec![]) });
+    }
     log!(logger, "Have {} endpoints. Must participate in distributed alg.", em.num_net_endpoints());
     let mut awaiting = HashSet::with_capacity(em.num_net_endpoints());
     // 1+ neighbors. Leader can only be learned by receiving messages
     // loop ends when I know my sink tree parent (implies leader was elected)
     let election_result: WaveState = {
         // initially: No parent, I'm the best leader.
         let mut best_wave = WaveState { parent: None, leader: connector_id };
         // start a wave for this initial state
         do_wave(em, &mut awaiting, &best_wave)?;
         // with 1+ neighbors, progress is only made in response to incoming messages
         em.undelay_all();
         'election: loop {
             log!(logger, "Election loop. awaiting {:?}...", awaiting.iter());
             let (recv_index, msg) = em.try_recv_any_setup(logger, deadline)?;
             log!(logger, "Received from index {:?} msg {:?}", &recv_index, &msg);
             match msg {
                 S(Sm::LeaderAnnounce { tree_leader }) => {
                     let election_result =
                         WaveState { leader: tree_leader, parent: Some(recv_index) };
                     log!(logger, "Election lost! Result {:?}", &election_result);
                     assert!(election_result.leader >= best_wave.leader);
                     assert_ne!(election_result.leader, connector_id);
                     break 'election election_result;
+                }
                 S(Sm::LeaderWave { wave_leader }) => {
                     use Ordering as O;
                     match wave_leader.cmp(&best_wave.leader) {
                         O::Less => log!(
                             logger,
                             "Ignoring wave with Id {:?}<{:?}",
                             wave_leader,
                             best_wave.leader
                         ),
                         O::Greater => {
                             log!(
                                 logger,
                                 "Joining wave with Id {:?}>{:?}",
                                 wave_leader,
                                 best_wave.leader
                             );
                             best_wave = WaveState { leader: wave_leader, parent: Some(recv_index) };
                             log!(logger, "New wave state {:?}", &best_wave);
                             do_wave(em, &mut awaiting, &best_wave)?;
                             if awaiting.is_empty() {
                                 log!(logger, "Special case! Only neighbor is parent. Replying to {:?} msg {:?}", recv_index, &msg);
                                 em.send_to_setup(recv_index, &msg)?;
+                            }
+                        }
                         O::Equal => {
                             assert!(awaiting.remove(&recv_index));
                             log!(
                                 logger,
                                 "Wave reply from index {:?} for leader {:?}. Now awaiting {} replies",
                                 recv_index,
                                 best_wave.leader,
                                 awaiting.len()
                             );
                             if awaiting.is_empty() {
                                 if let Some(parent) = best_wave.parent {
                                     log!(
                                         logger,
                                         "Sub-wave done! replying to parent {:?} msg {:?}",
                                         parent,
                                         &msg
                                     );
                                     em.send_to_setup(parent, &msg)?;
                                 } else {
                                     let election_result: WaveState = best_wave;
                                     log!(logger, "Election won! Result {:?}", &election_result);
                                     break 'election election_result;
+                                }
+                            }
+                        }
+                    }
+                }
                 msg @ S(Sm::YouAreMyParent) | msg @ S(Sm::MyPortInfo(_)) => {
                     log!(logger, "Endpont {:?} sent unexpected msg! {:?}", recv_index, &msg);
                     return Err(Ce::SetupAlgMisbehavior);
+                }
                 msg @ S(Sm::SessionScatter { .. })
                 | msg @ S(Sm::SessionGather { .. })
                 | msg @ Msg::CommMsg { .. } => {
                     log!(logger, "delaying msg {:?} during election algorithm", msg);
                     em.delayed_messages.push((recv_index, msg));
+                }
+            }
+        }
     };
     // starting algorithm 2. Send a message to every neighbor
     log!(logger, "Starting tree construction. Step 1: send one msg per neighbor");
     awaiting.clear();
     for index in em.index_iter() {
         if Some(index) == election_result.parent {
             em.send_to_setup(index, &S(Sm::YouAreMyParent))?;
         } else {
             awaiting.insert(index);
             em.send_to_setup(
                 index,
                 &S(Sm::LeaderAnnounce { tree_leader: election_result.leader }),
             )?;
+        }
+    }
     let mut children = vec![];
     em.undelay_all();
     while !awaiting.is_empty() {
         log!(logger, "Tree construction_loop loop. awaiting {:?}...", awaiting.iter());
         let (recv_index, msg) = em.try_recv_any_setup(logger, deadline)?;
         log!(logger, "Received from index {:?} msg {:?}", &recv_index, &msg);
         match msg {
             S(Sm::LeaderAnnounce { .. }) => {
                 // not a child
                 log!(
                     logger,
                     "Got reply from non-child index {:?}. Children: {:?}",
                     recv_index,
                     children.iter()
                 );
                 if !awaiting.remove(&recv_index) {
                     return Err(Ce::SetupAlgMisbehavior);
+                }

0 comments (0 inline, 0 general)