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

Changeset - b20c3a55156d

Parent rev.

Child rev.

[Not reviewed]

0 6 0

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

dummy scatter/gather wave in

6 files changed with 222 insertions and 78 deletions:

src/lib.rs

src/runtime/communication.rs

src/runtime/endpoints.rs

src/runtime/mod.rs

src/runtime/setup.rs

167

src/runtime/tests.rs

0 comments (0 inline, 0 general)

src/lib.rs

➞

Show inline comments

@@ @@ -7,10 +7,10 @@ mod runtime; @@
 // #[cfg(test)]
 // mod test;
 pub use common::{ConnectorId, EndpointPolarity, Polarity, PortId};
 pub use protocol::ProtocolDescription;
 pub use runtime::{error, Connector, FileLogger, VecLogger};
+pub use runtime::{error, Connector, DummyLogger, FileLogger, VecLogger};
 // #[cfg(feature = "ffi")]
 // pub use runtime::ffi;

src/runtime/communication.rs

➞

Show inline comments

@@ @@ -34,12 +34,15 @@ struct CyclicDrainer<'a, K: Eq + Hash, 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);
+}
 ////////////////
 impl Connector {
     pub fn gotten(&mut self, port: PortId) -> Result<&Payload, GottenError> {
         use GottenError::*;
         let Self { phased, .. } = self;
@@ @@ -119,28 +122,29 @@ impl Connector { @@
                     Ok(Some(ok_result)) => Ok(ok_result.batch_index),
                     Err(sync_error) => Err(sync_error.clone()),
+                }
+            }
+        }
+    }
     // TODO make cu immutable
     // 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);
         // 1. run all proto components to Nonsync blockers
         log!(
             cu.logger,
             "~~~ SYNC called with timeout {:?}; starting round {}",
             &timeout,
             comm.round_index
         );
         // 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() {
@@ @@ -259,13 +263,12 @@ impl Connector { @@
         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 ConnectorUnphased { port_info, proto_description, .. } = cu;
             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(
@@ @@ -512,13 +515,12 @@ impl Connector { @@
+}
 impl BranchingNative {
     fn feed_msg(
         &mut self,
         cu: &mut ConnectorUnphased,
         solution_storage: &mut SolutionStorage,
         // payloads_to_get: &mut Vec<(PortId, CommMsgContents)>,
         getter: PortId,
         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();
@@ @@ -614,13 +616,13 @@ impl BranchingNative { @@
 // },
 impl BranchingProtoComponent {
     fn drain_branches_to_blocked(
         cd: CyclicDrainer<Predicate, ProtoComponentBranch>,
         cu: &mut ConnectorUnphased,
         solution_storage: &mut SolutionStorage,
-        payloads_to_get: &mut Vec<(PortId, SendPayloadMsg)>,
+        payload_msg_sender: &mut impl PayloadMsgSender,
         proto_component_id: ProtoComponentId,
         ports: &HashSet<PortId>,
     ) {
         cd.cylic_drain(|mut predicate, mut branch, mut drainer| {
             let mut ctx = SyncProtoContext {
                     logger: &mut *cu.logger,
@@ @@ -680,27 +682,26 @@ impl BranchingProtoComponent { @@
                             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 getter = *cu.port_info.peers.get(&putter).unwrap();
                             let msg = SendPayloadMsg { predicate: predicate.clone(), payload };
-    payloads_to_get.push((getter, msg));
+                            payload_msg_sender.send(&cu.port_info, &putter, msg);
                             drainer.add_input(predicate, branch);
+                        }
+                    }
+                }
         });
+    }
     fn feed_msg(
         &mut self,
         cu: &mut ConnectorUnphased,
         solution_storage: &mut SolutionStorage,
         proto_component_id: ProtoComponentId,
-        payloads_to_get: &mut Vec<(PortId, SendPayloadMsg)>,
+        payload_msg_sender: &mut impl PayloadMsgSender,
         getter: PortId,
         send_payload_msg: SendPayloadMsg,
     ) {
         let logger = &mut *cu.logger;
         log!(
             logger,
@@ @@ -753,13 +754,13 @@ impl BranchingProtoComponent { @@
         let mut swap = HashMap::default();
         let cd = CyclicDrainer::new(&mut unblocked, &mut swap, &mut blocked);
         BranchingProtoComponent::drain_branches_to_blocked(
             cd,
             cu,
             solution_storage,
-            payloads_to_get,
+            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());
@@ @@ -803,13 +804,13 @@ impl SolutionStorage { @@
     fn clear(&mut self) {
         self.subtree_id_to_index.clear();
         self.subtree_solutions.clear();
         self.old_local.clear();
         self.new_local.clear();
+    }
-    pub(crate) fn reset(&mut self, subtree_ids: impl Iterator<Item = Route>) {
     fn reset(&mut self, subtree_ids: impl Iterator<Item = Route>) {
         self.subtree_id_to_index.clear();
         self.subtree_solutions.clear();
         self.old_local.clear();
         self.new_local.clear();
         for key in subtree_ids {
             self.subtree_id_to_index.insert(key, self.subtree_solutions.len());
@@ @@ -877,18 +878,24 @@ impl SolutionStorage { @@
                 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));
+    }
+}
 impl SyncProtoContext<'_> {
     pub fn is_firing(&mut self, port: PortId) -> Option<bool> {
+    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 fn read_msg(&mut self, port: PortId) -> Option<&Payload> {
+    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);

src/runtime/endpoints.rs

➞

Show inline comments

@@ @@ -14,13 +14,13 @@ enum TryRecyAnyError { @@
 /////////////////////
 fn would_block(err: &std::io::Error) -> bool {
     err.kind() == std::io::ErrorKind::WouldBlock
+}
 impl Endpoint {
     pub fn try_recv<T: serde::de::DeserializeOwned>(
+    pub(super) fn try_recv<T: serde::de::DeserializeOwned>(
         &mut self,
         logger: &mut dyn Logger,
     ) -> Result<Option<T>, EndpointError> {
         use EndpointError::*;
         // populate inbox as much as possible
         'read_loop: loop {
@@ @@ -47,47 +47,47 @@ impl Endpoint { @@
                 _ => Err(MalformedMessage),
                 // println!("SERDE ERRKIND {:?}", e);
                 // Err(MalformedMessage)
             },
+        }
+    }
     pub fn send<T: serde::ser::Serialize>(&mut self, msg: &T) -> Result<(), EndpointError> {
+    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 fn index_iter(&self) -> Range<usize> {
+    pub(super) fn index_iter(&self) -> Range<usize> {
 ..self.num_endpoints()
+    }
     pub fn num_endpoints(&self) -> usize {
+    pub(super) fn num_endpoints(&self) -> usize {
         self.endpoint_exts.len()
+    }
     pub fn send_to_setup(&mut self, index: usize, msg: &Msg) -> Result<(), ConnectError> {
+    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 fn send_to(&mut self, index: usize, msg: &Msg) -> Result<(), EndpointError> {
+    pub(super) fn send_to(&mut self, index: usize, msg: &Msg) -> Result<(), EndpointError> {
         self.endpoint_exts[index].endpoint.send(msg)
+    }
     pub fn try_recv_any_comms(
+    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 fn try_recv_any_setup(
+    pub(super) fn try_recv_any_setup(
         &mut self,
         logger: &mut dyn Logger,
         deadline: Option<Instant>,
     ) -> Result<(usize, Msg), ConnectError> {
         use {ConnectError as Ce, TryRecyAnyError as Trae};
         self.try_recv_any(logger, deadline).map_err(|err| match err {
@@ @@ -116,16 +116,16 @@ impl EndpointManager { @@
                 let endpoint = &mut self.endpoint_exts[index].endpoint;
                 if let Some(msg) = endpoint
                     .try_recv(logger)
                     .map_err(|error| Trea::EndpointError { error, index })?
+                {
                     endptlog!(logger, "RECV polled_undrained {:?}", &msg);
                     // if !endpoint.inbox.is_empty() {
                     // there may be another message waiting!
                     self.polled_undrained.insert(index);
                     // }
                     if !endpoint.inbox.is_empty() {
                         // there may be another message waiting!
                         self.polled_undrained.insert(index);
+                    }
                     return Ok((index, msg));
+                }
+            }
             // 3. 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(Trea::Timeout)?)
@@ @@ -140,23 +140,17 @@ impl EndpointManager { @@
                     logger,
                     "RECV poll event {:?} for endpoint index {:?}. undrained: {:?}",
                     &event,
                     index,
                     self.polled_undrained.iter()
                 );
                 if event.is_error() {
                     return Err(Trea::EndpointError {
                         error: EndpointError::BrokenEndpoint,
                         index,
                     });
+                }
+            }
             self.events.clear();
+        }
+    }
     pub fn undelay_all(&mut self) {
+    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
@@ @@ -171,13 +165,13 @@ impl Debug for Endpoint { @@
 impl<R: Read> From<R> for MonitoredReader<R> {
     fn from(r: R) -> Self {
         Self { r, bytes: 0 }
+    }
+}
 impl<R: Read> MonitoredReader<R> {
     pub fn bytes_read(&self) -> usize {
+    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)?;

src/runtime/mod.rs

➞

Show inline comments

@@ @@ -48,13 +48,19 @@ pub enum Msg { @@
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 pub 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)]
 pub struct SessionInfo {}
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 pub struct CommMsg {
     pub round_index: usize,
     pub contents: CommMsgContents,
+}
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
@@ @@ -189,13 +195,13 @@ pub struct SyncProtoContext<'a> { @@
     predicate: &'a Predicate,
     port_info: &'a PortInfo,
     inbox: &'a HashMap<PortId, Payload>,
+}
 ////////////////
 impl<T: std::cmp::Ord> VecSet<T> {
-    fn iter(&self) -> impl Iterator<Item = &T> {
+    fn iter(&self) -> std::slice::Iter<T> {
         self.vec.iter()
+    }
     fn contains(&self, element: &T) -> bool {
         self.vec.binary_search(element).is_ok()
+    }
     fn new(mut vec: Vec<T>) -> Self {

src/runtime/setup.rs

➞

Show inline comments

 use crate::common::*;
 use crate::runtime::*;
 use std::io::ErrorKind::WouldBlock;
 impl Connector {
     pub fn new_simple(
         proto_description: Arc<ProtocolDescription>,
         connector_id: ConnectorId,
     ) -> Self {
         let logger = Box::new(DummyLogger);
         // let logger = Box::new(DummyLogger);
         let surplus_sockets = 2;
         Self::new(logger, proto_description, connector_id, surplus_sockets)
+    }
     pub fn new(
         mut logger: Box<dyn Logger>,
         proto_description: Arc<ProtocolDescription>,
         connector_id: ConnectorId,
         surplus_sockets: u16,
     ) -> Self {
@@ @@ -58,51 +49,52 @@ impl Connector { @@
+            }
             ConnectorPhased::Communication { .. } => Err(()),
+        }
+    }
     pub fn connect(&mut self, timeout: Option<Duration>) -> Result<(), ConnectError> {
         use ConnectError::*;
-        let Self { unphased: up, phased } = self;
+        let Self { unphased: cu, phased } = self;
         match phased {
             ConnectorPhased::Communication { .. } => {
-                log!(up.logger, "Call to connecting in connected state");
+                log!(cu.logger, "Call to connecting in connected state");
                 Err(AlreadyConnected)
+            }
             ConnectorPhased::Setup { endpoint_setups, .. } => {
-                log!(up.logger, "~~~ CONNECT called timeout {:?}", timeout);
+                log!(cu.logger, "~~~ CONNECT called timeout {:?}", timeout);
                 let deadline = timeout.map(|to| Instant::now() + to);
                 // connect all endpoints in parallel; send and receive peer ids through ports
                 let mut endpoint_manager = new_endpoint_manager(
-                    &mut *up.logger,
+                    &mut *cu.logger,
                     endpoint_setups,
-                    &mut up.port_info,
+                    &mut cu.port_info,
                     deadline,
                 )?;
                 log!(
-                    up.logger,
+                    cu.logger,
                     "Successfully connected {} endpoints",
                     endpoint_manager.endpoint_exts.len()
                 );
                 // leader election and tree construction
                 let neighborhood = init_neighborhood(
                     up.id_manager.connector_id,
                     &mut *up.logger,
                     cu.id_manager.connector_id,
                     &mut *cu.logger,
                     &mut endpoint_manager,
                     deadline,
                 )?;
                 log!(up.logger, "Successfully created neighborhood {:?}", &neighborhood);
                 log!(up.logger, "connect() finished. setup phase complete");
                 // TODO session optimization goes here
                 self.phased = ConnectorPhased::Communication(ConnectorCommunication {
                 log!(cu.logger, "Successfully created neighborhood {:?}", &neighborhood);
                 let mut comm = ConnectorCommunication {
                     round_index: 0,
                     endpoint_manager,
                     neighborhood,
                     mem_inbox: Default::default(),
                     native_batches: vec![Default::default()],
                     round_result: Ok(None),
                 });
                 };
                 session_optimize(cu, &mut comm, deadline)?;
                 log!(cu.logger, "connect() finished. setup phase complete");
                 self.phased = ConnectorPhased::Communication(comm);
                 Ok(())
+            }
+        }
+    }
+}
@@ @@ -294,14 +286,14 @@ fn new_endpoint_manager( @@
 fn init_neighborhood(
     connector_id: ConnectorId,
     logger: &mut dyn Logger,
     em: &mut EndpointManager,
     deadline: Option<Instant>,
 ) -> Result<Neighborhood, ConnectError> {
     use {ConnectError::*, Msg::SetupMsg as S, SetupMsg::*};
     ////////////////////////////////
     use {ConnectError::*, Msg::SetupMsg as S, SetupMsg::*};
     #[derive(Debug)]
     struct WaveState {
         parent: Option<usize>,
         leader: ConnectorId,
+    }
     fn do_wave(
@@ @@ -413,16 +405,21 @@ fn init_neighborhood( @@
                                     break 'election election_result;
+                                }
+                            }
+                        }
+                    }
+                }
                 S(YouAreMyParent) | S(MyPortInfo(_)) => unreachable!(),
                 comm_msg @ Msg::CommMsg { .. } => {
                     log!(logger, "delaying msg {:?} during election algorithm", comm_msg);
                     em.delayed_messages.push((recv_index, comm_msg));
                 msg @ S(YouAreMyParent) | msg @ S(MyPortInfo(_)) => {
                     log!(logger, "Endpont {:?} sent unexpected msg! {:?}", recv_index, &msg);
                     return Err(SetupAlgMisbehavior);
+                }
                 msg @ S(SessionScatter { .. })
                 | msg @ S(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
@@ @@ -440,13 +437,12 @@ fn init_neighborhood( @@
     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(LeaderWave { .. }) => { /* old message */ }
             S(LeaderAnnounce { .. }) => {
                 // not a child
                 log!(
                     logger,
                     "Got reply from non-child index {:?}. Children: {:?}",
                     recv_index,
@@ @@ -465,19 +461,158 @@ fn init_neighborhood( @@
                         children.iter()
                     );
                     return Err(SetupAlgMisbehavior);
+                }
                 children.push(recv_index);
+            }
             S(MyPortInfo(_)) => unreachable!(),
             comm_msg @ Msg::CommMsg { .. } => {
                 log!(logger, "delaying msg {:?} during election algorithm", comm_msg);
                 em.delayed_messages.push((recv_index, comm_msg));
             msg @ S(MyPortInfo(_)) | msg @ S(LeaderWave { .. }) => {
                 log!(logger, "discarding old message {:?} during election", msg);
+            }
             msg @ S(SessionScatter { .. })
             | msg @ S(SessionGather { .. })
             | msg @ Msg::CommMsg { .. } => {
                 log!(logger, "delaying msg {:?} during election", msg);
                 em.delayed_messages.push((recv_index, msg));
+            }
+        }
+    }
     children.shrink_to_fit();
     let neighborhood =
         Neighborhood { parent: election_result.parent, children: VecSet::new(children) };
     log!(logger, "Neighborhood constructed {:?}", &neighborhood);
     Ok(neighborhood)
+}
 fn session_optimize(
     cu: &mut ConnectorUnphased,
     comm: &mut ConnectorCommunication,
     deadline: Option<Instant>,
 ) -> Result<(), ConnectError> {
     ////////////////////////////////////////
     use {ConnectError::*, Msg::SetupMsg as S, SetupMsg::*};
     ////////////////////////////////////////
     log!(cu.logger, "Beginning session optimization");
     // populate session_info_map from a message per child
     let mut unoptimized_map: HashMap<ConnectorId, SessionInfo> = Default::default();
     let mut awaiting: HashSet<usize> = comm.neighborhood.children.iter().copied().collect();
     comm.endpoint_manager.undelay_all();
     while !awaiting.is_empty() {
         log!(
             cu.logger,
             "Session gather loop. awaiting info from children {:?}...",
             awaiting.iter()
         );
         let (recv_index, msg) =
             comm.endpoint_manager.try_recv_any_setup(&mut *cu.logger, deadline)?;
         log!(cu.logger, "Received from index {:?} msg {:?}", &recv_index, &msg);
         match msg {
             S(SessionGather { unoptimized_map: child_unoptimized_map }) => {
                 if !awaiting.remove(&recv_index) {
                     log!(
                         cu.logger,
                         "Wasn't expecting session info from {:?}. Got {:?}",
                         recv_index,
                         &child_unoptimized_map
                     );
                     return Err(SetupAlgMisbehavior);
+                }
                 unoptimized_map.extend(child_unoptimized_map.into_iter());
+            }
             msg @ S(YouAreMyParent)
             | msg @ S(MyPortInfo(..))
             | msg @ S(LeaderAnnounce { .. })
             | msg @ S(LeaderWave { .. }) => {
                 log!(cu.logger, "discarding old message {:?} during election", msg);
+            }
             msg @ S(SessionScatter { .. }) => {
                 log!(
                     cu.logger,
                     "Endpoint {:?} sent unexpected scatter! {:?} I've not contributed yet!",
                     recv_index,
                     &msg
                 );
                 return Err(SetupAlgMisbehavior);
+            }
             msg @ Msg::CommMsg(..) => {
                 log!(cu.logger, "delaying msg {:?} during session optimization", msg);
                 comm.endpoint_manager.delayed_messages.push((recv_index, msg));
+            }
+        }
+    }
     log!(
         cu.logger,
         "Gathered all children's maps. ConnectorId set is... {:?}",
         unoptimized_map.keys()
     );
     let my_session_info = SessionInfo {};
     unoptimized_map.insert(cu.id_manager.connector_id, my_session_info);
     log!(cu.logger, "Inserting my own info. Unoptimized subtree map is {:?}", &unoptimized_map);
     // acquire the optimized info...
     let optimized_map = if let Some(parent) = comm.neighborhood.parent {
         // ... as a message from my parent
         log!(cu.logger, "Forwarding gathered info to parent {:?}", parent);
         let msg = S(SessionGather { unoptimized_map });
         comm.endpoint_manager.send_to_setup(parent, &msg)?;
         'scatter_loop: loop {
             log!(
                 cu.logger,
                 "Session scatter recv loop. awaiting info from children {:?}...",
                 awaiting.iter()
             );
             let (recv_index, msg) =
                 comm.endpoint_manager.try_recv_any_setup(&mut *cu.logger, deadline)?;
             log!(cu.logger, "Received from index {:?} msg {:?}", &recv_index, &msg);
             match msg {
                 S(SessionScatter { optimized_map }) => {
                     if recv_index != parent {
                         log!(cu.logger, "I expected the scatter from my parent only!");
                         return Err(SetupAlgMisbehavior);
+                    }
                     break 'scatter_loop optimized_map;
+                }
                 msg @ Msg::CommMsg { .. } => {
                     log!(cu.logger, "delaying msg {:?} during scatter recv", msg);
                     comm.endpoint_manager.delayed_messages.push((recv_index, msg));
+                }
                 msg @ S(SessionGather { .. })
                 | msg @ S(YouAreMyParent)
                 | msg @ S(MyPortInfo(..))
                 | msg @ S(LeaderAnnounce { .. })
                 | msg @ S(LeaderWave { .. }) => {
                     log!(cu.logger, "discarding old message {:?} during election", msg);
+                }
+            }
+        }
     } else {
         // by computing it myself
         log!(cu.logger, "I am the leader! I will optimize this session");
         leader_session_map_optimize(unoptimized_map)?
     };
     log!(
         cu.logger,
         "Optimized info map is {:?}. Sending to children {:?}",
         &optimized_map,
         comm.neighborhood.children.iter()
     );
     let optimized_info =
         optimized_map.get(&cu.id_manager.connector_id).expect("HEY NO INFO FOR ME?").clone();
     let msg = S(SessionScatter { optimized_map });
     for &child in comm.neighborhood.children.iter() {
         comm.endpoint_manager.send_to_setup(child, &msg)?;
+    }
     apply_optimizations(cu, comm, optimized_info)?;
     log!(cu.logger, "Session optimization complete");
     Ok(())
+}
 fn leader_session_map_optimize(
     unoptimized_map: HashMap<ConnectorId, SessionInfo>,
 ) -> Result<HashMap<ConnectorId, SessionInfo>, ConnectError> {
     Ok(unoptimized_map)
+}
 fn apply_optimizations(
     _cu: &mut ConnectorUnphased,
     _comm: &mut ConnectorCommunication,
     _session_info: SessionInfo,
 ) -> Result<(), ConnectError> {
     Ok(())
+}

src/runtime/tests.rs

➞

Show inline comments

+}
 //////////////////////////////////////////
 #[test]
 fn basic_connector() {
-    Connector::new_simple(MINIMAL_PROTO.clone(), 0);
+    Connector::new(Box::new(DummyLogger), MINIMAL_PROTO.clone(), 0, 0);
+}
 #[test]
 fn basic_logged_connector() {
     let test_log_path = Path::new("./logs/basic_logged_connector");
     file_logged_connector(0, test_log_path);
@@ @@ -83,15 +83,15 @@ fn single_node_connect() { @@
     let _ = c.new_net_port(Getter, sock_addr, Passive).unwrap();
     let _ = c.new_net_port(Putter, sock_addr, Active).unwrap();
     c.connect(Some(Duration::from_secs(1))).unwrap();
+}
 #[test]
-fn multithreaded_connect() {
+fn minimal_net_connect() {
     let sock_addr = next_test_addr();
-    let test_log_path = Path::new("./logs/multithreaded_connect");
+    let test_log_path = Path::new("./logs/minimal_net_connect");
     scope(|s| {
         s.spawn(|_| {
             let mut c = file_logged_connector(0, test_log_path);
             let _ = c.new_net_port(Getter, sock_addr, Active).unwrap();
             c.connect(Some(Duration::from_secs(1))).unwrap();
         });
@@ @@ -275,29 +275,31 @@ fn connector_pair_nondet() { @@
     })
     .unwrap();
+}
 #[test]
 fn cannot_use_moved_ports() {
     let test_log_path = Path::new("./logs/cannot_use_moved_ports"); /*
                                                                     native p|-->|g sync
                                                                     */
     /*
     native p|-->|g sync
     */
     let test_log_path = Path::new("./logs/cannot_use_moved_ports");
     let mut c = file_logged_connector(0, test_log_path);
     let [p, g] = c.new_port_pair();
     c.add_component(b"sync", &[g, p]).unwrap();
     c.connect(Some(Duration::from_secs(1))).unwrap();
     c.put(p, (b"hello" as &[_]).into()).unwrap_err();
     c.get(g).unwrap_err();
+}
 #[test]
 fn sync_sync() {
     let test_log_path = Path::new("./logs/sync_sync"); /*
                                                        native p0|-->|g0 sync
                                                               g1|<--|p1
                                                        */
     /*
     native p0|-->|g0 sync
            g1|<--|p1
     */
     let test_log_path = Path::new("./logs/sync_sync");
     let mut c = file_logged_connector(0, test_log_path);
     let [p0, g0] = c.new_port_pair();
     let [p1, g1] = c.new_port_pair();
     c.add_component(b"sync", &[g0, p1]).unwrap();
     c.connect(Some(Duration::from_secs(1))).unwrap();
     c.put(p0, (b"hello" as &[_]).into()).unwrap();
@@ @@ -330,17 +332,17 @@ fn double_net_connect() { @@
     })
     .unwrap();
+}
 #[test]
 fn distributed_msg_bounce() {
     let test_log_path = Path::new("./logs/distributed_msg_bounce");
     /*
     native[0] | sync 0.p|-->|1.p native[1]
 .g|<--|1.g
     */
     let test_log_path = Path::new("./logs/distributed_msg_bounce");
     let sock_addrs = [next_test_addr(), next_test_addr()];
     scope(|s| {
         s.spawn(|_| {
             /*
             native | sync p|-->
                    |      g|<--

0 comments (0 inline, 0 general)