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

Changeset - e8655e939407

Parent rev.

Child rev.

[Not reviewed]

0 3 0

Christopher Esterhuyse - 5 years ago 2020-03-06 10:56:51
christopher.esterhuyse@gmail.com

cleaned up

3 files changed with 17 insertions and 27 deletions:

src/runtime/communication.rs

src/runtime/mod.rs

src/runtime/setup.rs

0 comments (0 inline, 0 general)

src/runtime/communication.rs

➞

Show inline comments

@@ @@ -219,131 +219,141 @@ impl Controller { @@
             let poly_n = self.kick_off_native(sync_batches)?;
             log!(
                 &mut self.inner.logger,
                 "PolyN kicked off, and has branches with predicates... {:?}",
                 poly_n.branches.keys().collect::<Vec<_>>()
             );
             Some(poly_n)
         } else {
             log!(&mut self.inner.logger, "NO NATIVE COMPONENT");
             None
         };
         // 6. Kick off the synchronous round of each protocol actor
         //    If just one actor becomes inconsistent now, there can be no solution!
         //    TODO distinguish between completed and not completed poly_p's?
         log!(&mut self.inner.logger, "Kicking off {} PolyP's.", self.ephemeral.poly_ps.len());
         for (index, poly_p) in self.ephemeral.poly_ps.iter_mut().enumerate() {
             let my_subtree_id = SubtreeId::PolyP { index };
             let m_ctx = PolyPContext {
                 my_subtree_id,
                 inner: &mut self.inner,
                 solution_storage: &mut self.ephemeral.solution_storage,
             };
             use SyncRunResult as Srr;
             let blocker = poly_p.poly_run(m_ctx, &self.protocol_description)?;
             log!(&mut self.inner.logger, "... PolyP's poly_run got blocker {:?}", &blocker);
             match blocker {
                 Srr::NoBranches => return Err(SyncErr::Inconsistent),
                 Srr::AllBranchesComplete | Srr::BlockingForRecv => (),
+            }
+        }
         log!(&mut self.inner.logger, "All Poly machines have been kicked off!");
         // 7. `solution_storage` may have new solutions for this controller
         //    handle their discovery. LEADER => announce, otherwise => send to parent
+        {
             let peeked = self.ephemeral.solution_storage.peek_new_locals().collect::<Vec<_>>();
             log!(
                 &mut self.inner.logger,
                 "Got {} controller-local solutions before a single RECV: {:?}",
                 peeked.len(),
                 peeked
             );
+        }
         if self.handle_locals_maybe_decide()? {
             return Ok(());
+        }
         // 4. Receive incoming messages until the DECISION is made
+        // 4. Receive incoming messages until the DECISION is made OR some unrecoverable error
         log!(&mut self.inner.logger, "`No decision yet`. Time to recv messages");
         self.undelay_all();
         'recv_loop: loop {
             log!(&mut self.inner.logger, "`POLLING` with deadline {:?}...", deadline);
             let received = match deadline {
                 None => {
                     // we have personally timed out. perform a "long" poll.
                     self.recv(Instant::now() + Duration::from_secs(10))?.expect("DRIED UP")
+                }
                 Some(d) => match self.recv(d)? {
                     // we have not yet timed out. performed a time-limited poll
                     Some(received) => received,
                     None => {
                         // timed out! send a FAILURE message to the sink,
                         // and henceforth don't time out on polling.
                         deadline = None;
                         match self.inner.family.parent_ekey {
                             None => {
                                 // I am the sink! announce failure and return.
                                 return self.end_round_with_decision(Decision::Failure);
+                            }
                             Some(parent_ekey) => {
                                 // I am not the sink! send a failure message.
                                 let announcement = Msg::CommMsg(CommMsg {
                                     round_index: self.inner.round_index,
                                     contents: CommMsgContents::Failure,
                                 });
                                 log!(
                                     &mut self.inner.logger,
                                     "Forwarding {:?} to parent with ekey {:?}",
                                     &announcement,
                                     parent_ekey
                                 );
                                 self.inner
                                     .endpoint_exts
                                     .get_mut(parent_ekey)
                                     .expect("ss")
                                     .endpoint
                                     .send(announcement.clone())?;
                                 continue; // poll some more
+                            }
                             None => return self.end_round_with_decision(Decision::Failure),
+                        }
                         continue;
+                    }
                 },
                 None => self.recv(Instant::now() + Duration::from_secs(2))?.expect("DRIED UP"),
             };
             log!(&mut self.inner.logger, "::: message {:?}...", &received);
             let current_content = match received.msg {
                 Msg::SetupMsg(s) => {
                     // This occurs in the event the connector was malformed during connect()
                     println!("WASNT EXPECTING {:?}", s);
                     return Err(SyncErr::UnexpectedSetupMsg);
+                }
                 Msg::CommMsg(CommMsg { round_index, .. })
                     if round_index < self.inner.round_index =>
+                {
                     // Old message! Can safely discard
                     log!(&mut self.inner.logger, "...and its OLD! :(");
                     drop(received);
                     continue 'recv_loop;
+                }
                 Msg::CommMsg(CommMsg { round_index, .. })
                     if round_index > self.inner.round_index =>
+                {
                     // Message from a next round. Keep for later!
                     log!(&mut self.inner.logger, "... DELAY! :(");
                     self.delay(received);
                     continue 'recv_loop;
+                }
                 Msg::CommMsg(CommMsg { contents, round_index }) => {
                     log!(
                         &mut self.inner.logger,
                         "... its a round-appropriate CommMsg with key {:?}",
                         received.recipient
                     );
                     assert_eq!(round_index, self.inner.round_index);
                     contents
+                }
             };
             match current_content {
                 CommMsgContents::Failure => match self.inner.family.parent_ekey {
                     Some(parent_ekey) => {
                         let announcement = Msg::CommMsg(CommMsg {
                             round_index: self.inner.round_index,
                             contents: CommMsgContents::Failure,
                         });
                         log!(
                             &mut self.inner.logger,
                             "Forwarding {:?} to parent with ekey {:?}",
                             &announcement,
                             parent_ekey
                         );
                         self.inner
@@ @@ -653,54 +663,48 @@ impl SolutionStorage { @@
             for pred in set.iter() {
                 if let Some(elaborated) = pred.union_with(&partial) {
                     Self::elaborate_into_new_local_rec(
                         logger,
                         elaborated,
                         set_visitor.clone(),
                         old_local,
                         new_local,
+                    )
+                }
+            }
         } else {
             // recursive stop condition. `partial` is a local subtree solution
             if !old_local.contains(&partial) {
                 // ... and it hasn't been found before
                 log!(logger, "... storing NEW LOCAL SOLUTION {:?}", &partial);
                 new_local.insert(partial);
+            }
+        }
+    }
+}
 impl PolyContext for BranchPContext<'_, '_> {
     type D = ProtocolD;
     fn is_firing(&mut self, ekey: Key) -> Option<bool> {
         assert!(self.ekeys.contains(&ekey));
         let channel_id = self.m_ctx.inner.endpoint_exts.get(ekey).unwrap().info.channel_id;
         let val = self.predicate.query(channel_id);
         log!(
             &mut self.m_ctx.inner.logger,
             "!! PolyContext callback to is_firing by {:?}! returning {:?}",
             self.m_ctx.my_subtree_id,
             val,
         );
         val
+    }
     fn read_msg(&mut self, ekey: Key) -> Option<&Payload> {
         assert!(self.ekeys.contains(&ekey));
         let val = self.inbox.get(&ekey);
         log!(
             &mut self.m_ctx.inner.logger,
             "!! PolyContext callback to read_msg by {:?}! returning {:?}",
             self.m_ctx.my_subtree_id,
             val,
         );
         val
+    }
+}
 /*
 invariant: Controller.inner has stable MonoN/P states for which it will start the
 */

src/runtime/mod.rs

➞

Show inline comments

@@ @@ -50,117 +50,102 @@ pub struct Unconfigured { @@
+}
 #[derive(Debug)]
 pub struct Configured {
     controller_id: ControllerId,
     polarities: Vec<Polarity>,
     bindings: HashMap<usize, PortBinding>,
     protocol_description: Arc<ProtocolD>,
     main_component: Vec<u8>,
+}
 #[derive(Debug)]
 pub struct Connected {
     native_interface: Vec<(Key, Polarity)>,
     sync_batches: Vec<SyncBatch>,
     controller: Controller,
+}
 #[derive(Debug, Copy, Clone)]
 pub enum PortBinding {
     Native,
     Active(SocketAddr),
     Passive(SocketAddr),
+}
 #[derive(Debug)]
 struct Arena<T> {
     storage: Vec<T>,
+}
 #[derive(Debug)]
 struct ReceivedMsg {
     recipient: Key,
     msg: Msg,
+}
 #[derive(Debug)]
 struct MessengerState {
     poll: Poll,
     events: Events,
     delayed: Vec<ReceivedMsg>,
     undelayed: Vec<ReceivedMsg>,
     polled_undrained: IndexSet<Key>,
+}
 #[derive(Debug)]
 struct ChannelIdStream {
     controller_id: ControllerId,
     next_channel_index: ChannelIndex,
+}
 #[derive(Debug)]
 enum RoundHistory {
     Consistent {
         decision: Predicate,
         native_component: (MonoN, PolyN),
         protocol_components: Box<[(MonoP, PolyP)]>,
     },
     Inconsistent {
         error: SyncErr,
         subtree_solutions: SolutionStorage,
         native_component: PolyN,
         protocol_components: Box<[PolyP]>,
     },
+}
 #[derive(Debug)]
 struct Controller {
     protocol_description: Arc<ProtocolD>,
     inner: ControllerInner,
     ephemeral: ControllerEphemeral,
-    round_histories: Vec<RoundHistory>,
+    unrecoverable_error: Option<SyncErr>, // prevents future calls to Sync
+}
 #[derive(Debug)]
 struct ControllerInner {
     round_index: usize,
     channel_id_stream: ChannelIdStream,
     endpoint_exts: Arena<EndpointExt>,
     messenger_state: MessengerState,
     mono_n: MonoN,       // state at next round start
     mono_ps: Vec<MonoP>, // state at next round start
     family: ControllerFamily,
     logger: String,
+}
 /// This structure has its state entirely reset between synchronous rounds
 #[derive(Debug, Default)]
 struct ControllerEphemeral {
     solution_storage: SolutionStorage,
     poly_n: Option<PolyN>,
     poly_ps: Vec<PolyP>,
     mono_ps: Vec<MonoP>,
     ekey_to_holder: HashMap<Key, PolyId>,
+}
 #[derive(Debug)]
 struct ControllerFamily {
     parent_ekey: Option<Key>,
     children_ekeys: Vec<Key>,
+}
 #[derive(Debug)]
 pub(crate) enum SyncRunResult {
     BlockingForRecv,
     AllBranchesComplete,
     NoBranches,
+}
 // Used to identify poly actors
 #[derive(Debug, Copy, Clone, Eq, PartialEq, Ord, PartialOrd, Hash)]
 enum PolyId {
     N,
     P { index: usize },
+}
 #[derive(Debug, Copy, Clone, Eq, PartialEq, Ord, PartialOrd, Hash)]
 pub(crate) enum SubtreeId {
     PolyN,
     PolyP { index: usize },
     ChildController { ekey: Key },

src/runtime/setup.rs

➞

Show inline comments

@@ @@ -86,97 +86,98 @@ impl Controller { @@
                 PortBinding::Active(addr) => {
                     let ekey_proto = endpoint_ext_todos.alloc(EndpointExtTodo::ActiveConnecting {
                         addr,
                         polarity,
                         stream: TcpStream::connect(&addr).unwrap(),
                     });
                     ekeys_network.push(ekey_proto);
                     ekeys_proto.push(ekey_proto);
+                }
+            }
+        }
         log!(&mut logger, "{:03?} setup todos...", major);
         // 2. convert the arena to Arena<EndpointExt>  and return the
         let (mut messenger_state, mut endpoint_exts) =
             Self::finish_endpoint_ext_todos(major, &mut logger, endpoint_ext_todos, deadline)?;
         let n_mono = MonoN { ekeys: ekeys_native.into_iter().collect(), result: None };
         let p_monos = vec![MonoP {
             state: protocol_description.new_main_component(main_component, &ekeys_proto),
             ekeys: ekeys_proto.into_iter().collect(),
         }];
         // 6. Become a node in a sink tree, computing {PARENT, CHILDREN} from {NEIGHBORS}
         let family = Self::setup_sink_tree_family(
             major,
             &mut logger,
             &mut endpoint_exts,
             &mut messenger_state,
             ekeys_network,
             deadline,
         )?;
         log!(&mut logger, "CONNECT PHASE END! ~");
         let inner = ControllerInner {
             family,
             messenger_state,
             channel_id_stream,
             endpoint_exts,
             mono_ps: p_monos,
             mono_n: n_mono,
             round_index: 0,
             logger,
         };
         let controller = Self {
             protocol_description,
             inner,
             ephemeral: Default::default(),
             round_histories: vec![],
             // round_histories: vec![],
             unrecoverable_error: None,
         };
         Ok((controller, native_interface))
+    }
     fn test_stream_connectivity(stream: &mut TcpStream) -> bool {
         use std::io::Write;
         stream.write(&[]).is_ok()
+    }
     // inserts
     fn finish_endpoint_ext_todos(
         major: ControllerId,
         logger: &mut String,
         mut endpoint_ext_todos: Arena<EndpointExtTodo>,
         deadline: Instant,
     ) -> Result<(MessengerState, Arena<EndpointExt>), ConnectErr> {
         use {ConnectErr::*, EndpointExtTodo::*};
         // 1. define and setup a poller and event loop
         let edge = PollOpt::edge();
         let [ready_r, ready_w] = [Ready::readable(), Ready::writable()];
         let mut ms = MessengerState {
             poll: Poll::new().map_err(|_| PollInitFailed)?,
             events: Events::with_capacity(endpoint_ext_todos.len()),
             delayed: vec![],
             undelayed: vec![],
             polled_undrained: Default::default(),
         };
         // 2. Register all EndpointExtTodos with ms.poll. each has one of {Endpoint, TcpStream, TcpListener}
         // 3. store the keyset of EndpointExtTodos which are not Finished in `to_finish`.
         let mut to_finish = HashSet::<_>::default();
         log!(logger, "endpoint_ext_todos len {:?}", endpoint_ext_todos.len());
         for (key, t) in endpoint_ext_todos.iter() {
             let token = key.to_token();
             match t {
                 ActiveRecving { .. } | PassiveConnecting { .. } => unreachable!(),
                 Finished(EndpointExt { endpoint, .. }) => {
                     ms.poll.register(endpoint, token, ready_r, edge)
+                }
                 ActiveConnecting { stream, .. } => {
                     to_finish.insert(key);
                     ms.poll.register(stream, token, ready_w, edge)
+                }
                 PassiveAccepting { listener, .. } => {
                     to_finish.insert(key);
                     ms.poll.register(listener, token, ready_r, edge)
+                }

0 comments (0 inline, 0 general)