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

Changeset - 38c129959044

Parent rev.

Child rev.

[Not reviewed]

0 17 2

Max Henger - 3 years ago 2022-05-13 19:30:02
henger@cwi.nl

feat: transmitting ports

12 files changed:

src/protocol/eval/mod.rs

src/protocol/eval/value.rs

src/protocol/parser/mod.rs

src/runtime2/communication.rs

src/runtime2/component/component.rs

691

src/runtime2/component/component_context.rs

src/runtime2/component/component_internet.rs

src/runtime2/component/component_pdl.rs

347

src/runtime2/component/component_random.rs

src/runtime2/component/consensus.rs

src/runtime2/component/control_layer.rs

src/runtime2/component/mod.rs

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

0 comments (0 inline, 0 general)

src/protocol/eval/mod.rs

➞

Show inline comments

 /// eval
 ///
 /// Evaluator of the generated AST. Note that we use some misappropriated terms
 /// to describe where values live and what they do. This is a temporary
 /// implementation of an evaluator until some kind of appropriate bytecode or
 /// machine code is generated.
 ///
 /// Code is always executed within a "frame". For Reowolf the first frame is
 /// usually an executed component. All subsequent frames are function calls.
 /// Simple values live on the "stack". Each variable/parameter has a place on
 /// the stack where its values are stored. If the value is not a primitive, then
 /// its value will be stored in the "heap". Expressions are treated differently
 /// and use a separate "stack" for their evaluation.
 ///
 /// Since this is a value-based language, most values are copied. One has to be
 /// careful with values that reside in the "heap" and make sure that copies are
 /// properly removed from the heap..
 ///
 /// Just to reiterate: this is a temporary wasteful implementation. A proper
 /// implementation would fully fill out the type table with alignment/size/
 /// offset information and lay out bytecode.
 pub(crate) mod value;
 pub(crate) mod store;
 pub(crate) mod executor;
 pub(crate) mod error;
 pub use error::EvalError;
 pub use value::{PortId, Value, ValueGroup};
+pub use value::{PortId, Value, ValueId, ValueGroup};
 pub use executor::{EvalContinuation, EvalResult, Prompt};

src/protocol/eval/value.rs

➞

Show inline comments

 use std::collections::VecDeque;
 use super::store::*;
 use crate::protocol::ast::{
     AssignmentOperator,
     BinaryOperator,
     UnaryOperator,
     ConcreteType,
     ConcreteTypePart,
 };
 use crate::protocol::parser::token_parsing::*;
 pub type StackPos = u32;
 pub type HeapPos = u32;
 #[derive(Debug, Copy, Clone)]
 pub enum ValueId {
     Stack(StackPos), // place on stack
     Heap(HeapPos, u32), // allocated region + values within that region
+}
 #[derive(Debug, Copy, Clone, PartialEq, Eq)]
 pub struct PortId{
     pub(crate) id: u32
+}
 impl PortId {
     pub fn new(id: u32) -> Self {
         return Self{ id };
+    }
+}
 /// Represents a value stored on the stack or on the heap. Some values contain
 /// a `HeapPos`, implying that they're stored in the store's `Heap`. Clearing
 /// a `Value` with a `HeapPos` from a stack must also clear the associated
 /// region from the `Heap`.
 #[derive(Debug, Clone)]
 pub enum Value {
     // Special types, never encountered during evaluation if the compiler works correctly
     Unassigned,                 // Marker when variables are first declared, immediately followed by assignment
     PrevStackBoundary(isize),   // Marker for stack frame beginning, so we can pop stack values
     Ref(ValueId),               // Reference to a value, used by expressions producing references
     Binding(StackPos),          // Reference to a binding variable (reserved on the stack)
     // Builtin types
     Input(PortId),
     Output(PortId),
     Message(HeapPos),
     Null,
     Bool(bool),
     Char(char),
     String(HeapPos),
     UInt8(u8),
     UInt16(u16),
     UInt32(u32),
     UInt64(u64),
     SInt8(i8),
     SInt16(i16),
     SInt32(i32),
     SInt64(i64),
     Array(HeapPos),
     Tuple(HeapPos),
     // Instances of user-defined types
     Enum(i64),
     Union(i64, HeapPos),
     Struct(HeapPos),
+}
 union_cast_to_value_method_impl!(as_stack_boundary, isize, Value::PrevStackBoundary);
 union_cast_to_value_method_impl!(as_ref, ValueId, Value::Ref);
 union_cast_to_value_method_impl!(as_input, PortId, Value::Input);
 union_cast_to_value_method_impl!(as_output, PortId, Value::Output);
 union_cast_to_value_method_impl!(as_message, HeapPos, Value::Message);
 union_cast_to_value_method_impl!(as_bool, bool, Value::Bool);
 union_cast_to_value_method_impl!(as_char, char, Value::Char);
 union_cast_to_value_method_impl!(as_string, HeapPos, Value::String);
 union_cast_to_value_method_impl!(as_uint8, u8, Value::UInt8);
 union_cast_to_value_method_impl!(as_uint16, u16, Value::UInt16);
 union_cast_to_value_method_impl!(as_uint32, u32, Value::UInt32);
 union_cast_to_value_method_impl!(as_uint64, u64, Value::UInt64);
 union_cast_to_value_method_impl!(as_sint8, i8, Value::SInt8);
 union_cast_to_value_method_impl!(as_sint16, i16, Value::SInt16);
 union_cast_to_value_method_impl!(as_sint32, i32, Value::SInt32);
 union_cast_to_value_method_impl!(as_sint64, i64, Value::SInt64);
 union_cast_to_value_method_impl!(as_array, HeapPos, Value::Array);
 union_cast_to_value_method_impl!(as_tuple, HeapPos, Value::Tuple);
 union_cast_to_value_method_impl!(as_enum, i64, Value::Enum);
 union_cast_to_value_method_impl!(as_struct, HeapPos, Value::Struct);
 impl Value {
     pub(crate) fn as_union(&self) -> (i64, HeapPos) {
         match self {
             Value::Union(tag, v) => (*tag, *v),
             _ => panic!("called as_union on {:?}", self),
+        }
+    }
     pub(crate) fn as_port_id(&self) -> PortId {
         match self {
             Value::Input(v) => *v,
             Value::Output(v) => *v,
             _ => unreachable!(),
+        }
+    }
     pub(crate) fn is_integer(&self) -> bool {
         match self {
             Value::UInt8(_) | Value::UInt16(_) | Value::UInt32(_) | Value::UInt64(_) |
             Value::SInt8(_) | Value::SInt16(_) | Value::SInt32(_) | Value::SInt64(_) => true,
             _ => false
+        }
+    }
     pub(crate) fn is_unsigned_integer(&self) -> bool {
         match self {
             Value::UInt8(_) | Value::UInt16(_) | Value::UInt32(_) | Value::UInt64(_) => true,
             _ => false
+        }
+    }
     pub(crate) fn is_signed_integer(&self) -> bool {
         match self {
             Value::SInt8(_) | Value::SInt16(_) | Value::SInt32(_) | Value::SInt64(_) => true,
             _ => false
+        }
+    }
     pub(crate) fn as_unsigned_integer(&self) -> u64 {
         match self {
             Value::UInt8(v)  => *v as u64,
             Value::UInt16(v) => *v as u64,
             Value::UInt32(v) => *v as u64,
             Value::UInt64(v) => *v as u64,
             _ => unreachable!("called as_unsigned_integer on {:?}", self),
+        }
+    }
     pub(crate) fn as_signed_integer(&self) -> i64 {
         match self {
             Value::SInt8(v)  => *v as i64,
             Value::SInt16(v) => *v as i64,
             Value::SInt32(v) => *v as i64,
             Value::SInt64(v) => *v as i64,
             _ => unreachable!("called as_signed_integer on {:?}", self)
+        }
+    }
     /// Returns the heap position associated with the value. If the value
     /// doesn't store anything in the heap then we return `None`.
     pub(crate) fn get_heap_pos(&self) -> Option<HeapPos> {
         match self {
             Value::Message(v) => Some(*v),
             Value::String(v) => Some(*v),
             Value::Array(v) => Some(*v),
             Value::Tuple(v) => Some(*v),
             Value::Union(_, v) => Some(*v),
             Value::Struct(v) => Some(*v),
             _ => None
+        }
+    }
+}
 /// When providing arguments to a new component, or when transferring values
 /// from one component's store to a newly instantiated component, one has to
 /// transfer stack and heap values. This `ValueGroup` represents such a
 /// temporary group of values with potential heap allocations.
 ///
 /// Constructing such a ValueGroup manually requires some extra care to make
 /// sure all elements of `values` point to valid elements of `regions`.
 ///
 /// Again: this is a temporary thing, hopefully removed once we move to a
 /// bytecode interpreter.
 #[derive(Clone, Debug)]
 pub struct ValueGroup {
     pub(crate) values: Vec<Value>,
     pub(crate) regions: Vec<Vec<Value>>
+}
 impl ValueGroup {
     pub(crate) fn new_stack(values: Vec<Value>) -> Self {
         debug_assert!(values.iter().all(|v| v.get_heap_pos().is_none()));
         Self{
             values,
             regions: Vec::new(),
+        }
+    }
     pub(crate) fn from_store(store: &Store, values: &[Value]) -> Self {
         let mut group = ValueGroup{
             values: Vec::with_capacity(values.len()),
             regions: Vec::with_capacity(values.len()), // estimation
         };
         for value in values {
             let transferred = group.retrieve_value(value, store);
             group.values.push(transferred);
+        }
         group
+    }
     /// Creates a clone of the value group, but leaves the memory inside of the
     /// ValueGroup vectors allocated.
     pub(crate) fn take(&mut self) -> ValueGroup {
         let cloned = self.clone();
         self.values.clear();
         self.regions.clear();
         return cloned;
+    }
     /// Transfers a provided value from a store into a local value with its
     /// heap allocations (if any) stored in the ValueGroup. Calling this
     /// function will not store the returned value in the `values` member.
     fn retrieve_value(&mut self, value: &Value, from_store: &Store) -> Value {
         let value = from_store.maybe_read_ref(value);
         if let Some(heap_pos) = value.get_heap_pos() {
             // Value points to a heap allocation, so transfer the heap values
             // internally.
             let from_region = &from_store.heap_regions[heap_pos as usize].values;
             let mut new_region = Vec::with_capacity(from_region.len());
             for value in from_region {
                 let transferred = self.retrieve_value(value, from_store);
                 new_region.push(transferred);
+            }
             // Region is constructed, store internally and return the new value.
             let new_region_idx = self.regions.len() as HeapPos;
             self.regions.push(new_region);
             return match value {
                 Value::Message(_)    => Value::Message(new_region_idx),
                 Value::String(_)     => Value::String(new_region_idx),
                 Value::Array(_)      => Value::Array(new_region_idx),
                 Value::Tuple(_)      => Value::Tuple(new_region_idx),
                 Value::Union(tag, _) => Value::Union(*tag, new_region_idx),
                 Value::Struct(_)     => Value::Struct(new_region_idx),
                 _ => unreachable!(),
             };
         } else {
             return value.clone();
+        }
+    }
     /// Transfers the heap values and the stack values into the store. Stack
     /// values are pushed onto the Store's stack in the order in which they
     /// appear in the value group.
     pub(crate) fn into_store(self, store: &mut Store) {
         for value in &self.values {
             let transferred = self.provide_value(value, store);
             store.stack.push(transferred);
+        }
+    }
     /// Transfers the heap values into the store, but will put the stack values
     /// into the provided `VecDeque`. This is mainly used to merge `ValueGroup`
     /// instances retrieved by the code by `get` calls into the expression
     /// stack.
     pub(crate) fn into_stack(self, stack: &mut VecDeque<Value>, store: &mut Store) {
         for value in &self.values {
             let transferred = self.provide_value(value, store);
             stack.push_back(transferred);
+        }
+    }
     /// Retrieves a mutable reference to the value given its ValueId.
     pub(crate) fn get_value_mut(&mut self, id: ValueId) -> &mut Value {
         match id {
             ValueId::Stack(pos) => return &mut self.values[pos as usize],
             ValueId::Heap(heap_pos, pos) => return &mut self.regions[heap_pos as usize][pos as usize],
+        }
+    }
     fn provide_value(&self, value: &Value, to_store: &mut Store) -> Value {
         if let Some(from_heap_pos) = value.get_heap_pos() {
             let from_heap_pos = from_heap_pos as usize;
             let to_heap_pos = to_store.alloc_heap();
             let to_heap_pos_usize = to_heap_pos as usize;
             to_store.heap_regions[to_heap_pos_usize].values.reserve(self.regions[from_heap_pos].len());
             for value in &self.regions[from_heap_pos as usize] {
                 let transferred = self.provide_value(value, to_store);
                 to_store.heap_regions[to_heap_pos_usize].values.push(transferred);
+            }
             return match value {
                 Value::Message(_)    => Value::Message(to_heap_pos),
                 Value::String(_)     => Value::String(to_heap_pos),
                 Value::Array(_)      => Value::Array(to_heap_pos),
                 Value::Tuple(_)      => Value::Tuple(to_heap_pos),
                 Value::Union(tag, _) => Value::Union(*tag, to_heap_pos),
                 Value::Struct(_)     => Value::Struct(to_heap_pos),
                 _ => unreachable!(),
             };
         } else {
             return value.clone();
+        }
+    }
+}
 impl Default for ValueGroup {
     /// Returns an empty ValueGroup
     fn default() -> Self {
         Self { values: Vec::new(), regions: Vec::new() }
+    }
+}
 enum ValueKind { Message, String, Array }
 pub(crate) fn apply_assignment_operator(store: &mut Store, lhs: ValueId, op: AssignmentOperator, rhs: Value) {
     use AssignmentOperator as AO;
     macro_rules! apply_int_op {
         ($lhs:ident, $assignment_tokens:tt, $operator:ident, $rhs:ident) => {
             match $lhs {
                 Value::UInt8(v)  => { *v $assignment_tokens $rhs.as_uint8();  },
                 Value::UInt16(v) => { *v $assignment_tokens $rhs.as_uint16(); },
                 Value::UInt32(v) => { *v $assignment_tokens $rhs.as_uint32(); },
                 Value::UInt64(v) => { *v $assignment_tokens $rhs.as_uint64(); },
                 Value::SInt8(v)  => { *v $assignment_tokens $rhs.as_sint8();  },
                 Value::SInt16(v) => { *v $assignment_tokens $rhs.as_sint16(); },
                 Value::SInt32(v) => { *v $assignment_tokens $rhs.as_sint32(); },
                 Value::SInt64(v) => { *v $assignment_tokens $rhs.as_sint64(); },
                 _ => unreachable!("apply_assignment_operator {:?} on lhs {:?} and rhs {:?}", $operator, $lhs, $rhs),
+            }
+        }
+    }
     let lhs = store.read_mut_ref(lhs);
     let mut to_dealloc = None;
     match op {
         AO::Set => {
             match lhs {
                 Value::Unassigned => { *lhs = rhs; },
                 Value::Input(v)  => { *v = rhs.as_input(); },
                 Value::Output(v) => { *v = rhs.as_output(); },
                 Value::Message(v)  => { to_dealloc = Some(*v); *v = rhs.as_message(); },
                 Value::Bool(v)    => { *v = rhs.as_bool(); },
                 Value::Char(v) => { *v = rhs.as_char(); },
                 Value::String(v) => { *v = rhs.as_string().clone(); },
                 Value::UInt8(v) => { *v = rhs.as_uint8(); },
                 Value::UInt16(v) => { *v = rhs.as_uint16(); },
                 Value::UInt32(v) => { *v = rhs.as_uint32(); },
                 Value::UInt64(v) => { *v = rhs.as_uint64(); },
                 Value::SInt8(v) => { *v = rhs.as_sint8(); },
                 Value::SInt16(v) => { *v = rhs.as_sint16(); },
                 Value::SInt32(v) => { *v = rhs.as_sint32(); },
                 Value::SInt64(v) => { *v = rhs.as_sint64(); },
                 Value::Array(v) => { to_dealloc = Some(*v); *v = rhs.as_array(); },
                 Value::Tuple(v) => { to_dealloc = Some(*v); *v = rhs.as_tuple(); },
                 Value::Enum(v) => { *v = rhs.as_enum(); },
                 Value::Union(lhs_tag, lhs_heap_pos) => {
                     to_dealloc = Some(*lhs_heap_pos);
                     let (rhs_tag, rhs_heap_pos) = rhs.as_union();
                     *lhs_tag = rhs_tag;
                     *lhs_heap_pos = rhs_heap_pos;
+                }
                 Value::Struct(v) => { to_dealloc = Some(*v); *v = rhs.as_struct(); },
                 _ => unreachable!("apply_assignment_operator {:?} on lhs {:?} and rhs {:?}", op, lhs, rhs),
+            }
         },
         AO::Concatenated => {
             let lhs_heap_pos = lhs.get_heap_pos().unwrap() as usize;
             let rhs_heap_pos = rhs.get_heap_pos().unwrap() as usize;
             // To prevent borrowing crap, swap out heap region with a temp empty array
             let mut total = Vec::new();
             std::mem::swap(&mut total, &mut store.heap_regions[lhs_heap_pos].values);
             // Push everything onto the swapped vector
             let rhs_len = store.heap_regions[rhs_heap_pos].values.len();
             total.reserve(rhs_len);
             for value_idx in 0..rhs_len {
                 total.push(store.clone_value(store.heap_regions[rhs_heap_pos].values[value_idx].clone()));
+            }
             // Swap back in place
             std::mem::swap(&mut total, &mut store.heap_regions[lhs_heap_pos].values);
             // We took ownership of the RHS, but we copied it into the LHS, so
             // different form assignment we need to drop the RHS heap pos.
             to_dealloc = Some(rhs_heap_pos as u32);
         },
         AO::Multiplied =>   { apply_int_op!(lhs, *=,  op, rhs) },
         AO::Divided =>      { apply_int_op!(lhs, /=,  op, rhs) },
         AO::Remained =>     { apply_int_op!(lhs, %=,  op, rhs) },
         AO::Added =>        { apply_int_op!(lhs, +=,  op, rhs) },
         AO::Subtracted =>   { apply_int_op!(lhs, -=,  op, rhs) },
         AO::ShiftedLeft =>  { apply_int_op!(lhs, <<=, op, rhs) },
         AO::ShiftedRight => { apply_int_op!(lhs, >>=, op, rhs) },
         AO::BitwiseAnded => { apply_int_op!(lhs, &=,  op, rhs) },
         AO::BitwiseXored => { apply_int_op!(lhs, ^=,  op, rhs) },
         AO::BitwiseOred =>  { apply_int_op!(lhs, |=,  op, rhs) },
+    }
     if let Some(heap_pos) = to_dealloc {
         store.drop_heap_pos(heap_pos);
+    }
+}
 pub(crate) fn apply_binary_operator(store: &mut Store, lhs: &Value, op: BinaryOperator, rhs: &Value) -> Value {
     use BinaryOperator as BO;
     macro_rules! apply_int_op_and_return_self {
         ($lhs:ident, $operator_tokens:tt, $operator:ident, $rhs:ident) => {
             return match $lhs {
                 Value::UInt8(v)  => { Value::UInt8( *v $operator_tokens $rhs.as_uint8() ) },
                 Value::UInt16(v) => { Value::UInt16(*v $operator_tokens $rhs.as_uint16()) },
                 Value::UInt32(v) => { Value::UInt32(*v $operator_tokens $rhs.as_uint32()) },
                 Value::UInt64(v) => { Value::UInt64(*v $operator_tokens $rhs.as_uint64()) },
                 Value::SInt8(v)  => { Value::SInt8( *v $operator_tokens $rhs.as_sint8() ) },
                 Value::SInt16(v) => { Value::SInt16(*v $operator_tokens $rhs.as_sint16()) },
                 Value::SInt32(v) => { Value::SInt32(*v $operator_tokens $rhs.as_sint32()) },
                 Value::SInt64(v) => { Value::SInt64(*v $operator_tokens $rhs.as_sint64()) },
                 _ => unreachable!("apply_binary_operator {:?} on lhs {:?} and rhs {:?}", $operator, $lhs, $rhs)
             };
+        }
+    }
     macro_rules! apply_int_op_and_return_bool {
         ($lhs:ident, $operator_tokens:tt, $operator:ident, $rhs:ident) => {
             return match $lhs {
                 Value::UInt8(v)  => { Value::Bool(*v $operator_tokens $rhs.as_uint8() ) },
                 Value::UInt16(v) => { Value::Bool(*v $operator_tokens $rhs.as_uint16()) },
                 Value::UInt32(v) => { Value::Bool(*v $operator_tokens $rhs.as_uint32()) },
                 Value::UInt64(v) => { Value::Bool(*v $operator_tokens $rhs.as_uint64()) },
                 Value::SInt8(v)  => { Value::Bool(*v $operator_tokens $rhs.as_sint8() ) },
                 Value::SInt16(v) => { Value::Bool(*v $operator_tokens $rhs.as_sint16()) },
                 Value::SInt32(v) => { Value::Bool(*v $operator_tokens $rhs.as_sint32()) },
                 Value::SInt64(v) => { Value::Bool(*v $operator_tokens $rhs.as_sint64()) },
                 _ => unreachable!("apply_binary_operator {:?} on lhs {:?} and rhs {:?}", $operator, $lhs, $rhs)
             };
+        }
+    }
     // We need to handle concatenate in a special way because it needs the store
     // mutably.
     if op == BO::Concatenate {
         let target_heap_pos = store.alloc_heap();
         let lhs_heap_pos;
         let rhs_heap_pos;
         let lhs = store.maybe_read_ref(lhs);
         let rhs = store.maybe_read_ref(rhs);
         let value_kind;
         match lhs {
             Value::Message(lhs_pos) => {
                 lhs_heap_pos = *lhs_pos;
                 rhs_heap_pos = rhs.as_message();
                 value_kind = ValueKind::Message;
             },
             Value::String(lhs_pos) => {
                 lhs_heap_pos = *lhs_pos;
                 rhs_heap_pos = rhs.as_string();
                 value_kind = ValueKind::String;
             },
             Value::Array(lhs_pos) => {
                 lhs_heap_pos = *lhs_pos;
                 rhs_heap_pos = rhs.as_array();
                 value_kind = ValueKind::Array;
             },
             _ => unreachable!("apply_binary_operator {:?} on lhs {:?} and rhs {:?}", op, lhs, rhs)
+        }
         let lhs_heap_pos = lhs_heap_pos as usize;
         let rhs_heap_pos = rhs_heap_pos as usize;
         let mut concatenated = Vec::new();
         let lhs_len = store.heap_regions[lhs_heap_pos].values.len();
         let rhs_len = store.heap_regions[rhs_heap_pos].values.len();
         concatenated.reserve(lhs_len + rhs_len);
         for idx in 0..lhs_len {
             concatenated.push(store.clone_value(store.heap_regions[lhs_heap_pos].values[idx].clone()));
+        }
         for idx in 0..rhs_len {
             concatenated.push(store.clone_value(store.heap_regions[rhs_heap_pos].values[idx].clone()));
+        }
         store.heap_regions[target_heap_pos as usize].values = concatenated;
         return match value_kind{
             ValueKind::Message => Value::Message(target_heap_pos),
             ValueKind::String => Value::String(target_heap_pos),
             ValueKind::Array => Value::Array(target_heap_pos),
         };
+    }
     // If any of the values are references, retrieve the thing they're referring
     // to.
     let lhs = store.maybe_read_ref(lhs);
     let rhs = store.maybe_read_ref(rhs);
     match op {
         BO::Concatenate => unreachable!(),
         BO::LogicalOr => {
             return Value::Bool(lhs.as_bool() || rhs.as_bool());
         },
         BO::LogicalAnd => {
             return Value::Bool(lhs.as_bool() && rhs.as_bool());
         },
         BO::BitwiseOr        => { apply_int_op_and_return_self!(lhs, |,  op, rhs); },
         BO::BitwiseXor       => { apply_int_op_and_return_self!(lhs, ^,  op, rhs); },
         BO::BitwiseAnd       => { apply_int_op_and_return_self!(lhs, &,  op, rhs); },
         BO::Equality         => { Value::Bool(apply_equality_operator(store, lhs, rhs)) },
         BO::Inequality       => { Value::Bool(apply_inequality_operator(store, lhs, rhs)) },
         BO::LessThan         => { apply_int_op_and_return_bool!(lhs, <,  op, rhs); },
         BO::GreaterThan      => { apply_int_op_and_return_bool!(lhs, >,  op, rhs); },
         BO::LessThanEqual    => { apply_int_op_and_return_bool!(lhs, <=, op, rhs); },
         BO::GreaterThanEqual => { apply_int_op_and_return_bool!(lhs, >=, op, rhs); },
         BO::ShiftLeft        => { apply_int_op_and_return_self!(lhs, <<, op, rhs); },
         BO::ShiftRight       => { apply_int_op_and_return_self!(lhs, >>, op, rhs); },
         BO::Add              => { apply_int_op_and_return_self!(lhs, +,  op, rhs); },
         BO::Subtract         => { apply_int_op_and_return_self!(lhs, -,  op, rhs); },
         BO::Multiply         => { apply_int_op_and_return_self!(lhs, *,  op, rhs); },
         BO::Divide           => { apply_int_op_and_return_self!(lhs, /,  op, rhs); },
         BO::Remainder        => { apply_int_op_and_return_self!(lhs, %,  op, rhs); }
+    }
+}
 pub(crate) fn apply_unary_operator(store: &mut Store, op: UnaryOperator, value: &Value) -> Value {
     use UnaryOperator as UO;
     macro_rules! apply_int_expr_and_return {
         ($value:ident, $apply:tt, $op:ident) => {
             return match $value {
                 Value::UInt8(v)  => Value::UInt8($apply *v),
                 Value::UInt16(v) => Value::UInt16($apply *v),
                 Value::UInt32(v) => Value::UInt32($apply *v),
                 Value::UInt64(v) => Value::UInt64($apply *v),
                 Value::SInt8(v)  => Value::SInt8($apply *v),
                 Value::SInt16(v) => Value::SInt16($apply *v),
                 Value::SInt32(v) => Value::SInt32($apply *v),
                 Value::SInt64(v) => Value::SInt64($apply *v),
                 _ => unreachable!("apply_unary_operator {:?} on value {:?}", $op, $value),
+            }
+        }
+    }
     // If the value is a reference, retrieve the thing it is referring to
     let value = store.maybe_read_ref(value);
     match op {
         UO::Positive => {
             debug_assert!(value.is_integer());
             return value.clone();
         },
         UO::Negative => {
             // TODO: Error on negating unsigned integers
             return match value {
                 Value::SInt8(v) => Value::SInt8(-*v),
                 Value::SInt16(v) => Value::SInt16(-*v),
                 Value::SInt32(v) => Value::SInt32(-*v),
                 Value::SInt64(v) => Value::SInt64(-*v),
                 _ => unreachable!("apply_unary_operator {:?} on value {:?}", op, value),
+            }
         },
         UO::BitwiseNot => { apply_int_expr_and_return!(value, !, op); },
         UO::LogicalNot => { return Value::Bool(!value.as_bool()); },
+    }
+}
 pub(crate) fn apply_casting(store: &mut Store, output_type: &ConcreteType, subject: &Value) -> Result<Value, String> {
     // To simplify the casting logic: if the output type is not a simple
     // integer/boolean/character, then the type checker made sure that the two
     // types must be equal, hence we can do a simple clone.
     use ConcreteTypePart as CTP;
     let part = &output_type.parts[0];
     match part {
         CTP::Bool | CTP::Character |
         CTP::UInt8 | CTP::UInt16 | CTP::UInt32 | CTP::UInt64 |
         CTP::SInt8 | CTP::SInt16 | CTP::SInt32 | CTP::SInt64 => {
             // Do the checking of these below
             debug_assert_eq!(output_type.parts.len(), 1);
         },
         _ => {
             return Ok(store.clone_value(subject.clone()));
         },
+    }
     // Note: character is not included, needs per-type checking
     macro_rules! unchecked_cast {
         ($input: expr, $output_part: expr) => {
             return Ok(match $output_part {
                 CTP::UInt8 => Value::UInt8($input as u8),
                 CTP::UInt16 => Value::UInt16($input as u16),
                 CTP::UInt32 => Value::UInt32($input as u32),
                 CTP::UInt64 => Value::UInt64($input as u64),
                 CTP::SInt8 => Value::SInt8($input as i8),
                 CTP::SInt16 => Value::SInt16($input as i16),
                 CTP::SInt32 => Value::SInt32($input as i32),
                 CTP::SInt64 => Value::SInt64($input as i64),
                 _ => unreachable!()
             })
+        }
+    }
     macro_rules! from_unsigned_cast {
         ($input:expr, $input_type:ty, $output_part:expr) => {
+            {
                 let target_type_name = match $output_part {
                     CTP::Bool => return Ok(Value::Bool($input != 0)),
                     CTP::Character => if $input <= u8::MAX as $input_type {
                         return Ok(Value::Char(($input as u8) as char))
                     } else {
                         KW_TYPE_CHAR_STR
                     },
                     CTP::UInt8 => if $input <= u8::MAX as $input_type {
                         return Ok(Value::UInt8($input as u8))
                     } else {
                         KW_TYPE_UINT8_STR
                     },
                     CTP::UInt16 => if $input <= u16::MAX as $input_type {
                         return Ok(Value::UInt16($input as u16))
                     } else {
                         KW_TYPE_UINT16_STR
                     },
                     CTP::UInt32 => if $input <= u32::MAX as $input_type {
                         return Ok(Value::UInt32($input as u32))
                     } else {
                         KW_TYPE_UINT32_STR
                     },
                     CTP::UInt64 => return Ok(Value::UInt64($input as u64)), // any unsigned int to u64 is fine
                     CTP::SInt8 => if $input <= i8::MAX as $input_type {
                         return Ok(Value::SInt8($input as i8))
                     } else {
                         KW_TYPE_SINT8_STR
                     },
                     CTP::SInt16 => if $input <= i16::MAX as $input_type {
                         return Ok(Value::SInt16($input as i16))
                     } else {
                         KW_TYPE_SINT16_STR
                     },
                     CTP::SInt32 => if $input <= i32::MAX as $input_type {
                         return Ok(Value::SInt32($input as i32))
                     } else {
                         KW_TYPE_SINT32_STR
                     },
                     CTP::SInt64 => if $input <= i64::MAX as $input_type {
                         return Ok(Value::SInt64($input as i64))
                     } else {
                         KW_TYPE_SINT64_STR
                     },
                     _ => unreachable!(),
                 };
                 return Err(format!("value is '{}' which doesn't fit in a type '{}'", $input, target_type_name));
+            }
+        }
+    }
     macro_rules! from_signed_cast {
         // Programmer note: for signed checking we cannot do
         //  output_type::MAX as input_type,
         //

src/protocol/parser/mod.rs

➞

Show inline comments

 #[macro_use] mod visitor;
 pub(crate) mod symbol_table;
 pub(crate) mod type_table;
 pub(crate) mod tokens;
 pub(crate) mod token_parsing;
 pub(crate) mod pass_tokenizer;
 pub(crate) mod pass_symbols;
 pub(crate) mod pass_imports;
 pub(crate) mod pass_definitions;
 pub(crate) mod pass_definitions_types;
 pub(crate) mod pass_validation_linking;
 pub(crate) mod pass_rewriting;
 pub(crate) mod pass_typing;
 pub(crate) mod pass_stack_size;
 use tokens::*;
 use crate::collections::*;
 use visitor::Visitor;
 use pass_tokenizer::PassTokenizer;
 use pass_symbols::PassSymbols;
 use pass_imports::PassImport;
 use pass_definitions::PassDefinitions;
 use pass_validation_linking::PassValidationLinking;
 use pass_typing::{PassTyping, ResolveQueue};
 use pass_rewriting::PassRewriting;
 use pass_stack_size::PassStackSize;
 use symbol_table::*;
 use type_table::*;
 use crate::protocol::ast::*;
 use crate::protocol::input_source::*;
 use crate::protocol::ast_writer::ASTWriter;
 use crate::protocol::parser::type_table::PolymorphicVariable;
 use crate::protocol::token_writer::TokenWriter;
 const REOWOLF_PATH_ENV: &'static str = "REOWOLF_ROOT"; // first lookup reowolf path
 const REOWOLF_PATH_DIR: &'static str = "std"; // then try folder in current working directory
 #[derive(Debug, PartialEq, Eq, PartialOrd, Ord)]
 pub enum ModuleCompilationPhase {
     Tokenized,              // source is tokenized
     SymbolsScanned,         // all definitions are linked to their type class
     ImportsResolved,        // all imports are added to the symbol table
     DefinitionsParsed,      // produced the AST for the entire module
     TypesAddedToTable,      // added all definitions to the type table
     ValidatedAndLinked,     // AST is traversed and has linked the required AST nodes
     Typed,                  // Type inference and checking has been performed
     Rewritten,              // Special AST nodes are rewritten into regular AST nodes
     // When we continue with the compiler:
     // StackSize
+}
 pub struct Module {
     pub source: InputSource,
     pub tokens: TokenBuffer,
     pub is_compiler_file: bool, // TODO: @Hack for custom compiler-only types
     pub add_to_global_namespace: bool,
     pub root_id: RootId,
     pub name: Option<(PragmaId, StringRef<'static>)>,
     pub version: Option<(PragmaId, i64)>,
     pub phase: ModuleCompilationPhase,
+}
 pub struct TargetArch {
     pub void_type_id: TypeId,
     pub message_type_id: TypeId,
     pub bool_type_id: TypeId,
     pub uint8_type_id: TypeId,
     pub uint16_type_id: TypeId,
     pub uint32_type_id: TypeId,
     pub uint64_type_id: TypeId,
     pub sint8_type_id: TypeId,
     pub sint16_type_id: TypeId,
     pub sint32_type_id: TypeId,
     pub sint64_type_id: TypeId,
     pub char_type_id: TypeId,
     pub string_type_id: TypeId,
     pub array_type_id: TypeId,
     pub slice_type_id: TypeId,
     pub input_type_id: TypeId,
     pub output_type_id: TypeId,
     pub pointer_type_id: TypeId,
+}
 impl TargetArch {
     fn new() -> Self {
         return Self{
             void_type_id: TypeId::new_invalid(),
             bool_type_id: TypeId::new_invalid(),
             message_type_id: TypeId::new_invalid(),
             uint8_type_id: TypeId::new_invalid(),
             uint16_type_id: TypeId::new_invalid(),
             uint32_type_id: TypeId::new_invalid(),
             uint64_type_id: TypeId::new_invalid(),
             sint8_type_id: TypeId::new_invalid(),
             sint16_type_id: TypeId::new_invalid(),
             sint32_type_id: TypeId::new_invalid(),
             sint64_type_id: TypeId::new_invalid(),
             char_type_id: TypeId::new_invalid(),
             string_type_id: TypeId::new_invalid(),
             array_type_id: TypeId::new_invalid(),
             slice_type_id: TypeId::new_invalid(),
             input_type_id: TypeId::new_invalid(),
             output_type_id: TypeId::new_invalid(),
             pointer_type_id: TypeId::new_invalid(),
+        }
+    }
+}
 pub struct PassCtx<'a> {
     heap: &'a mut Heap,
     symbols: &'a mut SymbolTable,
     pool: &'a mut StringPool,
     arch: &'a TargetArch,
+}
 pub struct Parser {
     // Storage of all information created/gathered during compilation.
     pub(crate) heap: Heap,
     pub(crate) string_pool: StringPool, // Do not deallocate, holds all strings
     pub(crate) modules: Vec<Module>,
     pub(crate) symbol_table: SymbolTable,
     pub(crate) type_table: TypeTable,
     pub(crate) global_module_index: usize, // contains globals, implicitly imported everywhere
     // Compiler passes, used as little state machine that keep their memory
     // around.
     pass_tokenizer: PassTokenizer,
     pass_symbols: PassSymbols,
     pass_import: PassImport,
     pass_definitions: PassDefinitions,
     pass_validation: PassValidationLinking,
     pass_typing: PassTyping,
     pass_rewriting: PassRewriting,
     pass_stack_size: PassStackSize,
     // Compiler options
     pub write_tokens_to: Option<String>,
     pub write_ast_to: Option<String>,
     pub std_lib_dir: Option<String>,
     pub(crate) arch: TargetArch,
+}
 impl Parser {
     pub fn new(std_lib_dir: Option<String>) -> Result<Self, String> {
         let mut parser = Parser{
             heap: Heap::new(),
             string_pool: StringPool::new(),
             modules: Vec::new(),
             symbol_table: SymbolTable::new(),
             type_table: TypeTable::new(),
             global_module_index: 0,
             pass_tokenizer: PassTokenizer::new(),
             pass_symbols: PassSymbols::new(),
             pass_import: PassImport::new(),
             pass_definitions: PassDefinitions::new(),
             pass_validation: PassValidationLinking::new(),
             pass_typing: PassTyping::new(),
             pass_rewriting: PassRewriting::new(),
             pass_stack_size: PassStackSize::new(),
             write_tokens_to: None,
             write_ast_to: None,
             std_lib_dir,
             arch: TargetArch::new(),
         };
         parser.symbol_table.insert_scope(None, SymbolScope::Global);
         // Insert builtin types
         // TODO: At some point use correct values for size/alignment
         parser.arch.void_type_id    = insert_builtin_type(&mut parser.type_table, vec![ConcreteTypePart::Void], false, 0, 1);
         parser.arch.message_type_id = insert_builtin_type(&mut parser.type_table, vec![ConcreteTypePart::Message], false, 24, 8);
         parser.arch.bool_type_id    = insert_builtin_type(&mut parser.type_table, vec![ConcreteTypePart::Bool], false, 1, 1);
         parser.arch.uint8_type_id   = insert_builtin_type(&mut parser.type_table, vec![ConcreteTypePart::UInt8], false, 1, 1);
         parser.arch.uint16_type_id  = insert_builtin_type(&mut parser.type_table, vec![ConcreteTypePart::UInt16], false, 2, 2);
         parser.arch.uint32_type_id  = insert_builtin_type(&mut parser.type_table, vec![ConcreteTypePart::UInt32], false, 4, 4);
         parser.arch.uint64_type_id  = insert_builtin_type(&mut parser.type_table, vec![ConcreteTypePart::UInt64], false, 8, 8);
         parser.arch.sint8_type_id   = insert_builtin_type(&mut parser.type_table, vec![ConcreteTypePart::SInt8], false, 1, 1);
         parser.arch.sint16_type_id  = insert_builtin_type(&mut parser.type_table, vec![ConcreteTypePart::SInt16], false, 2, 2);
         parser.arch.sint32_type_id  = insert_builtin_type(&mut parser.type_table, vec![ConcreteTypePart::SInt32], false, 4, 4);
         parser.arch.sint64_type_id  = insert_builtin_type(&mut parser.type_table, vec![ConcreteTypePart::SInt64], false, 8, 8);
         parser.arch.char_type_id    = insert_builtin_type(&mut parser.type_table, vec![ConcreteTypePart::Character], false, 4, 4);
         parser.arch.string_type_id  = insert_builtin_type(&mut parser.type_table, vec![ConcreteTypePart::String], false, 24, 8);
         parser.arch.array_type_id   = insert_builtin_type(&mut parser.type_table, vec![ConcreteTypePart::Array, ConcreteTypePart::Void], true, 24, 8);
         parser.arch.slice_type_id   = insert_builtin_type(&mut parser.type_table, vec![ConcreteTypePart::Slice, ConcreteTypePart::Void], true, 16, 4);
         parser.arch.input_type_id   = insert_builtin_type(&mut parser.type_table, vec![ConcreteTypePart::Input, ConcreteTypePart::Void], true, 8, 8);
         parser.arch.output_type_id  = insert_builtin_type(&mut parser.type_table, vec![ConcreteTypePart::Output, ConcreteTypePart::Void], true, 8, 8);
         parser.arch.pointer_type_id = insert_builtin_type(&mut parser.type_table, vec![ConcreteTypePart::Pointer, ConcreteTypePart::Void], true, 8, 8);
         // Parse standard library
         parser.feed_standard_library()?;
         return Ok(parser)
+    }
     /// Feeds a new InputSource to the parser, which will tokenize it and store
     /// it internally for later parsing (when all modules are present). Returns
     /// the index of the new module.
     pub fn feed(&mut self, mut source: InputSource) -> Result<usize, ParseError> {
         return self.feed_internal(source, false, false);
+    }
     pub fn parse(&mut self) -> Result<(), ParseError> {
         let mut pass_ctx = PassCtx{
             heap: &mut self.heap,
             symbols: &mut self.symbol_table,
             pool: &mut self.string_pool,
             arch: &self.arch,
         };
         // Advance all modules to the phase where all symbols are scanned
         for module_idx in 0..self.modules.len() {
             self.pass_symbols.parse(&mut self.modules, module_idx, &mut pass_ctx)?;
+        }
         // With all symbols scanned, perform further compilation until we can
         // add all base types to the type table.
         for module_idx in 0..self.modules.len() {
             self.pass_import.parse(&mut self.modules, module_idx, &mut pass_ctx)?;
             self.pass_definitions.parse(&mut self.modules, module_idx, &mut pass_ctx)?;
+        }
         if let Some(filename) = &self.write_tokens_to {
             let mut writer = TokenWriter::new();
             let mut file = std::fs::File::create(std::path::Path::new(filename)).unwrap();
             writer.write(&mut file, &self.modules);
+        }
         // Add every known type to the type table
         self.type_table.build_base_types(&mut self.modules, &mut pass_ctx)?;
         // Continue compilation with the remaining phases now that the types
         // are all in the type table
         for module_idx in 0..self.modules.len() {
             let mut ctx = visitor::Ctx{
                 heap: &mut self.heap,
                 modules: &mut self.modules,
                 module_idx,
                 symbols: &mut self.symbol_table,
                 types: &mut self.type_table,
                 arch: &self.arch,
             };
             self.pass_validation.visit_module(&mut ctx)?;
+        }
         // Perform typechecking on all modules
         let mut queue = ResolveQueue::new();
         for module_idx in 0..self.modules.len() {
             let mut ctx = visitor::Ctx{
                 heap: &mut self.heap,
                 modules: &mut self.modules,
                 module_idx,
                 symbols: &mut self.symbol_table,
                 types: &mut self.type_table,
                 arch: &self.arch,
             };
             self.pass_typing.queue_module_definitions(&mut ctx, &mut queue);
         };
         while !queue.is_empty() {
             let top = queue.pop_front().unwrap();
             let mut ctx = visitor::Ctx{
                 heap: &mut self.heap,
                 modules: &mut self.modules,
                 module_idx: top.root_id.index as usize,
                 symbols: &mut self.symbol_table,
                 types: &mut self.type_table,
                 arch: &self.arch,
             };
             self.pass_typing.handle_module_definition(&mut ctx, &mut queue, top)?;
+        }
         // Rewrite nodes in tree, then prepare for execution of code
         for module_idx in 0..self.modules.len() {
             self.modules[module_idx].phase = ModuleCompilationPhase::Typed;
             let mut ctx = visitor::Ctx{
                 heap: &mut self.heap,
                 modules: &mut self.modules,
                 module_idx,
                 symbols: &mut self.symbol_table,
                 types: &mut self.type_table,
                 arch: &self.arch,
             };
             self.pass_rewriting.visit_module(&mut ctx)?;
             self.pass_stack_size.visit_module(&mut ctx)?;
+        }
         // Write out desired information
         if let Some(filename) = &self.write_ast_to {
             let mut writer = ASTWriter::new();
             let mut file = std::fs::File::create(std::path::Path::new(filename)).unwrap();
             writer.write_ast(&mut file, &self.heap);
+        }
         Ok(())
+    }
     /// Tries to find the standard library and add the files for parsing.
     fn feed_standard_library(&mut self) -> Result<(), String> {
         use std::env;
         use std::path::{Path, PathBuf};
         use std::fs;
         // Pair is (name, add_to_global_namespace)
         const FILES: [(&'static str, bool); 3] = [
             ("std.global.pdl", true),
             ("std.internet.pdl", false),
             ("std.random.pdl", false),
         ];
         // Determine base directory
         let (base_path, from_env) = if let Ok(path) = env::var(REOWOLF_PATH_ENV) {
             // Path variable is set
             (path, true)
         } else {
             let path = match self.std_lib_dir.take() {
                 Some(path) => path,
                 None => {
                     let mut path = String::with_capacity(REOWOLF_PATH_DIR.len() + 2);
                     path.push_str("./");
                     path.push_str(REOWOLF_PATH_DIR);
                     path
+                }
             };
             (path, false)
         };
         // Make sure directory exists
         let path = Path::new(&base_path);
         if !path.exists() {
             return Err(format!("std lib root directory '{}' does not exist", base_path));
             let from_env_message = if from_env {
                 format!(" (retrieved from the environment variable '{}')", REOWOLF_PATH_ENV)
             } else {
                 String::new()
             };
             return Err(format!("std lib root directory '{}'{} does not exist", base_path, from_env_message));
+        }
         // Try to load all standard library files. We might need a more unified
         // way to do this in the future (i.e. a "std" package, containing all
         // of the modules)
         let mut file_path = PathBuf::new();
         let mut first_file = true;
         for (file, add_to_global_namespace) in FILES {
             file_path.clear();
             file_path.push(path);
             file_path.push(file);
             let source = fs::read(file_path.as_path());
             if let Err(err) = source {
                 return Err(format!(
                     "failed to read std lib file '{}' in root directory '{}', because: {}",
                     file, base_path, err
                 ));
+            }
             let source = source.unwrap();
             let input_source = InputSource::new(file.to_string(), source);
             let module_index = self.feed_internal(input_source, true, add_to_global_namespace);
             if let Err(err) = module_index {
                 // A bit of a hack, but shouldn't really happen anyway: the
                 // compiler should ship with a decent standard library (at some
                 // point)
                 return Err(format!("{}", err));
+            }
             let module_index = module_index.unwrap();
             if first_file {
                 self.global_module_index = module_index;
                 first_file = false;
+            }
+        }
         return Ok(())
+    }
     fn feed_internal(&mut self, mut source: InputSource, is_compiler_file: bool, add_to_global_namespace: bool) -> Result<usize, ParseError> {
         let mut token_buffer = TokenBuffer::new();
         self.pass_tokenizer.tokenize(&mut source, &mut token_buffer)?;
         let module = Module{
             source,
             tokens: token_buffer,
             is_compiler_file,
             add_to_global_namespace,
             root_id: RootId::new_invalid(),
             name: None,
             version: None,
             phase: ModuleCompilationPhase::Tokenized,
         };
         let module_index = self.modules.len();
         self.modules.push(module);
         return Ok(module_index);
+    }
+}
 fn insert_builtin_type(type_table: &mut TypeTable, parts: Vec<ConcreteTypePart>, has_poly_var: bool, size: usize, alignment: usize) -> TypeId {
     const POLY_VARS: [PolymorphicVariable; 1] = [PolymorphicVariable{
         identifier: Identifier::new_empty(InputSpan::new()),
         is_in_use: false,
     }];
     let concrete_type = ConcreteType{ parts };
     let poly_var = if has_poly_var {
         POLY_VARS.as_slice()
     } else {
         &[]
     };
     return type_table.add_builtin_data_type(concrete_type, poly_var, size, alignment);
+}
@@ \ No newline at end of file @@

src/runtime2/communication.rs

➞

Show inline comments

 use crate::protocol::eval::*;
 use crate::protocol::eval::value::ValueId;
 use super::runtime::*;
 use super::component::*;
 // -----------------------------------------------------------------------------
 // Generic types
 // -----------------------------------------------------------------------------
 #[derive(Debug, Copy, Clone, PartialEq, Eq)]
 pub struct PortId(pub u32);
 impl PortId {
     /// This value is not significant, it is chosen to make debugging easier: a
     /// very large port number is more likely to shine a light on bugs.
     pub fn new_invalid() -> Self {
         return Self(u32::MAX);
+    }
+}
 pub struct Channel {
     pub putter_id: PortId,
     pub getter_id: PortId,
+}
 // -----------------------------------------------------------------------------
 // Data messages
 // -----------------------------------------------------------------------------
 #[derive(Debug)]
 pub struct DataMessage {
     pub data_header: MessageDataHeader,
     pub sync_header: MessageSyncHeader,
     pub content: ValueGroup,
     pub ports: Vec<TransmittedPort>,
+}
 #[derive(Debug)]
 pub enum PortAnnotationKind {
     Getter(PortAnnotationGetter),
     Putter(PortAnnotationPutter),
+}
 #[derive(Debug)]
 pub struct PortAnnotationGetter {
     pub self_comp_id: CompId,
     pub self_port_id: PortId,
     pub peer_comp_id: CompId,
     pub peer_port_id: PortId,
+}
 #[derive(Debug)]
 pub struct PortAnnotationPutter {
     pub self_comp_id: CompId,
     pub self_port_id: PortId,
+}
 #[derive(Debug)]
 pub struct MessageDataHeader {
     pub expected_mapping: Vec<(PortAnnotationKind, Option<u32>)>,
     pub new_mapping: u32,
     pub source_port: PortId,
     pub target_port: PortId,
+}
 #[derive(Debug)]
 pub struct TransmittedPort {
     pub locations: Vec<ValueId>, // within `content`
     pub messages: Vec<DataMessage>, // owned by previous component
     pub peer_comp: CompId,
     pub peer_port: PortId,
     pub kind: PortKind,
     pub state: PortState,
+}
 // -----------------------------------------------------------------------------
 // Sync messages
 // -----------------------------------------------------------------------------
 #[derive(Debug)]
 pub struct SyncMessage {
     pub sync_header: MessageSyncHeader,
     pub content: SyncMessageContent,
+}
 #[derive(Debug)]
 pub enum SyncLocalSolutionEntry {
     Putter(SyncSolutionPutterPort),
     Getter(SyncSolutionGetterPort),
+}
 pub type SyncLocalSolution = Vec<SyncLocalSolutionEntry>;
 /// Getter port in a solution. Upon receiving a message it is certain about who
 /// its peer is.
 #[derive(Debug)]
 pub struct SyncSolutionGetterPort {
     pub self_comp_id: CompId,
     pub self_port_id: PortId,
     pub peer_comp_id: CompId,
     pub peer_port_id: PortId,
     pub mapping: u32,
     pub failed: bool,
+}
 /// Putter port in a solution. A putter may not be certain about who its peer
 /// component/port is.
 #[derive(Debug)]
 pub struct SyncSolutionPutterPort {
     pub self_comp_id: CompId,
     pub self_port_id: PortId,
     pub mapping: u32,
     pub failed: bool,
+}
 #[derive(Debug)]
 pub struct SyncSolutionChannel {
     pub putter: Option<SyncSolutionPutterPort>,
     pub getter: Option<SyncSolutionGetterPort>,
+}
 #[derive(Debug, Copy, Clone, PartialEq, Eq)]
 pub enum SyncRoundDecision {
     None,
     Solution,
     Failure,
+}
 #[derive(Debug)]
 pub struct SyncPartialSolution {
     pub channel_mapping: Vec<SyncSolutionChannel>,
     pub decision: SyncRoundDecision,
+}
 impl Default for SyncPartialSolution {
     fn default() -> Self {
         return Self{
             channel_mapping: Vec::new(),
             decision: SyncRoundDecision::None,
+        }
+    }
+}
 #[derive(Debug)]
 pub enum SyncMessageContent {
     NotificationOfLeader,
     LocalSolution(CompId, SyncLocalSolution), // local solution of the specified component
     PartialSolution(SyncPartialSolution), // partial solution of multiple components
     GlobalSolution,
     GlobalFailure,
+}
 // -----------------------------------------------------------------------------
 // Control messages
 // -----------------------------------------------------------------------------
 #[derive(Debug)]
 pub struct ControlMessage {
     pub(crate) id: ControlId,
     pub sender_comp_id: CompId,
     pub target_port_id: Option<PortId>,
     pub content: ControlMessageContent,
+}
 /// Content of a control message. If the content refers to a port then the
 /// `target_port_id` field is the one that it refers to.
 #[derive(Copy, Clone, Debug)]
 pub enum ControlMessageContent {
     Ack,
     BlockPort,
     UnblockPort,
     ClosePort(ControlMessageClosePort),
     PortPeerChangedBlock,
     PortPeerChangedUnblock(PortId, CompId), // contains (new_port_id, new_component_id)
+}
 #[derive(Copy, Clone, Debug)]
 pub struct ControlMessageClosePort {
     pub closed_in_sync_round: bool, // needed to ensure correct handling of errors
+}
 // -----------------------------------------------------------------------------
 // Messages (generic)
 // -----------------------------------------------------------------------------
 #[derive(Debug)]
 pub struct MessageSyncHeader {
     pub sync_round: u32,
     pub sending_id: CompId,
     pub highest_id: CompId,
+}
 #[derive(Debug)]
 pub enum Message {
     Data(DataMessage),
     Sync(SyncMessage),
     Control(ControlMessage),
     Poll,
+}
 impl Message {
     pub(crate) fn target_port(&self) -> Option<PortId> {
         match self {
             Message::Data(v) =>
                 return Some(v.data_header.target_port),
             Message::Control(v) =>
                 return v.target_port_id,
             Message::Sync(_) =>
                 return None,
             Message::Poll =>
                 return None,
+        }
+    }
     pub(crate) fn modify_target_port(&mut self, port_id: PortId) {
         match self {
             Message::Data(v) =>
                 v.data_header.target_port = port_id,
             Message::Control(v) =>
                 v.target_port_id = Some(port_id),
             Message::Sync(_) => unreachable!(), // should never be called for this message type
             Message::Poll => unreachable!(),
+        }
+    }
+}

src/runtime2/component/component.rs

➞

Show inline comments

 /*
  * Default toolkit for creating components. Contains handlers for initiating and
  * responding to various events.
  */
 use std::fmt::{Display as FmtDisplay, Result as FmtResult, Formatter};
 use crate::protocol::eval::{Prompt, EvalError, ValueGroup, PortId as EvalPortId};
+use crate::protocol::eval::{Prompt, EvalError, ValueGroup, Value, ValueId, PortId as EvalPortId};
 use crate::protocol::*;
 use crate::runtime2::*;
 use crate::runtime2::communication::*;
 use super::{CompCtx, CompPDL, CompId};
 use super::component_context::*;
 use super::component_random::*;
 use super::component_internet::*;
 use super::control_layer::*;
 use super::consensus::*;
 pub enum CompScheduling {
     Immediate,
     Requeue,
     Sleep,
     Exit,
+}
 /// Potential error emitted by a component
 pub enum CompError {
     /// Error originating from the code executor. Hence has an associated
     /// source location.
     Executor(EvalError),
     /// Error originating from a component, but not necessarily associated with
     /// a location in the source.
     Component(String), // TODO: Maybe a different embedded value in the future?
     /// Pure runtime error. Not necessarily originating from the component
     /// itself. Should be treated as a very severe runtime-compromising error.
     Runtime(RtError),
+}
 impl FmtDisplay for CompError {
     fn fmt(&self, f: &mut Formatter<'_>) -> FmtResult {
         match self {
             CompError::Executor(v) => v.fmt(f),
             CompError::Component(v) => v.fmt(f),
             CompError::Runtime(v) => v.fmt(f),
+        }
+    }
+}
 /// Generic representation of a component (as viewed by a scheduler).
 pub(crate) trait Component {
     /// Called upon the creation of the component. Note that the scheduler
     /// context is officially running another component (the component that is
     /// creating the new component).
     fn on_creation(&mut self, comp_id: CompId, sched_ctx: &SchedulerCtx);
     /// Called when a component crashes or wishes to exit. So is not called
     /// right before destruction, other components may still hold a handle to
     /// the component and send it messages!
     fn on_shutdown(&mut self, sched_ctx: &SchedulerCtx);
     /// Called if the component is created by another component and the messages
     /// are being transferred between the two.
     fn adopt_message(&mut self, comp_ctx: &mut CompCtx, message: DataMessage);
     /// Called if the component receives a new message. The component is
     /// responsible for deciding where that messages goes.
     fn handle_message(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx, message: Message);
     /// Called if the component's routine should be executed. The return value
     /// can be used to indicate when the routine should be run again.
     fn run(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx) -> CompScheduling;
+}
 /// Representation of the generic operating mode of a component. Although not
 /// every state may be used by every kind of (builtin) component, this allows
 /// writing standard handlers for particular events in a component's lifetime.
 #[derive(Debug, Copy, Clone, PartialEq, Eq)]
 pub(crate) enum CompMode {
     NonSync, // not in sync mode
     Sync, // in sync mode, can interact with other components
     SyncEnd, // awaiting a solution, i.e. encountered the end of the sync block
     BlockedGet, // blocked because we need to receive a message on a particular port
     BlockedPut, // component is blocked because the port is blocked
     BlockedSelect, // waiting on message to complete the select statement
     PutPortsBlockedTransferredPorts, // sending a message with ports, those sent ports are (partly) blocked
     PutPortsBlockedAwaitingAcks, // sent out PPC message for blocking transferred ports, now awaiting Acks
     PutPortsBlockedSendingPort, // sending a message with ports, message sent through a still-blocked port
     NewComponentBlocked, // waiting until ports are in the appropriate state to create a new component
     StartExit, // temporary state: if encountered then we start the shutdown process.
     BusyExit, // temporary state: waiting for Acks for all the closed ports, potentially waiting for sync round to finish
     Exit, // exiting: shutdown process started, now waiting until the reference count drops to 0
+}
 impl CompMode {
     pub(crate) fn is_in_sync_block(&self) -> bool {
         use CompMode::*;
         match self {
             Sync | SyncEnd | BlockedGet | BlockedPut | BlockedSelect => true,
             NonSync | StartExit | BusyExit | Exit => false,
             Sync | SyncEnd | BlockedGet | BlockedPut | BlockedSelect |
                 PutPortsBlockedTransferredPorts |
                 PutPortsBlockedAwaitingAcks |
                 PutPortsBlockedSendingPort => true,
             NonSync | NewComponentBlocked | StartExit | BusyExit | Exit => false,
+        }
+    }
     pub(crate) fn is_busy_exiting(&self) -> bool {
         use CompMode::*;
         match self {
             NonSync | Sync | SyncEnd | BlockedGet | BlockedPut | BlockedSelect => false,
             NonSync | Sync | SyncEnd | BlockedGet | BlockedPut | BlockedSelect |
             PutPortsBlockedTransferredPorts |
             PutPortsBlockedAwaitingAcks |
             PutPortsBlockedSendingPort |
                 NewComponentBlocked => false,
             StartExit | BusyExit => true,
             Exit => false,
+        }
+    }
+}
 #[derive(Debug)]
 pub(crate) enum ExitReason {
     Termination, // regular termination of component
     ErrorInSync,
     ErrorNonSync,
+}
 impl ExitReason {
     pub(crate) fn is_in_sync(&self) -> bool {
         use ExitReason::*;
         match self {
             Termination | ErrorNonSync => false,
             ErrorInSync => true,
+        }
+    }
     pub(crate) fn is_error(&self) -> bool {
         use ExitReason::*;
         match self {
             Termination => false,
             ErrorInSync | ErrorNonSync => true,
+        }
+    }
+}
 /// Component execution state: the execution mode along with some descriptive
 /// fields. Fields are public for ergonomic reasons, use member functions when
 /// appropriate.
 pub(crate) struct CompExecState {
     pub mode: CompMode,
     pub mode_port: PortId, // valid if blocked on a port (put/get)
     pub mode_value: ValueGroup, // valid if blocked on a put
     pub mode_component: (ProcedureDefinitionId, TypeId),
     pub exit_reason: ExitReason, // valid if in StartExit/BusyExit/Exit mode
+}
 impl CompExecState {
     pub(crate) fn new() -> Self {
         return Self{
             mode: CompMode::NonSync,
             mode_port: PortId::new_invalid(),
             mode_value: ValueGroup::default(),
             mode_component: (ProcedureDefinitionId::new_invalid(), TypeId::new_invalid()),
             exit_reason: ExitReason::Termination,
+        }
+    }
     pub(crate) fn set_as_start_exit(&mut self, reason: ExitReason) {
         self.mode = CompMode::StartExit;
         self.exit_reason = reason;
+    }
     pub(crate) fn set_as_blocked_get(&mut self, port: PortId) {
         self.mode = CompMode::BlockedGet;
         self.mode_port = port;
         debug_assert!(self.mode_value.values.is_empty());
+    }
     pub(crate) fn set_as_create_component_blocked(
         &mut self, proc_id: ProcedureDefinitionId, type_id: TypeId,
         arguments: ValueGroup
     ) {
         self.mode = CompMode::NewComponentBlocked;
         self.mode_value = arguments;
         self.mode_component = (proc_id, type_id);
+    }
     pub(crate) fn is_blocked_on_get(&self, port: PortId) -> bool {
         return
             self.mode == CompMode::BlockedGet &&
             self.mode_port == port;
+    }
-    pub(crate) fn set_as_blocked_put(&mut self, port: PortId, value: ValueGroup) {
+    pub(crate) fn set_as_blocked_put_without_ports(&mut self, port: PortId, value: ValueGroup) {
         self.mode = CompMode::BlockedPut;
         self.mode_port = port;
         self.mode_value = value;
+    }
     pub(crate) fn is_blocked_on_put(&self, port: PortId) -> bool {
     pub(crate) fn set_as_blocked_put_with_ports(&mut self, port: PortId, value: ValueGroup) {
         self.mode = CompMode::PutPortsBlockedTransferredPorts;
         self.mode_port = port;
         self.mode_value = value;
+    }
     pub(crate) fn is_blocked_on_put_without_ports(&self, port: PortId) -> bool {
         return
             self.mode == CompMode::BlockedPut &&
             self.mode_port == port;
+    }
     pub(crate) fn is_blocked_on_create_component(&self) -> bool {
         return self.mode == CompMode::NewComponentBlocked;
+    }
+}
 // TODO: Replace when implementing port sending. Should probably be incorporated
 //  into CompCtx (and rename CompCtx into CompComms)
 pub(crate) type InboxMain = Vec<Option<DataMessage>>;
 pub(crate) type InboxMainRef = [Option<DataMessage>];
 pub(crate) type InboxBackup = Vec<DataMessage>;
 /// Creates a new component based on its definition. Meaning that if it is a
 /// user-defined component then we set up the PDL code state. Otherwise we
 /// construct a custom component. This does NOT take care of port and message
 /// management.
 pub(crate) fn create_component(
     protocol: &ProtocolDescription,
     definition_id: ProcedureDefinitionId, type_id: TypeId,
     arguments: ValueGroup, num_ports: usize
 ) -> Box<dyn Component> {
     let definition = &protocol.heap[definition_id];
     debug_assert!(definition.kind == ProcedureKind::Primitive || definition.kind == ProcedureKind::Composite);
     if definition.source.is_builtin() {
         // Builtin component
         let component: Box<dyn Component> = match definition.source {
             ProcedureSource::CompRandomU32 => Box::new(ComponentRandomU32::new(arguments)),
             ProcedureSource::CompTcpClient => Box::new(ComponentTcpClient::new(arguments)),
             _ => unreachable!(),
         };
         return component;
     } else {
         // User-defined component
         let prompt = Prompt::new(
             &protocol.types, &protocol.heap,
             definition_id, type_id, arguments
         );
         let component = CompPDL::new(prompt, num_ports);
         return Box::new(component);
+    }
+}
 // -----------------------------------------------------------------------------
 // Generic component messaging utilities (for sending and receiving)
 // -----------------------------------------------------------------------------
 /// Default handling of sending a data message. In case the port is blocked then
 /// the `ExecState` will become blocked as well. Note that
 /// `default_handle_control_message` will ensure that the port becomes
 /// unblocked if so instructed by the receiving component. The returned
 /// scheduling value must be used.
 #[must_use]
 pub(crate) fn default_send_data_message(
     exec_state: &mut CompExecState, transmitting_port_id: PortId,
     port_instruction: PortInstruction, value: ValueGroup,
     sched_ctx: &SchedulerCtx, consensus: &mut Consensus, comp_ctx: &mut CompCtx
     sched_ctx: &SchedulerCtx, consensus: &mut Consensus,
     control: &mut ControlLayer, comp_ctx: &mut CompCtx
 ) -> Result<CompScheduling, (PortInstruction, String)> {
     debug_assert_eq!(exec_state.mode, CompMode::Sync);
     let port_handle = comp_ctx.get_port_handle(transmitting_port_id);
     let port_info = comp_ctx.get_port_mut(port_handle);
     port_info.last_instruction = port_instruction;
     let port_info = comp_ctx.get_port(port_handle);
     debug_assert_eq!(port_info.kind, PortKind::Putter);
     let mut ports = Vec::new();
     find_ports_in_value_group(&value, &mut ports);
     if port_info.state.is_closed() {
         // Note: normally peer is eventually consistent, but if it has shut down
         // then we can be sure it is consistent (I think?)
         return Err((
             port_info.last_instruction,
             format!("Cannot send on this port, as the peer (id:{}) has shut down", port_info.peer_comp_id.0)
         ))
     } else if !ports.is_empty() {
         start_send_message_with_ports(
             transmitting_port_id, port_instruction, value, exec_state,
             comp_ctx, sched_ctx, control
         )?;
         return Ok(CompScheduling::Sleep);
     } else if port_info.state.is_blocked() {
         // Port is blocked, so we cannot send
-        exec_state.set_as_blocked_put(transmitting_port_id, value);
+        exec_state.set_as_blocked_put_without_ports(transmitting_port_id, value);
         return Ok(CompScheduling::Sleep);
     } else {
-        // Port is not blocked, so send to the peer
+        // Port is not blocked and no ports to transfer: send to the peer
         let peer_handle = comp_ctx.get_peer_handle(port_info.peer_comp_id);
         let peer_info = comp_ctx.get_peer(peer_handle);
         let annotated_message = consensus.annotate_data_message(comp_ctx, port_info, value);
         peer_info.handle.send_message_logged(sched_ctx, Message::Data(annotated_message), true);
         return Ok(CompScheduling::Immediate);
+    }
+}
 pub(crate) enum IncomingData {
     PlacedInSlot,
     SlotFull(DataMessage),
+}
 /// Default handling of receiving a data message. In case there is no room for
 /// the message it is returned from this function. Note that this function is
 /// different from PDL code performing a `get` on a port; this is the case where
 /// the message first arrives at the component.
 // NOTE: This is supposed to be a somewhat temporary implementation. It would be
 //  nicest if the sending component can figure out it cannot send any more data.
 #[must_use]
 pub(crate) fn default_handle_incoming_data_message(
     exec_state: &mut CompExecState, inbox_main: &mut InboxMain,
     comp_ctx: &mut CompCtx, incoming_message: DataMessage,
     sched_ctx: &SchedulerCtx, control: &mut ControlLayer
 ) -> IncomingData {
     let port_handle = comp_ctx.get_port_handle(incoming_message.data_header.target_port);
     let port_index = comp_ctx.get_port_index(port_handle);
     comp_ctx.get_port_mut(port_handle).received_message_for_sync = true;
     let port_value_slot = &mut inbox_main[port_index];
     let target_port_id = incoming_message.data_header.target_port;
     if port_value_slot.is_none() {
         // We can put the value in the slot
         *port_value_slot = Some(incoming_message);
         // Check if we're blocked on receiving this message.
         dbg_code!({
             // Our port cannot have been blocked itself, because we're able to
             // directly insert the message into its slot.
             assert!(!comp_ctx.get_port(port_handle).state.is_blocked());
         });
         if exec_state.is_blocked_on_get(target_port_id) {
             // Return to normal operation
             exec_state.mode = CompMode::Sync;
             exec_state.mode_port = PortId::new_invalid();
             debug_assert!(exec_state.mode_value.values.is_empty());
+        }
         return IncomingData::PlacedInSlot
     } else {
         // Slot is already full, so if the port was previously opened, it will
         // now become closed
         let port_info = comp_ctx.get_port_mut(port_handle);
         if port_info.state.is_open() {
             port_info.state.set(PortStateFlag::BlockedDueToFullBuffers);
             let (peer_handle, message) =
                 control.initiate_port_blocking(comp_ctx, port_handle);
             let peer = comp_ctx.get_peer(peer_handle);
             peer.handle.send_message_logged(sched_ctx, Message::Control(message), true);
+        }
         return IncomingData::SlotFull(incoming_message)
+    }
+}
 pub(crate) enum GetResult {
     Received(DataMessage),
     NoMessage,
     Error((PortInstruction, String)),
+}
 /// Default attempt at trying to receive from a port (i.e. through a `get`, or
 /// the equivalent operation for a builtin component). `target_port` is the port
 /// we're trying to receive from, and the `target_port_instruction` is the
 /// instruction we're attempting on this port.
 pub(crate) fn default_attempt_get(
     exec_state: &mut CompExecState, target_port: PortId, target_port_instruction: PortInstruction,
-    inbox_main: &mut InboxMainRef, inbox_backup: &mut InboxBackup, sched_ctx: &SchedulerCtx,
+    inbox_main: &mut InboxMain, inbox_backup: &mut InboxBackup, sched_ctx: &SchedulerCtx,
     comp_ctx: &mut CompCtx, control: &mut ControlLayer, consensus: &mut Consensus
 ) -> GetResult {
     let port_handle = comp_ctx.get_port_handle(target_port);
     let port_index = comp_ctx.get_port_index(port_handle);
     let port_info = comp_ctx.get_port_mut(port_handle);
     port_info.last_instruction = target_port_instruction;
     if port_info.state.is_closed() {
         let peer_id = port_info.peer_comp_id;
         return GetResult::Error((
             target_port_instruction,
             format!("Cannot get from this port, as the peer component (id:{}) closed the port", peer_id.0)
         ));
+    }
     if let Some(message) = &inbox_main[port_index] {
         if consensus.try_receive_data_message(sched_ctx, comp_ctx, message) {
             // We're allowed to receive this message
             let message = inbox_main[port_index].take().unwrap();
+            let mut message = inbox_main[port_index].take().unwrap();
             debug_assert_eq!(target_port, message.data_header.target_port);
             // Note: we can still run into an unrecoverable error when actually
             // receiving this message
             match default_handle_received_data_message(
                 target_port, target_port_instruction, inbox_main, inbox_backup,
                 comp_ctx, sched_ctx, control,
                 target_port, target_port_instruction,
                 &mut message, inbox_main, inbox_backup,
                 comp_ctx, sched_ctx, control, consensus
             ) {
                 Ok(()) => return GetResult::Received(message),
                 Err(location_and_message) => return GetResult::Error(location_and_message)
+            }
         } else {
             // We're not allowed to receive this message. This means that the
             // receiver is attempting to receive something out of order with
             // respect to the sender.
             return GetResult::Error((target_port_instruction, String::from(
                 "Cannot get from this port, as this causes a deadlock. This happens if you `get` in a different order as another component `put`s"
             )));
+        }
     } else {
         // We don't have a message waiting for us and the port is not blocked.
         // So enter the BlockedGet state
         exec_state.set_as_blocked_get(target_port);
         return GetResult::NoMessage;
+    }
+}
 /// Default handling that has been received through a `get`. Will check if any
 /// more messages are waiting, and if the corresponding port was blocked because
 /// of full buffers (hence, will use the control layer to make sure the peer
 /// will become unblocked).
 pub(crate) fn default_handle_received_data_message(
     targeted_port: PortId, port_instruction: PortInstruction,
     inbox_main: &mut InboxMainRef, inbox_backup: &mut InboxBackup,
     comp_ctx: &mut CompCtx, sched_ctx: &SchedulerCtx, control: &mut ControlLayer
     targeted_port: PortId, _port_instruction: PortInstruction, message: &mut DataMessage,
     inbox_main: &mut InboxMain, inbox_backup: &mut InboxBackup,
     comp_ctx: &mut CompCtx, sched_ctx: &SchedulerCtx, control: &mut ControlLayer,
     consensus: &mut Consensus
 ) -> Result<(), (PortInstruction, String)> {
     let port_handle = comp_ctx.get_port_handle(targeted_port);
     let port_index = comp_ctx.get_port_index(port_handle);
     let slot = &mut inbox_main[port_index];
     debug_assert!(slot.is_none()); // because we've just received from it
     debug_assert!(inbox_main[port_index].is_none()); // because we've just received from it
     // If we received any ports, add them to the port tracking and inbox struct.
     // Then notify the peers that they can continue sending to this port, but
     // now at a new address.
     for received_port in &mut message.ports {
         // Transfer messages to main/backup inbox
         let _new_inbox_index = inbox_main.len();
         if !received_port.messages.is_empty() {
             inbox_main.push(Some(received_port.messages.remove(0)));
             inbox_backup.extend(received_port.messages.drain(..));
         } else {
             inbox_main.push(None);
+        }
         // Create a new port locally
         let mut new_port_state = received_port.state;
         new_port_state.set(PortStateFlag::Received);
         let new_port_handle = comp_ctx.add_port(
             received_port.peer_comp, received_port.peer_port,
             received_port.kind, new_port_state
         );
         debug_assert_eq!(_new_inbox_index, comp_ctx.get_port_index(new_port_handle));
         comp_ctx.change_port_peer(sched_ctx, new_port_handle, Some(received_port.peer_comp));
         let new_port = comp_ctx.get_port(new_port_handle);
         // Add the port tho the consensus
         consensus.notify_received_port(_new_inbox_index, new_port_handle, comp_ctx);
         // Replace all references to the port in the received message
         for message_location in received_port.locations.iter().copied() {
             let value = message.content.get_value_mut(message_location);
             match value {
                 Value::Input(_) => {
                     debug_assert_eq!(new_port.kind, PortKind::Getter);
                     *value = Value::Input(port_id_to_eval(new_port.self_id));
                 },
                 Value::Output(_) => {
                     debug_assert_eq!(new_port.kind, PortKind::Putter);
                     *value = Value::Output(port_id_to_eval(new_port.self_id));
                 },
                 _ => unreachable!(),
+            }
+        }
         // Let the peer know that the port can now be used
         let peer_handle = comp_ctx.get_peer_handle(new_port.peer_comp_id);
         let peer_info = comp_ctx.get_peer(peer_handle);
         peer_info.handle.send_message_logged(sched_ctx, Message::Control(ControlMessage{
             id: ControlId::new_invalid(),
             sender_comp_id: comp_ctx.id,
             target_port_id: Some(new_port.peer_port_id),
             content: ControlMessageContent::PortPeerChangedUnblock(new_port.self_id, comp_ctx.id)
         }), true);
+    }
     // Modify last-known location where port instruction was retrieved
     let port_info = comp_ctx.get_port(port_handle);
     debug_assert_ne!(port_info.last_instruction, PortInstruction::None); // set by caller
     debug_assert!(port_info.state.is_open()); // checked by caller
     // Check if there are any more messages in the backup buffer
     for message_index in 0..inbox_backup.len() {
         let message = &inbox_backup[message_index];
         if message.data_header.target_port == targeted_port {
             // One more message, place it in the slot
             let message = inbox_backup.remove(message_index);
             debug_assert!(comp_ctx.get_port(port_handle).state.is_blocked()); // since we're removing another message from the backup
-            *slot = Some(message);
+            inbox_main[port_index] = Some(message);
             return Ok(());
+        }
+    }
     // Did not have any more messages, so if we were blocked, then we need to
     // unblock the port now (and inform the peer of this unblocking)
     if port_info.state.is_set(PortStateFlag::BlockedDueToFullBuffers) {
         let port_info = comp_ctx.get_port_mut(port_handle);
         port_info.state.clear(PortStateFlag::BlockedDueToFullBuffers);
         let (peer_handle, message) = control.cancel_port_blocking(comp_ctx, port_handle);
         let peer_info = comp_ctx.get_peer(peer_handle);
         peer_info.handle.send_message_logged(sched_ctx, Message::Control(message), true);
+    }
     return Ok(());
+}
 /// Handles control messages in the default way. Note that this function may
 /// take a lot of actions in the name of the caller: pending messages may be
 /// sent, ports may become blocked/unblocked, etc. So the execution
 /// (`CompExecState`), control (`ControlLayer`) and consensus (`Consensus`)
 /// state may all change.
 pub(crate) fn default_handle_control_message(
     exec_state: &mut CompExecState, control: &mut ControlLayer, consensus: &mut Consensus,
     message: ControlMessage, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx
     message: ControlMessage, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx,
     inbox_main: &mut InboxMain, inbox_backup: &mut InboxBackup
 ) -> Result<(), (PortInstruction, String)> {
     match message.content {
         ControlMessageContent::Ack => {
             default_handle_ack(control, message.id, sched_ctx, comp_ctx);
+            default_handle_ack(exec_state, control, message.id, sched_ctx, comp_ctx, consensus, inbox_main, inbox_backup)?;
         },
         ControlMessageContent::BlockPort => {
             // One of our messages was accepted, but the port should be
             // blocked.
             let port_to_block = message.target_port_id.unwrap();
             let port_handle = comp_ctx.get_port_handle(port_to_block);
             let port_info = comp_ctx.get_port_mut(port_handle);
             debug_assert_eq!(port_info.kind, PortKind::Putter);
             port_info.state.set(PortStateFlag::BlockedDueToFullBuffers);
         },
         ControlMessageContent::ClosePort(content) => {
             // Request to close the port. We immediately comply and remove
             // the component handle as well
             let port_to_close = message.target_port_id.unwrap();
             let port_handle = comp_ctx.get_port_handle(port_to_close);
             // We're closing the port, so we will always update the peer of the
             // port (in case of error messages)
             let port_info = comp_ctx.get_port_mut(port_handle);
             port_info.peer_comp_id = message.sender_comp_id;
             port_info.close_at_sync_end = true; // might be redundant (we might set it closed now)
             let peer_comp_id = port_info.peer_comp_id;
             let peer_handle = comp_ctx.get_peer_handle(peer_comp_id);
             // One exception to sending an `Ack` is if we just closed the
             // port ourselves, meaning that the `ClosePort` messages got
             // sent to one another.
             if let Some(control_id) = control.has_close_port_entry(port_handle, comp_ctx) {
                 // The two components (sender and this component) are closing
                 // the channel at the same time. So we don't care about the
                 // content of the `ClosePort` message.
                 default_handle_ack(control, control_id, sched_ctx, comp_ctx);
+                default_handle_ack(exec_state, control, control_id, sched_ctx, comp_ctx, consensus, inbox_main, inbox_backup)?;
             } else {
                 // Respond to the message
                 let port_info = comp_ctx.get_port(port_handle);
                 let last_instruction = port_info.last_instruction;
                 let port_has_had_message = port_info.received_message_for_sync;
                 default_send_ack(message.id, peer_handle, sched_ctx, comp_ctx);
                 comp_ctx.change_port_peer(sched_ctx, port_handle, None);
                 // Handle any possible error conditions (which boil down to: the
                 // port has been used, but the peer has died). If not in sync
                 // mode then we close the port immediately.
                 // Note that `port_was_used` does not mean that any messages
                 // were actually received. It might also mean that e.g. the
                 // component attempted a `get`, but there were no messages, so
                 // now it is in the `BlockedGet` state.
                 let port_was_used = last_instruction != PortInstruction::None;
                 if exec_state.mode.is_in_sync_block() {
                     let closed_during_sync_round = content.closed_in_sync_round && port_was_used;
                     let closed_before_sync_round = !content.closed_in_sync_round && !port_has_had_message;
+                    let closed_before_sync_round = !content.closed_in_sync_round && !port_has_had_message && port_was_used;
                     if closed_during_sync_round || closed_before_sync_round {
                         return Err((
                             last_instruction,
                             format!("Peer component (id:{}) shut down, so communication cannot (have) succeed(ed)", peer_comp_id.0)
                         ));
+                    }
                 } else {
                     let port_info = comp_ctx.get_port_mut(port_handle);
                     port_info.state.set(PortStateFlag::Closed);
+                }
+            }
         },
         ControlMessageContent::UnblockPort => {
             // We were previously blocked (or already closed)
             let port_to_unblock = message.target_port_id.unwrap();
             let port_handle = comp_ctx.get_port_handle(port_to_unblock);
             let port_info = comp_ctx.get_port_mut(port_handle);
             debug_assert_eq!(port_info.kind, PortKind::Putter);
             debug_assert!(port_info.state.is_set(PortStateFlag::BlockedDueToFullBuffers));
             port_info.state.clear(PortStateFlag::BlockedDueToFullBuffers);
             default_handle_recently_unblocked_port(exec_state, consensus, port_handle, sched_ctx, comp_ctx);
             default_handle_recently_unblocked_port(
                 exec_state, control, consensus, port_handle, sched_ctx,
                 comp_ctx, inbox_main, inbox_backup
             )?;
         },
         ControlMessageContent::PortPeerChangedBlock => {
             // The peer of our port has just changed. So we are asked to
             // temporarily block the port (while our original recipient is
             // potentially rerouting some of the in-flight messages) and
             // Ack. Then we wait for the `unblock` call.
             let port_to_change = message.target_port_id.unwrap();
             let port_handle = comp_ctx.get_port_handle(port_to_change);
             let port_info = comp_ctx.get_port_mut(port_handle);
             let peer_comp_id = port_info.peer_comp_id;
             port_info.state.set(PortStateFlag::BlockedDueToPeerChange);
             let peer_handle = comp_ctx.get_peer_handle(peer_comp_id);
             default_send_ack(message.id, peer_handle, sched_ctx, comp_ctx);
         },
         ControlMessageContent::PortPeerChangedUnblock(new_port_id, new_comp_id) => {
             let port_to_change = message.target_port_id.unwrap();
             let port_handle = comp_ctx.get_port_handle(port_to_change);
             let port_info = comp_ctx.get_port(port_handle);
             debug_assert!(port_info.state.is_set(PortStateFlag::BlockedDueToPeerChange));
             let old_peer_id = port_info.peer_comp_id;
             let port_info = comp_ctx.get_port_mut(port_handle);
             port_info.peer_port_id = new_port_id;
             port_info.state.clear(PortStateFlag::BlockedDueToPeerChange);
             comp_ctx.change_port_peer(sched_ctx, port_handle, Some(new_comp_id));
             default_handle_recently_unblocked_port(exec_state, consensus, port_handle, sched_ctx, comp_ctx);
             default_handle_recently_unblocked_port(
                 exec_state, control, consensus, port_handle, sched_ctx,
                 comp_ctx, inbox_main, inbox_backup
             )?;
+        }
+    }
     return Ok(());
+}
 /// Handles a component entering the synchronous block. Will ensure that the
 /// `Consensus` and the `ComponentCtx` are initialized properly.
 pub(crate) fn default_handle_sync_start(
     exec_state: &mut CompExecState, inbox_main: &mut InboxMainRef,
     sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, consensus: &mut Consensus
 ) {
     sched_ctx.info("Component starting sync mode");
     // If any messages are present for this sync round, set the appropriate flag
     // and notify the consensus handler of the present messages
     consensus.notify_sync_start(comp_ctx);
     for (port_index, message) in inbox_main.iter().enumerate() {
         if let Some(message) = message {
             consensus.handle_incoming_data_message(comp_ctx, message);
             let port_info = comp_ctx.get_port_by_index_mut(port_index);
             port_info.received_message_for_sync = true;
+        }
+    }
     // Modify execution state
     debug_assert_eq!(exec_state.mode, CompMode::NonSync);
     exec_state.mode = CompMode::Sync;
+}
 /// Handles a component that has reached the end of the sync block. This does
 /// not necessarily mean that the component will go into the `NonSync` mode, as
 /// it might have to wait for the leader to finish the round for everyone (see
 /// `default_handle_sync_decision`)
 pub(crate) fn default_handle_sync_end(
     exec_state: &mut CompExecState, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx,
     consensus: &mut Consensus
 ) {
     sched_ctx.info("Component ending sync mode (but possibly waiting for a solution)");
     debug_assert_eq!(exec_state.mode, CompMode::Sync);
     let decision = consensus.notify_sync_end_success(sched_ctx, comp_ctx);
     exec_state.mode = CompMode::SyncEnd;
     default_handle_sync_decision(sched_ctx, exec_state, comp_ctx, decision, consensus);
+}
 /// Handles a component initiating the exiting procedure, and closing all of its
 /// ports. Should only be called once per component (which is ensured by
 /// checking and modifying the mode in the execution state).
 #[must_use]
 pub(crate) fn default_handle_start_exit(
     exec_state: &mut CompExecState, control: &mut ControlLayer,
     sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, consensus: &mut Consensus
 ) -> CompScheduling {
     debug_assert_eq!(exec_state.mode, CompMode::StartExit);
     for port_index in 0..comp_ctx.num_ports() {
         let port_info = comp_ctx.get_port_by_index_mut(port_index);
         if port_info.state.is_blocked() {
             return CompScheduling::Sleep;
+        }
+    }
     sched_ctx.info(&format!("Component starting exit (reason: {:?})", exec_state.exit_reason));
     exec_state.mode = CompMode::BusyExit;
     let exit_inside_sync = exec_state.exit_reason.is_in_sync();
     // If exiting while inside sync mode, report to the leader of the current
     // round that we've failed.
     if exit_inside_sync {
         let decision = consensus.notify_sync_end_failure(sched_ctx, comp_ctx);
         default_handle_sync_decision(sched_ctx, exec_state, comp_ctx, decision, consensus);
+    }
     // Iterating over ports by index to work around borrowing rules
     for port_index in 0..comp_ctx.num_ports() {
         let port = comp_ctx.get_port_by_index_mut(port_index);
         if port.state.is_closed() || port.close_at_sync_end {
         println!("DEBUG: Considering port:\n{:?}", port);
         if port.state.is_closed() || port.state.is_set(PortStateFlag::Transmitted) || port.close_at_sync_end {
             // Already closed, or in the process of being closed
             continue;
+        }
         // Mark as closed
         let port_id = port.self_id;
         port.state.set(PortStateFlag::Closed);
         // Notify peer of closing
         let port_handle = comp_ctx.get_port_handle(port_id);
         let (peer, message) = control.initiate_port_closing(port_handle, exit_inside_sync, comp_ctx);
         let peer_info = comp_ctx.get_peer(peer);
         peer_info.handle.send_message_logged(sched_ctx, Message::Control(message), true);
+    }
     return CompScheduling::Immediate; // to check if we can shut down immediately
+}
 /// Handles a component waiting until all peers are notified that it is quitting
 /// (i.e. after calling `default_handle_start_exit`).
 #[must_use]
 pub(crate) fn default_handle_busy_exit(
     exec_state: &mut CompExecState, control: &ControlLayer,
     sched_ctx: &SchedulerCtx
 ) -> CompScheduling {
     debug_assert_eq!(exec_state.mode, CompMode::BusyExit);
     if control.has_acks_remaining() {
         sched_ctx.info("Component busy exiting, still has `Ack`s remaining");
         return CompScheduling::Sleep;
     } else {
         sched_ctx.info("Component busy exiting, now shutting down");
         exec_state.mode = CompMode::Exit;
         return CompScheduling::Exit;
+    }
+}
 /// Handles a potential synchronous round decision. If there was a decision then
 /// the `Some(success)` value indicates whether the round succeeded or not.
 /// Might also end up changing the `ExecState`.
 ///
 /// Might be called in two cases:
 /// 1. The component is in regular execution mode, at the end of a sync round,
 ///     and is waiting for a solution to the round.
 /// 2. The component has encountered an error during a sync round and is
 ///     exiting, hence is waiting for a "Failure" message from the leader.
 pub(crate) fn default_handle_sync_decision(
     sched_ctx: &SchedulerCtx, exec_state: &mut CompExecState, comp_ctx: &mut CompCtx,
     decision: SyncRoundDecision, consensus: &mut Consensus
 ) -> Option<bool> {
     let success = match decision {
         SyncRoundDecision::None => return None,
         SyncRoundDecision::Solution => true,
         SyncRoundDecision::Failure => false,
     };
     debug_assert!(
         exec_state.mode == CompMode::SyncEnd || (
             exec_state.mode.is_busy_exiting() && exec_state.exit_reason.is_error()
         ) || (
             exec_state.mode.is_in_sync_block() && decision == SyncRoundDecision::Failure
+        )
     );
     sched_ctx.info(&format!("Handling decision {:?} (in mode: {:?})", decision, exec_state.mode));
     consensus.notify_sync_decision(decision);
     if success {
         // We cannot get a success message if the component has encountered an
         // error.
         for port_index in 0..comp_ctx.num_ports() {
             let port_info = comp_ctx.get_port_by_index_mut(port_index);
             if port_info.close_at_sync_end {
                 port_info.state.set(PortStateFlag::Closed);
+            }
             port_info.state.clear(PortStateFlag::Received);
+        }
         debug_assert_eq!(exec_state.mode, CompMode::SyncEnd);
         exec_state.mode = CompMode::NonSync;
         return Some(true);
     } else {
         // We may get failure both in all possible cases. But we should only
         // modify the execution state if we're not already in exit mode
         if !exec_state.mode.is_busy_exiting() {
             sched_ctx.error("failed synchronous round, initiating exit");
             exec_state.set_as_start_exit(ExitReason::ErrorNonSync);
+        }
         return Some(false);
+    }
+}
 pub(crate) fn default_start_create_component(
     exec_state: &mut CompExecState, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx,
     control: &mut ControlLayer, inbox_main: &mut InboxMain, inbox_backup: &mut InboxBackup,
     definition_id: ProcedureDefinitionId, type_id: TypeId, arguments: ValueGroup
 ) {
     debug_assert_eq!(exec_state.mode, CompMode::NonSync);
     let mut transferred_ports = Vec::new();
     find_ports_in_value_group(&arguments, &mut transferred_ports);
     // Set execution state as waiting until we can create the component. If we
     // can do so right away, then we will.
     exec_state.set_as_create_component_blocked(definition_id, type_id, arguments);
     if ports_not_blocked(comp_ctx, &transferred_ports) {
         perform_create_component(exec_state, sched_ctx, comp_ctx, control, inbox_main, inbox_backup);
+    }
+}
 /// Actually creates a component (and assumes that the caller made sure that
 /// none of the ports are involved in a blocking operation).
 pub(crate) fn perform_create_component(
     exec_state: &mut CompExecState, sched_ctx: &SchedulerCtx, instantiator_ctx: &mut CompCtx,
     control: &mut ControlLayer, inbox_main: &mut InboxMain, inbox_backup: &mut InboxBackup
 ) {
     // Small internal utilities
     struct PortPair {
         instantiator_id: PortId,
         instantiator_handle: LocalPortHandle,
         created_id: PortId,
         created_handle: LocalPortHandle,
         is_open: bool,
+    }
     // Retrieve ports from the arguments
     debug_assert_eq!(exec_state.mode, CompMode::NewComponentBlocked);
     let (procedure_id, procedure_type_id) = exec_state.mode_component;
     let mut arguments = exec_state.mode_value.take();
     let mut ports = Vec::new();
     find_ports_in_value_group(&arguments, &mut ports);
     debug_assert!(ports_not_blocked(instantiator_ctx, &ports));
     // Reserve a location for the new component
     let reservation = sched_ctx.runtime.start_create_component();
     let mut created_ctx = CompCtx::new(&reservation);
     let mut port_pairs = Vec::with_capacity(ports.len());
     // Go over all the ports that will be transferred. Since the ports will get
     // a new ID in the new component, we will take care of that here.
     for (port_location, instantiator_port_id) in &ports {
         // Retrieve port information from instantiator
         let instantiator_port_id = *instantiator_port_id;
         let instantiator_port_handle = instantiator_ctx.get_port_handle(instantiator_port_id);
         let instantiator_port = instantiator_ctx.get_port(instantiator_port_handle);
         // Create port at created component
         let created_port_handle = created_ctx.add_port(
             instantiator_port.peer_comp_id, instantiator_port.peer_port_id,
             instantiator_port.kind, instantiator_port.state
         );
         let created_port = created_ctx.get_port(created_port_handle);
         let created_port_id = created_port.self_id;
         // Modify port ID in the arguments to the new component and store them
         // for later access
         let is_open = instantiator_port.state.is_open();
         port_pairs.push(PortPair{
             instantiator_id: instantiator_port_id,
             instantiator_handle: instantiator_port_handle,
             created_id: created_port_id,
             created_handle: created_port_handle,
             is_open,
         });
         for location in port_location.iter().copied() {
             let value = arguments.get_value_mut(location);
             match value {
                 Value::Input(id) => *id = port_id_to_eval(created_port_id),
                 Value::Output(id) => *id = port_id_to_eval(created_port_id),
                 _ => unreachable!(),
+            }
+        }
+    }
     // For each of the ports in the newly created component we set the peer to
     // the correct value. We will not yet change the peer on the instantiator's
     // ports (as we haven't yet stored the new component in the runtime's
     // component storage)
     let mut created_component_has_remote_peers = false;
     for pair in port_pairs.iter() {
         let instantiator_port_info = instantiator_ctx.get_port(pair.instantiator_handle);
         let created_port_info = created_ctx.get_port_mut(pair.created_handle);
         if created_port_info.peer_comp_id == instantiator_ctx.id {
             // The peer of the created component's port seems to be the
             // instantiator.
             let created_port_peer_index = port_pairs.iter()
                 .position(|v| v.instantiator_id == instantiator_port_info.peer_port_id);
             match created_port_peer_index {
                 Some(created_port_peer_index) => {
                     // However, the peer port is also moved to the new
                     // component, so the complete channel is owned by the new
                     // component.
                     let peer_pair = &port_pairs[created_port_peer_index];
                     created_port_info.peer_port_id = peer_pair.created_id;
                     created_port_info.peer_comp_id = reservation.id();
                 },
                 None => {
                     // Peer port remains with instantiator. However, we cannot
                     // set the peer on the instantiator yet, because the new
                     // component has not yet been stored in the runtime's
                     // component storage. So we do this later
                     created_port_info.peer_comp_id = instantiator_ctx.id;
                     if pair.is_open {
                         created_ctx.change_port_peer(sched_ctx, pair.created_handle, Some(instantiator_ctx.id));
+                    }
+                }
+            }
         } else {
             // Peer is a different component
             if pair.is_open {
                 // And the port is still open, so we need to notify the peer
                 let peer_handle = instantiator_ctx.get_peer_handle(created_port_info.peer_comp_id);
                 let peer_info = instantiator_ctx.get_peer(peer_handle);
                 created_ctx.change_port_peer(sched_ctx, pair.created_handle, Some(peer_info.id));
                 created_component_has_remote_peers = true;
+            }
+        }
+    }
     // Now we store the new component into the runtime's component storage using
     // the reservation.
     let component = create_component(
         &sched_ctx.runtime.protocol, procedure_id, procedure_type_id,
         arguments, port_pairs.len()
     );
     let (created_key, created_runtime_component) = sched_ctx.runtime.finish_create_component(
         reservation, component, created_ctx, false
     );
     let created_ctx = &mut created_runtime_component.ctx;
     let created_component = &mut created_runtime_component.component;
     created_component.on_creation(created_key.downgrade(), sched_ctx);
     // We now pass along the messages that the instantiator component still has
     // that belong to the new component. At the same time we'll take care of
     // setting the correct peer of the instantiator component
     for pair in port_pairs.iter() {
         // Transferring the messages and removing the port from the
         // instantiator component
         let instantiator_port_index = instantiator_ctx.get_port_index(pair.instantiator_handle);
         instantiator_ctx.change_port_peer(sched_ctx, pair.instantiator_handle, None);
         instantiator_ctx.remove_port(pair.instantiator_handle);
         if let Some(mut message) = inbox_main[instantiator_port_index].take() {
             message.data_header.target_port = pair.created_id;
             created_component.adopt_message(created_ctx, message);
+        }
         let mut message_index = 0;
         while message_index < inbox_backup.len() {
             let message = &inbox_backup[message_index];
             if message.data_header.target_port == pair.instantiator_id {
                 // Transfer the message
                 let mut message = inbox_backup.remove(message_index);
                 message.data_header.target_port = pair.created_id;
                 created_component.adopt_message(created_ctx, message);
             } else {
                 // Message does not belong to the port pair that we're
                 // transferring to the new component.
                 message_index += 1;
+            }
+        }
         // Here we take care of the case where the instantiator previously owned
         // both ends of the channel, but has transferred one port to the new
         // component (hence creating a channel between the instantiator
         // component and the new component).
         let created_port_info = created_ctx.get_port(pair.created_handle);
         if pair.is_open && created_port_info.peer_comp_id == instantiator_ctx.id {
             // Note: the port we're receiving here belongs to the instantiator
             // and is NOT in the "port_pairs" array.
             let instantiator_port_handle = instantiator_ctx.get_port_handle(created_port_info.peer_port_id);
             let instantiator_port_info = instantiator_ctx.get_port_mut(instantiator_port_handle);
             instantiator_port_info.peer_port_id = created_port_info.self_id;
             instantiator_ctx.change_port_peer(sched_ctx, instantiator_port_handle, Some(created_ctx.id));
+        }
+    }
     // Finally: if we did move ports around whose peers are different
     // components, then we'll initiate the appropriate protocol to notify them.
     if created_component_has_remote_peers {
         let schedule_entry_id = control.add_schedule_entry(created_ctx.id);
         for pair in &port_pairs {
             let port_info = created_ctx.get_port(pair.created_handle);
             if pair.is_open && port_info.peer_comp_id != instantiator_ctx.id && port_info.peer_comp_id != created_ctx.id {
                 // Peer component is not the instantiator, and it is not the
                 // new component itself
                 let message = control.add_reroute_entry(
                     instantiator_ctx.id, port_info.peer_port_id, port_info.peer_comp_id,
                     pair.instantiator_id, pair.created_id, created_ctx.id,
                     schedule_entry_id
                 );
                 let peer_handle = created_ctx.get_peer_handle(port_info.peer_comp_id);
                 let peer_info = created_ctx.get_peer(peer_handle);
                 peer_info.handle.send_message_logged(sched_ctx, message, true);
+            }
+        }
     } else {
         // We can schedule the component immediately, we do not have to wait
         // for any peers: there are none.
         sched_ctx.runtime.enqueue_work(created_key);
+    }
     exec_state.mode = CompMode::NonSync;
     exec_state.mode_component = (ProcedureDefinitionId::new_invalid(), TypeId::new_invalid());
+}
 pub(crate) fn ports_not_blocked(comp_ctx: &CompCtx, ports: &EncounteredPorts) -> bool {
     for (_port_locations, port_id) in ports {
         let port_handle = comp_ctx.get_port_handle(*port_id);
         let port_info = comp_ctx.get_port(port_handle);
         if port_info.state.is_blocked_due_to_port_change() {
             return false;
+        }
+    }
     return true;
+}
 /// Performs the default action of printing the provided error, and then putting
 /// the component in the state where it will shut down. Only to be used for
 /// builtin components: their error message construction is simpler (and more
 /// common) as they don't have any source code.
 pub(crate) fn default_handle_error_for_builtin(
     exec_state: &mut CompExecState, sched_ctx: &SchedulerCtx,
     location_and_message: (PortInstruction, String)
 ) {
     let (_location, message) = location_and_message;
     sched_ctx.error(&message);
     let exit_reason = if exec_state.mode.is_in_sync_block() {
         ExitReason::ErrorInSync
     } else {
         ExitReason::ErrorNonSync
     };
     exec_state.set_as_start_exit(exit_reason);
+}
 #[inline]
 pub(crate) fn default_handle_exit(_exec_state: &CompExecState) -> CompScheduling {
     debug_assert_eq!(_exec_state.mode, CompMode::Exit);
     return CompScheduling::Exit;
+}
 // -----------------------------------------------------------------------------
 // Internal messaging/state utilities
 // -----------------------------------------------------------------------------
 /// Sends a message without any transmitted ports. Does not check if sending
 /// is actually valid.
 fn send_message_without_ports(
     sending_port_handle: LocalPortHandle, value: ValueGroup,
     comp_ctx: &CompCtx, sched_ctx: &SchedulerCtx, consensus: &mut Consensus,
 ) {
     let port_info = comp_ctx.get_port(sending_port_handle);
     debug_assert!(port_info.state.can_send());
     let peer_handle = comp_ctx.get_peer_handle(port_info.peer_comp_id);
     let peer_info = comp_ctx.get_peer(peer_handle);
     let annotated_message = consensus.annotate_data_message(comp_ctx, port_info, value);
     peer_info.handle.send_message_logged(sched_ctx, Message::Data(annotated_message), true);
+}
 /// Prepares sending a message that contains ports. Only once a particular
 /// protocol has completed (where we notify all the peers that the ports will
 /// be transferred) will we actually send the message to the recipient.
 fn start_send_message_with_ports(
     sending_port_id: PortId, sending_port_instruction: PortInstruction, value: ValueGroup,
     exec_state: &mut CompExecState, comp_ctx: &mut CompCtx, sched_ctx: &SchedulerCtx,
     control: &mut ControlLayer
 ) -> Result<(), (PortInstruction, String)> {
     debug_assert_eq!(exec_state.mode, CompMode::Sync); // busy in sync, trying to send
     // Retrieve ports we're going to transfer
     let sending_port_handle = comp_ctx.get_port_handle(sending_port_id);
     let sending_port_info = comp_ctx.get_port_mut(sending_port_handle);
     sending_port_info.last_instruction = sending_port_instruction;
     let mut transmit_ports = Vec::new();
     find_ports_in_value_group(&value, &mut transmit_ports);
     debug_assert!(!transmit_ports.is_empty()); // required from caller
     // Enter the state where we'll wait until all transferred ports are not
     // blocked.
     exec_state.set_as_blocked_put_with_ports(sending_port_id, value);
     if ports_not_blocked(comp_ctx, &transmit_ports) {
         // Ports are not blocked, so we can send them right away.
         perform_send_message_with_ports_notify_peers(
             exec_state, comp_ctx, sched_ctx, control, transmit_ports
         )?;
     } // else: wait until they become unblocked
     return Ok(())
+}
 fn perform_send_message_with_ports_notify_peers(
     exec_state: &mut CompExecState, comp_ctx: &mut CompCtx, sched_ctx: &SchedulerCtx,
     control: &mut ControlLayer, transmit_ports: EncounteredPorts
 ) -> Result<(), (PortInstruction, String)> {
     // Check we're in the correct state in debug mode
     debug_assert_eq!(exec_state.mode, CompMode::PutPortsBlockedTransferredPorts);
     debug_assert!(ports_not_blocked(comp_ctx, &transmit_ports));
     // Set up the final Ack that triggers us to send our final message
     let unblock_put_control_id = control.add_unblock_put_with_ports_entry();
     for (_, port_id) in &transmit_ports {
         let transmit_port_handle = comp_ctx.get_port_handle(*port_id);
         let transmit_port_info = comp_ctx.get_port_mut(transmit_port_handle);
         let peer_comp_id = transmit_port_info.peer_comp_id;
         let peer_port_id = transmit_port_info.peer_port_id;
         // Note: we checked earlier that we are currently in sync mode. Now we
         // will check if we've already used the port we're about to transmit.
         if !transmit_port_info.last_instruction.is_none() {
             let sending_port_handle = comp_ctx.get_port_handle(exec_state.mode_port);
             let sending_port_instruction = comp_ctx.get_port(sending_port_handle).last_instruction;
             return Err((
                 sending_port_instruction,
                 String::from("Cannot transmit one of the ports in this message, as it is used in this sync round")
             ));
+        }
         if transmit_port_info.state.is_set(PortStateFlag::Transmitted) {
             let sending_port_handle = comp_ctx.get_port_handle(exec_state.mode_port);
             let sending_port_instruction = comp_ctx.get_port(sending_port_handle).last_instruction;
             return Err((
                 sending_port_instruction,
                 String::from("Cannot transmit one of the ports in this message, as that port is already transmitted")
             ));
+        }
         // Set the flag for transmission
         transmit_port_info.state.set(PortStateFlag::Transmitted);
         // Block the peer of the port
         let message = control.create_port_transfer_message(unblock_put_control_id, comp_ctx.id, peer_port_id);
         println!("DEBUG: Port transfer message\nControl ID: {:?}\nMessage: {:?}", unblock_put_control_id, message);
         let peer_handle = comp_ctx.get_peer_handle(peer_comp_id);
         let peer_info = comp_ctx.get_peer(peer_handle);
         peer_info.handle.send_message_logged(sched_ctx, message, true);
+    }
     // We've set up the protocol, once all the PPC's are blocked we are supposed
     // to transfer the message to the recipient. So store it temporarily
     exec_state.mode = CompMode::PutPortsBlockedAwaitingAcks;
     return Ok(());
+}
 /// Performs the transmission of a data message that contains ports. These were
 /// all stored in the component's execution state by the
 /// `prepare_send_message_with_ports` function. Port must be ready to send!
 fn perform_send_message_with_ports_to_receiver(
     exec_state: &mut CompExecState, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, consensus: &mut Consensus,
     inbox_main: &mut InboxMain, inbox_backup: &mut InboxBackup
 ) -> Result<(), (PortInstruction, String)> {
     debug_assert_eq!(exec_state.mode, CompMode::PutPortsBlockedSendingPort);
     // Find all ports again
     let mut transmit_ports = Vec::new();
     find_ports_in_value_group(&exec_state.mode_value, &mut transmit_ports);
     // Retrieve the port over which we're going to send the message
     let port_handle = comp_ctx.get_port_handle(exec_state.mode_port);
     let port_info = comp_ctx.get_port(port_handle);
     if !port_info.state.is_open() {
         return Err((
             port_info.last_instruction,
             String::from("cannot send over this port, as it is closed")
         ));
+    }
     debug_assert!(!port_info.state.is_blocked_due_to_port_change()); // caller should have checked this
     let peer_handle = comp_ctx.get_peer_handle(port_info.peer_comp_id);
     // Change state back to its default
     exec_state.mode = CompMode::Sync;
     let message_value = exec_state.mode_value.take();
     exec_state.mode_port = PortId::new_invalid();
     // Annotate the data message
     let mut annotated_message = consensus.annotate_data_message(comp_ctx, port_info, message_value);
     // And further enhance the message by adding data about the ports that are
     // being transferred
     for (port_locations, transmit_port_id) in transmit_ports {
         let transmit_port_handle = comp_ctx.get_port_handle(transmit_port_id);
         let transmit_port_info = comp_ctx.get_port(transmit_port_handle);
         let transmit_messages = take_port_messages(comp_ctx, transmit_port_id, inbox_main, inbox_backup);
         let mut transmit_port_state = transmit_port_info.state;
         debug_assert!(transmit_port_state.is_set(PortStateFlag::Transmitted));
         transmit_port_state.clear(PortStateFlag::Transmitted);
         annotated_message.ports.push(TransmittedPort{
             locations: port_locations,
             messages: transmit_messages,
             peer_comp: transmit_port_info.peer_comp_id,
             peer_port: transmit_port_info.peer_port_id,
             kind: transmit_port_info.kind,
             state: transmit_port_state
         });
         comp_ctx.change_port_peer(sched_ctx, transmit_port_handle, None);
+    }
     // And finally, send the message to the peer
     let peer_info = comp_ctx.get_peer(peer_handle);
     peer_info.handle.send_message_logged(sched_ctx, Message::Data(annotated_message), true);
     return Ok(());
+}
 /// Handles an `Ack` for the control layer.
 fn default_handle_ack(
     control: &mut ControlLayer, control_id: ControlId,
     sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx
 ) {
     exec_state: &mut CompExecState, control: &mut ControlLayer, control_id: ControlId,
     sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, consensus: &mut Consensus,
     inbox_main: &mut InboxMain, inbox_backup: &mut InboxBackup
 ) -> Result<(), (PortInstruction, String)>{
     // Since an `Ack` may cause another one, handle them in a loop
     let mut to_ack = control_id;
     loop {
         let (action, new_to_ack) = control.handle_ack(to_ack, sched_ctx, comp_ctx);
         match action {
             AckAction::SendMessage(target_comp, message) => {
                 // FIX @NoDirectHandle
                 let mut handle = sched_ctx.runtime.get_component_public(target_comp);
                 handle.send_message_logged(sched_ctx, Message::Control(message), true);
                 let _should_remove = handle.decrement_users();
                 debug_assert!(_should_remove.is_none());
             },
             AckAction::ScheduleComponent(to_schedule) => {
                 // FIX @NoDirectHandle
                 let mut handle = sched_ctx.runtime.get_component_public(to_schedule);
                 // Note that the component is intentionally not
                 // sleeping, so we just wake it up
                 debug_assert!(!handle.sleeping.load(std::sync::atomic::Ordering::Acquire));
                 let key = unsafe { to_schedule.upgrade() };
                 sched_ctx.runtime.enqueue_work(key);
                 let _should_remove = handle.decrement_users();
                 debug_assert!(_should_remove.is_none());
             },
             AckAction::UnblockPutWithPorts => {
                 // Send the message (containing ports) stored in the component
                 // execution state to the recipient
                 println!("DEBUG: Unblocking put with ports");
                 debug_assert_eq!(exec_state.mode, CompMode::PutPortsBlockedAwaitingAcks);
                 exec_state.mode = CompMode::PutPortsBlockedSendingPort;
                 let port_handle = comp_ctx.get_port_handle(exec_state.mode_port);
                 // Little bit of a hack, we didn't really unblock the sending
                 // port, but this will mesh nicely with waiting for the sending
                 // port to become unblocked.
                 default_handle_recently_unblocked_port(
                     exec_state, control, consensus, port_handle, sched_ctx,
                     comp_ctx, inbox_main, inbox_backup
                 )?;
             },
             AckAction::None => {}
+        }
         match new_to_ack {
             Some(new_to_ack) => to_ack = new_to_ack,
             None => break,
+        }
+    }
     return Ok(());
+}
 /// Little helper for sending the most common kind of `Ack`
 fn default_send_ack(
     causer_of_ack_id: ControlId, peer_handle: LocalPeerHandle,
     sched_ctx: &SchedulerCtx, comp_ctx: &CompCtx
 ) {
     let peer_info = comp_ctx.get_peer(peer_handle);
     peer_info.handle.send_message_logged(sched_ctx, Message::Control(ControlMessage{
         id: causer_of_ack_id,
         sender_comp_id: comp_ctx.id,
         target_port_id: None,
         content: ControlMessageContent::Ack
     }), true);
+}
 /// Handles the unblocking of a putter port. In case there is a pending message
 /// on that port then it will be sent.
 /// on that port then it will be sent. There are two reasons for calling this
 /// function: either a port was blocked (i.e. the Blocked state flag was
 /// cleared), or the component is ready to send a message containing ports
 /// (stored in the execution state). In this latter case we might still have
 /// a blocked port.
 fn default_handle_recently_unblocked_port(
     exec_state: &mut CompExecState, consensus: &mut Consensus,
+    exec_state: &mut CompExecState, control: &mut ControlLayer, consensus: &mut Consensus,
     port_handle: LocalPortHandle, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx,
 ) {
     inbox_main: &mut InboxMain, inbox_backup: &mut InboxBackup
 ) -> Result<(), (PortInstruction, String)> {
     let port_info = comp_ctx.get_port_mut(port_handle);
     let port_id = port_info.self_id;
     debug_assert!(!port_info.state.is_blocked()); // should have been done by the caller
     if exec_state.is_blocked_on_put(port_id) {
     if port_info.state.is_blocked() {
         // Port is still blocked. We wait until the next control message where
         // we unblock the port.
         return Ok(());
+    }
     if exec_state.is_blocked_on_put_without_ports(port_id) {
         // Annotate the message that we're going to send
         let port_info = comp_ctx.get_port(port_handle); // for immutable access
         debug_assert_eq!(port_info.kind, PortKind::Putter);
         let to_send = exec_state.mode_value.take();
         let to_send = consensus.annotate_data_message(comp_ctx, port_info, to_send);
         // Retrieve peer to send the message
         let peer_handle = comp_ctx.get_peer_handle(port_info.peer_comp_id);
         let peer_info = comp_ctx.get_peer(peer_handle);
         peer_info.handle.send_message_logged(sched_ctx, Message::Data(to_send), true);
         exec_state.mode = CompMode::Sync; // because we're blocked on a `put`, we must've started in the sync state.
         // Return to the regular execution mode
         exec_state.mode = CompMode::Sync;
         exec_state.mode_port = PortId::new_invalid();
     } else if exec_state.mode == CompMode::PutPortsBlockedTransferredPorts {
         // We are waiting until all of the transferred ports become unblocked,
         // check so here.
         let mut transfer_ports = Vec::new();
         find_ports_in_value_group(&exec_state.mode_value, &mut transfer_ports);
         if ports_not_blocked(comp_ctx, &transfer_ports) {
             perform_send_message_with_ports_notify_peers(
                 exec_state, comp_ctx, sched_ctx, control, transfer_ports
             )?;
+        }
     } else if exec_state.mode == CompMode::PutPortsBlockedSendingPort && exec_state.mode_port == port_id {
         // We checked above that the port became unblocked, so we can send the
         // message
         perform_send_message_with_ports_to_receiver(
             exec_state, sched_ctx, comp_ctx, consensus, inbox_main, inbox_backup
         )?;
     } else if exec_state.is_blocked_on_create_component() {
         let mut ports = Vec::new();
         find_ports_in_value_group(&exec_state.mode_value, &mut ports);
         if ports_not_blocked(comp_ctx, &ports) {
             perform_create_component(
                 exec_state, sched_ctx, comp_ctx, control, inbox_main, inbox_backup
             );
+        }
+    }
     return Ok(());
+}
 #[inline]
 pub(crate) fn port_id_from_eval(port_id: EvalPortId) -> PortId {
     return PortId(port_id.id);
+}
 #[inline]
 pub(crate) fn port_id_to_eval(port_id: PortId) -> EvalPortId {
     return EvalPortId{ id: port_id.0 };
+}
 // TODO: Optimize double vec
 type EncounteredPorts = Vec<(Vec<ValueId>, PortId)>;
 /// Recursively goes through the value group, attempting to find ports.
 /// Duplicates will only be added once.
 pub(crate) fn find_ports_in_value_group(value_group: &ValueGroup, ports: &mut EncounteredPorts) {
     // Helper to check a value for a port and recurse if needed.
     fn find_port_in_value(group: &ValueGroup, value: &Value, value_location: ValueId, ports: &mut EncounteredPorts) {
         match value {
             Value::Input(port_id) | Value::Output(port_id) => {
                 // This is an actual port
                 let cur_port = PortId(port_id.id);
                 for prev_port in ports.iter_mut() {
                     if prev_port.1 == cur_port {
                         // Already added
                         prev_port.0.push(value_location);
                         return;
+                    }
+                }
                 ports.push((vec![value_location], cur_port));
             },
             Value::Array(heap_pos) |
             Value::Message(heap_pos) |
             Value::String(heap_pos) |
             Value::Struct(heap_pos) |
             Value::Union(_, heap_pos) => {
                 // Reference to some dynamic thing which might contain ports,
                 // so recurse
                 let heap_region = &group.regions[*heap_pos as usize];
                 for (value_index, embedded_value) in heap_region.iter().enumerate() {
                     let value_location = ValueId::Heap(*heap_pos, value_index as u32);
                     find_port_in_value(group, embedded_value, value_location, ports);
+                }
             },
             _ => {}, // values we don't care about
+        }
+    }
     // Clear the ports, then scan all the available values
     ports.clear();
     for (value_index, value) in value_group.values.iter().enumerate() {
         find_port_in_value(value_group, value, ValueId::Stack(value_index as u32), ports);
+    }
+}
 /// Goes through the inbox of a component and takes out all the messages that
 /// are targeted at a specific port
 pub(crate) fn take_port_messages(
     comp_ctx: &CompCtx, port_id: PortId,
     inbox_main: &mut InboxMain, inbox_backup: &mut InboxBackup
 ) -> Vec<DataMessage> {
     let mut messages = Vec::new();
     let port_handle = comp_ctx.get_port_handle(port_id);
     let port_index = comp_ctx.get_port_index(port_handle);
     if let Some(message) = inbox_main[port_index].take() {
         messages.push(message);
+    }
     let mut message_index = 0;
     while message_index < inbox_backup.len() {
         let message = &inbox_backup[message_index];
         if message.data_header.target_port == port_id {
             let message = inbox_backup.remove(message_index);
             messages.push(message);
         } else {
             message_index += 1;
+        }
+    }
     return messages;
+}
@@ \ No newline at end of file @@

src/runtime2/component/component_context.rs

➞

Show inline comments

 use std::fmt::{Debug, Formatter, Result as FmtResult};
 use crate::runtime2::scheduler::*;
 use crate::runtime2::runtime::*;
 use crate::runtime2::communication::*;
 use crate::protocol::ExpressionId;
 /// Helper struct to remember when the last operation on the port took place.
 #[derive(Debug, PartialEq, Copy, Clone)]
 pub enum PortInstruction {
     None,
     NoSource,
     SourceLocation(ExpressionId),
 }
 impl PortInstruction {
     pub fn is_none(&self) -> bool {
         match self {
             PortInstruction::None => return true,
             _ => return false,
+        }
+    }
+}
 /// Directionality of a port
 #[derive(Debug, PartialEq, Eq, Clone, Copy)]
 pub enum PortKind {
     Putter,
     Getter,
 }
 /// Bitflags for port
 // TODO: Incorporate remaining flags from `Port` struct
 #[repr(u32)]
 #[derive(Debug, Copy, Clone)]
 pub enum PortStateFlag {
     Closed = 0x01, // If not closed, then the port is open
     BlockedDueToPeerChange = 0x02, // busy changing peers, hence use of port is temporarily blocked
     BlockedDueToFullBuffers = 0x04,
     Transmitted = 0x08, // Transmitted, so cannot be used anymore
     Received = 0x10, // Received, so cannot be used yet, only after the sync round
 }
 #[derive(Copy, Clone)]
 pub struct PortState {
     flags: u32
 }
 impl PortState {
     pub(crate) fn new() -> PortState {
         return PortState{ flags: 0 }
     }
     // high-level
     #[inline]
     pub fn is_open(&self) -> bool {
         return !self.is_closed();
     }
     #[inline]
     pub fn can_send(&self) -> bool {
         return
             !self.is_set(PortStateFlag::Closed) &&
             !self.is_set(PortStateFlag::Transmitted) &&
             !self.is_set(PortStateFlag::Received);
+    }
     #[inline]
     pub fn is_closed(&self) -> bool {
         return self.is_set(PortStateFlag::Closed);
     }
     #[inline]
     pub fn is_blocked(&self) -> bool {
         return
             self.is_set(PortStateFlag::BlockedDueToPeerChange) ||
             self.is_set(PortStateFlag::BlockedDueToFullBuffers);
     }
     #[inline]
     pub fn is_blocked_due_to_port_change(&self) -> bool {
         return self.is_set(PortStateFlag::BlockedDueToPeerChange);
+    }
     // lower-level utils
     #[inline]
     pub fn set(&mut self, flag: PortStateFlag) {
         self.flags |= flag as u32;
     }
     #[inline]
     pub fn clear(&mut self, flag: PortStateFlag) {
         self.flags &= !(flag as u32);
     }
     #[inline]
     pub const fn is_set(&self, flag: PortStateFlag) -> bool {
         return (self.flags & (flag as u32)) != 0;
     }
 }
 impl Debug for PortState {
     fn fmt(&self, f: &mut Formatter<'_>) -> FmtResult {
         use PortStateFlag::*;
         let mut s = f.debug_struct("PortState");
         for (flag_name, flag_value) in &[
             ("closed", Closed),
             ("blocked_peer_change", BlockedDueToPeerChange),
             ("blocked_full_buffers", BlockedDueToFullBuffers)
             ("blocked_full_buffers", BlockedDueToFullBuffers),
             ("transmitted", Transmitted),
         ] {
             s.field(flag_name, &self.is_set(*flag_value));
         }
         return s.finish();
     }
 }
 #[derive(Debug)]
 pub struct Port {
     // Identifiers
     pub self_id: PortId,
     pub peer_comp_id: CompId, // eventually consistent
     pub peer_port_id: PortId, // eventually consistent
     // Generic operating state
     pub kind: PortKind,
     pub state: PortState,
     // State tracking for error detection and error handling
     pub last_instruction: PortInstruction, // used during sync round to detect port-closed-during-sync errors
     pub received_message_for_sync: bool, // used during sync round to detect port-closed-before-sync errors
     pub close_at_sync_end: bool, // set during sync round when receiving a port-closed-after-sync message
     pub(crate) associated_with_peer: bool,
 }
 pub struct Peer {
     pub id: CompId,
     pub num_associated_ports: u32,
     pub(crate) handle: CompHandle,
 }
 /// Port and peer management structure. Will keep a local reference counter to
 /// the ports associate with peers, additionally manages the atomic reference
 /// counter associated with the peers' component handles.
 pub struct CompCtx {
     pub id: CompId,
     ports: Vec<Port>,
     peers: Vec<Peer>,
     port_id_counter: u32,
 }
 #[derive(Copy, Clone, PartialEq, Eq)]
 pub struct LocalPortHandle(PortId);
 #[derive(Copy, Clone)]
 pub struct LocalPeerHandle(CompId);
 impl CompCtx {
     /// Creates a new component context based on a reserved entry in the
     /// component store. This reservation is used such that we already know our
     /// assigned ID.
     pub(crate) fn new(reservation: &CompReserved) -> Self {
         return Self{
             id: reservation.id(),
             ports: Vec::new(),
             peers: Vec::new(),
             port_id_counter: 0,
         }
     }
     /// Creates a new channel that is fully owned by the component associated
     /// with this context.
     pub(crate) fn create_channel(&mut self) -> Channel {
         let putter_id = PortId(self.take_port_id());
         let getter_id = PortId(self.take_port_id());
         self.ports.push(Port{
             self_id: putter_id,
             peer_port_id: getter_id,
             kind: PortKind::Putter,
             state: PortState::new(),
             peer_comp_id: self.id,
             last_instruction: PortInstruction::None,
             close_at_sync_end: false,
             received_message_for_sync: false,
             associated_with_peer: false,
         });
         self.ports.push(Port{
             self_id: getter_id,
             peer_port_id: putter_id,
             kind: PortKind::Getter,
             state: PortState::new(),
             peer_comp_id: self.id,
             last_instruction: PortInstruction::None,
             close_at_sync_end: false,
             received_message_for_sync: false,
             associated_with_peer: false,
         });
         return Channel{ putter_id, getter_id };
     }
-    /// Adds a new port. Make sure to call `add_peer` afterwards.
+    /// Adds a new port. Make sure to call `change_peer` afterwards.
     pub(crate) fn add_port(&mut self, peer_comp_id: CompId, peer_port_id: PortId, kind: PortKind, state: PortState) -> LocalPortHandle {
         let self_id = PortId(self.take_port_id());
         self.ports.push(Port{
             self_id, peer_comp_id, peer_port_id, kind, state,
             last_instruction: PortInstruction::None,
             close_at_sync_end: false,
             received_message_for_sync: false,
             associated_with_peer: false,
         });
         return LocalPortHandle(self_id);
     }
     /// Removes a port. Make sure you called `remove_peer` first.
     /// Adds a self-reference. Called by the runtime/scheduler
     pub(crate) fn add_self_reference(&mut self, self_handle: CompHandle) {
         debug_assert_eq!(self.id, self_handle.id());
         debug_assert!(self.get_peer_index_by_id(self.id).is_none());
         self.peers.push(Peer{
             id: self.id,
             num_associated_ports: 0,
             handle: self_handle
         });
+    }
     /// Removes a self-reference. Called by the runtime/scheduler
     pub(crate) fn remove_self_reference(&mut self) -> Option<CompKey> {
         let self_index = self.get_peer_index_by_id(self.id).unwrap();
         let peer = &mut self.peers[self_index];
         let maybe_comp_key = peer.handle.decrement_users();
         self.peers.remove(self_index);
         return maybe_comp_key;
+    }
     /// Removes a port. Make sure you called `change_peer` first.
     pub(crate) fn remove_port(&mut self, port_handle: LocalPortHandle) -> Port {
         let port_index = self.must_get_port_index(port_handle);
         let port = self.ports.remove(port_index);
         dbg_code!(assert!(!port.associated_with_peer));
         return port;
     }
     /// Changes a peer
     pub(crate) fn change_port_peer(&mut self, sched_ctx: &SchedulerCtx, port_handle: LocalPortHandle, new_peer_comp_id: Option<CompId>) {
         // If port is currently associated with a peer, then remove that peer
         let port_index = self.get_port_index(port_handle);
         let port = &mut self.ports[port_index];
         let port_is_closed = port.state.is_closed();
         if port.associated_with_peer {
             // Remove old peer association
             port.associated_with_peer = false;
             let peer_comp_id = port.peer_comp_id;
             let peer_index = self.get_peer_index_by_id(peer_comp_id).unwrap();
             let peer = &mut self.peers[peer_index];
             peer.num_associated_ports -= 1;
             if peer.num_associated_ports == 0 {
                 let mut peer = self.peers.remove(peer_index);
                 if let Some(key) = peer.handle.decrement_users() {
                     sched_ctx.runtime.destroy_component(key);
                 }
             }
         }
         // If there is a new peer, then set it as the peer associated with the
         // port
         if let Some(peer_id) = new_peer_comp_id {
             let port = &mut self.ports[port_index];
             port.peer_comp_id = peer_id;
             if peer_id != self.id && !port_is_closed {
                 port.associated_with_peer = true;
                 match self.get_peer_index_by_id(peer_id) {
                     Some(index) => {
                         let peer = &mut self.peers[index];
                         peer.num_associated_ports += 1;
                     },
                     None => {
                         let handle = sched_ctx.runtime.get_component_public(peer_id);
                         self.peers.push(Peer {
                             id: peer_id,
                             num_associated_ports: 1,
                             handle
                         })
                     }
                 }
             }
         }
     }
     pub(crate) fn get_port_handle(&self, port_id: PortId) -> LocalPortHandle {
         return LocalPortHandle(port_id);
     }
     // should perhaps be revised, used in main inbox
     pub(crate) fn get_port_index(&self, port_handle: LocalPortHandle) -> usize {
         return self.must_get_port_index(port_handle);
     }
     pub(crate) fn get_peer_handle(&self, peer_id: CompId) -> LocalPeerHandle {
         return LocalPeerHandle(peer_id);
     }
     pub(crate) fn get_port(&self, port_handle: LocalPortHandle) -> &Port {
         let index = self.must_get_port_index(port_handle);
         return &self.ports[index];
     }
     pub(crate) fn get_port_mut(&mut self, port_handle: LocalPortHandle) -> &mut Port {
         let index = self.must_get_port_index(port_handle);
         return &mut self.ports[index];
     }
     pub(crate) fn get_port_by_index_mut(&mut self, index: usize) -> &mut Port {
         return &mut self.ports[index];
     }
     pub(crate) fn get_peer(&self, peer_handle: LocalPeerHandle) -> &Peer {
         let index = self.must_get_peer_index(peer_handle);
         return &self.peers[index];
     }
     pub(crate) fn get_peer_mut(&mut self, peer_handle: LocalPeerHandle) -> &mut Peer {
         let index = self.must_get_peer_index(peer_handle);
         return &mut self.peers[index];
     }
     #[inline]
     pub(crate) fn iter_ports(&self) -> impl Iterator<Item=&Port> {
         return self.ports.iter();
     }
     #[inline]
     pub(crate) fn iter_ports_mut(&mut self) -> impl Iterator<Item=&mut Port> {
         return self.ports.iter_mut();
     }
     #[inline]
     pub(crate) fn iter_peers(&self) -> impl Iterator<Item=&Peer> {
         return self.peers.iter();
     }
     #[inline]
     pub(crate) fn num_ports(&self) -> usize {
         return self.ports.len();
     }
     // -------------------------------------------------------------------------
     // Local utilities
     // -------------------------------------------------------------------------
     #[inline]
     fn requires_peer_reference(port: &Port, self_id: CompId, required_if_closed: bool) -> bool {
         return (!port.state.is_closed() || required_if_closed) && port.peer_comp_id != self_id;
     }
     fn must_get_port_index(&self, handle: LocalPortHandle) -> usize {
         for (index, port) in self.ports.iter().enumerate() {
             if port.self_id == handle.0 {
                 return index;
             }
         }
         unreachable!()
     }
     fn must_get_peer_index(&self, handle: LocalPeerHandle) -> usize {
         for (index, peer) in self.peers.iter().enumerate() {
             if peer.id == handle.0 {
                 return index;
             }
         }
         unreachable!()
     }
     fn get_peer_index_by_id(&self, comp_id: CompId) -> Option<usize> {
         for (index, peer) in self.peers.iter().enumerate() {
             if peer.id == comp_id {
                 return Some(index);
             }
         }
         return None;
     }
     fn take_port_id(&mut self) -> u32 {
         let port_id = self.port_id_counter;
         self.port_id_counter = self.port_id_counter.wrapping_add(1);
         return port_id;
     }
 }
@@ \ No newline at end of file @@

src/runtime2/component/component_internet.rs

➞

Show inline comments

 use crate::protocol::eval::{ValueGroup, Value};
 use crate::runtime2::*;
 use crate::runtime2::component::{CompCtx, CompId, PortInstruction};
 use crate::runtime2::stdlib::internet::*;
 use crate::runtime2::poll::*;
 use super::component::{self, *};
 use super::control_layer::*;
 use super::consensus::*;
 use std::io::ErrorKind as IoErrorKind;
 enum SocketState {
     Connected(SocketTcpClient),
     Error,
+}
 impl SocketState {
     fn get_socket(&self) -> &SocketTcpClient {
         match self {
             SocketState::Connected(v) => v,
             SocketState::Error => unreachable!(),
+        }
+    }
+}
 /// States from the point of view of the component that is connecting to this
 /// TCP component (i.e. from the point of view of attempting to interface with
 /// a socket).
 #[derive(PartialEq, Debug)]
 enum SyncState {
     AwaitingCmd,
     Getting,
     Putting,
     FinishSync,
     FinishSyncThenQuit,
+}
 pub struct ComponentTcpClient {
     // Properties for the tcp socket
     socket_state: SocketState,
     sync_state: SyncState,
     poll_ticket: Option<PollTicket>,
     inbox_main: InboxMain,
     inbox_backup: InboxBackup,
     pdl_input_port_id: PortId, // input from PDL, so transmitted over socket
     pdl_output_port_id: PortId, // output towards PDL, so received over socket
     input_union_send_tag_value: i64,
     input_union_receive_tag_value: i64,
     input_union_finish_tag_value: i64,
     input_union_shutdown_tag_value: i64,
     // Generic component state
     exec_state: CompExecState,
     control: ControlLayer,
     consensus: Consensus,
     // Temporary variables
     byte_buffer: Vec<u8>,
+}
 impl Component for ComponentTcpClient {
     fn on_creation(&mut self, id: CompId, sched_ctx: &SchedulerCtx) {
         // Retrieve type information for messages we're going to receive
         let pd = &sched_ctx.runtime.protocol;
         let cmd_type = pd.find_type(b"std.internet", b"Cmd")
             .expect("'Cmd' type in the 'std.internet' module");
         let cmd_type = cmd_type
             .as_union();
         self.input_union_send_tag_value = cmd_type.get_variant_tag_value(b"Send").unwrap();
         self.input_union_receive_tag_value = cmd_type.get_variant_tag_value(b"Receive").unwrap();
         self.input_union_finish_tag_value = cmd_type.get_variant_tag_value(b"Finish").unwrap();
         self.input_union_shutdown_tag_value = cmd_type.get_variant_tag_value(b"Shutdown").unwrap();
         // Register socket for async events
         if let SocketState::Connected(socket) = &self.socket_state {
             let self_handle = sched_ctx.runtime.get_component_public(id);
             let poll_ticket = sched_ctx.polling.register(socket, self_handle, true, true)
                 .expect("registering tcp component");
             debug_assert!(self.poll_ticket.is_none());
             self.poll_ticket = Some(poll_ticket);
+        }
+    }
     fn on_shutdown(&mut self, sched_ctx: &SchedulerCtx) {
         if let Some(poll_ticket) = self.poll_ticket.take() {
             sched_ctx.polling.unregister(poll_ticket)
                 .expect("unregistering tcp component");
+        }
+    }
     fn adopt_message(&mut self, _comp_ctx: &mut CompCtx, message: DataMessage) {
         let slot = &mut self.inbox_main[0];
         if slot.is_none() {
             *slot = Some(message);
         } else {
             self.inbox_backup.push(message);
+        }
+    }
     fn handle_message(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx, message: Message) {
         match message {
             Message::Data(message) => {
                 self.handle_incoming_data_message(sched_ctx, comp_ctx, message);
             },
             Message::Sync(message) => {
                 let decision = self.consensus.receive_sync_message(sched_ctx, comp_ctx, message);
                 component::default_handle_sync_decision(sched_ctx, &mut self.exec_state, comp_ctx, decision, &mut self.consensus);
             },
             Message::Control(message) => {
                 if let Err(location_and_message) = component::default_handle_control_message(
                     &mut self.exec_state, &mut self.control, &mut self.consensus,
                     message, sched_ctx, comp_ctx
+                    message, sched_ctx, comp_ctx, &mut self.inbox_main, &mut self.inbox_backup
                 ) {
                     component::default_handle_error_for_builtin(&mut self.exec_state, sched_ctx, location_and_message);
+                }
             },
             Message::Poll => {
                 sched_ctx.info("Received polling event");
             },
+        }
+    }
     fn run(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx) -> CompScheduling {
         sched_ctx.info(&format!("Running component ComponentTcpClient (mode: {:?}, sync state: {:?})", self.exec_state.mode, self.sync_state));
         match self.exec_state.mode {
             CompMode::BlockedSelect => {
             CompMode::BlockedSelect |
             CompMode::PutPortsBlockedTransferredPorts |
             CompMode::PutPortsBlockedAwaitingAcks |
             CompMode::PutPortsBlockedSendingPort |
             CompMode::NewComponentBlocked => {
                 // Not possible: we never enter this state
                 unreachable!();
             },
             CompMode::NonSync => {
                 // When in non-sync mode
                 match &mut self.socket_state {
                     SocketState::Connected(_socket) => {
                         if self.sync_state == SyncState::FinishSyncThenQuit {
                             // Previous request was to let the component shut down
                             self.exec_state.set_as_start_exit(ExitReason::Termination);
                         } else {
                             // Reset for a new request
                             self.sync_state = SyncState::AwaitingCmd;
                             component::default_handle_sync_start(
                                 &mut self.exec_state, &mut self.inbox_main, sched_ctx, comp_ctx, &mut self.consensus
                             );
+                        }
                         return CompScheduling::Immediate;
                     },
                     SocketState::Error => {
                         // Could potentially send an error message to the
                         // connected component.
                         self.exec_state.set_as_start_exit(ExitReason::ErrorNonSync);
                         return CompScheduling::Immediate;
+                    }
+                }
             },
             CompMode::Sync => {
                 // When in sync mode: wait for a command to come in
                 match self.sync_state {
                     SyncState::AwaitingCmd => {
                         match component::default_attempt_get(
                             &mut self.exec_state, self.pdl_input_port_id, PortInstruction::NoSource,
                             &mut self.inbox_main, &mut self.inbox_backup, sched_ctx, comp_ctx,
                             &mut self.control, &mut self.consensus
                         ) {
                             GetResult::Received(message) => {
                                 let (tag_value, embedded_heap_pos) = message.content.values[0].as_union();
                                 if tag_value == self.input_union_send_tag_value {
                                     // Retrieve bytes from the message
                                     self.byte_buffer.clear();
                                     let union_content = &message.content.regions[embedded_heap_pos as usize];
                                     debug_assert_eq!(union_content.len(), 1);
                                     let array_heap_pos = union_content[0].as_array();
                                     let array_values = &message.content.regions[array_heap_pos as usize];
                                     self.byte_buffer.reserve(array_values.len());
                                     for value in array_values {
                                         self.byte_buffer.push(value.as_uint8());
+                                    }
                                     self.sync_state = SyncState::Putting;
                                 } else if tag_value == self.input_union_receive_tag_value {
                                     // Component requires a `recv`
                                     self.sync_state = SyncState::Getting;
                                 } else if tag_value == self.input_union_finish_tag_value {
                                     // Component requires us to end the sync round
                                     self.sync_state = SyncState::FinishSync;
                                 } else if tag_value == self.input_union_shutdown_tag_value {
                                     // Component wants to close the connection
                                     self.sync_state = SyncState::FinishSyncThenQuit;
                                 } else {
                                     unreachable!("got tag_value {}", tag_value)
+                                }
                                 return CompScheduling::Immediate;
                             },
                             GetResult::NoMessage => {
                                 return CompScheduling::Sleep;
                             },
                             GetResult::Error(location_and_message) => {
                                 component::default_handle_error_for_builtin(&mut self.exec_state, sched_ctx, location_and_message);
                                 return CompScheduling::Immediate;
+                            }
+                        }
                     },
                     SyncState::Putting => {
                         // We're supposed to send a user-supplied message fully
                         // over the socket. But we might end up blocking. In
                         // that case the component goes to sleep until it is
                         // polled.
                         let socket = self.socket_state.get_socket();
                         while !self.byte_buffer.is_empty() {
                             match socket.send(&self.byte_buffer) {
                                 Ok(bytes_sent) => {
                                     self.byte_buffer.drain(..bytes_sent);
                                 },
                                 Err(err) => {
                                     if err.kind() == IoErrorKind::WouldBlock {
                                         return CompScheduling::Sleep; // wait until notified
                                     } else {
                                         todo!("handle socket.send error {:?}", err)
+                                    }
+                                }
+                            }
+                        }
                         // If here then we're done putting the data, we can
                         // finish the sync round
                         component::default_handle_sync_end(&mut self.exec_state, sched_ctx, comp_ctx, &mut self.consensus);
                         return CompScheduling::Requeue;
                     },
                     SyncState::Getting => {
                         // We're going to try and receive a single message. If
                         // this causes us to end up blocking the component
                         // goes to sleep until it is polled.
                         const BUFFER_SIZE: usize = 1024; // TODO: Move to config
                         let socket = self.socket_state.get_socket();
                         self.byte_buffer.resize(BUFFER_SIZE, 0);
                         match socket.receive(&mut self.byte_buffer) {
                             Ok(num_received) => {
                                 self.byte_buffer.resize(num_received, 0);
                                 let message_content = self.bytes_to_data_message_content(&self.byte_buffer);
                                 let send_result = component::default_send_data_message(&mut self.exec_state, self.pdl_output_port_id, PortInstruction::NoSource, message_content, sched_ctx, &mut self.consensus, comp_ctx);
                                 let send_result = component::default_send_data_message(
                                     &mut self.exec_state, self.pdl_output_port_id, PortInstruction::NoSource,
                                     message_content, sched_ctx, &mut self.consensus, &mut self.control, comp_ctx
                                 );
                                 if let Err(location_and_message) = send_result {
                                     component::default_handle_error_for_builtin(&mut self.exec_state, sched_ctx, location_and_message);
                                     return CompScheduling::Immediate;
                                 } else {
                                     let scheduling = send_result.unwrap();
                                     self.sync_state = SyncState::AwaitingCmd;
                                     return scheduling;
+                                }
                             },
                             Err(err) => {
                                 if err.kind() == IoErrorKind::WouldBlock {
                                     return CompScheduling::Sleep; // wait until polled
                                 } else {
                                     todo!("handle socket.receive error {:?}", err)
+                                }
+                            }
+                        }
                     },
                     SyncState::FinishSync | SyncState::FinishSyncThenQuit => {
                         component::default_handle_sync_end(&mut self.exec_state, sched_ctx, comp_ctx, &mut self.consensus);
                         return CompScheduling::Requeue;
                     },
+                }
             },
             CompMode::BlockedGet => {
                 // Entered when awaiting a new command
                 debug_assert_eq!(self.sync_state, SyncState::AwaitingCmd);
                 return CompScheduling::Sleep;
             },
             CompMode::SyncEnd | CompMode::BlockedPut =>
                 return CompScheduling::Sleep,
             CompMode::StartExit =>
                 return component::default_handle_start_exit(&mut self.exec_state, &mut self.control, sched_ctx, comp_ctx, &mut self.consensus),
             CompMode::BusyExit =>
                 return component::default_handle_busy_exit(&mut self.exec_state, &mut self.control, sched_ctx),
             CompMode::Exit =>
                 return component::default_handle_exit(&self.exec_state),
+        }
+    }
+}
 impl ComponentTcpClient {
     pub(crate) fn new(arguments: ValueGroup) -> Self {
         use std::net::{IpAddr, Ipv4Addr};
         debug_assert_eq!(arguments.values.len(), 4);
         // Parsing arguments
         let ip_heap_pos = arguments.values[0].as_array();
         let ip_elements = &arguments.regions[ip_heap_pos as usize];
         if ip_elements.len() != 4 {
             todo!("friendly error reporting: ip contains 4 octects");
+        }
         let ip_address = IpAddr::V4(Ipv4Addr::new(
             ip_elements[0].as_uint8(), ip_elements[1].as_uint8(),
             ip_elements[2].as_uint8(), ip_elements[3].as_uint8()
         ));
         let port = arguments.values[1].as_uint16();
         let input_port = component::port_id_from_eval(arguments.values[2].as_input());
         let output_port = component::port_id_from_eval(arguments.values[3].as_output());
         let socket = SocketTcpClient::new(ip_address, port);
         if let Err(socket) = socket {
             todo!("friendly error reporting: failed to open socket (reason: {:?})", socket);
+        }
         return Self{
             socket_state: SocketState::Connected(socket.unwrap()),
             sync_state: SyncState::AwaitingCmd,
             poll_ticket: None,
             inbox_main: vec![None],
             inbox_backup: Vec::new(),
             input_union_send_tag_value: -1,
             input_union_receive_tag_value: -1,
             input_union_finish_tag_value: -1,
             input_union_shutdown_tag_value: -1,
             pdl_input_port_id: input_port,
             pdl_output_port_id: output_port,
             exec_state: CompExecState::new(),
             control: ControlLayer::default(),
             consensus: Consensus::new(),
             byte_buffer: Vec::new(),
+        }
+    }
     // Handles incoming data from the PDL side (hence, going into the socket)
     fn handle_incoming_data_message(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, message: DataMessage) {
         if self.exec_state.mode.is_in_sync_block() {
             self.consensus.handle_incoming_data_message(comp_ctx, &message);
+        }
         match component::default_handle_incoming_data_message(
             &mut self.exec_state, &mut self.inbox_main, comp_ctx, message, sched_ctx, &mut self.control
         ) {
             IncomingData::PlacedInSlot => {},
             IncomingData::SlotFull(message) => {
                 self.inbox_backup.push(message);
+            }
+        }
+    }
     fn data_message_to_bytes(&self, message: DataMessage, bytes: &mut Vec<u8>) {
         debug_assert_eq!(message.data_header.target_port, self.pdl_input_port_id);
         debug_assert_eq!(message.content.values.len(), 1);
         if let Value::Array(array_pos) = message.content.values[0] {
             let region = &message.content.regions[array_pos as usize];
             bytes.reserve(region.len());
             for value in region {
                 bytes.push(value.as_uint8());
+            }
         } else {
             unreachable!();
+        }
+    }
     fn bytes_to_data_message_content(&self, buffer: &[u8]) -> ValueGroup {
         // Turn bytes into silly executor-style array
         let mut values = Vec::with_capacity(buffer.len());
         for byte in buffer.iter().copied() {
             values.push(Value::UInt8(byte));
+        }
         // Put in a value group
         let mut value_group = ValueGroup::default();
         value_group.regions.push(values);
         value_group.values.push(Value::Array(0));
         return value_group;
+    }
+}
@@ \ No newline at end of file @@

src/runtime2/component/component_pdl.rs

➞

Show inline comments

 use crate::random::Random;
 use crate::protocol::*;
 use crate::protocol::ast::ProcedureDefinitionId;
 use crate::protocol::eval::{
     PortId as EvalPortId, Prompt,
     ValueGroup, Value,
     EvalContinuation, EvalResult, EvalError
 };
 use crate::runtime2::runtime::CompId;
 use crate::runtime2::scheduler::SchedulerCtx;
 use crate::runtime2::communication::*;
 use super::component::{
     self,
     InboxMain, InboxBackup, GetResult,
     CompExecState, Component, CompScheduling, CompError, CompMode, ExitReason,
     port_id_from_eval, port_id_to_eval
 };
 use super::component_context::*;
 use super::control_layer::*;
 use super::consensus::Consensus;
 pub enum ExecStmt {
     CreatedChannel((Value, Value)),
     PerformedPut,
     PerformedGet(ValueGroup),
     PerformedSelectWait(u32),
     None,
+}
 impl ExecStmt {
     fn take(&mut self) -> ExecStmt {
         let mut value = ExecStmt::None;
         std::mem::swap(self, &mut value);
         return value;
+    }
     fn is_none(&self) -> bool {
         match self {
             ExecStmt::None => return true,
             _ => return false,
+        }
+    }
+}
 pub struct ExecCtx {
     stmt: ExecStmt,
+}
 impl RunContext for ExecCtx {
     fn performed_put(&mut self, _port: EvalPortId) -> bool {
         match self.stmt.take() {
             ExecStmt::None => return false,
             ExecStmt::PerformedPut => return true,
             _ => unreachable!(),
+        }
+    }
     fn performed_get(&mut self, _port: EvalPortId) -> Option<ValueGroup> {
         match self.stmt.take() {
             ExecStmt::None => return None,
             ExecStmt::PerformedGet(value) => return Some(value),
             _ => unreachable!(),
+        }
+    }
     fn fires(&mut self, _port: EvalPortId) -> Option<Value> {
         todo!("remove fires")
+    }
     fn performed_fork(&mut self) -> Option<bool> {
         todo!("remove fork")
+    }
     fn created_channel(&mut self) -> Option<(Value, Value)> {
         match self.stmt.take() {
             ExecStmt::None => return None,
             ExecStmt::CreatedChannel(ports) => return Some(ports),
             _ => unreachable!(),
+        }
+    }
     fn performed_select_wait(&mut self) -> Option<u32> {
         match self.stmt.take() {
             ExecStmt::None => return None,
             ExecStmt::PerformedSelectWait(selected_case) => Some(selected_case),
             _v => unreachable!(),
+        }
+    }
+}
 struct SelectCase {
     involved_ports: Vec<LocalPortHandle>,
+}
 // TODO: @Optimize, flatten cases into single array, have index-pointers to next case
 struct SelectState {
     cases: Vec<SelectCase>,
     next_case: u32,
     num_cases: u32,
     random: Random,
     candidates_workspace: Vec<usize>,
+}
 enum SelectDecision {
     None,
     Case(u32), // contains case index, should be passed along to PDL code
+}
 impl SelectState {
     fn new() -> Self {
         return Self{
             cases: Vec::new(),
             next_case: 0,
             num_cases: 0,
             random: Random::new(),
             candidates_workspace: Vec::new(),
+        }
+    }
     fn handle_select_start(&mut self, num_cases: u32) {
         self.cases.clear();
         self.next_case = 0;
         self.num_cases = num_cases;
+    }
     /// Register a port as belonging to a particular case. As for correctness of
     /// PDL code one cannot register the same port twice, this function might
     /// return an error
     fn register_select_case_port(&mut self, comp_ctx: &CompCtx, case_index: u32, _port_index: u32, port_id: PortId) -> Result<(), PortId> {
         // Retrieve case and port handle
         self.ensure_at_case(case_index);
         let cur_case = &mut self.cases[case_index as usize];
         let port_handle = comp_ctx.get_port_handle(port_id);
         debug_assert_eq!(cur_case.involved_ports.len(), _port_index as usize);
         // Make sure port wasn't added before, we disallow having the same port
         // in the same select guard twice.
         if cur_case.involved_ports.contains(&port_handle) {
             return Err(port_id);
+        }
         cur_case.involved_ports.push(port_handle);
         return Ok(());
+    }
     /// Notification that all ports have been registered and we should now wait
     /// until the appropriate messages have come in.
     fn handle_select_waiting_point(&mut self, inbox: &InboxMain, comp_ctx: &CompCtx) -> SelectDecision {
         if self.num_cases != self.next_case {
             // This happens when there are >=1 select cases written at the end
             // of the select block.
             self.ensure_at_case(self.num_cases - 1);
+        }
         return self.has_decision(inbox, comp_ctx);
+    }
     fn handle_updated_inbox(&mut self, inbox: &InboxMain, comp_ctx: &CompCtx) -> SelectDecision {
         return self.has_decision(inbox, comp_ctx);
+    }
     /// Internal helper, pushes empty cases inbetween last case and provided new
     /// case index.
     fn ensure_at_case(&mut self, new_case_index: u32) {
         // Push an empty case for all intermediate cases that were not
         // registered with a port.
         debug_assert!(new_case_index >= self.next_case && new_case_index < self.num_cases);
         for _ in self.next_case..new_case_index + 1 {
             self.cases.push(SelectCase{ involved_ports: Vec::new() });
+        }
         self.next_case = new_case_index + 1;
+    }
     /// Checks if a decision can be reached
     fn has_decision(&mut self, inbox: &InboxMain, comp_ctx: &CompCtx) -> SelectDecision {
         self.candidates_workspace.clear();
         if self.cases.is_empty() {
             // If there are no cases then we can immediately reach a "bogus
             // decision".
             return SelectDecision::Case(0);
+        }
         // Need to check for valid case
         'case_loop: for (case_index, case) in self.cases.iter().enumerate() {
             for port_handle in case.involved_ports.iter().copied() {
                 let port_index = comp_ctx.get_port_index(port_handle);
                 if inbox[port_index].is_none() {
                     // Condition not satisfied
                     continue 'case_loop;
+                }
+            }
             // If here then the case guard is satisfied
             self.candidates_workspace.push(case_index);
+        }
         if self.candidates_workspace.is_empty() {
             return SelectDecision::None;
         } else {
             let candidate_index = self.random.get_u64() as usize % self.candidates_workspace.len();
             return SelectDecision::Case(self.candidates_workspace[candidate_index] as u32);
+        }
+    }
+}
 pub(crate) struct CompPDL {
     pub exec_state: CompExecState,
     select_state: SelectState,
     pub prompt: Prompt,
     pub control: ControlLayer,
     pub consensus: Consensus,
     pub sync_counter: u32,
     pub exec_ctx: ExecCtx,
     // TODO: Temporary field, simulates future plans of having one storage place
     //  reserved per port.
     // Should be same length as the number of ports. Corresponding indices imply
     // message is intended for that port.
     pub inbox_main: InboxMain,
-    pub inbox_backup: Vec<DataMessage>,
+    pub inbox_backup: InboxBackup,
+}
 impl Component for CompPDL {
     fn on_creation(&mut self, _id: CompId, _sched_ctx: &SchedulerCtx) {
         // Intentionally empty
+    }
     fn on_shutdown(&mut self, _sched_ctx: &SchedulerCtx) {
         // Intentionally empty
+    }
     fn adopt_message(&mut self, comp_ctx: &mut CompCtx, message: DataMessage) {
         let port_handle = comp_ctx.get_port_handle(message.data_header.target_port);
         let port_index = comp_ctx.get_port_index(port_handle);
         if self.inbox_main[port_index].is_none() {
             self.inbox_main[port_index] = Some(message);
         } else {
             self.inbox_backup.push(message);
+        }
+    }
     fn handle_message(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx, mut message: Message) {
-        // sched_ctx.log(&format!("handling message: {:?}", message));
+        sched_ctx.debug(&format!("handling message: {:?}", message));
         if let Some(new_target) = self.control.should_reroute(&mut message) {
             let mut target = sched_ctx.runtime.get_component_public(new_target); // TODO: @NoDirectHandle
             sched_ctx.debug(&format!("rerouting to: {:?}", new_target));
             target.send_message_logged(sched_ctx, message, false); // not waking up: we schedule once we've received all PortPeerChanged Acks
             let _should_remove = target.decrement_users();
             debug_assert!(_should_remove.is_none());
             return;
+        }
         sched_ctx.debug("handling message myself");
         match message {
             Message::Data(message) => {
                 self.handle_incoming_data_message(sched_ctx, comp_ctx, message);
             },
             Message::Control(message) => {
                 if let Err(location_and_message) = component::default_handle_control_message(
                     &mut self.exec_state, &mut self.control, &mut self.consensus,
                     message, sched_ctx, comp_ctx
+                    message, sched_ctx, comp_ctx, &mut self.inbox_main, &mut self.inbox_backup
                 ) {
                     self.handle_generic_component_error(sched_ctx, location_and_message);
+                }
             },
             Message::Sync(message) => {
                 self.handle_incoming_sync_message(sched_ctx, comp_ctx, message);
             },
             Message::Poll => {
                 unreachable!(); // because we never register at the polling thread
+            }
+        }
+    }
     fn run(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx) -> CompScheduling {
         use EvalContinuation as EC;
         sched_ctx.info(&format!("Running component (mode: {:?})", self.exec_state.mode));
         // Depending on the mode don't do anything at all, take some special
         // actions, or fall through and run the PDL code.
         match self.exec_state.mode {
             CompMode::NonSync | CompMode::Sync => {
                 // continue and run PDL code
             },
             CompMode::SyncEnd | CompMode::BlockedGet | CompMode::BlockedPut | CompMode::BlockedSelect => {
             CompMode::SyncEnd | CompMode::BlockedGet |
             CompMode::BlockedPut | CompMode::BlockedSelect | CompMode::PutPortsBlockedTransferredPorts |
             CompMode::PutPortsBlockedAwaitingAcks | CompMode::PutPortsBlockedSendingPort |
             CompMode::NewComponentBlocked => {
                 return CompScheduling::Sleep;
+            }
             CompMode::StartExit => return component::default_handle_start_exit(
                 &mut self.exec_state, &mut self.control, sched_ctx, comp_ctx, &mut self.consensus
             ),
             CompMode::BusyExit => return component::default_handle_busy_exit(
                 &mut self.exec_state, &self.control, sched_ctx
             ),
             CompMode::Exit => return component::default_handle_exit(&self.exec_state),
+        }
         let run_result = self.execute_prompt(&sched_ctx);
         if let Err(error) = run_result {
             self.handle_component_error(sched_ctx, CompError::Executor(error));
             return CompScheduling::Immediate;
+        }
         let run_result = run_result.unwrap();
         match run_result {
             EC::Stepping => unreachable!(), // execute_prompt runs until this is no longer returned
             EC::BranchInconsistent | EC::NewFork | EC::BlockFires(_) => todo!("remove these"),
             // Results that can be returned in sync mode
             EC::SyncBlockEnd => {
                 component::default_handle_sync_end(&mut self.exec_state, sched_ctx, comp_ctx, &mut self.consensus);
                 return CompScheduling::Immediate;
             },
             EC::BlockGet(expr_id, port_id) => {
                 debug_assert_eq!(self.exec_state.mode, CompMode::Sync);
                 debug_assert!(self.exec_ctx.stmt.is_none());
                 let port_id = port_id_from_eval(port_id);
                 match component::default_attempt_get(
                     &mut self.exec_state, port_id, PortInstruction::SourceLocation(expr_id),
                     &mut self.inbox_main, &mut self.inbox_backup, sched_ctx, comp_ctx,
                     &mut self.control, &mut self.consensus
                 ) {
                     GetResult::Received(message) => {
                         self.exec_ctx.stmt = ExecStmt::PerformedGet(message.content);
                         return CompScheduling::Immediate;
                     },
                     GetResult::NoMessage => {
                         return CompScheduling::Sleep;
                     },
                     GetResult::Error(location_and_message) => {
                         self.handle_generic_component_error(sched_ctx, location_and_message);
                         return CompScheduling::Immediate;
+                    }
+                }
             },
             EC::Put(expr_id, port_id, value) => {
                 debug_assert_eq!(self.exec_state.mode, CompMode::Sync);
                 sched_ctx.info(&format!("Putting value {:?}", value));
                 // Send the message
                 let target_port_id = port_id_from_eval(port_id);
                 let send_result = component::default_send_data_message(
                     &mut self.exec_state, target_port_id,
                     PortInstruction::SourceLocation(expr_id), value,
                     sched_ctx, &mut self.consensus, comp_ctx
+                    sched_ctx, &mut self.consensus, &mut self.control, comp_ctx
                 );
                 if let Err(location_and_message) = send_result {
                     self.handle_generic_component_error(sched_ctx, location_and_message);
                     return CompScheduling::Immediate;
                 } else {
                     // When `run` is called again (potentially after becoming
                     // unblocked) we need to instruct the executor that we performed
                     // the `put`
                     let scheduling = send_result.unwrap();
                     self.exec_ctx.stmt = ExecStmt::PerformedPut;
                     return scheduling;
+                }
             },
             EC::SelectStart(num_cases, _num_ports) => {
                 debug_assert_eq!(self.exec_state.mode, CompMode::Sync);
                 self.select_state.handle_select_start(num_cases);
                 return CompScheduling::Requeue;
             },
             EC::SelectRegisterPort(expr_id, case_index, port_index, port_id) => {
                 debug_assert_eq!(self.exec_state.mode, CompMode::Sync);
                 let port_id = port_id_from_eval(port_id);
                 let port_handle = comp_ctx.get_port_handle(port_id);
                 // Note: we register the "last_instruction" here already. This
                 // way if we get a `ClosePort` message, the condition to fail
                 // the synchronous round is satisfied.
                 let port_info = comp_ctx.get_port_mut(port_handle);
                 port_info.last_instruction = PortInstruction::SourceLocation(expr_id);
                 let port_is_closed = port_info.state.is_closed();
                 // Register port as part of select guard
                 if let Err(_err) = self.select_state.register_select_case_port(comp_ctx, case_index, port_index, port_id) {
                     // Failure occurs if a port is used twice in the same guard
                     let protocol = &sched_ctx.runtime.protocol;
                     self.handle_component_error(sched_ctx, CompError::Executor(EvalError::new_error_at_expr(
                         &self.prompt, &protocol.modules, &protocol.heap, expr_id,
                         String::from("Cannot have the one port appear in the same guard twice")
                     )));
                 } else if port_is_closed {
                     // Port is closed
                     let peer_id = comp_ctx.get_port(port_handle).peer_comp_id;
                     let protocol = &sched_ctx.runtime.protocol;
                     self.handle_component_error(sched_ctx, CompError::Executor(EvalError::new_error_at_expr(
                         &self.prompt, &protocol.modules, &protocol.heap, expr_id,
                         format!("Cannot register port, as the peer component (id:{}) has shut down", peer_id.0)
                     )));
+                }
                 return CompScheduling::Immediate;
             },
             EC::SelectWait => {
                 debug_assert_eq!(self.exec_state.mode, CompMode::Sync);
                 let select_decision = self.select_state.handle_select_waiting_point(&self.inbox_main, comp_ctx);
                 if let SelectDecision::Case(case_index) = select_decision {
                     // Reached a conclusion, so we can continue immediately
                     self.exec_ctx.stmt = ExecStmt::PerformedSelectWait(case_index);
                     self.exec_state.mode = CompMode::Sync;
                     return CompScheduling::Immediate;
                 } else {
                     // No decision yet
                     self.exec_state.mode = CompMode::BlockedSelect;
                     return CompScheduling::Sleep;
+                }
             },
             // Results that can be returned outside of sync mode
             EC::ComponentTerminated => {
                 self.exec_state.set_as_start_exit(ExitReason::Termination);
                 return CompScheduling::Immediate;
             },
             EC::SyncBlockStart => {
                 component::default_handle_sync_start(
                     &mut self.exec_state, &mut self.inbox_main, sched_ctx, comp_ctx, &mut self.consensus
                 );
                 return CompScheduling::Immediate;
             },
             EC::NewComponent(definition_id, type_id, arguments) => {
                 debug_assert_eq!(self.exec_state.mode, CompMode::NonSync);
                 self.create_component_and_transfer_ports(
                     sched_ctx, comp_ctx,
                 component::default_start_create_component(
                     &mut self.exec_state, sched_ctx, comp_ctx, &mut self.control,
                     &mut self.inbox_main, &mut self.inbox_backup,
                     definition_id, type_id, arguments
                 );
                 return CompScheduling::Requeue;
             },
             EC::NewChannel => {
                 debug_assert_eq!(self.exec_state.mode, CompMode::NonSync);
                 debug_assert!(self.exec_ctx.stmt.is_none());
                 let channel = comp_ctx.create_channel();
                 self.exec_ctx.stmt = ExecStmt::CreatedChannel((
                     Value::Output(port_id_to_eval(channel.putter_id)),
                     Value::Input(port_id_to_eval(channel.getter_id))
                 ));
                 self.inbox_main.push(None);
                 self.inbox_main.push(None);
                 return CompScheduling::Immediate;
+            }
+        }
+    }
+}
 impl CompPDL {
     pub(crate) fn new(initial_state: Prompt, num_ports: usize) -> Self {
         let mut inbox_main = Vec::new();
         inbox_main.reserve(num_ports);
         for _ in 0..num_ports {
             inbox_main.push(None);
+        }
         return Self{
             exec_state: CompExecState::new(),
             select_state: SelectState::new(),
             prompt: initial_state,
             control: ControlLayer::default(),
             consensus: Consensus::new(),
             sync_counter: 0,
             exec_ctx: ExecCtx{
                 stmt: ExecStmt::None,
             },
             inbox_main,
             inbox_backup: Vec::new(),
+        }
+    }
     // -------------------------------------------------------------------------
     // Running component and handling changes in global component state
     // -------------------------------------------------------------------------
     fn execute_prompt(&mut self, sched_ctx: &SchedulerCtx) -> EvalResult {
         let mut step_result = EvalContinuation::Stepping;
         while let EvalContinuation::Stepping = step_result {
             step_result = self.prompt.step(
                 &sched_ctx.runtime.protocol.types, &sched_ctx.runtime.protocol.heap,
                 &sched_ctx.runtime.protocol.modules, &mut self.exec_ctx,
             )?;
+        }
         return Ok(step_result)
+    }
     // -------------------------------------------------------------------------
     // Handling messages
     // -------------------------------------------------------------------------
     /// Handles a message that came in through the public inbox. This function
     /// will handle putting it in the correct place, and potentially blocking
     /// the port in case too many messages are being received.
     fn handle_incoming_data_message(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, message: DataMessage) {
         use component::IncomingData;
         // Whatever we do, glean information from headers in message
         if self.exec_state.mode.is_in_sync_block() {
             self.consensus.handle_incoming_data_message(comp_ctx, &message);
+        }
         match component::default_handle_incoming_data_message(
             &mut self.exec_state, &mut self.inbox_main, comp_ctx, message,
             sched_ctx, &mut self.control
         ) {
             IncomingData::PlacedInSlot => {
                 if self.exec_state.mode == CompMode::BlockedSelect {
                     let select_decision = self.select_state.handle_updated_inbox(&self.inbox_main, comp_ctx);
                     if let SelectDecision::Case(case_index) = select_decision {
                         self.exec_ctx.stmt = ExecStmt::PerformedSelectWait(case_index);
                         self.exec_state.mode = CompMode::Sync;
+                    }
+                }
             },
             IncomingData::SlotFull(message) => {
                 self.inbox_backup.push(message);
+            }
+        }
+    }
     fn handle_incoming_sync_message(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, message: SyncMessage) {
         let decision = self.consensus.receive_sync_message(sched_ctx, comp_ctx, message);
         component::default_handle_sync_decision(sched_ctx, &mut self.exec_state, comp_ctx, decision, &mut self.consensus);
+    }
     /// Handles an error coming from the generic `component::handle_xxx`
     /// functions. Hence accepts argument as a tuple.
     fn handle_generic_component_error(&mut self, sched_ctx: &SchedulerCtx, location_and_message: (PortInstruction, String)) {
         // Retrieve location and message, display in terminal
         let (location, message) = location_and_message;
         let error = match location {
             PortInstruction::None => CompError::Component(message),
             PortInstruction::NoSource => unreachable!(), // for debugging: all in-sync errors are associated with a source location
             PortInstruction::SourceLocation(expression_id) => {
                 let protocol = &sched_ctx.runtime.protocol;
                 CompError::Executor(EvalError::new_error_at_expr(
                     &self.prompt, &protocol.modules, &protocol.heap,
                     expression_id, message
                 ))
+            }
         };
         self.handle_component_error(sched_ctx, error);
+    }
     fn handle_component_error(&mut self, sched_ctx: &SchedulerCtx, error: CompError) {
         sched_ctx.error(&format!("{}", error));
         // Set state to handle subsequent error
         let exit_reason = if self.exec_state.mode.is_in_sync_block() {
             ExitReason::ErrorInSync
         } else {
             ExitReason::ErrorNonSync
         };
         self.exec_state.set_as_start_exit(exit_reason);
+    }
     // -------------------------------------------------------------------------
     // Handling ports
     // -------------------------------------------------------------------------
     /// Creates a new component and transfers ports. Because of the stepwise
     /// process in which memory is allocated, ports are transferred, messages
     /// are exchanged, component lifecycle methods are called, etc. This
     /// function facilitates a lot of implicit assumptions (e.g. when the
     /// `Component::on_creation` method is called, the component is already
     /// registered at the runtime).
     fn create_component_and_transfer_ports(
         &mut self,
         sched_ctx: &SchedulerCtx, creator_ctx: &mut CompCtx,
         definition_id: ProcedureDefinitionId, type_id: TypeId, mut arguments: ValueGroup
     ) {
         struct PortPair{
             creator_handle: LocalPortHandle,
             creator_id: PortId,
             created_handle: LocalPortHandle,
             created_id: PortId,
+        }
         let mut opened_port_id_pairs = Vec::new();
         let mut closed_port_id_pairs = Vec::new();
         let reservation = sched_ctx.runtime.start_create_pdl_component();
         let mut created_ctx = CompCtx::new(&reservation);
         // let other_proc = &sched_ctx.runtime.protocol.heap[definition_id];
         // let self_proc = &sched_ctx.runtime.protocol.heap[self.prompt.frames[0].definition];
         // dbg_code!({
         //     sched_ctx.log(&format!(
         //         "DEBUG: Comp '{}' (ID {:?}) is creating comp '{}' (ID {:?})",
         //         self_proc.identifier.value.as_str(), creator_ctx.id,
         //         other_proc.identifier.value.as_str(), reservation.id()
         //     ));
         // });
         // Take all the ports ID that are in the `args` (and currently belong to
         // the creator component) and translate them into new IDs that are
         // associated with the component we're about to create
         let mut arg_iter = ValueGroupPortIter::new(&mut arguments);
         while let Some(port_reference) = arg_iter.next() {
             // Create port entry for new component
             let creator_port_id = port_reference.id;
             let creator_port_handle = creator_ctx.get_port_handle(creator_port_id);
             let creator_port = creator_ctx.get_port(creator_port_handle);
             let created_port_handle = created_ctx.add_port(
                 creator_port.peer_comp_id, creator_port.peer_port_id,
                 creator_port.kind, creator_port.state
             );
             let created_port = created_ctx.get_port(created_port_handle);
             let created_port_id = created_port.self_id;
             let port_id_pair = PortPair {
                 creator_handle: creator_port_handle,
                 creator_id: creator_port_id,
                 created_handle: created_port_handle,
                 created_id: created_port_id,
             };
             if creator_port.state.is_closed() {
                 closed_port_id_pairs.push(port_id_pair)
             } else {
                 opened_port_id_pairs.push(port_id_pair);
+            }
             // Modify value in arguments (bit dirty, but double vec in ValueGroup causes lifetime issues)
             let arg_value = if let Some(heap_pos) = port_reference.heap_pos {
                 &mut arg_iter.group.regions[heap_pos][port_reference.index]
             } else {
                 &mut arg_iter.group.values[port_reference.index]
             };
             match arg_value {
                 Value::Input(id) => *id = port_id_to_eval(created_port_id),
                 Value::Output(id) => *id = port_id_to_eval(created_port_id),
                 _ => unreachable!(),
+            }
+        }
         // For each transferred port pair set their peer components to the
         // correct values. This will only change the values for the ports of
         // the new component.
         let mut created_component_has_remote_peers = false;
         for pair in opened_port_id_pairs.iter() {
             let creator_port_info = creator_ctx.get_port(pair.creator_handle);
             let created_port_info = created_ctx.get_port_mut(pair.created_handle);
             if created_port_info.peer_comp_id == creator_ctx.id {
                 // Peer of the transferred port is the component that is
                 // creating the new component.
                 let created_peer_port_index = opened_port_id_pairs
                     .iter()
                     .position(|v| v.creator_id == creator_port_info.peer_port_id);
                 match created_peer_port_index {
                     Some(created_peer_port_index) => {
                         // Addendum to the above comment: but that port is also
                         // moving to the new component
                         let peer_pair = &opened_port_id_pairs[created_peer_port_index];
                         created_port_info.peer_port_id = peer_pair.created_id;
                         created_port_info.peer_comp_id = reservation.id();
                         todo!("either add 'self peer', or remove that idea from Ctx altogether");`
                     },
                     None => {
                         // Peer port remains with creator component.
                         created_port_info.peer_comp_id = creator_ctx.id;
                         created_ctx.change_port_peer(sched_ctx, pair.created_handle, Some(creator_ctx.id));
+                    }
+                }
             } else {
                 // Peer is a different component. We'll deal with sending the
                 // appropriate messages later
                 let peer_handle = creator_ctx.get_peer_handle(created_port_info.peer_comp_id);
                 let peer_info = creator_ctx.get_peer(peer_handle);
                 created_ctx.change_port_peer(sched_ctx, pair.created_handle, Some(peer_info.id));
                 created_component_has_remote_peers = true;
+            }
+        }
         // We'll now actually turn our reservation for a new component into an
         // actual component. Note that we initialize it as "not sleeping" as
         // its initial scheduling might be performed based on `Ack`s in response
         // to message exchanges between remote peers.
         let total_num_ports = opened_port_id_pairs.len() + closed_port_id_pairs.len();
         let component = component::create_component(&sched_ctx.runtime.protocol, definition_id, type_id, arguments, total_num_ports);
         let (created_key, component) = sched_ctx.runtime.finish_create_pdl_component(
             reservation, component, created_ctx, false,
         );
         component.component.on_creation(created_key.downgrade(), sched_ctx);
         // Now modify the creator's ports: remove every transferred port and
         // potentially remove the peer component.
         for pair in opened_port_id_pairs.iter() {
             // Remove peer if appropriate
             let creator_port_index = creator_ctx.get_port_index(pair.creator_handle);
             creator_ctx.change_port_peer(sched_ctx, pair.creator_handle, None);
             creator_ctx.remove_port(pair.creator_handle);
             // Transfer any messages
             if let Some(mut message) = self.inbox_main.remove(creator_port_index) {
                 message.data_header.target_port = pair.created_id;
                 component.component.adopt_message(&mut component.ctx, message)
+            }
             let mut message_index = 0;
             while message_index < self.inbox_backup.len() {
                 let message = &self.inbox_backup[message_index];
                 if message.data_header.target_port == pair.creator_id {
                     // transfer message
                     let mut message = self.inbox_backup.remove(message_index);
                     message.data_header.target_port = pair.created_id;
                     component.component.adopt_message(&mut component.ctx, message);
                 } else {
                     message_index += 1;
+                }
+            }
             let created_port_info = component.ctx.get_port(pair.created_handle);
             if created_port_info.peer_comp_id == creator_ctx.id {
                 // This handles the creation of a channel between the creator
                 // component and the newly created component. So if the creator
                 // had a `a -> b` channel, and `b` is moved to the new
                 // component, then `a` needs to set its peer component.
                 let peer_port_handle = creator_ctx.get_port_handle(created_port_info.peer_port_id);
                 let peer_port_info = creator_ctx.get_port_mut(peer_port_handle);
                 peer_port_info.peer_comp_id = component.ctx.id;
                 peer_port_info.peer_port_id = created_port_info.self_id;
                 creator_ctx.change_port_peer(sched_ctx, peer_port_handle, Some(component.ctx.id));
+            }
+        }
         // Do the same for the closed ports. Note that we might still have to
         // transfer messages that cause the new owner of the port to fail.
         for pair in closed_port_id_pairs.iter() {
             let port_index = creator_ctx.get_port_index(pair.creator_handle);
             creator_ctx.remove_port(pair.creator_handle);
             if let Some(mut message) = self.inbox_main.remove(port_index) {
                 message.data_header.target_port = pair.created_id;
                 component.component.adopt_message(&mut component.ctx, message);
+            }
             let mut message_index = 0;
             while message_index < self.inbox_backup.len() {
                 let message = &self.inbox_backup[message_index];
                 if message.data_header.target_port == pair.created_id {
                     // Transfer message
                     let mut message = self.inbox_backup.remove(message_index);
                     message.data_header.target_port = pair.created_id;
                     component.component.adopt_message(&mut component.ctx, message);
                 } else {
                     message_index += 1;
+                }
+            }
+        }
         // By now all ports and messages have been transferred. If there are any
         // peers that need to be notified about this new component, then we
         // initiate the protocol that will notify everyone here.
         if created_component_has_remote_peers {
             let created_ctx = &component.ctx;
             let schedule_entry_id = self.control.add_schedule_entry(created_ctx.id);
             for pair in opened_port_id_pairs.iter() {
                 let port_info = created_ctx.get_port(pair.created_handle);
                 if port_info.peer_comp_id != creator_ctx.id && port_info.peer_comp_id != created_ctx.id {
                     let message = self.control.add_reroute_entry(
                         creator_ctx.id, port_info.peer_port_id, port_info.peer_comp_id,
                         pair.creator_id, pair.created_id, created_ctx.id,
                         schedule_entry_id
                     );
                     let peer_handle = created_ctx.get_peer_handle(port_info.peer_comp_id);
                     let peer_info = created_ctx.get_peer(peer_handle);
                     peer_info.handle.send_message_logged(sched_ctx, message, true);
+                }
+            }
         } else {
             // Peer can be scheduled immediately
             sched_ctx.runtime.enqueue_work(created_key);
+        }
+    }
+}
 /// Recursively goes through the value group, attempting to find ports.
 /// Duplicates will only be added once.
 pub(crate) fn find_ports_in_value_group(value_group: &ValueGroup, ports: &mut Vec<PortId>) {
     // Helper to check a value for a port and recurse if needed.
     fn find_port_in_value(group: &ValueGroup, value: &Value, ports: &mut Vec<PortId>) {
         match value {
             Value::Input(port_id) | Value::Output(port_id) => {
                 // This is an actual port
                 let cur_port = PortId(port_id.id);
                 for prev_port in ports.iter() {
                     if *prev_port == cur_port {
                         // Already added
                         return;
+                    }
+                }
                 ports.push(cur_port);
             },
             Value::Array(heap_pos) |
             Value::Message(heap_pos) |
             Value::String(heap_pos) |
             Value::Struct(heap_pos) |
             Value::Union(_, heap_pos) => {
                 // Reference to some dynamic thing which might contain ports,
                 // so recurse
                 let heap_region = &group.regions[*heap_pos as usize];
                 for embedded_value in heap_region {
                     find_port_in_value(group, embedded_value, ports);
+                }
             },
             _ => {}, // values we don't care about
+        }
+    }
     // Clear the ports, then scan all the available values
     ports.clear();
     for value in &value_group.values {
         find_port_in_value(value_group, value, ports);
+    }
+}
 struct ValueGroupPortIter<'a> {
     group: &'a mut ValueGroup,
     heap_stack: Vec<(usize, usize)>,
     index: usize,
+}
 impl<'a> ValueGroupPortIter<'a> {
     fn new(group: &'a mut ValueGroup) -> Self {
         return Self{ group, heap_stack: Vec::new(), index: 0 }
+    }
+}
 struct ValueGroupPortRef {
     id: PortId,
     heap_pos: Option<usize>, // otherwise: on stack
     index: usize,
+}
 impl<'a> Iterator for ValueGroupPortIter<'a> {
     type Item = ValueGroupPortRef;
     fn next(&mut self) -> Option<Self::Item> {
         // Enter loop that keeps iterating until a port is found
         loop {
             if let Some(pos) = self.heap_stack.last() {
                 let (heap_pos, region_index) = *pos;
                 if region_index >= self.group.regions[heap_pos].len() {
                     self.heap_stack.pop();
                     continue;
+                }
                 let value = &self.group.regions[heap_pos][region_index];
                 self.heap_stack.last_mut().unwrap().1 += 1;
                 match value {
                     Value::Input(id) | Value::Output(id) => {
                         let id = PortId(id.id);
                         return Some(ValueGroupPortRef{
                             id,
                             heap_pos: Some(heap_pos),
                             index: region_index,
                         });
                     },
                     _ => {},
+                }
                 if let Some(heap_pos) = value.get_heap_pos() {
                     self.heap_stack.push((heap_pos as usize, 0));
+                }
             } else {
                 if self.index >= self.group.values.len() {
                     return None;
+                }
                 let value = &mut self.group.values[self.index];
                 self.index += 1;
                 match value {
                     Value::Input(id) | Value::Output(id) => {
                         let id = PortId(id.id);
                         return Some(ValueGroupPortRef{
                             id,
                             heap_pos: None,
                             index: self.index - 1
                         });
                     },
                     _ => {},
+                }
                 // Not a port, check if we need to enter a heap region
                 if let Some(heap_pos) = value.get_heap_pos() {
                     self.heap_stack.push((heap_pos as usize, 0));
                 } // else: just consider the next value
+            }
+        }
+    }
+}
@@ \ No newline at end of file @@

src/runtime2/component/component_random.rs

➞

Show inline comments

 use rand::prelude as random;
 use rand::RngCore;
 use crate::protocol::eval::{ValueGroup, Value};
 use crate::runtime2::*;
+use crate::runtime2::communication::*;
 use super::*;
 use super::component::{
     self,
     Component, CompExecState, CompScheduling,
     CompMode, ExitReason
 };
 use super::component::*;
 use super::control_layer::*;
 use super::consensus::*;
 /// TODO: Temporary component to figure out what to do with custom components.
 ///     This component sends random numbers between two u32 limits
 pub struct ComponentRandomU32 {
     // Properties for this specific component
     output_port_id: PortId,
     random_minimum: u32,
     random_maximum: u32,
     num_sends: u32,
     max_num_sends: u32,
     generator: random::ThreadRng,
     // Generic state-tracking
     exec_state: CompExecState,
     did_perform_send: bool, // when in sync mode
     control: ControlLayer,
     consensus: Consensus,
     inbox_main: InboxMain, // not used
     inbox_backup: InboxBackup, // not used
+}
 impl Component for ComponentRandomU32 {
     fn on_creation(&mut self, _id: CompId, _sched_ctx: &SchedulerCtx) {}
     fn on_shutdown(&mut self, sched_ctx: &SchedulerCtx) {}
     fn adopt_message(&mut self, _comp_ctx: &mut CompCtx, _message: DataMessage) {
         // Impossible since this component does not have any input ports in its
         // signature.
         unreachable!();
+    }
     fn handle_message(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx, message: Message) {
         match message {
             Message::Data(_message) => unreachable!(),
             Message::Sync(message) => {
                 let decision = self.consensus.receive_sync_message(sched_ctx, comp_ctx, message);
                 component::default_handle_sync_decision(sched_ctx, &mut self.exec_state, comp_ctx, decision, &mut self.consensus);
             },
             Message::Control(message) => {
                 if let Err(location_and_message) = component::default_handle_control_message(
                     &mut self.exec_state, &mut self.control, &mut self.consensus,
                     message, sched_ctx, comp_ctx
+                    message, sched_ctx, comp_ctx, &mut self.inbox_main, &mut self.inbox_backup
                 ) {
                     component::default_handle_error_for_builtin(&mut self.exec_state, sched_ctx, location_and_message);
+                }
             },
             Message::Poll => unreachable!(),
+        }
+    }
     fn run(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx) -> CompScheduling {
         sched_ctx.info(&format!("Running component ComponentRandomU32 (mode: {:?})", self.exec_state.mode));
         match self.exec_state.mode {
             CompMode::BlockedGet | CompMode::BlockedSelect => {
             CompMode::BlockedGet |
             CompMode::BlockedSelect |
             CompMode::PutPortsBlockedTransferredPorts |
             CompMode::PutPortsBlockedAwaitingAcks |
             CompMode::PutPortsBlockedSendingPort |
             CompMode::NewComponentBlocked => {
                 // impossible for this component, no input ports and no select
                 // blocks
                 unreachable!();
+            }
             CompMode::NonSync => {
                 // If in non-sync mode then we check if the arguments make sense
                 // (at some point in the future, this is just a testing
                 // component).
                 if self.random_minimum >= self.random_maximum {
                     // Could throw an evaluation error, but lets just panic
                     panic!("going to crash 'n burn your system now, please provide valid arguments");
+                }
                 if self.num_sends >= self.max_num_sends {
                     self.exec_state.set_as_start_exit(ExitReason::Termination);
                 } else {
                     sched_ctx.info("Entering sync mode");
                     self.did_perform_send = false;
                     component::default_handle_sync_start(
                         &mut self.exec_state, &mut [], sched_ctx, comp_ctx, &mut self.consensus
                     );
+                }
                 return CompScheduling::Immediate;
             },
             CompMode::Sync => {
                 // This component just sends a single message, then waits until
                 // consensus has been reached
                 if !self.did_perform_send {
                     sched_ctx.info("Sending random message");
                     let mut random = self.generator.next_u32() - self.random_minimum;
                     let random_delta = self.random_maximum - self.random_minimum;
                     random %= random_delta;
                     random += self.random_minimum;
                     let value_group = ValueGroup::new_stack(vec![Value::UInt32(random)]);
                     let send_result = component::default_send_data_message(
                         &mut self.exec_state, self.output_port_id,
                         PortInstruction::NoSource, value_group,
                         sched_ctx, &mut self.consensus, comp_ctx
                         sched_ctx, &mut self.consensus, &mut self.control,
                         comp_ctx
                     );
                     if let Err(location_and_message) = send_result {
                         component::default_handle_error_for_builtin(&mut self.exec_state, sched_ctx, location_and_message);
                         return CompScheduling::Immediate
                     } else {
                         // Blocked or not, we set `did_perform_send` to true. If
                         // blocked then the moment we become unblocked (and are back
                         // at the `Sync` mode) we have sent the message.
                         let scheduling = send_result.unwrap();
                         self.did_perform_send = true;
                         self.num_sends += 1;
                         return scheduling
+                    }
                 } else {
                     // Message was sent, finish this sync round
                     component::default_handle_sync_end(&mut self.exec_state, sched_ctx, comp_ctx, &mut self.consensus);
                     return CompScheduling::Requeue;
+                }
             },
             CompMode::SyncEnd | CompMode::BlockedPut => return CompScheduling::Sleep,
             CompMode::StartExit => return component::default_handle_start_exit(
                 &mut self.exec_state, &mut self.control, sched_ctx, comp_ctx, &mut self.consensus
             ),
             CompMode::BusyExit => return component::default_handle_busy_exit(
                 &mut self.exec_state, &self.control, sched_ctx
             ),
             CompMode::Exit => return component::default_handle_exit(&self.exec_state),
+        }
+    }
+}
 impl ComponentRandomU32 {
     pub(crate) fn new(arguments: ValueGroup) -> Self {
         debug_assert_eq!(arguments.values.len(), 4);
         debug_assert!(arguments.regions.is_empty());
         let port_id = component::port_id_from_eval(arguments.values[0].as_port_id());
         let minimum = arguments.values[1].as_uint32();
         let maximum = arguments.values[2].as_uint32();
         let num_sends = arguments.values[3].as_uint32();
         return Self{
             output_port_id: port_id,
             random_minimum: minimum,
             random_maximum: maximum,
             num_sends: 0,
             max_num_sends: num_sends,
             generator: random::thread_rng(),
             exec_state: CompExecState::new(),
             did_perform_send: false,
             control: ControlLayer::default(),
             consensus: Consensus::new(),
             inbox_main: Vec::new(),
             inbox_backup: Vec::new(),
+        }
+    }
+}
@@ \ No newline at end of file @@

src/runtime2/component/consensus.rs

➞

Show inline comments

 use crate::protocol::eval::ValueGroup;
 use crate::runtime2::scheduler::*;
 use crate::runtime2::runtime::*;
 use crate::runtime2::communication::*;
 use super::component_context::*;
 pub struct PortAnnotation {
     self_comp_id: CompId,
     self_port_id: PortId,
     peer_comp_id: CompId, // only valid for getter ports
     peer_port_id: PortId, // only valid for getter ports
     peer_discovered: bool, // only valid for getter ports
     mapping: Option<u32>,
     kind: PortKind,
+}
 impl PortAnnotation {
     fn new(comp_id: CompId, port_id: PortId, kind: PortKind) -> Self {
         return Self{
             self_comp_id: comp_id,
             self_port_id: port_id,
             peer_comp_id: CompId::new_invalid(),
             peer_port_id: PortId::new_invalid(),
             peer_discovered: false,
             mapping: None,
             kind,
+        }
+    }
+}
 #[derive(Debug, Eq, PartialEq)]
 enum Mode {
     NonSync,
     SyncBusy,
     SyncAwaitingSolution,
     SelectBusy,
     SelectWait,
+}
 struct SolutionCombiner {
     solution: SyncPartialSolution,
     matched_channels: usize,
+}
 impl SolutionCombiner {
     fn new() -> Self {
         return Self {
             solution: SyncPartialSolution::default(),
             matched_channels: 0,
+        }
+    }
     #[inline]
     fn has_contributions(&self) -> bool {
         return !self.solution.channel_mapping.is_empty();
+    }
     /// Returns a decision for the current round. If there is no decision (yet)
     /// then `RoundDecision::None` is returned.
     fn get_decision(&self) -> SyncRoundDecision {
         if self.matched_channels == self.solution.channel_mapping.len() {
             debug_assert_ne!(self.solution.decision, SyncRoundDecision::None);
             return self.solution.decision;
+        }
         return SyncRoundDecision::None; // even in case of failure: wait for everyone.
+    }
     fn combine_with_partial_solution(&mut self, partial: SyncPartialSolution) {
         debug_assert_ne!(self.solution.decision, SyncRoundDecision::Solution);
         debug_assert_ne!(partial.decision, SyncRoundDecision::Solution);
         if partial.decision == SyncRoundDecision::Failure {
             self.solution.decision = SyncRoundDecision::Failure;
+        }
         for entry in partial.channel_mapping {
             let channel_index = if entry.getter.is_some() && entry.putter.is_some() {
                 let channel_index = self.solution.channel_mapping.len();
                 self.solution.channel_mapping.push(entry);
                 channel_index
             } else if let Some(putter) = entry.putter {
                 self.combine_with_putter_port(putter)
             } else if let Some(getter) = entry.getter {
                 self.combine_with_getter_port(getter)
             } else {
                 unreachable!(); // both putter and getter are None
             };
             let channel = &self.solution.channel_mapping[channel_index];
             if let Some(consistent) = Self::channel_is_consistent(channel) {
                 if !consistent {
                     self.solution.decision = SyncRoundDecision::Failure;
+                }
                 self.matched_channels += 1;
+            }
+        }
         self.update_solution();
+    }
     /// Combines the currently stored global solution (if any) with the newly
     /// provided local solution. Make sure to check the `has_decision` return
     /// value afterwards.
     fn combine_with_local_solution(&mut self, _comp_id: CompId, solution: SyncLocalSolution) {
         debug_assert_ne!(self.solution.decision, SyncRoundDecision::Solution);
         // Combine partial solution with the local solution entries
         for entry in solution {
             // Match the current entry up with its peer endpoint, or add a new
             // entry.
             let channel_index = match entry {
                 SyncLocalSolutionEntry::Putter(putter) => {
                     self.combine_with_putter_port(putter)
                 },
                 SyncLocalSolutionEntry::Getter(getter) => {
                     self.combine_with_getter_port(getter)
+                }
             };
             // Check if channel is now consistent
             let channel = &self.solution.channel_mapping[channel_index];
             if let Some(consistent) = Self::channel_is_consistent(channel) {
                 if !consistent {
                     self.solution.decision = SyncRoundDecision::Failure;
+                }
                 self.matched_channels += 1;
+            }
+        }
         self.update_solution();
+    }
     /// Takes whatever partial solution is present in the solution combiner and
     /// returns it. The solution combiner's solution will end up being empty.
     /// This is used when a new leader is found and we need to pass along our
     /// partial results.
     fn take_partial_solution(&mut self) -> SyncPartialSolution {
         let mut partial_solution = SyncPartialSolution::default();
         std::mem::swap(&mut partial_solution, &mut self.solution);
         self.clear();
         return partial_solution;
+    }
     fn clear(&mut self) {
         self.solution.channel_mapping.clear();
         self.solution.decision = SyncRoundDecision::None;
         self.matched_channels = 0;
+    }
     // --- Small utilities for combining solutions
     fn combine_with_putter_port(&mut self, putter: SyncSolutionPutterPort) -> usize {
         let channel_index = self.get_channel_index_for_putter(putter.self_comp_id, putter.self_port_id);
         if let Some(channel_index) = channel_index {
             let channel = &mut self.solution.channel_mapping[channel_index];
             debug_assert!(channel.putter.is_none());
             channel.putter = Some(putter);
             return channel_index;
         } else {
             let channel_index = self.solution.channel_mapping.len();
             self.solution.channel_mapping.push(SyncSolutionChannel{
                 putter: Some(putter),
                 getter: None,
             });
             return channel_index;
+        }
+    }
     fn combine_with_getter_port(&mut self, getter: SyncSolutionGetterPort) -> usize {
         let channel_index = self.get_channel_index_for_getter(getter.peer_comp_id, getter.peer_port_id);
         if let Some(channel_index) = channel_index {
             let channel = &mut self.solution.channel_mapping[channel_index];
             debug_assert!(channel.getter.is_none());
             channel.getter = Some(getter);
             return channel_index;
         } else {
             let channel_index = self.solution.channel_mapping.len();
             self.solution.channel_mapping.push(SyncSolutionChannel{
                 putter: None,
                 getter: Some(getter)
             });
             return channel_index;
+        }
+    }
     /// Retrieve index of the channel containing a getter port that has received
     /// from the specified putter port.
     fn get_channel_index_for_putter(&self, putter_comp_id: CompId, putter_port_id: PortId) -> Option<usize> {
         for (channel_index, channel) in self.solution.channel_mapping.iter().enumerate() {
             if let Some(getter) = &channel.getter {
                 if getter.peer_comp_id == putter_comp_id && getter.peer_port_id == putter_port_id {
                     return Some(channel_index);
+                }
+            }
+        }
         return None;
+    }
     /// Retrieve index of the channel for a getter port. To find this channel
     /// the **peer** component/port IDs of the getter port are used.
     fn get_channel_index_for_getter(&self, peer_comp_id: CompId, peer_port_id: PortId) -> Option<usize> {
         for (channel_index, channel) in self.solution.channel_mapping.iter().enumerate() {
             if let Some(putter) = &channel.putter {
                 if putter.self_comp_id == peer_comp_id && putter.self_port_id == peer_port_id {
                     return Some(channel_index);
+                }
+            }
+        }
         return None;
+    }
     fn channel_is_consistent(channel: &SyncSolutionChannel) -> Option<bool> {
         if channel.putter.is_none() || channel.getter.is_none() {
             return None;
+        }
         let putter = channel.putter.as_ref().unwrap();
         let getter = channel.getter.as_ref().unwrap();
         return Some(
             !putter.failed &&
             !getter.failed &&
             putter.mapping == getter.mapping
         );
+    }
     /// Determines the global solution if all components have contributed their
     /// local solutions.
     fn update_solution(&mut self) {
         if self.matched_channels == self.solution.channel_mapping.len() {
             if self.solution.decision != SyncRoundDecision::Failure {
                 self.solution.decision = SyncRoundDecision::Solution;
+            }
+        }
+    }
+}
 /// Tracking consensus state
 pub struct Consensus {
     // General state of consensus manager
     mapping_counter: u32,
     mode: Mode,
     // State associated with sync round
     round_index: u32,
     highest_id: CompId,
     ports: Vec<PortAnnotation>,
     // State associated with arriving at a solution and being a (temporary)
     // leader in the consensus round
     solution: SolutionCombiner,
+}
 impl Consensus {
     pub(crate) fn new() -> Self {
         return Self{
             round_index: 0,
             highest_id: CompId::new_invalid(),
             ports: Vec::new(),
             mapping_counter: 0,
             mode: Mode::NonSync,
             solution: SolutionCombiner::new(),
+        }
+    }
     // -------------------------------------------------------------------------
     // Managing sync state
     // -------------------------------------------------------------------------
     /// Notifies the consensus management that the PDL code has reached the
     /// start of a sync block.
     pub(crate) fn notify_sync_start(&mut self, comp_ctx: &CompCtx) {
         debug_assert_eq!(self.mode, Mode::NonSync);
         self.highest_id = comp_ctx.id;
         self.mapping_counter = 0;
         self.mode = Mode::SyncBusy;
         // Make the internally stored port annotation array consistent with the
         // ports that the component currently owns. They should match by index
         // (i.e. annotation at index `i` corresponds to port `i` in `comp_ctx`).
         let mut needs_setting_ports = false;
         if comp_ctx.num_ports() != self.ports.len() {
             needs_setting_ports = true;
         } else {
             for (idx, port) in comp_ctx.iter_ports().enumerate() {
                 let comp_port_id = port.self_id;
                 let cons_port_id = self.ports[idx].self_port_id;
                 if comp_port_id != cons_port_id {
                     needs_setting_ports = true;
                     break;
+                }
+            }
+        }
         if needs_setting_ports {
             // Reset all ports
             self.ports.clear();
             self.ports.reserve(comp_ctx.num_ports());
             for port in comp_ctx.iter_ports() {
                 self.ports.push(PortAnnotation::new(comp_ctx.id, port.self_id, port.kind));
+            }
         } else {
             // Make sure that we consider all peers as undiscovered again
             for annotation in self.ports.iter_mut() {
                 annotation.peer_discovered = false;
+            }
+        }
+    }
     /// Notifies the consensus management that the PDL code has reached the end
     /// of a sync block. A local solution will be submitted, after which we wait
     /// until the participants in the round (hopefully) reach a conclusion.
     pub(crate) fn notify_sync_end_success(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &CompCtx) -> SyncRoundDecision {
         debug_assert_eq!(self.mode, Mode::SyncBusy);
         self.mode = Mode::SyncAwaitingSolution;
         let local_solution = self.generate_local_solution(comp_ctx, false);
         let decision = self.handle_local_solution(sched_ctx, comp_ctx, comp_ctx.id, local_solution, false);
         return decision;
+    }
     /// Notifies the consensus management that the component has encountered a
     /// critical error during the synchronous round. Hence we should report that
     /// we've failed and wait until all the participants have been notified of
     /// the error.
     pub(crate) fn notify_sync_end_failure(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &CompCtx) -> SyncRoundDecision {
         debug_assert_eq!(self.mode, Mode::SyncBusy);
         self.mode = Mode::SyncAwaitingSolution;
         let local_solution = self.generate_local_solution(comp_ctx, true);
         let decision = self.handle_local_solution(sched_ctx, comp_ctx, comp_ctx.id, local_solution, true);
         return decision;
+    }
     /// Notifies that a decision has been reached. Note that the caller should
     /// still take the appropriate actions based on the decision it is supplying
     /// to the consensus layer.
     pub(crate) fn notify_sync_decision(&mut self, _decision: SyncRoundDecision) {
         // Reset everything for the next round
         debug_assert_eq!(self.mode, Mode::SyncAwaitingSolution);
         self.mode = Mode::NonSync;
         self.round_index = self.round_index.wrapping_add(1);
         for port in self.ports.iter_mut() {
             port.mapping = None;
+        }
         self.solution.clear();
+    }
     pub(crate) fn notify_received_port(&mut self, _expected_index: usize, port_handle: LocalPortHandle, comp_ctx: &CompCtx) {
         debug_assert_eq!(_expected_index, self.ports.len());
         let port_info = comp_ctx.get_port(port_handle);
         self.ports.push(PortAnnotation{
             self_comp_id: comp_ctx.id,
             self_port_id: port_info.self_id,
             peer_comp_id: port_info.peer_comp_id,
             peer_port_id: port_info.peer_port_id,
             peer_discovered: false,
             mapping: None,
             kind: port_info.kind,
         });
+    }
     // -------------------------------------------------------------------------
     // Handling inbound and outbound messages
     // -------------------------------------------------------------------------
     /// Prepares a set of values to be sent of a channel.
     pub(crate) fn annotate_data_message(&mut self, comp_ctx: &CompCtx, port_info: &Port, content: ValueGroup) -> DataMessage {
         debug_assert_eq!(self.mode, Mode::SyncBusy); // can only send between sync start and sync end
         debug_assert!(self.ports.iter().any(|v| v.self_port_id == port_info.self_id));
         let data_header = self.create_data_header_and_update_mapping(port_info);
         let sync_header = self.create_sync_header(comp_ctx);
         return DataMessage{ data_header, sync_header, content };
         return DataMessage{
             data_header, sync_header, content,
             ports: Vec::new()
         };
+    }
     /// Handles the arrival of a new data message (needs to be called for every
     /// new data message, even though it might not end up being received). This
     /// is used to determine peers of `get`ter ports.
     // TODO: The use of this function is rather ugly. Find a more robust
     //  scheme about owners of `get`ter ports not knowing about their peers.
     pub(crate) fn handle_incoming_data_message(&mut self, comp_ctx: &CompCtx, message: &DataMessage) {
         let target_handle = comp_ctx.get_port_handle(message.data_header.target_port);
         let target_index = comp_ctx.get_port_index(target_handle);
         let annotation = &mut self.ports[target_index];
         debug_assert!(
             !annotation.peer_discovered || (
                 annotation.peer_comp_id == message.sync_header.sending_id &&
                 annotation.peer_port_id == message.data_header.source_port
+            )
         );
         annotation.peer_comp_id = message.sync_header.sending_id;
         annotation.peer_port_id = message.data_header.source_port;
         annotation.peer_discovered = true;
+    }
     /// Checks if the data message can be received (due to port annotations), if
     /// it can then `true` is returned and the caller is responsible for handing
     /// the message of to the PDL code. Otherwise the message cannot be
     /// received.
     pub(crate) fn try_receive_data_message(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, message: &DataMessage) -> bool {
         debug_assert_eq!(self.mode, Mode::SyncBusy);
         debug_assert!(self.ports.iter().any(|v| v.self_port_id == message.data_header.target_port));
         // Make sure the expected mapping matches the currently stored mapping
         for (peer_port_kind, expected_annotation) in &message.data_header.expected_mapping {
             // Determine our annotation, in order to do so we need to find the
             // port matching the peer ports
             let mut self_annotation = None;
             let mut self_annotation_found = false;
             match peer_port_kind {
                 PortAnnotationKind::Putter(peer_port) => {
                     for self_port in &self.ports {
                         if self_port.kind == PortKind::Getter &&
                             self_port.peer_discovered &&
                             self_port.peer_comp_id == peer_port.self_comp_id &&
                             self_port.peer_port_id == peer_port.self_port_id
+                        {
                             self_annotation = self_port.mapping;
                             self_annotation_found = true;
                             break;
+                        }
+                    }
                 },
                 PortAnnotationKind::Getter(peer_port) => {
                     if peer_port.peer_comp_id == comp_ctx.id {
                         // Peer indicates that we talked to it
                         let self_port_handle = comp_ctx.get_port_handle(peer_port.peer_port_id);
                         let self_port_index = comp_ctx.get_port_index(self_port_handle);
                         self_annotation = self.ports[self_port_index].mapping;
                         self_annotation_found = true;
+                    }
+                }
+            }
             if !self_annotation_found {
                 continue
+            }
             if self_annotation != *expected_annotation {
                 return false;
+            }
+        }
         // Expected mapping matches current mapping, so we will receive the message
         self.set_annotation(message.sync_header.sending_id, &message.data_header);
         // Handle the sync header embedded within the data message
         self.handle_sync_header(sched_ctx, comp_ctx, &message.sync_header);
         return true;
+    }
     /// Receives the sync message and updates the consensus state appropriately.
     pub(crate) fn receive_sync_message(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, message: SyncMessage) -> SyncRoundDecision {
         // Whatever happens: handle the sync header (possibly changing the
         // currently registered leader)
         self.handle_sync_header(sched_ctx, comp_ctx, &message.sync_header);
         match message.content {
             SyncMessageContent::NotificationOfLeader => {
                 return SyncRoundDecision::None;
             },
             SyncMessageContent::LocalSolution(solution_generator_id, local_solution) => {
                 return self.handle_local_solution(sched_ctx, comp_ctx, solution_generator_id, local_solution, false);
             },
             SyncMessageContent::PartialSolution(partial_solution) => {
                 return self.handle_partial_solution(sched_ctx, comp_ctx, partial_solution);
             },
             SyncMessageContent::GlobalSolution => {
                 debug_assert_eq!(self.mode, Mode::SyncAwaitingSolution); // leader can only find global- if we submitted local solution
                 return SyncRoundDecision::Solution;
             },
             SyncMessageContent::GlobalFailure => {
                 debug_assert_eq!(self.mode, Mode::SyncAwaitingSolution);
                 return SyncRoundDecision::Failure;
+            }
+        }
+    }
     fn handle_sync_header(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, header: &MessageSyncHeader) {
         if header.highest_id.0 > self.highest_id.0 {
             // Sender knows of someone with a higher ID. So store highest ID,
             // notify all peers, and forward local solutions
             self.highest_id = header.highest_id;
             for peer in comp_ctx.iter_peers() {
                 if peer.id == header.sending_id {
                     continue; // do not send to sender: it has the higher ID
+                }
                 // also: only send if we received a message in this round
                 let mut performed_communication = false; // TODO: Revise, temporary fix
                 for port in self.ports.iter() {
                     if port.peer_comp_id == peer.id && port.mapping.is_some() {
                         performed_communication = true;
                         break;
+                    }
+                }
                 if !performed_communication {
                     continue;
+                }
                 let message = SyncMessage{
                     sync_header: self.create_sync_header(comp_ctx),
                     content: SyncMessageContent::NotificationOfLeader,
                 };
                 peer.handle.send_message_logged(sched_ctx, Message::Sync(message), true);
+            }
             self.forward_partial_solution(sched_ctx, comp_ctx);
         } else if header.highest_id.0 < self.highest_id.0 {
             // Sender has a lower ID, so notify it of our higher one
             let message = SyncMessage{
                 sync_header: self.create_sync_header(comp_ctx),
                 content: SyncMessageContent::NotificationOfLeader,
             };
             let peer_handle = comp_ctx.get_peer_handle(header.sending_id);
             let peer_info = comp_ctx.get_peer(peer_handle);
             peer_info.handle.send_message_logged(sched_ctx, Message::Sync(message), true);
         } // else: exactly equal
+    }
     fn set_annotation(&mut self, source_comp_id: CompId, data_header: &MessageDataHeader) {
         for annotation in self.ports.iter_mut() {
             if annotation.self_port_id == data_header.target_port {
                 // Message should have already passed the `handle_new_data_message` function, so we
                 // should have already annotated the peer of the port.
                 debug_assert!(
                     annotation.peer_discovered &&
                     annotation.peer_comp_id == source_comp_id &&
                     annotation.peer_port_id == data_header.source_port
                 );
                 annotation.mapping = Some(data_header.new_mapping);
+            }
+        }
+    }
     // -------------------------------------------------------------------------
     // Leader-related methods
     // -------------------------------------------------------------------------
     fn forward_partial_solution(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx) {
         debug_assert_ne!(self.highest_id, comp_ctx.id); // not leader
         // Make sure that we have something to send
         if !self.solution.has_contributions() {
             return;
+        }
         // Swap the container with the partial solution and then send it along
         let partial_solution = self.solution.take_partial_solution();
         self.send_to_leader(sched_ctx, comp_ctx, Message::Sync(SyncMessage{
             sync_header: self.create_sync_header(comp_ctx),
             content: SyncMessageContent::PartialSolution(partial_solution),
         }));
+    }
     fn handle_local_solution(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &CompCtx, solution_sender_id: CompId, solution: SyncLocalSolution, fail_if_empty: bool) -> SyncRoundDecision {
         if self.highest_id == comp_ctx.id {
             // We are the leader
             self.solution.combine_with_local_solution(solution_sender_id, solution);
             let mut round_decision = self.solution.get_decision();
             if round_decision != SyncRoundDecision::None {
                 if fail_if_empty && self.solution.matched_channels == 0 {
                     // TODO: Not sure about this, bit of a hack. Situation is that a component
                     //  cannot interact with other components, but it is in a sync round, and has
                     //  failed that sync round.
                     round_decision = SyncRoundDecision::Failure;
+                }
                 self.broadcast_decision(sched_ctx, comp_ctx, round_decision);
+            }
             return round_decision;
         } else {
             // Forward the solution
             let message = SyncMessage{
                 sync_header: self.create_sync_header(comp_ctx),
                 content: SyncMessageContent::LocalSolution(solution_sender_id, solution),
             };
             self.send_to_leader(sched_ctx, comp_ctx, Message::Sync(message));
             return SyncRoundDecision::None;
+        }
+    }
     fn handle_partial_solution(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, solution: SyncPartialSolution) -> SyncRoundDecision {
         if self.highest_id == comp_ctx.id {
             // We are the leader, combine existing and new solution
             self.solution.combine_with_partial_solution(solution);
             let round_decision = self.solution.get_decision();
             if round_decision != SyncRoundDecision::None {
                 self.broadcast_decision(sched_ctx, comp_ctx, round_decision);
+            }
             return round_decision;
         } else {
             // Forward the partial solution
             let message = SyncMessage{
                 sync_header: self.create_sync_header(comp_ctx),
                 content: SyncMessageContent::PartialSolution(solution),
             };
             self.send_to_leader(sched_ctx, comp_ctx, Message::Sync(message));
             return SyncRoundDecision::None;
+        }
+    }
     fn broadcast_decision(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &CompCtx, decision: SyncRoundDecision) {
         debug_assert_eq!(self.highest_id, comp_ctx.id);
         let is_success = match decision {
             SyncRoundDecision::None => unreachable!(),
             SyncRoundDecision::Solution => true,
             SyncRoundDecision::Failure => false,
         };
         let mut peers = Vec::with_capacity(self.solution.solution.channel_mapping.len()); // TODO: @Performance
         for channel in self.solution.solution.channel_mapping.iter() {
             let getter = channel.getter.as_ref().unwrap();
             if getter.self_comp_id != comp_ctx.id && !peers.contains(&getter.self_comp_id) {
                 peers.push(getter.self_comp_id);
+            }
             if getter.peer_comp_id != comp_ctx.id && !peers.contains(&getter.peer_comp_id) {
                 peers.push(getter.peer_comp_id);
+            }
+        }
         for peer in peers {
             let mut handle = sched_ctx.runtime.get_component_public(peer);
             let message = Message::Sync(SyncMessage{
                 sync_header: self.create_sync_header(comp_ctx),
                 content: if is_success { SyncMessageContent::GlobalSolution } else { SyncMessageContent::GlobalFailure },
             });
             handle.send_message_logged(sched_ctx, message, true);
             let _should_remove = handle.decrement_users();
             debug_assert!(_should_remove.is_none());
+        }
+    }
     fn send_to_leader(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &CompCtx, message: Message) {
         debug_assert_ne!(self.highest_id, comp_ctx.id); // we're not the leader, // TODO: @NoDirectHandle
         let mut leader_info = sched_ctx.runtime.get_component_public(self.highest_id);
         leader_info.send_message_logged(sched_ctx, message, true);
         let should_remove = leader_info.decrement_users();
         if let Some(key) = should_remove {
             sched_ctx.runtime.destroy_component(key);
+        }
+    }
     // -------------------------------------------------------------------------
     // Small utilities
     // -------------------------------------------------------------------------
     fn generate_local_solution(&self, comp_ctx: &CompCtx, failed: bool) -> SyncLocalSolution {
         let mut local_solution = Vec::with_capacity(self.ports.len());
         for port in &self.ports {
             if let Some(mapping) = port.mapping {
                 let port_handle = comp_ctx.get_port_handle(port.self_port_id);
                 let port_info = comp_ctx.get_port(port_handle);
                 let new_entry = match port_info.kind {
                     PortKind::Putter => SyncLocalSolutionEntry::Putter(SyncSolutionPutterPort{
                         self_comp_id: comp_ctx.id,
                         self_port_id: port_info.self_id,
                         mapping,
                         failed
                     }),
                     PortKind::Getter => SyncLocalSolutionEntry::Getter(SyncSolutionGetterPort{
                         self_comp_id: comp_ctx.id,
                         self_port_id: port_info.self_id,
                         peer_comp_id: port.peer_comp_id,
                         peer_port_id: port.peer_port_id,
                         mapping,
                         failed
                     })
                 };
                 local_solution.push(new_entry);
+            }
+        }
         return local_solution;
+    }
     fn create_data_header_and_update_mapping(&mut self, port_info: &Port) -> MessageDataHeader {
         let mut expected_mapping = Vec::with_capacity(self.ports.len());
         let mut port_index = usize::MAX;
         for (index, port) in self.ports.iter().enumerate() {
             if port.self_port_id == port_info.self_id {
                 port_index = index; // remember for later updating
+            }
             // Add all of the
             let annotation_kind = match port.kind {
                 PortKind::Putter => {
                     PortAnnotationKind::Putter(PortAnnotationPutter{
                         self_comp_id: port.self_comp_id,
                         self_port_id: port.self_port_id
                     })
                 },
                 PortKind::Getter => {
                     if !port.peer_discovered {
                         continue;
+                    }
                     PortAnnotationKind::Getter(PortAnnotationGetter{
                         self_comp_id: port.self_comp_id,
                         self_port_id: port.self_port_id,
                         peer_comp_id: port.peer_comp_id,
                         peer_port_id: port.peer_port_id,
                     })
+                }
             };
             expected_mapping.push((annotation_kind, port.mapping));
+        }
         let new_mapping = self.take_mapping();
         self.ports[port_index].mapping = Some(new_mapping);
         debug_assert_eq!(port_info.kind, PortKind::Putter);
         return MessageDataHeader{
             expected_mapping,
             new_mapping,
             source_port: port_info.self_id,
             target_port: port_info.peer_port_id,
         };
+    }
     #[inline]
     fn create_sync_header(&self, comp_ctx: &CompCtx) -> MessageSyncHeader {
         return MessageSyncHeader{
             sync_round: self.round_index,
             sending_id: comp_ctx.id,
             highest_id: self.highest_id,
         };
+    }
     #[inline]
     fn take_mapping(&mut self) -> u32 {
         let mapping = self.mapping_counter;
         self.mapping_counter = self.mapping_counter.wrapping_add(1);
         return mapping;
+    }
+}
@@ \ No newline at end of file @@

src/runtime2/component/control_layer.rs

➞

Show inline comments

 use crate::runtime2::runtime::*;
 use crate::runtime2::communication::*;
 use crate::runtime2::component::*;
 use super::component_context::*;
 #[derive(Debug, PartialEq, Eq, Clone, Copy)]
 pub(crate) struct ControlId(u32);
+pub(crate) struct ControlId(pub(crate) u32);
 impl ControlId {
     /// Like other invalid IDs, this one doesn't care any significance, but is
     /// just set at u32::MAX to hopefully bring out bugs sooner.
     fn new_invalid() -> Self {
+    pub(crate) fn new_invalid() -> Self {
         return ControlId(u32::MAX);
+    }
+}
 struct ControlEntry {
     id: ControlId,
     ack_countdown: u32,
     content: ControlContent,
+}
 enum ControlContent {
     PeerChange(ContentPeerChange),
     ScheduleComponent(CompId),
     ClosedPort(PortId),
     UnblockPutWithPorts
+}
 struct ContentPeerChange {
     source_port: PortId,
     source_comp: CompId,
     old_target_port: PortId,
     new_target_port: PortId,
     new_target_comp: CompId,
     schedule_entry_id: ControlId,
+}
 struct ControlClosedPort {
     closed_port: PortId,
     exit_entry_id: Option<ControlId>,
+}
 pub(crate) enum AckAction {
     None,
     SendMessage(CompId, ControlMessage),
     ScheduleComponent(CompId),
     UnblockPutWithPorts,
+}
 /// Handling/sending control messages.
 pub(crate) struct ControlLayer {
     id_counter: ControlId,
     entries: Vec<ControlEntry>,
+}
 impl ControlLayer {
     pub(crate) fn should_reroute(&self, message: &mut Message) -> Option<CompId> {
         // Safety note: rerouting should occur during the time when we're
         // notifying a peer of a new component. During this period that
         // component hasn't been executed yet, so cannot have died yet.
         // FIX @NoDirectHandle
         let target_port = message.target_port();
         if target_port.is_none() {
             return None;
+        }
         let target_port = target_port.unwrap();
         for entry in &self.entries {
             if let ControlContent::PeerChange(entry) = &entry.content {
                 if entry.old_target_port == target_port {
                     message.modify_target_port(entry.new_target_port);
                     return Some(entry.new_target_comp);
+                }
+            }
+        }
         return None;
+    }
     /// Handles an acknowledgement. The returned action must be performed by the
     /// caller. The optionally returned `ControlId` must be used and passed to
     /// `handle_ack` again.
     pub(crate) fn handle_ack(&mut self, entry_id: ControlId, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx) -> (AckAction, Option<ControlId>) {
         let entry_index = self.get_entry_index_by_id(entry_id).unwrap();
         let entry = &mut self.entries[entry_index];
         debug_assert!(entry.ack_countdown > 0);
         entry.ack_countdown -= 1;
         if entry.ack_countdown != 0 {
             return (AckAction::None, None);
+        }
         let entry = self.entries.remove(entry_index);
         // All `Ack`s received, take action based on the kind of entry
         match entry.content {
             ControlContent::PeerChange(content) => {
                 // If change of peer is ack'd. Then we are certain we have
                 // rerouted all of the messages, and the sender's port can now
                 // be unblocked again.
                 let target_comp_id = content.source_comp;
                 let message_to_send = ControlMessage{
                     id: ControlId::new_invalid(),
                     sender_comp_id: comp_ctx.id,
                     target_port_id: Some(content.source_port),
                     content: ControlMessageContent::PortPeerChangedUnblock(
                         content.new_target_port,
                         content.new_target_comp
+                    )
                 };
                 let to_ack = content.schedule_entry_id;
                 return (
                     AckAction::SendMessage(target_comp_id, message_to_send),
                     Some(to_ack)
                 );
             },
             ControlContent::ScheduleComponent(to_schedule) => {
                 // If all change-of-peers are `Ack`d, then we're ready to
                 // schedule the component!
                 return (AckAction::ScheduleComponent(to_schedule), None);
             },
             ControlContent::ClosedPort(closed_port) => {
                 // If a closed port is Ack'd, then we remove the reference to
                 // that component.
                 let port_handle = comp_ctx.get_port_handle(closed_port);
                 debug_assert!(comp_ctx.get_port(port_handle).state.is_closed());
                 comp_ctx.change_port_peer(sched_ctx, port_handle, None);
                 return (AckAction::None, None);
             },
             ControlContent::UnblockPutWithPorts => {
                 return (AckAction::UnblockPutWithPorts, None);
+            }
+        }
+    }
     pub(crate) fn has_acks_remaining(&self) -> bool {
         return !self.entries.is_empty();
+    }
     // -------------------------------------------------------------------------
-    // Port transfer (due to component creation)
     // Port transfer
     // -------------------------------------------------------------------------
     /// Adds an entry that, when completely ack'd, will schedule a component.
     pub(crate) fn add_schedule_entry(&mut self, to_schedule_id: CompId) -> ControlId {
         let entry_id = self.take_id();
         self.entries.push(ControlEntry{
             id: entry_id,
             ack_countdown: 0, // incremented by calls to `add_reroute_entry`
             content: ControlContent::ScheduleComponent(to_schedule_id),
         });
         return entry_id;
+    }
     /// Removes a schedule entry. Only used if the caller preemptively called
     /// `add_schedule_entry`, but ended up not calling `add_reroute_entry`,
     /// hence the `ack_countdown` in the scheduling entry is at 0.
     pub(crate) fn remove_schedule_entry(&mut self, schedule_entry_id: ControlId) {
         let index = self.get_entry_index_by_id(schedule_entry_id).unwrap();
         debug_assert_eq!(self.entries[index].ack_countdown, 0);
         self.entries.remove(index);
     /// Adds an entry that returns the similarly named Ack action
     pub(crate) fn add_unblock_put_with_ports_entry(&mut self) -> ControlId {
         let entry_id = self.take_id();
         self.entries.push(ControlEntry{
             id: entry_id,
             ack_countdown: 0, // incremented by calls to `add_reroute_entry`
             content: ControlContent::UnblockPutWithPorts,
         });
         return entry_id;
+    }
     /// Adds a rerouting entry (used to ensure all messages will end up at a
     /// newly created component). Used when creating a new component.
     pub(crate) fn add_reroute_entry(
         &mut self, creator_comp_id: CompId,
         source_port_id: PortId, source_comp_id: CompId,
         old_target_port_id: PortId, new_target_port_id: PortId, new_comp_id: CompId,
         schedule_entry_id: ControlId,
     ) -> Message {
         let entry_id = self.take_id();
         self.entries.push(ControlEntry{
             id: entry_id,
             ack_countdown: 1,
             content: ControlContent::PeerChange(ContentPeerChange{
                 source_port: source_port_id,
                 source_comp: source_comp_id,
                 old_target_port: old_target_port_id,
                 new_target_port: new_target_port_id,
                 new_target_comp: new_comp_id,
                 schedule_entry_id,
             }),
         });
         // increment counter on schedule entry
         let entry_index = self.get_entry_index_by_id(schedule_entry_id).unwrap();
         self.entries[entry_index].ack_countdown += 1;
         return Message::Control(ControlMessage{
             id: entry_id,
             sender_comp_id: creator_comp_id,
             target_port_id: Some(source_port_id),
             content: ControlMessageContent::PortPeerChangedBlock
         })
+    }
     /// Creates a PortPeerChanged message (and increments ack-counter on a
     /// pre-created control entry) that is used as a preliminary step before
     /// transferring a port over a channel.
     pub(crate) fn create_port_transfer_message(
         &mut self, associated_control_id: ControlId,
         sender_comp_id: CompId, target_port_id: PortId
     ) -> Message {
         let entry_index = self.get_entry_index_by_id(associated_control_id).unwrap();
         self.entries[entry_index].ack_countdown += 1;
         return Message::Control(ControlMessage{
             id: associated_control_id,
             sender_comp_id,
             target_port_id: Some(target_port_id),
             content: ControlMessageContent::PortPeerChangedBlock
         })
+    }
     // -------------------------------------------------------------------------
     // Blocking, unblocking, and closing ports
     // -------------------------------------------------------------------------
     pub(crate) fn has_close_port_entry(&self, port_handle: LocalPortHandle, comp_ctx: &CompCtx) -> Option<ControlId> {
         let port = comp_ctx.get_port(port_handle);
         let port_id = port.self_id;
         for entry in self.entries.iter() {
             if let ControlContent::ClosedPort(entry_port_id) = &entry.content {
                 if *entry_port_id == port_id {
                     return Some(entry.id);
+                }
+            }
+        }
         return None;
+    }
     /// Initiates the control message procedures for closing a port. Caller must
     /// make sure that the port state has already been set to `Closed`.
     pub(crate) fn initiate_port_closing(&mut self, port_handle: LocalPortHandle, exit_inside_sync: bool, comp_ctx: &CompCtx) -> (LocalPeerHandle, ControlMessage) {
         let port = comp_ctx.get_port(port_handle);
         let peer_port_id = port.peer_port_id;
         debug_assert!(port.state.is_closed());
         // Construct the port-closing entry
         let entry_id = self.take_id();
         self.entries.push(ControlEntry{
             id: entry_id,
             ack_countdown: 1,
             content: ControlContent::ClosedPort(port.self_id),
         });
         // Return the messages notifying peer of the closed port
         let peer_handle = comp_ctx.get_peer_handle(port.peer_comp_id);
         return (
             peer_handle,
             ControlMessage{
                 id: entry_id,
                 sender_comp_id: comp_ctx.id,
                 target_port_id: Some(peer_port_id),
                 content: ControlMessageContent::ClosePort(ControlMessageClosePort{
                     closed_in_sync_round: exit_inside_sync,
                 }),
+            }
         );
+    }
     /// Generates the control message used to indicate to a peer that a port
     /// should be blocked (expects the caller to have set the port's state to
     /// blocked).
     pub(crate) fn initiate_port_blocking(&mut self, comp_ctx: &CompCtx, port_handle: LocalPortHandle) -> (LocalPeerHandle, ControlMessage) {
         let port_info = comp_ctx.get_port(port_handle);
         debug_assert_eq!(port_info.kind, PortKind::Getter); // because we're telling the putter to block
         debug_assert!(port_info.state.is_set(PortStateFlag::BlockedDueToFullBuffers)); // contract with caller
         let peer_port_id = port_info.peer_port_id;
         let peer_comp_id = port_info.peer_comp_id;
         let peer_handle = comp_ctx.get_peer_handle(peer_comp_id);
         return (
             peer_handle,
             ControlMessage{
                 id: ControlId::new_invalid(),
                 sender_comp_id: comp_ctx.id,
                 target_port_id: Some(peer_port_id),
                 content: ControlMessageContent::BlockPort,
+            }
         );
+    }
     /// Generates a messages used to indicate to a peer that a port should be
     /// unblocked again.
     pub(crate) fn cancel_port_blocking(&mut self, comp_ctx: &CompCtx, port_handle: LocalPortHandle) -> (LocalPeerHandle, ControlMessage) {
         let port_info = comp_ctx.get_port(port_handle);
         debug_assert_eq!(port_info.kind, PortKind::Getter); // because we're initiating the unblocking
         let peer_handle = comp_ctx.get_peer_handle(port_info.peer_comp_id);
         return (
             peer_handle,
             ControlMessage{
                 id: ControlId::new_invalid(),
                 sender_comp_id: comp_ctx.id,
                 target_port_id: Some(port_info.peer_port_id),
                 content: ControlMessageContent::UnblockPort,
+            }
         );
+    }
     // -------------------------------------------------------------------------
     // Internal utilities
     // -------------------------------------------------------------------------
     fn take_id(&mut self) -> ControlId {
         let id = self.id_counter;
         self.id_counter.0 = self.id_counter.0.wrapping_add(1);
         return id;
+    }
     fn get_entry_index_by_id(&self, entry_id: ControlId) -> Option<usize> {
         for (index, entry) in self.entries.iter().enumerate() {
             if entry.id == entry_id {
                 return Some(index);
+            }
+        }
         return None;
+    }
+}
 impl Default for ControlLayer {
     fn default() -> Self {
         return ControlLayer{
             id_counter: ControlId(0),
             entries: Vec::new(),
+        }
+    }
+}
@@ \ No newline at end of file @@

src/runtime2/component/mod.rs

➞

Show inline comments

 mod component_pdl;
 mod component_context;
 mod control_layer;
 mod consensus;
 mod component;
 mod component_random;
 mod component_internet;
 pub(crate) use component::{Component, CompScheduling};
 pub(crate) use component_pdl::{CompPDL};
 pub(crate) use component_context::{CompCtx, PortInstruction};
+pub(crate) use component_context::{CompCtx, PortInstruction, PortKind, PortState, PortStateFlag};
 pub(crate) use control_layer::{ControlId};
 use super::scheduler::*;
 use super::runtime::*;
 /// If the component is sleeping, then that flag will be atomically set to
 /// false. If we're the ones that made that happen then we add it to the work
 /// queue.
 pub(crate) fn wake_up_if_sleeping(runtime: &RuntimeInner, comp_id: CompId, handle: &CompHandle) {
     use std::sync::atomic::Ordering;
     let should_wake_up = handle.sleeping
         .compare_exchange(true, false, Ordering::AcqRel, Ordering::Acquire)
         .is_ok();
     if should_wake_up {
         let comp_key = unsafe{ comp_id.upgrade() };
         runtime.enqueue_work(comp_key);
+    }
+}
@@ \ No newline at end of file @@

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

0 comments (0 inline, 0 general)