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

Changeset - af5facdd41b1

Parent rev.

Child rev.

[Not reviewed]

0 6 0

Christopher Esterhuyse - 5 years ago 2020-03-06 09:06:49
christopher.esterhuyse@gmail.com

generalized announcement to Decision enum, which also covers TIMEOUT case

6 files changed with 129 insertions and 185 deletions:

src/runtime/communication.rs

src/runtime/connector.rs

src/runtime/endpoint.rs

src/runtime/mod.rs

src/runtime/serde.rs

src/runtime/setup.rs

0 comments (0 inline, 0 general)

src/runtime/communication.rs

➞

Show inline comments

 use crate::common::*;
 use crate::runtime::{actors::*, endpoint::*, errors::*, *};
 impl Controller {
     //usable BETWEEN rounds
     fn consistent(&self) -> bool {
         self.inner.mono_n.is_some()
+    }
     fn end_round_with_decision(&mut self, decision: Predicate) -> Result<(), SyncErr> {
     fn end_round_with_decision(&mut self, decision: Decision) -> Result<(), SyncErr> {
         log!(&mut self.inner.logger, "ENDING ROUND WITH DECISION! {:?}", &decision);
         // let mut table_row = HashMap::<Key, _>::default();
         let n_pair = {
             let poly_n = self.ephemeral.poly_n.take().unwrap();
             let mono_n = poly_n.choose_mono(&decision).unwrap();
             (mono_n, poly_n)
         };
         self.inner.mono_n = Some(n_pair.0.clone());
         let p_pairs = self
             .ephemeral
             .poly_ps
             .drain(..)
             .map(|poly_p| {
                 let mono_p = poly_p.choose_mono(&decision).unwrap();
                 (mono_p, poly_p)
             })
             .collect::<Vec<_>>()
             .into_boxed_slice();
         self.inner.mono_ps.extend(p_pairs.iter().map(|p_pair| p_pair.0.clone()));
         self.round_histories.push(RoundHistory::Consistent {
             decision: decision.clone(),
             native_component: n_pair,
             protocol_components: p_pairs,
         });
         let announcement =
             CommMsgContents::Announce { oracle: decision }.into_msg(self.inner.round_index);
         if let Decision::Success(predicate) = &decision {
             // overwrite MonoN/P
             self.inner.mono_n =
                 self.ephemeral.poly_n.take().unwrap().choose_mono(predicate).unwrap();
             self.inner.mono_ps.clear();
             self.inner.mono_ps.extend(
                 self.ephemeral
                     .poly_ps
                     .drain(..)
                     .map(|poly_p| poly_p.choose_mono(predicate).unwrap()),
+            )
+        }
         let announcement = CommMsgContents::Announce { decision }.into_msg(self.inner.round_index);
         for &child_ekey in self.inner.family.children_ekeys.iter() {
             log!(
                 &mut self.inner.logger,
                 "Forwarding {:?} to child with ekey {:?}",
                 &announcement,
                 child_ekey
@@ @@ -68,14 +49,15 @@ impl Controller { @@
                 parent_endpoint.send(msg)?;
+            }
             Ok(false)
         } else {
             // I have no parent -> I'm the leader
             assert!(self.inner.family.parent_ekey.is_none());
             let maybe_decision = self.ephemeral.solution_storage.iter_new_local_make_old().next();
             Ok(if let Some(decision) = maybe_decision {
             let maybe_predicate = self.ephemeral.solution_storage.iter_new_local_make_old().next();
             Ok(if let Some(predicate) = maybe_predicate {
                 let decision = Decision::Success(predicate);
                 log!(&mut self.inner.logger, "DECIDE ON {:?} AS LEADER!", &decision);
                 self.end_round_with_decision(decision)?;
                 true
             } else {
                 false
             })
@@ @@ -83,13 +65,13 @@ impl Controller { @@
+    }
     fn kick_off_native(
         &mut self,
         sync_batches: impl Iterator<Item = SyncBatch>,
     ) -> Result<PolyN, EndpointErr> {
-        let MonoN { ekeys, .. } = self.inner.mono_n.take().unwrap();
+        let MonoN { ekeys, .. } = self.inner.mono_n.clone();
         let Self { inner: ControllerInner { endpoint_exts, round_index, .. }, .. } = self;
         let mut branches = HashMap::<_, _>::default();
         for (sync_batch_index, SyncBatch { puts, gets }) in sync_batches.enumerate() {
             let ekey_to_channel_id = |ekey| endpoint_exts.get(ekey).unwrap().info.channel_id;
             let all_ekeys = ekeys.iter().copied();
             let all_channel_ids = all_ekeys.map(ekey_to_channel_id);
@@ @@ -140,76 +122,36 @@ impl Controller { @@
+            }
             branches.insert(predicate, branch);
+        }
         Ok(PolyN { ekeys, branches })
+    }
     pub fn sync_round(
         &mut self,
         deadline: Instant,
         sync_batches: Option<impl Iterator<Item = SyncBatch>>,
     ) -> Result<(), SyncErr> {
         if self.consistent() {
             match self.sync_round_inner(deadline, sync_batches) {
                 Ok(()) => return Ok(()),
                 Err(error) => {
                     log!(
                         &mut self.inner.logger,
                         "/\\/\\/\\/\\/\\/ Sync round failed! Preparing for diagnosis...",
                     );
                     let h = RoundHistory::Inconsistent {
                         error,
                         subtree_solutions: std::mem::take(&mut self.ephemeral.solution_storage),
                         native_component: self.ephemeral.poly_n.take().unwrap(),
                         protocol_components: std::mem::take(&mut self.ephemeral.poly_ps)
                             .into_boxed_slice(),
                     };
                     self.round_histories.push(h);
                     for (round_index, round) in self.round_histories.iter().enumerate() {
                         log!(&mut self.inner.logger, "round {}:{:#?}\n", round_index, round);
+                    }
+                }
+            }
+        }
         if let Some(RoundHistory::Inconsistent { error, .. }) =
             self.round_histories.iter().rev().next()
+        {
             Err(error.clone())
         } else {
             unreachable!()
+        }
+    }
     // Runs a synchronous round until all the actors are in decided state OR 1+ are inconsistent.
     // If a native requires setting up, arg `sync_batches` is Some, and those are used as the sync batches.
-    fn sync_round_inner(
+    pub fn sync_round(
         &mut self,
         deadline: Instant,
         sync_batches: Option<impl Iterator<Item = SyncBatch>>,
     ) -> Result<(), SyncErr> {
         assert!(self.ephemeral.is_clear());
         log!(
             &mut self.inner.logger,
             "~~~~~~~~ SYNC ROUND STARTS! ROUND={} ~~~~~~~~~",
             self.inner.round_index
         );
         assert!(self.ephemeral.is_clear());
         // 1. Run the Mono for each Mono actor (stored in `self.mono_ps`).
         //    Some actors are dropped. some new actors are created.
         //    Ultimately, we have 0 Mono actors and a list of unnamed sync_actors
         log!(&mut self.inner.logger, "Got {} MonoP's to run!", self.inner.mono_ps.len());
         self.ephemeral.poly_ps.clear();
         // let mut poly_ps: Vec<PolyP> = vec![];
         while let Some(mut mono_p) = self.inner.mono_ps.pop() {
         self.ephemeral.mono_ps.extend(self.inner.mono_ps.iter().cloned());
         log!(&mut self.inner.logger, "Got {} MonoP's to run!", self.ephemeral.mono_ps.len());
         while let Some(mut mono_p) = self.ephemeral.mono_ps.pop() {
             let mut m_ctx = MonoPContext {
                 ekeys: &mut mono_p.ekeys,
                 mono_ps: &mut self.ephemeral.mono_ps,
                 inner: &mut self.inner,
                 // endpoint_exts: &mut self.endpoint_exts,
                 // mono_ps: &mut self.mono_ps,
                 // channel_id_stream: &mut self.channel_id_stream,
             };
             // cross boundary into crate::protocol
             let blocker = mono_p.state.pre_sync_run(&mut m_ctx, &self.protocol_description);
             log!(&mut self.inner.logger, "... MonoP's pre_sync_run got blocker {:?}", &blocker);
             match blocker {
                 MonoBlocker::Inconsistent => return Err(SyncErr::Inconsistent),
@@ @@ -226,13 +168,13 @@ impl Controller { @@
         // 3. define the mapping from ekey -> actor
         //    this is needed during the event loop to determine which actor
         //    should receive the incoming message.
         //    TODO: store and update this mapping rather than rebuilding it each round.
         let ekey_to_holder: HashMap<Key, PolyId> = {
             use PolyId::*;
-            let n = self.inner.mono_n.iter().flat_map(|m| m.ekeys.iter().map(move |&e| (e, N)));
             let n = self.inner.mono_n.ekeys.iter().map(move |&e| (e, N));
             let p = self
                 .ephemeral
                 .poly_ps
                 .iter()
                 .enumerate()
                 .flat_map(|(index, m)| m.ekeys.iter().map(move |&e| (e, P { index })));
@@ @@ -244,13 +186,13 @@ impl Controller { @@
             &ekey_to_holder
         );
         // 4. Create the solution storage. it tracks the solutions of "subtrees"
         //    of the controller in the overlay tree.
         self.ephemeral.solution_storage.reset({
-            let n = self.inner.mono_n.iter().map(|_| SubtreeId::PolyN);
+            let n = std::iter::once(SubtreeId::PolyN);
             let m = (0..self.ephemeral.poly_ps.len()).map(|index| SubtreeId::PolyP { index });
             let c = self
                 .inner
                 .family
                 .children_ekeys
                 .iter()
@@ @@ -264,13 +206,12 @@ impl Controller { @@
             subtree_id_iter
         });
         // 5. kick off the synchronous round of the native actor if it exists
         log!(&mut self.inner.logger, "Kicking off native's synchronous round...");
         assert_eq!(sync_batches.is_some(), self.inner.mono_n.is_some()); // TODO better err
         self.ephemeral.poly_n = if let Some(sync_batches) = sync_batches {
             // using if let because of nested ? operator
             // TODO check that there are 1+ branches or NO SOLUTION
             let poly_n = self.kick_off_native(sync_batches)?;
             log!(
                 &mut self.inner.logger,
@@ @@ -358,12 +299,33 @@ impl Controller { @@
                     );
                     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
                             .endpoint_exts
                             .get_mut(parent_ekey)
                             .expect("ss")
                             .endpoint
                             .send(announcement.clone())?;
+                    }
                     None => return self.end_round_with_decision(Decision::Failure),
                 },
                 CommMsgContents::Elaborate { partial_oracle } => {
                     // Child controller submitted a subtree solution.
                     if !self.inner.family.children_ekeys.contains(&received.recipient) {
                         return Err(SyncErr::ElaborateFromNonChild);
+                    }
                     let subtree_id = SubtreeId::ChildController { ekey: received.recipient };
@@ @@ -379,23 +341,23 @@ impl Controller { @@
                         partial_oracle,
                     );
                     if self.handle_locals_maybe_decide()? {
                         return Ok(());
+                    }
+                }
-                CommMsgContents::Announce { oracle } => {
+                CommMsgContents::Announce { decision } => {
                     if self.inner.family.parent_ekey != Some(received.recipient) {
                         return Err(SyncErr::AnnounceFromNonParent);
+                    }
                     log!(
                         &mut self.inner.logger,
                         "Received ANNOUNCEMENT from from parent {:?}: {:?}",
                         received.recipient,
-                        &oracle
+                        &decision
                     );
-                    return self.end_round_with_decision(oracle);
+                    return self.end_round_with_decision(decision);
+                }
                 CommMsgContents::SendPayload { payload_predicate, payload } => {
                     assert_eq!(
                         Getter,
                         self.inner.endpoint_exts.get(received.recipient).unwrap().info.polarity
                     );
@@ @@ -486,19 +448,39 @@ impl Controller { @@
+                            }
+                        }
                     };
+                }
+            }
+        }
         // 'timeout_loop: loop {
         //     log!(&mut self.inner.logger, "`POLLING (already timed out)`...");
         //     let received = self.recv_blocking()?;
         //     log!(&mut self.inner.logger, "::: message {:?}...", &received);
         //     let current_content = match received.msg {
         //         Msg::SetupMsg(_) => {
         //             // This occurs in the event the connector was malformed during connect()
         //             return Err(SyncErr::UnexpectedSetupMsg);
         //         }
         //         Msg::CommMsg(CommMsg { round_index, contents }) => {
         //             if round_index > self.inner.round_index {
         //                 self.delay(received);
         //                 continue 'timeout_loop;
         //             } else {
         //                 contents
         //             }
         //         }
         //     };
         // }
+    }
+}
 impl ControllerEphemeral {
     fn is_clear(&self) -> bool {
         self.solution_storage.is_clear()
             && self.poly_n.is_none()
             && self.poly_ps.is_empty()
             && self.mono_ps.is_empty()
             && self.ekey_to_holder.is_empty()
+    }
     fn clear(&mut self) {
         self.solution_storage.clear();
         self.poly_n.take();
         self.poly_ps.clear();
@@ @@ -537,13 +519,13 @@ impl MonoContext for MonoPContext<'_> { @@
             &mut self.inner.logger,
             "!! MonoContext callback to new_component with ekeys {:?}!",
             &moved_ekeys,
         );
         if moved_ekeys.is_subset(self.ekeys) {
             self.ekeys.retain(|x| !moved_ekeys.contains(x));
-            self.inner.mono_ps.push(MonoP { state: init_state, ekeys: moved_ekeys });
             self.mono_ps.push(MonoP { state: init_state, ekeys: moved_ekeys });
         } else {
             panic!("MachineP attempting to move alien ekey!");
+        }
+    }
     fn new_channel(&mut self) -> [Key; 2] {
         let [a, b] = Endpoint::new_memory_pair();
@@ @@ -700,6 +682,12 @@ 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/connector.rs

➞

Show inline comments

@@ @@ -158,16 +158,13 @@ impl Connector { @@
             _ => return Err(NotConnected),
         };
         // do the synchronous round!
         connected.controller.sync_round(deadline, Some(connected.sync_batches.drain(..)))?;
         connected.sync_batches.push(SyncBatch::default());
         let mono_n = connected.controller.inner.mono_n.as_mut().unwrap();
         let result = mono_n.result.as_mut().unwrap();
         Ok(result.0)
         Ok(connected.controller.inner.mono_n.result.as_mut().unwrap().0)
+    }
     pub fn read_gotten(&self, native_port_index: usize) -> Result<&[u8], ReadGottenErr> {
         use ReadGottenErr::*;
         let connected = match self {
             Connector::Connected(connected) => connected,
@@ @@ -175,12 +172,11 @@ impl Connector { @@
         };
         let &(key, polarity) =
             connected.native_interface.get(native_port_index).ok_or(IndexOutOfBounds)?;
         if polarity != Getter {
             return Err(WrongPolarity);
+        }
         let mono_n = connected.controller.inner.mono_n.as_ref().expect("controller has no mono_n?");
         let result = mono_n.result.as_ref().ok_or(NoPreviousRound)?;
         let result = connected.controller.inner.mono_n.result.as_ref().ok_or(NoPreviousRound)?;
         let payload = result.1.get(&key).ok_or(DidNotGet)?;
         Ok(payload)
+    }
+}

src/runtime/endpoint.rs

➞

Show inline comments

@@ @@ -15,12 +15,18 @@ pub(crate) struct EndpointExt { @@
 #[derive(Debug, Copy, Clone)]
 pub struct EndpointInfo {
     pub polarity: Polarity,
     pub channel_id: ChannelId,
+}
 #[derive(Debug, Clone)]
 pub(crate) enum Decision {
     Failure,
     Success(Predicate),
+}
 #[derive(Clone, Debug)]
 pub(crate) enum Msg {
     SetupMsg(SetupMsg),
     CommMsg(CommMsg),
+}
 #[derive(Clone, Debug)]
@@ @@ -42,14 +48,15 @@ pub(crate) struct CommMsg { @@
     pub round_index: usize,
     pub contents: CommMsgContents,
+}
 #[derive(Clone, Debug)]
 pub(crate) enum CommMsgContents {
     SendPayload { payload_predicate: Predicate, payload: Payload },
     Elaborate { partial_oracle: Predicate },
     Announce { oracle: Predicate },
     Elaborate { partial_oracle: Predicate }, // SINKWARD
     Failure,                                 // SINKWARD
     Announce { decision: Decision },         // SINKAWAYS
+}
 pub struct NetworkEndpoint {
     stream: mio::net::TcpStream,
     inbox: Vec<u8>,
     outbox: Vec<u8>,

src/runtime/mod.rs

➞

Show inline comments

@@ @@ -120,24 +120,25 @@ struct Controller { @@
 #[derive(Debug)]
 struct ControllerInner {
     round_index: usize,
     channel_id_stream: ChannelIdStream,
     endpoint_exts: Arena<EndpointExt>,
     messenger_state: MessengerState,
     mono_n: Option<MonoN>,
     mono_ps: Vec<MonoP>,
     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>,
@@ @@ -165,12 +166,13 @@ pub(crate) enum SubtreeId { @@
     ChildController { ekey: Key },
+}
 pub(crate) struct MonoPContext<'a> {
     inner: &'a mut ControllerInner,
     ekeys: &'a mut HashSet<Key>,
     mono_ps: &'a mut Vec<MonoP>,
+}
 pub(crate) struct PolyPContext<'a> {
     my_subtree_id: SubtreeId,
     inner: &'a mut ControllerInner,
     solution_storage: &'a mut SolutionStorage,
+}
@@ @@ -243,46 +245,40 @@ trait Messengerlike { @@
+                }
                 Err(PollDeadlineErr::PollingFailed) => return Err(MessengerRecvErr::PollingFailed),
                 Err(PollDeadlineErr::Timeout) => return Ok(None),
+            }
+        }
+    }
     // attempt to receive a message from one of the endpoints before the deadline
     fn recv_until(
         &mut self,
         deadline: Option<Instant>,
     ) -> Result<Option<ReceivedMsg>, MessengerRecvErr> {
     fn recv_blocking(&mut self) -> Result<ReceivedMsg, MessengerRecvErr> {
         // try get something buffered
         if let Some(x) = self.get_state_mut().undelayed.pop() {
-            return Ok(Some(x));
             return Ok(x);
+        }
         loop {
             // polled_undrained may not be empty
             while let Some(eekey) = self.get_state_mut().polled_undrained.pop() {
                 if let Some(msg) = self
                     .get_endpoint_mut(eekey)
                     .recv()
                     .map_err(|e| MessengerRecvErr::EndpointErr(eekey, e))?
+                {
                     // this endpoint MAY still have messages! check again in future
                     self.get_state_mut().polled_undrained.insert(eekey);
-                    return Ok(Some(ReceivedMsg { recipient: eekey, msg }));
                     return Ok(ReceivedMsg { recipient: eekey, msg });
+                }
+            }
             let state = self.get_state_mut();
             match state.poll_events_until(deadline) {
                 Ok(()) => {
                     for e in state.events.iter() {
                         state.polled_undrained.insert(Key::from_token(e.token()));
+                    }
+                }
                 Err(PollDeadlineErr::PollingFailed) => return Err(MessengerRecvErr::PollingFailed),
                 Err(PollDeadlineErr::Timeout) => return Ok(None),
             state
                 .poll
                 .poll(&mut state.events, None)
                 .map_err(|_| MessengerRecvErr::PollingFailed)?;
             for e in state.events.iter() {
                 state.polled_undrained.insert(Key::from_token(e.token()));
+            }
+        }
+    }
+}
 /////////////////////////////////
@@ @@ -354,40 +350,20 @@ impl ChannelIdStream { @@
         self.next_channel_index += 1;
         ChannelId { controller_id: self.controller_id, channel_index: self.next_channel_index - 1 }
+    }
+}
 impl MessengerState {
     fn with_event_capacity(event_capacity: usize) -> Result<Self, std::io::Error> {
         Ok(Self {
             poll: Poll::new()?,
             events: Events::with_capacity(event_capacity),
             delayed: Default::default(),
             undelayed: Default::default(),
             polled_undrained: Default::default(),
         })
+    }
     // does NOT guarantee that events is non-empty
     fn poll_events(&mut self, deadline: Instant) -> Result<(), PollDeadlineErr> {
         use PollDeadlineErr::*;
         self.events.clear();
         let poll_timeout = deadline.checked_duration_since(Instant::now()).ok_or(Timeout)?;
         self.poll.poll(&mut self.events, Some(poll_timeout)).map_err(|_| PollingFailed)?;
         Ok(())
+    }
     fn poll_events_until(&mut self, deadline: Option<Instant>) -> Result<(), PollDeadlineErr> {
         use PollDeadlineErr::*;
         self.events.clear();
         let poll_timeout = if let Some(d) = deadline {
             Some(d.checked_duration_since(Instant::now()).ok_or(Timeout)?)
         } else {
             None
         };
         self.poll.poll(&mut self.events, poll_timeout).map_err(|_| PollingFailed)?;
         Ok(())
+    }
+}
 impl From<PollDeadlineErr> for ConnectErr {
     fn from(e: PollDeadlineErr) -> ConnectErr {
         match e {
             PollDeadlineErr::Timeout => ConnectErr::Timeout,
             PollDeadlineErr::PollingFailed => ConnectErr::PollingFailed,

src/runtime/serde.rs

➞

Show inline comments

 use crate::common::*;
 use crate::runtime::{
     endpoint::{CommMsg, CommMsgContents, EndpointInfo, Msg, SetupMsg},
+    endpoint::{CommMsg, CommMsgContents, Decision, EndpointInfo, Msg, SetupMsg},
     Predicate,
 };
 use byteorder::{BigEndian, ReadBytesExt, WriteBytesExt};
 use std::io::{ErrorKind::InvalidData, Read, Write};
 pub trait Ser<T>: Write {
@@ @@ -177,12 +177,33 @@ impl<R: Read> De<Predicate> for R { @@
         for _ in 0..len {
             assigned.insert(self.de()?, self.de()?);
+        }
         Ok(Predicate { assigned })
+    }
+}
 impl<W: Write> Ser<Decision> for W {
     fn ser(&mut self, t: &Decision) -> Result<(), std::io::Error> {
         match t {
             Decision::Failure => self.ser(&b'F'),
             Decision::Success(predicate) => {
                 self.ser(&b'S')?;
                 self.ser(predicate)
+            }
+        }
+    }
+}
 impl<R: Read> De<Decision> for R {
     fn de(&mut self) -> Result<Decision, std::io::Error> {
         let b: u8 = self.de()?;
         Ok(match b {
             b'F' => Decision::Failure,
             b'S' => Decision::Success(self.de()?),
             _ => return Err(InvalidData.into()),
         })
+    }
+}
 impl<W: Write> Ser<Polarity> for W {
     fn ser(&mut self, t: &Polarity) -> Result<(), std::io::Error> {
         self.ser(&match t {
             Polarity::Putter => b'P',
             Polarity::Getter => b'G',
@@ @@ -226,13 +247,14 @@ impl<W: Write> Ser<Msg> for W { @@
                 let zig = &ZigZag(*round_index as u64);
                 match contents {
                     SendPayload { payload_predicate, payload } => {
                         ser_seq![self, &4u8, zig, payload_predicate, payload]
+                    }
                     Elaborate { partial_oracle } => ser_seq![self, &5u8, zig, partial_oracle],
                     Announce { oracle } => ser_seq![self, &6u8, zig, oracle],
                     Announce { decision } => ser_seq![self, &6u8, zig, decision],
                     Failure => ser_seq![self, &7u8],
+                }
+            }
+        }
+    }
+}
 impl<R: Read> De<Msg> for R {
@@ @@ -249,60 +271,15 @@ impl<R: Read> De<Msg> for R { @@
             }),
             _ => {
                 let ZigZag(zig) = self.de()?;
                 let contents = match b {
 => SendPayload { payload_predicate: self.de()?, payload: self.de()? },
 => Elaborate { partial_oracle: self.de()? },
 => Announce { oracle: self.de()? },
 => Announce { decision: self.de()? },
 => Failure,
                     _ => return Err(InvalidData.into()),
                 };
                 Msg::CommMsg(CommMsg { round_index: zig as usize, contents })
+            }
         })
+    }
+}
 /////////////////
 // #[test]
 // fn my_serde() -> Result<(), std::io::Error> {
 //     let payload_predicate = Predicate {
 //         assigned: maplit::btreemap! { ChannelId {controller_id: 3, channel_index: 9} => false  },
 //     };
 //     let msg = Msg::CommMsg(CommMsg {
 //         round_index: !0,
 //         contents: CommMsgContents::SendPayload {
 //             payload_predicate,
 //             payload: (0..).take(2).collect(),
 //         },
 //     });
 //     let mut v = vec![];
 //     v.ser(&msg)?;
 //     print!("[");
 //     for (i, &x) in v.iter().enumerate() {
 //         print!("{:02x}", x);
 //         if i % 4 == 3 {
 //             print!(" ");
 //         }
 //     }
 //     println!("]");
 //     let msg2: Msg = (&v[..]).de()?;
 //     println!("msg2 {:#?}", msg2);
 //     Ok(())
 // }
 // #[test]
 // fn varint() {
 //     let mut v = vec![];
 //     v.ser(&ZigZag(!0)).unwrap();
 //     for (i, x) in v.iter_mut().enumerate() {
 //         print!("{:02x}", x);
 //         if i % 4 == 3 {
 //             print!(" ");
 //         }
 //     }
 //     *v.iter_mut().last().unwrap() |= 3;
 //     let ZigZag(x) = De::de(&mut &v[..]).unwrap();
 //     println!("");
 // }

src/runtime/setup.rs

➞

Show inline comments

@@ @@ -97,13 +97,13 @@ impl Controller { @@
         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 = Some(MonoN { ekeys: ekeys_native.into_iter().collect(), result: None });
         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}

0 comments (0 inline, 0 general)