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

@@ @@ -261,24 +261,36 @@ impl Controller { @@
             );
+        }
         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,
@@ @@ -289,19 +301,17 @@ impl Controller { @@
                                 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);
@@ @@ -695,12 +705,6 @@ impl PolyContext for BranchPContext<'_, '_> { @@
             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

@@ @@ -92,33 +92,18 @@ struct MessengerState { @@
 #[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>,

src/runtime/setup.rs

➞

Show inline comments

@@ @@ -128,13 +128,14 @@ impl Controller { @@
             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;

0 comments (0 inline, 0 general)