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();
@@ @@ -176,192 +176,203 @@ impl Value { @@
             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))
+            }

src/protocol/mod.rs

➞

Show inline comments

@@ @@ -12,193 +12,193 @@ lazy_static::lazy_static! { @@
     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)) => {

src/runtime/mod.rs

➞

Show inline comments

@@ @@ -564,194 +564,193 @@ impl Connector { @@
         // 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);
+                }

src/runtime/setup.rs

➞

Show inline comments

@@ @@ -177,311 +177,295 @@ impl Connector { @@
                 });
                 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;
+                        }

src/runtime/tests.rs

➞

Show inline comments

@@ @@ -1158,96 +1158,181 @@ fn xrouter_prim() { @@
     ";
     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)