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

Changeset - 4bfd6d133687

Parent rev.

Child rev.

[Not reviewed]

0 5 0

Christopher Esterhuyse - 5 years ago 2020-09-29 09:22:35
christopher.esterhuyse@gmail.com

more unit tests. minor bugfixes in protocol/eval

5 files changed with 109 insertions and 30 deletions:

src/protocol/eval.rs

src/protocol/mod.rs

src/runtime/mod.rs

src/runtime/setup.rs

src/runtime/tests.rs

0 comments (0 inline, 0 general)

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 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(0))))
+                    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
+                }
+            }
             Value::InputArray(_) => todo!(),
             Value::OutputArray(_) => todo!(),
             Value::MessageArray(_) => todo!(),
             Value::BooleanArray(_) => todo!(),
             Value::ByteArray(_) => todo!(),
             Value::ShortArray(_) => todo!(),
             Value::IntArray(_) => todo!(),
             Value::LongArray(_) => todo!(),
             _ => unreachable!(),
+        }
+    }
     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))) => {
                 Value::Boolean(BooleanValue(*s == *o))
+            }
             (Value::Short(ShortValue(s)), Value::Int(IntValue(o))) => {
                 Value::Boolean(BooleanValue(*s as i32 == *o))
+            }
             (Value::Short(ShortValue(s)), Value::Long(LongValue(o))) => {
                 Value::Boolean(BooleanValue(*s as i64 == *o))
+            }
             (Value::Int(IntValue(s)), Value::Byte(ByteValue(o))) => {
                 Value::Boolean(BooleanValue(*s == *o as i32))
+            }
             (Value::Int(IntValue(s)), Value::Short(ShortValue(o))) => {
                 Value::Boolean(BooleanValue(*s == *o as i32))
+            }
             (Value::Int(IntValue(s)), Value::Int(IntValue(o))) => {
                 Value::Boolean(BooleanValue(*s == *o))
+            }
             (Value::Int(IntValue(s)), Value::Long(LongValue(o))) => {
                 Value::Boolean(BooleanValue(*s as i64 == *o))
+            }
             (Value::Long(LongValue(s)), Value::Byte(ByteValue(o))) => {
                 Value::Boolean(BooleanValue(*s == *o as i64))
+            }
             (Value::Long(LongValue(s)), Value::Short(ShortValue(o))) => {
                 Value::Boolean(BooleanValue(*s == *o as i64))
+            }
             (Value::Long(LongValue(s)), Value::Int(IntValue(o))) => {
                 Value::Boolean(BooleanValue(*s == *o as i64))
+            }
             (Value::Long(LongValue(s)), Value::Long(LongValue(o))) => {
                 Value::Boolean(BooleanValue(*s == *o))
+            }
             (Value::Message(MessageValue(s)), Value::Message(MessageValue(o))) => {
                 Value::Boolean(BooleanValue(*s == *o))
+            }
             _ => unimplemented!(),
+        }
+    }
     fn neq(&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))) => {
                 Value::Boolean(BooleanValue(*s != *o))
+            }
             (Value::Short(ShortValue(s)), Value::Int(IntValue(o))) => {
                 Value::Boolean(BooleanValue(*s as i32 != *o))
+            }
             (Value::Short(ShortValue(s)), Value::Long(LongValue(o))) => {
                 Value::Boolean(BooleanValue(*s as i64 != *o))
+            }
             (Value::Int(IntValue(s)), Value::Byte(ByteValue(o))) => {
                 Value::Boolean(BooleanValue(*s != *o as i32))
+            }
             (Value::Int(IntValue(s)), Value::Short(ShortValue(o))) => {
                 Value::Boolean(BooleanValue(*s != *o as i32))
+            }
             (Value::Int(IntValue(s)), Value::Int(IntValue(o))) => {
                 Value::Boolean(BooleanValue(*s != *o))
+            }
             (Value::Int(IntValue(s)), Value::Long(LongValue(o))) => {
                 Value::Boolean(BooleanValue(*s as i64 != *o))
+            }
             (Value::Long(LongValue(s)), Value::Byte(ByteValue(o))) => {
                 Value::Boolean(BooleanValue(*s != *o as i64))
+            }
             (Value::Long(LongValue(s)), Value::Short(ShortValue(o))) => {
                 Value::Boolean(BooleanValue(*s != *o as i64))
+            }
             (Value::Long(LongValue(s)), Value::Int(IntValue(o))) => {
                 Value::Boolean(BooleanValue(*s != *o as i64))
+            }
             (Value::Long(LongValue(s)), Value::Long(LongValue(o))) => {
                 Value::Boolean(BooleanValue(*s != *o))
+            }
             (Value::Message(MessageValue(s)), Value::Message(MessageValue(o))) => {
                 Value::Boolean(BooleanValue(*s != *o))
+            }
             _ => unimplemented!(),
+        }
+    }
     fn lt(&self, other: &Value) -> Value {
         // TODO: match value directly (as done above)
         assert!(!self.exact_type().array);
         assert!(!other.exact_type().array);
         match (self.exact_type().primitive, other.exact_type().primitive) {
             (PrimitiveType::Byte, PrimitiveType::Byte) => {
                 Value::Boolean(BooleanValue(i8::from(self) < i8::from(other)))
+            }
             (PrimitiveType::Byte, PrimitiveType::Short) => {
                 Value::Boolean(BooleanValue(i16::from(self) < i16::from(other)))
+            }
             (PrimitiveType::Byte, PrimitiveType::Int) => {
                 Value::Boolean(BooleanValue(i32::from(self) < i32::from(other)))
+            }
             (PrimitiveType::Byte, PrimitiveType::Long) => {
                 Value::Boolean(BooleanValue(i64::from(self) < i64::from(other)))
+            }
             (PrimitiveType::Short, PrimitiveType::Byte) => {
                 Value::Boolean(BooleanValue(i16::from(self) < i16::from(other)))
+            }
             (PrimitiveType::Short, PrimitiveType::Short) => {
                 Value::Boolean(BooleanValue(i16::from(self) < i16::from(other)))
+            }
             (PrimitiveType::Short, PrimitiveType::Int) => {
                 Value::Boolean(BooleanValue(i32::from(self) < i32::from(other)))
+            }
             (PrimitiveType::Short, PrimitiveType::Long) => {
                 Value::Boolean(BooleanValue(i64::from(self) < i64::from(other)))
+            }
             (PrimitiveType::Int, PrimitiveType::Byte) => {
                 Value::Boolean(BooleanValue(i32::from(self) < i32::from(other)))
+            }
             (PrimitiveType::Int, PrimitiveType::Short) => {
                 Value::Boolean(BooleanValue(i32::from(self) < i32::from(other)))
+            }
             (PrimitiveType::Int, PrimitiveType::Int) => {
                 Value::Boolean(BooleanValue(i32::from(self) < i32::from(other)))
+            }
             (PrimitiveType::Int, PrimitiveType::Long) => {
                 Value::Boolean(BooleanValue(i64::from(self) < i64::from(other)))
+            }
             (PrimitiveType::Long, PrimitiveType::Byte) => {
                 Value::Boolean(BooleanValue(i64::from(self) < i64::from(other)))
+            }
             (PrimitiveType::Long, PrimitiveType::Short) => {
                 Value::Boolean(BooleanValue(i64::from(self) < i64::from(other)))
+            }
             (PrimitiveType::Long, PrimitiveType::Int) => {
                 Value::Boolean(BooleanValue(i64::from(self) < i64::from(other)))
+            }
             (PrimitiveType::Long, PrimitiveType::Long) => {
                 Value::Boolean(BooleanValue(i64::from(self) < i64::from(other)))
+            }
             _ => unimplemented!(),
+        }
+    }
     fn lte(&self, other: &Value) -> Value {
         assert!(!self.exact_type().array);
         assert!(!other.exact_type().array);
         match (self.exact_type().primitive, other.exact_type().primitive) {
             (PrimitiveType::Byte, PrimitiveType::Byte) => {
                 Value::Boolean(BooleanValue(i8::from(self) <= i8::from(other)))
+            }
             (PrimitiveType::Byte, PrimitiveType::Short) => {
                 Value::Boolean(BooleanValue(i16::from(self) <= i16::from(other)))
+            }
             (PrimitiveType::Byte, PrimitiveType::Int) => {
                 Value::Boolean(BooleanValue(i32::from(self) <= i32::from(other)))
+            }
             (PrimitiveType::Byte, PrimitiveType::Long) => {
                 Value::Boolean(BooleanValue(i64::from(self) <= i64::from(other)))
+            }
             (PrimitiveType::Short, PrimitiveType::Byte) => {
                 Value::Boolean(BooleanValue(i16::from(self) <= i16::from(other)))
+            }
             (PrimitiveType::Short, PrimitiveType::Short) => {
                 Value::Boolean(BooleanValue(i16::from(self) <= i16::from(other)))
+            }
             (PrimitiveType::Short, PrimitiveType::Int) => {
                 Value::Boolean(BooleanValue(i32::from(self) <= i32::from(other)))
+            }
             (PrimitiveType::Short, PrimitiveType::Long) => {
                 Value::Boolean(BooleanValue(i64::from(self) <= i64::from(other)))
+            }
             (PrimitiveType::Int, PrimitiveType::Byte) => {
                 Value::Boolean(BooleanValue(i32::from(self) <= i32::from(other)))
+            }
             (PrimitiveType::Int, PrimitiveType::Short) => {
                 Value::Boolean(BooleanValue(i32::from(self) <= i32::from(other)))
+            }
             (PrimitiveType::Int, PrimitiveType::Int) => {
                 Value::Boolean(BooleanValue(i32::from(self) <= i32::from(other)))
+            }
             (PrimitiveType::Int, PrimitiveType::Long) => {
                 Value::Boolean(BooleanValue(i64::from(self) <= i64::from(other)))
+            }
             (PrimitiveType::Long, PrimitiveType::Byte) => {
                 Value::Boolean(BooleanValue(i64::from(self) <= i64::from(other)))
+            }
             (PrimitiveType::Long, PrimitiveType::Short) => {
                 Value::Boolean(BooleanValue(i64::from(self) <= i64::from(other)))
+            }
             (PrimitiveType::Long, PrimitiveType::Int) => {
                 Value::Boolean(BooleanValue(i64::from(self) <= i64::from(other)))
+            }
             (PrimitiveType::Long, PrimitiveType::Long) => {
                 Value::Boolean(BooleanValue(i64::from(self) <= i64::from(other)))
+            }
             _ => unimplemented!(),
+        }
+    }
     fn gt(&self, other: &Value) -> Value {
         assert!(!self.exact_type().array);
         assert!(!other.exact_type().array);
         match (self.exact_type().primitive, other.exact_type().primitive) {
             (PrimitiveType::Byte, PrimitiveType::Byte) => {
                 Value::Boolean(BooleanValue(i8::from(self) > i8::from(other)))
+            }
             (PrimitiveType::Byte, PrimitiveType::Short) => {
                 Value::Boolean(BooleanValue(i16::from(self) > i16::from(other)))
+            }
             (PrimitiveType::Byte, PrimitiveType::Int) => {
                 Value::Boolean(BooleanValue(i32::from(self) > i32::from(other)))
+            }
             (PrimitiveType::Byte, PrimitiveType::Long) => {
                 Value::Boolean(BooleanValue(i64::from(self) > i64::from(other)))
+            }
             (PrimitiveType::Short, PrimitiveType::Byte) => {
                 Value::Boolean(BooleanValue(i16::from(self) > i16::from(other)))
+            }
             (PrimitiveType::Short, PrimitiveType::Short) => {
                 Value::Boolean(BooleanValue(i16::from(self) > i16::from(other)))
+            }
             (PrimitiveType::Short, PrimitiveType::Int) => {
                 Value::Boolean(BooleanValue(i32::from(self) > i32::from(other)))
+            }
             (PrimitiveType::Short, PrimitiveType::Long) => {
                 Value::Boolean(BooleanValue(i64::from(self) > i64::from(other)))
+            }
             (PrimitiveType::Int, PrimitiveType::Byte) => {
                 Value::Boolean(BooleanValue(i32::from(self) > i32::from(other)))
+            }
             (PrimitiveType::Int, PrimitiveType::Short) => {
                 Value::Boolean(BooleanValue(i32::from(self) > i32::from(other)))
+            }
             (PrimitiveType::Int, PrimitiveType::Int) => {
                 Value::Boolean(BooleanValue(i32::from(self) > i32::from(other)))
+            }
             (PrimitiveType::Int, PrimitiveType::Long) => {
                 Value::Boolean(BooleanValue(i64::from(self) > i64::from(other)))
+            }
             (PrimitiveType::Long, PrimitiveType::Byte) => {
                 Value::Boolean(BooleanValue(i64::from(self) > i64::from(other)))
+            }
             (PrimitiveType::Long, PrimitiveType::Short) => {
                 Value::Boolean(BooleanValue(i64::from(self) > i64::from(other)))
+            }
             (PrimitiveType::Long, PrimitiveType::Int) => {
                 Value::Boolean(BooleanValue(i64::from(self) > i64::from(other)))
+            }
             (PrimitiveType::Long, PrimitiveType::Long) => {
                 Value::Boolean(BooleanValue(i64::from(self) > i64::from(other)))
+            }
             _ => unimplemented!(),
+        }
+    }
     fn gte(&self, other: &Value) -> Value {
         assert!(!self.exact_type().array);
         assert!(!other.exact_type().array);
         match (self.exact_type().primitive, other.exact_type().primitive) {
             (PrimitiveType::Byte, PrimitiveType::Byte) => {
                 Value::Boolean(BooleanValue(i8::from(self) >= i8::from(other)))
+            }
             (PrimitiveType::Byte, PrimitiveType::Short) => {
                 Value::Boolean(BooleanValue(i16::from(self) >= i16::from(other)))
+            }
             (PrimitiveType::Byte, PrimitiveType::Int) => {
                 Value::Boolean(BooleanValue(i32::from(self) >= i32::from(other)))
+            }
             (PrimitiveType::Byte, PrimitiveType::Long) => {
                 Value::Boolean(BooleanValue(i64::from(self) >= i64::from(other)))
+            }
             (PrimitiveType::Short, PrimitiveType::Byte) => {
                 Value::Boolean(BooleanValue(i16::from(self) >= i16::from(other)))
+            }
             (PrimitiveType::Short, PrimitiveType::Short) => {
                 Value::Boolean(BooleanValue(i16::from(self) >= i16::from(other)))
+            }
             (PrimitiveType::Short, PrimitiveType::Int) => {
                 Value::Boolean(BooleanValue(i32::from(self) >= i32::from(other)))
+            }
             (PrimitiveType::Short, PrimitiveType::Long) => {
                 Value::Boolean(BooleanValue(i64::from(self) >= i64::from(other)))
+            }
             (PrimitiveType::Int, PrimitiveType::Byte) => {
                 Value::Boolean(BooleanValue(i32::from(self) >= i32::from(other)))
+            }
             (PrimitiveType::Int, PrimitiveType::Short) => {
                 Value::Boolean(BooleanValue(i32::from(self) >= i32::from(other)))
+            }
             (PrimitiveType::Int, PrimitiveType::Int) => {
                 Value::Boolean(BooleanValue(i32::from(self) >= i32::from(other)))
+            }
             (PrimitiveType::Int, PrimitiveType::Long) => {
                 Value::Boolean(BooleanValue(i64::from(self) >= i64::from(other)))
+            }
             (PrimitiveType::Long, PrimitiveType::Byte) => {
                 Value::Boolean(BooleanValue(i64::from(self) >= i64::from(other)))
+            }
             (PrimitiveType::Long, PrimitiveType::Short) => {
                 Value::Boolean(BooleanValue(i64::from(self) >= i64::from(other)))
+            }
             (PrimitiveType::Long, PrimitiveType::Int) => {
                 Value::Boolean(BooleanValue(i64::from(self) >= i64::from(other)))
+            }
             (PrimitiveType::Long, PrimitiveType::Long) => {
                 Value::Boolean(BooleanValue(i64::from(self) >= i64::from(other)))
+            }
             _ => unimplemented!(),
+        }
+    }
     fn as_boolean(&self) -> &BooleanValue {
         match self {
             Value::Boolean(result) => result,
             _ => panic!("Unable to cast `Value` to `BooleanValue`"),
+        }
+    }
+}
 impl From<bool> for Value {
     fn from(b: bool) -> Self {
         Value::Boolean(BooleanValue(b))
+    }
+}
 impl From<Value> for bool {
     fn from(val: Value) -> Self {
         match val {
             Value::Boolean(BooleanValue(b)) => b,
             _ => unimplemented!(),
+        }
+    }
+}
 impl From<&Value> for bool {
     fn from(val: &Value) -> Self {
         match val {
             Value::Boolean(BooleanValue(b)) => *b,
             _ => unimplemented!(),
+        }
+    }
+}
 impl From<Value> for i8 {
     fn from(val: Value) -> Self {
         match val {
             Value::Byte(ByteValue(b)) => b,
             _ => unimplemented!(),
+        }
+    }
+}
 impl From<&Value> for i8 {
     fn from(val: &Value) -> Self {
         match val {
             Value::Byte(ByteValue(b)) => *b,
             _ => unimplemented!(),
+        }
+    }
+}
 impl From<Value> for i16 {
     fn from(val: Value) -> Self {
         match val {
             Value::Byte(ByteValue(b)) => i16::from(b),
             Value::Short(ShortValue(s)) => s,
             _ => unimplemented!(),
+        }
+    }
+}
 impl From<&Value> for i16 {
     fn from(val: &Value) -> Self {
         match val {
             Value::Byte(ByteValue(b)) => i16::from(*b),
             Value::Short(ShortValue(s)) => *s,
             _ => unimplemented!(),
+        }
+    }
+}
 impl From<Value> for i32 {
     fn from(val: Value) -> Self {
         match val {
             Value::Byte(ByteValue(b)) => i32::from(b),
             Value::Short(ShortValue(s)) => i32::from(s),
             Value::Int(IntValue(i)) => i,
             _ => unimplemented!(),
+        }
+    }
+}
 impl From<&Value> for i32 {
     fn from(val: &Value) -> Self {
         match val {
             Value::Byte(ByteValue(b)) => i32::from(*b),
             Value::Short(ShortValue(s)) => i32::from(*s),
             Value::Int(IntValue(i)) => *i,
             _ => unimplemented!(),
+        }
+    }
+}
 impl From<Value> for i64 {
     fn from(val: Value) -> Self {
         match val {
             Value::Byte(ByteValue(b)) => i64::from(b),
             Value::Short(ShortValue(s)) => i64::from(s),
             Value::Int(IntValue(i)) => i64::from(i),
             Value::Long(LongValue(l)) => l,
             _ => unimplemented!(),
+        }
+    }
+}
 impl From<&Value> for i64 {
     fn from(val: &Value) -> Self {
         match val {
             Value::Byte(ByteValue(b)) => i64::from(*b),
             Value::Short(ShortValue(s)) => i64::from(*s),
             Value::Int(IntValue(i)) => i64::from(*i),
             Value::Long(LongValue(l)) => *l,
             _ => unimplemented!(),
+        }
+    }
+}
 impl ValueImpl for Value {
     fn exact_type(&self) -> Type {
         match self {
             Value::Input(val) => val.exact_type(),
             Value::Output(val) => val.exact_type(),
             Value::Message(val) => val.exact_type(),
             Value::Boolean(val) => val.exact_type(),
             Value::Byte(val) => val.exact_type(),
             Value::Short(val) => val.exact_type(),
             Value::Int(val) => val.exact_type(),
             Value::Long(val) => val.exact_type(),
             Value::InputArray(val) => val.exact_type(),
             Value::OutputArray(val) => val.exact_type(),
             Value::MessageArray(val) => val.exact_type(),
             Value::BooleanArray(val) => val.exact_type(),
             Value::ByteArray(val) => val.exact_type(),
             Value::ShortArray(val) => val.exact_type(),
             Value::IntArray(val) => val.exact_type(),
             Value::LongArray(val) => val.exact_type(),
+        }
+    }
     fn is_type_compatible(&self, t: &Type) -> bool {
         match self {
             Value::Input(val) => val.is_type_compatible(t),
             Value::Output(val) => val.is_type_compatible(t),
             Value::Message(val) => val.is_type_compatible(t),
             Value::Boolean(val) => val.is_type_compatible(t),
             Value::Byte(val) => val.is_type_compatible(t),
             Value::Short(val) => val.is_type_compatible(t),
             Value::Int(val) => val.is_type_compatible(t),
             Value::Long(val) => val.is_type_compatible(t),
             Value::InputArray(val) => val.is_type_compatible(t),
             Value::OutputArray(val) => val.is_type_compatible(t),
             Value::MessageArray(val) => val.is_type_compatible(t),
             Value::BooleanArray(val) => val.is_type_compatible(t),
             Value::ByteArray(val) => val.is_type_compatible(t),
             Value::ShortArray(val) => val.is_type_compatible(t),
             Value::IntArray(val) => val.is_type_compatible(t),
             Value::LongArray(val) => val.is_type_compatible(t),
+        }
+    }
+}
 impl Display for Value {
     fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
         let disp: &dyn Display;
         match self {
             Value::Input(val) => disp = val,
             Value::Output(val) => disp = val,
             Value::Message(val) => disp = val,
             Value::Boolean(val) => disp = val,
             Value::Byte(val) => disp = val,
             Value::Short(val) => disp = val,
             Value::Int(val) => disp = val,
             Value::Long(val) => disp = val,
             Value::InputArray(val) => disp = val,
             Value::OutputArray(val) => disp = val,
             Value::MessageArray(val) => disp = val,
             Value::BooleanArray(val) => disp = val,
             Value::ByteArray(val) => disp = val,
             Value::ShortArray(val) => disp = val,
             Value::IntArray(val) => disp = val,
             Value::LongArray(val) => disp = val,
+        }
         disp.fmt(f)
+    }
+}
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 pub struct InputValue(pub PortId);
 impl Display for InputValue {
     fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
         write!(f, "#in")
+    }
+}
 impl ValueImpl for InputValue {
     fn exact_type(&self) -> Type {
         Type::INPUT
+    }
     fn is_type_compatible(&self, t: &Type) -> bool {
         let Type { primitive, array } = t;
         if *array {
             return false;
+        }
         match primitive {
             PrimitiveType::Input => true,
             _ => false,
+        }
+    }
+}
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 pub struct OutputValue(pub PortId);
 impl Display for OutputValue {
     fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
         write!(f, "#out")
+    }
+}
 impl ValueImpl for OutputValue {
     fn exact_type(&self) -> Type {
         Type::OUTPUT
+    }
     fn is_type_compatible(&self, t: &Type) -> bool {
         let Type { primitive, array } = t;
         if *array {
             return false;
+        }
         match primitive {
             PrimitiveType::Output => true,
             _ => false,
+        }
+    }
+}
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 pub struct MessageValue(pub Option<Payload>);
 impl Display for MessageValue {
     fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
         match &self.0 {
             None => write!(f, "null"),
             Some(payload) => {
                 // format print up to 10 bytes
                 let mut slice = payload.as_slice();
                 if slice.len() > 10 {
                     slice = &slice[..10];
+                }
                 f.debug_list().entries(slice.iter().copied()).finish()
+            }
+        }
+    }
+}
 impl ValueImpl for MessageValue {
     fn exact_type(&self) -> Type {
         Type::MESSAGE
+    }
     fn is_type_compatible(&self, t: &Type) -> bool {
         let Type { primitive, array } = t;
         if *array {
             return false;
+        }
         match primitive {
             PrimitiveType::Message => true,
             _ => false,
+        }
+    }
+}
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 pub struct BooleanValue(bool);
 impl Display for BooleanValue {
     fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
         write!(f, "{}", self.0)
+    }
+}
 impl ValueImpl for BooleanValue {
     fn exact_type(&self) -> Type {
         Type::BOOLEAN
+    }
     fn is_type_compatible(&self, t: &Type) -> bool {
         let Type { primitive, array } = t;
         if *array {
             return false;
+        }
         match primitive {
             PrimitiveType::Boolean => true,
             PrimitiveType::Byte => true,
             PrimitiveType::Short => true,
             PrimitiveType::Int => true,
             PrimitiveType::Long => true,
             _ => false,
+        }
+    }
+}
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 pub struct ByteValue(i8);
 impl Display for ByteValue {
     fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
         write!(f, "{}", self.0)
+    }
+}
 impl ValueImpl for ByteValue {
     fn exact_type(&self) -> Type {
         Type::BYTE
+    }
     fn is_type_compatible(&self, t: &Type) -> bool {
         let Type { primitive, array } = t;
         if *array {
             return false;
+        }
         match primitive {
             PrimitiveType::Byte => true,
             PrimitiveType::Short => true,
             PrimitiveType::Int => true,
             PrimitiveType::Long => true,
             _ => false,
+        }
+    }
+}
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 pub struct ShortValue(i16);
 impl Display for ShortValue {
     fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
         write!(f, "{}", self.0)
+    }
+}
 impl ValueImpl for ShortValue {

src/protocol/mod.rs

➞

Show inline comments

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

src/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 UdpInBuffer {
     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>,
     udp_in_buffer: UdpInBuffer,
+}
 // 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 {
     map: HashMap<PortId, PortInfo>,
+}
 // 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);
+}
 // 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,
     NewControlMsg { net_index: usize, msg: CommCtrlMsg },
+}
 ////////////////
 fn err_would_block(err: &std::io::Error) -> bool {
     err.kind() == std::io::ErrorKind::WouldBlock
+}
 impl<T: std::cmp::Ord> VecSet<T> {
     fn new(mut vec: Vec<T>) -> Self {
         // establish the invariant
         vec.sort();
         vec.dedup();
         Self { vec }
+    }
     fn contains(&self, element: &T) -> bool {
         self.vec.binary_search(element).is_ok()
+    }
     // Insert the given element. Returns whether it was already present.
     fn insert(&mut self, element: T) -> bool {
         match self.vec.binary_search(&element) {
             Ok(_) => false,
             Err(index) => {
                 self.vec.insert(index, element);
                 true
+            }
+        }
+    }
     fn iter(&self) -> std::slice::Iter<T> {
         self.vec.iter()
+    }
     fn pop(&mut self) -> Option<T> {
         self.vec.pop()
+    }
+}
 impl PortInfoMap {
     fn ports_owned_by(&self, owner: ComponentId) -> impl Iterator<Item = &PortId> {
         self.map.iter().filter(move |(_, port_info)| port_info.owner == owner).map(|(port, _)| port)
+    }
     fn spec_var_for(&self, port: PortId) -> SpecVar {
         // Every port maps to a speculative variable
         // Two distinct ports map to the same variable
         // IFF they are two ends of the same logical channel.
         let info = self.map.get(&port).unwrap();
         SpecVar(match info.polarity {
             Getter => port,
             Putter => info.peer.unwrap(),
         })
+    }
+}
 impl SpecVarStream {
     fn next(&mut self) -> SpecVar {
         let phantom_port: PortId =
             Id { connector_id: self.connector_id, u32_suffix: self.port_suffix_stream.next() }
                 .into();
         SpecVar(phantom_port)
+    }
+}
 impl IdManager {
     fn new(connector_id: ConnectorId) -> Self {
         Self {
             connector_id,
             port_suffix_stream: Default::default(),
             component_suffix_stream: Default::default(),
+        }
+    }
     fn new_spec_var_stream(&self) -> SpecVarStream {
         // Spec var stream starts where the current port_id stream ends, with gap of SKIP_N.
         // This gap is entirely unnecessary (i.e. 0 is fine)
         // It's purpose is only to make SpecVars easier to spot in logs.
         // E.g. spot the spec var: { v0_0, v1_2, v1_103 }
         const SKIP_N: u32 = 100;
         let port_suffix_stream = self.port_suffix_stream.clone().n_skipped(SKIP_N);
         SpecVarStream { connector_id: self.connector_id, port_suffix_stream }
+    }
     fn new_port_id(&mut self) -> PortId {
         Id { connector_id: self.connector_id, u32_suffix: self.port_suffix_stream.next() }.into()
+    }
     fn new_component_id(&mut self) -> ComponentId {
         Id { connector_id: self.connector_id, u32_suffix: self.component_suffix_stream.next() }
             .into()
+    }
+}
 impl Drop for Connector {
     fn drop(&mut self) {
         log!(self.unphased.logger(), "Connector dropping. Goodbye!");
+    }
+}
 // Given a slice of ports, return the first, if any, port is present repeatedly
 fn duplicate_port(slice: &[PortId]) -> Option<PortId> {
     let mut vec = Vec::with_capacity(slice.len());
     for port in slice.iter() {
         match vec.binary_search(port) {
             Err(index) => vec.insert(index, *port),
             Ok(_) => return Some(*port),
+        }
+    }
     None
+}
 impl Connector {
     /// Generate a random connector identifier from the system's source of randomness.
     pub fn random_id() -> ConnectorId {
         type Bytes8 = [u8; std::mem::size_of::<ConnectorId>()];
         unsafe {
             let mut bytes = std::mem::MaybeUninit::<Bytes8>::uninit();
             // getrandom is the canonical crate for a small, secure rng
             getrandom::getrandom(&mut *bytes.as_mut_ptr()).unwrap();
             // safe! representations of all valid Byte8 values are valid ConnectorId values
             std::mem::transmute::<_, _>(bytes.assume_init())
+        }
+    }
     /// Returns true iff the connector is in connected state, i.e., it's setup phase is complete,
     /// and it is ready to participate in synchronous rounds of communication.
     pub fn is_connected(&self) -> bool {
         // If designed for Rust usage, connectors would be exposed as an enum type from the start.
         // consequently, this "phased" business would also include connector variants and this would
         // get a lot closer to the connector impl. itself.
         // Instead, the C-oriented implementation doesn't distinguish connector states as types,
         // and distinguish them as enum variants instead
         match self.phased {
             ConnectorPhased::Setup(..) => false,
             ConnectorPhased::Communication(..) => true,
+        }
+    }
     /// Enables the connector's current logger to be swapped out for another
     pub fn swap_logger(&mut self, mut new_logger: Box<dyn Logger>) -> Box<dyn Logger> {
         std::mem::swap(&mut self.unphased.logger, &mut new_logger);
         new_logger
+    }
     /// Access the connector's current logger
     pub fn get_logger(&mut self) -> &mut dyn Logger {
         &mut *self.unphased.logger
+    }
     /// Create a new synchronous channel, returning its ends as a pair of ports,
     /// with polarity output, input respectively. Available during either setup/communication phase.
     /// # Panics
     /// This function panics if the connector's (large) port id space is exhausted.
     pub fn new_port_pair(&mut self) -> [PortId; 2] {
         let cu = &mut self.unphased;
         // adds two new associated ports, related to each other, and exposed to the native
         let mut new_cid = || cu.ips.id_manager.new_port_id();
         // allocate two fresh port identifiers
         let [o, i] = [new_cid(), new_cid()];
         // store info for each:
         // - they are each others' peers
         // - they are owned by a local component with id `cid`
         // - polarity putter, getter respectively
         cu.ips.port_info.map.insert(
             o,
             PortInfo {
                 route: Route::LocalComponent,
                 peer: Some(i),
                 owner: cu.native_component_id,
                 polarity: Putter,
             },
         );
         cu.ips.port_info.map.insert(
             i,
             PortInfo {
                 route: Route::LocalComponent,
                 peer: Some(o),
                 owner: cu.native_component_id,
                 polarity: Getter,
             },
         );
         log!(cu.logger, "Added port pair (out->in) {:?} -> {:?}", o, i);
         [o, i]
+    }
     /// Instantiates a new component for the connector runtime to manage, and passing
     /// the given set of ports from the interface of the native component, to that of the
     /// newly created component (passing their ownership).
     /// # Errors
     /// Error is returned if the moved ports are not owned by the native component,
     /// if the given component name is not defined in the connector's protocol,
     /// the given sequence of ports contains a duplicate port,
     /// or if the component is unfit for instantiation with the given port sequence.
     /// # Panics
     /// This function panics if the connector's (large) component id space is exhausted.
     pub fn add_component(
         &mut self,
         identifier: &[u8],
         ports: &[PortId],
     ) -> Result<(), AddComponentError> {
         // Check for error cases first before modifying `cu`
         use AddComponentError as Ace;
         let cu = &self.unphased;
         if let Some(port) = duplicate_port(ports) {
             return Err(Ace::DuplicatePort(port));
+        }
         let expected_polarities = cu.proto_description.component_polarities(identifier)?;
         if expected_polarities.len() != ports.len() {
             return Err(Ace::WrongNumberOfParamaters { expected: expected_polarities.len() });
+        }
         for (&expected_polarity, &port) in expected_polarities.iter().zip(ports.iter()) {
             let info = cu.ips.port_info.map.get(&port).ok_or(Ace::UnknownPort(port))?;
             if info.owner != cu.native_component_id {
                 return Err(Ace::UnknownPort(port));
+            }
             if info.polarity != expected_polarity {
                 return Err(Ace::WrongPortPolarity { port, expected_polarity });
+            }
+        }
         // No errors! Time to modify `cu`
         // create a new component and identifier
         let cu = &mut self.unphased;
         let new_cid = cu.ips.id_manager.new_component_id();
         cu.proto_components
             .insert(new_cid, cu.proto_description.new_main_component(identifier, ports));
         cu.proto_components.insert(new_cid, cu.proto_description.new_component(identifier, ports));
         // update the ownership of moved ports
         for port in ports.iter() {
             match cu.ips.port_info.map.get_mut(port) {
                 Some(port_info) => port_info.owner = new_cid,
                 None => unreachable!(),
+            }
+        }
         Ok(())
+    }
+}
 impl Predicate {
     #[inline]
     pub fn singleton(k: SpecVar, v: SpecVal) -> Self {
         Self::default().inserted(k, v)
+    }
     #[inline]
     pub fn inserted(mut self, k: SpecVar, v: SpecVal) -> Self {
         self.assigned.insert(k, v);
         self
+    }
     // Return true whether `self` is a subset of `maybe_superset`
     pub fn assigns_subset(&self, maybe_superset: &Self) -> bool {
         for (var, val) in self.assigned.iter() {
             match maybe_superset.assigned.get(var) {
                 Some(val2) if val2 == val => {}
                 _ => return false, // var unmapped, or mapped differently
+            }
+        }
         // `maybe_superset` mirrored all my assignments!
         true
+    }
     /// Given the two predicates {self, other}, return that whose
     /// assignments are the union of those of both.
     fn assignment_union(&self, other: &Self) -> AssignmentUnionResult {
         use AssignmentUnionResult as Aur;
         // iterators over assignments of both predicates. Rely on SORTED ordering of BTreeMap's keys.
         let [mut s_it, mut o_it] = [self.assigned.iter(), other.assigned.iter()];
         let [mut s, mut o] = [s_it.next(), o_it.next()];
         // populate lists of assignments in self but not other and vice versa.
         // do this by incrementally unfolding the iterators, keeping an eye
         // on the ordering between the head elements [s, o].
         // whenever s<o, other is certainly missing element 's', etc.
         let [mut s_not_o, mut o_not_s] = [vec![], vec![]];
         loop {
             match [s, o] {
                 [None, None] => break, // both iterators are empty
                 [None, Some(x)] => {
                     // self's iterator is empty.
                     // all remaning elements are in other but not self
                     o_not_s.push(x);
                     o_not_s.extend(o_it);
                     break;
+                }
                 [Some(x), None] => {
                     // other's iterator is empty.
                     // all remaning elements are in self but not other
                     s_not_o.push(x);
                     s_not_o.extend(s_it);
                     break;
+                }
                 [Some((sid, sb)), Some((oid, ob))] => {
                     if sid < oid {
                         // o is missing this element
                         s_not_o.push((sid, sb));
                         s = s_it.next();
                     } else if sid > oid {
                         // s is missing this element
                         o_not_s.push((oid, ob));
                         o = o_it.next();
                     } else if sb != ob {
                         assert_eq!(sid, oid);
                         // both predicates assign the variable but differ on the value
                         // No predicate exists which satisfies both!
                         return Aur::Nonexistant;
                     } else {
                         // both predicates assign the variable to the same value
                         s = s_it.next();
                         o = o_it.next();
+                    }
+                }
+            }
+        }
         // Observed zero inconsistencies. A unified predicate exists...
         match [s_not_o.is_empty(), o_not_s.is_empty()] {
             [true, true] => Aur::Equivalent,       // ... equivalent to both.
             [false, true] => Aur::FormerNotLatter, // ... equivalent to self.
             [true, false] => Aur::LatterNotFormer, // ... equivalent to other.
             [false, false] => {
                 // ... which is the union of the predicates' assignments but
                 //     is equivalent to neither self nor other.
                 let mut new = self.clone();
                 for (&id, &b) in o_not_s {
                     new.assigned.insert(id, b);
+                }
                 Aur::New(new)
+            }
+        }
+    }
     // Compute the union of the assignments of the two given predicates, if it exists.
     // It doesn't exist if there is some value which the predicates assign to different values.
     pub(crate) fn union_with(&self, other: &Self) -> Option<Self> {
         let mut res = self.clone();
         for (&channel_id, &assignment_1) in other.assigned.iter() {
             match res.assigned.insert(channel_id, assignment_1) {
                 Some(assignment_2) if assignment_1 != assignment_2 => return None,
                 _ => {}
+            }
+        }
         Some(res)
+    }
     pub(crate) fn query(&self, var: SpecVar) -> Option<SpecVal> {
         self.assigned.get(&var).copied()
+    }
+}
 impl RoundCtx {
     // remove an arbitrary buffered message, along with the ID of the getter who receives it
     fn getter_pop(&mut self) -> Option<(PortId, SendPayloadMsg)> {
         self.payload_inbox.pop()
+    }
     // buffer a message along with the ID of the getter who receives it
     fn getter_push(&mut self, getter: PortId, msg: SendPayloadMsg) {
         self.payload_inbox.push((getter, msg));
+    }
     // buffer a message along with the ID of the putter who sent it
     fn putter_push(&mut self, cu: &mut impl CuUndecided, putter: PortId, msg: SendPayloadMsg) {
         if let Some(getter) = self.ips.port_info.map.get(&putter).unwrap().peer {
             log!(cu.logger(), "Putter add (putter:{:?} => getter:{:?})", putter, getter);
             self.getter_push(getter, msg);
         } else {
             log!(cu.logger(), "Putter {:?} has no known peer!", putter);
             panic!("Putter {:?} has no known peer!");
+        }
+    }
+}
 impl<T: Debug + std::cmp::Ord> Debug for VecSet<T> {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         f.debug_set().entries(self.vec.iter()).finish()
+    }
+}
 impl Debug for Predicate {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         struct Assignment<'a>((&'a SpecVar, &'a SpecVal));
         impl Debug for Assignment<'_> {
             fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
                 write!(f, "{:?}={:?}", (self.0).0, (self.0).1)
+            }
+        }
         f.debug_set().entries(self.assigned.iter().map(Assignment)).finish()
+    }
+}
 impl serde::Serialize for SerdeProtocolDescription {
     fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
     where
         S: serde::Serializer,
+    {
         let inner: &ProtocolDescription = &self.0;
         inner.serialize(serializer)
+    }
+}
 impl<'de> serde::Deserialize<'de> for SerdeProtocolDescription {
     fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
     where
         D: serde::Deserializer<'de>,
+    {
         let inner: ProtocolDescription = ProtocolDescription::deserialize(deserializer)?;
         Ok(Self(Arc::new(inner)))
+    }
+}
 impl IdParts for SpecVar {
     fn id_parts(self) -> (ConnectorId, U32Suffix) {
         self.0.id_parts()
+    }
+}
 impl Debug for SpecVar {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         let (a, b) = self.id_parts();
         write!(f, "v{}_{}", a, b)
+    }
+}
 impl SpecVal {
     const FIRING: Self = SpecVal(1);
     const SILENT: Self = SpecVal(0);
     fn is_firing(self) -> bool {
         self == Self::FIRING
         // all else treated as SILENT
+    }
     fn iter_domain() -> impl Iterator<Item = Self> {
         (0..).map(SpecVal)
+    }
+}
 impl Debug for SpecVal {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         self.0.fmt(f)
+    }
+}
 impl Default for UdpInBuffer {
     fn default() -> Self {
         let mut byte_vec = Vec::with_capacity(Self::CAPACITY);
         unsafe {
             // safe! this vector is guaranteed to have sufficient capacity
             byte_vec.set_len(Self::CAPACITY);
+        }
         Self { byte_vec }
+    }
+}
 impl UdpInBuffer {
     const CAPACITY: usize = u16::MAX as usize;
     fn as_mut_slice(&mut self) -> &mut [u8] {
         self.byte_vec.as_mut_slice()
+    }
+}
 impl Debug for UdpInBuffer {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         write!(f, "UdpInBuffer")
+    }
+}

src/runtime/setup.rs

➞

Show inline comments

 use crate::common::*;
 use crate::runtime::*;
 #[derive(Default)]
 struct ExtraPortInfo {
     info: HashMap<PortId, PortInfo>,
     peers: HashMap<PortId, PortId>,
+}
 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,
                 });
                 // 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,
                 });
                 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) => {
                 log!(cu.logger, "~~~ CONNECT called timeout {:?}", timeout);
                 let deadline = timeout.map(|to| Instant::now() + to);
                 // connect all endpoints in parallel; send and receive peer ids through ports
                 let (mut endpoint_manager, mut extra_port_info) = setup_endpoints_and_pair_ports(
                     &mut *cu.logger,
                     &setup.net_endpoint_setups,
                     &setup.udp_endpoint_setups,
                     &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(cu, &mut comm, &deadline)?;
+                }
                 log!(cu.logger, "connect() finished. setup phase complete");
                 // Connect procedure successful! Commit changes by...
                 // ... commiting new port info for ConnectorUnphased
                 for (port, info) in extra_port_info.info.drain() {
                     cu.ips.port_info.map.insert(port, info);
+                }
                 for (port, peer) in extra_port_info.peers.drain() {
                     cu.ips.port_info.map.get_mut(&port).unwrap().peer = Some(peer);
+                }
                 // ... replacing the connector's phase to "communication"
                 *phased = ConnectorPhased::Communication(Box::new(comm));
                 Ok(())
+            }
+        }
+    }
+}
 // 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: &PortInfoMap,
     deadline: &Option<Instant>,
 ) -> Result<(EndpointManager, ExtraPortInfo), ConnectError> {
     use ConnectError as Ce;
     const BOTH: Interest = Interest::READABLE.add(Interest::WRITABLE);
     const RETRY_PERIOD: Duration = Duration::from_millis(200);
     // The structure shared between this ("setup") thread and that of the waker.
     // The waker thread periodically sends signals.
     // struct WakerState {
     //     continue_signal: AtomicBool,
     //     waker: mio::Waker,
     // }
     // impl WakerState {
     //     // The waker thread runs this UNTIL the continue signal is set to false
     //     fn waker_loop(&self) {
     //         while self.continue_signal.load(SeqCst) {
     //             std::thread::sleep(WAKER_PERIOD);
     //             let _ = self.waker.wake();
     //         }
     //     }
     //     // The setup thread thread runs this to set the continue signal to false.
     //     fn waker_stop(&self) {
     //         self.continue_signal.store(false, SeqCst);
     //     }
     // }
     // 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 waker_state: Option<Arc<WakerState>> = None;
     let mut extra_port_info = ExtraPortInfo::default();
     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();
     // 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)
                     .expect("mio::TcpStream connect should not fail!");
                 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();
             Ok(UdpTodo { sock, getter_for_incoming: endpoint_setup.getter_for_incoming })
         })
         .collect::<Result<Vec<UdpTodo>, ConnectError>>()?;
     // Initially no net connections have failed, and all udp and net endpoint setups are incomplete
     let mut net_connect_to_retry: HashSet<usize> = Default::default();
     let mut setup_incomplete: HashSet<TokenTarget> = {
         let net_todo_targets_iter =
             (0..net_todos.len()).map(|index| TokenTarget::NetEndpoint { index });
         let udp_todo_targets_iter =
             (0..udp_todos.len()).map(|index| TokenTarget::UdpEndpoint { index });
         net_todo_targets_iter.chain(udp_todo_targets_iter).collect()
     };
     // progress by reacting to poll events. continue until every endpoint is set up
     while !setup_incomplete.is_empty() {
         // recompute the time left to poll for progress
         let remaining = if let Some(deadline) = deadline {
             deadline.checked_duration_since(Instant::now()).ok_or(Ce::Timeout)?.min(RETRY_PERIOD)
         } else {
             RETRY_PERIOD
         // recompute the timeout for the poll call
         let remaining = match (deadline, net_connect_to_retry.is_empty()) {
             (None, true) => None,
             (None, false) => Some(RETRY_PERIOD),
             (Some(deadline), is_empty) => {
                 let dur_to_timeout =
                     deadline.checked_duration_since(Instant::now()).ok_or(Ce::Timeout)?;
                 Some(if is_empty { dur_to_timeout } else { dur_to_timeout.min(RETRY_PERIOD) })
+            }
         };
         // block until either
         // (a) `events` has been populated with 1+ elements
         // (b) timeout elapses, or
         // (c) RETRY_PERIOD elapses
-        poll.poll(&mut events, Some(remaining)).map_err(|_| Ce::PollFailed)?;
         poll.poll(&mut events, remaining).map_err(|_| Ce::PollFailed)?;
         if last_retry_at.elapsed() > RETRY_PERIOD {
             // Retry all net connections and reset `last_retry_at`
             last_retry_at = Instant::now();
             for net_index in net_connect_to_retry.drain() {
                 // Restart connect procedure for this net endpoint
                 let net_todo = &mut net_todos[net_index];
                 log!(
                     logger,
                     "Restarting connection with endpoint {:?} {:?}",
                     net_index,
                     net_todo.endpoint_setup.sock_addr
                 );
                 match &mut net_todo.todo_endpoint {
                     NetTodoEndpoint::PeerInfoRecving(endpoint) => {
                         let mut new_stream = TcpStream::connect(net_todo.endpoint_setup.sock_addr)
                             .expect("mio::TcpStream connect should not fail!");
                         std::mem::swap(&mut endpoint.stream, &mut new_stream);
                         let token = TokenTarget::NetEndpoint { index: net_index }.into();
                         poll.registry().register(&mut endpoint.stream, token, BOTH).unwrap();
+                    }
                     _ => unreachable!(),
+                }
+            }
+        }
         for event in events.iter() {
             let token = event.token();
             // figure out which endpoint the event belonged to
             let token_target = TokenTarget::from(token);
             match token_target {
                 TokenTarget::UdpEndpoint { index } => {
                     // UdpEndpoints are easy to complete.
                     // Their setup event just has to succeed without error
                     if !setup_incomplete.contains(&token_target) {
                         // spurious wakeup. this endpoint has already been set up!
                         continue;
+                    }
                     let udp_todo: &UdpTodo = &udp_todos[index];
                     if event.is_error() {
                         return Err(Ce::BindFailed(udp_todo.sock.local_addr().unwrap()));
+                    }
                     setup_incomplete.remove(&token_target);
+                }
                 TokenTarget::NetEndpoint { index } => {
                     // NetEndpoints are complex to complete,
                     // they must accept/connect to their peer,
                     // and then exchange port info successfully
                     let net_todo = &mut net_todos[index];
                     if let NetTodoEndpoint::Accepting(listener) = &mut net_todo.todo_endpoint {
                         // Passive endpoint that will first try accept the peer's connection
                         match listener.accept() {
                             Err(e) if err_would_block(&e) => continue, // spurious wakeup
                             Err(_) => {
                                 log!(logger, "accept() failure on index {}", index);
                                 return Err(Ce::AcceptFailed(listener.local_addr().unwrap()));
+                            }
                             Ok((mut stream, peer_addr)) => {
                                 // successfully accepted the active peer
                                 // reusing the token, but now for the stream and not the listener
                                 poll.registry().deregister(listener).unwrap();
                                 poll.registry().register(&mut stream, token, BOTH).unwrap();
                                 log!(
                                     logger,
                                     "Endpoint[{}] accepted a connection from {:?}",
                                     index,
                                     peer_addr
                                 );
                                 let net_endpoint = NetEndpoint { stream, inbox: vec![] };
                                 net_todo.todo_endpoint =
                                     NetTodoEndpoint::PeerInfoRecving(net_endpoint);
+                            }
+                        }
+                    }
                     // OK now let's try and finish exchanging port info
                     if let NetTodoEndpoint::PeerInfoRecving(net_endpoint) =
                         &mut net_todo.todo_endpoint
+                    {
                         if event.is_error() {
                             // event signals some error! :(
                             if net_todo.endpoint_setup.endpoint_polarity
                                 == EndpointPolarity::Passive
+                            {
                                 // breaking as the acceptor is currently unrecoverable
                                 return Err(Ce::AcceptFailed(
                                     net_endpoint.stream.local_addr().unwrap(),
                                 ));
+                            }
                             // this actively-connecting endpoint failed to connect!
                             // We will schedule it for a retry
                             net_connect_to_retry.insert(index);
                             continue;
+                        }
                         // event wasn't ERROR
                         if net_connect_to_retry.contains(&index) {
                             // spurious wakeup. already scheduled to retry connect later
                             continue;
+                        }
                         if !setup_incomplete.contains(&token_target) {
                             // spurious wakeup. this endpoint has already been completed!
                             if event.is_readable() {
                                 net_polled_undrained.insert(index);
+                            }
                             continue;
+                        }
                         let local_info = port_info
                             .map
                             .get(&net_todo.endpoint_setup.getter_for_incoming)
                             .expect("Net Setup's getter port info isn't known"); // unreachable
                         if event.is_writable() && !net_todo.sent_local_port {
                             // can write and didn't send setup msg yet? Do so!
                             let msg = Msg::SetupMsg(SetupMsg::MyPortInfo(MyPortInfo {
                                 owner: local_info.owner,
                                 polarity: local_info.polarity,
                                 port: net_todo.endpoint_setup.getter_for_incoming,
                             }));
                             net_endpoint
                                 .send(&msg)
                                 .map_err(|e| {
                                     Ce::NetEndpointSetupError(
                                         net_endpoint.stream.local_addr().unwrap(),
                                         e,
+                                    )
                                 })
                                 .unwrap();
                             log!(logger, "endpoint[{}] sent msg {:?}", index, &msg);
                             net_todo.sent_local_port = true;
+                        }
                         if event.is_readable() && net_todo.recv_peer_port.is_none() {
                             // can read and didn't finish recving setup msg yet? Do so!
                             let maybe_msg = net_endpoint.try_recv(logger).map_err(|e| {
                                 Ce::NetEndpointSetupError(
                                     net_endpoint.stream.local_addr().unwrap(),
                                     e,
+                                )
                             })?;
                             if maybe_msg.is_some() && !net_endpoint.inbox.is_empty() {
                                 net_polled_undrained.insert(index);
+                            }
                             match maybe_msg {
                                 None => {} // msg deserialization incomplete
                                 Some(Msg::SetupMsg(SetupMsg::MyPortInfo(peer_info))) => {
                                     log!(
                                         logger,
                                         "endpoint[{}] got peer info {:?}",
                                         index,
                                         peer_info
                                     );
                                     if peer_info.polarity == local_info.polarity {
                                         return Err(ConnectError::PortPeerPolarityMismatch(
                                             net_todo.endpoint_setup.getter_for_incoming,
                                         ));
+                                    }
                                     net_todo.recv_peer_port = Some(peer_info.port);
                                     // finally learned the peer of this port!
                                     extra_port_info.peers.insert(
                                         net_todo.endpoint_setup.getter_for_incoming,
                                         peer_info.port,
                                     );
                                     // learned the info of this peer port
                                     if !port_info.map.contains_key(&peer_info.port) {
                                         let info = PortInfo {
                                             peer: Some(net_todo.endpoint_setup.getter_for_incoming),
                                             polarity: peer_info.polarity,
                                             owner: peer_info.owner,
                                             route: Route::NetEndpoint { index },
                                         };
                                         extra_port_info.info.insert(peer_info.port, info);
+                                    }
+                                }
                                 Some(inappropriate_msg) => {
                                     log!(
                                         logger,
                                         "delaying msg {:?} during channel setup phase",
                                         inappropriate_msg
                                     );
                                     delayed_messages.push((index, inappropriate_msg));
+                                }
+                            }
+                        }
                         // is the setup for this net_endpoint now complete?
                         if net_todo.sent_local_port && net_todo.recv_peer_port.is_some() {
                             // yes! connected, sent my info and received peer's info
                             setup_incomplete.remove(&token_target);
                             log!(logger, "endpoint[{}] is finished!", index);
+                        }
+                    }
+                }
+            }
+        }
         events.clear();
+    }
     log!(logger, "Endpoint setup complete! Cleaning up and building structures");
     let net_endpoint_exts = net_todos
         .into_iter()
         .enumerate()
         .map(|(index, NetTodo { todo_endpoint, endpoint_setup, .. })| NetEndpointExt {
             net_endpoint: match todo_endpoint {
                 NetTodoEndpoint::PeerInfoRecving(mut net_endpoint) => {
                     let token = TokenTarget::NetEndpoint { index }.into();
                     poll.registry()
                         .reregister(&mut net_endpoint.stream, token, Interest::READABLE)
                         .unwrap();
                     net_endpoint
+                }
                 _ => unreachable!(),
             },
             getter_for_incoming: endpoint_setup.getter_for_incoming,
         })
         .collect();
     let udp_endpoint_exts = udp_todos
         .into_iter()
         .enumerate()
         .map(|(index, udp_todo)| {
             let UdpTodo { mut sock, getter_for_incoming } = udp_todo;
             let token = TokenTarget::UdpEndpoint { index }.into();
             poll.registry().reregister(&mut sock, token, Interest::READABLE).unwrap();
             UdpEndpointExt {
                 sock,
                 outgoing_payloads: Default::default(),
                 received_this_round: false,
                 getter_for_incoming,
+            }
         })
         .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,
         },
         udp_in_buffer: Default::default(),
     };
     Ok((endpoint_manager, extra_port_info))
+}
 // 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,
     unoptimized_map: 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(unoptimized_map)
+}
 // 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
     cu.ips.port_info = port_info;
     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;
+    }
     Ok(())
+}

src/runtime/tests.rs

➞

Show inline comments

@@ @@ -486,768 +486,853 @@ fn child_timeout() { @@
             let mut c = file_logged_connector(999, test_log_path);
             let g = c.new_net_port(Getter, sock_addrs[0], Passive).unwrap();
             c.connect(SEC1).unwrap();
             c.get(g).unwrap(); // not matched by put
             c.sync(None).unwrap_err(); // no timeout
         });
     })
     .unwrap();
+}
 #[test]
 fn chain_connect() {
     let test_log_path = Path::new("./logs/chain_connect");
     let sock_addrs = [next_test_addr(), next_test_addr(), next_test_addr(), next_test_addr()];
     scope(|s| {
         s.spawn(|_| {
             let mut c = file_logged_connector(0, test_log_path);
             c.new_net_port(Putter, sock_addrs[0], Passive).unwrap();
             c.connect(SEC5).unwrap();
         });
         s.spawn(|_| {
             let mut c = file_logged_connector(10, test_log_path);
             c.new_net_port(Getter, sock_addrs[0], Active).unwrap();
             c.new_net_port(Putter, sock_addrs[1], Passive).unwrap();
             c.connect(SEC5).unwrap();
         });
         s.spawn(|_| {
             // LEADER
             let mut c = file_logged_connector(7, test_log_path);
             c.new_net_port(Getter, sock_addrs[1], Active).unwrap();
             c.new_net_port(Putter, sock_addrs[2], Passive).unwrap();
             c.connect(SEC5).unwrap();
         });
         s.spawn(|_| {
             let mut c = file_logged_connector(4, test_log_path);
             c.new_net_port(Getter, sock_addrs[2], Active).unwrap();
             c.new_net_port(Putter, sock_addrs[3], Passive).unwrap();
             c.connect(SEC5).unwrap();
         });
         s.spawn(|_| {
             let mut c = file_logged_connector(1, test_log_path);
             c.new_net_port(Getter, sock_addrs[3], Active).unwrap();
             c.connect(SEC5).unwrap();
         });
     })
     .unwrap();
+}
 #[test]
 fn net_self_loop() {
     let test_log_path = Path::new("./logs/net_self_loop");
     let sock_addrs = [next_test_addr()];
     let mut c = file_logged_connector(0, test_log_path);
     let p = c.new_net_port(Putter, sock_addrs[0], Active).unwrap();
     let g = c.new_net_port(Getter, sock_addrs[0], Passive).unwrap();
     c.connect(SEC1).unwrap();
     c.put(p, TEST_MSG.clone()).unwrap();
     c.get(g).unwrap();
     c.sync(MS300).unwrap();
+}
 #[test]
 fn nobody_connects_active() {
     let test_log_path = Path::new("./logs/nobody_connects_active");
     let sock_addrs = [next_test_addr()];
     let mut c = file_logged_connector(0, test_log_path);
     let _g = c.new_net_port(Getter, sock_addrs[0], Active).unwrap();
     c.connect(Some(Duration::from_secs(5))).unwrap_err();
+}
 #[test]
 fn nobody_connects_passive() {
     let test_log_path = Path::new("./logs/nobody_connects_passive");
     let sock_addrs = [next_test_addr()];
     let mut c = file_logged_connector(0, test_log_path);
     let _g = c.new_net_port(Getter, sock_addrs[0], Passive).unwrap();
     c.connect(Some(Duration::from_secs(5))).unwrap_err();
+}
 #[test]
 fn together() {
     let test_log_path = Path::new("./logs/together");
     let sock_addrs = [next_test_addr(), next_test_addr()];
     scope(|s| {
         s.spawn(|_| {
             let mut c = file_logged_connector(0, test_log_path);
             let [p0, p1] = c.new_port_pair();
             let p2 = c.new_net_port(Getter, sock_addrs[0], Passive).unwrap();
             let p3 = c.new_net_port(Putter, sock_addrs[1], Active).unwrap();
             let [p4, p5] = c.new_port_pair();
             c.add_component(b"together", &[p1, p2, p3, p4]).unwrap();
             c.connect(SEC1).unwrap();
             c.put(p0, TEST_MSG.clone()).unwrap();
             c.get(p5).unwrap();
             c.sync(MS300).unwrap();
             c.gotten(p5).unwrap();
         });
         s.spawn(|_| {
             let mut c = file_logged_connector(1, test_log_path);
             let [p0, p1] = c.new_port_pair();
             let p2 = c.new_net_port(Getter, sock_addrs[1], Passive).unwrap();
             let p3 = c.new_net_port(Putter, sock_addrs[0], Active).unwrap();
             let [p4, p5] = c.new_port_pair();
             c.add_component(b"together", &[p1, p2, p3, p4]).unwrap();
             c.connect(SEC1).unwrap();
             c.put(p0, TEST_MSG.clone()).unwrap();
             c.get(p5).unwrap();
             c.sync(MS300).unwrap();
             c.gotten(p5).unwrap();
         });
     })
     .unwrap();
+}
 #[test]
 fn native_batch_distinguish() {
     let test_log_path = Path::new("./logs/native_batch_distinguish");
     let mut c = file_logged_connector(0, test_log_path);
     c.connect(SEC1).unwrap();
     c.next_batch().unwrap();
     c.sync(SEC1).unwrap();
+}
 #[test]
 fn multirounds() {
     let test_log_path = Path::new("./logs/multirounds");
     let sock_addrs = [next_test_addr(), next_test_addr()];
     scope(|s| {
         s.spawn(|_| {
             let mut c = file_logged_connector(0, test_log_path);
             let p0 = c.new_net_port(Putter, sock_addrs[0], Active).unwrap();
             let p1 = c.new_net_port(Getter, sock_addrs[1], Passive).unwrap();
             c.connect(SEC1).unwrap();
             for _ in 0..10 {
                 c.put(p0, TEST_MSG.clone()).unwrap();
                 c.get(p1).unwrap();
                 c.sync(SEC1).unwrap();
+            }
         });
         s.spawn(|_| {
             let mut c = file_logged_connector(1, test_log_path);
             let p0 = c.new_net_port(Getter, sock_addrs[0], Passive).unwrap();
             let p1 = c.new_net_port(Putter, sock_addrs[1], Active).unwrap();
             c.connect(SEC1).unwrap();
             for _ in 0..10 {
                 c.get(p0).unwrap();
                 c.put(p1, TEST_MSG.clone()).unwrap();
                 c.sync(SEC1).unwrap();
+            }
         });
     })
     .unwrap();
+}
 #[test]
 fn multi_recover() {
     let test_log_path = Path::new("./logs/multi_recover");
     let sock_addrs = [next_test_addr(), next_test_addr()];
     let success_iter = [true, false].iter().copied().cycle().take(10);
     scope(|s| {
         s.spawn(|_| {
             let mut c = file_logged_connector(0, test_log_path);
             let p0 = c.new_net_port(Putter, sock_addrs[0], Active).unwrap();
             let p1 = c.new_net_port(Getter, sock_addrs[1], Passive).unwrap();
             c.connect(SEC1).unwrap();
             for succeeds in success_iter.clone() {
                 c.put(p0, TEST_MSG.clone()).unwrap();
                 if succeeds {
                     c.get(p1).unwrap();
+                }
                 let res = c.sync(MS300);
                 assert_eq!(res.is_ok(), succeeds);
+            }
         });
         s.spawn(|_| {
             let mut c = file_logged_connector(1, test_log_path);
             let p0 = c.new_net_port(Getter, sock_addrs[0], Passive).unwrap();
             let p1 = c.new_net_port(Putter, sock_addrs[1], Active).unwrap();
             c.connect(SEC1).unwrap();
             for succeeds in success_iter.clone() {
                 c.get(p0).unwrap();
                 c.put(p1, TEST_MSG.clone()).unwrap();
                 let res = c.sync(MS300);
                 assert_eq!(res.is_ok(), succeeds);
+            }
         });
     })
     .unwrap();
+}
 #[test]
 fn udp_self_connect() {
     let test_log_path = Path::new("./logs/udp_self_connect");
     let sock_addrs = [next_test_addr(), next_test_addr()];
     let mut c = file_logged_connector(0, test_log_path);
     c.new_udp_mediator_component(sock_addrs[0], sock_addrs[1]).unwrap();
     c.new_udp_mediator_component(sock_addrs[1], sock_addrs[0]).unwrap();
     c.connect(SEC1).unwrap();
+}
 #[test]
 fn solo_udp_put_success() {
     let test_log_path = Path::new("./logs/solo_udp_put_success");
     let sock_addrs = [next_test_addr(), next_test_addr()];
     let mut c = file_logged_connector(0, test_log_path);
     let [p0, _] = c.new_udp_mediator_component(sock_addrs[0], sock_addrs[1]).unwrap();
     c.connect(SEC1).unwrap();
     c.put(p0, TEST_MSG.clone()).unwrap();
     c.sync(MS300).unwrap();
+}
 #[test]
 fn solo_udp_get_fail() {
     let test_log_path = Path::new("./logs/solo_udp_get_fail");
     let sock_addrs = [next_test_addr(), next_test_addr()];
     let mut c = file_logged_connector(0, test_log_path);
     let [_, p0] = c.new_udp_mediator_component(sock_addrs[0], sock_addrs[1]).unwrap();
     c.connect(SEC1).unwrap();
     c.get(p0).unwrap();
     c.sync(MS300).unwrap_err();
+}
 #[test]
 fn reowolf_to_udp() {
     let test_log_path = Path::new("./logs/reowolf_to_udp");
     let sock_addrs = [next_test_addr(), next_test_addr()];
     let barrier = std::sync::Barrier::new(2);
     scope(|s| {
         s.spawn(|_| {
             barrier.wait();
             // reowolf thread
             let mut c = file_logged_connector(0, test_log_path);
             let [p0, _] = c.new_udp_mediator_component(sock_addrs[0], sock_addrs[1]).unwrap();
             c.connect(SEC1).unwrap();
             c.put(p0, TEST_MSG.clone()).unwrap();
             c.sync(MS300).unwrap();
             barrier.wait();
         });
         s.spawn(|_| {
             barrier.wait();
             // udp thread
             let udp = std::net::UdpSocket::bind(sock_addrs[1]).unwrap();
             udp.connect(sock_addrs[0]).unwrap();
             let mut buf = new_u8_buffer(256);
             let len = udp.recv(&mut buf).unwrap();
             assert_eq!(TEST_MSG_BYTES, &buf[0..len]);
             barrier.wait();
         });
     })
     .unwrap();
+}
 #[test]
 fn udp_to_reowolf() {
     let test_log_path = Path::new("./logs/udp_to_reowolf");
     let sock_addrs = [next_test_addr(), next_test_addr()];
     let barrier = std::sync::Barrier::new(2);
     scope(|s| {
         s.spawn(|_| {
             barrier.wait();
             // reowolf thread
             let mut c = file_logged_connector(0, test_log_path);
             let [_, p0] = c.new_udp_mediator_component(sock_addrs[0], sock_addrs[1]).unwrap();
             c.connect(SEC1).unwrap();
             c.get(p0).unwrap();
             c.sync(SEC5).unwrap();
             assert_eq!(c.gotten(p0).unwrap().as_slice(), TEST_MSG_BYTES);
             barrier.wait();
         });
         s.spawn(|_| {
             barrier.wait();
             // udp thread
             let udp = std::net::UdpSocket::bind(sock_addrs[1]).unwrap();
             udp.connect(sock_addrs[0]).unwrap();
             for _ in 0..15 {
                 udp.send(TEST_MSG_BYTES).unwrap();
                 std::thread::sleep(MS100.unwrap());
+            }
             barrier.wait();
         });
     })
     .unwrap();
+}
 #[test]
 fn udp_reowolf_swap() {
     let test_log_path = Path::new("./logs/udp_reowolf_swap");
     let sock_addrs = [next_test_addr(), next_test_addr()];
     let barrier = std::sync::Barrier::new(2);
     scope(|s| {
         s.spawn(|_| {
             barrier.wait();
             // reowolf thread
             let mut c = file_logged_connector(0, test_log_path);
             let [p0, p1] = c.new_udp_mediator_component(sock_addrs[0], sock_addrs[1]).unwrap();
             c.connect(SEC1).unwrap();
             c.put(p0, TEST_MSG.clone()).unwrap();
             c.get(p1).unwrap();
             c.sync(SEC5).unwrap();
             assert_eq!(c.gotten(p1).unwrap().as_slice(), TEST_MSG_BYTES);
             barrier.wait();
         });
         s.spawn(|_| {
             barrier.wait();
             // udp thread
             let udp = std::net::UdpSocket::bind(sock_addrs[1]).unwrap();
             udp.connect(sock_addrs[0]).unwrap();
             let mut buf = new_u8_buffer(256);
             for _ in 0..5 {
                 std::thread::sleep(Duration::from_millis(60));
                 udp.send(TEST_MSG_BYTES).unwrap();
+            }
             let len = udp.recv(&mut buf).unwrap();
             assert_eq!(TEST_MSG_BYTES, &buf[0..len]);
             barrier.wait();
         });
     })
     .unwrap();
+}
 #[test]
 fn example_pres_3() {
     let test_log_path = Path::new("./logs/example_pres_3");
     let sock_addrs = [next_test_addr(), next_test_addr()];
     scope(|s| {
         s.spawn(|_| {
             // "amy"
             let mut c = file_logged_connector(0, test_log_path);
             let p0 = c.new_net_port(Putter, sock_addrs[0], Active).unwrap();
             let p1 = c.new_net_port(Putter, sock_addrs[1], Active).unwrap();
             c.connect(SEC1).unwrap();
             // put {A} and FAIL
             c.put(p0, TEST_MSG.clone()).unwrap();
             c.sync(SEC1).unwrap_err();
             // put {B} and FAIL
             c.put(p1, TEST_MSG.clone()).unwrap();
             c.sync(SEC1).unwrap_err();
             // put {A, B} and SUCCEED
             c.put(p0, TEST_MSG.clone()).unwrap();
             c.put(p1, TEST_MSG.clone()).unwrap();
             c.sync(SEC1).unwrap();
         });
         s.spawn(|_| {
             // "bob"
             let mut c = file_logged_connector(1, test_log_path);
             let p0 = c.new_net_port(Getter, sock_addrs[0], Passive).unwrap();
             let p1 = c.new_net_port(Getter, sock_addrs[1], Passive).unwrap();
             c.connect(SEC1).unwrap();
             for _ in 0..2 {
                 // get {A, B} and FAIL
                 c.get(p0).unwrap();
                 c.get(p1).unwrap();
                 c.sync(SEC1).unwrap_err();
+            }
             // get {A, B} and SUCCEED
             c.get(p0).unwrap();
             c.get(p1).unwrap();
             c.sync(SEC1).unwrap();
         });
     })
     .unwrap();
+}
 #[test]
 fn ac_not_b() {
     let test_log_path = Path::new("./logs/ac_not_b");
     let sock_addrs = [next_test_addr(), next_test_addr()];
     scope(|s| {
         s.spawn(|_| {
             // "amy"
             let mut c = file_logged_connector(0, test_log_path);
             let p0 = c.new_net_port(Putter, sock_addrs[0], Active).unwrap();
             let p1 = c.new_net_port(Putter, sock_addrs[1], Active).unwrap();
             c.connect(SEC1).unwrap();
             // put both A and B
             c.put(p0, TEST_MSG.clone()).unwrap();
             c.put(p1, TEST_MSG.clone()).unwrap();
             c.sync(SEC1).unwrap_err();
         });
         s.spawn(|_| {
             // "bob"
             let pdl = b"
             primitive ac_not_b(in a, in b, out c){
                 // forward A to C but keep B silent
                 synchronous{ put(c, get(a)); }
             }";
             let pd = Arc::new(reowolf::ProtocolDescription::parse(pdl).unwrap());
             let mut c = file_logged_configured_connector(1, test_log_path, pd);
             let p0 = c.new_net_port(Getter, sock_addrs[0], Passive).unwrap();
             let p1 = c.new_net_port(Getter, sock_addrs[1], Passive).unwrap();
             let [a, b] = c.new_port_pair();
             c.add_component(b"ac_not_b", &[p0, p1, a]).unwrap();
             c.connect(SEC1).unwrap();
             c.get(b).unwrap();
             c.sync(SEC1).unwrap_err();
         });
     })
     .unwrap();
+}
 #[test]
 fn many_rounds_net() {
     let test_log_path = Path::new("./logs/many_rounds_net");
     let sock_addrs = [next_test_addr()];
     const NUM_ROUNDS: usize = 1_000;
     scope(|s| {
         s.spawn(|_| {
             let mut c = file_logged_connector(0, test_log_path);
             let p0 = c.new_net_port(Putter, sock_addrs[0], Active).unwrap();
             c.connect(SEC1).unwrap();
             for _ in 0..NUM_ROUNDS {
                 c.put(p0, TEST_MSG.clone()).unwrap();
                 c.sync(SEC1).unwrap();
+            }
         });
         s.spawn(|_| {
             let mut c = file_logged_connector(1, test_log_path);
             let p0 = c.new_net_port(Getter, sock_addrs[0], Passive).unwrap();
             c.connect(SEC1).unwrap();
             for _ in 0..NUM_ROUNDS {
                 c.get(p0).unwrap();
                 c.sync(SEC1).unwrap();
+            }
         });
     })
     .unwrap();
+}
 #[test]
 fn many_rounds_mem() {
     let test_log_path = Path::new("./logs/many_rounds_mem");
     const NUM_ROUNDS: usize = 1_000;
     let mut c = file_logged_connector(0, test_log_path);
     let [p0, p1] = c.new_port_pair();
     c.connect(SEC1).unwrap();
     for _ in 0..NUM_ROUNDS {
         c.put(p0, TEST_MSG.clone()).unwrap();
         c.get(p1).unwrap();
         c.sync(SEC1).unwrap();
+    }
+}
 #[test]
 fn pdl_reo_lossy() {
     let pdl = b"
     primitive lossy(in a, out b) {
         while(true) synchronous {
             msg m = null;
             if(fires(a)) {
                 m = get(a);
                 if(fires(b)) {
                     put(b, m);
+                }
+            }
+        }
+    }
     ";
     reowolf::ProtocolDescription::parse(pdl).unwrap();
+}
 #[test]
 fn pdl_reo_fifo1() {
     let pdl = b"
     primitive fifo1(in a, out b) {
         msg m = null;
         while(true) synchronous {
             if(m == null) {
                 if(fires(a)) m=get(a);
             } else {
                 if(fires(b)) put(b, m);
                 m = null;
+            }
+        }
+    }
     ";
     reowolf::ProtocolDescription::parse(pdl).unwrap();
+}
 #[test]
 fn pdl_reo_fifo1full() {
     let test_log_path = Path::new("./logs/pdl_reo_fifo1full");
     let pdl = b"
     primitive fifo1full(in a, out b) {
         msg m = create(0);
         while(true) synchronous {
             if(m == null) {
                 if(fires(a)) m=get(a);
             } else {
                 if(fires(b)) put(b, m);
                 m = null;
+            }
+        }
+    }
     ";
     let pd = reowolf::ProtocolDescription::parse(pdl).unwrap();
     let mut c = file_logged_configured_connector(0, test_log_path, Arc::new(pd));
     let [_p0, g0] = c.new_port_pair();
     let [p1, g1] = c.new_port_pair();
     c.add_component(b"fifo1full", &[g0, p1]).unwrap();
     c.connect(None).unwrap();
     c.get(g1).unwrap();
     c.sync(None).unwrap();
     assert_eq!(0, c.gotten(g1).unwrap().len());
+}
 #[test]
 fn pdl_msg_consensus() {
     let test_log_path = Path::new("./logs/pdl_msg_consensus");
     let pdl = b"
     primitive msgconsensus(in a, in b) {
         while(true) synchronous {
             msg x = get(a);
             msg y = get(b);
             assert(x == y);
+        }
+    }
     ";
     let pd = reowolf::ProtocolDescription::parse(pdl).unwrap();
     let mut c = file_logged_configured_connector(0, test_log_path, Arc::new(pd));
     let [p0, g0] = c.new_port_pair();
     let [p1, g1] = c.new_port_pair();
     c.add_component(b"msgconsensus", &[g0, g1]).unwrap();
     c.connect(None).unwrap();
     c.put(p0, Payload::from(b"HELLO" as &[_])).unwrap();
     c.put(p1, Payload::from(b"HELLO" as &[_])).unwrap();
     c.sync(SEC1).unwrap();
     c.put(p0, Payload::from(b"HEY" as &[_])).unwrap();
     c.put(p1, Payload::from(b"HELLO" as &[_])).unwrap();
     c.sync(SEC1).unwrap_err();
+}
 #[test]
 fn sequencer3_prim() {
     let test_log_path = Path::new("./logs/sequencer3_prim");
     let pdl = b"
     primitive sequencer3(out a, out b, out c) {
         int i = 0;
         while(true) synchronous {
             out to = a;
             if     (i==1) to = b;
             else if(i==2) to = c;
             if(fires(to)) {
                 put(to, create(0));
                 i = (i + 1)%3;
+            }
+        }
+    }
     ";
     let pd = reowolf::ProtocolDescription::parse(pdl).unwrap();
     let mut c = file_logged_configured_connector(0, test_log_path, Arc::new(pd));
     // setup a session between (a) native, and (b) sequencer3, connected by 3 ports.
     let [p0, g0] = c.new_port_pair();
     let [p1, g1] = c.new_port_pair();
     let [p2, g2] = c.new_port_pair();
     c.add_component(b"sequencer3", &[p0, p1, p2]).unwrap();
     c.connect(None).unwrap();
     let which_of_three = move |c: &mut Connector| {
         // setup three sync batches. sync. return which succeeded
         c.get(g0).unwrap();
         c.next_batch().unwrap();
         c.get(g1).unwrap();
         c.next_batch().unwrap();
         c.get(g2).unwrap();
         c.sync(None).unwrap()
     };
     const TEST_ROUNDS: usize = 50;
     // check that the batch index for rounds 0..TEST_ROUNDS are [0, 1, 2, 0, 1, 2, ...]
     for expected_batch_idx in (0..=2).cycle().take(TEST_ROUNDS) {
         // silent round
         assert_eq!(0, c.sync(None).unwrap());
         // non silent round
         assert_eq!(expected_batch_idx, which_of_three(&mut c));
+    }
+}
 #[test]
 fn sequencer3_comp() {
     let test_log_path = Path::new("./logs/sequencer3_comp");
     let pdl = b"
     primitive fifo1_init(msg m, in a, out b) {
         while(true) synchronous {
             if(m != null && fires(b)) {
                 put(b, m);
                 m = null;
             } else if (m == null && fires(a)) {
                 m = get(a);
+            }
+        }
+    }
     composite fifo1_full(in a, out b) {
         new fifo1_init(create(0), a, b);
+    }
     composite fifo1(in a, out b) {
         new fifo1_init(null, a, b);
+    }
     composite sequencer3(out a, out b, out c) {
         channel d -> e;
         channel f -> g;
         channel h -> i;
         channel j -> k;
         channel l -> m;
         channel n -> o;
         new fifo1_full(o, d);
         new replicator(e, f, a);
         new fifo1(g, h);
         new replicator(i, j, b);
         new fifo1(k, l);
         new replicator(m, n, c);
+    }
     ";
     let pd = reowolf::ProtocolDescription::parse(pdl).unwrap();
     let mut c = file_logged_configured_connector(0, test_log_path, Arc::new(pd));
     // setup a session between (a) native, and (b) sequencer3, connected by 3 ports.
     let [p0, g0] = c.new_port_pair();
     let [p1, g1] = c.new_port_pair();
     let [p2, g2] = c.new_port_pair();
     c.add_component(b"sequencer3", &[p0, p1, p2]).unwrap();
     c.connect(None).unwrap();
     let which_of_three = move |c: &mut Connector| {
         // setup three sync batches. sync. return which succeeded
         c.get(g0).unwrap();
         c.next_batch().unwrap();
         c.get(g1).unwrap();
         c.next_batch().unwrap();
         c.get(g2).unwrap();
         c.sync(SEC1).unwrap()
     };
     const TEST_ROUNDS: usize = 50;
     // check that the batch index for rounds 0..TEST_ROUNDS are [0, 1, 2, 0, 1, 2, ...]
     for expected_batch_idx in (0..=2).cycle().take(TEST_ROUNDS) {
         // silent round
         assert_eq!(0, c.sync(SEC1).unwrap());
         // non silent round
         assert_eq!(expected_batch_idx, which_of_three(&mut c));
+    }
+}
 enum XRouterItem {
     Silent,
     GetA,
     GetB,
+}
 // Hardcoded pseudo-random sequence of round behaviors for the native component
 const XROUTER_ITEMS: &[XRouterItem] = {
     use XRouterItem::{GetA as A, GetB as B, Silent as S};
     &[
         B, A, S, B, A, A, B, S, B, S, A, A, S, B, B, S, B, S, B, B, S, B, B, A, B, B, A, B, A, B,
         S, B, S, B, S, A, S, B, A, S, B, A, B, S, B, S, B, S, S, B, B, A, A, A, S, S, S, B, A, A,
         A, S, S, B, B, B, A, B, S, S, A, A, B, A, B, B, A, A, A, B, A, B, S, A, B, S, A, A, B, S,
+    ]
 };
 #[test]
 fn xrouter_prim() {
     let test_log_path = Path::new("./logs/xrouter_prim");
     let pdl = b"
     primitive xrouter(in a, out b, out c) {
         while(true) synchronous {
             if(fires(a)) {
                 if(fires(b)) put(b, get(a));
                 else         put(c, get(a));
+            }
+        }
+    }
     ";
     let pd = reowolf::ProtocolDescription::parse(pdl).unwrap();
     let mut c = file_logged_configured_connector(0, test_log_path, Arc::new(pd));
     // setup a session between (a) native, and (b) xrouter2, connected by 3 ports.
     let [p0, g0] = c.new_port_pair();
     let [p1, g1] = c.new_port_pair();
     let [p2, g2] = c.new_port_pair();
     c.add_component(b"xrouter", &[g0, p1, p2]).unwrap();
     c.connect(None).unwrap();
     let now = std::time::Instant::now();
     for item in XROUTER_ITEMS.iter() {
         match item {
             XRouterItem::Silent => {}
             XRouterItem::GetA => {
                 c.put(p0, TEST_MSG.clone()).unwrap();
                 c.get(g1).unwrap();
+            }
             XRouterItem::GetB => {
                 c.put(p0, TEST_MSG.clone()).unwrap();
                 c.get(g2).unwrap();
+            }
+        }
         assert_eq!(0, c.sync(SEC1).unwrap());
+    }
     println!("PRIM {:?}", now.elapsed());
+}
 #[test]
 fn xrouter_comp() {
     let test_log_path = Path::new("./logs/xrouter_comp");
     let pdl = b"
     primitive lossy(in a, out b) {
         while(true) synchronous {
             if(fires(a)) {
                 msg m = get(a);
                 if(fires(b)) put(b, m);
+            }
+        }
+    }
     primitive sync_drain(in a, in b) {
         while(true) synchronous {
             if(fires(a)) {
                 get(a);
                 get(b);
+            }
+        }
+    }
     composite xrouter(in a, out b, out c) {
         channel d -> e;
         channel f -> g;
         channel h -> i;
         channel j -> k;
         channel l -> m;
         channel n -> o;
         channel p -> q;
         channel r -> s;
         channel t -> u;
         new replicator(a, d, f);
         new replicator(g, t, h);
         new lossy(e, l);
         new lossy(i, j);
         new replicator(m, b, p);
         new replicator(k, n, c);
         new merger(q, o, r);
         new sync_drain(u, s);
+    }
     ";
     let pd = reowolf::ProtocolDescription::parse(pdl).unwrap();
     let mut c = file_logged_configured_connector(0, test_log_path, Arc::new(pd));
     // setup a session between (a) native, and (b) xrouter2, connected by 3 ports.
     let [p0, g0] = c.new_port_pair();
     let [p1, g1] = c.new_port_pair();
     let [p2, g2] = c.new_port_pair();
     c.add_component(b"xrouter", &[g0, p1, p2]).unwrap();
     c.connect(None).unwrap();
     let now = std::time::Instant::now();
     for item in XROUTER_ITEMS.iter() {
         match item {
             XRouterItem::Silent => {}
             XRouterItem::GetA => {
                 c.put(p0, TEST_MSG.clone()).unwrap();
                 c.get(g1).unwrap();
+            }
             XRouterItem::GetB => {
                 c.put(p0, TEST_MSG.clone()).unwrap();
                 c.get(g2).unwrap();
+            }
+        }
         assert_eq!(0, c.sync(SEC1).unwrap());
+    }
     println!("COMP {:?}", now.elapsed());
+}
 #[test]
 fn count_stream() {
     let test_log_path = Path::new("./logs/count_stream");
     let pdl = b"
     primitive count_stream(out o) {
         msg m = create(1);
         m[0] = 0;
         while(true) synchronous {
             put(o, m);
             m[0] += 1;
+        }
+    }
     ";
     let pd = reowolf::ProtocolDescription::parse(pdl).unwrap();
     let mut c = file_logged_configured_connector(0, test_log_path, Arc::new(pd));
     // setup a session between (a) native, and (b) sequencer3, connected by 3 ports.
     let [p0, g0] = c.new_port_pair();
     c.add_component(b"count_stream", &[p0]).unwrap();
     c.connect(None).unwrap();
     for expecting in 0u8..16 {
         c.get(g0).unwrap();
         c.sync(None).unwrap();
         assert_eq!(&[expecting], c.gotten(g0).unwrap().as_slice());
+    }
+}
 #[test]
 fn for_msg_byte() {
     let test_log_path = Path::new("./logs/for_msg_byte");
     let pdl = b"
     primitive for_msg_byte(out o) {
         byte i = 0;
         while(i<8) {
             msg m = create(1);
             m[0] = i;
             synchronous() put(o, m);
             i++;
+        }
+    }
     ";
     let pd = reowolf::ProtocolDescription::parse(pdl).unwrap();
     let mut c = file_logged_configured_connector(0, test_log_path, Arc::new(pd));
     // setup a session between (a) native, and (b) sequencer3, connected by 3 ports.
     let [p0, g0] = c.new_port_pair();
     c.add_component(b"for_msg_byte", &[p0]).unwrap();
     c.connect(None).unwrap();
     for expecting in 0u8..8 {
         c.get(g0).unwrap();
         c.sync(None).unwrap();
         assert_eq!(&[expecting], c.gotten(g0).unwrap().as_slice());
+    }
     c.sync(None).unwrap();
+}
 #[test]
 fn message_concat() {
     // Note: PDL quirks:
     // 1. declarations as first lines of a scope
     // 2. var names cannot be prefixed by types. Eg `msg_concat` prohibited.
     let test_log_path = Path::new("./logs/message_concat");
     let pdl = b"
     primitive message_concat(out o) {
         msg a = create(1);
         msg b = create(1);
         a[0] = 0;
         b[0] = 1;
         synchronous() put(o, a+b);
+    }
     ";
     let pd = reowolf::ProtocolDescription::parse(pdl).unwrap();
     let mut c = file_logged_configured_connector(0, test_log_path, Arc::new(pd));
     // setup a session between (a) native, and (b) sequencer3, connected by 3 ports.
     let [p0, g0] = c.new_port_pair();
     c.add_component(b"message_concat", &[p0]).unwrap();
     c.connect(None).unwrap();
     c.get(g0).unwrap();
     c.sync(None).unwrap();
     assert_eq!(&[0, 1], c.gotten(g0).unwrap().as_slice());
+}

0 comments (0 inline, 0 general)