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

Changeset - c2b16314e873

Parent rev.

Child rev.

[Not reviewed]

0 6 0

Christopher Esterhuyse - 5 years ago 2020-10-28 09:37:42
christopher.esterhuyse@gmail.com

cleanup of unused code

6 files changed with 3 insertions and 32 deletions:

src/common.rs

src/protocol/arena.rs

src/protocol/eval.rs

src/runtime/communication.rs

src/runtime/mod.rs

src/runtime/setup.rs

0 comments (0 inline, 0 general)

src/common.rs

➞

Show inline comments

 ///////////////////// PRELUDE /////////////////////
 pub(crate) use crate::protocol::{ComponentState, ProtocolDescription};
 pub(crate) use crate::runtime::{error::AddComponentError, NonsyncProtoContext, SyncProtoContext};
 pub(crate) use core::{
     cmp::Ordering,
     fmt::{Debug, Formatter},
     hash::Hash,
     ops::Range,
     time::Duration,
 };
 pub(crate) use maplit::hashmap;
 pub(crate) use mio::{
     net::{TcpListener, TcpStream},
     Events, Interest, Poll, Token,
 };
 pub(crate) use std::{
     collections::{BTreeMap, HashMap, HashSet},
     convert::TryInto,
     io::{Read, Write},
     net::SocketAddr,
     sync::Arc,
     time::Instant,
 };
 pub(crate) use Polarity::*;
 pub(crate) trait IdParts {
     fn id_parts(self) -> (ConnectorId, U32Suffix);
+}
 /// Used by various distributed algorithms to identify connectors.
 pub type ConnectorId = u32;
 /// Used in conjunction with the `ConnectorId` type to create identifiers for ports and components
 pub type U32Suffix = u32;
 #[derive(
     Copy, Clone, Eq, PartialEq, Ord, Hash, PartialOrd, serde::Serialize, serde::Deserialize,
 )]
 /// Generalization of a port/component identifier
 #[repr(C)]
 pub struct Id {
     pub(crate) connector_id: ConnectorId,
     pub(crate) u32_suffix: U32Suffix,
+}
 #[derive(Clone, Debug, Default)]
 pub struct U32Stream {
     next: u32,
+}
 /// Identifier of a component in a session
 #[derive(
     Copy, Clone, Eq, PartialEq, Ord, Hash, PartialOrd, serde::Serialize, serde::Deserialize,
 )]
 pub struct ComponentId(Id); // PUB because it can be returned by errors
 /// Identifier of a port in a session
 #[derive(
     Copy, Clone, Eq, PartialEq, Ord, Hash, PartialOrd, serde::Serialize, serde::Deserialize,
 )]
 #[repr(transparent)]
 pub struct PortId(Id);
 /// A safely aliasable heap-allocated payload of message bytes
 #[derive(Default, Eq, PartialEq, Clone, Ord, PartialOrd)]
 pub struct Payload(Arc<Vec<u8>>);
 #[derive(
     Debug, Eq, PartialEq, Clone, Hash, Copy, Ord, PartialOrd, serde::Serialize, serde::Deserialize,
 )]
 /// "Orientation" of a port, determining whether they can send or receive messages with `put` and `get` respectively.
 #[repr(C)]
 pub enum Polarity {
     Putter, // output port (from the perspective of the component)
     Getter, // input port (from the perspective of the component)
+}
 #[derive(
     Debug, Eq, PartialEq, Clone, Hash, Copy, Ord, PartialOrd, serde::Serialize, serde::Deserialize,
 )]
 /// "Orientation" of a transport-layer network endpoint, dictating how it's connection procedure should
 /// be conducted. Corresponds with connect() / accept() familiar to TCP socket programming.
 #[repr(C)]
 pub enum EndpointPolarity {
     Active,  // calls connect()
     Passive, // calls bind() listen() accept()
+}
 #[derive(Debug, Clone)]
 pub(crate) enum NonsyncBlocker {
     Inconsistent,
     ComponentExit,
     SyncBlockStart,
+}
 #[derive(Debug, Clone)]
 pub(crate) enum SyncBlocker {
     Inconsistent,
     SyncBlockEnd,
     CouldntReadMsg(PortId),
     CouldntCheckFiring(PortId),
     PutMsg(PortId, Payload),
     NondetChoice { n: u16 },
+}
 pub(crate) struct DenseDebugHex<'a>(pub &'a [u8]);
 pub(crate) struct DebuggableIter<I: Iterator<Item = T> + Clone, T: Debug>(pub(crate) I);
 ///////////////////// IMPL /////////////////////
 impl IdParts for Id {
     fn id_parts(self) -> (ConnectorId, U32Suffix) {
         (self.connector_id, self.u32_suffix)
+    }
+}
 impl IdParts for PortId {
     fn id_parts(self) -> (ConnectorId, U32Suffix) {
         self.0.id_parts()
+    }
+}
 impl IdParts for ComponentId {
     fn id_parts(self) -> (ConnectorId, U32Suffix) {
         self.0.id_parts()
+    }
+}
 impl U32Stream {
     pub(crate) fn next(&mut self) -> u32 {
         if self.next == u32::MAX {
             panic!("NO NEXT!")
+        }
         self.next += 1;
         self.next - 1
+    }
     pub(crate) fn n_skipped(mut self, n: u32) -> Self {
         self.next = self.next.saturating_add(n);
         self
+    }
+}
 impl From<Id> for PortId {
     fn from(id: Id) -> PortId {
         Self(id)
+    }
+}
 impl From<Id> for ComponentId {
     fn from(id: Id) -> Self {
         Self(id)
+    }
+}
 impl From<&[u8]> for Payload {
     fn from(s: &[u8]) -> Payload {
         Payload(Arc::new(s.to_vec()))
+    }
+}
 impl Payload {
     /// Create a new payload of uninitialized bytes with the given length.
     pub fn new(len: usize) -> Payload {
         let mut v = Vec::with_capacity(len);
         unsafe {
             v.set_len(len);
+        }
         Payload(Arc::new(v))
+    }
     /// Returns the length of the payload's byte sequence
     pub fn len(&self) -> usize {
         self.0.len()
+    }
     /// Allows shared reading of the payload's contents
     pub fn as_slice(&self) -> &[u8] {
         &self.0
+    }
     /// Allows mutation of the payload's contents.
     /// Results in a deep copy in the event this payload is aliased.
     pub fn as_mut_vec(&mut self) -> &mut Vec<u8> {
         Arc::make_mut(&mut self.0)
+    }
     /// Modifies this payload, concatenating the given immutable payload's contents.
     /// Results in a deep copy in the event this payload is aliased.
     pub fn concatenate_with(&mut self, other: &Self) {
         let bytes = other.as_slice().iter().copied();
         let me = self.as_mut_vec();
         me.extend(bytes);
+    }
+}
 impl serde::Serialize for Payload {
     fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::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) -> Result<Self, D::Error>
     where
         D: serde::Deserializer<'de>,
+    {
         let inner: Vec<u8> = Vec::deserialize(deserializer)?;
         Ok(Self(Arc::new(inner)))
+    }
+}
 impl From<Vec<u8>> for Payload {
     fn from(s: Vec<u8>) -> Self {
         Self(s.into())
+    }
+}
 impl Debug for PortId {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         let (a, b) = self.id_parts();
         write!(f, "pid{}_{}", a, b)
+    }
+}
 impl Debug for ComponentId {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         let (a, b) = self.id_parts();
         write!(f, "cid{}_{}", a, b)
+    }
+}
 impl Debug for Payload {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         write!(f, "Payload[{:?}]", DenseDebugHex(self.as_slice()))
+    }
+}
 impl std::ops::Not for Polarity {
     type Output = Self;
     fn not(self) -> Self::Output {
         use Polarity::*;
         match self {
             Putter => Getter,
             Getter => Putter,
+        }
+    }
+}
 impl Debug for DenseDebugHex<'_> {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         for b in self.0 {
             write!(f, "{:02X?}", b)?;
+        }
         Ok(())
+    }
+}
 impl<I: Iterator<Item = T> + Clone, T: Debug> Debug for DebuggableIter<I, T> {
     fn fmt(&self, f: &mut Formatter<'_>) -> Result<(), std::fmt::Error> {
         f.debug_list().entries(self.0.clone()).finish()
+    }
+}

src/protocol/arena.rs

➞

Show inline comments

 use crate::common::*;
 use core::hash::Hash;
 use core::marker::PhantomData;
 #[derive(serde::Serialize, serde::Deserialize)]
 pub struct Id<T> {
     index: u32,
     _phantom: PhantomData<T>,
+}
 #[derive(Debug, serde::Serialize, serde::Deserialize)]
 pub struct Arena<T> {
+pub(crate) 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);
+    }
+}
 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

 use std::collections::HashMap;
 use std::fmt;
 use std::fmt::{Debug, Display, Formatter};
 use std::{i16, i32, i64, i8};
 use crate::common::*;
 use crate::protocol::ast::*;
 // use crate::protocol::inputsource::*;
 // use crate::protocol::parser::*;
 use crate::protocol::EvalContext;
 const MAX_RECURSION: usize = 1024;
+// const MAX_RECURSION: usize = 1024;
 const BYTE_MIN: i64 = i8::MIN as i64;
 const BYTE_MAX: i64 = i8::MAX as i64;
 const SHORT_MIN: i64 = i16::MIN as i64;
 const SHORT_MAX: i64 = i16::MAX as i64;
 const INT_MIN: i64 = i32::MIN as i64;
 const INT_MAX: i64 = i32::MAX as i64;
 const MESSAGE_MAX_LENGTH: i64 = SHORT_MAX;
 const ONE: Value = Value::Byte(ByteValue(1));
 trait ValueImpl {
     fn exact_type(&self) -> Type;
     fn is_type_compatible(&self, t: &Type) -> bool;
+}
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 pub enum Value {
     Input(InputValue),
     Output(OutputValue),
     Message(MessageValue),
     Boolean(BooleanValue),
     Byte(ByteValue),
     Short(ShortValue),
     Int(IntValue),
     Long(LongValue),
     InputArray(InputArrayValue),
     OutputArray(OutputArrayValue),
     MessageArray(MessageArrayValue),
     BooleanArray(BooleanArrayValue),
     ByteArray(ByteArrayValue),
     ShortArray(ShortArrayValue),
     IntArray(IntArrayValue),
     LongArray(LongArrayValue),
+}
 impl Value {
     pub fn receive_message(buffer: &Payload) -> Value {
         Value::Message(MessageValue(Some(buffer.clone())))
+    }
     fn create_message(length: Value) -> Value {
         match length {
             Value::Byte(_) | Value::Short(_) | Value::Int(_) | Value::Long(_) => {
                 let length: i64 = i64::from(length);
                 if length < 0 || length > MESSAGE_MAX_LENGTH {
                     // Only messages within the expected length are allowed
                     Value::Message(MessageValue(None))
                 } else {
                     Value::Message(MessageValue(Some(Payload::new(length as usize))))
+                }
+            }
             _ => unimplemented!(),
+        }
+    }
     fn from_constant(constant: &Constant) -> Value {
         match constant {
             Constant::Null => Value::Message(MessageValue(None)),
             Constant::True => Value::Boolean(BooleanValue(true)),
             Constant::False => Value::Boolean(BooleanValue(false)),
             Constant::Integer(data) => {
                 // Convert raw ASCII data to UTF-8 string
                 let raw = String::from_utf8_lossy(data);
                 let val = raw.parse::<i64>().unwrap();
                 if val >= BYTE_MIN && val <= BYTE_MAX {
                     Value::Byte(ByteValue(val as i8))
                 } else if val >= SHORT_MIN && val <= SHORT_MAX {
                     Value::Short(ShortValue(val as i16))
                 } else if val >= INT_MIN && val <= INT_MAX {
                     Value::Int(IntValue(val as i32))
                 } else {
                     Value::Long(LongValue(val))
+                }
+            }
             Constant::Character(_data) => unimplemented!(),
+        }
+    }
     fn set(&mut self, index: &Value, value: &Value) -> Option<Value> {
         // The index must be of integer type, and non-negative
         let the_index: usize;
         match index {
             Value::Byte(_) | Value::Short(_) | Value::Int(_) | Value::Long(_) => {
                 let index = i64::from(index);
                 if index < 0 || index >= MESSAGE_MAX_LENGTH {
                     // It is inconsistent to update out of bounds
                     return None;
+                }
                 the_index = index.try_into().unwrap();
+            }
             _ => unreachable!(),
+        }
         // The subject must be either a message or an array
         // And the value and the subject must be compatible
         match (self, value) {
             (Value::Message(MessageValue(None)), _) => {
                 // It is inconsistent to update the null message
                 None
+            }
             (Value::Message(MessageValue(Some(payload))), Value::Byte(ByteValue(b))) => {
                 if *b < 0 {
                     // It is inconsistent to update with a negative value
                     return None;
+                }
                 if let Some(slot) = payload.as_mut_vec().get_mut(the_index) {
                     *slot = (*b).try_into().unwrap();
                     Some(value.clone())
                 } else {
                     // It is inconsistent to update out of bounds
                     None
+                }
+            }
             (Value::Message(MessageValue(Some(payload))), Value::Short(ShortValue(b))) => {
                 if *b < 0 || *b > BYTE_MAX as i16 {
                     // It is inconsistent to update with a negative value or a too large value
                     return None;
+                }
                 if let Some(slot) = payload.as_mut_vec().get_mut(the_index) {
                     *slot = (*b).try_into().unwrap();
                     Some(value.clone())
                 } else {
                     // It is inconsistent to update out of bounds
                     None
+                }
+            }
             (Value::InputArray(_), Value::Input(_)) => todo!(),
             (Value::OutputArray(_), Value::Output(_)) => todo!(),
             (Value::MessageArray(_), Value::Message(_)) => todo!(),
             (Value::BooleanArray(_), Value::Boolean(_)) => todo!(),
             (Value::ByteArray(_), Value::Byte(_)) => todo!(),
             (Value::ShortArray(_), Value::Short(_)) => todo!(),
             (Value::IntArray(_), Value::Int(_)) => todo!(),
             (Value::LongArray(_), Value::Long(_)) => todo!(),
             _ => unreachable!(),
+        }
+    }
     fn get(&self, index: &Value) -> Option<Value> {
         // The index must be of integer type, and non-negative
         let the_index: usize;
         match index {
             Value::Byte(_) | Value::Short(_) | Value::Int(_) | Value::Long(_) => {
                 let index = i64::from(index);
                 if index < 0 || index >= MESSAGE_MAX_LENGTH {
                     // It is inconsistent to update out of bounds
                     return None;
+                }
                 the_index = index.try_into().unwrap();
+            }
             _ => unreachable!(),
+        }
         // The subject must be either a message or an array
         match self {
             Value::Message(MessageValue(None)) => {
                 // It is inconsistent to read from the null message
                 None
+            }
             Value::Message(MessageValue(Some(payload))) => {
                 if let Some(slot) = payload.as_slice().get(the_index) {
                     Some(Value::Short(ShortValue((*slot).try_into().unwrap())))
                 } else {
                     // It is inconsistent to update out of bounds
                     None
+                }
+            }
             _ => panic!("Can only get from port value"),
+        }
+    }
     fn length(&self) -> Option<Value> {
         // The subject must be either a message or an array
         match self {
             Value::Message(MessageValue(None)) => {
                 // It is inconsistent to get length from the null message
                 None
+            }
             Value::Message(MessageValue(Some(buffer))) => {
                 Some(Value::Int(IntValue((buffer.len()).try_into().unwrap())))
+            }
             Value::InputArray(InputArrayValue(vec)) => {
                 Some(Value::Int(IntValue((vec.len()).try_into().unwrap())))
+            }
             Value::OutputArray(OutputArrayValue(vec)) => {
                 Some(Value::Int(IntValue((vec.len()).try_into().unwrap())))
+            }
             Value::MessageArray(MessageArrayValue(vec)) => {
                 Some(Value::Int(IntValue((vec.len()).try_into().unwrap())))
+            }
             Value::BooleanArray(BooleanArrayValue(vec)) => {
                 Some(Value::Int(IntValue((vec.len()).try_into().unwrap())))
+            }
             Value::ByteArray(ByteArrayValue(vec)) => {
                 Some(Value::Int(IntValue((vec.len()).try_into().unwrap())))
+            }
             Value::ShortArray(ShortArrayValue(vec)) => {
                 Some(Value::Int(IntValue((vec.len()).try_into().unwrap())))
+            }
             Value::IntArray(IntArrayValue(vec)) => {
                 Some(Value::Int(IntValue((vec.len()).try_into().unwrap())))
+            }
             Value::LongArray(LongArrayValue(vec)) => {
                 Some(Value::Int(IntValue((vec.len()).try_into().unwrap())))
+            }
             _ => unreachable!(),
+        }
+    }
     fn plus(&self, other: &Value) -> Value {
         match (self, other) {
             (Value::Byte(ByteValue(s)), Value::Byte(ByteValue(o))) => {
                 Value::Byte(ByteValue(*s + *o))
+            }
             (Value::Byte(ByteValue(s)), Value::Short(ShortValue(o))) => {
                 Value::Short(ShortValue(*s as i16 + *o))
+            }
             (Value::Byte(ByteValue(s)), Value::Int(IntValue(o))) => {
                 Value::Int(IntValue(*s as i32 + *o))
+            }
             (Value::Byte(ByteValue(s)), Value::Long(LongValue(o))) => {
                 Value::Long(LongValue(*s as i64 + *o))
+            }
             (Value::Short(ShortValue(s)), Value::Byte(ByteValue(o))) => {
                 Value::Short(ShortValue(*s + *o as i16))
+            }
             (Value::Short(ShortValue(s)), Value::Short(ShortValue(o))) => {
                 Value::Short(ShortValue(*s + *o))
+            }
             (Value::Short(ShortValue(s)), Value::Int(IntValue(o))) => {
                 Value::Int(IntValue(*s as i32 + *o))
+            }
             (Value::Short(ShortValue(s)), Value::Long(LongValue(o))) => {
                 Value::Long(LongValue(*s as i64 + *o))
+            }
             (Value::Int(IntValue(s)), Value::Byte(ByteValue(o))) => {
                 Value::Int(IntValue(*s + *o as i32))
+            }
             (Value::Int(IntValue(s)), Value::Short(ShortValue(o))) => {
                 Value::Int(IntValue(*s + *o as i32))
+            }
             (Value::Int(IntValue(s)), Value::Int(IntValue(o))) => Value::Int(IntValue(*s + *o)),
             (Value::Int(IntValue(s)), Value::Long(LongValue(o))) => {
                 Value::Long(LongValue(*s as i64 + *o))
+            }
             (Value::Long(LongValue(s)), Value::Byte(ByteValue(o))) => {
                 Value::Long(LongValue(*s + *o as i64))
+            }
             (Value::Long(LongValue(s)), Value::Short(ShortValue(o))) => {
                 Value::Long(LongValue(*s + *o as i64))
+            }
             (Value::Long(LongValue(s)), Value::Int(IntValue(o))) => {
                 Value::Long(LongValue(*s + *o as i64))
+            }
             (Value::Long(LongValue(s)), Value::Long(LongValue(o))) => {
                 Value::Long(LongValue(*s + *o))
+            }
             (Value::Message(MessageValue(s)), Value::Message(MessageValue(o))) => {
                 let payload = if let [Some(s), Some(o)] = [s, o] {
                     let mut payload = s.clone();
                     payload.concatenate_with(o);
                     Some(payload)
                 } else {
                     None
                 };
                 Value::Message(MessageValue(payload))
+            }
             _ => unimplemented!(),
+        }
+    }
     fn minus(&self, other: &Value) -> Value {
         match (self, other) {
             (Value::Byte(ByteValue(s)), Value::Byte(ByteValue(o))) => {
                 Value::Byte(ByteValue(*s - *o))
+            }
             (Value::Byte(ByteValue(s)), Value::Short(ShortValue(o))) => {
                 Value::Short(ShortValue(*s as i16 - *o))
+            }
             (Value::Byte(ByteValue(s)), Value::Int(IntValue(o))) => {
                 Value::Int(IntValue(*s as i32 - *o))
+            }
             (Value::Byte(ByteValue(s)), Value::Long(LongValue(o))) => {
                 Value::Long(LongValue(*s as i64 - *o))
+            }
             (Value::Short(ShortValue(s)), Value::Byte(ByteValue(o))) => {
                 Value::Short(ShortValue(*s - *o as i16))
+            }
             (Value::Short(ShortValue(s)), Value::Short(ShortValue(o))) => {
                 Value::Short(ShortValue(*s - *o))
+            }
             (Value::Short(ShortValue(s)), Value::Int(IntValue(o))) => {
                 Value::Int(IntValue(*s as i32 - *o))
+            }
             (Value::Short(ShortValue(s)), Value::Long(LongValue(o))) => {
                 Value::Long(LongValue(*s as i64 - *o))
+            }
             (Value::Int(IntValue(s)), Value::Byte(ByteValue(o))) => {
                 Value::Int(IntValue(*s - *o as i32))
+            }
             (Value::Int(IntValue(s)), Value::Short(ShortValue(o))) => {
                 Value::Int(IntValue(*s - *o as i32))
+            }
             (Value::Int(IntValue(s)), Value::Int(IntValue(o))) => Value::Int(IntValue(*s - *o)),
             (Value::Int(IntValue(s)), Value::Long(LongValue(o))) => {
                 Value::Long(LongValue(*s as i64 - *o))
+            }
             (Value::Long(LongValue(s)), Value::Byte(ByteValue(o))) => {
                 Value::Long(LongValue(*s - *o as i64))
+            }
             (Value::Long(LongValue(s)), Value::Short(ShortValue(o))) => {
                 Value::Long(LongValue(*s - *o as i64))
+            }
             (Value::Long(LongValue(s)), Value::Int(IntValue(o))) => {
                 Value::Long(LongValue(*s - *o as i64))
+            }
             (Value::Long(LongValue(s)), Value::Long(LongValue(o))) => {
                 Value::Long(LongValue(*s - *o))
+            }
             _ => unimplemented!(),
+        }
+    }
     fn modulus(&self, other: &Value) -> Value {
         match (self, other) {
             (Value::Byte(ByteValue(s)), Value::Byte(ByteValue(o))) => {
                 Value::Byte(ByteValue(*s % *o))
+            }
             (Value::Byte(ByteValue(s)), Value::Short(ShortValue(o))) => {
                 Value::Short(ShortValue(*s as i16 % *o))
+            }
             (Value::Byte(ByteValue(s)), Value::Int(IntValue(o))) => {
                 Value::Int(IntValue(*s as i32 % *o))
+            }
             (Value::Byte(ByteValue(s)), Value::Long(LongValue(o))) => {
                 Value::Long(LongValue(*s as i64 % *o))
+            }
             (Value::Short(ShortValue(s)), Value::Byte(ByteValue(o))) => {
                 Value::Short(ShortValue(*s % *o as i16))
+            }
             (Value::Short(ShortValue(s)), Value::Short(ShortValue(o))) => {
                 Value::Short(ShortValue(*s % *o))
+            }
             (Value::Short(ShortValue(s)), Value::Int(IntValue(o))) => {
                 Value::Int(IntValue(*s as i32 % *o))
+            }
             (Value::Short(ShortValue(s)), Value::Long(LongValue(o))) => {
                 Value::Long(LongValue(*s as i64 % *o))
+            }
             (Value::Int(IntValue(s)), Value::Byte(ByteValue(o))) => {
                 Value::Int(IntValue(*s % *o as i32))
+            }
             (Value::Int(IntValue(s)), Value::Short(ShortValue(o))) => {
                 Value::Int(IntValue(*s % *o as i32))
+            }
             (Value::Int(IntValue(s)), Value::Int(IntValue(o))) => Value::Int(IntValue(*s % *o)),
             (Value::Int(IntValue(s)), Value::Long(LongValue(o))) => {
                 Value::Long(LongValue(*s as i64 % *o))
+            }
             (Value::Long(LongValue(s)), Value::Byte(ByteValue(o))) => {
                 Value::Long(LongValue(*s % *o as i64))
+            }
             (Value::Long(LongValue(s)), Value::Short(ShortValue(o))) => {
                 Value::Long(LongValue(*s % *o as i64))
+            }
             (Value::Long(LongValue(s)), Value::Int(IntValue(o))) => {
                 Value::Long(LongValue(*s % *o as i64))
+            }
             (Value::Long(LongValue(s)), Value::Long(LongValue(o))) => {
                 Value::Long(LongValue(*s % *o))
+            }
             _ => unimplemented!(),
+        }
+    }
     fn eq(&self, other: &Value) -> Value {
         match (self, other) {
             (Value::Byte(ByteValue(s)), Value::Byte(ByteValue(o))) => {
                 Value::Boolean(BooleanValue(*s == *o))
+            }
             (Value::Byte(ByteValue(s)), Value::Short(ShortValue(o))) => {
                 Value::Boolean(BooleanValue(*s as i16 == *o))
+            }
             (Value::Byte(ByteValue(s)), Value::Int(IntValue(o))) => {
                 Value::Boolean(BooleanValue(*s as i32 == *o))
+            }
             (Value::Byte(ByteValue(s)), Value::Long(LongValue(o))) => {
                 Value::Boolean(BooleanValue(*s as i64 == *o))
+            }
             (Value::Short(ShortValue(s)), Value::Byte(ByteValue(o))) => {
                 Value::Boolean(BooleanValue(*s == *o as i16))
+            }
             (Value::Short(ShortValue(s)), Value::Short(ShortValue(o))) => {

src/runtime/communication.rs

➞

Show inline comments

@@ @@ -714,740 +714,716 @@ impl Connector { @@
                 match comm_ctrl_msg {
                     CommCtrlMsg::Suggest { suggestion } => {
                         // We receive the solution of another connector (part of the decision process)
                         // (only accept this through a child endpoint)
                         if comm.neighborhood.children.contains(&net_index) {
                             match suggestion {
                                 Decision::Success(predicate) => {
                                     // child solution contributes to local solution
                                     log!(cu.logger(), "Child provided solution {:?}", &predicate);
                                     let subtree_id = SubtreeId::NetEndpoint { index: net_index };
                                     rctx.solution_storage.submit_and_digest_subtree_solution(
                                         cu, subtree_id, predicate,
                                     );
+                                }
                                 Decision::Failure => {
                                     // Someone timed out! propagate this to parent or decide
                                     match comm.neighborhood.parent {
                                         None => {
                                             log!(cu.logger(), "I decide on my child's failure");
                                             break 'undecided Ok(Decision::Failure);
+                                        }
                                         Some(parent) => {
                                             log!(cu.logger(), "Forwarding failure through my parent endpoint {:?}", parent);
                                             if already_requested_failure.replace_with_true() {
                                                 Self::request_failure(cu, comm, parent)?
                                             } else {
                                                 log!(cu.logger(), "Already requested failure");
+                                            }
+                                        }
+                                    }
+                                }
+                            }
                         } else {
                             // Unreachable if all connectors are playing by the rules.
                             // Silently ignored instead of causing panic to make the
                             // runtime more robust against network fuzz
                             log!(
                                 cu.logger(),
                                 "Discarding suggestion {:?} from non-child endpoint idx {:?}",
                                 &suggestion,
                                 net_index
                             );
+                        }
+                    }
                     CommCtrlMsg::Announce { decision } => {
                         // Apparently this round is over! A decision has been reached
                         if Some(net_index) == comm.neighborhood.parent {
                             // We accept the decision because it comes from our parent.
                             // end this loop, and and the synchronous round
                             return Ok(decision);
                         } else {
                             // Again, unreachable if all connectors are playing by the rules
                             log!(
                                 cu.logger(),
                                 "Discarding announcement {:?} from non-parent endpoint idx {:?}",
                                 &decision,
                                 net_index
                             );
+                        }
+                    }
+                }
+            }
             log!(cu.logger(), "Endpoint msg recv done");
+        }
+    }
     // Send a failure request to my parent in the consensus tree
     fn request_failure(
         cu: &mut impl CuUndecided,
         comm: &mut ConnectorCommunication,
         parent: usize,
     ) -> Result<(), UnrecoverableSyncError> {
         log!(cu.logger(), "Forwarding to my parent {:?}", parent);
         let suggestion = Decision::Failure;
         let msg = Msg::CommMsg(CommMsg {
             round_index: comm.round_index,
             contents: CommMsgContents::CommCtrl(CommCtrlMsg::Suggest { suggestion }),
         });
         comm.endpoint_manager.send_to_comms(parent, &msg)
+    }
+}
 impl NativeBranch {
     fn is_ended(&self) -> bool {
         self.to_get.is_empty()
+    }
+}
 impl BranchingNative {
     // Feed the given payload to the native component
     // May result in discovering new component solutions,
     // or fork speculative branches if the message's predicate
     // is MORE SPECIFIC than the branches of the native
     fn feed_msg(
         &mut self,
         cu: &mut impl CuUndecided,
         rctx: &mut RoundCtx,
         getter: PortId,
         send_payload_msg: &SendPayloadMsg,
         bn_temp: MapTempGuard<'_, Predicate, NativeBranch>,
     ) {
         log!(cu.logger(), "feeding native getter {:?} {:?}", getter, &send_payload_msg);
         assert_eq!(Getter, rctx.ips.port_info.map.get(&getter).unwrap().polarity);
         let mut draining = bn_temp;
         let finished = &mut self.branches;
         std::mem::swap(draining.0, finished);
         // Visit all native's branches, and feed those whose current predicates are
         // consistent with that of the received message.
         for (predicate, mut branch) in draining.drain() {
             log!(cu.logger(), "visiting native branch {:?} with {:?}", &branch, &predicate);
             let var = rctx.ips.port_info.spec_var_for(getter);
             if predicate.query(var) != Some(SpecVal::FIRING) {
                 // optimization. Don't bother trying this branch,
                 // because the resulting branch would have an inconsistent predicate.
                 // the existing branch asserts the getter port is SILENT
                 log!(
                     cu.logger(),
                     "skipping branch with {:?} that doesn't want the message (fastpath)",
                     &predicate
                 );
                 Self::insert_branch_merging(finished, predicate, branch);
                 continue;
+            }
             // Define a little helper closure over `rctx`
             // for feeding the given branch this new payload,
             // and submitting any resulting solutions
             let mut feed_branch = |branch: &mut NativeBranch, predicate: &Predicate| {
                 // This branch notes the getter port as "gotten"
                 branch.to_get.remove(&getter);
                 if let Some(was) = branch.gotten.insert(getter, send_payload_msg.payload.clone()) {
                     // Sanity check. Payload mapping (Predicate,Port) should be unique each round
                     assert_eq!(&was, &send_payload_msg.payload);
+                }
                 if branch.is_ended() {
                     // That was the last message the branch was awaiting!
                     // Submitting new component solution.
                     log!(
                         cu.logger(),
                         "new native solution with {:?} is_ended() with gotten {:?}",
                         &predicate,
                         &branch.gotten
                     );
                     let subtree_id = SubtreeId::LocalComponent(cu.native_component_id());
                     rctx.solution_storage.submit_and_digest_subtree_solution(
                         cu,
                         subtree_id,
                         predicate.clone(),
                     );
                 } else {
                     // This branch still has ports awaiting their messages
                     log!(
                         cu.logger(),
                         "Fed native {:?} still has to_get {:?}",
                         &predicate,
                         &branch.to_get
                     );
+                }
             };
             use AssignmentUnionResult as Aur;
             match predicate.assignment_union(&send_payload_msg.predicate) {
                 Aur::Nonexistant => {
                     // The predicates of this branch and the payload are incompatible
                     // retain this branch as-is
                     log!(
                         cu.logger(),
                         "skipping branch with {:?} that doesn't want the message (slowpath)",
                         &predicate
                     );
                     Self::insert_branch_merging(finished, predicate, branch);
+                }
                 Aur::Equivalent | Aur::FormerNotLatter => {
                     // The branch's existing predicate "covers" (is at least as specific)
                     // as that of the payload. Can feed this branch the message without altering
                     // the branch predicate.
                     feed_branch(&mut branch, &predicate);
                     log!(cu.logger(), "branch pred covers it! Accept the msg");
                     Self::insert_branch_merging(finished, predicate, branch);
+                }
                 Aur::LatterNotFormer => {
                     // The predicates of branch and payload are compatible,
                     // but that of the payload is strictly more specific than that of the latter.
                     // FORK the branch, feed the fork the message, and give it the payload's predicate.
                     let mut branch2 = branch.clone();
                     let predicate2 = send_payload_msg.predicate.clone();
                     feed_branch(&mut branch2, &predicate2);
                     log!(
                         cu.logger(),
                         "payload pred {:?} covers branch pred {:?}",
                         &predicate2,
                         &predicate
                     );
                     Self::insert_branch_merging(finished, predicate, branch);
                     Self::insert_branch_merging(finished, predicate2, branch2);
+                }
                 Aur::New(predicate2) => {
                     // The predicates of branch and payload are compatible,
                     // but their union is some new predicate (both preds assign something new).
                     // FORK the branch, feed the fork the message, and give it the new predicate.
                     let mut branch2 = branch.clone();
                     feed_branch(&mut branch2, &predicate2);
                     log!(
                         cu.logger(),
                         "new subsuming pred created {:?}. forking and feeding",
                         &predicate2
                     );
                     Self::insert_branch_merging(finished, predicate, branch);
                     Self::insert_branch_merging(finished, predicate2, branch2);
+                }
+            }
+        }
+    }
     // Insert a new speculate branch into the given storage,
     // MERGING it with an existing branch if their predicate keys clash.
     fn insert_branch_merging(
         branches: &mut HashMap<Predicate, NativeBranch>,
         predicate: Predicate,
         mut branch: NativeBranch,
     ) {
         let e = branches.entry(predicate);
         use std::collections::hash_map::Entry;
         match e {
             Entry::Vacant(ev) => {
                 // no existing branch present. We insert it no problem. (The most common case)
                 ev.insert(branch);
+            }
             Entry::Occupied(mut eo) => {
                 // Oh dear, there is already a branch with this predicate.
                 // Rather than choosing either branch, we MERGE them.
                 // This means taking the UNION of their .gotten and the INTERSECTION of their .to_get
                 let old = eo.get_mut();
                 for (k, v) in branch.gotten.drain() {
                     if old.gotten.insert(k, v).is_none() {
                         // added a gotten element in `branch` not already in `old`
                         old.to_get.remove(&k);
+                    }
+                }
+            }
+        }
+    }
     // Given the predicate for the round's solution, collapse this
     // branching native to an ended branch whose predicate is consistent with it.
     // return as `RoundEndedNative` the result of a native completing successful round
     fn collapse_with(
         self,
         logger: &mut dyn Logger,
         solution_predicate: &Predicate,
     ) -> RoundEndedNative {
         log!(
             logger,
             "Collapsing native with {} branch preds {:?}",
             self.branches.len(),
             self.branches.keys()
         );
         for (branch_predicate, branch) in self.branches {
             log!(
                 logger,
                 "Considering native branch {:?} with to_get {:?} gotten {:?}",
                 &branch_predicate,
                 &branch.to_get,
                 &branch.gotten
             );
             if branch.is_ended() && branch_predicate.assigns_subset(solution_predicate) {
                 let NativeBranch { index, gotten, .. } = branch;
                 log!(logger, "Collapsed native has gotten {:?}", &gotten);
                 return RoundEndedNative { batch_index: index, gotten };
+            }
+        }
         log!(logger, "Native had no branches matching pred {:?}", solution_predicate);
         panic!("Native had no branches matching pred {:?}", solution_predicate);
+    }
+}
 impl BranchingProtoComponent {
     // Create a singleton-branch branching protocol component as
     // speculation begins, with the given protocol state.
     fn initial(state: ComponentState) -> Self {
         let branch = ProtoComponentBranch { state, inner: Default::default(), ended: false };
         Self { branches: hashmap! { Predicate::default() => branch } }
+    }
     // run all the given branches (cd.input) to their SyncBlocker,
     // populating cd.output by cyclically draining "input" -> "cd."input" / cd.output.
     // (to prevent concurrent r/w of one structure, we realize "input" as cd.input for reading and cd.swap for writing)
     // This procedure might lose branches, and it might create new branches.
     fn drain_branches_to_blocked(
         cd: CyclicDrainer<Predicate, ProtoComponentBranch>,
         cu: &mut impl CuUndecided,
         rctx: &mut RoundCtx,
         proto_component_id: ComponentId,
     ) -> Result<(), UnrecoverableSyncError> {
         // let CyclicDrainer { input, swap, output } = cd;
         while !cd.input.is_empty() {
             'branch_iter: for (mut predicate, mut branch) in cd.input.drain() {
                 let mut ctx = SyncProtoContext {
                     rctx,
                     predicate: &predicate,
                     branch_inner: &mut branch.inner,
                 };
                 // Run this component's state to the next syncblocker for handling
                 let blocker = branch.state.sync_run(&mut ctx, cu.proto_description());
                 log!(
                     cu.logger(),
                     "Proto component with id {:?} branch with pred {:?} hit blocker {:?}",
                     proto_component_id,
                     &predicate,
                     &blocker,
                 );
                 use SyncBlocker as B;
                 match blocker {
                     B::Inconsistent => drop((predicate, branch)), // EXPLICIT inconsistency
                     B::CouldntReadMsg(port) => {
                         // sanity check: `CouldntReadMsg` returned IFF the message is unavailable
                         assert!(!branch.inner.inbox.contains_key(&port));
                         // This branch hit a proper blocker: progress awaits the receipt of some message. Exit the cycle.
                         Self::insert_branch_merging(cd.output, predicate, branch);
+                    }
                     B::CouldntCheckFiring(port) => {
                         // sanity check: `CouldntCheckFiring` returned IFF the variable is speculatively assigned
                         let var = rctx.ips.port_info.spec_var_for(port);
                         assert!(predicate.query(var).is_none());
                         // speculate on the two possible values of `var`. Schedule both branches to be rerun.
                         Self::insert_branch_merging(
                             cd.swap,
                             predicate.clone().inserted(var, SpecVal::SILENT),
                             branch.clone(),
                         );
                         Self::insert_branch_merging(
                             cd.swap,
                             predicate.inserted(var, SpecVal::FIRING),
                             branch,
                         );
+                    }
                     B::PutMsg(putter, payload) => {
                         // sanity check: The given port indeed has `Putter` polarity
                         assert_eq!(Putter, rctx.ips.port_info.map.get(&putter).unwrap().polarity);
                         // assign FIRING to this port's associated firing variable
                         let var = rctx.ips.port_info.spec_var_for(putter);
                         let was = predicate.assigned.insert(var, SpecVal::FIRING);
                         if was == Some(SpecVal::SILENT) {
                             // Discard the branch, as it clearly has contradictory requirements for this value.
                             log!(cu.logger(), "Proto component {:?} tried to PUT on port {:?} when pred said var {:?}==Some(false). inconsistent!",
                             proto_component_id, putter, var);
                             drop((predicate, branch));
                         } else {
                             // Note that this port has put this round,
                             // and assert that this isn't its 2nd time putting this round (otheriwse PDL programming error)
                             assert!(branch.inner.did_put_or_get.insert(putter));
                             log!(cu.logger(), "Proto component {:?} with pred {:?} putting payload {:?} on port {:?} (using var {:?})",
                             proto_component_id, &predicate, &payload, putter, var);
                             // Send the given payload (by buffering it).
                             let msg = SendPayloadMsg { predicate: predicate.clone(), payload };
                             rctx.putter_push(cu, putter, msg);
                             // Branch can still make progress. Schedule to be rerun
                             Self::insert_branch_merging(cd.swap, predicate, branch);
+                        }
+                    }
                     B::SyncBlockEnd => {
                         // This branch reached the end of it's synchronous block
                         // assign all variables of owned ports that DIDN'T fire to SILENT
                         for port in rctx.ips.port_info.ports_owned_by(proto_component_id) {
                             let var = rctx.ips.port_info.spec_var_for(*port);
                             let actually_exchanged = branch.inner.did_put_or_get.contains(port);
                             let val = *predicate.assigned.entry(var).or_insert(SpecVal::SILENT);
                             let speculated_to_fire = val == SpecVal::FIRING;
                             if actually_exchanged != speculated_to_fire {
                                 log!(cu.logger(), "Inconsistent wrt. port {:?} var {:?} val {:?} actually_exchanged={}, speculated_to_fire={}",
                                 port, var, val, actually_exchanged, speculated_to_fire);
                                 // IMPLICIT inconsistency
                                 drop((predicate, branch));
                                 continue 'branch_iter;
+                            }
+                        }
                         // submit solution for this component
                         let subtree_id = SubtreeId::LocalComponent(proto_component_id);
                         rctx.solution_storage.submit_and_digest_subtree_solution(
                             cu,
                             subtree_id,
                             predicate.clone(),
                         );
                         branch.ended = true;
                         // This branch exits the cyclic drain
                         Self::insert_branch_merging(cd.output, predicate, branch);
+                    }
                     B::NondetChoice { n } => {
                         // This branch requested the creation of a new n-way nondeterministic
                         // fork of the branch with a fresh speculative variable.
                         // ... allocate a new speculative variable
                         let var = rctx.spec_var_stream.next();
                         // ... and for n distinct values, create a new forked branch,
                         // and schedule them to be rerun through the cyclic drain.
                         for val in SpecVal::iter_domain().take(n as usize) {
                             let predicate_n = predicate.clone().inserted(var, val);
                             let mut branch_n = branch.clone();
                             branch_n.inner.untaken_choice = Some(val.0);
                             Self::insert_branch_merging(cd.swap, predicate_n, branch_n);
+                        }
+                    }
+                }
+            }
             std::mem::swap(cd.input, cd.swap);
+        }
         Ok(())
+    }
     // Feed this branching protocol component the given message, and
     // then run all branches until they are once again blocked.
     fn feed_msg(
         &mut self,
         cu: &mut impl CuUndecided,
         rctx: &mut RoundCtx,
         proto_component_id: ComponentId,
         getter: PortId,
         send_payload_msg: &SendPayloadMsg,
         pcb_temps: MapTempsGuard<'_, Predicate, ProtoComponentBranch>,
     ) -> Result<(), UnrecoverableSyncError> {
         log!(
             cu.logger(),
             "feeding proto component {:?} getter {:?} {:?}",
             proto_component_id,
             getter,
             &send_payload_msg
         );
         let (mut unblocked, pcb_temps) = pcb_temps.split_first_mut();
         let (mut blocked, pcb_temps) = pcb_temps.split_first_mut();
         // partition drain from self.branches -> {unblocked, blocked} (not cyclic)
         log!(cu.logger(), "visiting {} blocked branches...", self.branches.len());
         for (predicate, mut branch) in self.branches.drain() {
             if branch.ended {
                 log!(cu.logger(), "Skipping ended branch with {:?}", &predicate);
                 Self::insert_branch_merging(&mut blocked, predicate, branch);
                 continue;
+            }
             use AssignmentUnionResult as Aur;
             log!(cu.logger(), "visiting branch with pred {:?}", &predicate);
             // We give each branch a chance to receive this message,
             // those that do are maybe UNBLOCKED, and all others remain BLOCKED.
             match predicate.assignment_union(&send_payload_msg.predicate) {
                 Aur::Nonexistant => {
                     // this branch does not receive the message. categorize into blocked.
                     log!(cu.logger(), "skipping branch");
                     Self::insert_branch_merging(&mut blocked, predicate, branch);
+                }
                 Aur::Equivalent | Aur::FormerNotLatter => {
                     // retain the existing predicate, but add this payload
                     log!(cu.logger(), "feeding this branch without altering its predicate");
                     branch.feed_msg(getter, send_payload_msg.payload.clone());
                     // this branch does receive the message. categorize into unblocked.
                     Self::insert_branch_merging(&mut unblocked, predicate, branch);
+                }
                 Aur::LatterNotFormer => {
                     // fork branch, give fork the message and payload predicate. original branch untouched
                     log!(cu.logger(), "Forking this branch, giving it the predicate of the msg");
                     let mut branch2 = branch.clone();
                     let predicate2 = send_payload_msg.predicate.clone();
                     branch2.feed_msg(getter, send_payload_msg.payload.clone());
                     // the branch that receives the message is unblocked, the original one is blocked
                     Self::insert_branch_merging(&mut blocked, predicate, branch);
                     Self::insert_branch_merging(&mut unblocked, predicate2, branch2);
+                }
                 Aur::New(predicate2) => {
                     // fork branch, give fork the message and the new predicate. original branch untouched
                     log!(cu.logger(), "Forking this branch with new predicate {:?}", &predicate2);
                     let mut branch2 = branch.clone();
                     branch2.feed_msg(getter, send_payload_msg.payload.clone());
                     // the branch that receives the message is unblocked, the original one is blocked
                     Self::insert_branch_merging(&mut blocked, predicate, branch);
                     Self::insert_branch_merging(&mut unblocked, predicate2, branch2);
+                }
+            }
+        }
         log!(cu.logger(), "blocked {:?} unblocked {:?}", blocked.len(), unblocked.len());
         // drain from unblocked --> blocked
         let (swap, _pcb_temps) = pcb_temps.split_first_mut(); // peel off ONE temp storage map
         let cd = CyclicDrainer { input: unblocked.0, swap: swap.0, output: blocked.0 };
         BranchingProtoComponent::drain_branches_to_blocked(cd, cu, rctx, proto_component_id)?;
         // swap the blocked branches back
         std::mem::swap(blocked.0, &mut self.branches);
         log!(cu.logger(), "component settles down with branches: {:?}", self.branches.keys());
         Ok(())
+    }
     // Insert a new speculate branch into the given storage,
     // MERGING it with an existing branch if their predicate keys clash.
     fn insert_branch_merging(
         branches: &mut HashMap<Predicate, ProtoComponentBranch>,
         predicate: Predicate,
         mut branch: ProtoComponentBranch,
     ) {
         let e = branches.entry(predicate);
         use std::collections::hash_map::Entry;
         match e {
             Entry::Vacant(ev) => {
                 // no existing branch present. We insert it no problem. (The most common case)
                 ev.insert(branch);
+            }
             Entry::Occupied(mut eo) => {
                 // Oh dear, there is already a branch with this predicate.
                 // Rather than choosing either branch, we MERGE them.
                 // This means keeping the existing one in-place, and giving it the UNION of the inboxes
                 let old = eo.get_mut();
                 for (k, v) in branch.inner.inbox.drain() {
                     old.inner.inbox.insert(k, v);
+                }
+            }
+        }
+    }
     // Given the predicate for the round's solution, collapse this
     // branching native to an ended branch whose predicate is consistent with it.
     fn collapse_with(
         self,
         logger: &mut dyn Logger,
         solution_predicate: &Predicate,
     ) -> ComponentState {
         let BranchingProtoComponent { branches } = self;
         for (branch_predicate, branch) in branches {
             if branch.ended && branch_predicate.assigns_subset(solution_predicate) {
                 let ProtoComponentBranch { state, .. } = branch;
                 return state;
+            }
+        }
         log!(logger, "ProtoComponent had no branches matching pred {:?}", solution_predicate);
         panic!("ProtoComponent had no branches matching pred {:?}", solution_predicate);
+    }
+}
 impl ProtoComponentBranch {
     // Feed this branch received message.
     // It's safe to receive the same message repeatedly,
     // but if we receive a message with different contents,
     // it's a sign something has gone wrong! keys of type (port, round, predicate)
     // should always map to at most one message value!
     fn feed_msg(&mut self, getter: PortId, payload: Payload) {
         let e = self.inner.inbox.entry(getter);
         use std::collections::hash_map::Entry;
         match e {
             Entry::Vacant(ev) => {
                 // new message
                 ev.insert(payload);
+            }
             Entry::Occupied(eo) => {
                 // redundant recv. can happen as a result of a
                 // component A having two branches X and Y related by
                 assert_eq!(eo.get(), &payload);
+            }
+        }
+    }
+}
 impl SolutionStorage {
     // Create a new solution storage, to manage the local solutions for
     // this connector and all of it's children (subtrees) in the solution tree.
     fn new(subtree_ids: impl Iterator<Item = SubtreeId>) -> Self {
         // For easy iteration, we store this SubtreeId => {Predicate}
         // structure instead as a pair of structures: a vector of predicate sets,
         // and a subtree_id-to-index lookup map
         let mut subtree_id_to_index: HashMap<SubtreeId, usize> = Default::default();
         let mut subtree_solutions = vec![];
         for id in subtree_ids {
             subtree_id_to_index.insert(id, subtree_solutions.len());
             subtree_solutions.push(Default::default())
+        }
         // new_local U old_local represents the solutions of this connector itself:
         // namely, those that can be created from the union of one element from each child's solution set.
         // The difference between new and old is that new stores those NOT YET sent over the network
         // to this connector's parent in the solution tree.
         // invariant: old_local and new_local have an empty intersection
         Self {
             subtree_solutions,
             subtree_id_to_index,
             old_local: Default::default(),
             new_local: Default::default(),
+        }
+    }
     // drain old_local to new_local, visiting all new additions to old_local
     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| {
             // rely on invariant: empty intersection between old and new local sets
             assert!(old_local.insert(local.clone()));
             local
         })
+    }
     // insert a solution for the given subtree ID,
     // AND update new_local to include any solutions that become
     // possible as a result of this new addition
     pub(crate) fn submit_and_digest_subtree_solution(
         &mut self,
         cu: &mut impl CuUndecided,
         subtree_id: SubtreeId,
         predicate: Predicate,
     ) {
         log!(cu.logger(), "++ new component solution {:?} {:?}", subtree_id, &predicate);
         let Self { subtree_solutions, new_local, old_local, subtree_id_to_index } = self;
         let index = subtree_id_to_index[&subtree_id];
         let was_new = subtree_solutions[index].insert(predicate.clone());
         if was_new {
             // This is a newly-added solution! update new_local
             // consider ALL consistent combinations of one element from each solution set
             // to our right or left in the solution-set vector
             // but with THIS PARTICULAR predicate from our own index.
             let left = 0..index;
             let right = (index + 1)..subtree_solutions.len();
             // iterator over SETS of solutions, one for every component except `subtree_id` (me)
             let set_visitor = left.chain(right).map(|index| &subtree_solutions[index]);
             // Recursively enumerate all solutions matching the description above,
             Self::elaborate_into_new_local_rec(cu, predicate, set_visitor, old_local, new_local);
+        }
+    }
     // Recursively build local solutions for this connector,
     // see `submit_and_digest_subtree_solution`
     fn elaborate_into_new_local_rec<'a, 'b>(
         cu: &mut impl CuUndecided,
         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 recursively creating combined solutions
             for pred in set.iter() {
                 if let Some(elaborated) = pred.union_with(&partial) {
                     Self::elaborate_into_new_local_rec(
                         cu,
                         elaborated,
                         set_visitor.clone(),
                         old_local,
                         new_local,
+                    )
+                }
+            }
         } else {
             // recursive stop condition. This is a solution for this connector...
             if !old_local.contains(&partial) {
                 // ... and it hasn't been found before
                 log!(cu.logger(), "storing NEW LOCAL SOLUTION {:?}", &partial);
                 new_local.insert(partial);
+            }
+        }
+    }
+}
 impl NonsyncProtoContext<'_> {
     // Facilitates callback from the component to the connector runtime,
     // creating a new component and changing the given port's ownership to that
     // of the new component.
     pub(crate) fn new_component(&mut self, moved_ports: HashSet<PortId>, state: ComponentState) {
         // Sanity check! The moved ports are owned by this component to begin with
         for port in moved_ports.iter() {
             assert_eq!(self.proto_component_id, self.ips.port_info.map.get(port).unwrap().owner);
+        }
         // Create the new component, and schedule it to be run
         let new_cid = self.ips.id_manager.new_component_id();
         log!(
             self.logger,
             "Component {:?} added new component {:?} with state {:?}, moving ports {:?}",
             self.proto_component_id,
             new_cid,
             &state,
             &moved_ports
         );
         self.unrun_components.push((new_cid, state));
         // Update the ownership of the moved ports
         for port in moved_ports.iter() {
             self.ips.port_info.map.get_mut(port).unwrap().owner = new_cid;
+        }
         if let Some(set) = self.ips.port_info.owned.get_mut(&self.proto_component_id) {
             set.retain(|x| !moved_ports.contains(x));
+        }
         self.ips.port_info.owned.insert(new_cid, moved_ports.clone());
+    }
     // Facilitates callback from the component to the connector runtime,
     // creating a new port-pair connected by an memory channel
     pub(crate) fn new_port_pair(&mut self) -> [PortId; 2] {
         // adds two new associated ports, related to each other, and exposed to the proto component
         let mut new_cid_fn = || self.ips.id_manager.new_port_id();
         let [o, i] = [new_cid_fn(), new_cid_fn()];
         self.ips.port_info.map.insert(
             o,
             PortInfo {
                 route: Route::LocalComponent,
                 peer: Some(i),
                 polarity: Putter,
                 owner: self.proto_component_id,
             },
         );
         self.ips.port_info.map.insert(
             i,
             PortInfo {
                 route: Route::LocalComponent,
                 peer: Some(o),
                 polarity: Getter,
                 owner: self.proto_component_id,
             },
         );
         self.ips
             .port_info
             .owned
             .entry(self.proto_component_id)
             .or_default()
             .extend([o, i].iter().copied());
         log!(
             self.logger,
             "Component {:?} port pair (out->in) {:?} -> {:?}",
             self.proto_component_id,
             o,
+            i
         );
         [o, i]
+    }
+}
 impl SyncProtoContext<'_> {
     // The component calls the runtime back, inspecting whether it's associated
     // preidcate has already determined a (speculative) value for the given port's firing variable.
     pub(crate) fn is_firing(&mut self, port: PortId) -> Option<bool> {
         let var = self.rctx.ips.port_info.spec_var_for(port);
         self.predicate.query(var).map(SpecVal::is_firing)
+    }
     // The component calls the runtime back, trying to inspect a port's message
     pub(crate) fn read_msg(&mut self, port: PortId) -> Option<&Payload> {
         let maybe_msg = self.branch_inner.inbox.get(&port);
         if maybe_msg.is_some() {
             // Make a note that this component has received
             // this port's message 1+ times this round
             self.branch_inner.did_put_or_get.insert(port);
+        }
         maybe_msg
+    }
     // NOT CURRENTLY USED
     // Once this component has injected a new nondeterministic branch with
     // SyncBlocker::NondetChoice, this is how the component retrieves it.
     // (Two step process necessary to get around mutable access rules,
     //  as injection of the nondeterministic choice modifies the
     //  branch predicate, forks the branch, etc.)
     pub(crate) fn take_choice(&mut self) -> Option<u16> {
         self.branch_inner.untaken_choice.take()
+    }
+}

src/runtime/mod.rs

➞

Show inline comments

 /// cbindgen:ignore
 mod communication;
 /// cbindgen:ignore
 mod endpoints;
 pub mod error;
 /// cbindgen:ignore
 mod logging;
 /// cbindgen:ignore
 mod setup;
 #[cfg(test)]
 mod tests;
 use crate::common::*;
 use error::*;
 use mio::net::UdpSocket;
 /// The interface between the user's application and a communication session,
 /// in which the application plays the part of a (native) component. This structure provides the application
 /// with functionality available to all components: the ability to add new channels (port pairs), and to
 /// instantiate new components whose definitions are defined in the connector's configured protocol
 /// description. Native components have the additional ability to add `dangling' ports backed by local/remote
 /// IP addresses, to be coupled with a counterpart once the connector's setup is completed by `connect`.
 /// This allows sets of applications to cooperate in constructing shared sessions that span the network.
 #[derive(Debug)]
 pub struct Connector {
     unphased: ConnectorUnphased,
     phased: ConnectorPhased,
+}
 /// Characterizes a type which can write lines of logging text.
 /// The implementations provided in the `logging` module are likely to be sufficient,
 /// but for added flexibility, users are able to implement their own loggers for use
 /// by connectors.
 pub trait Logger: Debug + Send + Sync {
     fn line_writer(&mut self) -> Option<&mut dyn std::io::Write>;
+}
 /// A logger that appends the logged strings to a growing byte buffer
 #[derive(Debug)]
 pub struct VecLogger(ConnectorId, Vec<u8>);
 /// A trivial logger that always returns None, such that no logging information is ever written.
 #[derive(Debug)]
 pub struct DummyLogger;
 /// A logger that writes the logged lines to a given file.
 #[derive(Debug)]
 pub struct FileLogger(ConnectorId, std::fs::File);
 // Interface between protocol state and the connector runtime BEFORE all components
 // ave begun their branching speculation. See ComponentState::nonsync_run.
 pub(crate) struct NonsyncProtoContext<'a> {
     ips: &'a mut IdAndPortState,
     logger: &'a mut dyn Logger,
     unrun_components: &'a mut Vec<(ComponentId, ComponentState)>, // lives for Nonsync phase
     proto_component_id: ComponentId,                              // KEY in id->component map
+}
 // Interface between protocol state and the connector runtime AFTER all components
 // have begun their branching speculation. See ComponentState::sync_run.
 pub(crate) struct SyncProtoContext<'a> {
     rctx: &'a RoundCtx,
     branch_inner: &'a mut ProtoComponentBranchInner, // sub-structure of component branch
     predicate: &'a Predicate,                        // KEY in pred->branch map
+}
 // The data coupled with a particular protocol component branch, but crucially omitting
 // the `ComponentState` such that this may be passed by reference to the state with separate
 // access control.
 #[derive(Default, Debug, Clone)]
 struct ProtoComponentBranchInner {
     untaken_choice: Option<u16>,
     did_put_or_get: HashSet<PortId>,
     inbox: HashMap<PortId, Payload>,
+}
 // A speculative variable that lives for the duration of the synchronous round.
 // Each is assigned a value in domain `SpecVal`.
 #[derive(
     Copy, Clone, Eq, PartialEq, Ord, Hash, PartialOrd, serde::Serialize, serde::Deserialize,
 )]
 struct SpecVar(PortId);
 // The codomain of SpecVal. Has two associated constants for values FIRING and SILENT,
 // but may also enumerate many more values to facilitate finer-grained nondeterministic branching.
 #[derive(
     Copy, Clone, Eq, PartialEq, Ord, Hash, PartialOrd, serde::Serialize, serde::Deserialize,
 )]
 struct SpecVal(u16);
 // Data associated with a successful synchronous round, retained afterwards such that the
 // native component can freely reflect on how it went, reading the messages received at their
 // inputs, and reflecting on which of their connector's synchronous batches succeeded.
 #[derive(Debug)]
 struct RoundEndedNative {
     batch_index: usize,
     gotten: HashMap<PortId, Payload>,
+}
 // Implementation of a set in terms of a vector (optimized for reading, not writing)
 #[derive(Default)]
 struct VecSet<T: std::cmp::Ord> {
     // invariant: ordered, deduplicated
     vec: Vec<T>,
+}
 // Allows a connector to remember how to forward payloads towards the component that
 // owns their destination port. `LocalComponent` corresponds with messages for components
 // managed by the connector itself (hinting for it to look it up in a local structure),
 // whereas the other variants direct the connector to forward the messages over the network.
 #[derive(Debug, Clone, Copy, Eq, PartialEq, Hash, serde::Serialize, serde::Deserialize)]
 enum Route {
     LocalComponent,
     NetEndpoint { index: usize },
     UdpEndpoint { index: usize },
+}
 // The outcome of a synchronous round, representing the distributed consensus.
 // In the success case, the attached predicate encodes a row in the session's trace table.
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 enum Decision {
     Failure, // some connector timed out!
     Success(Predicate),
+}
 // The type of control messages exchanged between connectors over the network
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 enum Msg {
     SetupMsg(SetupMsg),
     CommMsg(CommMsg),
+}
 // Control messages exchanged during the setup phase only
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 enum SetupMsg {
     MyPortInfo(MyPortInfo),
     LeaderWave { wave_leader: ConnectorId },
     LeaderAnnounce { tree_leader: ConnectorId },
     YouAreMyParent,
     SessionGather { unoptimized_map: HashMap<ConnectorId, SessionInfo> },
     SessionScatter { optimized_map: HashMap<ConnectorId, SessionInfo> },
+}
 // A data structure encoding the state of a connector, passed around
 // during the session optimization procedure.
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 struct SessionInfo {
     serde_proto_description: SerdeProtocolDescription,
     port_info: PortInfoMap,
     endpoint_incoming_to_getter: Vec<PortId>,
     proto_components: HashMap<ComponentId, ComponentState>,
+}
 // Newtype wrapper for an Arc<ProtocolDescription>,
 // such that it can be (de)serialized for transmission over the network.
 #[derive(Debug, Clone)]
 struct SerdeProtocolDescription(Arc<ProtocolDescription>);
 // Control message particular to the communication phase.
 // as such, it's annotated with a round_index
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 struct CommMsg {
     round_index: usize,
     contents: CommMsgContents,
+}
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 enum CommMsgContents {
     SendPayload(SendPayloadMsg),
     CommCtrl(CommCtrlMsg),
+}
 // Connector <-> connector control messages for use in the communication phase
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 enum CommCtrlMsg {
     Suggest { suggestion: Decision }, // child->parent
     Announce { decision: Decision },  // parent->child
+}
 // Speculative payload message, communicating the value for the given
 // port's message predecated on the given speculative variable assignments.
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 struct SendPayloadMsg {
     predicate: Predicate,
     payload: Payload,
+}
 // Return result of `Predicate::assignment_union`, communicating the contents
 // of the predicate which represents the (consistent) union of their mappings,
 // if it exists (no variable mapped distinctly by the input predicates)
 #[derive(Debug, PartialEq)]
 enum AssignmentUnionResult {
     FormerNotLatter,
     LatterNotFormer,
     Equivalent,
     New(Predicate),
     Nonexistant,
+}
 // One of two endpoints for a control channel with a connector on either end.
 // The underlying transport is TCP, so we use an inbox buffer to allow
 // discrete payload receipt.
 struct NetEndpoint {
     inbox: Vec<u8>,
     stream: TcpStream,
+}
 // Datastructure used during the setup phase representing a NetEndpoint TO BE SETUP
 #[derive(Debug, Clone)]
 struct NetEndpointSetup {
     getter_for_incoming: PortId,
     sock_addr: SocketAddr,
     endpoint_polarity: EndpointPolarity,
+}
 // Datastructure used during the setup phase representing a UdpEndpoint TO BE SETUP
 #[derive(Debug, Clone)]
 struct UdpEndpointSetup {
     getter_for_incoming: PortId,
     local_addr: SocketAddr,
     peer_addr: SocketAddr,
+}
 // NetEndpoint annotated with the ID of the port that receives payload
 // messages received through the endpoint. This approach assumes that NetEndpoints
 // DO NOT multiplex port->port channels, and so a mapping such as this is possible.
 // As a result, the messages themselves don't need to carry the PortID with them.
 #[derive(Debug)]
 struct NetEndpointExt {
     net_endpoint: NetEndpoint,
     getter_for_incoming: PortId,
+}
 // Endpoint for a "raw" UDP endpoint. Corresponds to the "Udp Mediator Component"
 // described in the literature.
 // It acts as an endpoint by receiving messages via the poller etc. (managed by EndpointManager),
 // It acts as a native component by managing a (speculative) set of payload messages (an outbox,
 //  protecting the peer on the other side of the network).
 #[derive(Debug)]
 struct UdpEndpointExt {
     sock: UdpSocket, // already bound and connected
     received_this_round: bool,
     outgoing_payloads: HashMap<Predicate, Payload>,
     getter_for_incoming: PortId,
+}
 // Meta-data for the connector: its role in the consensus tree.
 #[derive(Debug)]
 struct Neighborhood {
     parent: Option<usize>,
     children: VecSet<usize>,
+}
 // Manages the connector's ID, and manages allocations for connector/port IDs.
 #[derive(Debug, Clone)]
 struct IdManager {
     connector_id: ConnectorId,
     port_suffix_stream: U32Stream,
     component_suffix_stream: U32Stream,
+}
 // Newtype wrapper around a byte buffer, used for UDP mediators to receive incoming datagrams.
 struct IoByteBuffer {
     byte_vec: Vec<u8>,
+}
 // A generator of speculative variables. Created on-demand during the synchronous round
 // by the IdManager.
 #[derive(Debug)]
 struct SpecVarStream {
     connector_id: ConnectorId,
     port_suffix_stream: U32Stream,
+}
 // Manages the messy state of the various endpoints, pollers, buffers, etc.
 #[derive(Debug)]
 struct EndpointManager {
     // invariants:
     // 1. net and udp endpoints are registered with poll with tokens computed with TargetToken::into
     // 2. Events is empty
     poll: Poll,
     events: Events,
     delayed_messages: Vec<(usize, Msg)>,
     undelayed_messages: Vec<(usize, Msg)>, // ready to yield
     net_endpoint_store: EndpointStore<NetEndpointExt>,
     udp_endpoint_store: EndpointStore<UdpEndpointExt>,
     io_byte_buffer: IoByteBuffer,
+}
 // A storage of endpoints, which keeps track of which components have raised
 // an event during poll(), signifying that they need to be checked for new incoming data
 #[derive(Debug)]
 struct EndpointStore<T> {
     endpoint_exts: Vec<T>,
     polled_undrained: VecSet<usize>,
+}
 // The information associated with a port identifier, designed for local storage.
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 struct PortInfo {
     owner: ComponentId,
     peer: Option<PortId>,
     polarity: Polarity,
     route: Route,
+}
 // Similar to `PortInfo`, but designed for communication during the setup procedure.
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 struct MyPortInfo {
     polarity: Polarity,
     port: PortId,
     owner: ComponentId,
+}
 // Newtype around port info map, allowing the implementation of some
 // useful methods
 #[derive(Default, Debug, Clone, serde::Serialize, serde::Deserialize)]
 struct PortInfoMap {
     // invariant: self.invariant_preserved()
     // `owned` is redundant information, allowing for fast lookup
     // of a component's owned ports (which occurs during the sync round a lot)
     map: HashMap<PortId, PortInfo>,
     owned: HashMap<ComponentId, HashSet<PortId>>,
+}
 // A convenient substructure for containing port info and the ID manager.
 // Houses the bulk of the connector's persistent state between rounds.
 // It turns out several situations require access to both things.
 #[derive(Debug, Clone)]
 struct IdAndPortState {
     port_info: PortInfoMap,
     id_manager: IdManager,
+}
 // A component's setup-phase-specific data
 #[derive(Debug)]
 struct ConnectorCommunication {
     round_index: usize,
     endpoint_manager: EndpointManager,
     neighborhood: Neighborhood,
     native_batches: Vec<NativeBatch>,
     round_result: Result<Option<RoundEndedNative>, SyncError>,
+}
 // A component's data common to both setup and communication phases
 #[derive(Debug)]
 struct ConnectorUnphased {
     proto_description: Arc<ProtocolDescription>,
     proto_components: HashMap<ComponentId, ComponentState>,
     logger: Box<dyn Logger>,
     ips: IdAndPortState,
     native_component_id: ComponentId,
+}
 // A connector's phase-specific data
 #[derive(Debug)]
 enum ConnectorPhased {
     Setup(Box<ConnectorSetup>),
     Communication(Box<ConnectorCommunication>),
+}
 // A connector's setup-phase-specific data
 #[derive(Debug)]
 struct ConnectorSetup {
     net_endpoint_setups: Vec<NetEndpointSetup>,
     udp_endpoint_setups: Vec<UdpEndpointSetup>,
+}
 // A newtype wrapper for a map from speculative variable to speculative value
 // A missing mapping corresponds with "unspecified".
 #[derive(Default, Clone, Eq, PartialEq, Hash, serde::Serialize, serde::Deserialize)]
 struct Predicate {
     assigned: BTreeMap<SpecVar, SpecVal>,
+}
 // Identifies a child of this connector in the _solution tree_.
 // Each connector creates its own local solutions for the consensus procedure during `sync`,
 // from the solutions of its children. Those children are either locally-managed components,
 // (which are leaves in the solution tree), or other connectors reachable through the given
 // network endpoint (which are internal nodes in the solution tree).
 #[derive(Debug, Clone, Copy, Eq, PartialEq, Hash, serde::Serialize, serde::Deserialize)]
 enum SubtreeId {
     LocalComponent(ComponentId),
     NetEndpoint { index: usize },
+}
 // An accumulation of the connector's knowledge of all (a) the local solutions its children
 // in the solution tree have found, and (b) its own solutions derivable from those of its children.
 // This structure starts off each round with an empty set, and accumulates solutions as they are found
 // by local components, or received over the network in control messages.
 // IMPORTANT: solutions, once found, don't go away until the end of the round. That is to
 // say that these sets GROW until the round is over, and all solutions are reset.
 #[derive(Debug)]
 struct SolutionStorage {
     // invariant: old_local U new_local solutions are those that can be created from
     // the UNION of one element from each set in `subtree_solution`.
     // invariant is maintained by potentially populating new_local whenever subtree_solutions is populated.
     old_local: HashSet<Predicate>, // already sent to this connector's parent OR decided
     new_local: HashSet<Predicate>, // not yet sent to this connector's parent OR decided
     // this pair acts as SubtreeId -> HashSet<Predicate> which is friendlier to iteration
     subtree_solutions: Vec<HashSet<Predicate>>,
     subtree_id_to_index: HashMap<SubtreeId, usize>,
+}
 // Stores the transient data of a synchronous round.
 // Some of it is for bookkeeping, and the rest is a temporary mirror of fields of
 // `ConnectorUnphased`, such that any changes are safely contained within RoundCtx,
 // and can be undone if the round fails.
 struct RoundCtx {
     solution_storage: SolutionStorage,
     spec_var_stream: SpecVarStream,
     payload_inbox: Vec<(PortId, SendPayloadMsg)>,
     deadline: Option<Instant>,
     ips: IdAndPortState,
+}
 // A trait intended to limit the access of the ConnectorUnphased structure
 // such that we don't accidentally modify any important component/port data
 // while the results of the round are undecided. Why? Any actions during Connector::sync
 // are _speculative_ until the round is decided, and we need a safe way of rolling
 // back any changes.
 trait CuUndecided {
     fn logger(&mut self) -> &mut dyn Logger;
     fn proto_description(&self) -> &ProtocolDescription;
     fn native_component_id(&self) -> ComponentId;
     fn logger_and_protocol_description(&mut self) -> (&mut dyn Logger, &ProtocolDescription);
     fn logger_and_protocol_components(
         &mut self,
     ) -> (&mut dyn Logger, &mut HashMap<ComponentId, ComponentState>);
+}
 // Represents a set of synchronous port operations that the native component
 // has described as an "option" for completing during the synchronous rounds.
 // Operations contained here succeed together or not at all.
 // A native with N=2+ batches are expressing an N-way nondeterministic choice
 #[derive(Debug, Default)]
 struct NativeBatch {
     // invariant: putters' and getters' polarities respected
     to_put: HashMap<PortId, Payload>,
     to_get: HashSet<PortId>,
+}
 // Parallels a mio::Token type, but more clearly communicates
 // the way it identifies the evented structre it corresponds to.
 // See runtime/setup for methods converting between TokenTarget and mio::Token
 #[derive(Debug, Copy, Clone, Eq, PartialEq, Hash)]
 enum TokenTarget {
     NetEndpoint { index: usize },
     UdpEndpoint { index: usize },
+}
 // Returned by the endpoint manager as a result of comm_recv, telling the connector what happened,
 // such that it can know when to continue polling, and when to block.
 enum CommRecvOk {
     TimeoutWithoutNew,
     NewPayloadMsgs,

src/runtime/setup.rs

➞

Show inline comments

 use crate::common::*;
 use crate::runtime::*;
 impl TokenTarget {
     // subdivides the domain of usize into
     // [NET_ENDPOINT][UDP_ENDPOINT  ]
     // ^0            ^usize::MAX/2   ^usize::MAX
     const HALFWAY_INDEX: usize = usize::MAX / 2;
     const MAX_INDEX: usize = usize::MAX;
+}
 impl From<Token> for TokenTarget {
     fn from(Token(index): Token) -> Self {
         if let Some(shifted) = index.checked_sub(Self::HALFWAY_INDEX) {
             TokenTarget::UdpEndpoint { index: shifted }
         } else {
             TokenTarget::NetEndpoint { index }
+        }
+    }
+}
 impl Into<Token> for TokenTarget {
     fn into(self) -> Token {
         match self {
             TokenTarget::UdpEndpoint { index } => Token(index + Self::HALFWAY_INDEX),
             TokenTarget::NetEndpoint { index } => Token(index),
+        }
+    }
+}
 impl Connector {
     /// Create a new connector structure with the given protocol description (via Arc to facilitate sharing).
     /// The resulting connector will start in the setup phase, and cannot be used for communication until the
     /// `connect` procedure completes.
     /// # Safety
     /// The correctness of the system's underlying distributed algorithms requires that no two
     /// connectors have the same ID. If the user does not know the identifiers of other connectors in the
     /// system, it is advised to guess it using Connector::random_id (relying on the exceptionally low probability of an error).
     /// Sessions with duplicate connector identifiers will not result in any memory unsafety, but cannot be guaranteed
     /// to preserve their configured protocols.
     /// Fortunately, in most realistic cases, the presence of duplicate connector identifiers will result in an
     /// error during `connect`, observed as a peer misbehaving.
     pub fn new(
         mut logger: Box<dyn Logger>,
         proto_description: Arc<ProtocolDescription>,
         connector_id: ConnectorId,
     ) -> Self {
         log!(&mut *logger, "Created with connector_id {:?}", connector_id);
         let mut id_manager = IdManager::new(connector_id);
         let native_component_id = id_manager.new_component_id();
         Self {
             unphased: ConnectorUnphased {
                 proto_description,
                 proto_components: Default::default(),
                 logger,
                 native_component_id,
                 ips: IdAndPortState { id_manager, port_info: Default::default() },
             },
             phased: ConnectorPhased::Setup(Box::new(ConnectorSetup {
                 net_endpoint_setups: Default::default(),
                 udp_endpoint_setups: Default::default(),
             })),
+        }
+    }
     /// Conceptually, this returning [p0, g1] is sugar for:
     /// 1. create port pair [p0, g0]
     /// 2. create port pair [p1, g1]
     /// 3. create udp component with interface of moved ports [p1, g0]
     /// 4. return [p0, g1]
     pub fn new_udp_mediator_component(
         &mut self,
         local_addr: SocketAddr,
         peer_addr: SocketAddr,
     ) -> Result<[PortId; 2], WrongStateError> {
         let Self { unphased: cu, phased } = self;
         match phased {
             ConnectorPhased::Communication(..) => Err(WrongStateError),
             ConnectorPhased::Setup(setup) => {
                 let udp_index = setup.udp_endpoint_setups.len();
                 let udp_cid = cu.ips.id_manager.new_component_id();
                 // allocates 4 new port identifiers, two for each logical channel,
                 // one channel per direction (into and out of the component)
                 let mut npid = || cu.ips.id_manager.new_port_id();
                 let [nin, nout, uin, uout] = [npid(), npid(), npid(), npid()];
                 // allocate the native->udp_mediator channel's ports
                 cu.ips.port_info.map.insert(
                     nout,
                     PortInfo {
                         route: Route::LocalComponent,
                         polarity: Putter,
                         peer: Some(uin),
                         owner: cu.native_component_id,
                     },
                 );
                 cu.ips.port_info.map.insert(
                     uin,
                     PortInfo {
                         route: Route::UdpEndpoint { index: udp_index },
                         polarity: Getter,
                         peer: Some(uin),
                         owner: udp_cid,
                     },
                 );
                 // allocate the udp_mediator->native channel's ports
                 cu.ips.port_info.map.insert(
                     uout,
                     PortInfo {
                         route: Route::UdpEndpoint { index: udp_index },
                         polarity: Putter,
                         peer: Some(uin),
                         owner: udp_cid,
                     },
                 );
                 cu.ips.port_info.map.insert(
                     nin,
                     PortInfo {
                         route: Route::LocalComponent,
                         polarity: Getter,
                         peer: Some(uout),
                         owner: cu.native_component_id,
                     },
                 );
                 // allocate the two ports owned by the UdpMediator component
                 // Remember to setup this UdpEndpoint setup during `connect` later.
                 setup.udp_endpoint_setups.push(UdpEndpointSetup {
                     local_addr,
                     peer_addr,
                     getter_for_incoming: nin,
                 });
                 // update owned sets
                 cu.ips
                     .port_info
                     .owned
                     .entry(cu.native_component_id)
                     .or_default()
                     .extend([nin, nout].iter().copied());
                 cu.ips.port_info.owned.insert(udp_cid, maplit::hashset! {uin, uout});
                 // Return the native's output, input port pair
                 Ok([nout, nin])
+            }
+        }
+    }
     /// Adds a "dangling" port to the connector in the setup phase,
     /// to be formed into channel during the connect procedure with the given
     /// transport layer information.
     pub fn new_net_port(
         &mut self,
         polarity: Polarity,
         sock_addr: SocketAddr,
         endpoint_polarity: EndpointPolarity,
     ) -> Result<PortId, WrongStateError> {
         let Self { unphased: cu, phased } = self;
         match phased {
             ConnectorPhased::Communication(..) => Err(WrongStateError),
             ConnectorPhased::Setup(setup) => {
                 // allocate a single dangling port with a `None` peer (for now)
                 let new_pid = cu.ips.id_manager.new_port_id();
                 cu.ips.port_info.map.insert(
                     new_pid,
                     PortInfo {
                         route: Route::LocalComponent,
                         peer: None,
                         owner: cu.native_component_id,
                         polarity,
                     },
                 );
                 log!(
                     cu.logger,
                     "Added net port {:?} with polarity {:?} addr {:?} endpoint_polarity {:?}",
                     new_pid,
                     polarity,
                     &sock_addr,
                     endpoint_polarity
                 );
                 // Remember to setup this NetEndpoint setup during `connect` later.
                 setup.net_endpoint_setups.push(NetEndpointSetup {
                     sock_addr,
                     endpoint_polarity,
                     getter_for_incoming: new_pid,
                 });
                 // update owned set
                 cu.ips.port_info.owned.entry(cu.native_component_id).or_default().insert(new_pid);
                 Ok(new_pid)
+            }
+        }
+    }
     /// Finalizes the connector's setup procedure and forms a distributed system with
     /// all other connectors reachable through network channels. This procedure represents
     /// a synchronization barrier, and upon successful return, the connector can no longer add new network ports,
     /// but is ready to begin the first communication round.
     /// Initially, the connector has a singleton set of _batches_, the only element of which is empty.
     /// This single element starts off selected. The selected batch is modified with `put` and `get`,
     /// and new batches are added and selected with `next_batch`. See `sync` for an explanation of the
     /// purpose of these batches.
     pub fn connect(&mut self, timeout: Option<Duration>) -> Result<(), ConnectError> {
         use ConnectError as Ce;
         let Self { unphased: cu, phased } = self;
         match &phased {
             ConnectorPhased::Communication { .. } => {
                 log!(cu.logger, "Call to connecting in connected state");
                 Err(Ce::AlreadyConnected)
+            }
             ConnectorPhased::Setup(setup) => {
                 // Idea: Clone `self.unphased`, and then pass the replica to
                 // `connect_inner` to do the work, attempting to create a new connector structure
                 // in connected state without encountering any errors.
                 // If anything goes wrong during `connect_inner`, we simply keep the original `cu`.
                 // Ideally, we'd simply clone `cu` in its entirety.
                 // However, it isn't clonable, because of the pesky logger.
                 // Solution: the original and clone ConnectorUnphased structures
                 // 'share' the original logger by using `mem::swap` strategically to pass a dummy back and forth,
                 // such that the real logger is wherever we need it to be without violating any invariants.
                 let mut cu_clone = ConnectorUnphased {
                     logger: Box::new(DummyLogger),
                     proto_components: cu.proto_components.clone(),
                     native_component_id: cu.native_component_id.clone(),
                     ips: cu.ips.clone(),
                     proto_description: cu.proto_description.clone(),
                 };
                 // cu has REAL logger...
                 std::mem::swap(&mut cu.logger, &mut cu_clone.logger);
                 // ... cu_clone has REAL logger.
                 match Self::connect_inner(cu_clone, setup, timeout) {
                     Ok(connected_connector) => {
                         *self = connected_connector;
                         Ok(())
+                    }
                     Err((err, mut logger)) => {
                         // Put the original logger back in place (in self.unphased, AKA `cu`).
                         // cu_clone has REAL logger...
                         std::mem::swap(&mut cu.logger, &mut logger);
                         // ... cu has REAL logger.
                         Err(err)
+                    }
+                }
+            }
+        }
+    }
     // Given an immutable setup structure, and my own (cloned) ConnetorUnphased,
     // attempt to complete the setup procedure and return a new connector in Connected state.
     // If anything goes wrong, throw everything in the bin, except for the Logger, which is
     // the only structure that sees lasting effects of the failed attempt.
     fn connect_inner(
         mut cu: ConnectorUnphased,
         setup: &ConnectorSetup,
         timeout: Option<Duration>,
     ) -> Result<Self, (ConnectError, Box<dyn Logger>)> {
         log!(cu.logger, "~~~ CONNECT called timeout {:?}", timeout);
         let deadline = timeout.map(|to| Instant::now() + to);
         // `try_complete` is a helper function, which DOES NOT own `cu`, and returns ConnectError on err.
         // This outer function takes its output and wraps it alongside `cu` (which it owns)
         // as appropriate for Err(...) and OK(...) cases.
         let mut try_complete = || {
             // connect all endpoints in parallel; send and receive peer ids through ports
             let mut endpoint_manager = setup_endpoints_and_pair_ports(
                 &mut *cu.logger,
                 &setup.net_endpoint_setups,
                 &setup.udp_endpoint_setups,
                 &mut cu.ips.port_info,
                 &deadline,
             )?;
             log!(
                 cu.logger,
                 "Successfully connected {} endpoints. info now {:#?} {:#?}",
                 endpoint_manager.net_endpoint_store.endpoint_exts.len(),
                 &cu.ips.port_info,
                 &endpoint_manager,
             );
             // leader election and tree construction. Learn our role in the consensus tree,
             // from learning who are our children/parents (neighbors) in the consensus tree.
             let neighborhood = init_neighborhood(
                 cu.ips.id_manager.connector_id,
                 &mut *cu.logger,
                 &mut endpoint_manager,
                 &deadline,
             )?;
             log!(cu.logger, "Successfully created neighborhood {:?}", &neighborhood);
             // Put it all together with an initial round index of zero.
             let mut comm = ConnectorCommunication {
                 round_index: 0,
                 endpoint_manager,
                 neighborhood,
                 native_batches: vec![Default::default()],
                 round_result: Ok(None), // no previous round yet
             };
             if cfg!(feature = "session_optimization") {
                 // Perform the session optimization procedure, which may modify the
                 // internals of the connector, rerouting ports, moving around connectors etc.
                 session_optimize(&mut cu, &mut comm, &deadline)?;
+            }
             log!(cu.logger, "connect() finished. setup phase complete");
             Ok(comm)
         };
         match try_complete() {
             Ok(comm) => {
                 Ok(Self { unphased: cu, phased: ConnectorPhased::Communication(Box::new(comm)) })
+            }
             Err(err) => Err((err, cu.logger)),
+        }
+    }
+}
 // Given a set of net_ and udp_ endpoints to setup,
 // port information to flesh out (by discovering peers through channels)
 // and a deadline in which to do it,
 // try to return:
 // - An EndpointManager, containing all the set up endpoints
 // - new information about ports acquired through the newly-created channels
 fn setup_endpoints_and_pair_ports(
     logger: &mut dyn Logger,
     net_endpoint_setups: &[NetEndpointSetup],
     udp_endpoint_setups: &[UdpEndpointSetup],
     port_info: &mut PortInfoMap,
     deadline: &Option<Instant>,
 ) -> Result<EndpointManager, ConnectError> {
     use ConnectError as Ce;
     const BOTH: Interest = Interest::READABLE.add(Interest::WRITABLE);
     const RETRY_PERIOD: Duration = Duration::from_millis(200);
     // The data for a net endpoint's setup in progress
     struct NetTodo {
         // becomes completed once sent_local_port && recv_peer_port.is_some()
         // we send local port if we haven't already and we receive a writable event
         // we recv peer port if we haven't already and we receive a readbale event
         todo_endpoint: NetTodoEndpoint,
         endpoint_setup: NetEndpointSetup,
         sent_local_port: bool,          // true <-> I've sent my local port
         recv_peer_port: Option<PortId>, // Some(..) <-> I've received my peer's port
+    }
     // The data for a udp endpoint's setup in progress
     struct UdpTodo {
         // becomes completed once we receive our first writable event
         getter_for_incoming: PortId,
         sock: UdpSocket,
+    }
     // Substructure of `NetTodo`, which represents the endpoint itself
     enum NetTodoEndpoint {
         Accepting(TcpListener),       // awaiting it's peer initiating the connection
         PeerInfoRecving(NetEndpoint), // awaiting info about peer port through the channel
+    }
     ////////////////////////////////////////////
     // Start to construct our return values
     let mut poll = Poll::new().map_err(|_| Ce::PollInitFailed)?;
     let mut events =
         Events::with_capacity((net_endpoint_setups.len() + udp_endpoint_setups.len()) * 2 + 4);
     let [mut net_polled_undrained, udp_polled_undrained] = [VecSet::default(), VecSet::default()];
     let mut delayed_messages = vec![];
     let mut last_retry_at = Instant::now();
     let mut io_byte_buffer = IoByteBuffer::default();
     // Create net/udp todo structures, each already registered with poll
     let mut net_todos = net_endpoint_setups
         .iter()
         .enumerate()
         .map(|(index, endpoint_setup)| {
             let token = TokenTarget::NetEndpoint { index }.into();
             log!(logger, "Net endpoint {} beginning setup with {:?}", index, &endpoint_setup);
             let todo_endpoint = if let EndpointPolarity::Active = endpoint_setup.endpoint_polarity {
                 let mut stream = TcpStream::connect(endpoint_setup.sock_addr)
                     .map_err(|_| Ce::TcpInvalidConnect(endpoint_setup.sock_addr))?;
                 poll.registry().register(&mut stream, token, BOTH).unwrap();
                 NetTodoEndpoint::PeerInfoRecving(NetEndpoint { stream, inbox: vec![] })
             } else {
                 let mut listener = TcpListener::bind(endpoint_setup.sock_addr)
                     .map_err(|_| Ce::BindFailed(endpoint_setup.sock_addr))?;
                 poll.registry().register(&mut listener, token, BOTH).unwrap();
                 NetTodoEndpoint::Accepting(listener)
             };
             Ok(NetTodo {
                 todo_endpoint,
                 sent_local_port: false,
                 recv_peer_port: None,
                 endpoint_setup: endpoint_setup.clone(),
             })
         })
         .collect::<Result<Vec<NetTodo>, ConnectError>>()?;
     let udp_todos = udp_endpoint_setups
         .iter()
         .enumerate()
         .map(|(index, endpoint_setup)| {
             let mut sock = UdpSocket::bind(endpoint_setup.local_addr)
                 .map_err(|_| Ce::BindFailed(endpoint_setup.local_addr))?;
             sock.connect(endpoint_setup.peer_addr)
                 .map_err(|_| Ce::UdpConnectFailed(endpoint_setup.peer_addr))?;
             poll.registry()
                 .register(&mut sock, TokenTarget::UdpEndpoint { index }.into(), Interest::WRITABLE)
                 .unwrap();
@@ @@ -652,423 +651,423 @@ fn setup_endpoints_and_pair_ports( @@
         .collect();
     let endpoint_manager = EndpointManager {
         poll,
         events,
         undelayed_messages: delayed_messages, // no longer delayed
         delayed_messages: Default::default(),
         net_endpoint_store: EndpointStore {
             endpoint_exts: net_endpoint_exts,
             polled_undrained: net_polled_undrained,
         },
         udp_endpoint_store: EndpointStore {
             endpoint_exts: udp_endpoint_exts,
             polled_undrained: udp_polled_undrained,
         },
         io_byte_buffer,
     };
     Ok(endpoint_manager)
+}
 // Given a fully-formed endpoint manager,
 // construct the consensus tree with:
 // 1. decentralized leader election
 // 2. centralized tree construction
 fn init_neighborhood(
     connector_id: ConnectorId,
     logger: &mut dyn Logger,
     em: &mut EndpointManager,
     deadline: &Option<Instant>,
 ) -> Result<Neighborhood, ConnectError> {
     use {ConnectError as Ce, Msg::SetupMsg as S, SetupMsg as Sm};
     // storage structure for the state of a distributed wave
     // (for readability)
     #[derive(Debug)]
     struct WaveState {
         parent: Option<usize>,
         leader: ConnectorId,
+    }
     // kick off a leader-election wave rooted at myself
     // given the desired wave information
     // (e.g. don't inform my parent if they exist)
     fn do_wave(
         em: &mut EndpointManager,
         awaiting: &mut HashSet<usize>,
         ws: &WaveState,
     ) -> Result<(), ConnectError> {
         awaiting.clear();
         let msg = S(Sm::LeaderWave { wave_leader: ws.leader });
         for index in em.index_iter() {
             if Some(index) != ws.parent {
                 em.send_to_setup(index, &msg)?;
                 awaiting.insert(index);
+            }
+        }
         Ok(())
+    }
     ///////////////////////
     /*
     Conceptually, we have two distinct disstributed algorithms back-to-back
 . Leader election using echo algorithm with extinction.
         - Each connector initiates a wave tagged with their ID
         - Connectors participate in waves of GREATER ID, abandoning previous waves
         - Only the wave of the connector with GREATEST ID completes, whereupon they are the leader
 . Tree construction
         - The leader broadcasts their leadership with msg A
         - Upon receiving their first announcement, connectors reply B, and send A to all peers
         - A controller exits once they have received A or B from each neighbor
     The actual implementation is muddier, because non-leaders aren't aware of termiantion of algorithm 1,
     so they rely on receipt of the leader's announcement to realize that algorithm 2 has begun.
     NOTE the distinction between PARENT and LEADER
     */
     log!(logger, "beginning neighborhood construction");
     if em.num_net_endpoints() == 0 {
         log!(logger, "Edge case of no neighbors! No parent an no children!");
         return Ok(Neighborhood { parent: None, children: VecSet::new(vec![]) });
+    }
     log!(logger, "Have {} endpoints. Must participate in distributed alg.", em.num_net_endpoints());
     let mut awaiting = HashSet::with_capacity(em.num_net_endpoints());
     // 1+ neighbors. Leader can only be learned by receiving messages
     // loop ends when I know my sink tree parent (implies leader was elected)
     let election_result: WaveState = {
         // initially: No parent, I'm the best leader.
         let mut best_wave = WaveState { parent: None, leader: connector_id };
         // start a wave for this initial state
         do_wave(em, &mut awaiting, &best_wave)?;
         // with 1+ neighbors, progress is only made in response to incoming messages
         em.undelay_all();
         'election: loop {
             log!(logger, "Election loop. awaiting {:?}...", awaiting.iter());
             let (recv_index, msg) = em.try_recv_any_setup(logger, deadline)?;
             log!(logger, "Received from index {:?} msg {:?}", &recv_index, &msg);
             match msg {
                 S(Sm::LeaderAnnounce { tree_leader }) => {
                     // A neighbor explicitly tells me who is the leader
                     // they become my parent, and I adopt their announced leader
                     let election_result =
                         WaveState { leader: tree_leader, parent: Some(recv_index) };
                     log!(logger, "Election lost! Result {:?}", &election_result);
                     assert!(election_result.leader >= best_wave.leader);
                     assert_ne!(election_result.leader, connector_id);
                     break 'election election_result;
+                }
                 S(Sm::LeaderWave { wave_leader }) => {
                     use Ordering as O;
                     match wave_leader.cmp(&best_wave.leader) {
                         O::Less => log!(
                             logger,
                             "Ignoring wave with Id {:?}<{:?}",
                             wave_leader,
                             best_wave.leader
                         ),
                         O::Greater => {
                             log!(
                                 logger,
                                 "Joining wave with Id {:?}>{:?}",
                                 wave_leader,
                                 best_wave.leader
                             );
                             best_wave = WaveState { leader: wave_leader, parent: Some(recv_index) };
                             log!(logger, "New wave state {:?}", &best_wave);
                             do_wave(em, &mut awaiting, &best_wave)?;
                             if awaiting.is_empty() {
                                 log!(logger, "Special case! Only neighbor is parent. Replying to {:?} msg {:?}", recv_index, &msg);
                                 em.send_to_setup(recv_index, &msg)?;
+                            }
+                        }
                         O::Equal => {
                             assert!(awaiting.remove(&recv_index));
                             log!(
                                 logger,
                                 "Wave reply from index {:?} for leader {:?}. Now awaiting {} replies",
                                 recv_index,
                                 best_wave.leader,
                                 awaiting.len()
                             );
                             if awaiting.is_empty() {
                                 if let Some(parent) = best_wave.parent {
                                     log!(
                                         logger,
                                         "Sub-wave done! replying to parent {:?} msg {:?}",
                                         parent,
                                         &msg
                                     );
                                     em.send_to_setup(parent, &msg)?;
                                 } else {
                                     let election_result: WaveState = best_wave;
                                     log!(logger, "Election won! Result {:?}", &election_result);
                                     break 'election election_result;
+                                }
+                            }
+                        }
+                    }
+                }
                 msg @ S(Sm::YouAreMyParent) | msg @ S(Sm::MyPortInfo(_)) => {
                     log!(logger, "Endpont {:?} sent unexpected msg! {:?}", recv_index, &msg);
                     return Err(Ce::SetupAlgMisbehavior);
+                }
                 msg @ S(Sm::SessionScatter { .. })
                 | msg @ S(Sm::SessionGather { .. })
                 | msg @ Msg::CommMsg { .. } => {
                     log!(logger, "delaying msg {:?} during election algorithm", msg);
                     em.delayed_messages.push((recv_index, msg));
+                }
+            }
+        }
     };
     // starting algorithm 2. Send a message to every neighbor
     // namely, send "YouAreMyParent" to parent (if they exist),
     // and LeaderAnnounce to everyone else
     log!(logger, "Starting tree construction. Step 1: send one msg per neighbor");
     awaiting.clear();
     for index in em.index_iter() {
         if Some(index) == election_result.parent {
             em.send_to_setup(index, &S(Sm::YouAreMyParent))?;
         } else {
             awaiting.insert(index);
             em.send_to_setup(
                 index,
                 &S(Sm::LeaderAnnounce { tree_leader: election_result.leader }),
             )?;
+        }
+    }
     // Receive one message from each neighbor to learn
     // whether they consider me their parent or not.
     let mut children = vec![];
     em.undelay_all();
     while !awaiting.is_empty() {
         log!(logger, "Tree construction_loop loop. awaiting {:?}...", awaiting.iter());
         let (recv_index, msg) = em.try_recv_any_setup(logger, deadline)?;
         log!(logger, "Received from index {:?} msg {:?}", &recv_index, &msg);
         match msg {
             S(Sm::LeaderAnnounce { .. }) => {
                 // `recv_index` is not my child
                 log!(
                     logger,
                     "Got reply from non-child index {:?}. Children: {:?}",
                     recv_index,
                     children.iter()
                 );
                 if !awaiting.remove(&recv_index) {
                     return Err(Ce::SetupAlgMisbehavior);
+                }
+            }
             S(Sm::YouAreMyParent) => {
                 if !awaiting.remove(&recv_index) {
                     log!(
                         logger,
                         "Got reply from child index {:?}. Children before... {:?}",
                         recv_index,
                         children.iter()
                     );
                     return Err(Ce::SetupAlgMisbehavior);
+                }
                 // `recv_index` is my child
                 children.push(recv_index);
+            }
             msg @ S(Sm::MyPortInfo(_)) | msg @ S(Sm::LeaderWave { .. }) => {
                 log!(logger, "discarding old message {:?} during election", msg);
+            }
             msg @ S(Sm::SessionScatter { .. })
             | msg @ S(Sm::SessionGather { .. })
             | msg @ Msg::CommMsg { .. } => {
                 log!(logger, "delaying msg {:?} during election", msg);
                 em.delayed_messages.push((recv_index, msg));
+            }
+        }
+    }
     // Neighborhood complete!
     children.shrink_to_fit();
     let neighborhood =
         Neighborhood { parent: election_result.parent, children: VecSet::new(children) };
     log!(logger, "Neighborhood constructed {:?}", &neighborhood);
     Ok(neighborhood)
+}
 // Connectors collect a map of type ConnectorId=>SessionInfo,
 // representing a global view of the session's state at the leader.
 // The leader rewrites its contents however they like (currently: nothing happens)
 // and the map is again broadcasted, for each peer to make their local changes to
 // reflect the results of the rewrite.
 fn session_optimize(
     cu: &mut ConnectorUnphased,
     comm: &mut ConnectorCommunication,
     deadline: &Option<Instant>,
 ) -> Result<(), ConnectError> {
     use {ConnectError as Ce, Msg::SetupMsg as S, SetupMsg as Sm};
     log!(cu.logger, "Beginning session optimization");
     // populate session_info_map from a message per child
     let mut unoptimized_map: HashMap<ConnectorId, SessionInfo> = Default::default();
     let mut awaiting: HashSet<usize> = comm.neighborhood.children.iter().copied().collect();
     comm.endpoint_manager.undelay_all();
     while !awaiting.is_empty() {
         log!(
             cu.logger,
             "Session gather loop. awaiting info from children {:?}...",
             awaiting.iter()
         );
         let (recv_index, msg) =
             comm.endpoint_manager.try_recv_any_setup(&mut *cu.logger, deadline)?;
         log!(cu.logger, "Received from index {:?} msg {:?}", &recv_index, &msg);
         match msg {
             S(Sm::SessionGather { unoptimized_map: child_unoptimized_map }) => {
                 if !awaiting.remove(&recv_index) {
                     log!(
                         cu.logger,
                         "Wasn't expecting session info from {:?}. Got {:?}",
                         recv_index,
                         &child_unoptimized_map
                     );
                     return Err(Ce::SetupAlgMisbehavior);
+                }
                 unoptimized_map.extend(child_unoptimized_map.into_iter());
+            }
             msg @ S(Sm::YouAreMyParent)
             | msg @ S(Sm::MyPortInfo(..))
             | msg @ S(Sm::LeaderAnnounce { .. })
             | msg @ S(Sm::LeaderWave { .. }) => {
                 log!(cu.logger, "discarding old message {:?} during election", msg);
+            }
             msg @ S(Sm::SessionScatter { .. }) => {
                 log!(
                     cu.logger,
                     "Endpoint {:?} sent unexpected scatter! {:?} I've not contributed yet!",
                     recv_index,
                     &msg
                 );
                 return Err(Ce::SetupAlgMisbehavior);
+            }
             msg @ Msg::CommMsg(..) => {
                 log!(cu.logger, "delaying msg {:?} during session optimization", msg);
                 comm.endpoint_manager.delayed_messages.push((recv_index, msg));
+            }
+        }
+    }
     log!(
         cu.logger,
         "Gathered all children's maps. ConnectorId set is... {:?}",
         unoptimized_map.keys()
     );
     // add my own session info to the map
     let my_session_info = SessionInfo {
         port_info: cu.ips.port_info.clone(),
         proto_components: cu.proto_components.clone(),
         serde_proto_description: SerdeProtocolDescription(cu.proto_description.clone()),
         endpoint_incoming_to_getter: comm
             .endpoint_manager
             .net_endpoint_store
             .endpoint_exts
             .iter()
             .map(|ee| ee.getter_for_incoming)
             .collect(),
     };
     unoptimized_map.insert(cu.ips.id_manager.connector_id, my_session_info);
     log!(cu.logger, "Inserting my own info. Unoptimized subtree map is {:?}", &unoptimized_map);
     // acquire the optimized info...
     let optimized_map = if let Some(parent) = comm.neighborhood.parent {
         // ... as a message from my parent
         log!(cu.logger, "Forwarding gathered info to parent {:?}", parent);
         let msg = S(Sm::SessionGather { unoptimized_map });
         comm.endpoint_manager.send_to_setup(parent, &msg)?;
         'scatter_loop: loop {
             log!(
                 cu.logger,
                 "Session scatter recv loop. awaiting info from children {:?}...",
                 awaiting.iter()
             );
             let (recv_index, msg) =
                 comm.endpoint_manager.try_recv_any_setup(&mut *cu.logger, deadline)?;
             log!(cu.logger, "Received from index {:?} msg {:?}", &recv_index, &msg);
             match msg {
                 S(Sm::SessionScatter { optimized_map }) => {
                     if recv_index != parent {
                         log!(cu.logger, "I expected the scatter from my parent only!");
                         return Err(Ce::SetupAlgMisbehavior);
+                    }
                     break 'scatter_loop optimized_map;
+                }
                 msg @ Msg::CommMsg { .. } => {
                     log!(cu.logger, "delaying msg {:?} during scatter recv", msg);
                     comm.endpoint_manager.delayed_messages.push((recv_index, msg));
+                }
                 msg @ S(Sm::SessionGather { .. })
                 | msg @ S(Sm::YouAreMyParent)
                 | msg @ S(Sm::MyPortInfo(..))
                 | msg @ S(Sm::LeaderAnnounce { .. })
                 | msg @ S(Sm::LeaderWave { .. }) => {
                     log!(cu.logger, "discarding old message {:?} during election", msg);
+                }
+            }
+        }
     } else {
         // by computing it myself
         log!(cu.logger, "I am the leader! I will optimize this session");
         leader_session_map_optimize(&mut *cu.logger, unoptimized_map)?
     };
     log!(
         cu.logger,
         "Optimized info map is {:?}. Sending to children {:?}",
         &optimized_map,
         comm.neighborhood.children.iter()
     );
     log!(cu.logger, "All session info dumped!: {:#?}", &optimized_map);
     // extract my own ConnectorId's entry
     let optimized_info =
         optimized_map.get(&cu.ips.id_manager.connector_id).expect("HEY NO INFO FOR ME?").clone();
     // broadcast the optimized session info to my children
     let msg = S(Sm::SessionScatter { optimized_map });
     for &child in comm.neighborhood.children.iter() {
         comm.endpoint_manager.send_to_setup(child, &msg)?;
+    }
     // apply local optimizations
     apply_my_optimizations(cu, comm, optimized_info)?;
     log!(cu.logger, "Session optimizations applied");
     Ok(())
+}
 // Defines the optimization function, consuming an optimized map,
 // and returning an optimized map.
 fn leader_session_map_optimize(
     logger: &mut dyn Logger,
-    mut m: HashMap<ConnectorId, SessionInfo>,
     m: HashMap<ConnectorId, SessionInfo>,
 ) -> Result<HashMap<ConnectorId, SessionInfo>, ConnectError> {
     log!(logger, "Session map optimize START");
     // currently, it's the identity function
     log!(logger, "Session map optimize END");
     Ok(m)
+}
 // Modify the given connector's internals to reflect
 // the given session info
 fn apply_my_optimizations(
     cu: &mut ConnectorUnphased,
     comm: &mut ConnectorCommunication,
     session_info: SessionInfo,
 ) -> Result<(), ConnectError> {
     let SessionInfo {
         proto_components,
         port_info,
         serde_proto_description,
         endpoint_incoming_to_getter,
     } = session_info;
     // simply overwrite the contents
     println!("BEFORE: {:#?}\n{:#?}", cu, comm);
     cu.ips.port_info = port_info;
     assert!(cu.ips.port_info.invariant_preserved());
     cu.proto_components = proto_components;
     cu.proto_description = serde_proto_description.0;
     for (ee, getter) in comm
         .endpoint_manager
         .net_endpoint_store
         .endpoint_exts
         .iter_mut()
         .zip(endpoint_incoming_to_getter)
+    {
         ee.getter_for_incoming = getter;
+    }
     // println!("AFTER: {:#?}\n{:#?}", cu, comm);
     Ok(())
+}

0 comments (0 inline, 0 general)