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

Changeset - 44a98be4e4b4

Parent rev.

Child rev.

[Not reviewed]

0 17 2

Christopher Esterhuyse - 5 years ago 2020-06-19 17:04:03
christopher.esterhuyse@gmail.com

beginning large overhaul: moving to globally-unique ports & port -> endpoint route mappings

18 files changed:

Cargo.toml

src/common.rs

src/lib.rs

src/macros.rs

src/protocol/arena.rs

src/protocol/eval.rs

src/protocol/inputsource.rs

src/protocol/lexer.rs

115

src/protocol/mod.rs

src/protocol/parser.rs

src/runtime/actors.rs

src/runtime/communication.rs

src/runtime/errors.rs

src/runtime/mod.rs

737

493

src/runtime/my_tests.rs

src/runtime/serde.rs

src/runtime/setup.rs

src/runtime/setup2.rs

275

Changeset was too big and was cut off... Show full diff anyway

0 comments (0 inline, 0 general)

Cargo.toml

➞

Show inline comments

 [package]
 name = "reowolf_rs"
 version = "0.1.3"
 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.2.1"
 serde = { version = "1.0.112", features = ["derive"] }
 getrandom = "0.1.14" # tiny crate. used to guess controller-id
 take_mut = "0.2.2"
 indexmap = "1.3.0" # hashsets/hashmaps with efficient arbitrary element removal
 # network
 integer-encoding = "1.0.7"
 byteorder = "1.3.2"
-mio = "0.6.21" # migrate to mio v0.7.0 when it stabilizes
 mio = "0.6.21"
 mio-extras = "2.0.6"
 mio07 = { version = "0.7.0", package = "mio", features = ["tcp", "os-poll"] }
 # protocol
 # id-arena = "2.2.1"
 backtrace = "0.3"
 [dev-dependencies]
 test-generator = "0.3.0"
 crossbeam-utils = "0.7.0"
 lazy_static = "1.4.0"
 [lib]
 # compile target: dynamically linked library using C ABI
 crate-type = ["cdylib"]
 [features]
 default = ["ffi"]
 ffi = [] # no feature dependencies
@@ \ No newline at end of file @@

src/common.rs

➞

Show inline comments

 ///////////////////// PRELUDE /////////////////////
 pub use crate::protocol::{ComponentState, ProtocolDescription};
 pub use crate::runtime::{NonsyncContext, SyncContext};
 pub use core::{
     cmp::Ordering,
     fmt::{Debug, Formatter},
     hash::{Hash, Hasher},
     ops::{Range, RangeFrom},
     time::Duration,
 };
 pub use indexmap::{IndexMap, IndexSet};
 pub use maplit::{hashmap, hashset};
 pub use mio::{
     net::{TcpListener, TcpStream},
     Event, Evented, Events, Poll, PollOpt, Ready, Token,
 };
 pub use std::{
     collections::{hash_map::Entry, BTreeMap, HashMap, HashSet},
     convert::TryInto,
     io::{Read, Write},
     net::SocketAddr,
     sync::Arc,
     time::Instant,
 };
 pub use Polarity::*;
 ///////////////////// DEFS /////////////////////
 pub type ControllerId = u32;
-pub type ChannelIndex = u32;
+pub type PortSuffix = u32;
 #[derive(Debug, Copy, Clone, Eq, PartialEq, Ord, Hash, PartialOrd)]
 // globally unique
 #[derive(
     Copy, Clone, Eq, PartialEq, Ord, Hash, PartialOrd, serde::Serialize, serde::Deserialize,
 )]
 pub struct PortId {
     pub(crate) controller_id: ControllerId,
-    pub(crate) port_index: u32,
+    pub(crate) port_index: PortSuffix,
+}
 #[derive(Debug, Clone, Eq, PartialEq, Ord, PartialOrd)]
 pub struct Payload(Arc<Vec<u8>>);
 /// This is a unique identifier for a channel (i.e., port).
 #[derive(Debug, Eq, PartialEq, Clone, Hash, Copy, Ord, PartialOrd)]
 pub struct ChannelId {
     pub(crate) controller_id: ControllerId,
     pub(crate) channel_index: ChannelIndex,
+}
 #[derive(Debug, Eq, PartialEq, Clone, Hash, Copy, Ord, PartialOrd)]
 #[derive(
     Debug, Eq, PartialEq, Clone, Hash, Copy, Ord, PartialOrd, serde::Serialize, serde::Deserialize,
 )]
 pub enum Polarity {
     Putter, // output port (from the perspective of the component)
     Getter, // input port (from the perspective of the component)
+}
 #[derive(
     Eq, PartialEq, Ord, PartialOrd, Hash, Copy, Clone, serde::Serialize, serde::Deserialize,
 )]
 #[repr(C)]
 pub struct Port(pub u32); // ports are COPY
 #[derive(Eq, PartialEq, Copy, Clone, Debug)]
-pub enum MainComponentErr {
+pub enum AddComponentError {
     NoSuchComponent,
     NonPortTypeParameters,
-    CannotMovePort(Port),
+    CannotMovePort(PortId),
     WrongNumberOfParamaters { expected: usize },
     UnknownPort(Port),
     WrongPortPolarity { param_index: usize, port: Port },
     DuplicateMovedPort(Port),
+}
 pub trait ProtocolDescription: Sized {
     type S: ComponentState<D = Self>;
     fn parse(pdl: &[u8]) -> Result<Self, String>;
     fn component_polarities(&self, identifier: &[u8]) -> Result<Vec<Polarity>, MainComponentErr>;
     fn new_main_component(&self, identifier: &[u8], ports: &[Port]) -> Self::S;
+}
 pub trait ComponentState: Sized + Clone {
     type D: ProtocolDescription;
     fn pre_sync_run<C: MonoContext<D = Self::D, S = Self>>(
         &mut self,
         runtime_ctx: &mut C,
         protocol_description: &Self::D,
     ) -> MonoBlocker;
     fn sync_run<C: PolyContext<D = Self::D>>(
         &mut self,
         runtime_ctx: &mut C,
         protocol_description: &Self::D,
     ) -> PolyBlocker;
     UnknownPort(PortId),
     WrongPortPolarity { port: PortId, expected_polarity: Polarity },
     DuplicateMovedPort(PortId),
+}
 #[derive(Debug, Clone)]
-pub enum MonoBlocker {
+pub enum NonsyncBlocker {
     Inconsistent,
     ComponentExit,
     SyncBlockStart,
+}
 #[derive(Debug, Clone)]
-pub enum PolyBlocker {
+pub enum SyncBlocker {
     Inconsistent,
     SyncBlockEnd,
     CouldntReadMsg(Port),
     CouldntCheckFiring(Port),
     PutMsg(Port, Payload),
+}
 pub trait MonoContext {
     type D: ProtocolDescription;
     type S: ComponentState<D = Self::D>;
     fn new_component(&mut self, moved_ports: HashSet<Port>, init_state: Self::S);
     fn new_channel(&mut self) -> [Port; 2];
     fn new_random(&mut self) -> u64;
+}
 pub trait PolyContext {
     type D: ProtocolDescription;
     fn is_firing(&mut self, port: Port) -> Option<bool>;
     fn read_msg(&mut self, port: Port) -> Option<&Payload>;
     CouldntReadMsg(PortId),
     CouldntCheckFiring(PortId),
     PutMsg(PortId, Payload),
+}
 ///////////////////// IMPL /////////////////////
 impl Payload {
     pub fn new(len: usize) -> Payload {
         let mut v = Vec::with_capacity(len);
         unsafe {
             v.set_len(len);
+        }
         Payload(Arc::new(v))
+    }
     pub fn len(&self) -> usize {
         self.0.len()
+    }
     pub fn as_slice(&self) -> &[u8] {
         &self.0
+    }
     pub fn as_mut_slice(&mut self) -> &mut [u8] {
         Arc::make_mut(&mut self.0) as _
+    }
     pub fn concat_with(&mut self, other: &Self) {
         let bytes = other.as_slice().iter().copied();
         let me = Arc::make_mut(&mut self.0);
         me.extend(bytes);
+    }
+}
 impl serde::Serialize for Payload {
     fn serialize<S>(
         &self,
         serializer: S,
     ) -> std::result::Result<<S as serde::Serializer>::Ok, <S as serde::Serializer>::Error>
     where
         S: serde::Serializer,
+    {
         let inner: &Vec<u8> = &self.0;
         inner.serialize(serializer)
+    }
+}
 impl<'de> serde::Deserialize<'de> for Payload {
     fn deserialize<D>(
         deserializer: D,
     ) -> std::result::Result<Self, <D as serde::Deserializer<'de>>::Error>
     where
         D: serde::Deserializer<'de>,
+    {
         let inner: Vec<u8> = Vec::deserialize(deserializer)?;
         Ok(Self(Arc::new(inner)))
+    }
+}
 impl std::iter::FromIterator<u8> for Payload {
     fn from_iter<I: IntoIterator<Item = u8>>(it: I) -> Self {
         Self(Arc::new(it.into_iter().collect()))
+    }
+}
 impl From<Vec<u8>> for Payload {
     fn from(s: Vec<u8>) -> Self {
         Self(s.into())
+    }
+}
-impl Debug for Port {
+impl Debug for PortId {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         write!(f, "Port({})", self.0)
+    }
+}
 impl Port {
     pub fn from_raw(raw: u32) -> Self {
         Self(raw)
+    }
     pub fn to_raw(self) -> u32 {
         self.0
+    }
     pub fn to_token(self) -> mio::Token {
         mio::Token(self.0.try_into().unwrap())
+    }
     pub fn from_token(t: mio::Token) -> Self {
         Self(t.0.try_into().unwrap())
         write!(f, "PortId({},{})", self.controller_id, self.port_index)
+    }
+}

src/lib.rs

➞

Show inline comments

 #[macro_use]
 mod macros;
 mod common;
 mod protocol;
 mod runtime;
 #[cfg(test)]
 mod test;
 // #[cfg(test)]
 // mod test;
 pub use runtime::{errors, Connector, PortBinding};
 pub use common::Polarity;
 pub use protocol::ProtocolDescription;
 pub use runtime::{Connector, EndpointSetup, StringLogger};
 #[cfg(feature = "ffi")]
 pub use runtime::ffi;
 // #[cfg(feature = "ffi")]
 // pub use runtime::ffi;

src/macros.rs

➞

Show inline comments

 macro_rules! log {
     ($logger:expr, $($arg:tt)*) => {{
         use std::fmt::Write;
         writeln!($logger, $($arg)*).unwrap();
         let _ = write!($logger.line_writer(), $($arg)*).unwrap();
     }};
+}
 macro_rules! assert_let {
     ($pat:pat = $expr:expr => $work:expr) => {
         if let $pat = $expr {
             $work
         } else {
             panic!("assert_let failed");
+        }
     };
+}
 // macro_rules! assert_let {
 //     ($pat:pat = $expr:expr => $work:expr) => {
 //         if let $pat = $expr {
 //             $work
 //         } else {
 //             panic!("assert_let failed");
 //         }
 //     };
 // }
 #[test]
 fn assert_let() {
     let x = Some(5);
     let z = assert_let![Some(y) = x => {
         println!("{:?}", y);
     }];
     println!("{:?}", z);
+}
 // #[test]
 // fn assert_let() {
 //     let x = Some(5);
 //     let z = assert_let![Some(y) = x => {
 //         println!("{:?}", y);
 //         3
 //     }];
 //     println!("{:?}", z);
 // }
 #[test]
 #[should_panic]
 fn must_let_panic() {
     let x: Option<u32> = None;
     assert_let![Some(y) = x => {
         println!("{:?}", y);
     }];
+}
 // #[test]
 // #[should_panic]
 // fn must_let_panic() {
 //     let x: Option<u32> = None;
 //     assert_let![Some(y) = x => {
 //         println!("{:?}", y);
 //     }];
 // }

src/protocol/arena.rs

➞

Show inline comments

 use crate::common::*;
 use core::hash::Hash;
 use core::marker::PhantomData;
-#[derive(Debug, serde::Serialize, serde::Deserialize)]
 #[derive(serde::Serialize, serde::Deserialize)]
 pub struct Id<T> {
     index: u32,
     _phantom: PhantomData<T>,
+}
 #[derive(Debug, serde::Serialize, serde::Deserialize)]
 pub struct Arena<T> {
     store: Vec<T>,
+}
 //////////////////////////////////
 impl<T> Debug for Id<T> {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         f.debug_struct("Id").field("index", &self.index).finish()
+    }
+}
 impl<T> Clone for Id<T> {
     fn clone(&self) -> Self {
         *self
+    }
+}
 impl<T> Copy for Id<T> {}
 impl<T> PartialEq for Id<T> {
     fn eq(&self, other: &Self) -> bool {
         self.index.eq(&other.index)
+    }
+}
 impl<T> Eq for Id<T> {}
 impl<T> Hash for Id<T> {
     fn hash<H: std::hash::Hasher>(&self, h: &mut H) {
         self.index.hash(h);
+    }
+}
 #[derive(Debug, serde::Serialize, serde::Deserialize)]
 pub struct Arena<T> {
     store: Vec<T>,
+}
 impl<T> Arena<T> {
     pub fn new() -> Self {
         Self { store: vec![] }
+    }
     pub fn alloc_with_id(&mut self, f: impl FnOnce(Id<T>) -> T) -> Id<T> {
         use std::convert::TryFrom;
         let id = Id {
             index: u32::try_from(self.store.len()).expect("Out of capacity!"),
             _phantom: Default::default(),
         };
         self.store.push(f(id));
         id
+    }
     pub fn iter(&self) -> impl Iterator<Item = (Id<T>, &T)> {
         (0..).map(|index| Id { index, _phantom: Default::default() }).zip(self.store.iter())
+    }
+}
 impl<T> core::ops::Index<Id<T>> for Arena<T> {
     type Output = T;
     fn index(&self, id: Id<T>) -> &Self::Output {
         self.store.index(id.index as usize)
+    }
+}
 impl<T> core::ops::IndexMut<Id<T>> for Arena<T> {
     fn index_mut(&mut self, id: Id<T>) -> &mut Self::Output {
         self.store.index_mut(id.index as usize)
+    }
+}

src/protocol/eval.rs

➞

Show inline comments

@@ @@ -697,410 +697,410 @@ impl Value { @@
+            }
             (PrimitiveType::Long, PrimitiveType::Short) => {
                 Value::Boolean(BooleanValue(i64::from(self) >= i64::from(other)))
+            }
             (PrimitiveType::Long, PrimitiveType::Int) => {
                 Value::Boolean(BooleanValue(i64::from(self) >= i64::from(other)))
+            }
             (PrimitiveType::Long, PrimitiveType::Long) => {
                 Value::Boolean(BooleanValue(i64::from(self) >= i64::from(other)))
+            }
             _ => unimplemented!(),
+        }
+    }
     fn as_boolean(&self) -> &BooleanValue {
         match self {
             Value::Boolean(result) => result,
             _ => panic!("Unable to cast `Value` to `BooleanValue`"),
+        }
+    }
+}
 impl From<bool> for Value {
     fn from(b: bool) -> Self {
         Value::Boolean(BooleanValue(b))
+    }
+}
 impl From<Value> for bool {
     fn from(val: Value) -> Self {
         match val {
             Value::Boolean(BooleanValue(b)) => b,
             _ => unimplemented!(),
+        }
+    }
+}
 impl From<&Value> for bool {
     fn from(val: &Value) -> Self {
         match val {
             Value::Boolean(BooleanValue(b)) => *b,
             _ => unimplemented!(),
+        }
+    }
+}
 impl From<Value> for i8 {
     fn from(val: Value) -> Self {
         match val {
             Value::Byte(ByteValue(b)) => b,
             _ => unimplemented!(),
+        }
+    }
+}
 impl From<&Value> for i8 {
     fn from(val: &Value) -> Self {
         match val {
             Value::Byte(ByteValue(b)) => *b,
             _ => unimplemented!(),
+        }
+    }
+}
 impl From<Value> for i16 {
     fn from(val: Value) -> Self {
         match val {
             Value::Byte(ByteValue(b)) => i16::from(b),
             Value::Short(ShortValue(s)) => s,
             _ => unimplemented!(),
+        }
+    }
+}
 impl From<&Value> for i16 {
     fn from(val: &Value) -> Self {
         match val {
             Value::Byte(ByteValue(b)) => i16::from(*b),
             Value::Short(ShortValue(s)) => *s,
             _ => unimplemented!(),
+        }
+    }
+}
 impl From<Value> for i32 {
     fn from(val: Value) -> Self {
         match val {
             Value::Byte(ByteValue(b)) => i32::from(b),
             Value::Short(ShortValue(s)) => i32::from(s),
             Value::Int(IntValue(i)) => i,
             _ => unimplemented!(),
+        }
+    }
+}
 impl From<&Value> for i32 {
     fn from(val: &Value) -> Self {
         match val {
             Value::Byte(ByteValue(b)) => i32::from(*b),
             Value::Short(ShortValue(s)) => i32::from(*s),
             Value::Int(IntValue(i)) => *i,
             _ => unimplemented!(),
+        }
+    }
+}
 impl From<Value> for i64 {
     fn from(val: Value) -> Self {
         match val {
             Value::Byte(ByteValue(b)) => i64::from(b),
             Value::Short(ShortValue(s)) => i64::from(s),
             Value::Int(IntValue(i)) => i64::from(i),
             Value::Long(LongValue(l)) => l,
             _ => unimplemented!(),
+        }
+    }
+}
 impl From<&Value> for i64 {
     fn from(val: &Value) -> Self {
         match val {
             Value::Byte(ByteValue(b)) => i64::from(*b),
             Value::Short(ShortValue(s)) => i64::from(*s),
             Value::Int(IntValue(i)) => i64::from(*i),
             Value::Long(LongValue(l)) => *l,
             _ => unimplemented!(),
+        }
+    }
+}
 impl ValueImpl for Value {
     fn exact_type(&self) -> Type {
         match self {
             Value::Input(val) => val.exact_type(),
             Value::Output(val) => val.exact_type(),
             Value::Message(val) => val.exact_type(),
             Value::Boolean(val) => val.exact_type(),
             Value::Byte(val) => val.exact_type(),
             Value::Short(val) => val.exact_type(),
             Value::Int(val) => val.exact_type(),
             Value::Long(val) => val.exact_type(),
             Value::InputArray(val) => val.exact_type(),
             Value::OutputArray(val) => val.exact_type(),
             Value::MessageArray(val) => val.exact_type(),
             Value::BooleanArray(val) => val.exact_type(),
             Value::ByteArray(val) => val.exact_type(),
             Value::ShortArray(val) => val.exact_type(),
             Value::IntArray(val) => val.exact_type(),
             Value::LongArray(val) => val.exact_type(),
+        }
+    }
     fn is_type_compatible(&self, t: &Type) -> bool {
         match self {
             Value::Input(val) => val.is_type_compatible(t),
             Value::Output(val) => val.is_type_compatible(t),
             Value::Message(val) => val.is_type_compatible(t),
             Value::Boolean(val) => val.is_type_compatible(t),
             Value::Byte(val) => val.is_type_compatible(t),
             Value::Short(val) => val.is_type_compatible(t),
             Value::Int(val) => val.is_type_compatible(t),
             Value::Long(val) => val.is_type_compatible(t),
             Value::InputArray(val) => val.is_type_compatible(t),
             Value::OutputArray(val) => val.is_type_compatible(t),
             Value::MessageArray(val) => val.is_type_compatible(t),
             Value::BooleanArray(val) => val.is_type_compatible(t),
             Value::ByteArray(val) => val.is_type_compatible(t),
             Value::ShortArray(val) => val.is_type_compatible(t),
             Value::IntArray(val) => val.is_type_compatible(t),
             Value::LongArray(val) => val.is_type_compatible(t),
+        }
+    }
+}
 impl Display for Value {
     fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
         let disp: &dyn Display;
         match self {
             Value::Input(val) => disp = val,
             Value::Output(val) => disp = val,
             Value::Message(val) => disp = val,
             Value::Boolean(val) => disp = val,
             Value::Byte(val) => disp = val,
             Value::Short(val) => disp = val,
             Value::Int(val) => disp = val,
             Value::Long(val) => disp = val,
             Value::InputArray(val) => disp = val,
             Value::OutputArray(val) => disp = val,
             Value::MessageArray(val) => disp = val,
             Value::BooleanArray(val) => disp = val,
             Value::ByteArray(val) => disp = val,
             Value::ShortArray(val) => disp = val,
             Value::IntArray(val) => disp = val,
             Value::LongArray(val) => disp = val,
+        }
         disp.fmt(f)
+    }
+}
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
-pub struct InputValue(pub Port);
+pub struct InputValue(pub PortId);
 impl Display for InputValue {
     fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
         write!(f, "#in")
+    }
+}
 impl ValueImpl for InputValue {
     fn exact_type(&self) -> Type {
         Type::INPUT
+    }
     fn is_type_compatible(&self, t: &Type) -> bool {
         let Type { primitive, array } = t;
         if *array {
             return false;
+        }
         match primitive {
             PrimitiveType::Input => true,
             _ => false,
+        }
+    }
+}
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
-pub struct OutputValue(pub Port);
+pub struct OutputValue(pub PortId);
 impl Display for OutputValue {
     fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
         write!(f, "#out")
+    }
+}
 impl ValueImpl for OutputValue {
     fn exact_type(&self) -> Type {
         Type::OUTPUT
+    }
     fn is_type_compatible(&self, t: &Type) -> bool {
         let Type { primitive, array } = t;
         if *array {
             return false;
+        }
         match primitive {
             PrimitiveType::Output => true,
             _ => false,
+        }
+    }
+}
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 pub struct MessageValue(pub Option<Payload>);
 impl Display for MessageValue {
     fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
         match &self.0 {
             None => write!(f, "null"),
             Some(payload) => {
                 // format print up to 10 bytes
                 let mut slice = payload.as_slice();
                 if slice.len() > 10 {
                     slice = &slice[..10];
+                }
                 f.debug_list().entries(slice.iter().copied()).finish()
+            }
+        }
+    }
+}
 impl ValueImpl for MessageValue {
     fn exact_type(&self) -> Type {
         Type::MESSAGE
+    }
     fn is_type_compatible(&self, t: &Type) -> bool {
         let Type { primitive, array } = t;
         if *array {
             return false;
+        }
         match primitive {
             PrimitiveType::Message => true,
             _ => false,
+        }
+    }
+}
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 pub struct BooleanValue(bool);
 impl Display for BooleanValue {
     fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
         write!(f, "{}", self.0)
+    }
+}
 impl ValueImpl for BooleanValue {
     fn exact_type(&self) -> Type {
         Type::BOOLEAN
+    }
     fn is_type_compatible(&self, t: &Type) -> bool {
         let Type { primitive, array } = t;
         if *array {
             return false;
+        }
         match primitive {
             PrimitiveType::Boolean => true,
             PrimitiveType::Byte => true,
             PrimitiveType::Short => true,
             PrimitiveType::Int => true,
             PrimitiveType::Long => true,
             _ => false,
+        }
+    }
+}
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 pub struct ByteValue(i8);
 impl Display for ByteValue {
     fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
         write!(f, "{}", self.0)
+    }
+}
 impl ValueImpl for ByteValue {
     fn exact_type(&self) -> Type {
         Type::BYTE
+    }
     fn is_type_compatible(&self, t: &Type) -> bool {
         let Type { primitive, array } = t;
         if *array {
             return false;
+        }
         match primitive {
             PrimitiveType::Byte => true,
             PrimitiveType::Short => true,
             PrimitiveType::Int => true,
             PrimitiveType::Long => true,
             _ => false,
+        }
+    }
+}
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 pub struct ShortValue(i16);
 impl Display for ShortValue {
     fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
         write!(f, "{}", self.0)
+    }
+}
 impl ValueImpl for ShortValue {
     fn exact_type(&self) -> Type {
         Type::SHORT
+    }
     fn is_type_compatible(&self, t: &Type) -> bool {
         let Type { primitive, array } = t;
         if *array {
             return false;
+        }
         match primitive {
             PrimitiveType::Short => true,
             PrimitiveType::Int => true,
             PrimitiveType::Long => true,
             _ => false,
+        }
+    }
+}
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 pub struct IntValue(i32);
 impl Display for IntValue {
     fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
         write!(f, "{}", self.0)
+    }
+}
 impl ValueImpl for IntValue {
     fn exact_type(&self) -> Type {
         Type::INT
+    }
     fn is_type_compatible(&self, t: &Type) -> bool {
         let Type { primitive, array } = t;
         if *array {
             return false;
+        }
         match primitive {
             PrimitiveType::Int => true,
             PrimitiveType::Long => true,
             _ => false,
+        }
+    }
+}
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 pub struct LongValue(i64);
 impl Display for LongValue {
     fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
         write!(f, "{}", self.0)
+    }
+}
 impl ValueImpl for LongValue {
     fn exact_type(&self) -> Type {
         Type::LONG
+    }
     fn is_type_compatible(&self, t: &Type) -> bool {
         let Type { primitive, array } = t;
         if *array {
             return false;
+        }
         match primitive {
             PrimitiveType::Long => true,
             _ => false,
+        }
+    }
+}
@@ @@ -1614,227 +1614,227 @@ pub enum EvalContinuation { @@
 pub struct Prompt {
     definition: DefinitionId,
     store: Store,
     position: Option<StatementId>,
+}
 impl Prompt {
     pub fn new(h: &Heap, def: DefinitionId, args: &Vec<Value>) -> Self {
         let mut prompt =
             Prompt { definition: def, store: Store::new(), position: Some((&h[def]).body()) };
         prompt.set_arguments(h, args);
         prompt
+    }
     fn set_arguments(&mut self, h: &Heap, args: &Vec<Value>) {
         let def = &h[self.definition];
         let params = def.parameters();
         assert_eq!(params.len(), args.len());
         for (param, value) in params.iter().zip(args.iter()) {
             let hparam = &h[*param];
             let type_annot = &h[hparam.type_annotation];
             assert!(value.is_type_compatible(&type_annot.the_type));
             self.store.initialize(h, param.upcast(), value.clone());
+        }
+    }
     pub fn step(&mut self, h: &Heap, ctx: &mut EvalContext) -> EvalResult {
         if self.position.is_none() {
             return Err(EvalContinuation::Terminal);
+        }
         let stmt = &h[self.position.unwrap()];
         match stmt {
             Statement::Block(stmt) => {
                 // Continue to first statement
                 self.position = Some(stmt.first());
                 Err(EvalContinuation::Stepping)
+            }
             Statement::Local(stmt) => {
                 match stmt {
                     LocalStatement::Memory(stmt) => {
                         // Evaluate initial expression
                         let value = self.store.eval(h, ctx, stmt.initial)?;
                         // Update store
                         self.store.initialize(h, stmt.variable.upcast(), value);
+                    }
                     LocalStatement::Channel(stmt) => {
                         let [from, to] = ctx.new_channel();
                         // Store the values in the declared variables
                         self.store.initialize(h, stmt.from.upcast(), from);
                         self.store.initialize(h, stmt.to.upcast(), to);
+                    }
+                }
                 // Continue to next statement
                 self.position = stmt.next();
                 Err(EvalContinuation::Stepping)
+            }
             Statement::Skip(stmt) => {
                 // Continue to next statement
                 self.position = stmt.next;
                 Err(EvalContinuation::Stepping)
+            }
             Statement::Labeled(stmt) => {
                 // Continue to next statement
                 self.position = Some(stmt.body);
                 Err(EvalContinuation::Stepping)
+            }
             Statement::If(stmt) => {
                 // Evaluate test
                 let value = self.store.eval(h, ctx, stmt.test)?;
                 // Continue with either branch
                 if value.as_boolean().0 {
                     self.position = Some(stmt.true_body);
                 } else {
                     self.position = Some(stmt.false_body);
+                }
                 Err(EvalContinuation::Stepping)
+            }
             Statement::EndIf(stmt) => {
                 // Continue to next statement
                 self.position = stmt.next;
                 Err(EvalContinuation::Stepping)
+            }
             Statement::While(stmt) => {
                 // Evaluate test
                 let value = self.store.eval(h, ctx, stmt.test)?;
                 // Either continue with body, or go to next
                 if value.as_boolean().0 {
                     self.position = Some(stmt.body);
                 } else {
                     self.position = stmt.next.map(|x| x.upcast());
+                }
                 Err(EvalContinuation::Stepping)
+            }
             Statement::EndWhile(stmt) => {
                 // Continue to next statement
                 self.position = stmt.next;
                 Err(EvalContinuation::Stepping)
+            }
             Statement::Synchronous(stmt) => {
                 // Continue to next statement, and signal upward
                 self.position = Some(stmt.body);
                 Err(EvalContinuation::SyncBlockStart)
+            }
             Statement::EndSynchronous(stmt) => {
                 // Continue to next statement, and signal upward
                 self.position = stmt.next;
                 Err(EvalContinuation::SyncBlockEnd)
+            }
             Statement::Break(stmt) => {
                 // Continue to end of while
                 self.position = stmt.target.map(EndWhileStatementId::upcast);
                 Err(EvalContinuation::Stepping)
+            }
             Statement::Continue(stmt) => {
                 // Continue to beginning of while
                 self.position = stmt.target.map(WhileStatementId::upcast);
                 Err(EvalContinuation::Stepping)
+            }
             Statement::Assert(stmt) => {
                 // Evaluate expression
                 let value = self.store.eval(h, ctx, stmt.expression)?;
                 if value.as_boolean().0 {
                     // Continue to next statement
                     self.position = stmt.next;
                     Err(EvalContinuation::Stepping)
                 } else {
                     // Assertion failed: inconsistent
                     Err(EvalContinuation::Inconsistent)
+                }
+            }
             Statement::Return(stmt) => {
                 // Evaluate expression
                 let value = self.store.eval(h, ctx, stmt.expression)?;
                 // Done with evaluation
                 Ok(value)
+            }
             Statement::Goto(stmt) => {
                 // Continue to target
                 self.position = stmt.target.map(|x| x.upcast());
                 Err(EvalContinuation::Stepping)
+            }
             Statement::New(stmt) => {
                 let expr = &h[stmt.expression];
                 let mut args = Vec::new();
                 for &arg in expr.arguments.iter() {
                     let value = self.store.eval(h, ctx, arg)?;
                     args.push(value);
+                }
                 self.position = stmt.next;
                 Err(EvalContinuation::NewComponent(expr.declaration.unwrap(), args))
+            }
             Statement::Put(stmt) => {
                 // Evaluate port and message
                 let port = self.store.eval(h, ctx, stmt.port)?;
                 let message = self.store.eval(h, ctx, stmt.message)?;
                 // Continue to next statement
                 self.position = stmt.next;
                 // Signal the put upwards
                 Err(EvalContinuation::Put(port, message))
+            }
             Statement::Expression(stmt) => {
                 // Evaluate expression
                 let value = self.store.eval(h, ctx, stmt.expression)?;
                 // Continue to next statement
                 self.position = stmt.next;
                 Err(EvalContinuation::Stepping)
+            }
+        }
+    }
     fn compute_function(h: &Heap, fun: FunctionId, args: &Vec<Value>) -> Option<Value> {
         let mut prompt = Self::new(h, fun.upcast(), args);
         let mut context = EvalContext::None;
         loop {
             let result = prompt.step(h, &mut context);
             match result {
                 Ok(val) => return Some(val),
                 Err(cont) => match cont {
                     EvalContinuation::Stepping => continue,
                     EvalContinuation::Inconsistent => return None,
                     // Functions never terminate without returning
                     EvalContinuation::Terminal => unreachable!(),
                     // Functions never encounter any blocking behavior
                     EvalContinuation::SyncBlockStart => unreachable!(),
                     EvalContinuation::SyncBlockEnd => unreachable!(),
                     EvalContinuation::NewComponent(_, _) => unreachable!(),
                     EvalContinuation::BlockFires(val) => unreachable!(),
                     EvalContinuation::BlockGet(val) => unreachable!(),
                     EvalContinuation::Put(port, msg) => unreachable!(),
                 },
+            }
+        }
+    }
+}
 #[cfg(test)]
 mod tests {
     extern crate test_generator;
     use std::fs::File;
     use std::io::Read;
     use std::path::Path;
     use test_generator::test_resources;
     use super::*;
     #[test_resources("testdata/eval/positive/*.pdl")]
     fn batch1(resource: &str) {
         let path = Path::new(resource);
         let expect = path.with_extension("txt");
         let mut heap = Heap::new();
         let mut source = InputSource::from_file(&path).unwrap();
         let mut parser = Parser::new(&mut source);
         let pd = parser.parse(&mut heap).unwrap();
         let def = heap[pd].get_definition_ident(&heap, b"test").unwrap();
         let fun = heap[def].as_function().this;
         let args = Vec::new();
         let result = Prompt::compute_function(&heap, fun, &args).unwrap();
         let valstr: String = format!("{}", result);
         println!("{}", valstr);
         let mut cev: Vec<u8> = Vec::new();
         let mut f = File::open(expect).unwrap();
         f.read_to_end(&mut cev).unwrap();
         let lavstr = String::from_utf8_lossy(&cev);
         println!("{}", lavstr);
         assert_eq!(valstr, lavstr);
+    }
+}
 // #[cfg(test)]
 // mod tests {
 //     extern crate test_generator;
 //     use std::fs::File;
 //     use std::io::Read;
 //     use std::path::Path;
 //     use test_generator::test_resources;
 //     use super::*;
 //     #[test_resources("testdata/eval/positive/*.pdl")]
 //     fn batch1(resource: &str) {
 //         let path = Path::new(resource);
 //         let expect = path.with_extension("txt");
 //         let mut heap = Heap::new();
 //         let mut source = InputSource::from_file(&path).unwrap();
 //         let mut parser = Parser::new(&mut source);
 //         let pd = parser.parse(&mut heap).unwrap();
 //         let def = heap[pd].get_definition_ident(&heap, b"test").unwrap();
 //         let fun = heap[def].as_function().this;
 //         let args = Vec::new();
 //         let result = Prompt::compute_function(&heap, fun, &args).unwrap();
 //         let valstr: String = format!("{}", result);
 //         println!("{}", valstr);
 //         let mut cev: Vec<u8> = Vec::new();
 //         let mut f = File::open(expect).unwrap();
 //         f.read_to_end(&mut cev).unwrap();
 //         let lavstr = String::from_utf8_lossy(&cev);
 //         println!("{}", lavstr);
 //         assert_eq!(valstr, lavstr);
 //     }
 // }

src/protocol/inputsource.rs

➞

Show inline comments

@@ @@ -122,285 +122,285 @@ impl fmt::Display for InputSource { @@
 #[derive(Debug, Clone, Copy, Default, serde::Serialize, serde::Deserialize)]
 pub struct InputPosition {
     line: usize,
     column: usize,
     offset: usize,
+}
 impl InputPosition {
     fn context<'a>(&self, source: &'a InputSource) -> &'a [u8] {
         let start = self.offset - (self.column - 1);
         let mut end = self.offset;
         while end < source.input.len() {
             let cur = (*source.input)[end];
             if cur == b'\n' || cur == b'\r' {
                 break;
+            }
             end += 1;
+        }
         &source.input[start..end]
+    }
     fn parse_error<S: ToString>(&self, message: S) -> ParseError {
         ParseError { position: *self, message: message.to_string(), backtrace: Backtrace::new() }
+    }
     fn eval_error<S: ToString>(&self, message: S) -> EvalError {
         EvalError { position: *self, message: message.to_string(), backtrace: Backtrace::new() }
+    }
+}
 impl fmt::Display for InputPosition {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         write!(f, "{}:{}", self.line, self.column)
+    }
+}
 pub trait SyntaxElement {
     fn position(&self) -> InputPosition;
     fn error<S: ToString>(&self, message: S) -> EvalError {
         self.position().eval_error(message)
+    }
+}
 #[derive(Debug, Clone)]
 pub struct ParseError {
     position: InputPosition,
     message: String,
     backtrace: Backtrace,
+}
 impl ParseError {
     pub fn new<S: ToString>(position: InputPosition, message: S) -> ParseError {
         ParseError { position, message: message.to_string(), backtrace: Backtrace::new() }
+    }
     // Diagnostic methods
     pub fn write<A: io::Write>(&self, source: &InputSource, writer: &mut A) -> io::Result<()> {
         if !source.filename.is_empty() {
             writeln!(
                 writer,
                 "Parse error at {}:{}: {}",
                 source.filename, self.position, self.message
             )?;
         } else {
             writeln!(writer, "Parse error at {}: {}", self.position, self.message)?;
+        }
         let line = self.position.context(source);
         writeln!(writer, "{}", String::from_utf8_lossy(line))?;
         let mut arrow: Vec<u8> = Vec::new();
         for pos in 1..self.position.column {
             let c = line[pos - 1];
             if c == b'\t' {
                 arrow.push(b'\t')
             } else {
                 arrow.push(b' ')
+            }
+        }
         arrow.push(b'^');
         writeln!(writer, "{}", String::from_utf8_lossy(&arrow))
+    }
     pub fn print(&self, source: &InputSource) {
         self.write(source, &mut std::io::stdout()).unwrap()
+    }
     pub fn display<'a>(&'a self, source: &'a InputSource) -> DisplayParseError<'a> {
         DisplayParseError::new(self, source)
+    }
+}
 impl From<ParseError> for io::Error {
     fn from(_: ParseError) -> io::Error {
         io::Error::new(io::ErrorKind::InvalidInput, "parse error")
+    }
+}
 #[derive(Clone, Copy)]
 pub struct DisplayParseError<'a> {
     error: &'a ParseError,
     source: &'a InputSource,
+}
 impl DisplayParseError<'_> {
     fn new<'a>(error: &'a ParseError, source: &'a InputSource) -> DisplayParseError<'a> {
         DisplayParseError { error, source }
+    }
+}
 impl fmt::Display for DisplayParseError<'_> {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         let mut vec: Vec<u8> = Vec::new();
         match self.error.write(self.source, &mut vec) {
             Err(_) => {
                 return fmt::Result::Err(fmt::Error);
+            }
             Ok(_) => {}
+        }
         write!(f, "{}", String::from_utf8_lossy(&vec))
+    }
+}
 #[derive(Debug, Clone)]
 pub struct EvalError {
     position: InputPosition,
     message: String,
     backtrace: Backtrace,
+}
 impl EvalError {
     pub fn new<S: ToString>(position: InputPosition, message: S) -> EvalError {
         EvalError { position, message: message.to_string(), backtrace: Backtrace::new() }
+    }
     // Diagnostic methods
     pub fn write<A: io::Write>(&self, source: &InputSource, writer: &mut A) -> io::Result<()> {
         if !source.filename.is_empty() {
             writeln!(
                 writer,
                 "Evaluation error at {}:{}: {}",
                 source.filename, self.position, self.message
             )?;
         } else {
             writeln!(writer, "Evaluation error at {}: {}", self.position, self.message)?;
+        }
         let line = self.position.context(source);
         writeln!(writer, "{}", String::from_utf8_lossy(line))?;
         let mut arrow: Vec<u8> = Vec::new();
         for pos in 1..self.position.column {
             let c = line[pos - 1];
             if c == b'\t' {
                 arrow.push(b'\t')
             } else {
                 arrow.push(b' ')
+            }
+        }
         arrow.push(b'^');
         writeln!(writer, "{}", String::from_utf8_lossy(&arrow))
+    }
     pub fn print(&self, source: &InputSource) {
         self.write(source, &mut std::io::stdout()).unwrap()
+    }
     pub fn display<'a>(&'a self, source: &'a InputSource) -> DisplayEvalError<'a> {
         DisplayEvalError::new(self, source)
+    }
+}
 impl From<EvalError> for io::Error {
     fn from(_: EvalError) -> io::Error {
         io::Error::new(io::ErrorKind::InvalidInput, "eval error")
+    }
+}
 #[derive(Clone, Copy)]
 pub struct DisplayEvalError<'a> {
     error: &'a EvalError,
     source: &'a InputSource,
+}
 impl DisplayEvalError<'_> {
     fn new<'a>(error: &'a EvalError, source: &'a InputSource) -> DisplayEvalError<'a> {
         DisplayEvalError { error, source }
+    }
+}
 impl fmt::Display for DisplayEvalError<'_> {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         let mut vec: Vec<u8> = Vec::new();
         match self.error.write(self.source, &mut vec) {
             Err(_) => {
                 return fmt::Result::Err(fmt::Error);
+            }
             Ok(_) => {}
+        }
         write!(f, "{}", String::from_utf8_lossy(&vec))
+    }
+}
 #[cfg(test)]
 mod tests {
     use super::*;
 // #[cfg(test)]
 // mod tests {
 //     use super::*;
     #[test]
     fn test_from_string() {
         let mut is = InputSource::from_string("#version 100\n").unwrap();
         assert!(is.input.len() == 13);
         assert!(is.line == 1);
         assert!(is.column == 1);
         assert!(is.offset == 0);
         let ps = is.pos();
         assert!(ps.line == 1);
         assert!(ps.column == 1);
         assert!(ps.offset == 0);
         assert!(is.next() == Some(b'#'));
         is.consume();
         assert!(is.next() == Some(b'v'));
         assert!(is.lookahead(1) == Some(b'e'));
         is.consume();
         assert!(is.next() == Some(b'e'));
         is.consume();
         assert!(is.next() == Some(b'r'));
         is.consume();
         assert!(is.next() == Some(b's'));
         is.consume();
         assert!(is.next() == Some(b'i'));
         is.consume();
+        {
             let ps = is.pos();
             assert_eq!(b"#version 100", ps.context(&is));
             let er = is.error("hello world!");
             let mut vec: Vec<u8> = Vec::new();
             er.write(&is, &mut vec).unwrap();
             assert_eq!(
                 "Parse error at 1:7: hello world!\n#version 100\n      ^\n",
                 String::from_utf8_lossy(&vec)
             );
+        }
         assert!(is.next() == Some(b'o'));
         is.consume();
         assert!(is.next() == Some(b'n'));
         is.consume();
         assert!(is.input.len() == 13);
         assert!(is.line == 1);
         assert!(is.column == 9);
         assert!(is.offset == 8);
         assert!(is.next() == Some(b' '));
         is.consume();
         assert!(is.next() == Some(b'1'));
         is.consume();
         assert!(is.next() == Some(b'0'));
         is.consume();
         assert!(is.next() == Some(b'0'));
         is.consume();
         assert!(is.input.len() == 13);
         assert!(is.line == 1);
         assert!(is.column == 13);
         assert!(is.offset == 12);
         assert!(is.next() == Some(b'\n'));
         is.consume();
         assert!(is.input.len() == 13);
         assert!(is.line == 2);
         assert!(is.column == 1);
         assert!(is.offset == 13);
+        {
             let ps = is.pos();
             assert_eq!(b"", ps.context(&is));
+        }
         assert!(is.next() == None);
         is.consume();
         assert!(is.next() == None);
+    }
 //     #[test]
 //     fn test_from_string() {
 //         let mut is = InputSource::from_string("#version 100\n").unwrap();
 //         assert!(is.input.len() == 13);
 //         assert!(is.line == 1);
 //         assert!(is.column == 1);
 //         assert!(is.offset == 0);
 //         let ps = is.pos();
 //         assert!(ps.line == 1);
 //         assert!(ps.column == 1);
 //         assert!(ps.offset == 0);
 //         assert!(is.next() == Some(b'#'));
 //         is.consume();
 //         assert!(is.next() == Some(b'v'));
 //         assert!(is.lookahead(1) == Some(b'e'));
 //         is.consume();
 //         assert!(is.next() == Some(b'e'));
 //         is.consume();
 //         assert!(is.next() == Some(b'r'));
 //         is.consume();
 //         assert!(is.next() == Some(b's'));
 //         is.consume();
 //         assert!(is.next() == Some(b'i'));
 //         is.consume();
 //         {
 //             let ps = is.pos();
 //             assert_eq!(b"#version 100", ps.context(&is));
 //             let er = is.error("hello world!");
 //             let mut vec: Vec<u8> = Vec::new();
 //             er.write(&is, &mut vec).unwrap();
 //             assert_eq!(
 //                 "Parse error at 1:7: hello world!\n#version 100\n      ^\n",
 //                 String::from_utf8_lossy(&vec)
 //             );
 //         }
 //         assert!(is.next() == Some(b'o'));
 //         is.consume();
 //         assert!(is.next() == Some(b'n'));
 //         is.consume();
 //         assert!(is.input.len() == 13);
 //         assert!(is.line == 1);
 //         assert!(is.column == 9);
 //         assert!(is.offset == 8);
 //         assert!(is.next() == Some(b' '));
 //         is.consume();
 //         assert!(is.next() == Some(b'1'));
 //         is.consume();
 //         assert!(is.next() == Some(b'0'));
 //         is.consume();
 //         assert!(is.next() == Some(b'0'));
 //         is.consume();
 //         assert!(is.input.len() == 13);
 //         assert!(is.line == 1);
 //         assert!(is.column == 13);
 //         assert!(is.offset == 12);
 //         assert!(is.next() == Some(b'\n'));
 //         is.consume();
 //         assert!(is.input.len() == 13);
 //         assert!(is.line == 2);
 //         assert!(is.column == 1);
 //         assert!(is.offset == 13);
 //         {
 //             let ps = is.pos();
 //             assert_eq!(b"", ps.context(&is));
 //         }
 //         assert!(is.next() == None);
 //         is.consume();
 //         assert!(is.next() == None);
 //     }
     #[test]
     fn test_split() {
         let mut is = InputSource::from_string("#version 100\n").unwrap();
         let backup = is.clone();
         assert!(is.next() == Some(b'#'));
         is.consume();
         assert!(is.next() == Some(b'v'));
         is.consume();
         assert!(is.next() == Some(b'e'));
         is.consume();
         is = backup;
         assert!(is.next() == Some(b'#'));
         is.consume();
         assert!(is.next() == Some(b'v'));
         is.consume();
         assert!(is.next() == Some(b'e'));
         is.consume();
+    }
+}
 //     #[test]
 //     fn test_split() {
 //         let mut is = InputSource::from_string("#version 100\n").unwrap();
 //         let backup = is.clone();
 //         assert!(is.next() == Some(b'#'));
 //         is.consume();
 //         assert!(is.next() == Some(b'v'));
 //         is.consume();
 //         assert!(is.next() == Some(b'e'));
 //         is.consume();
 //         is = backup;
 //         assert!(is.next() == Some(b'#'));
 //         is.consume();
 //         assert!(is.next() == Some(b'v'));
 //         is.consume();
 //         assert!(is.next() == Some(b'e'));
 //         is.consume();
 //     }
 // }

src/protocol/lexer.rs

➞

Show inline comments

@@ @@ -1467,313 +1467,313 @@ impl Lexer<'_> { @@
         self.consume_whitespace(false)?;
         let label = self.consume_identifier(h)?;
         self.consume_whitespace(false)?;
         self.consume_string(b";")?;
         Ok(h.alloc_goto_statement(|this| GotoStatement { this, position, label, target: None }))
+    }
     fn consume_new_statement(&mut self, h: &mut Heap) -> Result<NewStatementId, ParseError> {
         let position = self.source.pos();
         self.consume_keyword(b"new")?;
         self.consume_whitespace(false)?;
         let expression = self.consume_call_expression(h)?;
         self.consume_whitespace(false)?;
         self.consume_string(b";")?;
         Ok(h.alloc_new_statement(|this| NewStatement { this, position, expression, next: None }))
+    }
     fn consume_put_statement(&mut self, h: &mut Heap) -> Result<PutStatementId, ParseError> {
         let position = self.source.pos();
         self.consume_keyword(b"put")?;
         self.consume_whitespace(false)?;
         self.consume_string(b"(")?;
         let port = self.consume_expression(h)?;
         self.consume_whitespace(false)?;
         self.consume_string(b",")?;
         self.consume_whitespace(false)?;
         let message = self.consume_expression(h)?;
         self.consume_whitespace(false)?;
         self.consume_string(b")")?;
         self.consume_whitespace(false)?;
         self.consume_string(b";")?;
         Ok(h.alloc_put_statement(|this| PutStatement { this, position, port, message, next: None }))
+    }
     fn consume_expression_statement(
         &mut self,
         h: &mut Heap,
     ) -> Result<ExpressionStatementId, ParseError> {
         let position = self.source.pos();
         let expression = self.consume_expression(h)?;
         self.consume_whitespace(false)?;
         self.consume_string(b";")?;
         Ok(h.alloc_expression_statement(|this| ExpressionStatement {
             this,
             position,
             expression,
             next: None,
         }))
+    }
     // ====================
     // Symbol definitions
     // ====================
     fn has_symbol_definition(&self) -> bool {
         self.has_keyword(b"composite")
             || self.has_keyword(b"primitive")
             || self.has_type_keyword()
             || self.has_identifier()
+    }
     fn consume_symbol_definition(&mut self, h: &mut Heap) -> Result<DefinitionId, ParseError> {
         if self.has_keyword(b"composite") || self.has_keyword(b"primitive") {
             Ok(self.consume_component_definition(h)?.upcast())
         } else {
             Ok(self.consume_function_definition(h)?.upcast())
+        }
+    }
     fn consume_component_definition(&mut self, h: &mut Heap) -> Result<ComponentId, ParseError> {
         if self.has_keyword(b"composite") {
             Ok(self.consume_composite_definition(h)?.upcast())
         } else {
             Ok(self.consume_primitive_definition(h)?.upcast())
+        }
+    }
     fn consume_composite_definition(&mut self, h: &mut Heap) -> Result<CompositeId, ParseError> {
         let position = self.source.pos();
         self.consume_keyword(b"composite")?;
         self.consume_whitespace(true)?;
         let identifier = self.consume_identifier(h)?;
         self.consume_whitespace(false)?;
         let mut parameters = Vec::new();
         self.consume_parameters(h, &mut parameters)?;
         self.consume_whitespace(false)?;
         let body = self.consume_block_statement(h)?;
         Ok(h.alloc_composite(|this| Composite { this, position, identifier, parameters, body }))
+    }
     fn consume_primitive_definition(&mut self, h: &mut Heap) -> Result<PrimitiveId, ParseError> {
         let position = self.source.pos();
         self.consume_keyword(b"primitive")?;
         self.consume_whitespace(true)?;
         let identifier = self.consume_identifier(h)?;
         self.consume_whitespace(false)?;
         let mut parameters = Vec::new();
         self.consume_parameters(h, &mut parameters)?;
         self.consume_whitespace(false)?;
         let body = self.consume_block_statement(h)?;
         Ok(h.alloc_primitive(|this| Primitive { this, position, identifier, parameters, body }))
+    }
     fn consume_function_definition(&mut self, h: &mut Heap) -> Result<FunctionId, ParseError> {
         let position = self.source.pos();
         let return_type = self.consume_type_annotation(h)?;
         self.consume_whitespace(true)?;
         let identifier = self.consume_identifier(h)?;
         self.consume_whitespace(false)?;
         let mut parameters = Vec::new();
         self.consume_parameters(h, &mut parameters)?;
         self.consume_whitespace(false)?;
         let body = self.consume_block_statement(h)?;
         Ok(h.alloc_function(|this| Function {
             this,
             position,
             return_type,
             identifier,
             parameters,
             body,
         }))
+    }
     fn has_pragma(&self) -> bool {
         if let Some(c) = self.source.next() {
             c == b'#'
         } else {
             false
+        }
+    }
     fn consume_pragma(&mut self, h: &mut Heap) -> Result<PragmaId, ParseError> {
         let position = self.source.pos();
         let next = self.source.next();
         if next != Some(b'#') {
             return Err(self.source.error("Expected pragma"));
+        }
         self.source.consume();
         if !is_vchar(self.source.next()) {
             return Err(self.source.error("Expected pragma"));
+        }
         let value = self.consume_line()?;
         Ok(h.alloc_pragma(|this| Pragma { this, position, value }))
+    }
     fn has_import(&self) -> bool {
         self.has_keyword(b"import")
+    }
     fn consume_import(&mut self, h: &mut Heap) -> Result<ImportId, ParseError> {
         let position = self.source.pos();
         self.consume_keyword(b"import")?;
         self.consume_whitespace(true)?;
         let mut value = Vec::new();
         let mut ident = self.consume_ident()?;
         value.append(&mut ident);
         while self.has_string(b".") {
             self.consume_string(b".")?;
             value.push(b'.');
             ident = self.consume_ident()?;
             value.append(&mut ident);
+        }
         self.consume_whitespace(false)?;
         self.consume_string(b";")?;
         Ok(h.alloc_import(|this| Import { this, position, value }))
+    }
     pub fn consume_protocol_description(&mut self, h: &mut Heap) -> Result<RootId, ParseError> {
         let position = self.source.pos();
         let mut pragmas = Vec::new();
         let mut imports = Vec::new();
         let mut definitions = Vec::new();
         self.consume_whitespace(false)?;
         while self.has_pragma() {
             let pragma = self.consume_pragma(h)?;
             pragmas.push(pragma);
             self.consume_whitespace(false)?;
+        }
         while self.has_import() {
             let import = self.consume_import(h)?;
             imports.push(import);
             self.consume_whitespace(false)?;
+        }
         // do-while block
         while {
             let def = self.consume_symbol_definition(h)?;
             definitions.push(def);
             self.consume_whitespace(false)?;
             self.has_symbol_definition()
         } {}
         // end of file
         if !self.source.is_eof() {
             return Err(self.source.error("Expected end of file"));
+        }
         Ok(h.alloc_protocol_description(|this| Root {
             this,
             position,
             pragmas,
             imports,
             definitions,
             declarations: Vec::new(),
         }))
+    }
+}
 #[cfg(test)]
 mod tests {
     use crate::protocol::ast::Expression::*;
     use crate::protocol::{ast, lexer::*};
 // #[cfg(test)]
 // mod tests {
 //     use crate::protocol::ast::Expression::*;
 //     use crate::protocol::{ast, lexer::*};
     #[test]
     fn test_lowercase() {
         assert_eq!(lowercase(b'a'), b'a');
         assert_eq!(lowercase(b'A'), b'a');
         assert_eq!(lowercase(b'z'), b'z');
         assert_eq!(lowercase(b'Z'), b'z');
+    }
 //     #[test]
 //     fn test_lowercase() {
 //         assert_eq!(lowercase(b'a'), b'a');
 //         assert_eq!(lowercase(b'A'), b'a');
 //         assert_eq!(lowercase(b'z'), b'z');
 //         assert_eq!(lowercase(b'Z'), b'z');
 //     }
     #[test]
     fn test_basic_expression() {
         let mut h = Heap::new();
         let mut is = InputSource::from_string("a+b;").unwrap();
         let mut lex = Lexer::new(&mut is);
         match lex.consume_expression(&mut h) {
             Ok(expr) => {
                 println!("{:?}", expr);
                 if let Binary(bin) = &h[expr] {
                     if let Variable(left) = &h[bin.left] {
                         if let Variable(right) = &h[bin.right] {
                             assert_eq!("a", format!("{}", h[left.identifier]));
                             assert_eq!("b", format!("{}", h[right.identifier]));
                             assert_eq!(Some(b';'), is.next());
                             return;
+                        }
+                    }
+                }
                 assert!(false);
+            }
             Err(err) => {
                 err.print(&is);
                 assert!(false);
+            }
+        }
+    }
 //     #[test]
 //     fn test_basic_expression() {
 //         let mut h = Heap::new();
 //         let mut is = InputSource::from_string("a+b;").unwrap();
 //         let mut lex = Lexer::new(&mut is);
 //         match lex.consume_expression(&mut h) {
 //             Ok(expr) => {
 //                 println!("{:?}", expr);
 //                 if let Binary(bin) = &h[expr] {
 //                     if let Variable(left) = &h[bin.left] {
 //                         if let Variable(right) = &h[bin.right] {
 //                             assert_eq!("a", format!("{}", h[left.identifier]));
 //                             assert_eq!("b", format!("{}", h[right.identifier]));
 //                             assert_eq!(Some(b';'), is.next());
 //                             return;
 //                         }
 //                     }
 //                 }
 //                 assert!(false);
 //             }
 //             Err(err) => {
 //                 err.print(&is);
 //                 assert!(false);
 //             }
 //         }
 //     }
     #[test]
     fn test_paren_expression() {
         let mut h = Heap::new();
         let mut is = InputSource::from_string("(true)").unwrap();
         let mut lex = Lexer::new(&mut is);
         match lex.consume_paren_expression(&mut h) {
             Ok(expr) => {
                 println!("{:#?}", expr);
                 if let Constant(con) = &h[expr] {
                     if let ast::Constant::True = con.value {
                         return;
+                    }
+                }
                 assert!(false);
+            }
             Err(err) => {
                 err.print(&is);
                 assert!(false);
+            }
+        }
+    }
 //     #[test]
 //     fn test_paren_expression() {
 //         let mut h = Heap::new();
 //         let mut is = InputSource::from_string("(true)").unwrap();
 //         let mut lex = Lexer::new(&mut is);
 //         match lex.consume_paren_expression(&mut h) {
 //             Ok(expr) => {
 //                 println!("{:#?}", expr);
 //                 if let Constant(con) = &h[expr] {
 //                     if let ast::Constant::True = con.value {
 //                         return;
 //                     }
 //                 }
 //                 assert!(false);
 //             }
 //             Err(err) => {
 //                 err.print(&is);
 //                 assert!(false);
 //             }
 //         }
 //     }
     #[test]
     fn test_expression() {
         let mut h = Heap::new();
         let mut is = InputSource::from_string("(x(1+5,get(y))-w[5])+z++\n").unwrap();
         let mut lex = Lexer::new(&mut is);
         match lex.consume_expression(&mut h) {
             Ok(expr) => {
                 println!("{:#?}", expr);
+            }
             Err(err) => {
                 err.print(&is);
                 assert!(false);
+            }
+        }
+    }
 //     #[test]
 //     fn test_expression() {
 //         let mut h = Heap::new();
 //         let mut is = InputSource::from_string("(x(1+5,get(y))-w[5])+z++\n").unwrap();
 //         let mut lex = Lexer::new(&mut is);
 //         match lex.consume_expression(&mut h) {
 //             Ok(expr) => {
 //                 println!("{:#?}", expr);
 //             }
 //             Err(err) => {
 //                 err.print(&is);
 //                 assert!(false);
 //             }
 //         }
 //     }
     #[test]
     fn test_basic_statement() {
         let mut h = Heap::new();
         let mut is = InputSource::from_string("while (true) { skip; }").unwrap();
         let mut lex = Lexer::new(&mut is);
         match lex.consume_statement(&mut h) {
             Ok(stmt) => {
                 println!("{:#?}", stmt);
                 if let Statement::While(w) = &h[stmt] {
                     if let Expression::Constant(_) = h[w.test] {
                         if let Statement::Block(_) = h[w.body] {
                             return;
+                        }
+                    }
+                }
                 assert!(false);
+            }
             Err(err) => {
                 err.print(&is);
                 assert!(false);
+            }
+        }
+    }
 //     #[test]
 //     fn test_basic_statement() {
 //         let mut h = Heap::new();
 //         let mut is = InputSource::from_string("while (true) { skip; }").unwrap();
 //         let mut lex = Lexer::new(&mut is);
 //         match lex.consume_statement(&mut h) {
 //             Ok(stmt) => {
 //                 println!("{:#?}", stmt);
 //                 if let Statement::While(w) = &h[stmt] {
 //                     if let Expression::Constant(_) = h[w.test] {
 //                         if let Statement::Block(_) = h[w.body] {
 //                             return;
 //                         }
 //                     }
 //                 }
 //                 assert!(false);
 //             }
 //             Err(err) => {
 //                 err.print(&is);
 //                 assert!(false);
 //             }
 //         }
 //     }
     #[test]
     fn test_statement() {
         let mut h = Heap::new();
         let mut is = InputSource::from_string(
             "label: while (true) { if (x++ > y[0]) break label; else continue; }\n",
+        )
         .unwrap();
         let mut lex = Lexer::new(&mut is);
         match lex.consume_statement(&mut h) {
             Ok(stmt) => {
                 println!("{:#?}", stmt);
+            }
             Err(err) => {
                 err.print(&is);
                 assert!(false);
+            }
+        }
+    }
+}
 //     #[test]
 //     fn test_statement() {
 //         let mut h = Heap::new();
 //         let mut is = InputSource::from_string(
 //             "label: while (true) { if (x++ > y[0]) break label; else continue; }\n",
 //         )
 //         .unwrap();
 //         let mut lex = Lexer::new(&mut is);
 //         match lex.consume_statement(&mut h) {
 //             Ok(stmt) => {
 //                 println!("{:#?}", stmt);
 //             }
 //             Err(err) => {
 //                 err.print(&is);
 //                 assert!(false);
 //             }
 //         }
 //     }
 // }

src/protocol/mod.rs

➞

Show inline comments

 mod arena;
 mod ast;
 mod eval;
 pub mod inputsource;
 mod lexer;
 mod library;
 mod parser;
 use crate::common::*;
 use crate::protocol::ast::*;
 use crate::protocol::eval::*;
 use crate::protocol::inputsource::*;
 use crate::protocol::parser::*;
 #[derive(serde::Serialize, serde::Deserialize)]
-pub struct ProtocolDescriptionImpl {
+pub struct ProtocolDescription {
     heap: Heap,
     source: InputSource,
     root: RootId,
+}
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 pub struct ComponentState {
     prompt: Prompt,
+}
 pub enum EvalContext<'a> {
     Nonsync(&'a mut NonsyncContext<'a>),
     Sync(&'a mut SyncContext<'a>),
     None,
+}
 //////////////////////////////////////////////
-impl std::fmt::Debug for ProtocolDescriptionImpl {
+impl std::fmt::Debug for ProtocolDescription {
     fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
         write!(f, "Protocol")
+    }
+}
 impl ProtocolDescription for ProtocolDescriptionImpl {
     type S = ComponentStateImpl;
     fn parse(buffer: &[u8]) -> Result<Self, String> {
 impl ProtocolDescription {
     pub fn parse(buffer: &[u8]) -> Result<Self, String> {
         let mut heap = Heap::new();
         let mut source = InputSource::from_buffer(buffer).unwrap();
         let mut parser = Parser::new(&mut source);
         match parser.parse(&mut heap) {
             Ok(root) => {
-                return Ok(ProtocolDescriptionImpl { heap, source, root });
+                return Ok(ProtocolDescription { heap, source, root });
+            }
             Err(err) => {
                 let mut vec: Vec<u8> = Vec::new();
                 err.write(&source, &mut vec).unwrap();
                 Err(String::from_utf8_lossy(&vec).to_string())
+            }
+        }
+    }
     fn component_polarities(&self, identifier: &[u8]) -> Result<Vec<Polarity>, MainComponentErr> {
     pub fn component_polarities(
         &self,
         identifier: &[u8],
     ) -> Result<Vec<Polarity>, AddComponentError> {
         use AddComponentError::*;
         let h = &self.heap;
         let root = &h[self.root];
         let def = root.get_definition_ident(h, identifier);
         if def.is_none() {
-            return Err(MainComponentErr::NoSuchComponent);
             return Err(NoSuchComponent);
+        }
         let def = &h[def.unwrap()];
         if !def.is_component() {
-            return Err(MainComponentErr::NoSuchComponent);
             return Err(NoSuchComponent);
+        }
         for &param in def.parameters().iter() {
             let param = &h[param];
             let type_annot = &h[param.type_annotation];
             if type_annot.the_type.array {
-                return Err(MainComponentErr::NonPortTypeParameters);
                 return Err(NonPortTypeParameters);
+            }
             match type_annot.the_type.primitive {
                 PrimitiveType::Input | PrimitiveType::Output => continue,
                 _ => {
-                    return Err(MainComponentErr::NonPortTypeParameters);
                     return Err(NonPortTypeParameters);
+                }
+            }
+        }
         let mut result = Vec::new();
         for &param in def.parameters().iter() {
             let param = &h[param];
             let type_annot = &h[param.type_annotation];
             let ptype = &type_annot.the_type.primitive;
             if ptype == &PrimitiveType::Input {
                 result.push(Polarity::Getter)
             } else if ptype == &PrimitiveType::Output {
                 result.push(Polarity::Putter)
             } else {
                 unreachable!()
+            }
+        }
         Ok(result)
+    }
     // expects port polarities to be correct
-    fn new_main_component(&self, identifier: &[u8], ports: &[Port]) -> ComponentStateImpl {
+    pub fn new_main_component(&self, identifier: &[u8], ports: &[PortId]) -> ComponentState {
         let mut args = Vec::new();
         for (&x, y) in ports.iter().zip(self.component_polarities(identifier).unwrap()) {
             match y {
                 Polarity::Getter => args.push(Value::Input(InputValue(x))),
                 Polarity::Putter => args.push(Value::Output(OutputValue(x))),
+            }
+        }
         let h = &self.heap;
         let root = &h[self.root];
         let def = root.get_definition_ident(h, identifier).unwrap();
-        ComponentStateImpl { prompt: Prompt::new(h, def, &args) }
+        ComponentState { prompt: Prompt::new(h, def, &args) }
+    }
+}
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 pub struct ComponentStateImpl {
     prompt: Prompt,
+}
 impl ComponentState for ComponentStateImpl {
     type D = ProtocolDescriptionImpl;
     fn pre_sync_run<C: MonoContext<D = ProtocolDescriptionImpl, S = Self>>(
         &mut self,
         context: &mut C,
         pd: &ProtocolDescriptionImpl,
     ) -> MonoBlocker {
         let mut context = EvalContext::Mono(context);
 impl ComponentState {
     pub fn nonsync_run<'a: 'b, 'b>(
         &'a mut self,
         context: &'b mut NonsyncContext<'b>,
         pd: &'a ProtocolDescription,
     ) -> NonsyncBlocker {
         let mut context = EvalContext::Nonsync(context);
         loop {
             let result = self.prompt.step(&pd.heap, &mut context);
             match result {
                 // In component definitions, there are no return statements
                 Ok(_) => unreachable!(),
                 Err(cont) => match cont {
                     EvalContinuation::Stepping => continue,
                     EvalContinuation::Inconsistent => return MonoBlocker::Inconsistent,
                     EvalContinuation::Terminal => return MonoBlocker::ComponentExit,
                     EvalContinuation::SyncBlockStart => return MonoBlocker::SyncBlockStart,
                     EvalContinuation::Inconsistent => return NonsyncBlocker::Inconsistent,
                     EvalContinuation::Terminal => return NonsyncBlocker::ComponentExit,
                     EvalContinuation::SyncBlockStart => return NonsyncBlocker::SyncBlockStart,
                     // Not possible to end sync block if never entered one
                     EvalContinuation::SyncBlockEnd => unreachable!(),
                     EvalContinuation::NewComponent(decl, args) => {
                         // Look up definition (TODO for now, assume it is a definition)
                         let h = &pd.heap;
                         let def = h[decl].as_defined().definition;
-                        let init_state = ComponentStateImpl { prompt: Prompt::new(h, def, &args) };
+                        let init_state = ComponentState { prompt: Prompt::new(h, def, &args) };
                         context.new_component(&args, init_state);
                         // Continue stepping
                         continue;
+                    }
                     // Outside synchronous blocks, no fires/get/put happens
                     EvalContinuation::BlockFires(_) => unreachable!(),
                     EvalContinuation::BlockGet(_) => unreachable!(),
                     EvalContinuation::Put(_, _) => unreachable!(),
                 },
+            }
+        }
+    }
     fn sync_run<C: PolyContext<D = ProtocolDescriptionImpl>>(
         &mut self,
         context: &mut C,
         pd: &ProtocolDescriptionImpl,
     ) -> PolyBlocker {
         let mut context = EvalContext::Poly(context);
     pub fn sync_run<'a: 'b, 'b>(
         &'a mut self,
         context: &'b mut SyncContext<'b>,
         pd: &'a ProtocolDescription,
     ) -> SyncBlocker {
         let mut context = EvalContext::Sync(context);
         loop {
             let result = self.prompt.step(&pd.heap, &mut context);
             match result {
                 // Inside synchronous blocks, there are no return statements
                 Ok(_) => unreachable!(),
                 Err(cont) => match cont {
                     EvalContinuation::Stepping => continue,
-                    EvalContinuation::Inconsistent => return PolyBlocker::Inconsistent,
+                    EvalContinuation::Inconsistent => return SyncBlocker::Inconsistent,
                     // First need to exit synchronous block before definition may end
                     EvalContinuation::Terminal => unreachable!(),
                     // No nested synchronous blocks
                     EvalContinuation::SyncBlockStart => unreachable!(),
-                    EvalContinuation::SyncBlockEnd => return PolyBlocker::SyncBlockEnd,
+                    EvalContinuation::SyncBlockEnd => return SyncBlocker::SyncBlockEnd,
                     // Not possible to create component in sync block
                     EvalContinuation::NewComponent(_, _) => unreachable!(),
                     EvalContinuation::BlockFires(port) => match port {
                         Value::Output(OutputValue(port)) => {
-                            return PolyBlocker::CouldntCheckFiring(port);
+                            return SyncBlocker::CouldntCheckFiring(port);
+                        }
                         Value::Input(InputValue(port)) => {
-                            return PolyBlocker::CouldntCheckFiring(port);
+                            return SyncBlocker::CouldntCheckFiring(port);
+                        }
                         _ => unreachable!(),
                     },
                     EvalContinuation::BlockGet(port) => match port {
                         Value::Output(OutputValue(port)) => {
-                            return PolyBlocker::CouldntReadMsg(port);
+                            return SyncBlocker::CouldntReadMsg(port);
+                        }
                         Value::Input(InputValue(port)) => {
-                            return PolyBlocker::CouldntReadMsg(port);
+                            return SyncBlocker::CouldntReadMsg(port);
+                        }
                         _ => unreachable!(),
                     },
                     EvalContinuation::Put(port, message) => {
                         let value;
                         match port {
                             Value::Output(OutputValue(port_value)) => {
                                 value = port_value;
+                            }
                             Value::Input(InputValue(port_value)) => {
                                 value = port_value;
+                            }
                             _ => unreachable!(),
+                        }
                         let payload;
                         match message {
                             Value::Message(MessageValue(None)) => {
                                 // Putting a null message is inconsistent
-                                return PolyBlocker::Inconsistent;
+                                return SyncBlocker::Inconsistent;
+                            }
                             Value::Message(MessageValue(Some(buffer))) => {
                                 // Create a copy of the payload
                                 payload = buffer.clone();
+                            }
                             _ => unreachable!(),
+                        }
-                        return PolyBlocker::PutMsg(value, payload);
+                        return SyncBlocker::PutMsg(value, payload);
+                    }
                 },
+            }
+        }
+    }
+}
 pub enum EvalContext<'a> {
     Mono(&'a mut dyn MonoContext<D = ProtocolDescriptionImpl, S = ComponentStateImpl>),
     Poly(&'a mut dyn PolyContext<D = ProtocolDescriptionImpl>),
     None,
+}
 impl EvalContext<'_> {
     // fn random(&mut self) -> LongValue {
     //     match self {
     //         EvalContext::None => unreachable!(),
     //         EvalContext::Mono(_context) => todo!(),
     //         EvalContext::Poly(_) => unreachable!(),
     //         EvalContext::Nonsync(_context) => todo!(),
     //         EvalContext::Sync(_) => unreachable!(),
     //     }
     // }
-    fn new_component(&mut self, args: &[Value], init_state: ComponentStateImpl) -> () {
+    fn new_component(&mut self, args: &[Value], init_state: ComponentState) -> () {
         match self {
             EvalContext::None => unreachable!(),
-            EvalContext::Mono(context) => {
+            EvalContext::Nonsync(context) => {
                 let mut moved_ports = HashSet::new();
                 for arg in args.iter() {
                     match arg {
                         Value::Output(OutputValue(port)) => {
                             moved_ports.insert(*port);
+                        }
                         Value::Input(InputValue(port)) => {
                             moved_ports.insert(*port);
+                        }
                         _ => {}
+                    }
+                }
                 context.new_component(moved_ports, init_state)
+            }
-            EvalContext::Poly(_) => unreachable!(),
+            EvalContext::Sync(_) => unreachable!(),
+        }
+    }
     fn new_channel(&mut self) -> [Value; 2] {
         match self {
             EvalContext::None => unreachable!(),
-            EvalContext::Mono(context) => {
+            EvalContext::Nonsync(context) => {
                 let [from, to] = context.new_channel();
                 let from = Value::Output(OutputValue(from));
                 let to = Value::Input(InputValue(to));
                 return [from, to];
+            }
-            EvalContext::Poly(_) => unreachable!(),
+            EvalContext::Sync(_) => unreachable!(),
+        }
+    }
     fn fires(&mut self, port: Value) -> Option<Value> {
         match self {
             EvalContext::None => unreachable!(),
             EvalContext::Mono(_) => unreachable!(),
             EvalContext::Poly(context) => match port {
             EvalContext::Nonsync(_) => unreachable!(),
             EvalContext::Sync(context) => match port {
                 Value::Output(OutputValue(port)) => context.is_firing(port).map(Value::from),
                 Value::Input(InputValue(port)) => context.is_firing(port).map(Value::from),
                 _ => unreachable!(),
             },
+        }
+    }
     fn get(&mut self, port: Value) -> Option<Value> {
         match self {
             EvalContext::None => unreachable!(),
             EvalContext::Mono(_) => unreachable!(),
             EvalContext::Poly(context) => match port {
             EvalContext::Nonsync(_) => unreachable!(),
             EvalContext::Sync(context) => match port {
                 Value::Output(OutputValue(port)) => {
                     context.read_msg(port).map(Value::receive_message)
+                }
                 Value::Input(InputValue(port)) => {
                     context.read_msg(port).map(Value::receive_message)
+                }
                 _ => unreachable!(),
             },
+        }
+    }
+}

src/protocol/parser.rs

➞

Show inline comments

@@ @@ -1386,498 +1386,500 @@ impl Visitor for ResolveLabels { @@
         if let Some(label) = h[stmt].label {
             let target = self.get_target(h, label)?;
             let target = &h[h[target].body];
             if !target.is_while() {
                 return Err(ParseError::new(
                     h[stmt].position,
                     "Illegal continue: target not a while statement",
                 ));
+            }
             the_while = target.as_while();
         // TODO: check if continue is nested under while
         } else {
             if self.while_enclosure.is_none() {
                 return Err(ParseError::new(
                     h[stmt].position,
                     "Illegal continue: no surrounding while statement",
                 ));
+            }
             the_while = &h[self.while_enclosure.unwrap()];
             // continue is always nested under while, by recursive vistor
+        }
         if the_while.in_sync != self.sync_enclosure {
             return Err(ParseError::new(
                 h[stmt].position,
                 "Illegal continue: synchronous statement escape",
             ));
+        }
         h[stmt].target = Some(the_while.this);
         Ok(())
+    }
     fn visit_synchronous_statement(
         &mut self,
         h: &mut Heap,
         stmt: SynchronousStatementId,
     ) -> VisitorResult {
         assert!(self.sync_enclosure.is_none());
         self.sync_enclosure = Some(stmt);
         recursive_synchronous_statement(self, h, stmt)?;
         self.sync_enclosure = None;
         Ok(())
+    }
     fn visit_goto_statement(&mut self, h: &mut Heap, stmt: GotoStatementId) -> VisitorResult {
         let target = self.get_target(h, h[stmt].label)?;
         if h[target].in_sync != self.sync_enclosure {
             return Err(ParseError::new(
                 h[stmt].position,
                 "Illegal goto: synchronous statement escape",
             ));
+        }
         h[stmt].target = Some(target);
         Ok(())
+    }
     fn visit_expression(&mut self, h: &mut Heap, expr: ExpressionId) -> VisitorResult {
         Ok(())
+    }
+}
 struct AssignableExpressions {
     assignable: bool,
+}
 impl AssignableExpressions {
     fn new() -> Self {
         AssignableExpressions { assignable: false }
+    }
     fn error(&self, position: InputPosition) -> VisitorResult {
         Err(ParseError::new(position, "Unassignable expression"))
+    }
+}
 impl Visitor for AssignableExpressions {
     fn visit_assignment_expression(
         &mut self,
         h: &mut Heap,
         expr: AssignmentExpressionId,
     ) -> VisitorResult {
         if self.assignable {
             self.error(h[expr].position)
         } else {
             self.assignable = true;
             self.visit_expression(h, h[expr].left)?;
             self.assignable = false;
             self.visit_expression(h, h[expr].right)
+        }
+    }
     fn visit_conditional_expression(
         &mut self,
         h: &mut Heap,
         expr: ConditionalExpressionId,
     ) -> VisitorResult {
         if self.assignable {
             self.error(h[expr].position)
         } else {
             recursive_conditional_expression(self, h, expr)
+        }
+    }
     fn visit_binary_expression(&mut self, h: &mut Heap, expr: BinaryExpressionId) -> VisitorResult {
         if self.assignable {
             self.error(h[expr].position)
         } else {
             recursive_binary_expression(self, h, expr)
+        }
+    }
     fn visit_unary_expression(&mut self, h: &mut Heap, expr: UnaryExpressionId) -> VisitorResult {
         if self.assignable {
             self.error(h[expr].position)
         } else {
             match h[expr].operation {
                 UnaryOperation::PostDecrement
                 | UnaryOperation::PreDecrement
                 | UnaryOperation::PostIncrement
                 | UnaryOperation::PreIncrement => {
                     self.assignable = true;
                     recursive_unary_expression(self, h, expr)?;
                     self.assignable = false;
                     Ok(())
+                }
                 _ => recursive_unary_expression(self, h, expr),
+            }
+        }
+    }
     fn visit_indexing_expression(
         &mut self,
         h: &mut Heap,
         expr: IndexingExpressionId,
     ) -> VisitorResult {
         let old = self.assignable;
         self.assignable = false;
         recursive_indexing_expression(self, h, expr)?;
         self.assignable = old;
         Ok(())
+    }
     fn visit_slicing_expression(
         &mut self,
         h: &mut Heap,
         expr: SlicingExpressionId,
     ) -> VisitorResult {
         let old = self.assignable;
         self.assignable = false;
         recursive_slicing_expression(self, h, expr)?;
         self.assignable = old;
         Ok(())
+    }
     fn visit_select_expression(&mut self, h: &mut Heap, expr: SelectExpressionId) -> VisitorResult {
         if h[expr].field.is_length() && self.assignable {
             return self.error(h[expr].position);
+        }
         let old = self.assignable;
         self.assignable = false;
         recursive_select_expression(self, h, expr)?;
         self.assignable = old;
         Ok(())
+    }
     fn visit_array_expression(&mut self, h: &mut Heap, expr: ArrayExpressionId) -> VisitorResult {
         if self.assignable {
             self.error(h[expr].position)
         } else {
             recursive_array_expression(self, h, expr)
+        }
+    }
     fn visit_call_expression(&mut self, h: &mut Heap, expr: CallExpressionId) -> VisitorResult {
         if self.assignable {
             self.error(h[expr].position)
         } else {
             recursive_call_expression(self, h, expr)
+        }
+    }
     fn visit_constant_expression(
         &mut self,
         h: &mut Heap,
         expr: ConstantExpressionId,
     ) -> VisitorResult {
         if self.assignable {
             self.error(h[expr].position)
         } else {
             Ok(())
+        }
+    }
     fn visit_variable_expression(
         &mut self,
         h: &mut Heap,
         expr: VariableExpressionId,
     ) -> VisitorResult {
         Ok(())
+    }
+}
 struct IndexableExpressions {
     indexable: bool,
+}
 impl IndexableExpressions {
     fn new() -> Self {  IndexableExpressions { indexable: false }  }
     fn new() -> Self {
         IndexableExpressions { indexable: false }
+    }
     fn error(&self, position: InputPosition) -> VisitorResult {
         Err(ParseError::new(position, "Unindexable expression"))
+    }
+}
 impl Visitor for IndexableExpressions {
     fn visit_assignment_expression(
         &mut self,
         h: &mut Heap,
         expr: AssignmentExpressionId,
     ) -> VisitorResult {
         if self.indexable {
             self.error(h[expr].position)
         } else {
             recursive_assignment_expression(self, h, expr)
+        }
+    }
     fn visit_conditional_expression(
         &mut self,
         h: &mut Heap,
         expr: ConditionalExpressionId,
     ) -> VisitorResult {
         let old = self.indexable;
         self.indexable = false;
         self.visit_expression(h, h[expr].test)?;
         self.indexable = old;
         self.visit_expression(h, h[expr].true_expression)?;
         self.visit_expression(h, h[expr].false_expression)
+    }
     fn visit_binary_expression(&mut self, h: &mut Heap, expr: BinaryExpressionId) -> VisitorResult {
         if self.indexable && h[expr].operation != BinaryOperator::Concatenate {
             self.error(h[expr].position)
         } else {
             recursive_binary_expression(self, h, expr)
+        }
+    }
     fn visit_unary_expression(&mut self, h: &mut Heap, expr: UnaryExpressionId) -> VisitorResult {
         if self.indexable {
             self.error(h[expr].position)
         } else {
             recursive_unary_expression(self, h, expr)
+        }
+    }
     fn visit_indexing_expression(
         &mut self,
         h: &mut Heap,
         expr: IndexingExpressionId,
     ) -> VisitorResult {
         let old = self.indexable;
         self.indexable = true;
         self.visit_expression(h, h[expr].subject)?;
         self.indexable = false;
         self.visit_expression(h, h[expr].index)?;
         self.indexable = old;
         Ok(())
+    }
     fn visit_slicing_expression(
         &mut self,
         h: &mut Heap,
         expr: SlicingExpressionId,
     ) -> VisitorResult {
         let old = self.indexable;
         self.indexable = true;
         self.visit_expression(h, h[expr].subject)?;
         self.indexable = false;
         self.visit_expression(h, h[expr].from_index)?;
         self.visit_expression(h, h[expr].to_index)?;
         self.indexable = old;
         Ok(())
+    }
     fn visit_select_expression(&mut self, h: &mut Heap, expr: SelectExpressionId) -> VisitorResult {
         let old = self.indexable;
         self.indexable = false;
         recursive_select_expression(self, h, expr)?;
         self.indexable = old;
         Ok(())
+    }
     fn visit_array_expression(&mut self, h: &mut Heap, expr: ArrayExpressionId) -> VisitorResult {
         let old = self.indexable;
         self.indexable = false;
         recursive_array_expression(self, h, expr)?;
         self.indexable = old;
         Ok(())
+    }
     fn visit_call_expression(&mut self, h: &mut Heap, expr: CallExpressionId) -> VisitorResult {
         let old = self.indexable;
         self.indexable = false;
         recursive_call_expression(self, h, expr)?;
         self.indexable = old;
         Ok(())
+    }
     fn visit_constant_expression(
         &mut self,
         h: &mut Heap,
         expr: ConstantExpressionId,
     ) -> VisitorResult {
         if self.indexable {
             self.error(h[expr].position)
         } else {
             Ok(())
+        }
+    }
+}
 struct SelectableExpressions {
     selectable: bool,
+}
 impl SelectableExpressions {
     fn new() -> Self {
         SelectableExpressions { selectable: false }
+    }
     fn error(&self, position: InputPosition) -> VisitorResult {
         Err(ParseError::new(position, "Unselectable expression"))
+    }
+}
 impl Visitor for SelectableExpressions {
     fn visit_assignment_expression(
         &mut self,
         h: &mut Heap,
         expr: AssignmentExpressionId,
     ) -> VisitorResult {
         // left-hand side of assignment can be skipped
         let old = self.selectable;
         self.selectable = false;
         self.visit_expression(h, h[expr].right)?;
         self.selectable = old;
         Ok(())
+    }
     fn visit_conditional_expression(
         &mut self,
         h: &mut Heap,
         expr: ConditionalExpressionId,
     ) -> VisitorResult {
         let old = self.selectable;
         self.selectable = false;
         self.visit_expression(h, h[expr].test)?;
         self.selectable = old;
         self.visit_expression(h, h[expr].true_expression)?;
         self.visit_expression(h, h[expr].false_expression)
+    }
     fn visit_binary_expression(&mut self, h: &mut Heap, expr: BinaryExpressionId) -> VisitorResult {
         if self.selectable && h[expr].operation != BinaryOperator::Concatenate {
             self.error(h[expr].position)
         } else {
             recursive_binary_expression(self, h, expr)
+        }
+    }
     fn visit_unary_expression(&mut self, h: &mut Heap, expr: UnaryExpressionId) -> VisitorResult {
         if self.selectable {
             self.error(h[expr].position)
         } else {
             recursive_unary_expression(self, h, expr)
+        }
+    }
     fn visit_indexing_expression(
         &mut self,
         h: &mut Heap,
         expr: IndexingExpressionId,
     ) -> VisitorResult {
         let old = self.selectable;
         self.selectable = false;
         recursive_indexing_expression(self, h, expr)?;
         self.selectable = old;
         Ok(())
+    }
     fn visit_slicing_expression(
         &mut self,
         h: &mut Heap,
         expr: SlicingExpressionId,
     ) -> VisitorResult {
         let old = self.selectable;
         self.selectable = false;
         recursive_slicing_expression(self, h, expr)?;
         self.selectable = old;
         Ok(())
+    }
     fn visit_select_expression(&mut self, h: &mut Heap, expr: SelectExpressionId) -> VisitorResult {
         let old = self.selectable;
         self.selectable = false;
         recursive_select_expression(self, h, expr)?;
         self.selectable = old;
         Ok(())
+    }
     fn visit_array_expression(&mut self, h: &mut Heap, expr: ArrayExpressionId) -> VisitorResult {
         let old = self.selectable;
         self.selectable = false;
         recursive_array_expression(self, h, expr)?;
         self.selectable = old;
         Ok(())
+    }
     fn visit_call_expression(&mut self, h: &mut Heap, expr: CallExpressionId) -> VisitorResult {
         let old = self.selectable;
         self.selectable = false;
         recursive_call_expression(self, h, expr)?;
         self.selectable = old;
         Ok(())
+    }
     fn visit_constant_expression(
         &mut self,
         h: &mut Heap,
         expr: ConstantExpressionId,
     ) -> VisitorResult {
         if self.selectable {
             self.error(h[expr].position)
         } else {
             Ok(())
+        }
+    }
+}
 pub struct Parser<'a> {
     source: &'a mut InputSource,
+}
 impl<'a> Parser<'a> {
     pub fn new(source: &'a mut InputSource) -> Self {
         Parser { source }
+    }
     pub fn parse(&mut self, h: &mut Heap) -> Result<RootId, ParseError> {
         let mut lex = Lexer::new(self.source);
         let pd = lex.consume_protocol_description(h)?;
         NestedSynchronousStatements::new().visit_protocol_description(h, pd)?;
         ChannelStatementOccurrences::new().visit_protocol_description(h, pd)?;
         FunctionStatementReturns::new().visit_protocol_description(h, pd)?;
         ComponentStatementReturnNew::new().visit_protocol_description(h, pd)?;
         CheckBuiltinOccurrences::new().visit_protocol_description(h, pd)?;
         BuildSymbolDeclarations::new().visit_protocol_description(h, pd)?;
         LinkCallExpressions::new().visit_protocol_description(h, pd)?;
         BuildScope::new().visit_protocol_description(h, pd)?;
         ResolveVariables::new().visit_protocol_description(h, pd)?;
         LinkStatements::new().visit_protocol_description(h, pd)?;
         BuildLabels::new().visit_protocol_description(h, pd)?;
         ResolveLabels::new().visit_protocol_description(h, pd)?;
         AssignableExpressions::new().visit_protocol_description(h, pd)?;
         IndexableExpressions::new().visit_protocol_description(h, pd)?;
         SelectableExpressions::new().visit_protocol_description(h, pd)?;
         Ok(pd)
+    }
+}
 #[cfg(test)]
 mod tests {
     extern crate test_generator;
     use std::fs::File;
     use std::io::Read;
     use std::path::Path;
     use test_generator::test_resources;
     use super::*;
     #[test_resources("testdata/parser/positive/*.pdl")]
     fn batch1(resource: &str) {
         let path = Path::new(resource);
         let mut heap = Heap::new();
         let mut source = InputSource::from_file(&path).unwrap();
         let mut parser = Parser::new(&mut source);
         match parser.parse(&mut heap) {
             Ok(_) => {}
             Err(err) => {
                 println!("{}", err.display(&source));
                 println!("{:?}", err);
                 assert!(false);
+            }
+        }
+    }
     #[test_resources("testdata/parser/negative/*.pdl")]
     fn batch2(resource: &str) {
         let path = Path::new(resource);
         let expect = path.with_extension("txt");
         let mut heap = Heap::new();
         let mut source = InputSource::from_file(&path).unwrap();
         let mut parser = Parser::new(&mut source);
         match parser.parse(&mut heap) {
             Ok(pd) => {
                 println!("{:?}", heap[pd]);
                 println!("Expected parse error:");
                 let mut cev: Vec<u8> = Vec::new();
                 let mut f = File::open(expect).unwrap();
                 f.read_to_end(&mut cev).unwrap();
                 println!("{}", String::from_utf8_lossy(&cev));
                 assert!(false);
+            }
             Err(err) => {
                 println!("{:?}", err);
                 let mut vec: Vec<u8> = Vec::new();
                 err.write(&source, &mut vec).unwrap();
                 println!("{}", String::from_utf8_lossy(&vec));
                 let mut cev: Vec<u8> = Vec::new();
                 let mut f = File::open(expect).unwrap();
                 f.read_to_end(&mut cev).unwrap();
                 println!("{}", String::from_utf8_lossy(&cev));
                 assert_eq!(vec, cev);
+            }
+        }
+    }
+}
 // #[cfg(test)]
 // mod tests {
 //     extern crate test_generator;
 //     use std::fs::File;
 //     use std::io::Read;
 //     use std::path::Path;
 //     use test_generator::test_resources;
 //     use super::*;
 //     #[test_resources("testdata/parser/positive/*.pdl")]
 //     fn batch1(resource: &str) {
 //         let path = Path::new(resource);
 //         let mut heap = Heap::new();
 //         let mut source = InputSource::from_file(&path).unwrap();
 //         let mut parser = Parser::new(&mut source);
 //         match parser.parse(&mut heap) {
 //             Ok(_) => {}
 //             Err(err) => {
 //                 println!("{}", err.display(&source));
 //                 println!("{:?}", err);
 //                 assert!(false);
 //             }
 //         }
 //     }
 //     #[test_resources("testdata/parser/negative/*.pdl")]
 //     fn batch2(resource: &str) {
 //         let path = Path::new(resource);
 //         let expect = path.with_extension("txt");
 //         let mut heap = Heap::new();
 //         let mut source = InputSource::from_file(&path).unwrap();
 //         let mut parser = Parser::new(&mut source);
 //         match parser.parse(&mut heap) {
 //             Ok(pd) => {
 //                 println!("{:?}", heap[pd]);
 //                 println!("Expected parse error:");
 //                 let mut cev: Vec<u8> = Vec::new();
 //                 let mut f = File::open(expect).unwrap();
 //                 f.read_to_end(&mut cev).unwrap();
 //                 println!("{}", String::from_utf8_lossy(&cev));
 //                 assert!(false);
 //             }
 //             Err(err) => {
 //                 println!("{:?}", err);
 //                 let mut vec: Vec<u8> = Vec::new();
 //                 err.write(&source, &mut vec).unwrap();
 //                 println!("{}", String::from_utf8_lossy(&vec));
 //                 let mut cev: Vec<u8> = Vec::new();
 //                 let mut f = File::open(expect).unwrap();
 //                 f.read_to_end(&mut cev).unwrap();
 //                 println!("{}", String::from_utf8_lossy(&cev));
 //                 assert_eq!(vec, cev);
 //             }
 //         }
 //     }
 // }

src/runtime/actors.rs

➞

Show inline comments

 use crate::common::*;
 use crate::runtime::{endpoint::*, *};
 #[derive(Debug, Clone)]
 pub(crate) struct MonoN {
     pub ports: HashSet<Port>,
     pub result: Option<(usize, HashMap<Port, Payload>)>,
     pub ports: HashSet<PortId>,
     pub result: Option<(usize, HashMap<PortId, Payload>)>,
+}
 #[derive(Debug)]
 pub(crate) struct PolyN {
-    pub ports: HashSet<Port>,
+    pub ports: HashSet<PortId>,
     pub branches: HashMap<Predicate, BranchN>,
+}
 #[derive(Debug, Clone)]
 pub(crate) struct BranchN {
     pub to_get: HashSet<Port>,
     pub gotten: HashMap<Port, Payload>,
     pub to_get: HashSet<PortId>,
     pub gotten: HashMap<PortId, Payload>,
     pub sync_batch_index: usize,
+}
 #[derive(Debug, Clone)]
 pub struct MonoP {
     pub state: ProtocolS,
-    pub ports: HashSet<Port>,
+    pub ports: HashSet<PortId>,
+}
 #[derive(Debug)]
 pub(crate) struct PolyP {
     pub incomplete: HashMap<Predicate, BranchP>,
     pub complete: HashMap<Predicate, BranchP>,
-    pub ports: HashSet<Port>,
+    pub ports: HashSet<PortId>,
+}
 #[derive(Debug, Clone)]
 pub(crate) struct BranchP {
     pub blocking_on: Option<Port>,
     pub outbox: HashMap<Port, Payload>,
     pub inbox: HashMap<Port, Payload>,
     pub blocking_on: Option<PortId>,
     pub outbox: HashMap<PortId, Payload>,
     pub inbox: HashMap<PortId, Payload>,
     pub state: ProtocolS,
+}
 //////////////////////////////////////////////////////////////////
 impl PolyP {
     pub(crate) fn poly_run(
         &mut self,
         m_ctx: PolyPContext,
         protocol_description: &ProtocolD,
     ) -> Result<SyncRunResult, EndpointErr> {
         let to_run: Vec<_> = self.incomplete.drain().collect();
         self.poly_run_these_branches(m_ctx, protocol_description, to_run)
+    }
     pub(crate) fn poly_run_these_branches(
         &mut self,
         mut m_ctx: PolyPContext,
         protocol_description: &ProtocolD,
         mut to_run: Vec<(Predicate, BranchP)>,
     ) -> Result<SyncRunResult, EndpointErr> {
         use SyncRunResult as Srr;
         log!(&mut m_ctx.inner.logger, "~ Running branches for PolyP {:?}!", m_ctx.my_subtree_id,);
         'to_run_loop: while let Some((mut predicate, mut branch)) = to_run.pop() {
             let mut r_ctx = BranchPContext {
                 m_ctx: m_ctx.reborrow(),
                 ports: &self.ports,
                 predicate: &predicate,
                 inbox: &branch.inbox,
             };
             use PolyBlocker as Sb;
             let blocker = branch.state.sync_run(&mut r_ctx, protocol_description);
             log!(
                 &mut r_ctx.m_ctx.inner.logger,
                 "~ ... ran PolyP {:?} with branch pred {:?} to blocker {:?}",
                 r_ctx.m_ctx.my_subtree_id,
                 &predicate,
                 &blocker
             );
             match blocker {
                 Sb::Inconsistent => {} // DROP
                 Sb::CouldntReadMsg(port) => {
                     assert!(self.ports.contains(&port));
                     let channel_id =
                         r_ctx.m_ctx.inner.endpoint_exts.get(port).unwrap().info.channel_id;
                     log!(
                         &mut r_ctx.m_ctx.inner.logger,
                         "~ ... {:?} couldnt read msg for port {:?}. has inbox {:?}",
                         r_ctx.m_ctx.my_subtree_id,
                         channel_id,
                         &branch.inbox,
                     );
                     if predicate.replace_assignment(channel_id, true) != Some(false) {
                         // don't rerun now. Rerun at next `sync_run`
                         log!(&mut m_ctx.inner.logger, "~ ... Delay {:?}", m_ctx.my_subtree_id,);
                         branch.blocking_on = Some(port);
                         self.incomplete.insert(predicate, branch);
                     } else {
                         log!(&mut m_ctx.inner.logger, "~ ... Drop {:?}", m_ctx.my_subtree_id,);
+                    }
                     // ELSE DROP
+                }
                 Sb::CouldntCheckFiring(port) => {
                     assert!(self.ports.contains(&port));
                     let channel_id =
                         r_ctx.m_ctx.inner.endpoint_exts.get(port).unwrap().info.channel_id;
                     // split the branch!
                     let branch_f = branch.clone();
                     let mut predicate_f = predicate.clone();
                     if predicate_f.replace_assignment(channel_id, false).is_some() {
                         panic!("OI HANS QUERY FIRST!");
+                    }
                     assert!(predicate.replace_assignment(channel_id, true).is_none());
                     to_run.push((predicate, branch));
                     to_run.push((predicate_f, branch_f));
+                }
                 Sb::SyncBlockEnd => {
                     let ControllerInner { logger, endpoint_exts, .. } = m_ctx.inner;
                     log!(
                         logger,
                         "~ ... ran {:?} reached SyncBlockEnd with pred {:?} ...",
                         m_ctx.my_subtree_id,
                         &predicate,
                     );
                     // come up with the predicate for this local solution
                     for port in self.ports.iter() {
                         let channel_id = endpoint_exts.get(*port).unwrap().info.channel_id;
                         let fired =
                             branch.inbox.contains_key(port) || branch.outbox.contains_key(port);
                         match predicate.query(channel_id) {
                             Some(true) => {
                                 if !fired {
                                     // This branch should have fired but didn't!
                                     log!(
                                         logger,
                                         "~ ... ... should have fired {:?} and didn't! pruning!",
                                         channel_id,
                                     );
                                     continue 'to_run_loop;
+                                }
+                            }
                             Some(false) => {
                                 if fired {
                                     println!(
                                         "pred {:#?} in {:#?} out {:#?}",
                                         &predicate,
                                         branch.inbox.get(port),
                                         branch.outbox.get(port)
                                     );
                                     panic!("channel_id {:?} fired (based on outbox/inbox) but the predicate had Some(false)!" ,channel_id)
+                                }
+                            }
                             None => {
                                 predicate.replace_assignment(channel_id, false);
                                 if fired {
                                     println!(
                                         "pred {:#?} in {:#?} out {:#?}",
                                         &predicate,
                                         branch.inbox.get(port),
                                         branch.outbox.get(port)
                                     );
                                     panic!("channel_id {:?} fired (based on outbox/inbox) but the predicate had None!" ,channel_id)
+                                }
+                            }
+                        }
+                    }
                     log!(logger, "~ ... ... and finished just fine!",);
                     m_ctx.solution_storage.submit_and_digest_subtree_solution(
                         &mut m_ctx.inner.logger,
                         m_ctx.my_subtree_id,
                         predicate.clone(),
                     );
                     self.complete.insert(predicate, branch);
+                }
                 Sb::PutMsg(port, payload) => {
                     assert!(self.ports.contains(&port));
                     let EndpointExt { info, endpoint } =
                         m_ctx.inner.endpoint_exts.get_mut(port).unwrap();
                     if predicate.replace_assignment(info.channel_id, true) != Some(false) {
                         branch.outbox.insert(port, payload.clone());
                         let msg = CommMsgContents::SendPayload {
                             payload_predicate: predicate.clone(),
                             payload,
+                        }
                         .into_msg(m_ctx.inner.round_index);
                         log!(
                             &mut m_ctx.inner.logger,
                             "~ ... ... PolyP sending msg {:?} to {:?} ({:?}) now!",
                             &msg,
                             port,
                             (info.channel_id.controller_id, info.channel_id.channel_index),
                         );
                         endpoint.send(msg)?;
                         to_run.push((predicate, branch));
+                    }
                     // ELSE DROP
+                }
+            }
+        }
         // all in self.incomplete most recently returned Blocker::CouldntReadMsg
         Ok(if self.incomplete.is_empty() {
             if self.complete.is_empty() {
                 Srr::NoBranches
             } else {
                 Srr::AllBranchesComplete
+            }
         } else {
             Srr::BlockingForRecv
         })
+    }
     pub(crate) fn poly_recv_run(
         &mut self,
         m_ctx: PolyPContext,
         protocol_description: &ProtocolD,
-        port: Port,
+        port: PortId,
         payload_predicate: Predicate,
         payload: Payload,
     ) -> Result<SyncRunResult, EndpointErr> {
         // try exact match
         let to_run = if self.complete.contains_key(&payload_predicate) {
             // exact match with stopped machine
             log!(
                 &mut m_ctx.inner.logger,
                 "... poly_recv_run matched stopped machine exactly! nothing to do here",
             );
             vec![]
         } else if let Some(mut branch) = self.incomplete.remove(&payload_predicate) {
             // exact match with running machine
             log!(
                 &mut m_ctx.inner.logger,
                 "... poly_recv_run matched running machine exactly! pred is {:?}",
                 &payload_predicate
             );
             branch.inbox.insert(port, payload);
             if branch.blocking_on == Some(port) {
                 branch.blocking_on = None;
                 vec![(payload_predicate, branch)]
             } else {
                 vec![]
+            }
         } else {
             log!(
                 &mut m_ctx.inner.logger,
                 "... poly_recv_run didn't have any exact matches... Let's try feed it to all branches",
             );
             let mut incomplete2 = HashMap::<_, _>::default();
             let to_run = self
                 .incomplete
                 .drain()
                 .filter_map(|(old_predicate, mut branch)| {
                     use CommonSatResult as Csr;
                     match old_predicate.common_satisfier(&payload_predicate) {
                         Csr::FormerNotLatter | Csr::Equivalent => {
                             log!(
                                 &mut m_ctx.inner.logger,
                                 "... poly_recv_run This branch is compatible unaltered! branch pred: {:?}",
                                 &old_predicate
                             );
                             // old_predicate COVERS the assumptions of payload_predicate
                             if let Some(prev_payload) = branch.inbox.get(&port) {
                                 // Incorrect to receive two distinct messages in same branch!
                                 assert_eq!(prev_payload, &payload);
+                            }
                             branch.inbox.insert(port, payload.clone());
                             if branch.blocking_on == Some(port) {
                                 // run.
                                 branch.blocking_on = None;
                                 Some((old_predicate, branch))
                             } else {
                                 // don't bother running. its awaiting something else
                                 incomplete2.insert(old_predicate, branch);
                                 None
+                            }
+                        }
                         Csr::New(new) => {
                             log!(
                                 &mut m_ctx.inner.logger,
                                 "... poly_recv_run payloadpred {:?} and branchpred {:?} satisfied by new pred {:?}. FORKING",
                                 &payload_predicate,
                                 &old_predicate,
                                 &new,
                             );
                             // payload_predicate has new assumptions. FORK!
                             let mut payload_branch = branch.clone();
                             if let Some(prev_payload) = payload_branch.inbox.get(&port) {
                                 // Incorrect to receive two distinct messages in same branch!
                                 assert_eq!(prev_payload, &payload);
+                            }
                             payload_branch.inbox.insert(port, payload.clone());
                             // put the original back untouched
                             incomplete2.insert(old_predicate, branch);
                             if payload_branch.blocking_on == Some(port) {
                                 // run the fork
                                 payload_branch.blocking_on = None;
                                 Some((new, payload_branch))
                             } else {
                                 // don't bother running. its awaiting something else
                                 incomplete2.insert(new, payload_branch);
                                 None
+                            }
+                        }
                         Csr::LatterNotFormer => {
                             log!(
                                 &mut m_ctx.inner.logger,
                                 "... poly_recv_run payloadpred {:?} subsumes branch pred {:?}. FORKING",
                                 &old_predicate,
                                 &payload_predicate,
                             );
                             // payload_predicate has new assumptions. FORK!
                             let mut payload_branch = branch.clone();
                             if let Some(prev_payload) = payload_branch.inbox.get(&port) {
                                 // Incorrect to receive two distinct messages in same branch!
                                 assert_eq!(prev_payload, &payload);
+                            }
                             payload_branch.inbox.insert(port, payload.clone());
                             // put the original back untouched
                             incomplete2.insert(old_predicate.clone(), branch);
                             if payload_branch.blocking_on == Some(port) {
                                 // run the fork
                                 payload_branch.blocking_on = None;
                                 Some((payload_predicate.clone(), payload_branch))
                             } else {
                                 // don't bother running. its awaiting something else
                                 incomplete2.insert(payload_predicate.clone(), payload_branch);
                                 None
+                            }
+                        }
                         Csr::Nonexistant => {
                             log!(
                                 &mut m_ctx.inner.logger,
                                 "... poly_recv_run SKIPPING because branchpred={:?}. payloadpred={:?}",
                                 &old_predicate,
                                 &payload_predicate,
                             );
                             // predicates contradict
                             incomplete2.insert(old_predicate, branch);
                             None
+                        }
+                    }
                 })
                 .collect();
             std::mem::swap(&mut self.incomplete, &mut incomplete2);
             to_run
         };
         log!(
             &mut m_ctx.inner.logger,
             "... DONE FEEDING BRANCHES. {} branches to run!",
             to_run.len(),
         );
         self.poly_run_these_branches(m_ctx, protocol_description, to_run)
+    }
     pub(crate) fn choose_mono(&self, decision: &Predicate) -> Option<MonoP> {
         self.complete
             .iter()
             .find(|(p, _)| decision.satisfies(p))
             .map(|(_, branch)| MonoP { state: branch.state.clone(), ports: self.ports.clone() })
+    }
+}
 impl PolyN {
     pub fn sync_recv(
         &mut self,
-        port: Port,
+        port: PortId,
         logger: &mut String,
         payload: Payload,
         payload_predicate: Predicate,
         solution_storage: &mut SolutionStorage,
     ) {
         let mut branches2: HashMap<_, _> = Default::default();
         for (old_predicate, mut branch) in self.branches.drain() {
             use CommonSatResult as Csr;
             let case = old_predicate.common_satisfier(&payload_predicate);
             let mut report_if_solution =
                 |branch: &BranchN, pred: &Predicate, logger: &mut String| {
                     if branch.to_get.is_empty() {
                         log!(logger, "Native reporting solution with inbox {:#?}", &branch.gotten);
                         solution_storage.submit_and_digest_subtree_solution(
                             logger,
                             SubtreeId::PolyN,
                             pred.clone(),
                         );
+                    }
                 };
             log!(
                 logger,
                 "Feeding msg {:?} {:?} to native branch with pred {:?}. Predicate case {:?}",
                 &payload_predicate,
                 &payload,
                 &old_predicate,
                 &case
             );
             match case {
                 Csr::Nonexistant => { /* skip branch */ }
                 Csr::LatterNotFormer | Csr::Equivalent => {
                     // Feed the message to this branch in-place. no need to modify pred.
                     if branch.to_get.remove(&port) {
                         branch.gotten.insert(port, payload.clone());
                         report_if_solution(&branch, &old_predicate, logger);
+                    }
+                }
                 Csr::FormerNotLatter => {
                     // create a new branch with the payload_predicate.
                     let mut forked = branch.clone();
                     if forked.to_get.remove(&port) {
                         forked.gotten.insert(port, payload.clone());
                         report_if_solution(&forked, &payload_predicate, logger);
                         branches2.insert(payload_predicate.clone(), forked);
+                    }
+                }
                 Csr::New(new) => {
                     // create a new branch with the newly-created predicate
                     let mut forked = branch.clone();
                     if forked.to_get.remove(&port) {
                         forked.gotten.insert(port, payload.clone());
                         report_if_solution(&forked, &new, logger);
                         branches2.insert(new.clone(), forked);
+                    }
+                }
+            }
             // unlike PolyP machines, Native branches do NOT become inconsistent
             branches2.insert(old_predicate, branch);
+        }
         log!(logger, "Native now has branches {:#?}", &branches2);
         std::mem::swap(&mut branches2, &mut self.branches);
+    }
     pub fn choose_mono(&self, decision: &Predicate) -> Option<MonoN> {
         self.branches
             .iter()
             .find(|(p, branch)| branch.to_get.is_empty() && decision.satisfies(p))
             .map(|(_, branch)| {
                 let BranchN { gotten, sync_batch_index, .. } = branch.clone();
                 MonoN { ports: self.ports.clone(), result: Some((sync_batch_index, gotten)) }
             })
+    }
+}

src/runtime/communication.rs

➞

Show inline comments

@@ @@ -12,728 +12,728 @@ impl Controller { @@
                     poly_n.choose_mono(predicate).unwrap_or_else(|| {
                         panic!(
                             "Ending round with decision pred {:#?} but poly_n has branches {:#?}. My log is... {}",
                             &predicate, &poly_n.branches, &self.inner.logger
                         );
                     })
                 };
                 self.inner.mono_ps.clear();
                 self.inner.mono_ps.extend(
                     self.ephemeral
                         .poly_ps
                         .drain(..)
                         .map(|poly_p| poly_p.choose_mono(predicate).unwrap()),
                 );
                 Ok(())
+            }
             Decision::Failure => Err(SyncErr::Timeout),
         };
         let announcement = CommMsgContents::Announce { decision }.into_msg(self.inner.round_index);
         for &child_port in self.inner.family.children_ports.iter() {
             log!(
                 &mut self.inner.logger,
                 "Forwarding {:?} to child with port {:?}",
                 &announcement,
                 child_port
             );
             self.inner
                 .endpoint_exts
                 .get_mut(child_port)
                 .expect("eefef")
                 .endpoint
                 .send(announcement.clone())?;
+        }
         self.inner.round_index += 1;
         self.ephemeral.clear();
         ret
+    }
     // Drain self.ephemeral.solution_storage and handle the new locals. Return decision if one is found
     fn handle_locals_maybe_decide(&mut self) -> Result<bool, SyncErr> {
         if let Some(parent_port) = self.inner.family.parent_port {
             // I have a parent -> I'm not the leader
             let parent_endpoint =
                 &mut self.inner.endpoint_exts.get_mut(parent_port).expect("huu").endpoint;
             for partial_oracle in self.ephemeral.solution_storage.iter_new_local_make_old() {
                 let msg =
                     CommMsgContents::Elaborate { partial_oracle }.into_msg(self.inner.round_index);
                 log!(&mut self.inner.logger, "Sending {:?} to parent {:?}", &msg, parent_port);
                 parent_endpoint.send(msg)?;
+            }
             Ok(false)
         } else {
             // I have no parent -> I'm the leader
             assert!(self.inner.family.parent_port.is_none());
             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
             })
+        }
+    }
     fn kick_off_native(
         &mut self,
         sync_batches: impl Iterator<Item = SyncBatch>,
     ) -> Result<PolyN, EndpointErr> {
         let MonoN { ports, .. } = 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 port_to_channel_id = |port| endpoint_exts.get(port).unwrap().info.channel_id;
             let all_ports = ports.iter().copied();
             let all_channel_ids = all_ports.map(port_to_channel_id);
             let mut predicate = Predicate::new_trivial();
             // assign TRUE for puts and gets
             let true_ports = puts.keys().chain(gets.iter()).copied();
             let true_channel_ids = true_ports.clone().map(port_to_channel_id);
             predicate.batch_assign_nones(true_channel_ids, true);
             // assign FALSE for all in interface not assigned true
             predicate.batch_assign_nones(all_channel_ids.clone(), false);
             if branches.contains_key(&predicate) {
                 // TODO what do I do with redundant predicates?
                 unimplemented!(
                     "Duplicate predicate {:#?}!\nHaving multiple batches with the same
                     predicate requires the support of oracle boolean variables",
                     &predicate,
+                )
+            }
             let branch = BranchN { to_get: gets, gotten: Default::default(), sync_batch_index };
             for (port, payload) in puts {
                 log!(
                     &mut self.inner.logger,
                     "... ... Initial native put msg {:?} pred {:?} batch {:?}",
                     &payload,
                     &predicate,
                     sync_batch_index,
                 );
                 let msg =
                     CommMsgContents::SendPayload { payload_predicate: predicate.clone(), payload }
                         .into_msg(*round_index);
                 endpoint_exts.get_mut(port).unwrap().endpoint.send(msg)?;
+            }
             log!(
                 &mut self.inner.logger,
                 "... Initial native branch batch index={} with pred {:?}",
                 sync_batch_index,
                 &predicate
             );
             if branch.to_get.is_empty() {
                 self.ephemeral.solution_storage.submit_and_digest_subtree_solution(
                     &mut self.inner.logger,
                     SubtreeId::PolyN,
                     predicate.clone(),
                 );
+            }
             branches.insert(predicate, branch);
+        }
         Ok(PolyN { ports, branches })
+    }
     pub fn sync_round(
         &mut self,
         deadline: Option<Instant>,
         sync_batches: Option<impl Iterator<Item = SyncBatch>>,
     ) -> Result<(), SyncErr> {
         if let Some(e) = self.unrecoverable_error {
             return Err(e.clone());
+        }
         self.sync_round_inner(deadline, sync_batches).map_err(move |e| match e {
             SyncErr::Timeout => e, // this isn't unrecoverable
             _ => {
                 // Must set unrecoverable error! and tear down our net channels
                 self.unrecoverable_error = Some(e);
                 self.ephemeral.clear();
                 self.inner.endpoint_exts = Default::default();
+                e
+            }
         })
+    }
     // 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(
         &mut self,
         mut deadline: Option<Instant>,
         sync_batches: Option<impl Iterator<Item = SyncBatch>>,
     ) -> Result<(), SyncErr> {
         log!(
             &mut self.inner.logger,
             "~~~~~~~~ SYNC ROUND STARTS! ROUND={} ~~~~~~~~~",
             self.inner.round_index
         );
         assert!(self.ephemeral.is_clear());
         assert!(self.unrecoverable_error.is_none());
         // 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
         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 {
                 ports: &mut mono_p.ports,
                 mono_ps: &mut self.ephemeral.mono_ps,
                 inner: &mut self.inner,
             };
             // 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),
                 MonoBlocker::ComponentExit => drop(mono_p),
                 MonoBlocker::SyncBlockStart => self.ephemeral.poly_ps.push(mono_p.into()),
+            }
+        }
         log!(
             &mut self.inner.logger,
             "Finished running all MonoPs! Have {} PolyPs waiting",
             self.ephemeral.poly_ps.len()
         );
         // 3. define the mapping from port -> 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 port_to_holder: HashMap<Port, PolyId> = {
+        let port_to_holder: HashMap<PortId, PolyId> = {
             use PolyId::*;
             let n = self.inner.mono_n.ports.iter().map(move |&e| (e, N));
             let p = self
                 .ephemeral
                 .poly_ps
                 .iter()
                 .enumerate()
                 .flat_map(|(index, m)| m.ports.iter().map(move |&e| (e, P { index })));
             n.chain(p).collect()
         };
         log!(
             &mut self.inner.logger,
             "SET OF PolyPs and MonoPs final! port lookup map is {:?}",
             &port_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 = std::iter::once(SubtreeId::PolyN);
             let m = (0..self.ephemeral.poly_ps.len()).map(|index| SubtreeId::PolyP { index });
             let c = self
                 .inner
                 .family
                 .children_ports
                 .iter()
                 .map(|&port| SubtreeId::ChildController { port });
             let subtree_id_iter = n.chain(m).chain(c);
             log!(
                 &mut self.inner.logger,
                 "Solution Storage has subtree Ids: {:?}",
                 &subtree_id_iter.clone().collect::<Vec<_>>()
             );
             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...");
         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,
                 "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 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_port {
                             None => {
                                 // I am the sink! announce failure and return.
                                 return self.end_round_with_decision(Decision::Failure);
+                            }
                             Some(parent_port) => {
                                 // 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 port {:?}",
                                     &announcement,
                                     parent_port
                                 );
                                 self.inner
                                     .endpoint_exts
                                     .get_mut(parent_port)
                                     .expect("ss")
                                     .endpoint
                                     .send(announcement.clone())?;
                                 continue; // poll some more
+                            }
+                        }
+                    }
                 },
             };
             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 port {:?}",
                         received.recipient
                     );
                     assert_eq!(round_index, self.inner.round_index);
                     contents
+                }
             };
             match current_content {
                 CommMsgContents::Failure => match self.inner.family.parent_port {
                     Some(parent_port) => {
                         let announcement = Msg::CommMsg(CommMsg {
                             round_index: self.inner.round_index,
                             contents: CommMsgContents::Failure,
                         });
                         log!(
                             &mut self.inner.logger,
                             "Forwarding {:?} to parent with port {:?}",
                             &announcement,
                             parent_port
                         );
                         self.inner
                             .endpoint_exts
                             .get_mut(parent_port)
                             .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_ports.contains(&received.recipient) {
                         return Err(SyncErr::ElaborateFromNonChild);
+                    }
                     let subtree_id = SubtreeId::ChildController { port: received.recipient };
                     log!(
                         &mut self.inner.logger,
                         "Received elaboration from child for subtree {:?}: {:?}",
                         subtree_id,
                         &partial_oracle
                     );
                     self.ephemeral.solution_storage.submit_and_digest_subtree_solution(
                         &mut self.inner.logger,
                         subtree_id,
                         partial_oracle,
                     );
                     if self.handle_locals_maybe_decide()? {
                         return Ok(());
+                    }
+                }
                 CommMsgContents::Announce { decision } => {
                     if self.inner.family.parent_port != Some(received.recipient) {
                         return Err(SyncErr::AnnounceFromNonParent);
+                    }
                     log!(
                         &mut self.inner.logger,
                         "Received ANNOUNCEMENT from from parent {:?}: {:?}",
                         received.recipient,
                         &decision
                     );
                     return self.end_round_with_decision(decision);
+                }
                 CommMsgContents::SendPayload { payload_predicate, payload } => {
                     // check that we expect to be able to receive payloads from this sender
                     assert_eq!(
                         Getter,
                         self.inner.endpoint_exts.get(received.recipient).unwrap().info.polarity
                     );
                     // message for some actor. Feed it to the appropriate actor
                     // and then give them another chance to run.
                     let subtree_id = port_to_holder.get(&received.recipient);
                     log!(
                         &mut self.inner.logger,
                         "Received SendPayload for subtree {:?} with pred {:?} and payload {:?}",
                         subtree_id,
                         &payload_predicate,
                         &payload
                     );
                     let channel_id = self
                         .inner
                         .endpoint_exts
                         .get(received.recipient)
                         .expect("UEHFU")
                         .info
                         .channel_id;
                     if payload_predicate.query(channel_id) != Some(true) {
                         // sender didn't preserve the invariant
                         return Err(SyncErr::PayloadPremiseExcludesTheChannel(channel_id));
+                    }
                     match subtree_id {
                         None => {
                             // this happens when a message is sent to a component that has exited.
                             // It's safe to drop this message;
                             // The sender branch will certainly not be part of the solution
+                        }
                         Some(PolyId::N) => {
                             // Message for NativeMachine
                             self.ephemeral.poly_n.as_mut().unwrap().sync_recv(
                                 received.recipient,
                                 &mut self.inner.logger,
                                 payload,
                                 payload_predicate,
                                 &mut self.ephemeral.solution_storage,
                             );
                             if self.handle_locals_maybe_decide()? {
                                 return Ok(());
+                            }
+                        }
                         Some(PolyId::P { index }) => {
                             // Message for protocol actor
                             let poly_p = &mut self.ephemeral.poly_ps[*index];
                             let m_ctx = PolyPContext {
                                 my_subtree_id: SubtreeId::PolyP { index: *index },
                                 inner: &mut self.inner,
                                 solution_storage: &mut self.ephemeral.solution_storage,
                             };
                             use SyncRunResult as Srr;
                             let blocker = poly_p.poly_recv_run(
                                 m_ctx,
                                 &self.protocol_description,
                                 received.recipient,
                                 payload_predicate,
                                 payload,
                             )?;
                             log!(
                                 &mut self.inner.logger,
                                 "... Fed the msg to PolyP {:?} and ran it to blocker {:?}",
                                 subtree_id,
                                 blocker
                             );
                             match blocker {
                                 Srr::NoBranches => return Err(SyncErr::Inconsistent),
                                 Srr::BlockingForRecv | Srr::AllBranchesComplete => {
+                                    {
                                         let peeked = self
                                             .ephemeral
                                             .solution_storage
                                             .peek_new_locals()
                                             .collect::<Vec<_>>();
                                         log!(
                                             &mut self.inner.logger,
                                             "Got {} new controller-local solutions from RECV: {:?}",
                                             peeked.len(),
                                             peeked
                                         );
+                                    }
                                     if self.handle_locals_maybe_decide()? {
                                         return Ok(());
+                                    }
+                                }
+                            }
+                        }
                     };
+                }
+            }
+        }
+    }
+}
 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.port_to_holder.is_empty()
+    }
     fn clear(&mut self) {
         self.solution_storage.clear();
         self.poly_n.take();
         self.poly_ps.clear();
         self.port_to_holder.clear();
+    }
+}
 impl Into<PolyP> for MonoP {
     fn into(self) -> PolyP {
         PolyP {
             complete: Default::default(),
             incomplete: hashmap! {
                 Predicate::new_trivial() =>
                 BranchP {
                     state: self.state,
                     inbox: Default::default(),
                     outbox: Default::default(),
                     blocking_on: None,
+                }
             },
             ports: self.ports,
+        }
+    }
+}
 impl From<EndpointErr> for SyncErr {
     fn from(e: EndpointErr) -> SyncErr {
         SyncErr::EndpointErr(e)
+    }
+}
 impl MonoContext for MonoPContext<'_> {
     type D = ProtocolD;
     type S = ProtocolS;
-    fn new_component(&mut self, moved_ports: HashSet<Port>, init_state: Self::S) {
+    fn new_component(&mut self, moved_ports: HashSet<PortId>, init_state: Self::S) {
         log!(
             &mut self.inner.logger,
             "!! MonoContext callback to new_component with ports {:?}!",
             &moved_ports,
         );
         if moved_ports.is_subset(self.ports) {
             self.ports.retain(|x| !moved_ports.contains(x));
             self.mono_ps.push(MonoP { state: init_state, ports: moved_ports });
         } else {
             panic!("MachineP attempting to move alien port!");
+        }
+    }
-    fn new_channel(&mut self) -> [Port; 2] {
+    fn new_channel(&mut self) -> [PortId; 2] {
         let [a, b] = Endpoint::new_memory_pair();
         let channel_id = self.inner.channel_id_stream.next();
         let mut clos = |endpoint, polarity| {
             let endpoint_ext =
                 EndpointExt { info: EndpointInfo { polarity, channel_id }, endpoint };
             let port = self.inner.endpoint_exts.alloc(endpoint_ext);
             let endpoint = &self.inner.endpoint_exts.get(port).unwrap().endpoint;
-            let token = Port::to_token(port);
+            let token = PortId::to_token(port);
             self.inner
                 .messenger_state
                 .poll
                 .register(endpoint, token, Ready::readable(), PollOpt::edge())
                 .expect("AAGAGGGGG");
             self.ports.insert(port);
             port
         };
         let [kp, kg] = [clos(a, Putter), clos(b, Getter)];
         log!(
             &mut self.inner.logger,
             "!! MonoContext callback to new_channel. returning ports {:?}!",
             [kp, kg],
         );
         [kp, kg]
+    }
     fn new_random(&mut self) -> u64 {
         type Bytes8 = [u8; std::mem::size_of::<u64>()];
         let mut bytes = Bytes8::default();
         getrandom::getrandom(&mut bytes).unwrap();
         let val = unsafe { std::mem::transmute::<Bytes8, _>(bytes) };
         log!(
             &mut self.inner.logger,
             "!! MonoContext callback to new_random. returning val {:?}!",
             val,
         );
         val
+    }
+}
 impl SolutionStorage {
     fn is_clear(&self) -> bool {
         self.subtree_id_to_index.is_empty()
             && self.subtree_solutions.is_empty()
             && self.old_local.is_empty()
             && self.new_local.is_empty()
+    }
     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 = SubtreeId>) {
         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());
             self.subtree_solutions.push(Default::default())
+        }
+    }
     pub(crate) fn peek_new_locals(&self) -> impl Iterator<Item = &Predicate> + '_ {
         self.new_local.iter()
+    }
     pub(crate) fn iter_new_local_make_old(&mut self) -> impl Iterator<Item = Predicate> + '_ {
         let Self { old_local, new_local, .. } = self;
         new_local.drain().map(move |local| {
             old_local.insert(local.clone());
             local
         })
+    }
     pub(crate) fn submit_and_digest_subtree_solution(
         &mut self,
         logger: &mut String,
         subtree_id: SubtreeId,
         predicate: Predicate,
     ) {
         log!(logger, "NEW COMPONENT SOLUTION {:?} {:?}", subtree_id, &predicate);
         let index = self.subtree_id_to_index[&subtree_id];
         let left = 0..index;
         let right = (index + 1)..self.subtree_solutions.len();
         let Self { subtree_solutions, new_local, old_local, .. } = self;
         let was_new = subtree_solutions[index].insert(predicate.clone());
         if was_new {
             let set_visitor = left.chain(right).map(|index| &subtree_solutions[index]);
             Self::elaborate_into_new_local_rec(
                 logger,
                 predicate,
                 set_visitor,
                 old_local,
                 new_local,
             );
+        }
+    }
     fn elaborate_into_new_local_rec<'a, 'b>(
         logger: &mut String,
         partial: Predicate,
         mut set_visitor: impl Iterator<Item = &'b HashSet<Predicate>> + Clone,
         old_local: &'b HashSet<Predicate>,
         new_local: &'a mut HashSet<Predicate>,
     ) {
         if let Some(set) = set_visitor.next() {
             // incomplete solution. keep traversing
             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, port: Port) -> Option<bool> {
+    fn is_firing(&mut self, port: PortId) -> Option<bool> {
         assert!(self.ports.contains(&port));
         let channel_id = self.m_ctx.inner.endpoint_exts.get(port).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, port: Port) -> Option<&Payload> {
+    fn read_msg(&mut self, port: PortId) -> Option<&Payload> {
         assert!(self.ports.contains(&port));
         let val = self.inbox.get(&port);
         log!(
             &mut self.m_ctx.inner.logger,
             "!! PolyContext callback to read_msg by {:?}! returning {:?}",
             self.m_ctx.my_subtree_id,
             val,
         );
         val
+    }
+}

src/runtime/errors.rs

➞

Show inline comments

 use crate::common::*;
 #[derive(Debug, Copy, Clone, PartialEq, Eq)]
 pub enum PortBindErr {
     AlreadyConnected,
     IndexOutOfBounds,
     NotConfigured,
     ParseErr,
     AlreadyConfigured,
+}
 #[derive(Debug, Copy, Clone, PartialEq, Eq)]
 pub enum ReadGottenErr {
     NotConnected,
     IndexOutOfBounds,
     WrongPolarity,
     NoPreviousRound,
     DidNotGet,
+}
 #[derive(Debug, Copy, Clone, PartialEq, Eq)]
 pub enum PortOpErr {
     IndexOutOfBounds,
     NotConnected,
     WrongPolarity,
     DuplicateOperation,
+}
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum ConfigErr {
     AlreadyConnected,
     ParseErr(String),
     AlreadyConfigured,
     NoSuchComponent,
     NonPortTypeParameters,
+}
 #[derive(Debug, Copy, Clone, PartialEq, Eq)]
 pub enum ConnectErr {
     PortNotBound { native_index: usize },
     NotConfigured,
     AlreadyConnected,
     MetaProtocolDeviation,
     Disconnected,
     PollInitFailed,
     MessengerRecvErr(MessengerRecvErr),
     Timeout,
     PollingFailed,
     PolarityMatched(SocketAddr),
     AcceptFailed(SocketAddr),
     PassiveConnectFailed(SocketAddr),
     BindFailed(SocketAddr),
+}
 #[derive(Debug, Copy, Clone, PartialEq, Eq)]
 pub enum PollDeadlineErr {
     PollingFailed,
     Timeout,
+}
 #[derive(Debug, Copy, Clone, PartialEq, Eq)]
 pub enum EndpointErr {
     Disconnected,
     MetaProtocolDeviation,
+}
 #[derive(Debug, Copy, Clone, PartialEq, Eq)]
 pub enum SyncErr {
     NotConnected,
     MessengerRecvErr(MessengerRecvErr),
     Inconsistent,
     Timeout,
     ElaborateFromNonChild,
     AnnounceFromNonParent,
     PayloadPremiseExcludesTheChannel(ChannelId),
     UnexpectedSetupMsg,
     EndpointErr(EndpointErr),
     EvalErr(EvalErr),
+}
 #[derive(Debug, Copy, Clone, PartialEq, Eq)]
 pub enum EvalErr {
     ComponentExitWhileBranching,
+}
 #[derive(Debug, Copy, Clone, PartialEq, Eq)]
 pub enum MessengerRecvErr {
     PollingFailed,
-    EndpointErr(Port, EndpointErr),
+    EndpointErr(PortId, EndpointErr),
+}
 impl From<MainComponentErr> for ConfigErr {
     fn from(e: MainComponentErr) -> Self {
         use ConfigErr as C;
         use MainComponentErr as M;
         match e {
             M::NoSuchComponent => C::NoSuchComponent,
             M::NonPortTypeParameters => C::NonPortTypeParameters,
             _ => todo!(),
+        }
+    }
+}

src/runtime/mod.rs

➞

Show inline comments

 #[cfg(feature = "ffi")]
 pub mod ffi;
 mod actors;
 pub(crate) mod communication;
 pub(crate) mod connector;
 pub(crate) mod endpoint;
 pub mod errors;
 mod serde;
 pub(crate) mod setup;
 // #[cfg(feature = "ffi")]
 // pub mod ffi;
 // mod actors;
 // pub(crate) mod communication;
 // pub(crate) mod connector;
 // pub(crate) mod endpoint;
 // pub mod errors;
 // mod serde;
 mod my_tests;
 mod setup2;
 // pub(crate) mod setup;
 // mod v2;
 pub(crate) type ProtocolD = crate::protocol::ProtocolDescriptionImpl;
 pub(crate) type ProtocolS = crate::protocol::ComponentStateImpl;
 use crate::common::*;
 use actors::*;
 use endpoint::*;
 use errors::*;
 // use actors::*;
 // use endpoint::*;
 // use errors::*;
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 pub(crate) enum Decision {
     Failure,
     Success(Predicate),
+}
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 pub(crate) enum Msg {
     SetupMsg(SetupMsg),
     CommMsg(CommMsg),
+}
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 pub struct MyPortInfo {
     polarity: Polarity,
     port: PortId,
+}
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 pub(crate) enum SetupMsg {
     // sent by the passive endpoint to the active endpoint
     // MyPortInfo(MyPortInfo),
     LeaderEcho { maybe_leader: ControllerId },
     LeaderAnnounce { leader: ControllerId },
     YouAreMyParent,
+}
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 pub(crate) struct CommMsg {
     pub round_index: usize,
     pub contents: CommMsgContents,
+}
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 pub(crate) enum CommMsgContents {
     SendPayload { payload_predicate: Predicate, payload: Payload },
     Elaborate { partial_oracle: Predicate }, // SINKWARD
     Failure,                                 // SINKWARD
     Announce { decision: Decision },         // SINKAWAYS
+}
 #[derive(Debug, PartialEq)]
 pub(crate) enum CommonSatResult {
     FormerNotLatter,
     LatterNotFormer,
     Equivalent,
     New(Predicate),
     Nonexistant,
+}
 #[derive(Clone, Eq, PartialEq, Hash)]
 pub(crate) struct Predicate {
     pub assigned: BTreeMap<ChannelId, bool>,
 pub struct Endpoint {
     inbox: Vec<u8>,
     stream: mio07::net::TcpStream,
+}
 #[derive(Debug, Default)]
 struct SyncBatch {
     puts: HashMap<Port, Payload>,
     gets: HashSet<Port>,
 pub struct IntStream {
     next: u32,
+}
 #[derive(Debug)]
 pub enum Connector {
     Unconfigured(Unconfigured),
     Configured(Configured),
     Connected(Connected), // TODO consider boxing. currently takes up a lot of stack space
 pub struct IdManager {
     controller_id: ControllerId,
     port_suffix_stream: IntStream,
+}
 #[derive(Debug)]
 pub struct Unconfigured {
     pub controller_id: ControllerId,
 pub struct ProtoComponent {
     state: ComponentState,
     ports: HashSet<PortId>,
+}
 #[derive(Debug)]
 pub struct Configured {
     controller_id: ControllerId,
     polarities: Vec<Polarity>,
     bindings: HashMap<usize, PortBinding>,
     protocol_description: Arc<ProtocolD>,
     main_component: Vec<u8>,
     logger: String,
 pub enum InpRoute {
     NativeComponent,
     ProtoComponent { index: usize },
     Endpoint { index: usize },
+}
 #[derive(Debug)]
 pub struct Connected {
     native_interface: Vec<(Port, Polarity)>,
     sync_batches: Vec<SyncBatch>,
     // controller is cooperatively scheduled with the native application
     // (except for transport layer behind Endpoints, which are managed by the OS)
     // control flow is passed to the controller during methods on Connector (primarily, connect and sync).
     controller: Controller,
 pub trait Logger: Debug {
     fn line_writer(&mut self) -> &mut dyn std::fmt::Write;
     fn dump_log(&self, w: &mut dyn std::io::Write);
+}
 #[derive(Debug, Copy, Clone)]
 pub enum PortBinding {
     Native,
     Active(SocketAddr),
     Passive(SocketAddr),
 #[derive(Debug, Clone)]
 pub struct EndpointSetup {
     pub polarity: Polarity,
     pub sock_addr: SocketAddr,
     pub is_active: bool,
+}
 #[derive(Debug)]
 struct Arena<T> {
     storage: Vec<T>,
 pub struct EndpointExt {
     endpoint: Endpoint,
     inp_for_emerging_msgs: PortId,
+}
 #[derive(Debug)]
 struct ReceivedMsg {
     recipient: Port,
     msg: Msg,
 pub struct Neighborhood {
     parent: Option<usize>,
     children: Vec<usize>, // ordered, deduplicated
+}
 #[derive(Debug)]
 struct MessengerState {
     poll: Poll,
     events: Events,
     delayed: Vec<ReceivedMsg>,
     undelayed: Vec<ReceivedMsg>,
     polled_undrained: IndexSet<Port>,
 pub struct MemInMsg {
     inp: PortId,
     msg: Payload,
+}
 #[derive(Debug)]
 struct ChannelIdStream {
     controller_id: ControllerId,
     next_channel_index: ChannelIndex,
 pub struct EndpointPoller {
     poll: mio07::Poll,
     events: mio07::Events,
     undrained_endpoints: HashSet<usize>,
     delayed_inp_messages: Vec<(PortId, Msg)>,
+}
 #[derive(Debug)]
 struct Controller {
     protocol_description: Arc<ProtocolD>,
     inner: ControllerInner,
     ephemeral: ControllerEphemeral,
     unrecoverable_error: Option<SyncErr>, // prevents future calls to Sync
 pub struct Connector {
     logger: Box<dyn Logger>,
     proto_description: Arc<ProtocolDescription>,
     id_manager: IdManager,
     native_ports: HashSet<PortId>,
     proto_components: Vec<ProtoComponent>,
     outp_to_inp: HashMap<PortId, PortId>,
     inp_to_route: HashMap<PortId, InpRoute>,
     phased: ConnectorPhased,
+}
 #[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>,
     port_to_holder: HashMap<Port, PolyId>,
 pub enum ConnectorPhased {
     Setup {
         endpoint_setups: Vec<(PortId, EndpointSetup)>,
         surplus_sockets: u16,
     },
     Communication {
         endpoint_poller: EndpointPoller,
         endpoint_exts: Vec<EndpointExt>,
         neighborhood: Neighborhood,
         mem_inbox: Vec<MemInMsg>,
     },
+}
 #[derive(Debug)]
 struct ControllerFamily {
     parent_port: Option<Port>,
     children_ports: Vec<Port>,
 pub struct StringLogger(ControllerId, String);
 #[derive(Clone, Eq, PartialEq, Hash, serde::Serialize, serde::Deserialize)]
 pub(crate) struct Predicate {
     pub assigned: BTreeMap<PortId, bool>,
+}
 #[derive(Debug)]
 pub(crate) enum SyncRunResult {
     BlockingForRecv,
     AllBranchesComplete,
     NoBranches,
 #[derive(Debug, Default)]
 struct SyncBatch {
     puts: HashMap<PortId, Payload>,
     gets: HashSet<PortId>,
+}
 // Used to identify poly actors
 #[derive(Debug, Copy, Clone, Eq, PartialEq, Ord, PartialOrd, Hash)]
 enum PolyId {
     N,
     P { index: usize },
 pub struct MonitoredReader<R: Read> {
     bytes: usize,
     r: R,
+}
 #[derive(Debug, Copy, Clone, Eq, PartialEq, Ord, PartialOrd, Hash)]
 pub(crate) enum SubtreeId {
     PolyN,
     PolyP { index: usize },
     ChildController { port: Port },
 pub enum EndpointRecvErr {
     MalformedMessage,
     BrokenEndpoint,
+}
 pub(crate) struct MonoPContext<'a> {
     inner: &'a mut ControllerInner,
     ports: &'a mut HashSet<Port>,
     mono_ps: &'a mut Vec<MonoP>,
 pub struct SyncContext<'a> {
     connector: &'a mut Connector,
+}
 pub(crate) struct PolyPContext<'a> {
     my_subtree_id: SubtreeId,
     inner: &'a mut ControllerInner,
     solution_storage: &'a mut SolutionStorage,
 pub struct NonsyncContext<'a> {
     connector: &'a mut Connector,
+}
 impl PolyPContext<'_> {
     #[inline(always)]
     fn reborrow<'a>(&'a mut self) -> PolyPContext<'a> {
         let Self { solution_storage, my_subtree_id, inner } = self;
         PolyPContext { solution_storage, my_subtree_id: *my_subtree_id, inner }
 ////////////////
 impl Debug for Endpoint {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         f.debug_struct("Endpoint").field("inbox", &self.inbox).finish()
+    }
+}
 struct BranchPContext<'m, 'r> {
     m_ctx: PolyPContext<'m>,
     ports: &'r HashSet<Port>,
     predicate: &'r Predicate,
     inbox: &'r HashMap<Port, Payload>,
+}
 #[derive(Default)]
 pub(crate) struct SolutionStorage {
     old_local: HashSet<Predicate>,
     new_local: HashSet<Predicate>,
     // this pair acts as SubtreeId -> HashSet<Predicate> which is friendlier to iteration
     subtree_solutions: Vec<HashSet<Predicate>>,
     subtree_id_to_index: HashMap<SubtreeId, usize>,
+}
 trait Messengerlike {
     fn get_state_mut(&mut self) -> &mut MessengerState;
     fn get_endpoint_mut(&mut self, eport: Port) -> &mut Endpoint;
     fn delay(&mut self, received: ReceivedMsg) {
         self.get_state_mut().delayed.push(received);
 impl NonsyncContext<'_> {
     pub fn new_component(&mut self, moved_ports: HashSet<PortId>, init_state: ComponentState) {
         todo!()
+    }
     fn undelay_all(&mut self) {
         let MessengerState { delayed, undelayed, .. } = self.get_state_mut();
         undelayed.extend(delayed.drain(..))
     pub fn new_channel(&mut self) -> [PortId; 2] {
         todo!()
+    }
     fn send(&mut self, to: Port, msg: Msg) -> Result<(), EndpointErr> {
         self.get_endpoint_mut(to).send(msg)
+    }
     // attempt to receive a message from one of the endpoints before the deadline
     fn recv(&mut self, deadline: Instant) -> Result<Option<ReceivedMsg>, MessengerRecvErr> {
         // try get something buffered
         if let Some(x) = self.get_state_mut().undelayed.pop() {
             return Ok(Some(x));
+        }
         loop {
             // polled_undrained may not be empty
             while let Some(eport) = self.get_state_mut().polled_undrained.pop() {
                 if let Some(msg) = self
                     .get_endpoint_mut(eport)
                     .recv()
                     .map_err(|e| MessengerRecvErr::EndpointErr(eport, e))?
+                {
                     // this endpoint MAY still have messages! check again in future
                     self.get_state_mut().polled_undrained.insert(eport);
                     return Ok(Some(ReceivedMsg { recipient: eport, msg }));
+                }
+            }
             let state = self.get_state_mut();
             match state.poll_events(deadline) {
                 Ok(()) => {
                     for e in state.events.iter() {
                         state.polled_undrained.insert(Port::from_token(e.token()));
+                    }
+                }
                 Err(PollDeadlineErr::PollingFailed) => return Err(MessengerRecvErr::PollingFailed),
                 Err(PollDeadlineErr::Timeout) => return Ok(None),
+            }
+        }
+}
 impl SyncContext<'_> {
     pub fn is_firing(&mut self, port: PortId) -> Option<bool> {
         todo!()
+    }
     fn recv_blocking(&mut self) -> Result<ReceivedMsg, MessengerRecvErr> {
         // try get something buffered
         if let Some(x) = self.get_state_mut().undelayed.pop() {
             return Ok(x);
+        }
         loop {
             // polled_undrained may not be empty
             while let Some(eport) = self.get_state_mut().polled_undrained.pop() {
                 if let Some(msg) = self
                     .get_endpoint_mut(eport)
                     .recv()
                     .map_err(|e| MessengerRecvErr::EndpointErr(eport, e))?
+                {
                     // this endpoint MAY still have messages! check again in future
                     self.get_state_mut().polled_undrained.insert(eport);
                     return Ok(ReceivedMsg { recipient: eport, msg });
+                }
+            }
             let state = self.get_state_mut();
             state
                 .poll
                 .poll(&mut state.events, None)
                 .map_err(|_| MessengerRecvErr::PollingFailed)?;
             for e in state.events.iter() {
                 state.polled_undrained.insert(Port::from_token(e.token()));
+            }
+        }
     pub fn read_msg(&mut self, port: PortId) -> Option<&Payload> {
         todo!()
+    }
+}
 /////////////////////////////////
 impl Debug for SolutionStorage {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         f.pad("Solutions: [")?;
         for (subtree_id, &index) in self.subtree_id_to_index.iter() {
             let sols = &self.subtree_solutions[index];
             f.write_fmt(format_args!("{:?}: {:?}, ", subtree_id, sols))?;
+        }
         f.pad("]")
 impl<R: Read> From<R> for MonitoredReader<R> {
     fn from(r: R) -> Self {
         Self { r, bytes: 0 }
+    }
+}
 impl From<EvalErr> for SyncErr {
     fn from(e: EvalErr) -> SyncErr {
         SyncErr::EvalErr(e)
 impl<R: Read> MonitoredReader<R> {
     pub fn bytes_read(&self) -> usize {
         self.bytes
+    }
+}
 impl From<MessengerRecvErr> for SyncErr {
     fn from(e: MessengerRecvErr) -> SyncErr {
         SyncErr::MessengerRecvErr(e)
 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 From<MessengerRecvErr> for ConnectErr {
     fn from(e: MessengerRecvErr) -> ConnectErr {
         ConnectErr::MessengerRecvErr(e)
 impl Into<Msg> for SetupMsg {
     fn into(self) -> Msg {
         Msg::SetupMsg(self)
+    }
+}
 impl<T> Default for Arena<T> {
     fn default() -> Self {
         Self { storage: vec![] }
 impl Debug for Predicate {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         f.pad("{")?;
         for (port, &v) in self.assigned.iter() {
             f.write_fmt(format_args!("{:?}=>{}, ", port, if v { 'T' } else { 'F' }))?
+        }
         f.pad("}")
+    }
+}
 impl<T> Arena<T> {
     pub fn alloc(&mut self, t: T) -> Port {
         self.storage.push(t);
         let l: u32 = self.storage.len().try_into().unwrap();
         Port::from_raw(l - 1u32)
+    }
     pub fn get(&self, key: Port) -> Option<&T> {
         self.storage.get(key.to_raw() as usize)
+    }
     pub fn get_mut(&mut self, key: Port) -> Option<&mut T> {
         self.storage.get_mut(key.to_raw() as usize)
+    }
     pub fn type_convert<X>(self, f: impl FnMut((Port, T)) -> X) -> Arena<X> {
         Arena { storage: self.keyspace().zip(self.storage.into_iter()).map(f).collect() }
+    }
     pub fn iter(&self) -> impl Iterator<Item = (Port, &T)> {
         self.keyspace().zip(self.storage.iter())
+    }
     pub fn len(&self) -> usize {
         self.storage.len()
+    }
     pub fn keyspace(&self) -> impl Iterator<Item = Port> {
         (0u32..self.storage.len().try_into().unwrap()).map(Port::from_raw)
 impl StringLogger {
     pub fn new(controller_id: ControllerId) -> Self {
         Self(controller_id, String::default())
+    }
+}
 impl ChannelIdStream {
     fn new(controller_id: ControllerId) -> Self {
         Self { controller_id, next_channel_index: 0 }
 impl Logger for StringLogger {
     fn line_writer(&mut self) -> &mut dyn std::fmt::Write {
         use std::fmt::Write;
         let _ = write!(&mut self.1, "\nCID({}): ", self.0);
         self
+    }
     fn next(&mut self) -> ChannelId {
         self.next_channel_index += 1;
         ChannelId { controller_id: self.controller_id, channel_index: self.next_channel_index - 1 }
     fn dump_log(&self, w: &mut dyn std::io::Write) {
         let _ = w.write(self.1.as_bytes());
+    }
+}
 impl MessengerState {
     // 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(())
 impl std::fmt::Write for StringLogger {
     fn write_str(&mut self, s: &str) -> Result<(), std::fmt::Error> {
         self.1.write_str(s)
+    }
+}
 impl From<PollDeadlineErr> for ConnectErr {
     fn from(e: PollDeadlineErr) -> ConnectErr {
         match e {
             PollDeadlineErr::Timeout => ConnectErr::Timeout,
             PollDeadlineErr::PollingFailed => ConnectErr::PollingFailed,
 impl IntStream {
     fn next(&mut self) -> u32 {
         if self.next == u32::MAX {
             panic!("NO NEXT!")
+        }
         self.next += 1;
         self.next - 1
+    }
+}
 impl std::ops::Not for Polarity {
     type Output = Self;
     fn not(self) -> Self::Output {
         use Polarity::*;
         match self {
             Putter => Getter,
             Getter => Putter,
+        }
 impl IdManager {
     fn next_port(&mut self) -> PortId {
         let port_suffix = self.port_suffix_stream.next();
         let controller_id = self.controller_id;
         PortId { controller_id, port_index: port_suffix }
+    }
+}
 impl Predicate {
     // returns true IFF self.unify would return Equivalent OR FormerNotLatter
     pub fn satisfies(&self, other: &Self) -> bool {
         let mut s_it = self.assigned.iter();
         let mut s = if let Some(s) = s_it.next() {
+            s
         } else {
             return other.assigned.is_empty();
         };
         for (oid, ob) in other.assigned.iter() {
             while s.0 < oid {
                 s = if let Some(s) = s_it.next() {
+                    s
                 } else {
                     return false;
                 };
+            }
             if s.0 > oid || s.1 != ob {
                 return false;
     fn new(controller_id: ControllerId) -> Self {
         Self { controller_id, port_suffix_stream: Default::default() }
+    }
+}
 impl Endpoint {
     fn try_recv<T: serde::de::DeserializeOwned>(&mut self) -> Result<Option<T>, EndpointRecvErr> {
         use EndpointRecvErr::*;
         // populate inbox as much as possible
         'read_loop: loop {
             match self.stream.read_to_end(&mut self.inbox) {
                 Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => break 'read_loop,
                 Ok(0) => break 'read_loop,
                 Ok(_) => (),
                 Err(_e) => return Err(BrokenEndpoint),
+            }
+        }
         true
+    }
     /// Given self and other, two predicates, return the most general Predicate possible, N
     /// such that n.satisfies(self) && n.satisfies(other).
     /// If none exists Nonexistant is returned.
     /// If the resulting predicate is equivlanet to self, other, or both,
     /// FormerNotLatter, LatterNotFormer and Equivalent are returned respectively.
     /// otherwise New(N) is returned.
     pub fn common_satisfier(&self, other: &Self) -> CommonSatResult {
         use CommonSatResult::*;
         // iterators over assignments of both predicates. Rely on SORTED ordering of BTreeMap's keys.
         let [mut s_it, mut o_it] = [self.assigned.iter(), other.assigned.iter()];
         let [mut s, mut o] = [s_it.next(), o_it.next()];
         // lists of assignments in self but not other and vice versa.
         let [mut s_not_o, mut o_not_s] = [vec![], vec![]];
         loop {
             match [s, o] {
                 [None, None] => break,
                 [None, Some(x)] => {
                     o_not_s.push(x);
                     o_not_s.extend(o_it);
                     break;
+                }
                 [Some(x), None] => {
                     s_not_o.push(x);
                     s_not_o.extend(s_it);
                     break;
+                }
                 [Some((sid, sb)), Some((oid, ob))] => {
                     if sid < oid {
                         // o is missing this element
                         s_not_o.push((sid, sb));
                         s = s_it.next();
                     } else if sid > oid {
                         // s is missing this element
                         o_not_s.push((oid, ob));
                         o = o_it.next();
                     } else if sb != ob {
                         assert_eq!(sid, oid);
                         // both predicates assign the variable but differ on the value
                         return Nonexistant;
                     } else {
                         // both predicates assign the variable to the same value
                         s = s_it.next();
                         o = o_it.next();
+                    }
+                }
         let mut monitored = MonitoredReader::from(&self.inbox[..]);
         match bincode::deserialize_from(&mut monitored) {
             Ok(msg) => {
                 let msg_size = monitored.bytes_read();
                 self.inbox.drain(0..(msg_size.try_into().unwrap()));
                 Ok(Some(msg))
+            }
+        }
         // Observed zero inconsistencies. A unified predicate exists...
         match [s_not_o.is_empty(), o_not_s.is_empty()] {
             [true, true] => Equivalent,       // ... equivalent to both.
             [false, true] => FormerNotLatter, // ... equivalent to self.
             [true, false] => LatterNotFormer, // ... equivalent to other.
             [false, false] => {
                 // ... which is the union of the predicates' assignments but
                 //     is equivalent to neither self nor other.
                 let mut new = self.clone();
                 for (&id, &b) in o_not_s {
                     new.assigned.insert(id, b);
             Err(e) => match *e {
                 bincode::ErrorKind::Io(k) if k.kind() == std::io::ErrorKind::UnexpectedEof => {
                     Ok(None)
+                }
                 New(new)
+            }
+        }
+    }
     pub fn iter_matching(&self, value: bool) -> impl Iterator<Item = ChannelId> + '_ {
         self.assigned
             .iter()
             .filter_map(move |(&channel_id, &b)| if b == value { Some(channel_id) } else { None })
+    }
     pub fn batch_assign_nones(
         &mut self,
         channel_ids: impl Iterator<Item = ChannelId>,
         value: bool,
     ) {
         for channel_id in channel_ids {
             self.assigned.entry(channel_id).or_insert(value);
+        }
+    }
     pub fn replace_assignment(&mut self, channel_id: ChannelId, value: bool) -> Option<bool> {
         self.assigned.insert(channel_id, value)
+    }
     pub fn union_with(&self, other: &Self) -> Option<Self> {
         let mut res = self.clone();
         for (&channel_id, &assignment_1) in other.assigned.iter() {
             match res.assigned.insert(channel_id, assignment_1) {
                 Some(assignment_2) if assignment_1 != assignment_2 => return None,
                 _ => {}
+            }
                 _ => Err(MalformedMessage),
                 // println!("SERDE ERRKIND {:?}", e);
                 // Err(MalformedMessage)
             },
+        }
         Some(res)
+    }
     pub fn query(&self, x: ChannelId) -> Option<bool> {
         self.assigned.get(&x).copied()
+    }
     pub fn new_trivial() -> Self {
         Self { assigned: Default::default() }
+    }
+}
 impl Debug for Predicate {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         f.pad("{")?;
         for (ChannelId { controller_id, channel_index }, &v) in self.assigned.iter() {
             f.write_fmt(format_args!(
                 "({:?},{:?})=>{}, ",
                 controller_id,
                 channel_index,
                 if v { 'T' } else { 'F' }
             ))?
+        }
         f.pad("}")
+    }
+}
 #[test]
 fn pred_sat() {
     use maplit::btreemap;
     let mut c = ChannelIdStream::new(0);
     let ch = std::iter::repeat_with(move || c.next()).take(5).collect::<Vec<_>>();
     let p = Predicate::new_trivial();
     let p_0t = Predicate { assigned: btreemap! { ch[0] => true } };
     let p_0f = Predicate { assigned: btreemap! { ch[0] => false } };
     let p_0f_3f = Predicate { assigned: btreemap! { ch[0] => false, ch[3] => false } };
     let p_0f_3t = Predicate { assigned: btreemap! { ch[0] => false, ch[3] => true } };
     assert!(p.satisfies(&p));
     assert!(p_0t.satisfies(&p_0t));
     assert!(p_0f.satisfies(&p_0f));
     assert!(p_0f_3f.satisfies(&p_0f_3f));
     assert!(p_0f_3t.satisfies(&p_0f_3t));
     assert!(p_0t.satisfies(&p));
     assert!(p_0f.satisfies(&p));
     assert!(p_0f_3f.satisfies(&p_0f));
     assert!(p_0f_3t.satisfies(&p_0f));
     assert!(!p.satisfies(&p_0t));
     assert!(!p.satisfies(&p_0f));
     assert!(!p_0f.satisfies(&p_0t));
     assert!(!p_0t.satisfies(&p_0f));
     assert!(!p_0f_3f.satisfies(&p_0f_3t));
     assert!(!p_0f_3t.satisfies(&p_0f_3f));
     assert!(!p_0t.satisfies(&p_0f_3f));
     assert!(!p_0f.satisfies(&p_0f_3f));
     assert!(!p_0t.satisfies(&p_0f_3t));
     assert!(!p_0f.satisfies(&p_0f_3t));
+}
 #[test]
 fn pred_common_sat() {
     use maplit::btreemap;
     use CommonSatResult::*;
     let mut c = ChannelIdStream::new(0);
     let ch = std::iter::repeat_with(move || c.next()).take(5).collect::<Vec<_>>();
     let p = Predicate::new_trivial();
     let p_0t = Predicate { assigned: btreemap! { ch[0] => true } };
     let p_0f = Predicate { assigned: btreemap! { ch[0] => false } };
     let p_3f = Predicate { assigned: btreemap! { ch[3] => false } };
     let p_0f_3f = Predicate { assigned: btreemap! { ch[0] => false, ch[3] => false } };
     let p_0f_3t = Predicate { assigned: btreemap! { ch[0] => false, ch[3] => true } };
     assert_eq![p.common_satisfier(&p), Equivalent];
     assert_eq![p_0t.common_satisfier(&p_0t), Equivalent];
     assert_eq![p.common_satisfier(&p_0t), LatterNotFormer];
     assert_eq![p_0t.common_satisfier(&p), FormerNotLatter];
     assert_eq![p_0t.common_satisfier(&p_0f), Nonexistant];
     assert_eq![p_0f_3t.common_satisfier(&p_0f_3f), Nonexistant];
     assert_eq![p_0f_3t.common_satisfier(&p_3f), Nonexistant];
     assert_eq![p_3f.common_satisfier(&p_0f_3t), Nonexistant];
     assert_eq![p_0f.common_satisfier(&p_3f), New(p_0f_3f)];
+}
     fn send<T: serde::ser::Serialize>(&mut self, msg: &T) -> Result<(), ()> {
         bincode::serialize_into(&mut self.stream, msg).map_err(drop)
+    }
+}
 impl Connector {
     fn get_logger(&self) -> &dyn Logger {
         &*self.logger
+    }
+}
 // #[derive(Debug)]
 // pub enum Connector {
 //     Unconfigured(Unconfigured),
 //     Configured(Configured),
 //     Connected(Connected), // TODO consider boxing. currently takes up a lot of stack space
 // }
 // #[derive(Debug)]
 // pub struct Unconfigured {
 //     pub controller_id: ControllerId,
 // }
 // #[derive(Debug)]
 // pub struct Configured {
 //     controller_id: ControllerId,
 //     polarities: Vec<Polarity>,
 //     bindings: HashMap<usize, PortBinding>,
 //     protocol_description: Arc<ProtocolD>,
 //     main_component: Vec<u8>,
 //     logger: String,
 // }
 // #[derive(Debug)]
 // pub struct Connected {
 //     native_interface: Vec<(PortId, Polarity)>,
 //     sync_batches: Vec<SyncBatch>,
 //     // controller is cooperatively scheduled with the native application
 //     // (except for transport layer behind Endpoints, which are managed by the OS)
 //     // control flow is passed to the controller during methods on Connector (primarily, connect and sync).
 //     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: PortId,
 //     msg: Msg,
 // }
 // #[derive(Debug)]
 // struct MessengerState {
 //     poll: Poll,
 //     events: Events,
 //     delayed: Vec<ReceivedMsg>,
 //     undelayed: Vec<ReceivedMsg>,
 //     polled_undrained: IndexSet<PortId>,
 // }
 // #[derive(Debug)]
 // struct ChannelIdStream {
 //     controller_id: ControllerId,
 //     next_channel_index: ChannelIndex,
 // }
 // #[derive(Debug)]
 // struct Controller {
 //     protocol_description: Arc<ProtocolD>,
 //     inner: ControllerInner,
 //     ephemeral: ControllerEphemeral,
 //     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>,
 //     port_to_holder: HashMap<PortId, PolyId>,
 // }
 // #[derive(Debug)]
 // struct ControllerFamily {
 //     parent_port: Option<PortId>,
 //     children_ports: Vec<PortId>,
 // }
 // #[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 { port: PortId },
 // }
 // pub(crate) struct MonoPContext<'a> {
 //     inner: &'a mut ControllerInner,
 //     ports: &'a mut HashSet<PortId>,
 //     mono_ps: &'a mut Vec<MonoP>,
 // }
 // pub(crate) struct PolyPContext<'a> {
 //     my_subtree_id: SubtreeId,
 //     inner: &'a mut ControllerInner,
 //     solution_storage: &'a mut SolutionStorage,
 // }
 // impl PolyPContext<'_> {
 //     #[inline(always)]
 //     fn reborrow<'a>(&'a mut self) -> PolyPContext<'a> {
 //         let Self { solution_storage, my_subtree_id, inner } = self;
 //         PolyPContext { solution_storage, my_subtree_id: *my_subtree_id, inner }
 //     }
 // }
 // struct BranchPContext<'m, 'r> {
 //     m_ctx: PolyPContext<'m>,
 //     ports: &'r HashSet<PortId>,
 //     predicate: &'r Predicate,
 //     inbox: &'r HashMap<PortId, Payload>,
 // }
 // #[derive(Default)]
 // pub(crate) struct SolutionStorage {
 //     old_local: HashSet<Predicate>,
 //     new_local: HashSet<Predicate>,
 //     // this pair acts as SubtreeId -> HashSet<Predicate> which is friendlier to iteration
 //     subtree_solutions: Vec<HashSet<Predicate>>,
 //     subtree_id_to_index: HashMap<SubtreeId, usize>,
 // }
 // trait Messengerlike {
 //     fn get_state_mut(&mut self) -> &mut MessengerState;
 //     fn get_endpoint_mut(&mut self, eport: PortId) -> &mut Endpoint;
 //     fn delay(&mut self, received: ReceivedMsg) {
 //         self.get_state_mut().delayed.push(received);
 //     }
 //     fn undelay_all(&mut self) {
 //         let MessengerState { delayed, undelayed, .. } = self.get_state_mut();
 //         undelayed.extend(delayed.drain(..))
 //     }
 //     fn send(&mut self, to: PortId, msg: Msg) -> Result<(), EndpointErr> {
 //         self.get_endpoint_mut(to).send(msg)
 //     }
 //     // attempt to receive a message from one of the endpoints before the deadline
 //     fn recv(&mut self, deadline: Instant) -> Result<Option<ReceivedMsg>, MessengerRecvErr> {
 //         // try get something buffered
 //         if let Some(x) = self.get_state_mut().undelayed.pop() {
 //             return Ok(Some(x));
 //         }
 //         loop {
 //             // polled_undrained may not be empty
 //             while let Some(eport) = self.get_state_mut().polled_undrained.pop() {
 //                 if let Some(msg) = self
 //                     .get_endpoint_mut(eport)
 //                     .recv()
 //                     .map_err(|e| MessengerRecvErr::EndpointErr(eport, e))?
 //                 {
 //                     // this endpoint MAY still have messages! check again in future
 //                     self.get_state_mut().polled_undrained.insert(eport);
 //                     return Ok(Some(ReceivedMsg { recipient: eport, msg }));
 //                 }
 //             }
 //             let state = self.get_state_mut();
 //             match state.poll_events(deadline) {
 //                 Ok(()) => {
 //                     for e in state.events.iter() {
 //                         state.polled_undrained.insert(PortId::from_token(e.token()));
 //                     }
 //                 }
 //                 Err(PollDeadlineErr::PollingFailed) => return Err(MessengerRecvErr::PollingFailed),
 //                 Err(PollDeadlineErr::Timeout) => return Ok(None),
 //             }
 //         }
 //     }
 //     fn recv_blocking(&mut self) -> Result<ReceivedMsg, MessengerRecvErr> {
 //         // try get something buffered
 //         if let Some(x) = self.get_state_mut().undelayed.pop() {
 //             return Ok(x);
 //         }
 //         loop {
 //             // polled_undrained may not be empty
 //             while let Some(eport) = self.get_state_mut().polled_undrained.pop() {
 //                 if let Some(msg) = self
 //                     .get_endpoint_mut(eport)
 //                     .recv()
 //                     .map_err(|e| MessengerRecvErr::EndpointErr(eport, e))?
 //                 {
 //                     // this endpoint MAY still have messages! check again in future
 //                     self.get_state_mut().polled_undrained.insert(eport);
 //                     return Ok(ReceivedMsg { recipient: eport, msg });
 //                 }
 //             }
 //             let state = self.get_state_mut();
 //             state
 //                 .poll
 //                 .poll(&mut state.events, None)
 //                 .map_err(|_| MessengerRecvErr::PollingFailed)?;
 //             for e in state.events.iter() {
 //                 state.polled_undrained.insert(PortId::from_token(e.token()));
 //             }
 //         }
 //     }
 // }
 // /////////////////////////////////
 // impl Debug for SolutionStorage {
 //     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
 //         f.pad("Solutions: [")?;
 //         for (subtree_id, &index) in self.subtree_id_to_index.iter() {
 //             let sols = &self.subtree_solutions[index];
 //             f.write_fmt(format_args!("{:?}: {:?}, ", subtree_id, sols))?;
 //         }
 //         f.pad("]")
 //     }
 // }
 // impl From<EvalErr> for SyncErr {
 //     fn from(e: EvalErr) -> SyncErr {
 //         SyncErr::EvalErr(e)
 //     }
 // }
 // impl From<MessengerRecvErr> for SyncErr {
 //     fn from(e: MessengerRecvErr) -> SyncErr {
 //         SyncErr::MessengerRecvErr(e)
 //     }
 // }
 // impl From<MessengerRecvErr> for ConnectErr {
 //     fn from(e: MessengerRecvErr) -> ConnectErr {
 //         ConnectErr::MessengerRecvErr(e)
 //     }
 // }
 // impl<T> Default for Arena<T> {
 //     fn default() -> Self {
 //         Self { storage: vec![] }
 //     }
 // }
 // impl<T> Arena<T> {
 //     pub fn alloc(&mut self, t: T) -> PortId {
 //         self.storage.push(t);
 //         let l: u32 = self.storage.len().try_into().unwrap();
 //         PortId::from_raw(l - 1u32)
 //     }
 //     pub fn get(&self, key: PortId) -> Option<&T> {
 //         self.storage.get(key.to_raw() as usize)
 //     }
 //     pub fn get_mut(&mut self, key: PortId) -> Option<&mut T> {
 //         self.storage.get_mut(key.to_raw() as usize)
 //     }
 //     pub fn type_convert<X>(self, f: impl FnMut((PortId, T)) -> X) -> Arena<X> {
 //         Arena { storage: self.keyspace().zip(self.storage.into_iter()).map(f).collect() }
 //     }
 //     pub fn iter(&self) -> impl Iterator<Item = (PortId, &T)> {
 //         self.keyspace().zip(self.storage.iter())
 //     }
 //     pub fn len(&self) -> usize {
 //         self.storage.len()
 //     }
 //     pub fn keyspace(&self) -> impl Iterator<Item = PortId> {
 //         (0u32..self.storage.len().try_into().unwrap()).map(PortId::from_raw)
 //     }
 // }
 // impl ChannelIdStream {
 //     fn new(controller_id: ControllerId) -> Self {
 //         Self { controller_id, next_channel_index: 0 }
 //     }
 //     fn next(&mut self) -> ChannelId {
 //         self.next_channel_index += 1;
 //         ChannelId { controller_id: self.controller_id, channel_index: self.next_channel_index - 1 }
 //     }
 // }
 // impl MessengerState {
 //     // 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(())
 //     }
 // }
 // impl From<PollDeadlineErr> for ConnectErr {
 //     fn from(e: PollDeadlineErr) -> ConnectErr {
 //         match e {
 //             PollDeadlineErr::Timeout => ConnectErr::Timeout,
 //             PollDeadlineErr::PollingFailed => ConnectErr::PollingFailed,
 //         }
 //     }
 // }
 // impl std::ops::Not for Polarity {
 //     type Output = Self;
 //     fn not(self) -> Self::Output {
 //         use Polarity::*;
 //         match self {
 //             Putter => Getter,
 //             Getter => Putter,
 //         }
 //     }
 // }
 // impl Predicate {
 //     // returns true IFF self.unify would return Equivalent OR FormerNotLatter
 //     pub fn satisfies(&self, other: &Self) -> bool {
 //         let mut s_it = self.assigned.iter();
 //         let mut s = if let Some(s) = s_it.next() {
 //             s
 //         } else {
 //             return other.assigned.is_empty();
 //         };
 //         for (oid, ob) in other.assigned.iter() {
 //             while s.0 < oid {
 //                 s = if let Some(s) = s_it.next() {
 //                     s
 //                 } else {
 //                     return false;
 //                 };
 //             }
 //             if s.0 > oid || s.1 != ob {
 //                 return false;
 //             }
 //         }
 //         true
 //     }
 //     /// Given self and other, two predicates, return the most general Predicate possible, N
 //     /// such that n.satisfies(self) && n.satisfies(other).
 //     /// If none exists Nonexistant is returned.
 //     /// If the resulting predicate is equivlanet to self, other, or both,
 //     /// FormerNotLatter, LatterNotFormer and Equivalent are returned respectively.
 //     /// otherwise New(N) is returned.
 //     pub fn common_satisfier(&self, other: &Self) -> CommonSatResult {
 //         use CommonSatResult::*;
 //         // iterators over assignments of both predicates. Rely on SORTED ordering of BTreeMap's keys.
 //         let [mut s_it, mut o_it] = [self.assigned.iter(), other.assigned.iter()];
 //         let [mut s, mut o] = [s_it.next(), o_it.next()];
 //         // lists of assignments in self but not other and vice versa.
 //         let [mut s_not_o, mut o_not_s] = [vec![], vec![]];
 //         loop {
 //             match [s, o] {
 //                 [None, None] => break,
 //                 [None, Some(x)] => {
 //                     o_not_s.push(x);
 //                     o_not_s.extend(o_it);
 //                     break;
 //                 }
 //                 [Some(x), None] => {
 //                     s_not_o.push(x);
 //                     s_not_o.extend(s_it);
 //                     break;
 //                 }
 //                 [Some((sid, sb)), Some((oid, ob))] => {
 //                     if sid < oid {
 //                         // o is missing this element
 //                         s_not_o.push((sid, sb));
 //                         s = s_it.next();
 //                     } else if sid > oid {
 //                         // s is missing this element
 //                         o_not_s.push((oid, ob));
 //                         o = o_it.next();
 //                     } else if sb != ob {
 //                         assert_eq!(sid, oid);
 //                         // both predicates assign the variable but differ on the value
 //                         return Nonexistant;
 //                     } else {
 //                         // both predicates assign the variable to the same value
 //                         s = s_it.next();
 //                         o = o_it.next();
 //                     }
 //                 }
 //             }
 //         }
 //         // Observed zero inconsistencies. A unified predicate exists...
 //         match [s_not_o.is_empty(), o_not_s.is_empty()] {
 //             [true, true] => Equivalent,       // ... equivalent to both.
 //             [false, true] => FormerNotLatter, // ... equivalent to self.
 //             [true, false] => LatterNotFormer, // ... equivalent to other.
 //             [false, false] => {
 //                 // ... which is the union of the predicates' assignments but
 //                 //     is equivalent to neither self nor other.
 //                 let mut new = self.clone();
 //                 for (&id, &b) in o_not_s {
 //                     new.assigned.insert(id, b);
 //                 }
 //                 New(new)
 //             }
 //         }
 //     }
 //     pub fn iter_matching(&self, value: bool) -> impl Iterator<Item = ChannelId> + '_ {
 //         self.assigned
 //             .iter()
 //             .filter_map(move |(&channel_id, &b)| if b == value { Some(channel_id) } else { None })
 //     }
 //     pub fn batch_assign_nones(
 //         &mut self,
 //         channel_ids: impl Iterator<Item = ChannelId>,
 //         value: bool,
 //     ) {
 //         for channel_id in channel_ids {
 //             self.assigned.entry(channel_id).or_insert(value);
 //         }
 //     }
 //     pub fn replace_assignment(&mut self, channel_id: ChannelId, value: bool) -> Option<bool> {
 //         self.assigned.insert(channel_id, value)
 //     }
 //     pub fn union_with(&self, other: &Self) -> Option<Self> {
 //         let mut res = self.clone();
 //         for (&channel_id, &assignment_1) in other.assigned.iter() {
 //             match res.assigned.insert(channel_id, assignment_1) {
 //                 Some(assignment_2) if assignment_1 != assignment_2 => return None,
 //                 _ => {}
 //             }
 //         }
 //         Some(res)
 //     }
 //     pub fn query(&self, x: ChannelId) -> Option<bool> {
 //         self.assigned.get(&x).copied()
 //     }
 //     pub fn new_trivial() -> Self {
 //         Self { assigned: Default::default() }
 //     }
 // }
 // #[test]
 // fn pred_sat() {
 //     use maplit::btreemap;
 //     let mut c = ChannelIdStream::new(0);
 //     let ch = std::iter::repeat_with(move || c.next()).take(5).collect::<Vec<_>>();
 //     let p = Predicate::new_trivial();
 //     let p_0t = Predicate { assigned: btreemap! { ch[0] => true } };
 //     let p_0f = Predicate { assigned: btreemap! { ch[0] => false } };
 //     let p_0f_3f = Predicate { assigned: btreemap! { ch[0] => false, ch[3] => false } };
 //     let p_0f_3t = Predicate { assigned: btreemap! { ch[0] => false, ch[3] => true } };
 //     assert!(p.satisfies(&p));
 //     assert!(p_0t.satisfies(&p_0t));
 //     assert!(p_0f.satisfies(&p_0f));
 //     assert!(p_0f_3f.satisfies(&p_0f_3f));
 //     assert!(p_0f_3t.satisfies(&p_0f_3t));
 //     assert!(p_0t.satisfies(&p));
 //     assert!(p_0f.satisfies(&p));
 //     assert!(p_0f_3f.satisfies(&p_0f));
 //     assert!(p_0f_3t.satisfies(&p_0f));
 //     assert!(!p.satisfies(&p_0t));
 //     assert!(!p.satisfies(&p_0f));
 //     assert!(!p_0f.satisfies(&p_0t));
 //     assert!(!p_0t.satisfies(&p_0f));
 //     assert!(!p_0f_3f.satisfies(&p_0f_3t));
 //     assert!(!p_0f_3t.satisfies(&p_0f_3f));
 //     assert!(!p_0t.satisfies(&p_0f_3f));
 //     assert!(!p_0f.satisfies(&p_0f_3f));
 //     assert!(!p_0t.satisfies(&p_0f_3t));
 //     assert!(!p_0f.satisfies(&p_0f_3t));
 // }
 // #[test]
 // fn pred_common_sat() {
 //     use maplit::btreemap;
 //     use CommonSatResult::*;
 //     let mut c = ChannelIdStream::new(0);
 //     let ch = std::iter::repeat_with(move || c.next()).take(5).collect::<Vec<_>>();
 //     let p = Predicate::new_trivial();
 //     let p_0t = Predicate { assigned: btreemap! { ch[0] => true } };
 //     let p_0f = Predicate { assigned: btreemap! { ch[0] => false } };
 //     let p_3f = Predicate { assigned: btreemap! { ch[3] => false } };
 //     let p_0f_3f = Predicate { assigned: btreemap! { ch[0] => false, ch[3] => false } };
 //     let p_0f_3t = Predicate { assigned: btreemap! { ch[0] => false, ch[3] => true } };
 //     assert_eq![p.common_satisfier(&p), Equivalent];
 //     assert_eq![p_0t.common_satisfier(&p_0t), Equivalent];
 //     assert_eq![p.common_satisfier(&p_0t), LatterNotFormer];
 //     assert_eq![p_0t.common_satisfier(&p), FormerNotLatter];
 //     assert_eq![p_0t.common_satisfier(&p_0f), Nonexistant];
 //     assert_eq![p_0f_3t.common_satisfier(&p_0f_3f), Nonexistant];
 //     assert_eq![p_0f_3t.common_satisfier(&p_3f), Nonexistant];
 //     assert_eq![p_3f.common_satisfier(&p_0f_3t), Nonexistant];
 //     assert_eq![p_0f.common_satisfier(&p_3f), New(p_0f_3f)];
 // }

src/runtime/my_tests.rs

➞

Show inline comments

@@ new file 100644 @@
 use crate as reowolf;
 use reowolf::Polarity::*;
 use std::net::SocketAddr;
 use std::{sync::Arc, time::Duration};
 fn next_test_addr() -> SocketAddr {
     use std::{
         net::{Ipv4Addr, SocketAddrV4},
         sync::atomic::{AtomicU16, Ordering::SeqCst},
     };
     static TEST_PORT: AtomicU16 = AtomicU16::new(5_000);
     let port = TEST_PORT.fetch_add(1, SeqCst);
     SocketAddrV4::new(Ipv4Addr::LOCALHOST, port).into()
+}
 lazy_static::lazy_static! {
     static ref MINIMAL_PROTO: Arc<reowolf::ProtocolDescription> =
         { Arc::new(reowolf::ProtocolDescription::parse(b"").unwrap()) };
+}
 #[test]
 fn simple_connector() {
     let c = reowolf::Connector::new_simple(MINIMAL_PROTO.clone(), 0);
     println!("{:#?}", c);
+}
 #[test]
 fn add_port_pair() {
     let mut c = reowolf::Connector::new_simple(MINIMAL_PROTO.clone(), 0);
     let [_, _] = c.add_port_pair();
     let [_, _] = c.add_port_pair();
     println!("{:#?}", c);
+}
 #[test]
 fn add_sync() {
     let mut c = reowolf::Connector::new_simple(MINIMAL_PROTO.clone(), 0);
     let [o, i] = c.add_port_pair();
     c.add_component(b"sync", &[i, o]).unwrap();
     println!("{:#?}", c);
+}
 #[test]
 fn add_net_port() {
     let mut c = reowolf::Connector::new_simple(MINIMAL_PROTO.clone(), 0);
     let sock_addr = next_test_addr();
     let _ = c
         .add_net_port(reowolf::EndpointSetup { polarity: Getter, sock_addr, is_active: false })
         .unwrap();
     let _ = c
         .add_net_port(reowolf::EndpointSetup { polarity: Putter, sock_addr, is_active: true })
         .unwrap();
     println!("{:#?}", c);
+}
 #[test]
 fn trivial_connect() {
     let mut c = reowolf::Connector::new_simple(MINIMAL_PROTO.clone(), 0);
     c.connect(Duration::from_secs(1)).unwrap();
     println!("{:#?}", c);
+}
 #[test]
 fn single_node_connect() {
     let mut c = reowolf::Connector::new_simple(MINIMAL_PROTO.clone(), 0);
     let sock_addr = next_test_addr();
     let _ = c
         .add_net_port(reowolf::EndpointSetup { polarity: Getter, sock_addr, is_active: false })
         .unwrap();
     let _ = c
         .add_net_port(reowolf::EndpointSetup { polarity: Putter, sock_addr, is_active: true })
         .unwrap();
     c.connect(Duration::from_secs(1)).unwrap();
     println!("{:#?}", c);
     c.get_logger().dump_log(&mut std::io::stdout().lock());
+}

src/runtime/serde.rs

➞

Show inline comments

 use crate::common::*;
 use crate::runtime::{
     endpoint::{CommMsg, CommMsgContents, Decision, EndpointInfo, Msg, SetupMsg},
     Predicate,
 };
 use crate::runtime::*;
 use byteorder::{BigEndian, ReadBytesExt, WriteBytesExt};
 use std::io::{ErrorKind::InvalidData, Read, Write};
 pub trait Ser<T>: Write {
     fn ser(&mut self, t: &T) -> Result<(), std::io::Error>;
+}
 pub trait De<T>: Read {
     fn de(&mut self) -> Result<T, std::io::Error>;
+}
 pub struct MonitoredReader<R: Read> {
     bytes: usize,
     r: R,
+}
 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 {
         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)
+    }
+}
 /////////////////////////////////////////
 struct VarLenInt(u64);
 macro_rules! ser_seq {
     ( $w:expr ) => {{
         io::Result::Ok(())
     }};
     ( $w:expr, $first:expr ) => {{
         $w.ser($first)
     }};
     ( $w:expr, $first:expr, $( $x:expr ),+ ) => {{
         $w.ser($first)?;
         ser_seq![$w, $( $x ),*]
     }};
+}
 /////////////////////////////////////////
 impl<W: Write> Ser<PortId> for W {
     fn ser(&mut self, t: &PortId) -> Result<(), std::io::Error> {
         self.ser(&t.controller_id)?;
         self.ser(&VarLenInt(t.port_index as u64))
+    }
+}
 impl<R: Read> De<PortId> for R {
     fn de(&mut self) -> Result<PortId, std::io::Error> {
         Ok(PortId { controller_id: self.de()?, port_index: De::<VarLenInt>::de(self)?.0 as u32 })
+    }
+}
 impl<W: Write> Ser<bool> for W {
     fn ser(&mut self, t: &bool) -> Result<(), std::io::Error> {
         self.ser(&match t {
             true => b'T',
             false => b'F',
         })
+    }
+}
 impl<R: Read> De<bool> for R {
     fn de(&mut self) -> Result<bool, std::io::Error> {
         let b: u8 = self.de()?;
         Ok(match b {
             b'T' => true,
             b'F' => false,
             _ => return Err(InvalidData.into()),
         })
+    }
+}
 impl<W: Write> Ser<u8> for W {
     fn ser(&mut self, t: &u8) -> Result<(), std::io::Error> {
         self.write_u8(*t)
+    }
+}
 impl<R: Read> De<u8> for R {
     fn de(&mut self) -> Result<u8, std::io::Error> {
         self.read_u8()
+    }
+}
 impl<W: Write> Ser<u16> for W {
     fn ser(&mut self, t: &u16) -> Result<(), std::io::Error> {
         self.write_u16::<BigEndian>(*t)
+    }
+}
 impl<R: Read> De<u16> for R {
     fn de(&mut self) -> Result<u16, std::io::Error> {
         self.read_u16::<BigEndian>()
+    }
+}
 impl<W: Write> Ser<u32> for W {
     fn ser(&mut self, t: &u32) -> Result<(), std::io::Error> {
         self.write_u32::<BigEndian>(*t)
+    }
+}
 impl<R: Read> De<u32> for R {
     fn de(&mut self) -> Result<u32, std::io::Error> {
         self.read_u32::<BigEndian>()
+    }
+}
 impl<W: Write> Ser<u64> for W {
     fn ser(&mut self, t: &u64) -> Result<(), std::io::Error> {
         self.write_u64::<BigEndian>(*t)
+    }
+}
 impl<R: Read> De<u64> for R {
     fn de(&mut self) -> Result<u64, std::io::Error> {
         self.read_u64::<BigEndian>()
+    }
+}
 impl<W: Write> Ser<Payload> for W {
     fn ser(&mut self, t: &Payload) -> Result<(), std::io::Error> {
         self.ser(&VarLenInt(t.len() as u64))?;
         for byte in t.as_slice() {
             self.ser(byte)?;
+        }
         Ok(())
+    }
+}
 impl<R: Read> De<Payload> for R {
     fn de(&mut self) -> Result<Payload, std::io::Error> {
         let VarLenInt(len) = self.de()?;
         let mut x = Vec::with_capacity(len as usize);
         for _ in 0..len {
             x.push(self.de()?);
+        }
         Ok(x.into())
+    }
+}
 impl<W: Write> Ser<VarLenInt> for W {
     fn ser(&mut self, t: &VarLenInt) -> Result<(), std::io::Error> {
         integer_encoding::VarIntWriter::write_varint(self, t.0).map(|_| ())
+    }
+}
 impl<R: Read> De<VarLenInt> for R {
     fn de(&mut self) -> Result<VarLenInt, std::io::Error> {
         integer_encoding::VarIntReader::read_varint(self).map(VarLenInt)
+    }
+}
 impl<W: Write> Ser<ChannelId> for W {
     fn ser(&mut self, t: &ChannelId) -> Result<(), std::io::Error> {
         self.ser(&t.controller_id)?;
         self.ser(&VarLenInt(t.channel_index as u64))
+    }
+}
 impl<R: Read> De<ChannelId> for R {
     fn de(&mut self) -> Result<ChannelId, std::io::Error> {
         Ok(ChannelId {
             controller_id: self.de()?,
             channel_index: De::<VarLenInt>::de(self)?.0 as ChannelIndex,
         })
+    }
+}
 impl<W: Write> Ser<Predicate> for W {
     fn ser(&mut self, t: &Predicate) -> Result<(), std::io::Error> {
         self.ser(&VarLenInt(t.assigned.len() as u64))?;
         for (channel_id, boolean) in &t.assigned {
             ser_seq![self, channel_id, boolean]?;
+        }
         Ok(())
+    }
+}
 impl<R: Read> De<Predicate> for R {
     fn de(&mut self) -> Result<Predicate, std::io::Error> {
         let VarLenInt(len) = self.de()?;
-        let mut assigned = BTreeMap::<ChannelId, bool>::default();
+        let mut assigned = BTreeMap::<PortId, bool>::default();
         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',
         })
+    }
+}
 impl<R: Read> De<Polarity> for R {
     fn de(&mut self) -> Result<Polarity, std::io::Error> {
         let b: u8 = self.de()?;
         Ok(match b {
             b'P' => Polarity::Putter,
             b'G' => Polarity::Getter,
             _ => return Err(InvalidData.into()),
         })
+    }
+}
 impl<W: Write> Ser<EndpointInfo> for W {
     fn ser(&mut self, t: &EndpointInfo) -> Result<(), std::io::Error> {
         let EndpointInfo { channel_id, polarity } = t;
         ser_seq![self, channel_id, polarity]
+    }
+}
 impl<R: Read> De<EndpointInfo> for R {
     fn de(&mut self) -> Result<EndpointInfo, std::io::Error> {
         Ok(EndpointInfo { channel_id: self.de()?, polarity: self.de()? })
+    }
+}
 impl<W: Write> Ser<Msg> for W {
     fn ser(&mut self, t: &Msg) -> Result<(), std::io::Error> {
         use {CommMsgContents::*, SetupMsg::*};
         match t {
             Msg::SetupMsg(s) => match s {
                 // [flag, data]
-                ChannelSetup { info } => ser_seq![self, &0u8, info],
+                MyPortInfo { polarity, port } => ser_seq![self, &0u8, polarity, port],
                 LeaderEcho { maybe_leader } => ser_seq![self, &1u8, maybe_leader],
                 LeaderAnnounce { leader } => ser_seq![self, &2u8, leader],
                 YouAreMyParent => ser_seq![self, &3u8],
             },
             Msg::CommMsg(CommMsg { round_index, contents }) => {
                 // [flag, round_num, data]
                 let varlenint = &VarLenInt(*round_index as u64);
                 match contents {
                     SendPayload { payload_predicate, payload } => {
                         ser_seq![self, &4u8, varlenint, payload_predicate, payload]
+                    }
                     Elaborate { partial_oracle } => ser_seq![self, &5u8, varlenint, partial_oracle],
                     Announce { decision } => ser_seq![self, &6u8, varlenint, decision],
                     Failure => ser_seq![self, &7u8, varlenint],
+                }
+            }
+        }
+    }
+}
 impl<R: Read> De<Msg> for R {
     fn de(&mut self) -> Result<Msg, std::io::Error> {
         use {CommMsgContents::*, SetupMsg::*};
         let b: u8 = self.de()?;
         Ok(match b {
 ..=3 => Msg::SetupMsg(match b {
                 // [flag, data]
-u8 => ChannelSetup { info: self.de()? },
+u8 => MyPortInfo { polarity: self.de()?, port: self.de()? },
 u8 => LeaderEcho { maybe_leader: self.de()? },
 u8 => LeaderAnnounce { leader: self.de()? },
 u8 => YouAreMyParent,
                 _ => unreachable!(),
             }),
 ..=7 => {
                 // [flag, round_num, data]
                 let VarLenInt(varlenint) = self.de()?;
                 let contents = match b {
 u8 => SendPayload { payload_predicate: self.de()?, payload: self.de()? },
 u8 => Elaborate { partial_oracle: self.de()? },
 u8 => Announce { decision: self.de()? },
 u8 => Failure,
                     _ => unreachable!(),
                 };
                 Msg::CommMsg(CommMsg { round_index: varlenint as usize, contents })
+            }
             _ => return Err(InvalidData.into()),
         })
+    }
+}

src/runtime/setup.rs

➞

Show inline comments

 use crate::common::*;
 use crate::runtime::{
     actors::{MonoN, MonoP},
     endpoint::*,
     errors::*,
     *,
 };
 #[derive(Debug)]
 enum EndpointExtTodo {
     Finished(EndpointExt),
     ActiveConnecting { addr: SocketAddr, polarity: Polarity, stream: TcpStream },
     ActiveRecving { addr: SocketAddr, polarity: Polarity, endpoint: Endpoint },
     PassiveAccepting { addr: SocketAddr, info: EndpointInfo, listener: TcpListener },
     PassiveConnecting { addr: SocketAddr, info: EndpointInfo, stream: TcpStream },
+}
 ///////////////////// IMPL /////////////////////
 impl Controller {
     // Given port bindings and a protocol config, create a connector with 1 native node
     pub fn connect(
         major: ControllerId,
         main_component: &[u8],
         protocol_description: Arc<ProtocolD>,
         bound_proto_interface: &[(PortBinding, Polarity)],
         logger: &mut String,
         deadline: Instant,
-    ) -> Result<(Self, Vec<(Port, Polarity)>), ConnectErr> {
+    ) -> Result<(Self, Vec<(PortId, Polarity)>), ConnectErr> {
         use ConnectErr::*;
         log!(logger, "CONNECT PHASE START! MY CID={:?} STARTING LOGGER ~", major);
         let mut channel_id_stream = ChannelIdStream::new(major);
         let mut endpoint_ext_todos = Arena::default();
         let mut ports_native = vec![];
         let mut ports_proto = vec![];
         let mut ports_network = vec![];
         let mut native_interface = vec![];
         /*
 .  - allocate an EndpointExtTodo for every native and interface port
             - store all the resulting ports in two portlists for the interfaces of the native and proto components
                 native: [a, c,    f]
                          |  |     |
                          |  |     |
                 proto:  [b, d, e, g]
                                ^todo
                 arena: <A,B,C,D,E,F,G>
         */
         for &(binding, polarity) in bound_proto_interface.iter() {
             match binding {
                 PortBinding::Native => {
                     let channel_id = channel_id_stream.next();
                     let ([port_native, port_proto], native_polarity) = {
                         let [p, g] = Endpoint::new_memory_pair();
                         let mut endpoint_to_port = |endpoint, polarity| {
                             endpoint_ext_todos.alloc(EndpointExtTodo::Finished(EndpointExt {
                                 endpoint,
                                 info: EndpointInfo { polarity, channel_id },
                             }))
                         };
                         let pport = endpoint_to_port(p, Putter);
                         let gport = endpoint_to_port(g, Getter);
                         let port_pair = match polarity {
                             Putter => [gport, pport],
                             Getter => [pport, gport],
                         };
                         (port_pair, !polarity)
                     };
                     native_interface.push((port_native, native_polarity));
                     ports_native.push(port_native);
                     ports_proto.push(port_proto);
+                }
                 PortBinding::Passive(addr) => {
                     let channel_id = channel_id_stream.next();
                     let port_proto = endpoint_ext_todos.alloc(EndpointExtTodo::PassiveAccepting {
                         addr,
                         info: EndpointInfo { polarity, channel_id },
                         listener: TcpListener::bind(&addr).map_err(|_| BindFailed(addr))?,
                     });
                     ports_network.push(port_proto);
                     ports_proto.push(port_proto);
+                }
                 PortBinding::Active(addr) => {
                     let port_proto = endpoint_ext_todos.alloc(EndpointExtTodo::ActiveConnecting {
                         addr,
                         polarity,
                         stream: TcpStream::connect(&addr).unwrap(),
                     });
                     ports_network.push(port_proto);
                     ports_proto.push(port_proto);
+                }
+            }
+        }
         log!(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, logger, endpoint_ext_todos, deadline)?;
         let n_mono = MonoN { ports: ports_native.into_iter().collect(), result: None };
         let p_monos = vec![MonoP {
             state: protocol_description.new_main_component(main_component, &ports_proto),
             ports: ports_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,
             logger,
             &mut endpoint_exts,
             &mut messenger_state,
             ports_network,
             deadline,
         )?;
         log!(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 mut l = String::default();
                 std::mem::swap(&mut l, logger);
+                l
             },
         };
         let controller = Self {
             protocol_description,
             inner,
             ephemeral: Default::default(),
             // round_histories: vec![],
             unrecoverable_error: None,
         };
         Ok((controller, native_interface))
+    }
     // with mio v0.6 attempting to read bytes into a nonempty buffer appears to
     // be the only reliably platform-independent means of testing the connectivity of
     // a mio::TcpStream (see Self::connection_testing_read).
     // as this unavoidably MAY read some crucial payload bytes, we have to be careful
     // to pass these potentially populated buffers into the Endpoint, or bytes may be lost.
     // This is done with Endpoint::from_fresh_stream_and_inbox.
     fn connection_testing_read(stream: &mut TcpStream, inbox: &mut Vec<u8>) -> std::io::Result<()> {
         inbox.clear();
         use std::io::Read;
         match stream.read_to_end(inbox) {
             Ok(0) => unreachable!("Ok(0) on read should return Err instead!"),
             Ok(_) => Ok(()),
             Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => Ok(()),
             Err(e) => Err(e),
+        }
+    }
     // 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 portset 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 (port, t) in endpoint_ext_todos.iter() {
             let token = port.to_token();
             match t {
                 ActiveRecving { .. } | PassiveConnecting { .. } => unreachable!(),
                 Finished(EndpointExt { endpoint, .. }) => {
                     ms.poll.register(endpoint, token, ready_r, edge)
+                }
                 ActiveConnecting { stream, .. } => {
                     to_finish.insert(port);
                     ms.poll.register(stream, token, ready_w, edge)
+                }
                 PassiveAccepting { listener, .. } => {
                     to_finish.insert(port);
                     ms.poll.register(listener, token, ready_r, edge)
+                }
+            }
             .expect("register first");
+        }
         // invariant: every EndpointExtTodo has one thing registered with mio
         // 4. until all in endpoint_ext_todos are Finished variant, handle events
         let mut polled_undrained_later = IndexSet::<_>::default();
         let mut backoff_millis = 10;
         // see Self::connection_testing_read for why we populate Endpoint inboxes here.
         let mut next_inbox = vec![];
         while !to_finish.is_empty() {
             ms.poll_events(deadline).map_err(|e| {
                 log!(logger, "{:03?} timing out", major);
+                e
             })?;
             for event in ms.events.iter() {
                 log!(logger, "event {:#?}", event);
                 let token = event.token();
-                let port = Port::from_token(token);
+                let port = PortId::from_token(token);
                 let entry = endpoint_ext_todos.get_mut(port).unwrap();
                 match entry {
                     Finished(_) => {
                         polled_undrained_later.insert(port);
+                    }
                     PassiveAccepting { addr, listener, .. } => {
                         log!(logger, "{:03?} start PassiveAccepting...", major);
                         assert!(event.readiness().is_readable());
                         let (stream, _peer_addr) =
                             listener.accept().map_err(|_| AcceptFailed(*addr))?;
                         ms.poll.deregister(listener).expect("wer");
                         ms.poll.register(&stream, token, ready_w, edge).expect("3y5");
                         take_mut::take(entry, |e| {
                             assert_let![PassiveAccepting { addr, info, .. } = e => {
                                 PassiveConnecting { addr, info, stream }
                             }]
                         });
                         log!(logger, "{:03?} ... end PassiveAccepting", major);
+                    }
                     PassiveConnecting { addr, stream, .. } => {
                         log!(logger, "{:03?} start PassiveConnecting...", major);
                         assert!(event.readiness().is_writable());
                         if Self::connection_testing_read(stream, &mut next_inbox).is_err() {
                             return Err(PassiveConnectFailed(*addr));
+                        }
                         ms.poll.reregister(stream, token, ready_r, edge).expect("52");
                         let mut res = Ok(());
                         take_mut::take(entry, |e| {
                             let mut inbox = vec![];
                             std::mem::swap(&mut inbox, &mut next_inbox);
                             assert_let![PassiveConnecting { info, stream, .. } = e => {
                                 let mut endpoint = Endpoint::from_fresh_stream_and_inbox(stream, inbox);
                                 let msg = Msg::SetupMsg(SetupMsg::ChannelSetup { info });
                                 res = endpoint.send(msg);
                                 Finished(EndpointExt { info, endpoint })
                             }]
                         });
                         res?;
                         log!(logger, "{:03?} ... end PassiveConnecting", major);
                         assert!(to_finish.remove(&port));
+                    }
                     ActiveConnecting { addr, stream, .. } => {
                         log!(logger, "{:03?} start ActiveConnecting...", major);
                         assert!(event.readiness().is_writable());
                         if Self::connection_testing_read(stream, &mut next_inbox).is_ok() {
                             // connect successful
                             log!(logger, "Connectivity test passed");
                             ms.poll.reregister(stream, token, ready_r, edge).expect("52");
                             take_mut::take(entry, |e| {
                                 let mut inbox = vec![];
                                 std::mem::swap(&mut inbox, &mut next_inbox);
                                 assert_let![ActiveConnecting { stream, polarity, addr } = e => {
                                     let endpoint = Endpoint::from_fresh_stream_and_inbox(stream, inbox);
                                     ActiveRecving { endpoint, polarity, addr }
                                 }]
                             });
                         } else {
                             // connect failure. retry!
                             log!(logger, "CONNECT FAIL");
                             ms.poll.deregister(stream).expect("wt");
                             std::thread::sleep(Duration::from_millis(backoff_millis));
                             backoff_millis = ((backoff_millis as f32) * 1.2) as u64 + 3;
                             let mut new_stream = TcpStream::connect(addr).unwrap();
                             ms.poll.register(&new_stream, token, ready_w, edge).expect("PAC 3");
                             std::mem::swap(stream, &mut new_stream);
+                        }
                         log!(logger, "{:03?} ... end ActiveConnecting", major);
+                    }
                     ActiveRecving { addr, polarity, endpoint } => {
                         log!(logger, "{:03?} start ActiveRecving...", major);
                         assert!(event.readiness().is_readable());
                         'recv_loop: while let Some(msg) = endpoint.recv()? {
                             if let Msg::SetupMsg(SetupMsg::ChannelSetup { info }) = msg {
                                 if info.polarity == *polarity {
                                     return Err(PolarityMatched(*addr));
+                                }
                                 take_mut::take(entry, |e| {
                                     assert_let![ActiveRecving { polarity, endpoint, .. } = e => {
                                         let info = EndpointInfo { polarity, channel_id: info.channel_id };
                                         Finished(EndpointExt { info, endpoint })
                                     }]
                                 });
                                 ms.polled_undrained.insert(port);
                                 assert!(to_finish.remove(&port));
                                 break 'recv_loop;
                             } else {
                                 ms.delayed.push(ReceivedMsg { recipient: port, msg });
+                            }
+                        }
                         log!(logger, "{:03?} ... end ActiveRecving", major);
+                    }
+                }
+            }
+        }
         for port in polled_undrained_later {
             ms.polled_undrained.insert(port);
+        }
         let endpoint_exts = endpoint_ext_todos.type_convert(|(_, todo)| match todo {
             Finished(endpoint_ext) => endpoint_ext,
             _ => unreachable!(),
         });
         Ok((ms, endpoint_exts))
+    }
     fn setup_sink_tree_family(
         major: ControllerId,
         logger: &mut String,
         endpoint_exts: &mut Arena<EndpointExt>,
         messenger_state: &mut MessengerState,
-        neighbors: Vec<Port>,
+        neighbors: Vec<PortId>,
         deadline: Instant,
     ) -> Result<ControllerFamily, ConnectErr> {
         use {ConnectErr::*, Msg::SetupMsg as S, SetupMsg::*};
         log!(logger, "neighbors {:?}", &neighbors);
         let mut messenger = (messenger_state, endpoint_exts);
         impl Messengerlike for (&mut MessengerState, &mut Arena<EndpointExt>) {
             fn get_state_mut(&mut self) -> &mut MessengerState {
                 self.0
+            }
-            fn get_endpoint_mut(&mut self, port: Port) -> &mut Endpoint {
+            fn get_endpoint_mut(&mut self, port: PortId) -> &mut Endpoint {
                 &mut self.1.get_mut(port).expect("OUT OF BOUNDS").endpoint
+            }
+        }
         // 1. broadcast my ID as the first echo. await reply from all in net_portlist
         let echo = S(LeaderEcho { maybe_leader: major });
         let mut awaiting = IndexSet::with_capacity(neighbors.len());
         for &n in neighbors.iter() {
             log!(logger, "{:?}'s initial echo to {:?}, {:?}", major, n, &echo);
             messenger.send(n, echo.clone())?;
             awaiting.insert(n);
+        }
         // 2. Receive incoming replies. whenever a higher-id echo arrives,
         //    adopt it as leader, sender as parent, and reset the await set.
-        let mut parent: Option<Port> = None;
+        let mut parent: Option<PortId> = None;
         let mut my_leader = major;
         messenger.undelay_all();
         'echo_loop: while !awaiting.is_empty() || parent.is_some() {
             let ReceivedMsg { recipient, msg } = messenger.recv(deadline)?.ok_or(Timeout)?;
             log!(logger, "{:?} GOT {:?} {:?}", major, &recipient, &msg);
             match msg {
                 S(LeaderAnnounce { leader }) => {
                     // someone else completed the echo and became leader first!
                     // the sender is my parent
                     parent = Some(recipient);
                     my_leader = leader;
                     awaiting.clear();
                     break 'echo_loop;
+                }
                 S(LeaderEcho { maybe_leader }) => {
                     use Ordering::*;
                     match maybe_leader.cmp(&my_leader) {
                         Less => { /* ignore */ }
                         Equal => {
                             awaiting.remove(&recipient);
                             if awaiting.is_empty() {
                                 if let Some(p) = parent {
                                     // return the echo to my parent
                                     messenger.send(p, S(LeaderEcho { maybe_leader }))?;
                                 } else {
                                     // DECIDE!
                                     break 'echo_loop;
+                                }
+                            }
+                        }
                         Greater => {
                             // join new echo
                             log!(logger, "{:?} setting leader to {:?}", major, recipient);
                             parent = Some(recipient);
                             my_leader = maybe_leader;
                             let echo = S(LeaderEcho { maybe_leader: my_leader });
                             awaiting.clear();
                             if neighbors.len() == 1 {
                                 // immediately reply to parent
                                 log!(
                                     logger,
                                     "{:?} replying echo to parent {:?} immediately",
                                     major,
                                     recipient
                                 );
                                 messenger.send(recipient, echo.clone())?;
                             } else {
                                 for &n in neighbors.iter() {
                                     if n != recipient {
                                         log!(
                                             logger,
                                             "{:?} repeating echo {:?} to {:?}",
                                             major,
                                             &echo,
+                                            n
                                         );
                                         messenger.send(n, echo.clone())?;
                                         awaiting.insert(n);
+                                    }
+                                }
+                            }
+                        }
+                    }
+                }
                 msg => messenger.delay(ReceivedMsg { recipient, msg }),
+            }
+        }
         match parent {
             None => assert_eq!(
                 my_leader, major,
                 "I've got no parent, but I consider {:?} the leader?",
                 my_leader
             ),
             Some(parent) => assert_ne!(
                 my_leader, major,
                 "I have {:?} as parent, but I consider myself ({:?}) the leader?",
                 parent, major
             ),
+        }
         log!(logger, "{:?} DONE WITH ECHO! Leader has cid={:?}", major, my_leader);
         // 3. broadcast leader announcement (except to parent: confirm they are your parent)
         //    in this loop, every node sends 1 message to each neighbor
         let msg_for_non_parents = S(LeaderAnnounce { leader: my_leader });
         for &k in neighbors.iter() {
             let msg =
                 if Some(k) == parent { S(YouAreMyParent) } else { msg_for_non_parents.clone() };
             log!(logger, "{:?} ANNOUNCING to {:?} {:?}", major, k, &msg);
             messenger.send(k, msg)?;
+        }
         // await 1 message from all non-parents
         for &n in neighbors.iter() {
             if Some(n) != parent {
                 awaiting.insert(n);
+            }
+        }
         let mut children = Vec::default();
         messenger.undelay_all();
         while !awaiting.is_empty() {
             let ReceivedMsg { recipient, msg } = messenger.recv(deadline)?.ok_or(Timeout)?;
             match msg {
                 S(YouAreMyParent) => {
                     assert!(awaiting.remove(&recipient));
                     children.push(recipient);
+                }
                 S(SetupMsg::LeaderAnnounce { leader }) => {
                     assert!(awaiting.remove(&recipient));
                     assert!(leader == my_leader);
                     assert!(Some(recipient) != parent);
                     // they wouldn't send me this if they considered me their parent
+                }
                 _ => messenger.delay(ReceivedMsg { recipient, msg }),
+            }
+        }
         Ok(ControllerFamily { parent_port: parent, children_ports: children })
+    }
+}
 impl Messengerlike for Controller {
     fn get_state_mut(&mut self) -> &mut MessengerState {
         &mut self.inner.messenger_state
+    }
-    fn get_endpoint_mut(&mut self, port: Port) -> &mut Endpoint {
+    fn get_endpoint_mut(&mut self, port: PortId) -> &mut Endpoint {
         &mut self.inner.endpoint_exts.get_mut(port).expect("OUT OF BOUNDS").endpoint
+    }
+}

src/runtime/setup2.rs

➞

Show inline comments

@@ new file 100644 @@
 use crate::common::*;
 use crate::runtime::*;
 impl Connector {
     pub fn new_simple(
         proto_description: Arc<ProtocolDescription>,
         controller_id: ControllerId,
     ) -> Self {
         let logger = Box::new(StringLogger::new(controller_id));
         let surplus_sockets = 8;
         Self::new(logger, proto_description, controller_id, surplus_sockets)
+    }
     pub fn new(
         logger: Box<dyn Logger>,
         proto_description: Arc<ProtocolDescription>,
         controller_id: ControllerId,
         surplus_sockets: u16,
     ) -> Self {
         Self {
             logger,
             proto_description,
             id_manager: IdManager::new(controller_id),
             native_ports: Default::default(),
             proto_components: Default::default(),
             outp_to_inp: Default::default(),
             inp_to_route: Default::default(),
             phased: ConnectorPhased::Setup { endpoint_setups: Default::default(), surplus_sockets },
+        }
+    }
     pub fn add_port_pair(&mut self) -> [PortId; 2] {
         let o = self.id_manager.next_port();
         let i = self.id_manager.next_port();
         self.outp_to_inp.insert(o, i);
         self.inp_to_route.insert(i, InpRoute::NativeComponent);
         self.native_ports.insert(o);
         self.native_ports.insert(i);
         log!(self.logger, "Added port pair (out->in) {:?} -> {:?}", o, i);
         [o, i]
+    }
     pub fn add_net_port(&mut self, endpoint_setup: EndpointSetup) -> Result<PortId, ()> {
         match &mut self.phased {
             ConnectorPhased::Setup { endpoint_setups, .. } => {
                 let p = self.id_manager.next_port();
                 self.native_ports.insert(p);
                 log!(self.logger, "Added net port {:?} with info {:?} ", p, &endpoint_setup);
                 endpoint_setups.push((p, endpoint_setup));
                 Ok(p)
+            }
             ConnectorPhased::Communication { .. } => Err(()),
+        }
+    }
     fn check_polarity(&self, port: &PortId) -> Polarity {
         if let ConnectorPhased::Setup { endpoint_setups, .. } = &self.phased {
             for (setup_port, EndpointSetup { polarity, .. }) in endpoint_setups.iter() {
                 if setup_port == port {
                     // special case. this port's polarity isn't reflected by
                     // self.inp_to_route or self.outp_to_inp, because its still not paired to a peer
                     return *polarity;
+                }
+            }
+        }
         if self.outp_to_inp.contains_key(port) {
             Polarity::Putter
         } else {
             assert!(self.inp_to_route.contains_key(port));
             Polarity::Getter
+        }
+    }
     pub fn add_component(
         &mut self,
         identifier: &[u8],
         ports: &[PortId],
     ) -> Result<(), AddComponentError> {
         use AddComponentError::*;
         let polarities = self.proto_description.component_polarities(identifier)?;
         if polarities.len() != ports.len() {
             return Err(WrongNumberOfParamaters { expected: polarities.len() });
+        }
         for (&expected_polarity, port) in polarities.iter().zip(ports.iter()) {
             if !self.native_ports.contains(port) {
                 return Err(UnknownPort(*port));
+            }
             if expected_polarity != self.check_polarity(port) {
                 return Err(WrongPortPolarity { port: *port, expected_polarity });
+            }
+        }
         // ok!
         let state = self.proto_description.new_main_component(identifier, ports);
         let proto_component = ProtoComponent { ports: ports.iter().copied().collect(), state };
         let proto_component_index = self.proto_components.len();
         self.proto_components.push(proto_component);
         for port in ports.iter() {
             if let Polarity::Getter = self.check_polarity(port) {
                 self.inp_to_route
                     .insert(*port, InpRoute::ProtoComponent { index: proto_component_index });
+            }
+        }
         Ok(())
+    }
     pub fn connect(&mut self, timeout: Duration) -> Result<(), ()> {
         match &mut self.phased {
             ConnectorPhased::Communication { .. } => {
                 log!(self.logger, "Call to connecting in connected state");
                 Err(())
+            }
             ConnectorPhased::Setup { endpoint_setups, .. } => {
                 log!(self.logger, "Call to connecting in setup state. Timeout {:?}", timeout);
                 let deadline = Instant::now() + timeout;
                 // connect all endpoints in parallel; send and receive peer ids through ports
                 let (mut endpoint_exts, mut endpoint_poller) = init_endpoints(
                     &mut *self.logger,
                     endpoint_setups,
                     &mut self.inp_to_route,
                     deadline,
                 )?;
                 log!(self.logger, "Successfully connected {} endpoints", endpoint_exts.len());
                 // leader election and tree construction
                 let neighborhood =
                     init_neighborhood(&mut *self.logger, &mut endpoint_exts, &mut endpoint_poller)?;
                 log!(self.logger, "Successfully created neighborhood {:?}", &neighborhood);
                 // TODO session optimization goes here
                 self.phased = ConnectorPhased::Communication {
                     endpoint_poller,
                     endpoint_exts,
                     neighborhood,
                     mem_inbox: Default::default(),
                 };
                 Ok(())
+            }
+        }
+    }
+}
 fn init_endpoints(
     logger: &mut dyn Logger,
     endpoint_setups: &[(PortId, EndpointSetup)],
     inp_to_route: &mut HashMap<PortId, InpRoute>,
     deadline: Instant,
 ) -> Result<(Vec<EndpointExt>, EndpointPoller), ()> {
     use mio07::{
         net::{TcpListener, TcpStream},
         Events, Interest, Poll, Token,
     };
     const BOTH: Interest = Interest::READABLE.add(Interest::WRITABLE);
     struct Todo {
         todo_endpoint: TodoEndpoint,
         endpoint_setup: EndpointSetup,
         local_port: PortId,
         sent_local_port: bool,          // true <-> I've sent my local port
         recv_peer_port: Option<PortId>, // Some(..) <-> I've received my peer's port
+    }
     enum TodoEndpoint {
         Listener(TcpListener),
         Endpoint(Endpoint),
+    }
     fn init(
         token: Token,
         local_port: PortId,
         endpoint_setup: &EndpointSetup,
         poll: &mut Poll,
     ) -> Result<Todo, ()> {
         let todo_endpoint = if endpoint_setup.is_active {
             let mut stream = TcpStream::connect(endpoint_setup.sock_addr).map_err(drop)?;
             poll.registry().register(&mut stream, token, BOTH).unwrap();
             TodoEndpoint::Endpoint(Endpoint { stream, inbox: vec![] })
         } else {
             let mut listener = TcpListener::bind(endpoint_setup.sock_addr).map_err(drop)?;
             poll.registry().register(&mut listener, token, BOTH).unwrap();
             TodoEndpoint::Listener(listener)
         };
         Ok(Todo {
             todo_endpoint,
             endpoint_setup: endpoint_setup.clone(),
             local_port,
             sent_local_port: false,
             recv_peer_port: None,
         })
     };
     ////////////////////////
     let mut ep = EndpointPoller {
         poll: Poll::new().map_err(drop)?,
         events: Events::with_capacity(64),
         undrained_endpoints: Default::default(),
         delayed_inp_messages: Default::default(),
     };
     let mut todos = endpoint_setups
         .iter()
         .enumerate()
         .map(|(index, (local_port, endpoint_setup))| {
             init(Token(index), *local_port, endpoint_setup, &mut ep.poll)
         })
         .collect::<Result<Vec<Todo>, _>>()?;
     let mut unfinished: HashSet<usize> = (0..todos.len()).collect();
     while !unfinished.is_empty() {
         let remaining = deadline.checked_duration_since(Instant::now()).ok_or(())?;
         ep.poll.poll(&mut ep.events, Some(remaining)).map_err(drop)?;
         for event in ep.events.iter() {
             let token = event.token();
             let Token(index) = token;
             let todo: &mut Todo = &mut todos[index];
             if let TodoEndpoint::Listener(listener) = &mut todo.todo_endpoint {
                 let (mut stream, peer_addr) = listener.accept().map_err(drop)?;
                 ep.poll.registry().deregister(listener).unwrap();
                 ep.poll.registry().register(&mut stream, token, BOTH).unwrap();
                 log!(logger, "Endpoint({}) accepted a connection from {:?}", index, peer_addr);
                 let endpoint = Endpoint { stream, inbox: vec![] };
                 todo.todo_endpoint = TodoEndpoint::Endpoint(endpoint);
+            }
             match todo {
                 Todo {
                     todo_endpoint: TodoEndpoint::Endpoint(endpoint),
                     local_port,
                     endpoint_setup,
                     sent_local_port,
                     recv_peer_port,
                 } => {
                     if !unfinished.contains(&index) {
                         continue;
+                    }
                     if event.is_writable() && !*sent_local_port {
                         let msg =
                             MyPortInfo { polarity: endpoint_setup.polarity, port: *local_port };
                         endpoint.send(&msg)?;
                         log!(logger, "endpoint[{}] sent peer info {:?}", index, &msg);
                         *sent_local_port = true;
+                    }
                     if event.is_readable() && recv_peer_port.is_none() {
                         ep.undrained_endpoints.insert(index);
                         if let Some(peer_port_info) =
                             endpoint.try_recv::<MyPortInfo>().map_err(drop)?
+                        {
                             log!(logger, "endpoint[{}] got peer info {:?}", index, peer_port_info);
                             assert!(peer_port_info.polarity != endpoint_setup.polarity);
                             if let Putter = endpoint_setup.polarity {
                                 inp_to_route.insert(*local_port, InpRoute::Endpoint { index });
+                            }
                             *recv_peer_port = Some(peer_port_info.port);
+                        }
+                    }
                     if *sent_local_port && recv_peer_port.is_some() {
                         unfinished.remove(&index);
                         log!(logger, "endpoint[{}] is finished!", index);
+                    }
+                }
                 Todo { todo_endpoint: TodoEndpoint::Listener(_), .. } => unreachable!(),
+            }
+        }
         ep.events.clear();
+    }
     let endpoint_exts = todos
         .into_iter()
         .map(|Todo { todo_endpoint, recv_peer_port, .. }| EndpointExt {
             endpoint: match todo_endpoint {
                 TodoEndpoint::Endpoint(endpoint) => endpoint,
                 TodoEndpoint::Listener(..) => unreachable!(),
             },
             inp_for_emerging_msgs: recv_peer_port.unwrap(),
         })
         .collect();
     Ok((endpoint_exts, ep))
+}
 fn init_neighborhood(
     logger: &mut dyn Logger,
     endpoint_exts: &mut [EndpointExt],
     endpoint_poller: &mut EndpointPoller,
 ) -> Result<Neighborhood, ()> {
     log!(logger, "Time to construct my neighborhood");
     let parent = None;
     let children = Default::default();
     Ok(Neighborhood { parent, children })
+}

Changeset was too big and was cut off... Show full diff anyway

0 comments (0 inline, 0 general)