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

Changeset - 7d01f1245b7c

Parent rev.

Child rev.

[Not reviewed]

0 12 0

mh - 4 years ago 2021-10-20 20:26:23
contact@maxhenger.nl

Everything compiles again, pending restructuring of shared runtime objects

10 files changed:

src/collections/raw_vec.rs

src/protocol/parser/type_table.rs

src/runtime2/connector.rs

src/runtime2/global_store.rs

src/runtime2/inbox.rs

src/runtime2/messages.rs

src/runtime2/mod.rs

114

src/runtime2/native.rs

src/runtime2/port.rs

src/runtime2/runtime.rs

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

0 comments (0 inline, 0 general)

src/collections/raw_vec.rs

➞

Show inline comments

 use std::{mem, ptr, cmp};
 use std::alloc::{Layout, alloc, dealloc};
 #[derive(Debug)]
 enum AllocError {
     CapacityOverflow,
+}
 /// Generic raw vector. It has a base pointer, a capacity and a length. Basic
 /// operations are supported, but the user of the structure is responsible for
 /// ensuring that no illegal mutable access occurs.
 /// A lot of the logic is simply stolen from the std lib. The destructor will
 /// free the backing memory, but will not run any destructors.
 pub struct RawVec<T: Sized> {
     base: *mut T,
     cap: usize,
     len: usize,
+}
 impl<T: Sized> RawVec<T> {
     const T_ALIGNMENT: usize = mem::align_of::<T>();
     const T_SIZE: usize = mem::size_of::<T>();
     const GROWTH_RATE: usize = 2;
     pub fn new() -> Self {
         Self{
             base: ptr::null_mut(),
             cap: 0,
             len: 0,
+        }
+    }
     pub fn with_capacity(capacity: usize) -> Self {
         // Could be done a bit more efficiently
         let mut result = Self::new();
         result.ensure_space(capacity);
+        result.ensure_space(capacity).unwrap();
         return result;
+    }
     pub unsafe fn get(&self, idx: usize) -> *const T {
         debug_assert!(idx < self.len);
         return self.base.add(idx);
+    }
     pub unsafe fn get_mut(&self, idx: usize) -> *mut T {
         debug_assert!(idx < self.len);
         return self.base.add(idx);
+    }
     pub fn push(&mut self, item: T) {
         self.ensure_space(1);
+        self.ensure_space(1).unwrap();
         unsafe {
             let target = self.base.add(self.len);
             std::ptr::write(target, item);
             self.len += 1;
+        }
+    }
     pub fn len(&self) -> usize {
         return self.len;
+    }
     fn ensure_space(&mut self, additional: usize) -> Result<(), AllocError>{
         debug_assert!(Self::T_SIZE != 0);
         debug_assert!(self.cap >= self.len);
         if self.cap - self.len < additional {
             // Need to resize. Note that due to all checked conditions we have
             // that new_cap >= 1.
             debug_assert!(additional > 0);
             let new_cap = self.len.checked_add(additional).unwrap();
             let new_cap = cmp::max(new_cap, self.cap * Self::GROWTH_RATE);
             let layout = Layout::array::<T>(new_cap)
                 .map_err(|_| AllocError::CapacityOverflow)?;
             debug_assert_eq!(new_cap * Self::T_SIZE, layout.size());
             unsafe {
                 // Allocate new storage, transfer bits, deallocate old store
                 let new_base = alloc(layout);
                 if self.cap > 0 {
                     let old_base = self.base as *mut u8;
                     let (old_size, old_layout) = self.current_layout();
                     ptr::copy_nonoverlapping(new_base, old_base, old_size);
                     dealloc(old_base, old_layout);
+                }
                 self.base = new_base;
+                self.base = new_base as *mut T;
                 self.cap = new_cap;
+            }
         } // else: still enough space
         return Ok(());
+    }
     #[inline]
     fn current_layout(&self) -> (usize, Layout) {
         debug_assert!(Self::T_SIZE > 0);
         let old_size = self.cap * Self::T_SIZE;
         unsafe {
             return (
                 old_size,
                 Layout::from_size_align_unchecked(old_size, Self::T_ALIGNMENT)
             );
+        }
+    }
+}
 impl<T: Sized> Drop for RawVec<T> {
     fn drop(&mut self) {
         if self.cap > 0 {
             debug_assert!(!self.base.is_null());
             let (_, layout) = self.current_layout();
             unsafe {
                 dealloc(self.base, layout);
+                dealloc(self.base as *mut u8, layout);
                 if cfg!(debug_assertions) {
                     self.base = ptr::null_mut();
+                }
+            }
+        }
+    }
+}
@@ \ No newline at end of file @@

src/protocol/parser/type_table.rs

➞

Show inline comments

 /**
  * type_table.rs
+ *
  * The type table is a lookup from AST definition (which contains just what the
  * programmer typed) to a type with additional information computed (e.g. the
  * byte size and offsets of struct members). The type table should be considered
  * the authoritative source of information on types by the compiler (not the
  * AST itself!).
+ *
  * The type table operates in two modes: one is where we just look up the type,
  * check its fields for correctness and mark whether it is polymorphic or not.
  * The second one is where we compute byte sizes, alignment and offsets.
+ *
  * The basic algorithm for type resolving and computing byte sizes is to
  * recursively try to lay out each member type of a particular type. This is
  * done in a stack-like fashion, where each embedded type pushes a breadcrumb
  * unto the stack. We may discover a cycle in embedded types (we call this a
  * "type loop"). After which the type table attempts to break the type loop by
  * making specific types heap-allocated. Upon doing so we know their size
  * because their stack-size is now based on pointers. Hence breaking the type
  * loop required for computing the byte size of types.
+ *
  * The reason for these type shenanigans is because PDL is a value-based
  * language, but we would still like to be able to express recursively defined
  * types like trees or linked lists. Hence we need to insert pointers somewhere
  * to break these cycles.
+ *
  * We will insert these pointers into the variants of unions. However note that
  * we can only compute the stack size of a union until we've looked at *all*
  * variants. Hence we perform an initial pass where we detect type loops, a
  * second pass where we compute the stack sizes of everything, and a third pass
  * where we actually compute the size of the heap allocations for unions.
+ *
  * As a final bit of global documentation: non-polymorphic types will always
  * have one "monomorph" entry. This contains the non-polymorphic type's memory
  * layout.
  */
 use std::fmt::{Formatter, Result as FmtResult};
 use std::collections::HashMap;
 use crate::protocol::ast::*;
 use crate::protocol::parser::symbol_table::SymbolScope;
 use crate::protocol::input_source::ParseError;
 use crate::protocol::parser::*;
 //------------------------------------------------------------------------------
 // Defined Types
 //------------------------------------------------------------------------------
 #[derive(Copy, Clone, PartialEq, Eq)]
 pub enum TypeClass {
     Enum,
     Union,
     Struct,
     Function,
     Component
+}
 impl TypeClass {
     pub(crate) fn display_name(&self) -> &'static str {
         match self {
             TypeClass::Enum => "enum",
             TypeClass::Union => "union",
             TypeClass::Struct => "struct",
             TypeClass::Function => "function",
             TypeClass::Component => "component",
+        }
+    }
     pub(crate) fn is_data_type(&self) -> bool {
         match self {
             TypeClass::Enum | TypeClass::Union | TypeClass::Struct => true,
             TypeClass::Function | TypeClass::Component => false,
+        }
+    }
+}
 impl std::fmt::Display for TypeClass {
     fn fmt(&self, f: &mut Formatter<'_>) -> FmtResult {
         write!(f, "{}", self.display_name())
+    }
+}
 /// Struct wrapping around a potentially polymorphic type. If the type does not
 /// have any polymorphic arguments then it will not have any monomorphs and
 /// `is_polymorph` will be set to `false`. A type with polymorphic arguments
 /// only has `is_polymorph` set to `true` if the polymorphic arguments actually
 /// appear in the types associated types (function return argument, struct
 /// field, enum variant, etc.). Otherwise the polymorphic argument is just a
 /// marker and does not influence the bytesize of the type.
 pub struct DefinedType {
     pub(crate) ast_root: RootId,
     pub(crate) ast_definition: DefinitionId,
     pub(crate) definition: DefinedTypeVariant,
     pub(crate) poly_vars: Vec<PolymorphicVariable>,
     pub(crate) is_polymorph: bool,
+}
 impl DefinedType {
     /// Returns the number of monomorphs that are instantiated. Remember that
     /// during the type loop detection, and the memory layout phase we will
     /// pre-allocate monomorphs which are not yet fully laid out in memory.
     pub(crate) fn num_monomorphs(&self) -> usize {
         use DefinedTypeVariant as DTV;
         match &self.definition {
             DTV::Enum(def) => def.monomorphs.len(),
             DTV::Union(def) => def.monomorphs.len(),
             DTV::Struct(def) => def.monomorphs.len(),
             DTV::Function(_) | DTV::Component(_) => unreachable!(),
+        }
+    }
     /// Returns the index at which a monomorph occurs. Will only check the
     /// polymorphic arguments that are in use (none of the, in rust lingo,
     /// phantom types). If the type is not polymorphic and its memory has been
     /// layed out, then this will always return `Some(0)`.
     pub(crate) fn get_monomorph_index(&self, concrete_type: &ConcreteType) -> Option<usize> {
         use DefinedTypeVariant as DTV;
         // Helper to compare two types, while disregarding the polymorphic
         // variables that are not in use.
         let concrete_types_match = |type_a: &ConcreteType, type_b: &ConcreteType, check_if_poly_var_is_used: bool| -> bool {
             let mut a_iter = type_a.embedded_iter(0).enumerate();
             let mut b_iter = type_b.embedded_iter(0);
             while let Some((section_idx, a_section)) = a_iter.next() {
                 let b_section = b_iter.next().unwrap();
                 if check_if_poly_var_is_used && !self.poly_vars[section_idx].is_in_use {
                     continue;
+                }
                 if a_section != b_section {
                     return false;
+                }
+            }
             return true;
         };
         // Check check if type is polymorphic to some degree at all
         if cfg!(debug_assertions) {
             if let ConcreteTypePart::Instance(definition_id, num_poly_args) = concrete_type.parts[0] {
                 assert_eq!(definition_id, self.ast_definition);
                 assert_eq!(num_poly_args as usize, self.poly_vars.len());
             } else {
                 assert!(false, "concrete type {:?} is not a user-defined type", concrete_type);
+            }
+        }
         match &self.definition {
             DTV::Enum(definition) => {
                 // Special case, enum is never a "true polymorph"
                 debug_assert!(!self.is_polymorph);
                 if definition.monomorphs.is_empty() {
                     return None
                 } else {
                     return Some(0)
+                }
             },
             DTV::Union(definition) => {
                 for (monomorph_idx, monomorph) in definition.monomorphs.iter().enumerate() {
                     if concrete_types_match(&monomorph.concrete_type, concrete_type, true) {
                         return Some(monomorph_idx);
+                    }
+                }
             },
             DTV::Struct(definition) => {
                 for (monomorph_idx, monomorph) in definition.monomorphs.iter().enumerate() {
                     if concrete_types_match(&monomorph.concrete_type, concrete_type, true) {
                         return Some(monomorph_idx);
+                    }
+                }
             },
             DTV::Function(definition) => {
                 for (monomorph_idx, monomorph) in definition.monomorphs.iter().enumerate() {
                     if concrete_types_match(&monomorph.concrete_type, concrete_type, false) {
                         return Some(monomorph_idx)
+                    }
+                }
+            }
             DTV::Component(definition) => {
                 for (monomorph_idx, monomorph) in definition.monomorphs.iter().enumerate() {
                     if concrete_types_match(&monomorph.concrete_type, concrete_type, false) {
                         return Some(monomorph_idx)
+                    }
+                }
+            }
+        }
         // Nothing matched
         return None;
+    }
     /// Retrieves size and alignment of the particular type's monomorph if it
     /// has been layed out in memory.
     pub(crate) fn get_monomorph_size_alignment(&self, idx: usize) -> Option<(usize, usize)> {
         use DefinedTypeVariant as DTV;
         let (size, alignment) = match &self.definition {
             DTV::Enum(def) => {
                 debug_assert!(idx == 0);
                 (def.size, def.alignment)
             },
             DTV::Union(def) => {
                 let monomorph = &def.monomorphs[idx];
                 (monomorph.stack_size, monomorph.stack_alignment)
             },
             DTV::Struct(def) => {
                 let monomorph = &def.monomorphs[idx];
                 (monomorph.size, monomorph.alignment)
             },
             DTV::Function(_) | DTV::Component(_) => {
                 // Type table should never be able to arrive here during layout
                 // of types. Types may only contain function prototypes.
                 unreachable!("retrieving size and alignment of procedure type");
+            }
         };
         if size == 0 && alignment == 0 {
             // The "marker" for when the type has not been layed out yet. Even
             // for zero-size types we will set alignment to `1` to simplify
             // alignment calculations.
             return None;
         } else {
             return Some((size, alignment));
+        }
+    }
+}
 pub enum DefinedTypeVariant {
     Enum(EnumType),
     Union(UnionType),
     Struct(StructType),
     Function(FunctionType),
     Component(ComponentType)
+}
 impl DefinedTypeVariant {
     pub(crate) fn type_class(&self) -> TypeClass {
         match self {
             DefinedTypeVariant::Enum(_) => TypeClass::Enum,
             DefinedTypeVariant::Union(_) => TypeClass::Union,
             DefinedTypeVariant::Struct(_) => TypeClass::Struct,
             DefinedTypeVariant::Function(_) => TypeClass::Function,
             DefinedTypeVariant::Component(_) => TypeClass::Component
+        }
+    }
     pub(crate) fn as_struct(&self) -> &StructType {
         match self {
             DefinedTypeVariant::Struct(v) => v,
             _ => unreachable!("Cannot convert {} to struct variant", self.type_class())
+        }
+    }
     pub(crate) fn as_struct_mut(&mut self) -> &mut StructType {
         match self {
             DefinedTypeVariant::Struct(v) => v,
             _ => unreachable!("Cannot convert {} to struct variant", self.type_class())
+        }
+    }
     pub(crate) fn as_enum(&self) -> &EnumType {
         match self {
             DefinedTypeVariant::Enum(v) => v,
             _ => unreachable!("Cannot convert {} to enum variant", self.type_class())
+        }
+    }
     pub(crate) fn as_enum_mut(&mut self) -> &mut EnumType {
         match self {
             DefinedTypeVariant::Enum(v) => v,
             _ => unreachable!("Cannot convert {} to enum variant", self.type_class())
+        }
+    }
     pub(crate) fn as_union(&self) -> &UnionType {
         match self {
             DefinedTypeVariant::Union(v) => v,
             _ => unreachable!("Cannot convert {} to union variant", self.type_class())
+        }
+    }
     pub(crate) fn as_union_mut(&mut self) -> &mut UnionType {
         match self {
             DefinedTypeVariant::Union(v) => v,
             _ => unreachable!("Cannot convert {} to union variant", self.type_class())
+        }
+    }
     pub(crate) fn procedure_monomorphs(&self) -> &Vec<ProcedureMonomorph> {
         use DefinedTypeVariant::*;
         match self {
             Function(v) => &v.monomorphs,
             Component(v) => &v.monomorphs,
             _ => unreachable!("cannot get procedure monomorphs from {}", self.type_class()),
+        }
+    }
     pub(crate) fn procedure_monomorphs_mut(&mut self) -> &mut Vec<ProcedureMonomorph> {
         use DefinedTypeVariant::*;
         match self {
             Function(v) => &mut v.monomorphs,
             Component(v) => &mut v.monomorphs,
             _ => unreachable!("cannot get procedure monomorphs from {}", self.type_class()),
+        }
+    }
+}
 pub struct PolymorphicVariable {
     identifier: Identifier,
     is_in_use: bool, // a polymorphic argument may be defined, but not used by the type definition
+}
 /// Data associated with a monomorphized procedure type. Has the wrong name,
 /// because it will also be used to store expression data for a non-polymorphic
 /// procedure. (in that case, there will only ever be one)
 pub struct ProcedureMonomorph {
     // Expression data for one particular monomorph
     pub concrete_type: ConcreteType,
     pub arg_types: Vec<ConcreteType>,
     pub expr_data: Vec<MonomorphExpression>,
+}
 /// `EnumType` is the classical C/C++ enum type. It has various variants with
 /// an assigned integer value. The integer values may be user-defined,
 /// compiler-defined, or a mix of the two. If a user assigns the same enum
 /// value multiple times, we assume the user is an expert and we consider both
 /// variants to be equal to one another.
 pub struct EnumType {
     pub variants: Vec<EnumVariant>,
     pub monomorphs: Vec<EnumMonomorph>,
     pub minimum_tag_value: i64,
     pub maximum_tag_value: i64,
     pub tag_type: ConcreteType,
     pub size: usize,
     pub alignment: usize,
+}
 // TODO: Also support maximum u64 value
 pub struct EnumVariant {
     pub identifier: Identifier,
     pub value: i64,
+}
 pub struct EnumMonomorph {
     pub concrete_type: ConcreteType,
+}
 /// `UnionType` is the algebraic datatype (or sum type, or discriminated union).
 /// A value is an element of the union, identified by its tag, and may contain
 /// a single subtype.
 /// For potentially infinite types (i.e. a tree, or a linked list) only unions
 /// can break the infinite cycle. So when we lay out these unions in memory we
 /// will reserve enough space on the stack for all union variants that do not
 /// cause "type loops" (i.e. a union `A` with a variant containing a struct
 /// `B`). And we will reserve enough space on the heap (and store a pointer in
 /// the union) for all variants which do cause type loops (i.e. a union `A`
 /// with a variant to a struct `B` that contains the union `A` again).
 pub struct UnionType {
     pub variants: Vec<UnionVariant>,
     pub monomorphs: Vec<UnionMonomorph>,
     pub tag_type: ConcreteType,
     pub tag_size: usize,
+}
 pub struct UnionVariant {
     pub identifier: Identifier,
     pub embedded: Vec<ParserType>, // zero-length does not have embedded values
     pub tag_value: i64,
+}
 pub struct UnionMonomorph {
     pub concrete_type: ConcreteType,
     pub variants: Vec<UnionMonomorphVariant>,
     // stack_size is the size of the union on the stack, includes the tag
     pub stack_size: usize,
     pub stack_alignment: usize,
     // heap_size contains the allocated size of the union in the case it
     // is used to break a type loop. If it is 0, then it doesn't require
     // allocation and lives entirely on the stack.
     pub heap_size: usize,
     pub heap_alignment: usize,
+}
 pub struct UnionMonomorphVariant {
     pub lives_on_heap: bool,
     pub embedded: Vec<UnionMonomorphEmbedded>,
+}
 pub struct UnionMonomorphEmbedded {
     pub concrete_type: ConcreteType,
     // Note that the meaning of the offset (and alignment) depend on whether or
     // not the variant lives on the stack/heap. If it lives on the stack then
     // they refer to the offset from the start of the union value (so the first
     // embedded type lives at a non-zero offset, because the union tag sits in
     // the front). If it lives on the heap then it refers to the offset from the
     // allocated memory region (so the first embedded type lives at a 0 offset).
     pub size: usize,
     pub alignment: usize,
     pub offset: usize,
+}
 /// `StructType` is a generic C-like struct type (or record type, or product
 /// type) type.
 pub struct StructType {
     pub fields: Vec<StructField>,
     pub monomorphs: Vec<StructMonomorph>,
+}
 pub struct StructField {
     pub identifier: Identifier,
     pub parser_type: ParserType,
+}
 pub struct StructMonomorph {
     pub concrete_type: ConcreteType,
     pub fields: Vec<StructMonomorphField>,
     pub size: usize,
     pub alignment: usize,
+}
 pub struct StructMonomorphField {
     pub concrete_type: ConcreteType,
     pub size: usize,
     pub alignment: usize,
     pub offset: usize,
+}
 /// `FunctionType` is what you expect it to be: a particular function's
 /// signature.
 pub struct FunctionType {
     pub return_types: Vec<ParserType>,
     pub arguments: Vec<FunctionArgument>,
     pub monomorphs: Vec<ProcedureMonomorph>,
+}
 pub struct ComponentType {
     pub variant: ComponentVariant,
     pub arguments: Vec<FunctionArgument>,
     pub monomorphs: Vec<ProcedureMonomorph>
+}
 pub struct FunctionArgument {
     identifier: Identifier,
     parser_type: ParserType,
+}
 /// Represents the data associated with a single expression after type inference
 /// for a monomorph (or just the normal expression types, if dealing with a
 /// non-polymorphic function/component).
 pub struct MonomorphExpression {
     // The output type of the expression. Note that for a function it is not the
     // function's signature but its return type
     pub(crate) expr_type: ConcreteType,
     // Has multiple meanings: the field index for select expressions, the
     // monomorph index for polymorphic function calls or literals. Negative
     // values are never used, but used to catch programming errors.
     pub(crate) field_or_monomorph_idx: i32,
+}
 //------------------------------------------------------------------------------
 // Type table
 //------------------------------------------------------------------------------
 // Programmer note: keep this struct free of dynamically allocated memory
 #[derive(Clone)]
 struct TypeLoopBreadcrumb {
     definition_id: DefinitionId,
     monomorph_idx: usize,
     next_member: usize,
     next_embedded: usize, // for unions, the index into the variant's embedded types
+}
 #[derive(Clone)]
 struct MemoryBreadcrumb {
     definition_id: DefinitionId,
     monomorph_idx: usize,
     next_member: usize,
     next_embedded: usize,
     first_size_alignment_idx: usize,
+}
 #[derive(Debug, PartialEq, Eq)]
 enum TypeLoopResult {
     TypeExists,
     PushBreadcrumb(DefinitionId, ConcreteType),
     TypeLoop(usize), // index into vec of breadcrumbs at which the type matched
+}
 enum MemoryLayoutResult {
     TypeExists(usize, usize), // (size, alignment)
     PushBreadcrumb(MemoryBreadcrumb),
+}
 // TODO: @Optimize, initial memory-unoptimized implementation
 struct TypeLoopEntry {
     definition_id: DefinitionId,
     monomorph_idx: usize,
     is_union: bool,
+}
 struct TypeLoop {
     members: Vec<TypeLoopEntry>
+}
 pub struct TypeTable {
     /// Lookup from AST DefinitionId to a defined type. Considering possible
     /// polymorphs is done inside the `DefinedType` struct.
     lookup: HashMap<DefinitionId, DefinedType>,
     /// Breadcrumbs left behind while trying to find type loops. Also used to
     /// determine sizes of types when all type loops are detected.
     type_loop_breadcrumbs: Vec<TypeLoopBreadcrumb>,
     type_loops: Vec<TypeLoop>,
     /// Stores all encountered types during type loop detection. Used afterwards
     /// to iterate over all types in order to compute size/alignment.
     encountered_types: Vec<TypeLoopEntry>,
     /// Breadcrumbs and temporary storage during memory layout computation.
     memory_layout_breadcrumbs: Vec<MemoryBreadcrumb>,
     size_alignment_stack: Vec<(usize, usize)>,
+}
 impl TypeTable {
     /// Construct a new type table without any resolved types.
     pub(crate) fn new() -> Self {
         Self{
             lookup: HashMap::new(),
             type_loop_breadcrumbs: Vec::with_capacity(32),
             type_loops: Vec::with_capacity(8),
             encountered_types: Vec::with_capacity(32),
             memory_layout_breadcrumbs: Vec::with_capacity(32),
             size_alignment_stack: Vec::with_capacity(64),
+        }
+    }
     /// Iterates over all defined types (polymorphic and non-polymorphic) and
     /// add their types in two passes. In the first pass we will just add the
     /// base types (we will not consider monomorphs, and we will not compute
     /// byte sizes). In the second pass we will compute byte sizes of
     /// non-polymorphic types, and potentially the monomorphs that are embedded
     /// in those types.
     pub(crate) fn build_base_types(&mut self, modules: &mut [Module], ctx: &mut PassCtx) -> Result<(), ParseError> {
         // Make sure we're allowed to cast root_id to index into ctx.modules
         debug_assert!(modules.iter().all(|m| m.phase >= ModuleCompilationPhase::DefinitionsParsed));
         debug_assert!(self.lookup.is_empty());
         if cfg!(debug_assertions) {
             for (index, module) in modules.iter().enumerate() {
                 debug_assert_eq!(index, module.root_id.index as usize);
+            }
+        }
         // Use context to guess hashmap size of the base types
         let reserve_size = ctx.heap.definitions.len();
         self.lookup.reserve(reserve_size);
         // Resolve all base types
         for definition_idx in 0..ctx.heap.definitions.len() {
             let definition_id = ctx.heap.definitions.get_id(definition_idx);
             let definition = &ctx.heap[definition_id];
             match definition {
                 Definition::Enum(_) => self.build_base_enum_definition(modules, ctx, definition_id)?,
                 Definition::Union(_) => self.build_base_union_definition(modules, ctx, definition_id)?,
                 Definition::Struct(_) => self.build_base_struct_definition(modules, ctx, definition_id)?,
                 Definition::Function(_) => self.build_base_function_definition(modules, ctx, definition_id)?,
                 Definition::Component(_) => self.build_base_component_definition(modules, ctx, definition_id)?,
+            }
+        }
         debug_assert_eq!(self.lookup.len(), reserve_size, "mismatch in reserved size of type table"); // NOTE: Temp fix for builtin functions
         for module in modules.iter_mut() {
             module.phase = ModuleCompilationPhase::TypesAddedToTable;
+        }
         // Go through all types again, lay out all types that are not
         // polymorphic. This might cause us to lay out types that are monomorphs
         // of polymorphic types.
         for definition_idx in 0..ctx.heap.definitions.len() {
             let definition_id = ctx.heap.definitions.get_id(definition_idx);
             let poly_type = self.lookup.get(&definition_id).unwrap();
             // Here we explicitly want to instantiate types which have no
             // polymorphic arguments (even if it has phantom polymorphic
             // arguments) because otherwise the user will see very weird
             // error messages.
             if poly_type.definition.type_class().is_data_type() && poly_type.poly_vars.is_empty() && poly_type.num_monomorphs() == 0 {
                 self.detect_and_resolve_type_loops_for(
                     modules, ctx.heap,
                     ConcreteType{
                         parts: vec![ConcreteTypePart::Instance(definition_id, 0)]
                     },
                 )?;
                 self.lay_out_memory_for_encountered_types(ctx.arch);
+            }
+        }
         Ok(())
+    }
     /// Retrieves base definition from type table. We must be able to retrieve
     /// it as we resolve all base types upon type table construction (for now).
     /// However, in the future we might do on-demand type resolving, so return
     /// an option anyway
     pub(crate) fn get_base_definition(&self, definition_id: &DefinitionId) -> Option<&DefinedType> {
         self.lookup.get(&definition_id)
+    }
     /// Returns the index into the monomorph type array if the procedure type
     /// already has a (reserved) monomorph.
     pub(crate) fn get_procedure_monomorph_index(&self, definition_id: &DefinitionId, types: &ConcreteType) -> Option<i32> {
         let def = self.lookup.get(definition_id).unwrap();
         let monos = def.definition.procedure_monomorphs();
         return monos.iter()
             .position(|v| v.concrete_type == *types)
             .map(|v| v as i32);
+    }
     /// Returns a mutable reference to a procedure's monomorph expression data.
     /// Used by typechecker to fill in previously reserved type information
     pub(crate) fn get_procedure_expression_data_mut(&mut self, definition_id: &DefinitionId, monomorph_idx: i32) -> &mut ProcedureMonomorph {
         debug_assert!(monomorph_idx >= 0);
         let def = self.lookup.get_mut(definition_id).unwrap();
         let monomorphs = def.definition.procedure_monomorphs_mut();
         return &mut monomorphs[monomorph_idx as usize];
+    }
     pub(crate) fn get_procedure_expression_data(&self, definition_id: &DefinitionId, monomorph_idx: i32) -> &ProcedureMonomorph {
         debug_assert!(monomorph_idx >= 0);
         let def = self.lookup.get(definition_id).unwrap();
         let monomorphs = def.definition.procedure_monomorphs();
         return &monomorphs[monomorph_idx as usize];
+    }
     /// Reserves space for a monomorph of a polymorphic procedure. The index
     /// will point into a (reserved) slot of the array of expression types. The
     /// monomorph may NOT exist yet (because the reservation implies that we're
     /// going to be performing typechecking on it, and we don't want to
     /// check the same monomorph twice)
     pub(crate) fn reserve_procedure_monomorph_index(&mut self, definition_id: &DefinitionId, concrete_type: ConcreteType) -> i32 {
         let def = self.lookup.get_mut(definition_id).unwrap();
         let mono_types = def.definition.procedure_monomorphs_mut();
         debug_assert!(def.is_polymorph == (concrete_type.parts.len() != 1));
         debug_assert!(!mono_types.iter().any(|v| v.concrete_type == concrete_type));
         let mono_idx = mono_types.len();
         mono_types.push(ProcedureMonomorph{
             concrete_type,
             arg_types: Vec::new(),
             expr_data: Vec::new(),
         });
         return mono_idx as i32;
+    }
     /// Adds a datatype polymorph to the type table. Will not add the
     /// monomorph if it is already present, or if the type's polymorphic
     /// variables are all unused.
     /// TODO: Fix signature
     pub(crate) fn add_data_monomorph(
         &mut self, modules: &[Module], heap: &Heap, arch: &TargetArch, definition_id: DefinitionId, concrete_type: ConcreteType
     ) -> Result<i32, ParseError> {
         debug_assert_eq!(definition_id, get_concrete_type_definition(&concrete_type));
         // Check if the monomorph already exists
         let poly_type = self.lookup.get_mut(&definition_id).unwrap();
         if let Some(idx) = poly_type.get_monomorph_index(&concrete_type) {
             return Ok(idx as i32);
+        }
         // Doesn't exist, so instantiate a monomorph and determine its memory
         // layout.
         self.detect_and_resolve_type_loops_for(modules, heap, concrete_type)?;
         debug_assert_eq!(self.encountered_types[0].definition_id, definition_id);
         let mono_idx = self.encountered_types[0].monomorph_idx;
         self.lay_out_memory_for_encountered_types(arch);
         return Ok(mono_idx as i32);
+    }
     //--------------------------------------------------------------------------
     // Building base types
     //--------------------------------------------------------------------------
     /// Builds the base type for an enum. Will not compute byte sizes
     fn build_base_enum_definition(&mut self, modules: &[Module], ctx: &mut PassCtx, definition_id: DefinitionId) -> Result<(), ParseError> {
         debug_assert!(!self.lookup.contains_key(&definition_id), "base enum already built");
         let definition = ctx.heap[definition_id].as_enum();
         let root_id = definition.defined_in;
         // Determine enum variants
         let mut enum_value = -1;
         let mut variants = Vec::with_capacity(definition.variants.len());
         for variant in &definition.variants {
             if enum_value == i64::MAX {
                 let source = &modules[definition.defined_in.index as usize].source;
                 return Err(ParseError::new_error_str_at_span(
                     source, variant.identifier.span,
                     "this enum variant has an integer value that is too large"
                 ));
+            }
             enum_value += 1;
             if let EnumVariantValue::Integer(explicit_value) = variant.value {
                 enum_value = explicit_value;
+            }
             variants.push(EnumVariant{
                 identifier: variant.identifier.clone(),
                 value: enum_value,
             });
+        }
         // Determine tag size
         let mut min_enum_value = 0;
         let mut max_enum_value = 0;
         if !variants.is_empty() {
             min_enum_value = variants[0].value;
             max_enum_value = variants[0].value;
             for variant in variants.iter().skip(1) {
                 min_enum_value = min_enum_value.min(variant.value);
                 max_enum_value = max_enum_value.max(variant.value);
+            }
+        }
         let (tag_type, size_and_alignment) = Self::variant_tag_type_from_values(min_enum_value, max_enum_value);
         // Enum names and polymorphic args do not conflict
         Self::check_identifier_collision(
             modules, root_id, &variants, |variant| &variant.identifier, "enum variant"
         )?;
         // Polymorphic arguments cannot appear as embedded types, because
         // they can only consist of integer variants.
         Self::check_poly_args_collision(modules, ctx, root_id, &definition.poly_vars)?;
         let poly_vars = Self::create_polymorphic_variables(&definition.poly_vars);
         self.lookup.insert(definition_id, DefinedType {
             ast_root: root_id,
             ast_definition: definition_id,
             definition: DefinedTypeVariant::Enum(EnumType{
                 variants,
                 monomorphs: Vec::new(),
                 minimum_tag_value: min_enum_value,
                 maximum_tag_value: max_enum_value,
                 tag_type,
                 size: size_and_alignment,
                 alignment: size_and_alignment
             }),
             poly_vars,
             is_polymorph: false,
         });
         return Ok(());
+    }
     /// Builds the base type for a union. Will compute byte sizes.
     fn build_base_union_definition(&mut self, modules: &[Module], ctx: &mut PassCtx, definition_id: DefinitionId) -> Result<(), ParseError> {
         debug_assert!(!self.lookup.contains_key(&definition_id), "base union already built");
         let definition = ctx.heap[definition_id].as_union();
         let root_id = definition.defined_in;
         // Check all variants and their embedded types
         let mut variants = Vec::with_capacity(definition.variants.len());
         let mut tag_counter = 0;
         for variant in &definition.variants {
             for embedded in &variant.value {
                 Self::check_member_parser_type(
                     modules, ctx, root_id, embedded, false
                 )?;
+            }
             variants.push(UnionVariant{
                 identifier: variant.identifier.clone(),
                 embedded: variant.value.clone(),
                 tag_value: tag_counter,
             });
             tag_counter += 1;
+        }
         let mut max_tag_value = 0;
         if tag_counter != 0 {
             max_tag_value = tag_counter - 1
+        }
         let (tag_type, tag_size) = Self::variant_tag_type_from_values(0, max_tag_value);
         // Make sure there are no conflicts in identifiers
         Self::check_identifier_collision(
             modules, root_id, &variants, |variant| &variant.identifier, "union variant"
         )?;
         Self::check_poly_args_collision(modules, ctx, root_id, &definition.poly_vars)?;
         // Construct internal representation of union
         let mut poly_vars = Self::create_polymorphic_variables(&definition.poly_vars);
         for variant in &definition.variants {
             for embedded in &variant.value {
                 Self::mark_used_polymorphic_variables(&mut poly_vars, embedded);
+            }
+        }
         let is_polymorph = poly_vars.iter().any(|arg| arg.is_in_use);
         self.lookup.insert(definition_id, DefinedType{
             ast_root: root_id,
             ast_definition: definition_id,
             definition: DefinedTypeVariant::Union(UnionType{
                 variants,
                 monomorphs: Vec::new(),
                 tag_type,
                 tag_size,
             }),
             poly_vars,
             is_polymorph
         });
         return Ok(());
+    }
     /// Builds base struct type. Will not compute byte sizes.
     fn build_base_struct_definition(&mut self, modules: &[Module], ctx: &mut PassCtx, definition_id: DefinitionId) -> Result<(), ParseError> {
         debug_assert!(!self.lookup.contains_key(&definition_id), "base struct already built");
         let definition = ctx.heap[definition_id].as_struct();
         let root_id = definition.defined_in;
         // Check all struct fields and construct internal representation
         let mut fields = Vec::with_capacity(definition.fields.len());
         for field in &definition.fields {
             Self::check_member_parser_type(
                 modules, ctx, root_id, &field.parser_type, false
             )?;
             fields.push(StructField{
                 identifier: field.field.clone(),
                 parser_type: field.parser_type.clone(),
             });
+        }
         // Make sure there are no conflicting variables
         Self::check_identifier_collision(
             modules, root_id, &fields, |field| &field.identifier, "struct field"
         )?;
         Self::check_poly_args_collision(modules, ctx, root_id, &definition.poly_vars)?;
         // Construct base type in table
         let mut poly_vars = Self::create_polymorphic_variables(&definition.poly_vars);
         for field in &fields {
             Self::mark_used_polymorphic_variables(&mut poly_vars, &field.parser_type);
+        }
         let is_polymorph = poly_vars.iter().any(|arg| arg.is_in_use);
         self.lookup.insert(definition_id, DefinedType{
             ast_root: root_id,
             ast_definition: definition_id,
             definition: DefinedTypeVariant::Struct(StructType{
                 fields,
                 monomorphs: Vec::new(),
             }),
             poly_vars,
             is_polymorph
         });
         return Ok(())
+    }
     /// Builds base function type.
     fn build_base_function_definition(&mut self, modules: &[Module], ctx: &mut PassCtx, definition_id: DefinitionId) -> Result<(), ParseError> {
         debug_assert!(!self.lookup.contains_key(&definition_id), "base function already built");
         let definition = ctx.heap[definition_id].as_function();
         let root_id = definition.defined_in;
         // Check and construct return types and argument types.
         debug_assert_eq!(definition.return_types.len(), 1, "not one return type"); // TODO: @ReturnValues
         for return_type in &definition.return_types {
             Self::check_member_parser_type(
                 modules, ctx, root_id, return_type, definition.builtin
             )?;
+        }
         let mut arguments = Vec::with_capacity(definition.parameters.len());
         for parameter_id in &definition.parameters {
             let parameter = &ctx.heap[*parameter_id];
             Self::check_member_parser_type(
                 modules, ctx, root_id, &parameter.parser_type, definition.builtin
             )?;
             arguments.push(FunctionArgument{
                 identifier: parameter.identifier.clone(),
                 parser_type: parameter.parser_type.clone(),
             });
+        }
         // Check conflict of identifiers
         Self::check_identifier_collision(
             modules, root_id, &arguments, |arg| &arg.identifier, "function argument"
         )?;
         Self::check_poly_args_collision(modules, ctx, root_id, &definition.poly_vars)?;
         // Construct internal representation of function type
         let mut poly_vars = Self::create_polymorphic_variables(&definition.poly_vars);
         for return_type in &definition.return_types {
             Self::mark_used_polymorphic_variables(&mut poly_vars, return_type);
+        }
         for argument in &arguments {
             Self::mark_used_polymorphic_variables(&mut poly_vars, &argument.parser_type);
+        }
         let is_polymorph = poly_vars.iter().any(|arg| arg.is_in_use);
         self.lookup.insert(definition_id, DefinedType{
             ast_root: root_id,
             ast_definition: definition_id,
             definition: DefinedTypeVariant::Function(FunctionType{
                 return_types: definition.return_types.clone(),
                 arguments,
                 monomorphs: Vec::new(),
             }),
             poly_vars,
             is_polymorph
         });
         return Ok(());
+    }
     /// Builds base component type.
     fn build_base_component_definition(&mut self, modules: &[Module], ctx: &mut PassCtx, definition_id: DefinitionId) -> Result<(), ParseError> {
         debug_assert!(!self.lookup.contains_key(&definition_id), "base component already built");
         let definition = &ctx.heap[definition_id].as_component();
         let root_id = definition.defined_in;
         // Check the argument types
         let mut arguments = Vec::with_capacity(definition.parameters.len());
         for parameter_id in &definition.parameters {
             let parameter = &ctx.heap[*parameter_id];
             Self::check_member_parser_type(
                 modules, ctx, root_id, &parameter.parser_type, false
             )?;
             arguments.push(FunctionArgument{
                 identifier: parameter.identifier.clone(),
                 parser_type: parameter.parser_type.clone(),
             });
+        }
         // Check conflict of identifiers
         Self::check_identifier_collision(
             modules, root_id, &arguments, |arg| &arg.identifier, "connector argument"
         )?;
         Self::check_poly_args_collision(modules, ctx, root_id, &definition.poly_vars)?;
         // Construct internal representation of component
         let mut poly_vars = Self::create_polymorphic_variables(&definition.poly_vars);
         for argument in &arguments {
             Self::mark_used_polymorphic_variables(&mut poly_vars, &argument.parser_type);
+        }
         let is_polymorph = poly_vars.iter().any(|arg| arg.is_in_use);
         self.lookup.insert(definition_id, DefinedType{
             ast_root: root_id,
             ast_definition: definition_id,
             definition: DefinedTypeVariant::Component(ComponentType{
                 variant: definition.variant,
                 arguments,
                 monomorphs: Vec::new()
             }),
             poly_vars,
             is_polymorph
         });
         Ok(())
+    }
     /// Will check if the member type (field of a struct, embedded type in a
     /// union variant) is valid.
     fn check_member_parser_type(
         modules: &[Module], ctx: &PassCtx, base_definition_root_id: RootId,
         member_parser_type: &ParserType, allow_special_compiler_types: bool
     ) -> Result<(), ParseError> {
         use ParserTypeVariant as PTV;
         for element in &member_parser_type.elements {
             match element.variant {
                 // Special cases
                 PTV::Void | PTV::InputOrOutput | PTV::ArrayLike | PTV::IntegerLike => {
                     if !allow_special_compiler_types {
                         unreachable!("compiler-only ParserTypeVariant in member type");
+                    }
                 },
                 // Builtin types, always valid
                 PTV::Message | PTV::Bool |
                 PTV::UInt8 | PTV::UInt16 | PTV::UInt32 | PTV::UInt64 |
                 PTV::SInt8 | PTV::SInt16 | PTV::SInt32 | PTV::SInt64 |
                 PTV::Character | PTV::String |
                 PTV::Array | PTV::Input | PTV::Output |
                 // Likewise, polymorphic variables are always valid
                 PTV::PolymorphicArgument(_, _) => {},
                 // Types that are not constructable, or types that are not
                 // allowed (and checked earlier)
                 PTV::IntegerLiteral | PTV::Inferred => {
                     unreachable!("illegal ParserTypeVariant within type definition");
                 },
                 // Finally, user-defined types
                 PTV::Definition(definition_id, _) => {
                     let definition = &ctx.heap[definition_id];
                     if !(definition.is_struct() || definition.is_enum() || definition.is_union()) {
                         let source = &modules[base_definition_root_id.index as usize].source;
                         return Err(ParseError::new_error_str_at_span(
                             source, element.element_span, "expected a datatype (a struct, enum or union)"
                         ));
+                    }
                     // Otherwise, we're fine
+                }
+            }
+        }
         // If here, then all elements check out
         return Ok(());
+    }
     /// Go through a list of identifiers and ensure that all identifiers have
     /// unique names
     fn check_identifier_collision<T: Sized, F: Fn(&T) -> &Identifier>(
         modules: &[Module], root_id: RootId, items: &[T], getter: F, item_name: &'static str
     ) -> Result<(), ParseError> {
         for (item_idx, item) in items.iter().enumerate() {
             let item_ident = getter(item);
             for other_item in &items[0..item_idx] {
                 let other_item_ident = getter(other_item);
                 if item_ident == other_item_ident {
                     let module_source = &modules[root_id.index as usize].source;
                     return Err(ParseError::new_error_at_span(
                         module_source, item_ident.span, format!("This {} is defined more than once", item_name)
                     ).with_info_at_span(
                         module_source, other_item_ident.span, format!("The other {} is defined here", item_name)
                     ));
+                }
+            }
+        }
         Ok(())
+    }
     /// Go through a list of polymorphic arguments and make sure that the
     /// arguments all have unique names, and the arguments do not conflict with
     /// any symbols defined at the module scope.
     fn check_poly_args_collision(
         modules: &[Module], ctx: &PassCtx, root_id: RootId, poly_args: &[Identifier]
     ) -> Result<(), ParseError> {
         // Make sure polymorphic arguments are unique and none of the
         // identifiers conflict with any imported scopes
         for (arg_idx, poly_arg) in poly_args.iter().enumerate() {
             for other_poly_arg in &poly_args[..arg_idx] {
                 if poly_arg == other_poly_arg {
                     let module_source = &modules[root_id.index as usize].source;
                     return Err(ParseError::new_error_str_at_span(
                         module_source, poly_arg.span,
                         "This polymorphic argument is defined more than once"
                     ).with_info_str_at_span(
                         module_source, other_poly_arg.span,
                         "It conflicts with this polymorphic argument"
                     ));
+                }
+            }
             // Check if identifier conflicts with a symbol defined or imported
             // in the current module
             if let Some(symbol) = ctx.symbols.get_symbol_by_name(SymbolScope::Module(root_id), poly_arg.value.as_bytes()) {
                 // We have a conflict
                 let module_source = &modules[root_id.index as usize].source;
                 let introduction_span = symbol.variant.span_of_introduction(ctx.heap);
                 return Err(ParseError::new_error_str_at_span(
                     module_source, poly_arg.span,
                     "This polymorphic argument conflicts with another symbol"
                 ).with_info_str_at_span(
                     module_source, introduction_span,
                     "It conflicts due to this symbol"
                 ));
+            }
+        }
         // All arguments are fine
         Ok(())
+    }
     //--------------------------------------------------------------------------
     // Detecting type loops
     //--------------------------------------------------------------------------
     /// Internal function that will detect type loops and check if they're
     /// resolvable. If so then the appropriate union variants will be marked as
     /// "living on heap". If not then a `ParseError` will be returned
     fn detect_and_resolve_type_loops_for(&mut self, modules: &[Module], heap: &Heap, concrete_type: ConcreteType) -> Result<(), ParseError> {
         use DefinedTypeVariant as DTV;
         debug_assert!(self.type_loop_breadcrumbs.is_empty());
         debug_assert!(self.type_loops.is_empty());
         debug_assert!(self.encountered_types.is_empty());
         // Push the initial breadcrumb
         let initial_breadcrumb = self.check_member_for_type_loops(&concrete_type);
         if let TypeLoopResult::PushBreadcrumb(definition_id, concrete_type) = initial_breadcrumb {
             self.handle_new_breadcrumb_for_type_loops(definition_id, concrete_type);
         } else {
             unreachable!();
+        }
         // Enter into the main resolving loop
         while !self.type_loop_breadcrumbs.is_empty() {
             // Because we might be modifying the breadcrumb array we need to
             let breadcrumb_idx = self.type_loop_breadcrumbs.len() - 1;
             let mut breadcrumb = self.type_loop_breadcrumbs[breadcrumb_idx].clone();
             let poly_type = self.lookup.get(&breadcrumb.definition_id).unwrap();
             let poly_type_definition_id = poly_type.ast_definition;
             let resolve_result = match &poly_type.definition {
                 DTV::Enum(_) => {
                     TypeLoopResult::TypeExists
                 },
                 DTV::Union(definition) => {
                     let monomorph = &definition.monomorphs[breadcrumb.monomorph_idx];
                     let num_variants = monomorph.variants.len();
                     let mut union_result = TypeLoopResult::TypeExists;
                     'member_loop: while breadcrumb.next_member < num_variants {
                         let mono_variant = &monomorph.variants[breadcrumb.next_member];
                         let num_embedded = mono_variant.embedded.len();
                         while breadcrumb.next_embedded < num_embedded {
                             let mono_embedded = &mono_variant.embedded[breadcrumb.next_embedded];
                             union_result = self.check_member_for_type_loops(&mono_embedded.concrete_type);
                             if union_result != TypeLoopResult::TypeExists {
                                 // In type loop or new breadcrumb pushed, so
                                 // break out of the resolving loop
                                 break 'member_loop;
+                            }
                             breadcrumb.next_embedded += 1;
+                        }
                         breadcrumb.next_embedded = 0;
                         breadcrumb.next_member += 1
+                    }
                     union_result
                 },
                 DTV::Struct(definition) => {
                     let monomorph = &definition.monomorphs[breadcrumb.monomorph_idx];
                     let num_fields = monomorph.fields.len();
                     let mut struct_result = TypeLoopResult::TypeExists;
                     while breadcrumb.next_member < num_fields {
                         let mono_field = &monomorph.fields[breadcrumb.next_member];
                         struct_result = self.check_member_for_type_loops(&mono_field.concrete_type);
                         if struct_result != TypeLoopResult::TypeExists {
                             // Type loop or breadcrumb pushed, so break out of
                             // the resolving loop
                             break;
+                        }
                         breadcrumb.next_member += 1;
+                    }
                     struct_result
                 },
                 DTV::Function(_) | DTV::Component(_) => unreachable!(),
             };
             // Handle the result of attempting to resolve the current breadcrumb
             match resolve_result {
                 TypeLoopResult::TypeExists => {
                     // We finished parsing the type
                     self.type_loop_breadcrumbs.pop();
                 },
                 TypeLoopResult::PushBreadcrumb(definition_id, concrete_type) => {
                     // We recurse into the member type.
                     self.type_loop_breadcrumbs[breadcrumb_idx] = breadcrumb;
                     self.handle_new_breadcrumb_for_type_loops(definition_id, concrete_type);
                 },
                 TypeLoopResult::TypeLoop(first_idx) => {
                     // Because we will be modifying breadcrumbs within the
                     // type-loop handling code, put back the modified breadcrumb
                     self.type_loop_breadcrumbs[breadcrumb_idx] = breadcrumb;
                     // We're in a type loop. Add the type loop
                     let mut loop_members = Vec::with_capacity(self.type_loop_breadcrumbs.len() - first_idx);
                     let mut contains_union = false;
                     for breadcrumb_idx in first_idx..self.type_loop_breadcrumbs.len() {
                         let breadcrumb = &mut self.type_loop_breadcrumbs[breadcrumb_idx];
                         let mut is_union = false;
                         let entry = self.lookup.get_mut(&breadcrumb.definition_id).unwrap();
                         match &mut entry.definition {
                             DTV::Union(definition) => {
                                 // Mark the currently processed variant as requiring heap
                                 // allocation, then advance the *embedded* type. The loop above
                                 // will then take care of advancing it to the next *member*.
                                 let monomorph = &mut definition.monomorphs[breadcrumb.monomorph_idx];
                                 let variant = &mut monomorph.variants[breadcrumb.next_member];
                                 variant.lives_on_heap = true;
                                 breadcrumb.next_embedded += 1;
                                 is_union = true;
                                 contains_union = true;
                             },
                             _ => {}, // else: we don't care for now
+                        }
                         loop_members.push(TypeLoopEntry{
                             definition_id: breadcrumb.definition_id,
                             monomorph_idx: breadcrumb.monomorph_idx,
                             is_union
                         });
+                    }
                     let new_type_loop = TypeLoop{ members: loop_members };
                     if !contains_union {
                         // No way to (potentially) break the union. So return a
                         // type loop error. This is because otherwise our
                         // breadcrumb resolver ends up in an infinite loop.
                         return Err(construct_type_loop_error(
                             self, &new_type_loop, modules, heap
                         ));
+                    }
                     self.type_loops.push(new_type_loop);
+                }
+            }
+        }
         // All breadcrumbs have been cleared. So now `type_loops` contains all
         // of the encountered type loops, and `encountered_types` contains a
         // list of all unique monomorphs we encountered.
         // The next step is to figure out if all of the type loops can be
         // broken. A type loop can be broken if at least one union exists in the
         // loop and that union ended up having variants that are not part of
         // a type loop.
         fn type_loop_source_span_and_message<'a>(
             modules: &'a [Module], heap: &Heap, defined_type: &DefinedType, monomorph_idx: usize, index_in_loop: usize
         ) -> (&'a InputSource, InputSpan, String) {
             // Note: because we will discover the type loop the *first* time we
             // instantiate a monomorph with the provided polymorphic arguments
             // (not all arguments are actually used in the type). We don't have
             // to care about a second instantiation where certain unused
             // polymorphic arguments are different.
             let monomorph_type = match &defined_type.definition {
                 DTV::Union(definition) => &definition.monomorphs[monomorph_idx].concrete_type,
                 DTV::Struct(definition) => &definition.monomorphs[monomorph_idx].concrete_type,
                 DTV::Enum(_) | DTV::Function(_) | DTV::Component(_) =>
                     unreachable!(), // impossible to have an enum/procedure in a type loop
             };
             let type_name = monomorph_type.display_name(&heap);
             let message = if index_in_loop == 0 {
                 format!(
                     "encountered an infinitely large type for '{}' (which can be fixed by \
                     introducing a union type that has a variant whose embedded types are \
                     not part of a type loop, or do not have embedded types)",
                     type_name
+                )
             } else if index_in_loop == 1 {
                 format!("because it depends on the type '{}'", type_name)
             } else {
                 format!("which depends on the type '{}'", type_name)
             };
             let ast_definition = &heap[defined_type.ast_definition];
             let ast_root_id = ast_definition.defined_in();
             return (
                 &modules[ast_root_id.index as usize].source,
                 ast_definition.identifier().span,
                 message
             );
+        }
         fn construct_type_loop_error(table: &TypeTable, type_loop: &TypeLoop, modules: &[Module], heap: &Heap) -> ParseError {
             let first_entry = &type_loop.members[0];
             let first_type = table.lookup.get(&first_entry.definition_id).unwrap();
             let (first_module, first_span, first_message) = type_loop_source_span_and_message(
                 modules, heap, first_type, first_entry.monomorph_idx, 0
             );
             let mut parse_error = ParseError::new_error_at_span(first_module, first_span, first_message);
             for member_idx in 1..type_loop.members.len() {
                 let entry = &type_loop.members[member_idx];
                 let entry_type = table.lookup.get(&first_entry.definition_id).unwrap();
                 let (module, span, message) = type_loop_source_span_and_message(
                     modules, heap, entry_type, entry.monomorph_idx, member_idx
                 );
                 parse_error = parse_error.with_info_at_span(module, span, message);
+            }
             parse_error
+        }
         for type_loop in &self.type_loops {
             let mut can_be_broken = false;
             debug_assert!(!type_loop.members.is_empty());
             for entry in &type_loop.members {
                 if entry.is_union {
                     let base_type = self.lookup.get(&entry.definition_id).unwrap();
                     let monomorph = &base_type.definition.as_union().monomorphs[entry.monomorph_idx];
                     debug_assert!(!monomorph.variants.is_empty()); // otherwise it couldn't be part of the type loop
                     let has_stack_variant = monomorph.variants.iter().any(|variant| !variant.lives_on_heap);
                     if has_stack_variant {
                         can_be_broken = true;
+                    }
+                }
+            }
             if !can_be_broken {
                 // Construct a type loop error
                 return Err(construct_type_loop_error(self, type_loop, modules, heap));
+            }
+        }
         // If here, then all type loops have been resolved and we can lay out
         // all of the members
         self.type_loops.clear();
         return Ok(());
+    }
     /// Checks if the specified type needs to be resolved (i.e. we need to push
     /// a breadcrumb), is already resolved (i.e. we can continue with the next
     /// member of the currently considered type) or is in the process of being
     /// resolved (i.e. we're in a type loop). Because of borrowing rules we
     /// don't do any modifications of internal types here. Hence: if we
     /// return `PushBreadcrumb` then call `handle_new_breadcrumb_for_type_loops`
     /// to take care of storing the appropriate types.
     fn check_member_for_type_loops(&self, definition_type: &ConcreteType) -> TypeLoopResult {
         use ConcreteTypePart as CTP;
         // We're only interested in user-defined types, so exit if it is a
         // builtin of some sort.
         debug_assert!(!definition_type.parts.is_empty());
         let definition_id = match &definition_type.parts[0] {
             CTP::Instance(definition_id, _) |
             CTP::Function(definition_id, _) |
             CTP::Component(definition_id, _) => {
                 *definition_id
             },
             _ => {
                 return TypeLoopResult::TypeExists
             },
         };
         let base_type = self.lookup.get(&definition_id).unwrap();
         if let Some(mono_idx) = base_type.get_monomorph_index(&definition_type) {
             // Monomorph is already known. Check if it is present in the
             // breadcrumbs. If so, then we are in a type loop
             for (breadcrumb_idx, breadcrumb) in self.type_loop_breadcrumbs.iter().enumerate() {
                 if breadcrumb.definition_id == definition_id && breadcrumb.monomorph_idx == mono_idx {
                     return TypeLoopResult::TypeLoop(breadcrumb_idx);
+                }
+            }
             return TypeLoopResult::TypeExists;
+        }
         // Type is not yet known, so we need to insert it into the lookup and
         // push a new breadcrumb.
         return TypeLoopResult::PushBreadcrumb(definition_id, definition_type.clone());
+    }
     /// Handles the `PushBreadcrumb` result for a `check_member_for_type_loops`
     /// call.
     fn handle_new_breadcrumb_for_type_loops(&mut self, definition_id: DefinitionId, definition_type: ConcreteType) {
         use DefinedTypeVariant as DTV;
         let base_type = self.lookup.get_mut(&definition_id).unwrap();
         let mut is_union = false;
         let monomorph_idx = match &mut base_type.definition {
             DTV::Enum(definition) => {
                 debug_assert!(definition.monomorphs.is_empty());
                 definition.monomorphs.push(EnumMonomorph{
                     concrete_type: definition_type,
                 });
             },
             DTV::Union(definition) => {
                 // Create all the variants with their concrete types
                 let mut mono_variants = Vec::with_capacity(definition.variants.len());
                 for poly_variant in &definition.variants {
                     let mut mono_embedded = Vec::with_capacity(poly_variant.embedded.len());
                     for poly_embedded in &poly_variant.embedded {
                         let mono_concrete = Self::construct_concrete_type(poly_embedded, &definition_type);
                         mono_embedded.push(UnionMonomorphEmbedded{
                             concrete_type: mono_concrete,
                             size: 0,
                             alignment: 0,
                             offset: 0
                         });
+                    }
                     mono_variants.push(UnionMonomorphVariant{
                         lives_on_heap: false,
                         embedded: mono_embedded,
                     })
+                }
                 let mono_idx = definition.monomorphs.len();
                 definition.monomorphs.push(UnionMonomorph{
                     concrete_type: definition_type,
                     variants: mono_variants,
                     stack_size: 0,
                     stack_alignment: 0,
                     heap_size: 0,
                     heap_alignment: 0
                 });
                 is_union = true;
                 mono_idx
             },
             DTV::Struct(definition) => {
                 let mut mono_fields = Vec::with_capacity(definition.fields.len());
                 for poly_field in &definition.fields {
                     let mono_concrete = Self::construct_concrete_type(&poly_field.parser_type, &definition_type);
                     mono_fields.push(StructMonomorphField{
                         concrete_type: mono_concrete,
                         size: 0,
                         alignment: 0,
                         offset: 0
                     })
+                }
                 let mono_idx = definition.monomorphs.len();
                 definition.monomorphs.push(StructMonomorph{
                     concrete_type: definition_type,
                     fields: mono_fields,
                     size: 0,
                     alignment: 0
                 });
                 mono_idx
             },
             DTV::Function(_) | DTV::Component(_) => {
                 unreachable!("pushing type resolving breadcrumb for procedure type")
             },
         };
         self.encountered_types.push(TypeLoopEntry{
             definition_id,
             monomorph_idx,
             is_union,
         });
         self.type_loop_breadcrumbs.push(TypeLoopBreadcrumb{
             definition_id,
             monomorph_idx,
             next_member: 0,
             next_embedded: 0,
         });
+    }
     /// Constructs a concrete type out of a parser type for a struct field or
     /// union embedded type. It will do this by looking up the polymorphic
     /// variables in the supplied concrete type. The assumption is that the
     /// polymorphic variable's indices correspond to the subtrees in the
     /// concrete type.
     fn construct_concrete_type(member_type: &ParserType, container_type: &ConcreteType) -> ConcreteType {
         use ParserTypeVariant as PTV;
         use ConcreteTypePart as CTP;
         // TODO: Combine with code in pass_typing.rs
         fn parser_to_concrete_part(part: &ParserTypeVariant) -> Option<ConcreteTypePart> {
             match part {
                 PTV::Void      => Some(CTP::Void),
                 PTV::Message   => Some(CTP::Message),
                 PTV::Bool      => Some(CTP::Bool),
                 PTV::UInt8     => Some(CTP::UInt8),
                 PTV::UInt16    => Some(CTP::UInt16),
                 PTV::UInt32    => Some(CTP::UInt32),
                 PTV::UInt64    => Some(CTP::UInt64),
                 PTV::SInt8     => Some(CTP::SInt8),
                 PTV::SInt16    => Some(CTP::SInt16),
                 PTV::SInt32    => Some(CTP::SInt32),
                 PTV::SInt64    => Some(CTP::SInt64),
                 PTV::Character => Some(CTP::Character),
                 PTV::String    => Some(CTP::String),
                 PTV::Array     => Some(CTP::Array),
                 PTV::Input     => Some(CTP::Input),
                 PTV::Output    => Some(CTP::Output),
                 PTV::Definition(definition_id, num) => Some(CTP::Instance(*definition_id, *num)),
                 _              => None
+            }
+        }
         let mut parts = Vec::with_capacity(member_type.elements.len()); // usually a correct estimation, might not be
         for member_part in &member_type.elements {
             // Check if we have a regular builtin type
             if let Some(part) = parser_to_concrete_part(&member_part.variant) {
                 parts.push(part);
                 continue;
+            }
             // Not builtin, but if all code is working correctly, we only care
             // about the polymorphic argument at this point.
             if let PTV::PolymorphicArgument(_container_definition_id, poly_arg_idx) = member_part.variant {
                 debug_assert_eq!(_container_definition_id, get_concrete_type_definition(container_type));
                 let mut container_iter = container_type.embedded_iter(0);
                 for _ in 0..poly_arg_idx {
                     container_iter.next();
+                }
                 let poly_section = container_iter.next().unwrap();
                 parts.extend(poly_section);
                 continue;
+            }
             unreachable!("unexpected type part {:?} from {:?}", member_part, member_type);
+        }
         return ConcreteType{ parts };
+    }
     //--------------------------------------------------------------------------
     // Determining memory layout for types
     //--------------------------------------------------------------------------
     fn lay_out_memory_for_encountered_types(&mut self, arch: &TargetArch) {
         use DefinedTypeVariant as DTV;
         // Just finished type loop detection, so we're left with the encountered
         // types only
         debug_assert!(self.type_loops.is_empty());
         debug_assert!(!self.encountered_types.is_empty());
         debug_assert!(self.memory_layout_breadcrumbs.is_empty());
         debug_assert!(self.size_alignment_stack.is_empty());
         // Push the first entry (the type we originally started with when we
         // were detecting type loops)
         let first_entry = &self.encountered_types[0];
         self.memory_layout_breadcrumbs.push(MemoryBreadcrumb{
             definition_id: first_entry.definition_id,
             monomorph_idx: first_entry.monomorph_idx,
             next_member: 0,
             next_embedded: 0,
             first_size_alignment_idx: 0,
         });
         // Enter the main resolving loop
         'breadcrumb_loop: while !self.memory_layout_breadcrumbs.is_empty() {
             let cur_breadcrumb_idx = self.memory_layout_breadcrumbs.len() - 1;
             let mut breadcrumb = self.memory_layout_breadcrumbs[cur_breadcrumb_idx].clone();
             let poly_type = self.lookup.get(&breadcrumb.definition_id).unwrap();
             match &poly_type.definition {
                 DTV::Enum(definition) => {
                     // Size should already be computed
                     debug_assert!(definition.size != 0 && definition.alignment != 0);
                 },
                 DTV::Union(definition) => {
                     // Retrieve size/alignment of each embedded type. We do not
                     // compute the offsets or total type sizes yet.
                     let mono_type = &definition.monomorphs[breadcrumb.monomorph_idx];
                     let num_variants = mono_type.variants.len();
                     while breadcrumb.next_member < num_variants {
                         let mono_variant = &mono_type.variants[breadcrumb.next_member];
                         if mono_variant.lives_on_heap {
                             // To prevent type loops we made this a heap-
                             // allocated variant. This implies we cannot
                             // compute sizes of members at this point.
                         } else {
                             let num_embedded = mono_variant.embedded.len();
                             while breadcrumb.next_embedded < num_embedded {
                                 let mono_embedded = &mono_variant.embedded[breadcrumb.next_embedded];
                                 match self.get_memory_layout_or_breadcrumb(arch, &mono_embedded.concrete_type) {
                                     MemoryLayoutResult::TypeExists(size, alignment) => {
                                         self.size_alignment_stack.push((size, alignment));
                                     },
                                     MemoryLayoutResult::PushBreadcrumb(new_breadcrumb) => {
                                         self.memory_layout_breadcrumbs[cur_breadcrumb_idx] = breadcrumb;
                                         self.memory_layout_breadcrumbs.push(new_breadcrumb);
                                         continue 'breadcrumb_loop;
+                                    }
+                                }
                                 breadcrumb.next_embedded += 1;
+                            }
+                        }
                         breadcrumb.next_member += 1;
                         breadcrumb.next_embedded = 0;
+                    }
                     // If here then we can at least compute the stack size of
                     // the type, we'll have to come back at the very end to
                     // fill in the heap size/alignment/offset of each heap-
                     // allocated variant.
                     let mut max_size = definition.tag_size;
                     let mut max_alignment = definition.tag_size;
                     let poly_type = self.lookup.get_mut(&breadcrumb.definition_id).unwrap();
                     let definition = poly_type.definition.as_union_mut();
                     let mono_type = &mut definition.monomorphs[breadcrumb.monomorph_idx];
                     let mut size_alignment_idx = breadcrumb.first_size_alignment_idx;
                     for variant in &mut mono_type.variants {
                         // We're doing stack computations, so always start with
                         // the tag size/alignment.
                         let mut variant_offset = definition.tag_size;
                         let mut variant_alignment = definition.tag_size;
                         if variant.lives_on_heap {
                             // Variant lives on heap, so just a pointer
                             let (ptr_size, ptr_align) = arch.pointer_size_alignment;
                             align_offset_to(&mut variant_offset, ptr_align);
                             variant_offset += ptr_size;
                             variant_alignment = variant_alignment.max(ptr_align);
                         } else {
                             // Variant lives on stack, so walk all embedded
                             // types.
                             for embedded in &mut variant.embedded {
                                 let (size, alignment) = self.size_alignment_stack[size_alignment_idx];
                                 embedded.size = size;
                                 embedded.alignment = alignment;
                                 size_alignment_idx += 1;
                                 align_offset_to(&mut variant_offset, alignment);
                                 embedded.offset = variant_offset;
                                 variant_offset += size;
                                 variant_alignment = variant_alignment.max(alignment);
+                            }
                         };
                         max_size = max_size.max(variant_offset);
                         max_alignment = max_alignment.max(variant_alignment);
+                    }
                     mono_type.stack_size = max_size;
                     mono_type.stack_alignment = max_alignment;
                     self.size_alignment_stack.truncate(breadcrumb.first_size_alignment_idx);
                 },
                 DTV::Struct(definition) => {
                     // Retrieve size and alignment of each struct member. We'll
                     // compute the offsets once all of those are known
                     let mono_type = &definition.monomorphs[breadcrumb.monomorph_idx];
                     let num_fields = mono_type.fields.len();
                     while breadcrumb.next_member < num_fields {
                         let mono_field = &mono_type.fields[breadcrumb.next_member];
                         match self.get_memory_layout_or_breadcrumb(arch, &mono_field.concrete_type) {
                             MemoryLayoutResult::TypeExists(size, alignment) => {
                                 self.size_alignment_stack.push((size, alignment))
                             },
                             MemoryLayoutResult::PushBreadcrumb(new_breadcrumb) => {
                                 self.memory_layout_breadcrumbs[cur_breadcrumb_idx] = breadcrumb;
                                 self.memory_layout_breadcrumbs.push(new_breadcrumb);
                                 continue 'breadcrumb_loop;
                             },
+                        }
                         breadcrumb.next_member += 1;
+                    }
                     // Compute offsets and size of total type
                     let mut cur_offset = 0;
                     let mut max_alignment = 1;
                     let poly_type = self.lookup.get_mut(&breadcrumb.definition_id).unwrap();
                     let definition = poly_type.definition.as_struct_mut();
                     let mono_type = &mut definition.monomorphs[breadcrumb.monomorph_idx];
                     let mut size_alignment_idx = breadcrumb.first_size_alignment_idx;
                     for field in &mut mono_type.fields {
                         let (size, alignment) = self.size_alignment_stack[size_alignment_idx];
                         field.size = size;
                         field.alignment = alignment;
                         size_alignment_idx += 1;
                         align_offset_to(&mut cur_offset, alignment);
                         field.offset = cur_offset;
                         cur_offset += size;
                         max_alignment = max_alignment.max(alignment);
+                    }
                     mono_type.size = cur_offset;
                     mono_type.alignment = max_alignment;
                     self.size_alignment_stack.truncate(breadcrumb.first_size_alignment_idx);
                 },
                 DTV::Function(_) | DTV::Component(_) => {
                     unreachable!();
+                }
+            }
             // If here, then we completely layed out the current type. So move
             // to the next breadcrumb
             self.memory_layout_breadcrumbs.pop();
+        }
         debug_assert!(self.size_alignment_stack.is_empty());
         // If here then all types have been layed out. What remains is to
         // compute the sizes/alignment/offsets of the heap variants of the
         // unions we have encountered.
         for entry in &self.encountered_types {
             if !entry.is_union {
                 continue;
+            }
             // First pass, use buffer to store size/alignment to prevent
             // borrowing issues.
             let poly_type = self.lookup.get(&entry.definition_id).unwrap();
             let definition = poly_type.definition.as_union();
             let mono_type = &definition.monomorphs[entry.monomorph_idx];
             for variant in &mono_type.variants {
                 if !variant.lives_on_heap {
                     continue;
+                }
                 debug_assert!(!variant.embedded.is_empty());
                 for embedded in &variant.embedded {
                     match self.get_memory_layout_or_breadcrumb(arch, &embedded.concrete_type) {
                         MemoryLayoutResult::TypeExists(size, alignment) => {
                             self.size_alignment_stack.push((size, alignment));
                         },
                         _ => unreachable!(),
+                    }
+                }
+            }
             // Second pass, apply the size/alignment values in our buffer
             let poly_type = self.lookup.get_mut(&entry.definition_id).unwrap();
             let definition = poly_type.definition.as_union_mut();
             let mono_type = &mut definition.monomorphs[entry.monomorph_idx];
             let mut max_size = 0;
             let mut max_alignment = 1;
             let mut size_alignment_idx = 0;
             for variant in &mut mono_type.variants {
                 if !variant.lives_on_heap {
                     continue;
+                }
                 let mut variant_offset = 0;
                 let mut variant_alignment = 1;
                 for embedded in &mut variant.embedded {
                     let (size, alignment) = self.size_alignment_stack[size_alignment_idx];
                     embedded.size = size;
                     embedded.alignment = alignment;
                     size_alignment_idx += 1;
                     align_offset_to(&mut variant_offset, alignment);
                     embedded.alignment = variant_offset;
                     variant_offset += size;
                     variant_alignment = variant_alignment.max(alignment);
+                }
                 max_size = max_size.max(variant_offset);
                 max_alignment = max_alignment.max(variant_alignment);
+            }
             if max_size != 0 {
                 // At least one entry lives on the heap
                 mono_type.heap_size = max_size;
                 mono_type.heap_alignment = max_alignment;
+            }
+        }
         // And now, we're actually, properly, done
         self.encountered_types.clear();
+    }
     fn get_memory_layout_or_breadcrumb(&self, arch: &TargetArch, concrete_type: &ConcreteType) -> MemoryLayoutResult {
         use ConcreteTypePart as CTP;
         // Before we do any fancy shenanigans, we need to check if the concrete
         // type actually requires laying out memory.
         debug_assert!(!concrete_type.parts.is_empty());
         let (builtin_size, builtin_alignment) = match concrete_type.parts[0] {
             CTP::Void   => (0, 1),
             CTP::Message => arch.array_size_alignment,
             CTP::Bool   => (1, 1),
             CTP::UInt8  => (1, 1),
             CTP::UInt16 => (2, 2),
             CTP::UInt32 => (4, 4),
             CTP::UInt64 => (8, 8),
             CTP::SInt8  => (1, 1),
             CTP::SInt16 => (2, 2),
             CTP::SInt32 => (4, 4),
             CTP::SInt64 => (8, 8),
             CTP::Character => (4, 4),
             CTP::String => arch.string_size_alignment,
             CTP::Array => arch.array_size_alignment,
             CTP::Slice => arch.array_size_alignment,
             CTP::Input => arch.port_size_alignment,
             CTP::Output => arch.port_size_alignment,
             CTP::Instance(definition_id, _) => {
                 // Special case where we explicitly return to simplify the
                 // return case for the builtins.
                 let entry = self.lookup.get(&definition_id).unwrap();
                 let monomorph_idx = entry.get_monomorph_index(concrete_type).unwrap();
                 if let Some((size, alignment)) = entry.get_monomorph_size_alignment(monomorph_idx) {
                     // Type has been layed out in memory
                     return MemoryLayoutResult::TypeExists(size, alignment);
                 } else {
                     return MemoryLayoutResult::PushBreadcrumb(MemoryBreadcrumb{
                         definition_id,
                         monomorph_idx,
                         next_member: 0,
                         next_embedded: 0,
                         first_size_alignment_idx: self.size_alignment_stack.len(),
                     });
+                }
             },
             CTP::Function(_, _) | CTP::Component(_, _) => {
                 todo!("storage for 'function pointers'");
+            }
         };
         return MemoryLayoutResult::TypeExists(builtin_size, builtin_alignment);
+    }
     /// Returns tag concrete type (always a builtin integer type), the size of
     /// that type in bytes (and implicitly, its alignment)
     fn variant_tag_type_from_values(min_val: i64, max_val: i64) -> (ConcreteType, usize) {
         debug_assert!(min_val <= max_val);
         let (part, size) = if min_val >= 0 {
             // Can be an unsigned integer
             if max_val <= (u8::MAX as i64) {
                 (ConcreteTypePart::UInt8, 1)
             } else if max_val <= (u16::MAX as i64) {
                 (ConcreteTypePart::UInt16, 2)
             } else if max_val <= (u32::MAX as i64) {
                 (ConcreteTypePart::UInt32, 4)
             } else {
                 (ConcreteTypePart::UInt64, 8)
+            }
         } else {
             // Must be a signed integer
             if min_val >= (i8::MIN as i64) && max_val <= (i8::MAX as i64) {
                 (ConcreteTypePart::SInt8, 1)
             } else if min_val >= (i16::MIN as i64) && max_val <= (i16::MAX as i64) {
                 (ConcreteTypePart::SInt16, 2)
             } else if min_val >= (i32::MIN as i64) && max_val <= (i32::MAX as i64) {
                 (ConcreteTypePart::SInt32, 4)
             } else {
                 (ConcreteTypePart::SInt64, 8)
+            }
         };
         return (ConcreteType{ parts: vec![part] }, size);
+    }
     //--------------------------------------------------------------------------
     // Small utilities
     //--------------------------------------------------------------------------
     fn create_polymorphic_variables(variables: &[Identifier]) -> Vec<PolymorphicVariable> {
         let mut result = Vec::with_capacity(variables.len());
         for variable in variables.iter() {
             result.push(PolymorphicVariable{ identifier: variable.clone(), is_in_use: false });
+        }
         result
+    }
     fn mark_used_polymorphic_variables(poly_vars: &mut Vec<PolymorphicVariable>, parser_type: &ParserType) {
         for element in &parser_type.elements {
             if let ParserTypeVariant::PolymorphicArgument(_, idx) = &element.variant {
                 poly_vars[*idx as usize].is_in_use = true;
+            }
+        }
+    }
+}
 #[inline] fn align_offset_to(offset: &mut usize, alignment: usize) {
     debug_assert!(alignment > 0);
     let alignment_min_1 = alignment - 1;
     *offset += alignment_min_1;
     *offset &= !(alignment_min_1);
+}
 #[inline] fn get_concrete_type_definition(concrete: &ConcreteType) -> DefinitionId {
     if let ConcreteTypePart::Instance(definition_id, _) = concrete.parts[0] {
         return definition_id;
     } else {
         debug_assert!(false, "passed {:?} to the type table", concrete);
         return DefinitionId::new_invalid()
+    }
+}
@@ \ No newline at end of file @@

src/runtime2/connector.rs

➞

Show inline comments

 use std::collections::HashMap;
 use std::sync::atomic::AtomicBool;
 use crate::{PortId, ProtocolDescription};
 use crate::protocol::{ComponentState, RunContext, RunResult};
 use crate::protocol::eval::{Prompt, Value, ValueGroup};
 use crate::runtime2::inbox::{MessageContents, SolutionMessage};
+use crate::runtime2::inbox::{Message, MessageContents, SolutionMessage};
 use crate::runtime2::native::Connector;
 use crate::runtime2::port::{Port, PortKind};
 use crate::runtime2::scheduler::ConnectorCtx;
 use super::global_store::ConnectorId;
 use super::inbox::{
     PrivateInbox, PublicInbox, DataMessage, SyncMessage,
     SyncBranchConstraint, SyncConnectorSolution
 };
 use super::port::PortIdLocal;
 /// Represents the identifier of a branch (the index within its container). An
 /// ID of `0` generally means "no branch" (e.g. no parent, or a port did not
 /// yet receive anything from any branch).
 #[derive(Clone, Copy, PartialEq, Eq)]
-pub(crate) struct BranchId {
 pub struct BranchId {
     pub index: u32,
+}
 impl BranchId {
     fn new_invalid() -> Self {
         Self{ index: 0 }
+    }
     fn new(index: u32) -> Self {
         debug_assert!(index != 0);
         Self{ index }
+    }
     #[inline]
     pub(crate) fn is_valid(&self) -> bool {
         return self.index != 0;
+    }
+}
 #[derive(Debug, PartialEq, Eq)]
 pub(crate) enum SpeculativeState {
     // Non-synchronous variants
     RunningNonSync,         // regular execution of code
     Error,                  // encountered a runtime error
     Finished,               // finished executing connector's code
     // Synchronous variants
     RunningInSync,          // running within a sync block
     HaltedAtBranchPoint,    // at a branching point (at a `get` call)
     ReachedSyncEnd,         // reached end of sync block, branch represents a local solution
     Inconsistent,           // branch can never represent a local solution, so halted
+}
 pub(crate) struct Branch {
     index: BranchId,
     parent_index: BranchId,
     // Code execution state
     code_state: ComponentState,
     sync_state: SpeculativeState,
     next_branch_in_queue: Option<u32>,
     // Message/port state
     received: HashMap<PortIdLocal, DataMessage>, // TODO: @temporary, remove together with fires()
     ports_delta: Vec<PortOwnershipDelta>,
+}
 impl Branch {
     /// Constructs a non-sync branch. It is assumed that the code is at the
     /// first instruction
     pub(crate) fn new_initial_branch(component_state: ComponentState) -> Self {
         Branch{
             index: BranchId::new_invalid(),
             parent_index: BranchId::new_invalid(),
             code_state: component_state,
             sync_state: SpeculativeState::RunningNonSync,
             next_branch_in_queue: None,
             received: HashMap::new(),
             ports_delta: Vec::new(),
+        }
+    }
     /// Constructs a sync branch. The provided branch is assumed to be the
     /// parent of the new branch within the execution tree.
     fn new_sync_branching_from(new_index: u32, parent_branch: &Branch) -> Self {
         debug_assert!(
             (parent_branch.sync_state == SpeculativeState::RunningNonSync && !parent_branch.parent_index.is_valid()) ||
             (parent_branch.sync_state == SpeculativeState::HaltedAtBranchPoint)
         );
         Branch{
             index: BranchId::new(new_index),
             parent_index: parent_branch.index,
             code_state: parent_branch.code_state.clone(),
             sync_state: SpeculativeState::RunningInSync,
             next_branch_in_queue: None,
             received: parent_branch.received.clone(),
             ports_delta: parent_branch.ports_delta.clone(),
+        }
+    }
     fn commit_to_sync(&mut self) {
         self.index = BranchId::new(0);
         self.parent_index = BranchId::new_invalid();
         self.sync_state = SpeculativeState::RunningNonSync;
         self.next_branch_in_queue = None;
         self.received.clear();
         self.ports_delta.clear();
+    }
+}
 #[derive(Clone)]
 struct PortAssignment {
     is_assigned: bool,
     last_registered_branch_id: BranchId, // invalid branch ID implies not assigned yet
     num_times_fired: u32,
+}
 impl PortAssignment {
     fn new_unassigned() -> Self {
         Self{
             is_assigned: false,
             last_registered_branch_id: BranchId::new_invalid(),
             num_times_fired: 0,
+        }
+    }
     #[inline]
     fn mark_speculative(&mut self, num_times_fired: u32) {
         debug_assert!(!self.last_registered_branch_id.is_valid());
         self.is_assigned = true;
         self.num_times_fired = num_times_fired;
+    }
     #[inline]
     fn mark_definitive(&mut self, branch_id: BranchId, num_times_fired: u32) {
         self.is_assigned = true;
         self.last_registered_branch_id = branch_id;
         self.num_times_fired = num_times_fired;
+    }
+}
 #[derive(Clone)]
 struct PortOwnershipDelta {
     acquired: bool, // if false, then released ownership
     port_id: PortIdLocal,
+}
 #[derive(Debug)]
 enum PortOwnershipError {
     UsedInInteraction(PortIdLocal),
     AlreadyGivenAway(PortIdLocal)
+}
 /// Contains a description of the port mapping during a particular sync session.
 /// TODO: Extend documentation
 pub(crate) struct ConnectorPorts {
     // Essentially a mapping from `port_index` to `port_id`.
     pub owned_ports: Vec<PortIdLocal>,
     // Contains P*B entries, where P is the number of ports and B is the number
     // of branches. One can find the appropriate mapping of port p at branch b
     // at linear index `b*P+p`.
-    pub port_mapping: Vec<PortAssignment>
     port_mapping: Vec<PortAssignment>
+}
 impl ConnectorPorts {
     /// Constructs the initial ports object. Assumes the presence of the
     /// non-sync branch at index 0. Will initialize all entries for the non-sync
     /// branch.
     fn new(owned_ports: Vec<PortIdLocal>) -> Self {
         let num_ports = owned_ports.len();
         let mut port_mapping = Vec::with_capacity(num_ports);
         for _ in 0..num_ports {
             port_mapping.push(PortAssignment::new_unassigned());
+        }
         Self{ owned_ports, port_mapping }
+    }
     /// Prepares the port mapping for a new branch. Assumes that there is no
     /// intermediate branch index that we have skipped.
     fn prepare_sync_branch(&mut self, parent_branch_idx: u32, new_branch_idx: u32) {
         let num_ports = self.owned_ports.len();
         let parent_base_idx = parent_branch_idx as usize * num_ports;
         let new_base_idx = new_branch_idx as usize * num_ports;
         debug_assert!(parent_branch_idx < new_branch_idx);
         debug_assert!(new_base_idx == self.port_mapping.len());
         self.port_mapping.reserve(num_ports);
         for offset in 0..num_ports {
             let parent_port = &self.port_mapping[parent_base_idx + offset];
             self.port_mapping.push(parent_port.clone());
             let parent_port = parent_port.clone();
             self.port_mapping.push(parent_port);
+        }
+    }
     /// Adds a new port. Caller must make sure that the connector is not in the
     /// sync phase.
     fn add_port(&mut self, port_id: PortIdLocal) {
         debug_assert!(self.port_mapping.len() == self.owned_ports.len());
         debug_assert!(!self.owned_ports.contains(&port_id));
         self.owned_ports.push(port_id);
         self.port_mapping.push(PortAssignment::new_unassigned());
+    }
     /// Commits to a particular branch. Essentially just removes the port
     /// mapping information generated during the sync phase.
     fn commit_to_sync(&mut self) {
         self.port_mapping.truncate(self.owned_ports.len());
         debug_assert!(self.port_mapping.iter().all(|v| {
             !v.is_assigned && !v.last_registered_branch_id.is_valid()
         }));
+    }
     /// Removes a particular port from the connector. May only be done if the
     /// connector is in non-sync mode
     fn remove_port(&mut self, port_id: PortIdLocal) {
         debug_assert!(self.port_mapping.len() == self.owned_ports.len()); // in non-sync mode
         let port_index = self.get_port_index(port_id).unwrap();
         self.owned_ports.remove(port_index);
         self.port_mapping.remove(port_index);
+    }
     /// Retrieves the index associated with a port id. Note that the port might
     /// not exist (yet) if a speculative branch has just received the port.
     /// TODO: But then again, one cannot use that port, right?
     #[inline]
     fn get_port_index(&self, port_id: PortIdLocal) -> Option<usize> {
         for (idx, port) in self.owned_ports.iter().enumerate() {
             if *port == port_id {
                 return Some(idx)
+            }
+        }
         return None
+    }
     /// Retrieves the ID associated with the port at the provided index
     #[inline]
     fn get_port_id(&self, port_index: usize) -> PortIdLocal {
         return self.owned_ports[port_index];
+    }
     #[inline]
     fn get_port(&self, branch_idx: u32, port_idx: usize) -> &PortAssignment {
         let mapped_idx = self.mapped_index(branch_idx, port_idx);
         return &self.port_mapping[mapped_idx];
+    }
     #[inline]
     fn get_port_mut(&mut self, branch_idx: u32, port_idx: usize) -> &mut PortAssignment {
         let mapped_idx = self.mapped_index(branch_idx, port_idx);
         return &mut self.port_mapping[mapped_idx];
+    }
     #[inline]
     fn num_ports(&self) -> usize {
         return self.owned_ports.len();
+    }
     // Function for internal use: retrieve index in flattened port mapping array
     // based on branch/port index.
     #[inline]
     fn mapped_index(&self, branch_idx: u32, port_idx: usize) -> usize {
         let branch_idx = branch_idx as usize;
         let num_ports = self.owned_ports.len();
         debug_assert!(port_idx < num_ports);
         debug_assert!((branch_idx + 1) * num_ports <= self.port_mapping.len());
         return branch_idx * num_ports + port_idx;
+    }
+}
 struct BranchQueue {
     first: u32,
     last: u32,
+}
 impl BranchQueue {
     #[inline]
     fn new() -> Self {
         Self{ first: 0, last: 0 }
+    }
     #[inline]
     fn is_empty(&self) -> bool {
         debug_assert!((self.first == 0) == (self.last == 0));
         return self.first == 0;
+    }
     #[inline]
     fn clear(&mut self) {
         self.first = 0;
         self.last = 0;
+    }
+}
 /// Public fields of the connector that can be freely shared between multiple
 /// threads.
 pub(crate) struct ConnectorPublic {
     pub inbox: PublicInbox,
     pub sleeping: AtomicBool,
+}
 impl ConnectorPublic {
     pub fn new() -> Self {
+    pub fn new(initialize_as_sleeping: bool) -> Self {
         ConnectorPublic{
             inbox: PublicInbox::new(),
-            sleeping: AtomicBool::new(false),
+            sleeping: AtomicBool::new(initialize_as_sleeping),
+        }
+    }
+}
 // TODO: Maybe prevent false sharing by aligning `public` to next cache line.
 // TODO: Do this outside of the connector, create a wrapping struct
 pub(crate) struct ConnectorPDL {
     // State and properties of connector itself
     in_sync: bool,
     // Branch management
     branches: Vec<Branch>, // first branch is always non-speculative one
     sync_active: BranchQueue,
     sync_pending_get: BranchQueue,
     sync_finished: BranchQueue,
     sync_finished_last_handled: u32, // TODO: Change to BranchId?
     cur_round: u32,
     // Port/message management
     pub committed_to: Option<(ConnectorId, u64)>,
     pub inbox: PrivateInbox,
     pub ports: ConnectorPorts,
+}
 struct TempCtx {}
 impl RunContext for TempCtx {
     fn did_put(&mut self, port: PortId) -> bool {
         todo!()
+    }
     fn get(&mut self, port: PortId) -> Option<ValueGroup> {
         todo!()
+    }
     fn fires(&mut self, port: PortId) -> Option<Value> {
         todo!()
+    }
     fn get_channel(&mut self) -> Option<(Value, Value)> {
         todo!()
+    }
+}
 impl Connector for ConnectorPDL {
-    fn handle_message(&mut self, message: MessageContents, ctx: &ConnectorCtx, delta_state: &mut RunDeltaState) {
+    fn handle_message(&mut self, message: Message, ctx: &ConnectorCtx, delta_state: &mut RunDeltaState) {
         use MessageContents as MC;
         match message {
             MC::Data(message) => self.handle_data_message(message),
             MC::Sync(message) => self.handle_sync_message(message, ctx, delta_state),
             MC::RequestCommit(message) => self.handle_request_commit_message(message, ctx, delta_state),
             MC::ConfirmCommit(message) => self.handle_confirm_commit_message(message, ctx, delta_state),
         match message.contents {
             MC::Data(content) => self.handle_data_message(message.receiving_port, content),
             MC::Sync(content) => self.handle_sync_message(content, ctx, delta_state),
             MC::RequestCommit(content) => self.handle_request_commit_message(content, ctx, delta_state),
             MC::ConfirmCommit(content) => self.handle_confirm_commit_message(content, ctx, delta_state),
             MC::Control(_) | MC::Ping => {},
+        }
+    }
     fn run(&mut self, pd: &ProtocolDescription, ctx: &ConnectorCtx, delta_state: &mut RunDeltaState) -> ConnectorScheduling {
         if self.in_sync {
             let scheduling = self.run_in_speculative_mode(pd, ctx, delta_state);
             // When in speculative mode we might have generated new sync
             // solutions, we need to turn them into proposed solutions here.
             if self.sync_finished_last_handled != self.sync_finished.last {
                 // Retrieve first element in queue
                 let mut next_id;
                 if self.sync_finished_last_handled == 0 {
                     next_id = self.sync_finished.first;
                 } else {
                     let last_handled = &self.branches[self.sync_finished_last_handled as usize];
                     debug_assert!(last_handled.next_branch_in_queue.is_some()); // because "last handled" != "last in queue"
                     next_id = last_handled.next_branch_in_queue.unwrap();
+                }
                 loop {
                     let branch_id = BranchId::new(next_id);
                     let branch = &self.branches[next_id as usize];
                     let branch_next = branch.next_branch_in_queue;
                     // Turn local solution into a message and send it along
                     // TODO: Like `ports` access, also revise the construction of this `key`, should not be needed
                     let solution_message = self.generate_initial_solution_for_branch(branch_id, ctx);
                     if let Some(valid_solution) = solution_message {
                         self.submit_sync_solution(valid_solution, ctx, delta_state);
                     } else {
                         // Branch is actually invalid, but we only just figured
                         // it out. We need to mark it as invalid to prevent
                         // future use
                         Self::remove_branch_from_queue(&mut self.branches, &mut self.sync_finished, branch_id);
                         if branch_id.index == self.sync_finished_last_handled {
                             self.sync_finished_last_handled = self.sync_finished.last;
+                        }
                         let branch = &mut self.branches[next_id as usize];
                         branch.sync_state = SpeculativeState::Inconsistent;
+                    }
                     match branch_next {
                         Some(id) => next_id = id,
                         None => break,
+                    }
+                }
                 self.sync_finished_last_handled = next_id;
+            }
             return scheduling;
         } else {
             let scheduling = self.run_in_deterministic_mode(pd, ctx, delta_state);
             return scheduling;
+        }
+    }
+}
 impl ConnectorPDL {
     /// Constructs a representation of a connector. The assumption is that the
     /// initial branch is at the first instruction of the connector's code,
     /// hence is in a non-sync state.
     pub fn new(initial_branch: Branch, owned_ports: Vec<PortIdLocal>) -> Self {
         Self{
             in_sync: false,
             branches: vec![initial_branch],
             sync_active: BranchQueue::new(),
             sync_pending_get: BranchQueue::new(),
             sync_finished: BranchQueue::new(),
             sync_finished_last_handled: 0, // none at all
             cur_round: 0,
             committed_to: None,
             inbox: PrivateInbox::new(),
             ports: ConnectorPorts::new(owned_ports),
+        }
+    }
     pub fn is_in_sync_mode(&self) -> bool {
         return self.in_sync;
+    }
     // -------------------------------------------------------------------------
     // Handling connector messages
     // -------------------------------------------------------------------------
     #[inline]
     pub fn handle_data_message(&mut self, message: DataMessage) {
         self.inbox.insert_message(message);
     pub fn handle_data_message(&mut self, target_port: PortIdLocal, message: DataMessage) {
         self.inbox.insert_message(target_port, message);
+    }
     /// Accepts a synchronous message and combines it with the locally stored
     /// solution(s). Then queue new `Sync`/`Solution` messages when appropriate.
     pub fn handle_sync_message(&mut self, message: SyncMessage, ctx: &ConnectorCtx, results: &mut RunDeltaState) {
         debug_assert!(!message.to_visit.contains(&ctx.id)); // own ID already removed
         debug_assert!(message.constraints.iter().any(|v| v.connector_id == ctx.id)); // we have constraints
         // TODO: Optimize, use some kind of temp workspace vector
         let mut execution_path_branch_ids = Vec::new();
         if self.sync_finished_last_handled != 0 {
             // We have some solutions to match against
             let constraints_index = message.constraints
                 .iter()
                 .position(|v| v.connector_id == ctx.id)
                 .unwrap();
             let constraints = &message.constraints[constraints_index].constraints;
             debug_assert!(!constraints.is_empty());
             // Note that we only iterate over the solutions we've already
             // handled ourselves, not necessarily
             let mut branch_index = self.sync_finished.first;
             'branch_loop: loop {
                 // Load solution branch
                 let branch = &self.branches[branch_index as usize];
                 execution_path_branch_ids.clear();
                 self.branch_ids_of_execution_path(BranchId::new(branch_index), &mut execution_path_branch_ids);
                 // Check if the branch matches all of the applied constraints
                 for constraint in constraints {
                     match constraint {
                         SyncBranchConstraint::SilentPort(silent_port_id) => {
                             let port_index = self.ports.get_port_index(*silent_port_id);
                             if port_index.is_none() {
                                 // Nefarious peer
                                 continue 'branch_loop;
+                            }
                             let port_index = port_index.unwrap();
                             let mapping = self.ports.get_port(branch_index, port_index);
                             debug_assert!(mapping.is_assigned);
                             if mapping.num_times_fired != 0 {
                                 // Not silent, constraint not satisfied
                                 continue 'branch_loop;
+                            }
                         },
                         SyncBranchConstraint::BranchNumber(expected_branch_id) => {
                             if !execution_path_branch_ids.contains(expected_branch_id) {
                                 // Not the expected execution path, constraint not satisfied
                                 continue 'branch_loop;
+                            }
                         },
                         SyncBranchConstraint::PortMapping(port_id, expected_branch_id) => {
                             let port_index = self.ports.get_port_index(*port_id);
                             if port_index.is_none() {
                                 // Nefarious peer
                                 continue 'branch_loop;
+                            }
                             let port_index = port_index.unwrap();
                             let mapping = self.ports.get_port(branch_index, port_index);
                             if mapping.last_registered_branch_id != *expected_branch_id {
                                 // Not the expected interaction on this port, constraint not satisfied
                                 continue 'branch_loop;
+                            }
                         },
+                    }
+                }
                 // If here, then all of the external constraints were satisfied
                 // for the current branch. But the branch itself also imposes
                 // constraints. So while building up the new solution, make sure
                 // that those are satisfied as well.
                 // TODO: Code below can probably be merged with initial solution
                 //  generation.
                 // - clone old solution so we can add to it
                 let mut new_solution = message.clone();
                 // - determine the initial port mapping
                 let num_ports = self.ports.num_ports();
                 let mut new_solution_mapping = Vec::with_capacity(num_ports);
                 for port_index in 0..self.ports.num_ports() {
                     let port_id = self.ports.get_port_id(port_index);
                     let mapping = self.ports.get_port(branch_index, port_index);
                     new_solution_mapping.push((port_id, mapping.last_registered_branch_id));
+                }
                 // - replace constraints with a local solution
                 new_solution.constraints.remove(constraints_index);
                 new_solution.local_solutions.push(SyncConnectorSolution{
                     connector_id: ctx.id,
                     terminating_branch_id: BranchId::new(branch_index),
                     execution_branch_ids: execution_path_branch_ids.clone(),
                     final_port_mapping: new_solution_mapping,
                 });
                 // - do a second pass on the ports to generate and add the
                 //   constraints that should be applied to other connectors
                 for port_index in 0..self.ports.num_ports() {
                     let port_id = self.ports.get_port_id(port_index);
                     let (peer_connector_id, peer_port_id, peer_is_getter) = {
                         let port = ctx.get_port(port_id);
                         (port.peer_connector, port.peer_id, port.kind == PortKind::Putter)
                     };
                     let mapping = self.ports.get_port(branch_index, port_index);
                     let constraint = if mapping.num_times_fired == 0 {
                         SyncBranchConstraint::SilentPort(peer_port_id)
                     } else {
                         if peer_is_getter {
                             SyncBranchConstraint::PortMapping(peer_port_id, mapping.last_registered_branch_id)
                         } else {
                             SyncBranchConstraint::BranchNumber(mapping.last_registered_branch_id)
+                        }
                     };
                     match new_solution.add_or_check_constraint(peer_connector_id, constraint) {
                         Err(_) => continue 'branch_loop,
                         Ok(false) => continue 'branch_loop,
                         Ok(true) => {},
+                    }
+                }
                 // If here, then the newly generated solution is completely
                 // compatible.
                 let next_branch = branch.next_branch_in_queue;
                 self.submit_sync_solution(new_solution, ctx, results);
                 // Consider the next branch
                 if branch_index == self.sync_finished_last_handled {
                     // At the end of the previously handled solutions
                     break;
+                }
                 debug_assert!(branch.next_branch_in_queue.is_some()); // because we cannot be at the end of the queue
                 branch_index = branch.next_branch_in_queue.unwrap();
                 debug_assert!(next_branch.is_some()); // because we cannot be at the end of the queue
                 branch_index = next_branch.unwrap();
+            }
+        }
+    }
     fn handle_request_commit_message(&mut self, mut message: SolutionMessage, ctx: &ConnectorCtx, delta_state: &mut RunDeltaState) {
         let should_propagate_message = match &self.committed_to {
             Some((previous_origin, previous_comparison)) => {
                 // Already committed to something. So will commit to this if it
                 // takes precedence over the current solution
                 message.comparison_number > *previous_comparison ||
                     (message.comparison_number == *previous_comparison && message.connector_origin.0 > previous_origin.0)
             },
             None => {
                 // Not yet committed to a solution, so commit to this one
                 true
+            }
         };
         if should_propagate_message {
             self.committed_to = Some((message.connector_origin, message.comparison_number));
             if message.to_visit.is_empty() {
                 // Visited all of the connectors, so every connector can now
                 // apply the solution
                 // TODO: Use temporary workspace
                 let mut to_visit = Vec::with_capacity(message.local_solutions.len() - 1);
                 for (connector_id, _) in &message.local_solutions {
                     if *connector_id != ctx.id {
                         to_visit.push(*connector_id);
+                    }
+                }
                 message.to_visit = to_visit;
                 self.handle_confirm_commit_message(message.clone(), ctx, delta_state);
                 delta_state.outbox.push(MessageContents::ConfirmCommit(message));
             } else {
                 // Not yet visited all of the connectors
                 delta_state.outbox.push(MessageContents::RequestCommit(message));
+            }
+        }
+    }
-    fn handle_confirm_commit_message(&mut self, message: SolutionMessage, ctx: &ConnectorCtx, delta_state: &mut RunDeltaState) {
+    fn handle_confirm_commit_message(&mut self, message: SolutionMessage, ctx: &ConnectorCtx, _delta_state: &mut RunDeltaState) {
         // Make sure this is the message we actually committed to. As long as
         // we're running on a single machine this is fine.
         // TODO: Take care of nefarious peers
         let (expected_connector_id, expected_comparison_number) =
             self.committed_to.unwrap();
         assert_eq!(message.connector_origin, expected_connector_id);
         assert_eq!(message.comparison_number, expected_comparison_number);
         // Find the branch we're supposed to commit to
         let (_, branch_id) = message.local_solutions
             .iter()
             .find(|(id, _)| *id == ctx.id)
             .unwrap();
         let branch_id = *branch_id;
         // Commit to the branch. That is: move the solution branch to the first
         // of the connector's branches
         self.in_sync = false;
         self.branches.swap(0, branch_id.index as usize);
         self.branches.truncate(1); // TODO: Or drain and do not deallocate?
         let solution = &mut self.branches[0];
         // Clear all of the other sync-related variables
         self.sync_active.clear();
         self.sync_pending_get.clear();
         self.sync_finished.clear();
         self.sync_finished_last_handled = 0;
         self.cur_round += 1;
         self.committed_to = None;
         self.inbox.clear();
         self.ports.commit_to_sync();
         // Add/remove any of the ports we lost during the sync phase
         for port_delta in &solution.ports_delta {
             if port_delta.acquired {
                 self.ports.add_port(port_delta.port_id);
             } else {
                 self.ports.remove_port(port_delta.port_id);
+            }
+        }
         solution.commit_to_sync();
+    }
     // -------------------------------------------------------------------------
     // Executing connector code
     // -------------------------------------------------------------------------
     /// Runs the connector in synchronous mode. Potential changes to the global
     /// system's state are added to the `RunDeltaState` object by the connector,
     /// where it is the caller's responsibility to immediately take care of
     /// those changes. The return value indicates when (and if) the connector
     /// needs to be scheduled again.
-    pub fn run_in_speculative_mode(&mut self, pd: &ProtocolDescription, context: &ConnectorCtx, results: &mut RunDeltaState) -> ConnectorScheduling {
+    pub fn run_in_speculative_mode(&mut self, pd: &ProtocolDescription, _context: &ConnectorCtx, results: &mut RunDeltaState) -> ConnectorScheduling {
         debug_assert!(self.in_sync);
         debug_assert!(!self.sync_active.is_empty());
         let branch = Self::pop_branch_from_queue(&mut self.branches, &mut self.sync_active);
         // Run the branch to the next blocking point
         let mut run_context = TempCtx{};
         let run_result = branch.code_state.run(&mut run_context, pd);
         // Match statement contains `return` statements only if the particular
         // run result behind handled requires an immediate re-run of the
         // connector.
         match run_result {
             RunResult::BranchInconsistent => {
                 // Speculative branch became inconsistent
                 branch.sync_state = SpeculativeState::Inconsistent;
             },
             RunResult::BranchMissingPortState(port_id) => {
                 // Branch called `fires()` on a port that does not yet have an
                 // assigned speculative value. So we need to create those
                 // branches
                 let local_port_id = PortIdLocal::new(port_id.0.u32_suffix);
                 let local_port_index = self.ports.get_port_index(local_port_id).unwrap();
                 debug_assert!(self.ports.owned_ports.contains(&local_port_id));
                 let silent_branch = &*branch;
                 // Create a copied branch who will have the port set to firing
                 let firing_index = self.branches.len() as u32;
                 let mut firing_branch = Branch::new_sync_branching_from(firing_index, silent_branch);
                 self.ports.prepare_sync_branch(branch.index.index, firing_index);
                 // Create two copied branches, one silent and one firing
                 branch.sync_state = SpeculativeState::HaltedAtBranchPoint;
                 let parent_branch_id = branch.index;
                 let parent_branch = &self.branches[parent_branch_id.index as usize];
                 let firing_port = self.ports.get_port_mut(firing_index, local_port_index);
                 firing_port.mark_speculative(1);
                 let silent_index = self.branches.len() as u32;
                 let firing_index = silent_index + 1;
                 let silent_branch = Branch::new_sync_branching_from(silent_index, parent_branch);
                 self.ports.prepare_sync_branch(parent_branch.index.index, silent_index);
                 // Assign the old branch a silent value
                 let silent_port = self.ports.get_port_mut(silent_branch.index.index, local_port_index);
                 let firing_branch = Branch::new_sync_branching_from(firing_index, parent_branch);
                 self.ports.prepare_sync_branch(parent_branch.index.index, firing_index);
                 // Assign the port values of the two new branches
                 let silent_port = self.ports.get_port_mut(silent_index, local_port_index);
                 silent_port.mark_speculative(0);
                 let firing_port = self.ports.get_port_mut(firing_index, local_port_index);
                 firing_port.mark_speculative(1);
                 // Run both branches again
                 Self::push_branch_into_queue(&mut self.branches, &mut self.sync_active, silent_branch.index);
                 Self::push_branch_into_queue(&mut self.branches, &mut self.sync_active, firing_branch.index);
                 let silent_branch_id = silent_branch.index;
                 self.branches.push(silent_branch);
                 let firing_branch_id = firing_branch.index;
                 self.branches.push(firing_branch);
                 Self::push_branch_into_queue(&mut self.branches, &mut self.sync_active, silent_branch_id);
                 Self::push_branch_into_queue(&mut self.branches, &mut self.sync_active, firing_branch_id);
                 return ConnectorScheduling::Immediate;
             },
             RunResult::BranchMissingPortValue(port_id) => {
                 // Branch performed a `get` on a port that has not yet received
                 // a value in its inbox.
                 let local_port_id = PortIdLocal::new(port_id.0.u32_suffix);
                 let local_port_index = self.ports.get_port_index(local_port_id);
                 if local_port_index.is_none() {
                     todo!("deal with the case where the port is acquired");
+                }
                 let local_port_index = local_port_index.unwrap();
                 let port_mapping = self.ports.get_port_mut(branch.index.index, local_port_index);
                 // Check for port mapping assignment and, if present, if it is
                 // consistent
                 let is_valid_get = if port_mapping.is_assigned {
                     assert!(port_mapping.num_times_fired <= 1); // temporary, until we get rid of `fires`
                     port_mapping.num_times_fired == 1
                 } else {
                     // Not yet assigned
                     port_mapping.mark_speculative(1);
                     true
                 };
                 if is_valid_get {
                     // Mark as a branching point for future messages
                     branch.sync_state = SpeculativeState::HaltedAtBranchPoint;
                     Self::push_branch_into_queue(&mut self.branches, &mut self.sync_pending_get, branch.index);
                     let branch_id = branch.index;
                     Self::push_branch_into_queue(&mut self.branches, &mut self.sync_pending_get, branch_id);
                     // But if some messages can be immediately applied, do so
                     // now.
                     let messages = self.inbox.get_messages(local_port_id, port_mapping.last_registered_branch_id);
                     let mut did_have_messages = false;
                     for message in messages {
                         did_have_messages = true;
                         // For each message prepare a new branch to execute
                         let parent_branch = &self.branches[branch_id.index as usize];
                         let new_branch_index = self.branches.len() as u32;
                         let mut new_branch = Branch::new_sync_branching_from(new_branch_index, branch);
                         self.ports.prepare_sync_branch(branch.index.index, new_branch_index);
                         let mut new_branch = Branch::new_sync_branching_from(new_branch_index, parent_branch);
                         self.ports.prepare_sync_branch(branch_id.index, new_branch_index);
                         let port_mapping = self.ports.get_port_mut(new_branch_index, local_port_index);
                         port_mapping.last_registered_branch_id = message.sender_cur_branch_id;
                         debug_assert!(port_mapping.is_assigned && port_mapping.num_times_fired == 1);
                         new_branch.received.insert(local_port_id, message.clone());
                         // If the message contains any ports then they will now
                         // be owned by the new branch
                         debug_assert!(results.ports.is_empty());
                         find_ports_in_value_group(&message.message, &mut results.ports);
                         Self::acquire_ports_during_sync(&mut self.ports, &mut new_branch, &results.ports);
                         results.ports.clear();
                         // Schedule the new branch
                         debug_assert!(new_branch.sync_state == SpeculativeState::RunningInSync);
-                        Self::push_branch_into_queue(&mut self.branches, &mut self.sync_active, new_branch.index);
+                        let new_branch_id = new_branch.index;
                         self.branches.push(new_branch);
                         Self::push_branch_into_queue(&mut self.branches, &mut self.sync_active, new_branch_id);
+                    }
                     if did_have_messages {
                         // If we did create any new branches, then we can run
                         // them immediately.
                         return ConnectorScheduling::Immediate;
+                    }
                 } else {
                     branch.sync_state = SpeculativeState::Inconsistent;
+                }
             },
             RunResult::BranchAtSyncEnd => {
                 // Branch is done, go through all of the ports that are not yet
                 // assigned and map them to non-firing.
                 for port_idx in 0..self.ports.num_ports() {
                     let port_mapping = self.ports.get_port_mut(branch.index.index, port_idx);
                     if !port_mapping.is_assigned {
                         port_mapping.mark_speculative(0);
+                    }
+                }
                 let branch_id = branch.index;
                 branch.sync_state = SpeculativeState::ReachedSyncEnd;
-                Self::push_branch_into_queue(&mut self.branches, &mut self.sync_finished, branch.index);
+                Self::push_branch_into_queue(&mut self.branches, &mut self.sync_finished, branch_id);
             },
             RunResult::BranchPut(port_id, value_group) => {
                 // Branch performed a `put` on a particualar port.
                 let local_port_id = PortIdLocal{ index: port_id.0.u32_suffix };
                 let local_port_index = self.ports.get_port_index(local_port_id);
                 if local_port_index.is_none() {
                     todo!("handle case where port was received before (i.e. in ports_delta)")
+                }
                 let local_port_index = local_port_index.unwrap();
                 // Check the port mapping for consistency
                 // TODO: For now we can only put once, so that simplifies stuff
                 let port_mapping = self.ports.get_port_mut(branch.index.index, local_port_index);
                 let is_valid_put = if port_mapping.is_assigned {
                     // Already assigned, so must be speculative and one time
                     // firing, otherwise we are `put`ing multiple times.
                     if port_mapping.last_registered_branch_id.is_valid() {
                         // Already did a `put`
                         todo!("handle error through RunDeltaState");
                     } else {
                         // Valid if speculatively firing
                         port_mapping.num_times_fired == 1
+                    }
                 } else {
                     // Not yet assigned, do so now
                     true
                 };
                 if is_valid_put {
                     // Put in run results for thread to pick up and transfer to
                     // the correct connector inbox.
                     port_mapping.mark_definitive(branch.index, 1);
                     let message = DataMessage{
                         sending_port: local_port_id,
                         sender_prev_branch_id: BranchId::new_invalid(),
                         sender_cur_branch_id: branch.index,
                         message: value_group,
                     };
                     // If the message contains any ports then we release our
                     // ownership over them in this branch
                     debug_assert!(results.ports.is_empty());
                     find_ports_in_value_group(&message.message, &mut results.ports);
                     Self::release_ports_during_sync(&mut self.ports, branch, &results.ports);
+                    Self::release_ports_during_sync(&mut self.ports, branch, &results.ports).unwrap();
                     results.ports.clear();
                     results.outbox.push(MessageContents::Data(message));
                     return ConnectorScheduling::Immediate
                 } else {
                     branch.sync_state = SpeculativeState::Inconsistent;
+                }
             },
             _ => unreachable!("unexpected run result '{:?}' while running in sync mode", run_result),
+        }
         // Not immediately scheduling, so schedule again if there are more
         // branches to run
         if self.sync_active.is_empty() {
             return ConnectorScheduling::NotNow;
         } else {
             return ConnectorScheduling::Later;
+        }
+    }
     /// Runs the connector in non-synchronous mode.
-    pub fn run_in_deterministic_mode(&mut self, pd: &ProtocolDescription, context: &ConnectorCtx, results: &mut RunDeltaState) -> ConnectorScheduling {
+    pub fn run_in_deterministic_mode(&mut self, pd: &ProtocolDescription, _context: &ConnectorCtx, results: &mut RunDeltaState) -> ConnectorScheduling {
         debug_assert!(!self.in_sync);
         debug_assert!(self.sync_active.is_empty() && self.sync_pending_get.is_empty() && self.sync_finished.is_empty());
         debug_assert!(self.branches.len() == 1);
         let branch = &mut self.branches[0];
         debug_assert!(branch.sync_state == SpeculativeState::RunningNonSync);
         let mut run_context = TempCtx{};
         let run_result = branch.code_state.run(&mut run_context, pd);
         match run_result {
             RunResult::ComponentTerminated => {
                 // Need to wait until all children are terminated
                 // TODO: Think about how to do this?
                 branch.sync_state = SpeculativeState::Finished;
                 return ConnectorScheduling::NotNow;
             },
             RunResult::ComponentAtSyncStart => {
                 // Prepare for sync execution and reschedule immediately
                 self.in_sync = true;
                 let first_sync_branch = Branch::new_sync_branching_from(1, branch);
-                Self::push_branch_into_queue(&mut self.branches, &mut self.sync_active, first_sync_branch.index);
+                let first_sync_branch_id = first_sync_branch.index;
                 self.branches.push(first_sync_branch);
                 Self::push_branch_into_queue(&mut self.branches, &mut self.sync_active, first_sync_branch_id);
                 return ConnectorScheduling::Later;
             },
             RunResult::NewComponent(definition_id, monomorph_idx, arguments) => {
                 // Construction of a new component. Find all references to ports
                 // inside of the arguments
                 debug_assert!(results.ports.is_empty());
                 find_ports_in_value_group(&arguments, &mut results.ports);
                 if !results.ports.is_empty() {
                     // Ports changing ownership
                     if let Err(_) = Self::release_ports_during_non_sync(&mut self.ports, branch, &results.ports) {
                         todo!("fatal error handling");
+                    }
+                }
                 // Add connector for later execution
                 let new_connector_state = ComponentState {
                     prompt: Prompt::new(&pd.types, &pd.heap, definition_id, monomorph_idx, arguments)
                 };
                 let new_connector_ports = results.ports.clone(); // TODO: Do something with this
                 let new_connector_branch = Branch::new_initial_branch(new_connector_state);
                 let new_connector = ConnectorPDL::new(new_connector_branch, new_connector_ports);
                 results.new_connectors.push(new_connector);
                 return ConnectorScheduling::Later;
             },
             RunResult::NewChannel => {
                 // Need to prepare a new channel
                 todo!("adding channels to some global context");
                 return ConnectorScheduling::Later;
             },
             _ => unreachable!("unexpected run result '{:?}' while running in non-sync mode", run_result),
+        }
+    }
     // -------------------------------------------------------------------------
     // Internal helpers
     // -------------------------------------------------------------------------
     // Helpers for management of the branches and their internally stored
     // `next_branch_in_queue` and the `BranchQueue` objects. Essentially forming
     // linked lists inside of the vector of branches.
     /// Pops from front of linked-list branch queue.
     fn pop_branch_from_queue<'a>(branches: &'a mut Vec<Branch>, queue: &mut BranchQueue) -> &'a mut Branch {
         debug_assert!(queue.first != 0);
         let branch = &mut branches[queue.first as usize];
         queue.first = branch.next_branch_in_queue.unwrap_or(0);
         branch.next_branch_in_queue = None;
         if queue.first == 0 {
             // No more entries in queue
             debug_assert_eq!(queue.last, branch.index.index);
             queue.last = 0;
+        }
         return branch;
+    }
     /// Pushes branch at the end of the linked-list branch queue.
     fn push_branch_into_queue(
         branches: &mut Vec<Branch>, queue: &mut BranchQueue, to_push: BranchId,
     ) {
         debug_assert!(to_push.is_valid());
         let to_push = to_push.index;
         if queue.last == 0 {
             // No branches in the queue at all
             debug_assert_eq!(queue.first, 0);
             branches[to_push as usize].next_branch_in_queue = None;
             queue.first = to_push;
             queue.last = to_push;
         } else {
             // Pre-existing branch in the queue
             debug_assert_ne!(queue.first, 0);
             branches[queue.last as usize].next_branch_in_queue = Some(to_push);
             queue.last = to_push;
+        }
+    }
     /// Removes branch from linked-list queue. Walks through the entire list to
     /// find the element (!). Assumption is that this is not called often.
     fn remove_branch_from_queue(
         branches: &mut Vec<Branch>, queue: &mut BranchQueue, to_delete: BranchId,
     ) {
         debug_assert!(to_delete.is_valid()); // we're deleting a valid item
         debug_assert!(queue.first != 0 && queue.last != 0); // queue isn't empty to begin with
         // Retrieve branch and its next element
         let branch_to_delete = &mut branches[to_delete.index as usize];
         let branch_next_index_option = branch_to_delete.next_branch_in_queue;
         let branch_next_index_unwrapped = branch_next_index_option.unwrap_or(0);
         branch_to_delete.next_branch_in_queue = None;
         // Walk through all elements in queue to find branch to delete
         let mut prev_index = 0;
         let mut next_index = queue.first;
         while next_index != 0 {
             if next_index == to_delete.index {
                 // Found the element we're going to delete
                 // - check if at the first element or not
                 if prev_index == 0 {
                     queue.first = branch_next_index_unwrapped;
                 } else {
                     let prev_branch = &mut branches[prev_index as usize];
                     prev_branch.next_branch_in_queue = branch_next_index_option;
+                }
                 // - check if at last element or not (also takes care of "no elements left in queue")
                 if branch_next_index_option.is_none() {
                     queue.last = prev_index;
+                }
                 return;
+            }
             prev_index = next_index;
             let entry = &branches[next_index as usize];
             next_index = entry.next_branch_in_queue.unwrap_or(0);
+        }
         // If here, then we didn't find the element
         panic!("branch does not exist in provided queue");
+    }
     // Helpers for local port management. Specifically for adopting/losing
     // ownership over ports, and for checking if specific ports can be sent
     // over another port.
     /// Releasing ownership of ports while in non-sync mode. This only occurs
     /// while instantiating new connectors
     fn release_ports_during_non_sync(ports: &mut ConnectorPorts, branch: &mut Branch, port_ids: &[PortIdLocal]) -> Result<(), PortOwnershipError> {
         debug_assert!(!branch.index.is_valid()); // branch in non-sync mode
         for port_id in port_ids {
             // We must own the port, or something is wrong with our code
             todo!("Set up some kind of message router");
             debug_assert!(ports.get_port_index(*port_id).is_some());
             ports.remove_port(*port_id);
+        }
         return Ok(())
+    }
     /// Releasing ownership of ports during a sync-session. Will provide an
     /// error if the port was already used during a sync block.
     fn release_ports_during_sync(ports: &mut ConnectorPorts, branch: &mut Branch, port_ids: &[PortIdLocal]) -> Result<(), PortOwnershipError> {
         todo!("unfinished: add port properties during final solution-commit msgs");
         debug_assert!(branch.index.is_valid()); // branch in sync mode
         for port_id in port_ids {
             match ports.get_port_index(*port_id) {
                 Some(port_index) => {
                     // We (used to) own the port. Make sure it is not given away
                     // already and not used to put/get data.
                     let port_mapping = ports.get_port(branch.index.index, port_index);
                     if port_mapping.is_assigned && port_mapping.num_times_fired != 0 {
                         // Already used
                         return Err(PortOwnershipError::UsedInInteraction(*port_id));
+                    }
                     for delta in &branch.ports_delta {
                         if delta.port_id == *port_id {
                             // We cannot have acquired this port, because the
                             // call to `ports.get_port_index` returned an index.
                             debug_assert!(!delta.acquired);
                             return Err(PortOwnershipError::AlreadyGivenAway(*port_id));
+                        }
+                    }
                     branch.ports_delta.push(PortOwnershipDelta{
                         acquired: false,
                         port_id: *port_id,
                     });
                 },
                 None => {
                     // Not in port mapping, so we must have acquired it before,
                     // remove the acquirement.
                     let mut to_delete_index: isize = -1;
                     for (delta_idx, delta) in branch.ports_delta.iter().enumerate() {
                         if delta.port_id == *port_id {
                             debug_assert!(delta.acquired);
                             to_delete_index = delta_idx as isize;
                             break;
+                        }
+                    }
                     debug_assert!(to_delete_index != -1);
                     branch.ports_delta.remove(to_delete_index as usize);
+                }
+            }
+        }
         return Ok(())
+    }
     /// Acquiring ports during a sync-session.
     fn acquire_ports_during_sync(ports: &mut ConnectorPorts, branch: &mut Branch, port_ids: &[PortIdLocal]) -> Result<(), PortOwnershipError> {
         todo!("unfinished: add port properties during final solution-commit msgs");
         debug_assert!(branch.index.is_valid()); // branch in sync mode
         'port_loop: for port_id in port_ids {
             for (delta_idx, delta) in branch.ports_delta.iter().enumerate() {
                 if delta.port_id == *port_id {
                     if delta.acquired {
                         // Somehow already received this port.
                         // TODO: @security
                         todo!("take care of nefarious peers");
                     } else {
                         // Sending ports to ourselves
                         debug_assert!(ports.get_port_index(*port_id).is_some());
                         branch.ports_delta.remove(delta_idx);
                         continue 'port_loop;
+                    }
+                }
+            }
             // If here then we can safely acquire the new port
             branch.ports_delta.push(PortOwnershipDelta{
                 acquired: true,
                 port_id: *port_id,
             });
+        }
         return Ok(())
+    }
     // Helpers for generating and handling sync messages (and the solutions that
     // are described by those sync messages)
     /// Generates the initial solution for a finished sync branch. If initial
     /// local solution is valid, then the appropriate message is returned.
     /// Otherwise the initial solution is inconsistent.
     fn generate_initial_solution_for_branch(&self, branch_id: BranchId, ctx: &ConnectorCtx) -> Option<SyncMessage> {
         // Retrieve branchg
         debug_assert!(branch_id.is_valid()); // because we're supposed to be in sync mode
         let branch = &self.branches[branch_id.index as usize];
         debug_assert_eq!(branch.sync_state, SpeculativeState::ReachedSyncEnd);
         // Set up storage (this is also the storage for all of the connectors
         // that will be visited, hence the initial size approximation)
         let mut all_branch_ids = Vec::new();
         self.branch_ids_of_execution_path(branch_id, &mut all_branch_ids);
         let num_ports = self.ports.num_ports();
         let approximate_peers = num_ports;
         let mut initial_solution_port_mapping = Vec::with_capacity(num_ports);
         for port_idx in 0..self.ports.num_ports() {
             let port_id = self.ports.get_port_id(port_idx);
             let port_desc = self.ports.get_port(branch_id.index, port_idx);
             // Note: if assigned then we expect a valid branch ID. Otherwise we have the "invalid
             // branch" as ID, marking that we want it to be silent
             debug_assert!(port_desc.is_assigned == port_desc.last_registered_branch_id.is_valid());
             initial_solution_port_mapping.push((port_id, port_desc.last_registered_branch_id));
+        }
         let initial_local_solution = SyncConnectorSolution{
             connector_id: ctx.id,
             terminating_branch_id: branch_id,
             execution_branch_ids: all_branch_ids,
             final_port_mapping: initial_solution_port_mapping,
         };
         let mut sync_message = SyncMessage::new(initial_local_solution, approximate_peers);
         // Turn local port mapping into constraints on other connectors
         // - constraints on other components due to transferred ports
         for port_delta in &branch.ports_delta {
             // For transferred ports we always have two constraints: one for the
             // sender and one for the receiver, ensuring it was not used.
             // TODO: This will fail if a port is passed around multiple times.
             //  maybe a special "passed along" entry in `ports_delta`.
             if !sync_message.check_constraint(ctx.id, SyncBranchConstraint::SilentPort(port_delta.port_id)).unwrap() {
                 return None;
+            }
             // Might need to check if we own the other side of the channel
             let port = ctx.get_port(port_delta.port_id);
             if !sync_message.add_or_check_constraint(port.peer_connector, SyncBranchConstraint::SilentPort(port.peer_id)).unwrap() {
                 return None;
+            }
+        }
         // - constraints on other components due to owned ports
         for port_index in 0..self.ports.num_ports() {
             let port_id = self.ports.get_port_id(port_index);
             let port_mapping = self.ports.get_port(branch_id.index, port_index);
             let port = ctx.get_port(port_id);
             let constraint = if port_mapping.is_assigned {
                 if port.kind == PortKind::Getter {
                     SyncBranchConstraint::BranchNumber(port_mapping.last_registered_branch_id)
                 } else {
                     SyncBranchConstraint::PortMapping(port.peer_id, port_mapping.last_registered_branch_id)
+                }
             } else {
                 SyncBranchConstraint::SilentPort(port.peer_id)
             };
             if !sync_message.add_or_check_constraint(port.peer_connector, constraint).unwrap() {
                 return None;
+            }
+        }
         return Some(sync_message);
+    }
     fn submit_sync_solution(&mut self, partial_solution: SyncMessage, ctx: &ConnectorCtx, results: &mut RunDeltaState) {
         if partial_solution.to_visit.is_empty() {
             // Solution is completely consistent. So ask everyone to commit
             // TODO: Maybe another package for random?
             let comparison_number: u64 = unsafe {
                 let mut random_array = [0u8; 8];
                 getrandom::getrandom(&mut random_array);
+                getrandom::getrandom(&mut random_array).unwrap();
                 std::mem::transmute(random_array)
             };
             let num_local = partial_solution.local_solutions.len();
             let mut full_solution = SolutionMessage{
                 comparison_number,
                 connector_origin: ctx.id,
                 local_solutions: Vec::with_capacity(num_local),
                 to_visit: Vec::with_capacity(num_local - 1),
             };
             for local_solution in &partial_solution.local_solutions {
                 full_solution.local_solutions.push((local_solution.connector_id, local_solution.terminating_branch_id));
                 if local_solution.connector_id != ctx.id {
                     full_solution.to_visit.push(local_solution.connector_id);
+                }
+            }
             debug_assert!(self.committed_to.is_none());
             self.committed_to = Some((full_solution.connector_origin, full_solution.comparison_number));
             results.outbox.push(MessageContents::RequestCommit(full_solution));
         } else {
             // Still have connectors to visit
             results.outbox.push(MessageContents::Sync(partial_solution));
+        }
+    }
     fn branch_ids_of_execution_path(&self, leaf_branch_id: BranchId, parents: &mut Vec<BranchId>) {
         debug_assert!(parents.is_empty());
         let mut next_branch_id = leaf_branch_id;
         debug_assert!(next_branch_id.is_valid());
         while next_branch_id.is_valid() {
             parents.push(next_branch_id);
             let branch = &self.branches[next_branch_id.index as usize];
             next_branch_id = branch.parent_index;
+        }
+    }
+}
 /// A data structure passed to a connector whose code is being executed that is
 /// used to queue up various state changes that have to be applied after
 /// running, e.g. the messages the have to be transferred to other connectors.
 // TODO: Come up with a better name
 pub(crate) struct RunDeltaState {
     // Variables that allow the thread running the connector to pick up global
     // state changes and try to apply them.
     pub outbox: Vec<MessageContents>,
     pub new_connectors: Vec<ConnectorPDL>,
     pub new_ports: Vec<Port>,
     // Workspaces
     pub ports: Vec<PortIdLocal>,
+}
 impl RunDeltaState {
     /// Constructs a new `RunDeltaState` object with the default amount of
     /// reserved memory
     pub fn new() -> Self {
         RunDeltaState{
             outbox: Vec::with_capacity(64),
             new_connectors: Vec::new(),
             new_ports: Vec::new(),
             ports: Vec::with_capacity(64),
+        }
+    }
+}
 #[derive(Eq, PartialEq)]
 pub(crate) enum ConnectorScheduling {
     Immediate,      // Run again, immediately
     Later,          // Schedule for running, at some later point in time
     NotNow,         // Do not reschedule for running
+}
 /// 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<PortIdLocal>) {
     // Helper to check a value for a port and recurse if needed.
     fn find_port_in_value(group: &ValueGroup, value: &Value, ports: &mut Vec<PortIdLocal>) {
         match value {
             Value::Input(port_id) | Value::Output(port_id) => {
                 // This is an actual port
                 let cur_port = PortIdLocal::new(port_id.0.u32_suffix);
                 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);
+    }
+}
@@ \ No newline at end of file @@

src/runtime2/global_store.rs

➞

Show inline comments

 use std::ptr;
 use std::sync::{Arc, RwLock};
 use std::sync::atomic::{AtomicBool, AtomicU32};
 use crate::collections::{MpmcQueue, RawVec};
 use super::connector::{ConnectorPDL, ConnectorPublic};
 use super::scheduler::Router;
 use crate::ProtocolDescription;
 use crate::runtime2::connector::{ConnectorScheduling, RunDeltaState};
 use crate::runtime2::inbox::MessageContents;
 use crate::runtime2::native::{Connector, ConnectorApplication};
 use crate::runtime2::scheduler::ConnectorCtx;
 /// A kind of token that, once obtained, allows mutable access to a connector.
 /// We're trying to use move semantics as much as possible: the owner of this
 /// key is the only one that may execute the connector's code.
 pub(crate) struct ConnectorKey {
     pub index: u32, // of connector
+}
 impl ConnectorKey {
     /// Downcasts the `ConnectorKey` type, which can be used to obtain mutable
     /// access, to a "regular ID" which can be used to obtain immutable access.
     #[inline]
     pub fn downcast(&self) -> ConnectorId {
         return ConnectorId(self.index);
+    }
     /// Turns the `ConnectorId` into a `ConnectorKey`, marked as unsafe as it
     /// bypasses the type-enforced `ConnectorKey`/`ConnectorId` system
     #[inline]
     pub unsafe fn from_id(id: ConnectorId) -> ConnectorKey {
         return ConnectorKey{ index: id.0 };
+    }
+}
 /// A kind of token that allows shared access to a connector. Multiple threads
 /// may hold this
 #[derive(Debug, Copy, Clone, PartialEq, Eq)]
 pub(crate) struct ConnectorId(pub u32);
 use super::scheduler::{Router, ConnectorCtx};
 use super::connector::{ConnectorPDL, ConnectorPublic, ConnectorScheduling, RunDeltaState};
 use super::inbox::Message;
 use super::native::{Connector, ConnectorApplication};
 impl ConnectorId {
     // TODO: Like the other `new_invalid`, maybe remove
     #[inline]
     pub fn new_invalid() -> ConnectorId {
         return ConnectorId(u32::MAX);
+    }
     #[inline]
     pub(crate) fn is_valid(&self) -> bool {
         return self.0 != u32::MAX;
+    }
+}
 // TODO: Change this, I hate this. But I also don't want to put `public` and
 //  `router` of `ScheduledConnector` back into `Connector`. The reason I don't
 //  want `Box<dyn Connector>` everywhere is because of the v-table overhead. But
 //  to truly design this properly I need some benchmarks.
 pub enum ConnectorVariant {
     UserDefined(ConnectorPDL),
     Native(Box<dyn Connector>),
+}
 impl Connector for ConnectorVariant {
     fn handle_message(&mut self, message: MessageContents, ctx: &ConnectorCtx, delta_state: &mut RunDeltaState) {
         match self {
             ConnectorVariant::UserDefined(c) => c.handle_message(message, ctx, delta_state),
             ConnectorVariant::Native(c) => c.handle_message(message, ctx, delta_state),
+        }
+    }
     fn run(&mut self, protocol_description: &ProtocolDescription, ctx: &ConnectorCtx, delta_state: &mut RunDeltaState) -> ConnectorScheduling {
         match self {
             ConnectorVariant::UserDefined(c) => c.run(protocol_description, ctx, delta_state),
             ConnectorVariant::Native(c) => c.run(protocol_description, ctx, delta_state),
+        }
+    }
+}
 pub struct ScheduledConnector {
     pub connector: ConnectorVariant, // access by connector
     pub context: ConnectorCtx, // mutable access by scheduler, immutable by connector
     pub public: ConnectorPublic, // accessible by all schedulers and connectors
     pub router: Router,
+}
 /// The registry containing all connectors. The idea here is that when someone
 /// owns a `ConnectorKey`, then one has unique access to that connector.
 /// Otherwise one has shared access.
 ///
 /// This datastructure is built to be wrapped in a RwLock.
 pub(crate) struct ConnectorStore {
     pub(crate) port_counter: Arc<AtomicU32>,
     inner: RwLock<ConnectorStoreInner>,
+}
 struct ConnectorStoreInner {
     connectors: RawVec<*mut ScheduledConnector>,
     free: Vec<usize>,
+}
 impl ConnectorStore {
     fn with_capacity(capacity: usize) -> Self {
         return Self{
             port_counter: Arc::new(AtomicU32::new(0)),
             inner: RwLock::new(ConnectorStoreInner {
                 connectors: RawVec::with_capacity(capacity),
                 free: Vec::with_capacity(capacity),
             }),
         };
+    }
     /// Retrieves the shared members of the connector.
     pub(crate) fn get_shared(&self, connector_id: ConnectorId) -> &'static ConnectorPublic {
         let lock = self.inner.read().unwrap();
         unsafe {
             let connector = lock.connectors.get(connector_id.0 as usize);
             debug_assert!(!connector.is_null());
             return &(**connector).public;
+        }
+    }
     /// Retrieves a particular connector. Only the thread that pulled the
     /// associated key out of the execution queue should (be able to) call this.
     pub(crate) fn get_mut(&self, key: &ConnectorKey) -> &'static mut ScheduledConnector {
         let lock = self.inner.read().unwrap();
         unsafe {
             let connector = lock.connectors.get_mut(key.index as usize);
             debug_assert!(!connector.is_null());
             return &mut (**connector);
+        }
+    }
     pub(crate) fn create_interface(&self, connector: ConnectorApplication) -> ConnectorKey {
         // Connector interface does not own any initial ports, and cannot be
         // created by another connector
         let key = self.create_connector_raw(ConnectorVariant::Native(Box::new(connector)));
+        let key = self.create_connector_raw(ConnectorVariant::Native(Box::new(connector)), true);
         return key;
+    }
     /// Create a new connector, returning the key that can be used to retrieve
     /// and/or queue it. The caller must make sure that the constructed
     /// connector's code is initialized with the same ports as the ports in the
     /// `initial_ports` array.
     /// `initial_ports` array. Furthermore the connector is initialized as not
     /// sleeping, so MUST be put on the connector queue by the caller.
     pub(crate) fn create_pdl(&self, created_by: &mut ScheduledConnector, connector: ConnectorPDL) -> ConnectorKey {
         let key = self.create_connector_raw(ConnectorVariant::UserDefined(connector));
+        let key = self.create_connector_raw(ConnectorVariant::UserDefined(connector), false);
         let new_connector = self.get_mut(&key);
         // Transferring ownership of ports (and crashing if there is a
         // programmer's mistake in port management)
         match &new_connector.connector {
             ConnectorVariant::UserDefined(connector) => {
                 for port_id in &connector.ports.owned_ports {
                     let mut port = created_by.context.remove_port(*port_id);
                     new_connector.context.add_port(port);
+                }
             },
             ConnectorVariant::Native(_) => unreachable!(),
+        }
         return key;
+    }
     pub(crate) fn destroy(&self, key: ConnectorKey) {
         let lock = self.inner.write().unwrap();
+        let mut lock = self.inner.write().unwrap();
         unsafe {
             let connector = lock.connectors.get_mut(key.index as usize);
             ptr::drop_in_place(*connector);
             // Note: but not deallocating!
+        }
         lock.free.push(key.index as usize);
+    }
     /// Creates a connector but does not set its initial ports
     fn create_connector_raw(&self, connector: ConnectorVariant) -> ConnectorKey {
+    fn create_connector_raw(&self, connector: ConnectorVariant, initialize_as_sleeping: bool) -> ConnectorKey {
         // Creation of the connector in the global store, requires a lock
         let index;
+        {
             let lock = self.inner.write().unwrap();
+            let mut lock = self.inner.write().unwrap();
             let connector = ScheduledConnector {
                 connector,
                 context: ConnectorCtx::new(self.port_counter.clone()),
                 public: ConnectorPublic::new(),
+                public: ConnectorPublic::new(initialize_as_sleeping),
                 router: Router::new(),
             };
             if lock.free.is_empty() {
                 let connector = Box::into_raw(Box::new(connector));
                 unsafe {
                     // Cheating a bit here. Anyway, move to heap, store in list
                     index = lock.connectors.len();
                     lock.connectors.push(connector);
+                }
                 index = lock.connectors.len();
                 lock.connectors.push(connector);
             } else {
                 index = lock.free.pop().unwrap();
                 unsafe {
                     let target = lock.connectors.get_mut(index);
                     debug_assert!(!target.is_null());
                     ptr::write(*target, connector);
+                }
+            }
+        }
         // Generate key and retrieve the connector to set its ID
         let key = ConnectorKey{ index: index as u32 };
         let new_connector = self.get_mut(&key);
         new_connector.context.id = key.downcast();
         // Return the connector key
         return key;
+    }
+}
 impl Drop for ConnectorStore {
     fn drop(&mut self) {
         let lock = self.inner.write().unwrap();
         for idx in 0..lock.connectors.len() {
             unsafe {
                 let memory = *lock.connectors.get_mut(idx);
                 let _ = Box::from_raw(memory); // takes care of deallocation
+            }
+        }
+    }
+}
 /// Global store of connectors, ports and queues that are used by the sceduler
 /// threads. The global store has the appearance of a thread-safe datatype, but
 /// one needs to be careful using it.
 ///
 /// TODO: @docs
 /// TODO: @Optimize, very lazy implementation of concurrent datastructures.
 ///     This includes the `should_exit` and `did_exit` pair!
 pub struct GlobalStore {
+pub(crate) struct GlobalStore {
     pub connector_queue: MpmcQueue<ConnectorKey>,
     pub connectors: ConnectorStore,
     pub should_exit: AtomicBool,    // signal threads to exit
+}
 impl GlobalStore {
     pub fn new() -> Self {
+    pub(crate) fn new() -> Self {
         Self{
             connector_queue: MpmcQueue::with_capacity(256),
             connectors: ConnectorStore::with_capacity(256),
             should_exit: AtomicBool::new(false),
+        }
+    }
+}
@@ \ No newline at end of file @@

src/runtime2/inbox.rs

➞

Show inline comments

 /**
 inbox.rs
 Contains various types of inboxes and message types for the connectors. There
 are two kinds of inboxes:
 The `PublicInbox` is a simple message queue. Messages are put in by various
 threads, and they're taken out by a single thread. These messages may contain
 control messages and may be filtered or redirected by the scheduler.
 The `PrivateInbox` is a temporary storage for all messages that are received
 within a certain sync-round.
 **/
 use std::collections::VecDeque;
 use std::sync::Mutex;
 use crate::protocol::eval::ValueGroup;
 use super::connector::BranchId;
 use super::port::PortIdLocal;
 use super::global_store::ConnectorId;
 /// A message that has been delivered (after being imbued with the receiving
 /// port by the scheduler) to a connector.
 #[derive(Clone)]
 pub struct DataMessage {
     pub sending_port: PortIdLocal,
     pub sender_prev_branch_id: BranchId,
     pub sender_cur_branch_id: BranchId,
     pub message: ValueGroup,
+}
 #[derive(Clone)]
 pub enum SyncBranchConstraint {
     SilentPort(PortIdLocal),
     BranchNumber(BranchId),
     PortMapping(PortIdLocal, BranchId),
+}
 #[derive(Clone)]
 pub struct SyncConnectorSolution {
     pub connector_id: ConnectorId,
     pub terminating_branch_id: BranchId,
     pub execution_branch_ids: Vec<BranchId>, // no particular ordering of IDs enforced
     pub final_port_mapping: Vec<(PortIdLocal, BranchId)>
+}
 #[derive(Clone)]
 pub struct SyncConnectorConstraints {
     pub connector_id: ConnectorId,
     pub constraints: Vec<SyncBranchConstraint>,
+}
 #[derive(Clone)]
 pub struct SyncMessage {
     pub local_solutions: Vec<SyncConnectorSolution>,
     pub constraints: Vec<SyncConnectorConstraints>,
     pub to_visit: Vec<ConnectorId>,
+}
 // TODO: Shouldn't really be here, right?
 impl SyncMessage {
     /// Creates a new sync message. Assumes that it is created by a connector
     /// that has just encountered a new local solution.
     pub(crate) fn new(initial_solution: SyncConnectorSolution, approximate_peers: usize) -> Self {
         let mut local_solutions = Vec::with_capacity(approximate_peers);
         local_solutions.push(initial_solution);
         return Self{
             local_solutions,
             constraints: Vec::with_capacity(approximate_peers),
             to_visit: Vec::with_capacity(approximate_peers),
         };
+    }
     /// Checks if a connector has already provided a local solution
     pub(crate) fn has_local_solution_for(&self, connector_id: ConnectorId) -> bool {
         return self.local_solutions
             .iter()
             .any(|v| v.connector_id == connector_id);
+    }
     /// Adds a new constraint. If the connector has already provided a local
     /// solution then the constraint will be checked. Otherwise the constraint
     /// will be added to the solution. If this is the first constraint for a
     /// connector then it will be added to the connectors that still have to be
     /// visited.
     ///
     /// If this returns true then the constraint was added, or the local
     /// solution for the specified connector satisfies the constraint. If this
     /// function returns an error then we're dealing with a nefarious peer.
     pub(crate) fn add_or_check_constraint(
         &mut self, connector_id: ConnectorId, constraint: SyncBranchConstraint
     ) -> Result<bool, ()> {
         if self.has_local_solution_for(connector_id) {
             return self.check_constraint(connector_id, constraint);
         } else {
             self.add_constraint(connector_id, constraint);
             return Ok(true);
+        }
+    }
     /// Pushes a new connector constraint. Caller must ensure that the solution
     /// has not yet arrived at the specified connector (because then it would no
     /// longer have constraints, but a proposed solution instead).
     pub(crate) fn add_constraint(&mut self, connector_id: ConnectorId, constraint: SyncBranchConstraint) {
         debug_assert!(!self.has_local_solution_for(connector_id));
         let position = self.constraints
             .iter()
             .position(|v| v.connector_id == connector_id);
         match position {
             Some(index) => {
                 // Has pre-existing constraints
                 debug_assert!(self.to_visit.contains(&connector_id));
                 let entry = &mut self.constraints[index];
                 entry.constraints.push(constraint);
             },
             None => {
                 debug_assert!(!self.to_visit.contains(&connector_id));
                 self.constraints.push(SyncConnectorConstraints{
                     connector_id,
                     constraints: vec![constraint],
                 });
                 self.to_visit.push(connector_id);
+            }
+        }
+    }
     /// Checks if a constraint is satisfied by a solution. Caller must make sure
     /// that a local solution has already been provided. Will return an error
     /// value only if the provided constraint does not make sense (i.e. a
     /// nefarious peer has supplied a constraint with a port we do not own).
     pub(crate) fn check_constraint(&self, connector_id: ConnectorId, constraint: SyncBranchConstraint) -> Result<bool, ()>  {
         debug_assert!(self.has_local_solution_for(connector_id));
         let entry = self.local_solutions
             .iter()
             .find(|v| v.connector_id == connector_id)
             .unwrap();
         match constraint {
             SyncBranchConstraint::SilentPort(silent_port_id) => {
                 for (port_id, mapped_id) in &entry.final_port_mapping {
                     if *port_id == silent_port_id {
                         // If silent, then mapped ID is invalid
                         return Ok(!mapped_id.is_valid())
+                    }
+                }
                 return Err(());
             },
             SyncBranchConstraint::BranchNumber(expected_branch_id) => {
                 return Ok(entry.execution_branch_ids.contains(&expected_branch_id));
             },
             SyncBranchConstraint::PortMapping(port_id, expected_branch_id) => {
                 for (port_id, mapped_id) in &entry.final_port_mapping {
                     if port_id == port_id {
                         return Ok(*mapped_id == expected_branch_id);
+                    }
+                }
                 return Err(());
             },
+        }
+    }
+}
 #[derive(Clone)]
 pub struct SolutionMessage {
     pub comparison_number: u64,
     pub connector_origin: ConnectorId,
     pub local_solutions: Vec<(ConnectorId, BranchId)>,
     pub to_visit: Vec<ConnectorId>,
+}
 /// A control message. These might be sent by the scheduler to notify eachother
 /// of asynchronous state changes.
 #[derive(Clone)]
 pub struct ControlMessage {
     pub id: u32, // generic identifier, used to match request to response
     pub content: ControlMessageVariant,
+}
 #[derive(Clone)]
 pub enum ControlMessageVariant {
     ChangePortPeer(PortIdLocal, ConnectorId), // specified port has a new peer, sent to owner of said port
     Ack, // acknowledgement of previous control message, matching occurs through control message ID.
+}
 /// Generic message contents.
 #[derive(Clone)]
 pub enum MessageContents {
     Data(DataMessage),              // data message, handled by connector
     Sync(SyncMessage),              // sync message, handled by both connector/scheduler
     RequestCommit(SolutionMessage), // solution message, requesting participants to commit
     ConfirmCommit(SolutionMessage), // solution message, confirming a solution everyone committed to
     Control(ControlMessage),        // control message, handled by scheduler
     Ping,                           // ping message, intentionally waking up a connector (used for native connectors)
+}
 pub struct Message {
     pub sending_connector: ConnectorId,
     pub receiving_port: PortIdLocal, // may be invalid (in case of messages targeted at the connector)
     pub contents: MessageContents,
+}
 /// The public inbox of a connector. The thread running the connector that owns
 /// this inbox may retrieved from it. Non-owning threads may only put new
 /// messages inside of it.
 // TODO: @Optimize, lazy concurrency. Probably ringbuffer with read/write heads.
 //  Should behave as a MPSC queue.
 pub struct PublicInbox {
     messages: Mutex<VecDeque<Message>>,
+}
 impl PublicInbox {
     pub fn new() -> Self {
         Self{
             messages: Mutex::new(VecDeque::new()),
+        }
+    }
     pub fn insert_message(&self, message: Message) {
         let mut lock = self.messages.lock().unwrap();
         lock.push_back(message);
+    }
     pub fn take_message(&self) -> Option<Message> {
         let mut lock = self.messages.lock().unwrap();
         return lock.pop_front();
+    }
     pub fn is_empty(&self) -> bool {
         let lock = self.messages.lock().unwrap();
         return lock.is_empty();
+    }
+}
 pub struct PrivateInbox {
+pub(crate) struct PrivateInbox {
     // "Normal" messages, intended for a PDL protocol. These need to stick
     // around during an entire sync-block (to handle `put`s for which the
     // corresponding `get`s have not yet been reached).
     messages: Vec<DataMessage>,
+    messages: Vec<(PortIdLocal, DataMessage)>,
     len_read: usize,
+}
 impl PrivateInbox {
     pub fn new() -> Self {
         Self{
             messages: Vec::new(),
             len_read: 0,
+        }
+    }
     /// Will insert the message into the inbox. Only exception is when the tuple
     /// (prev_branch_id, cur_branch_id, receiving_port_id) already exists, then
     /// nothing is inserted..
     pub fn insert_message(&mut self, message: DataMessage) {
         for existing in self.messages.iter() {
     pub(crate) fn insert_message(&mut self, target_port: PortIdLocal, message: DataMessage) {
         for (existing_target_port, existing) in self.messages.iter() {
             if existing.sender_prev_branch_id == message.sender_prev_branch_id &&
                     existing.sender_cur_branch_id == message.sender_cur_branch_id &&
-                    existing.sending_port == message.sending_port {
+                    *existing_target_port == target_port {
                 // Message was already received
                 return;
+            }
+        }
         self.messages.push(message);
+        self.messages.push((target_port, message));
+    }
     /// Retrieves all previously read messages that satisfy the provided
     /// speculative conditions. Note that the inbox remains read-locked until
     /// the returned iterator is dropped. Should only be called by the
     /// inbox-reader (i.e. the thread executing a connector's PDL code).
     ///
     /// This function should only be used to check if already-received messages
     /// could be received by a newly encountered `get` call in a connector's
     /// PDL code.
     pub fn get_messages(&self, port_id: PortIdLocal, prev_branch_id: BranchId) -> InboxMessageIter {
+    pub(crate) fn get_messages(&self, port_id: PortIdLocal, prev_branch_id: BranchId) -> InboxMessageIter {
         return InboxMessageIter{
             messages: &self.messages,
             next_index: 0,
             max_index: self.len_read,
             match_port_id: port_id,
             match_prev_branch_id: prev_branch_id,
         };
+    }
     /// Retrieves the next unread message. Should only be called by the
     /// inbox-reader.
     pub fn next_message(&mut self) -> Option<&DataMessage> {
+    pub(crate) fn next_message(&mut self) -> Option<&DataMessage> {
         if self.len_read == self.messages.len() {
             return None;
+        }
         let to_return = &self.messages[self.len_read];
+        let (_, to_return) = &self.messages[self.len_read];
         self.len_read += 1;
         return Some(to_return);
+    }
     /// Simply empties the inbox
     pub fn clear(&mut self) {
+    pub(crate) fn clear(&mut self) {
         self.messages.clear();
         self.len_read = 0;
+    }
+}
 /// Iterator over previously received messages in the inbox.
 pub struct InboxMessageIter<'i> {
     messages: &'i Vec<DataMessage>,
 pub(crate) struct InboxMessageIter<'i> {
     messages: &'i Vec<(PortIdLocal, DataMessage)>,
     next_index: usize,
     max_index: usize,
     match_port_id: PortIdLocal,
     match_prev_branch_id: BranchId,
+}
 impl<'i> Iterator for InboxMessageIter<'i> {
     type Item = &'i DataMessage;
     fn next(&mut self) -> Option<Self::Item> {
         // Loop until match is found or at end of messages
         while self.next_index < self.max_index {
             let cur_message = &self.messages[self.next_index];
             if cur_message.receiving_port == self.match_port_id && cur_message.sender_prev_branch_id == self.match_prev_branch_id {
             let (target_port, cur_message) = &self.messages[self.next_index];
             if *target_port == self.match_port_id && cur_message.sender_prev_branch_id == self.match_prev_branch_id {
                 // Found a match
                 break;
+            }
             self.next_index += 1;
+        }
         if self.next_index == self.max_index {
             return None;
+        }
         let message = &self.messages[self.next_index];
+        let (_, message) = &self.messages[self.next_index];
         self.next_index += 1;
         return Some(message);
+    }
+}
@@ \ No newline at end of file @@

src/runtime2/messages.rs

➞

Show inline comments

 use std::collections::hash_map::Entry;
 use std::collections::HashMap;
 use crate::PortId;
 use crate::protocol::*;
 use crate::protocol::eval::*;
 /// A message residing in a connector's inbox (waiting to be put into some kind
 /// of speculative branch), or a message waiting to be sent.
 #[derive(Clone)]
 pub struct BufferedMessage {
     pub(crate) sending_port: PortId,
     pub(crate) receiving_port: PortId,
     pub(crate) peer_prev_branch_id: Option<u32>,
     pub(crate) peer_cur_branch_id: u32,
     pub(crate) message: ValueGroup,
+}
 /// A connector's global inbox. Any received message ends up here. This is
 /// because a message might be received before a branch arrives at the
 /// corresponding `get()` that is supposed to receive that message. Hence we
 /// need to store it for all future branches that might be able to receive it.
 pub struct ConnectorInbox {
     // TODO: @optimize, HashMap + Vec is a bit stupid.
     messages: HashMap<PortAction, Vec<BufferedMessage>>
+}
 /// An action performed on a port. Unsure about this
 #[derive(PartialEq, Eq, Hash)]
 struct PortAction {
     port_id: u32,
     prev_branch_id: Option<u32>,
+}
 // TODO: @remove
 impl ConnectorInbox {
     pub fn new() -> Self {
         Self {
             messages: HashMap::new(),
+        }
+    }
     /// Inserts a new message into the inbox.
     pub fn insert_message(&mut self, message: BufferedMessage) {
         // TODO: @error - Messages are received from actors we generally cannot
         //  trust, and may be unreliable, so messages may be received multiple
         //  times or have spoofed branch IDs. Debug asserts are present for the
         //  initial implementation.
         // If it is the first message on the port, then we cannot possible have
         // a previous port mapping on that port.
         let port_action = PortAction{
             port_id: message.receiving_port.0.u32_suffix,
             prev_branch_id: message.peer_prev_branch_id,
         };
         match self.messages.entry(port_action) {
             Entry::Occupied(mut entry) => {
                 let entry = entry.get_mut();
                 debug_assert!(
                     entry.iter()
                         .find(|v| v.peer_cur_branch_id == message.peer_cur_branch_id)
                         .is_none(),
                     "inbox already contains sent message (same new branch ID)"
                 );
                 entry.push(message);
             },
             Entry::Vacant(entry) => {
                 entry.insert(vec![message]);
+            }
+        }
+    }
     /// Checks if the provided port (and the branch id mapped to that port)
     /// correspond to any messages in the inbox.
     pub fn find_matching_message(&self, port_id: u32, prev_branch_id_at_port: Option<u32>) -> Option<&[BufferedMessage]> {
         let port_action = PortAction{
             port_id,
             prev_branch_id: prev_branch_id_at_port,
         };
         match self.messages.get(&port_action) {
             Some(messages) => return Some(messages.as_slice()),
             None => return None,
+        }
+    }
     pub fn clear(&mut self) {
         self.messages.clear();
+    }
+}
 /// A connector's outbox. A temporary storage for messages that are sent by
 /// branches performing `put`s until we're done running all branches and can
 /// actually transmit the messages.
 pub struct ConnectorOutbox {
     messages: Vec<BufferedMessage>,
+}
 impl ConnectorOutbox {
     pub fn new() -> Self {
         Self{
             messages: Vec::new(),
+        }
+    }
     pub fn insert_message(&mut self, message: BufferedMessage) {
         // TODO: @error - Depending on the way we implement the runtime in the
         //  future we might end up not trusting "our own code" (i.e. in case
         //  the connectors we are running are described by foreign code)
         debug_assert!(
             self.messages.iter()
                 .find(|v|
                     v.sending_port == message.sending_port &&
                     v.peer_prev_branch_id == message.peer_prev_branch_id
+                )
                 .is_none(),
             "messages was already registered for sending"
         );
         self.messages.push(message);
+    }
     pub fn take_next_message_to_send(&mut self) -> Option<BufferedMessage> {
         self.messages.pop()
+    }
     pub fn clear(&mut self) {
         self.messages.clear();
+    }
+}
@@ \ No newline at end of file @@

src/runtime2/mod.rs

➞

Show inline comments

 // Structure of module
 mod runtime;
 mod messages;
 mod connector;
 mod native;
 mod port;
 mod global_store;
 mod scheduler;
 mod inbox;
 #[cfg(test)] mod tests;
 // Imports
 use std::sync::{Arc, Mutex};
 use std::sync::atomic::Ordering;
+use std::sync::atomic::{AtomicU32, Ordering};
 use std::thread::{self, JoinHandle};
 use crate::ProtocolDescription;
 use global_store::{ConnectorVariant, GlobalStore};
 use scheduler::Scheduler;
 use native::{ConnectorApplication, ApplicationInterface};
 // Runtime API
 // TODO: Exit condition is very dirty. Take into account:
 //  - Connector hack with &'static references. May only destroy (unforced) if all connectors are done working
 //  - Running schedulers: schedulers need to be signaled that they should exit, then wait until all are done
 //  - User-owned interfaces: As long as these are owned user may still decide to create new connectors.
 /// A kind of token that, once obtained, allows mutable access to a connector.
 /// We're trying to use move semantics as much as possible: the owner of this
 /// key is the only one that may execute the connector's code.
 pub(crate) struct ConnectorKey {
     pub index: u32, // of connector
+}
 impl ConnectorKey {
     /// Downcasts the `ConnectorKey` type, which can be used to obtain mutable
     /// access, to a "regular ID" which can be used to obtain immutable access.
     #[inline]
     pub fn downcast(&self) -> ConnectorId {
         return ConnectorId(self.index);
+    }
     /// Turns the `ConnectorId` into a `ConnectorKey`, marked as unsafe as it
     /// bypasses the type-enforced `ConnectorKey`/`ConnectorId` system
     #[inline]
     pub unsafe fn from_id(id: ConnectorId) -> ConnectorKey {
         return ConnectorKey{ index: id.0 };
+    }
+}
 /// A kind of token that allows shared access to a connector. Multiple threads
 /// may hold this
 #[derive(Debug, Copy, Clone, PartialEq, Eq)]
 pub struct ConnectorId(pub u32);
 impl ConnectorId {
     // TODO: Like the other `new_invalid`, maybe remove
     #[inline]
     pub fn new_invalid() -> ConnectorId {
         return ConnectorId(u32::MAX);
+    }
     #[inline]
     pub(crate) fn is_valid(&self) -> bool {
         return self.0 != u32::MAX;
+    }
+}
 // TODO: Change this, I hate this. But I also don't want to put `public` and
 //  `router` of `ScheduledConnector` back into `Connector`. The reason I don't
 //  want `Box<dyn Connector>` everywhere is because of the v-table overhead. But
 //  to truly design this properly I need some benchmarks.
 pub(crate) enum ConnectorVariant {
     UserDefined(ConnectorPDL),
     Native(Box<dyn Connector>),
+}
 impl Connector for ConnectorVariant {
     fn handle_message(&mut self, message: Message, ctx: &ConnectorCtx, delta_state: &mut RunDeltaState) {
         match self {
             ConnectorVariant::UserDefined(c) => c.handle_message(message, ctx, delta_state),
             ConnectorVariant::Native(c) => c.handle_message(message, ctx, delta_state),
+        }
+    }
     fn run(&mut self, protocol_description: &ProtocolDescription, ctx: &ConnectorCtx, delta_state: &mut RunDeltaState) -> ConnectorScheduling {
         match self {
             ConnectorVariant::UserDefined(c) => c.run(protocol_description, ctx, delta_state),
             ConnectorVariant::Native(c) => c.run(protocol_description, ctx, delta_state),
+        }
+    }
+}
 pub(crate) struct ScheduledConnector {
     pub connector: ConnectorVariant, // access by connector
     pub context: ConnectorCtx, // mutable access by scheduler, immutable by connector
     pub public: ConnectorPublic, // accessible by all schedulers and connectors
     pub router: Router,
+}
 /// Externally facing runtime.
 pub struct Runtime {
     inner: Arc<RuntimeInner>,
+}
 pub(crate) struct RuntimeInner {
-    pub(crate) global_store: GlobalStore,
+    // Protocol
     pub(crate) protocol_description: ProtocolDescription,
     schedulers: Mutex<Vec<JoinHandle<()>>>, // TODO: Revise, make exit condition something like: all interfaces dropped
     // Storage of connectors in a kind of freelist. Note the vector of points to
     // ensure pointer stability: the vector can be changed but the entries
     // themselves remain valid.
     pub connectors_list: RawVec<*mut ScheduledConnector>,
     pub connectors_free: Vec<usize>,
     pub(crate) global_store: GlobalStore,
     schedulers: Mutex<Vec<JoinHandle<()>>>,
     active_interfaces: AtomicU32, // active API interfaces that can add connectors/channels
+}
 impl RuntimeInner {
     #[inline]
     pub(crate) fn increment_active_interfaces(&self) {
         let _old_num = self.active_interfaces.fetch_add(1, Ordering::SeqCst);
         debug_assert_ne!(_old_num, 1); // once it hits 0, it stays zero
+    }
     pub(crate) fn decrement_active_interfaces(&self) {
         let old_num = self.active_interfaces.fetch_sub(1, Ordering::SeqCst);
         debug_assert!(old_num > 0);
         if old_num == 1 {
             // Became 0
             // TODO: Check num connectors, if 0, then set exit flag
+        }
+    }
+}
 // TODO: Come back to this at some point
 unsafe impl Send for RuntimeInner {}
 unsafe impl Sync for RuntimeInner {}
 impl Runtime {
-    pub fn new(num_threads: usize, protocol_description: ProtocolDescription) -> Runtime {
+    pub fn new(num_threads: u32, protocol_description: ProtocolDescription) -> Runtime {
         // Setup global state
         assert!(num_threads > 0, "need a thread to run connectors");
         let runtime_inner = Arc::new(RuntimeInner{
             global_store: GlobalStore::new(),
             protocol_description,
             schedulers: Mutex::new(Vec::new()),
             active_interfaces: AtomicU32::new(1), // we are the active interface
         });
         // Launch threads
+        {
             let mut schedulers = Vec::with_capacity(num_threads);
             for _ in 0..num_threads {
             let mut schedulers = Vec::with_capacity(num_threads as usize);
             for thread_index in 0..num_threads {
                 let cloned_runtime_inner = runtime_inner.clone();
                 let thread = thread::spawn(move || {
                     let mut scheduler = Scheduler::new(cloned_runtime_inner);
                     scheduler.run();
                 });
                 let thread = thread::Builder::new()
                     .name(format!("thread-{}", thread_index))
                     .spawn(move || {
                         let mut scheduler = Scheduler::new(cloned_runtime_inner, thread_index);
                         scheduler.run();
                     })
                     .unwrap();
                 schedulers.push(thread);
+            }
             let mut lock = runtime_inner.schedulers.lock().unwrap();
             *lock = schedulers;
+        }
         // Return runtime
         return Runtime{ inner: runtime_inner };
+    }
     /// Returns a new interface through which channels and connectors can be
     /// created.
     pub fn create_interface(&self) -> ApplicationInterface {
         let (connector, mut interface) = ConnectorApplication::new(self.inner.clone());
         let connector_key = self.inner.global_store.connectors.create_interface(connector);
         interface.set_connector_id(connector_key.downcast());
         // Note that we're not scheduling. That is done by the interface in case
         // it is actually needed.
         return interface;
+    }
+}
 impl Drop for Runtime {
     fn drop(&mut self) {
         self.inner.global_store.should_exit.store(true, Ordering::Release);
         let mut schedulers = self.inner.schedulers.lock().unwrap();
         for scheduler in schedulers.drain(..) {
             scheduler.join();
+            scheduler.join().unwrap();
+        }
+    }
+}
@@ \ No newline at end of file @@

src/runtime2/native.rs

➞

Show inline comments

 use std::collections::VecDeque;
 use std::sync::{Arc, Mutex, Condvar};
 use std::sync::atomic::Ordering;
 use crate::protocol::ComponentCreationError;
 use crate::protocol::ComponentCreationError;
 use crate::protocol::eval::ValueGroup;
 use crate::ProtocolDescription;
 use crate::runtime2::connector::{Branch, find_ports_in_value_group};
 use crate::runtime2::global_store::ConnectorKey;
 use crate::runtime2::inbox::MessageContents;
 use crate::runtime2::port::{Port, PortKind};
 use crate::runtime2::scheduler::ConnectorCtx;
 use super::RuntimeInner;
 use super::global_store::ConnectorId;
 use super::port::{Channel, PortIdLocal};
 use super::connector::{ConnectorPDL, ConnectorScheduling, RunDeltaState};
 use super::inbox::Message;
 /// Generic connector interface from the scheduler's point of view.
 pub trait Connector {
+pub(crate) trait Connector {
     /// Handle a new message (preprocessed by the scheduler). You probably only
     /// want to handle `Data`, `Sync`, and `Solution` messages. The others are
     /// intended for the scheduler itself.
-    fn handle_message(&mut self, message: MessageContents, ctx: &ConnectorCtx, delta_state: &mut RunDeltaState);
+    fn handle_message(&mut self, message: Message, ctx: &ConnectorCtx, delta_state: &mut RunDeltaState);
     /// Should run the connector's behaviour up until the next blocking point.
     fn run(&mut self, protocol_description: &ProtocolDescription, ctx: &ConnectorCtx, delta_state: &mut RunDeltaState) -> ConnectorScheduling;
+}
 type SyncDone = Arc<(Mutex<bool>, Condvar)>;
-type JobQueue = Arc<Mutex<Vec<ApplicationJob>>>;
+type JobQueue = Arc<Mutex<VecDeque<ApplicationJob>>>;
 enum ApplicationJob {
     NewChannel((Port, Port)),
     NewConnector(ConnectorPDL),
+}
 /// The connector which an application can directly interface with. Once may set
 /// up the next synchronous round, and retrieve the data afterwards.
 pub struct ConnectorApplication {
     sync_done: SyncDone,
     job_queue: JobQueue,
+}
 impl ConnectorApplication {
     pub(crate) fn new(runtime: Arc<RuntimeInner>) -> (Self, ApplicationInterface) {
         let sync_done = Arc::new(( Mutex::new(false), Condvar::new() ));
-        let job_queue = Arc::new(Mutex::new(Vec::with_capacity(32)));
+        let job_queue = Arc::new(Mutex::new(VecDeque::with_capacity(32)));
         let connector = ConnectorApplication { sync_done: sync_done.clone(), job_queue: job_queue.clone() };
         let interface = ApplicationInterface::new(sync_done, job_queue, runtime);
         return (connector, interface);
+    }
+}
 impl Connector for ConnectorApplication {
     fn handle_message(&mut self, message: MessageContents, ctx: &ConnectorCtx, delta_state: &mut RunDeltaState) {
         todo!("handling messages in ConnectorApplication (API for runtime)")
     fn handle_message(&mut self, message: Message, _ctx: &ConnectorCtx, _delta_state: &mut RunDeltaState) {
         use MessageContents as MC;
         match message.contents {
             MC::Data(_) => unreachable!("data message in API connector"),
             MC::Sync(_) | MC::RequestCommit(_) | MC::ConfirmCommit(_) => {
                 // Handling sync in API
             },
             MC::Control(_) => {},
             MC::Ping => {},
+        }
+    }
     fn run(&mut self, protocol_description: &ProtocolDescription, ctx: &ConnectorCtx, delta_state: &mut RunDeltaState) -> ConnectorScheduling {
         let mut queue = self.job_queue.lock().unwrap();
-        while let Some(job) = queue.pop() {
+        while let Some(job) = queue.pop_front() {
             match job {
                 ApplicationJob::NewChannel((endpoint_a, endpoint_b)) => {
                     println!("DEBUG: API adopting ports");
                     delta_state.new_ports.reserve(2);
                     delta_state.new_ports.push(endpoint_a);
                     delta_state.new_ports.push(endpoint_b);
+                }
                 ApplicationJob::NewConnector(connector) => {
                     println!("DEBUG: API creating connector");
                     delta_state.new_connectors.push(connector);
+                }
+            }
+        }
         return ConnectorScheduling::NotNow;
+    }
+}
 /// The interface to a `ApplicationConnector`. This allows setting up the
 /// interactions the `ApplicationConnector` performs within a synchronous round.
 pub struct ApplicationInterface {
     sync_done: SyncDone,
     job_queue: JobQueue,
     runtime: Arc<RuntimeInner>,
     connector_id: ConnectorId,
     owned_ports: Vec<PortIdLocal>,
+}
 impl ApplicationInterface {
     pub(crate) fn new(sync_done: SyncDone, job_queue: JobQueue, runtime: Arc<RuntimeInner>) -> Self {
     fn new(sync_done: SyncDone, job_queue: JobQueue, runtime: Arc<RuntimeInner>) -> Self {
         runtime.active_interfaces += 1;
         return Self{
             sync_done, job_queue, runtime,
             connector_id: ConnectorId::new_invalid(),
             owned_ports: Vec::new(),
+        }
+    }
     /// Creates a new channel.
     pub fn create_channel(&mut self) -> Channel {
         // TODO: Duplicated logic in scheduler
         let getter_id = self.runtime.global_store.connectors.port_counter.fetch_add(2, Ordering::SeqCst);
         let putter_id = PortIdLocal::new(getter_id + 1);
         let getter_id = PortIdLocal::new(getter_id);
         // Create ports and add a job such that they are transferred to the
         // API component. (note that we do not send a ping, this is only
         // necessary once we create a connector)
         let getter_port = Port{
             self_id: getter_id,
             peer_id: putter_id,
             kind: PortKind::Getter,
             peer_connector: self.connector_id,
         };
         let putter_port = Port{
             self_id: putter_id,
             peer_id: getter_id,
             kind: PortKind::Putter,
             peer_connector: self.connector_id,
         };
+        {
             let mut lock = self.job_queue.lock().unwrap();
-            lock.push(ApplicationJob::NewChannel((getter_port, putter_port)));
+            lock.push_back(ApplicationJob::NewChannel((getter_port, putter_port)));
+        }
         // Add to owned ports for error checking while creating a connector
         self.owned_ports.reserve(2);
         self.owned_ports.push(putter_id);
         self.owned_ports.push(getter_id);
         return Channel{ putter_id, getter_id };
+    }
     /// Creates a new connector. Note that it is not scheduled immediately, but
     /// depends on the `ApplicationConnector` to run, followed by the created
     /// connector being scheduled.
     // TODO: Optimize by yanking out scheduler logic for common use.
     pub fn create_connector(&mut self, module: &str, routine: &str, arguments: ValueGroup) -> Result<(), ComponentCreationError> {
         // Retrieve ports and make sure that we own the ones that are currently
         // specified. This is also checked by the scheduler, but that is done
         // asynchronously.
         let mut initial_ports = Vec::new();
         find_ports_in_value_group(&arguments, &mut initial_ports);
         for port_to_remove in &initial_ports {
             match self.owned_ports.iter().position(|v| v == port_to_remove) {
                 Some(index_to_remove) => {
                     // We own the port, so continue
                     self.owned_ports.remove(index_to_remove);
                 },
                 None => {
                     // We don't own the port
                     return Err(ComponentCreationError::UnownedPort);
+                }
+            }
+        }
         let state = self.runtime.protocol_description.new_component_v2(module.as_bytes(), routine.as_bytes(), arguments)?;
         let connector = ConnectorPDL::new(Branch::new_initial_branch(state), initial_ports);
         // Put on job queue
+        {
             let mut queue = self.job_queue.lock().unwrap();
-            queue.push(ApplicationJob::NewConnector(connector));
+            queue.push_back(ApplicationJob::NewConnector(connector));
+        }
         // Send ping message to wake up connector
         let connector = self.runtime.global_store.connectors.get_shared(self.connector_id);
         connector.inbox.insert_message(Message{
             sending_connector: ConnectorId::new_invalid(),
             receiving_port: PortIdLocal::new_invalid(),
             contents: MessageContents::Ping,
         });
         let should_wake_up = connector.sleeping
             .compare_exchange(true, false, Ordering::SeqCst, Ordering::Acquire)
             .is_ok();
         if should_wake_up {
             println!("DEBUG: Waking up connector");
             let key = unsafe{ ConnectorKey::from_id(self.connector_id) };
             self.runtime.global_store.connector_queue.push_back(key);
         } else {
             println!("DEBUG: NOT waking up connector");
+        }
         return Ok(());
+    }
     /// Check if the next sync-round is finished.
     pub fn try_wait(&self) -> bool {
         let (is_done, _) = &*self.sync_done;
         let lock = is_done.lock().unwrap();
         return *lock;
+    }
     /// Wait until the next sync-round is finished
     pub fn wait(&self) {
         let (is_done, condition) = &*self.sync_done;
         let lock = is_done.lock().unwrap();
         condition.wait_while(lock, |v| !*v); // wait while not done
+        condition.wait_while(lock, |v| !*v).unwrap(); // wait while not done
+    }
     /// Called by runtime to set associated connector's ID.
     pub(crate) fn set_connector_id(&mut self, id: ConnectorId) {
         self.connector_id = id;
+    }
+}
 impl Drop for ApplicationInterface {
     fn drop(&mut self) {
+    }
+}
@@ \ No newline at end of file @@

src/runtime2/port.rs

➞

Show inline comments

 use super::global_store::ConnectorId;
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
-pub(crate) struct PortIdLocal {
 pub struct PortIdLocal {
     pub index: u32,
+}
 impl PortIdLocal {
     pub fn new(id: u32) -> Self {
         Self{ index: id }
+    }
     // TODO: Unsure about this, maybe remove, then also remove all struct
     //  instances where I call this
     pub fn new_invalid() -> Self {
         Self{ index: u32::MAX }
+    }
     pub fn is_valid(&self) -> bool {
         return self.index != u32::MAX;
+    }
+}
 #[derive(Eq, PartialEq)]
 pub enum PortKind {
     Putter,
     Getter,
+}
 /// Represents a port inside of the runtime. May be without owner if it is
 /// created by the application interfacing with the runtime, instead of being
 /// created by a connector.
 pub struct Port {
     pub self_id: PortIdLocal,
     pub peer_id: PortIdLocal,
     pub kind: PortKind,
     pub peer_connector: ConnectorId, // might be temporarily inconsistent while peer port is sent around in non-sync phase.
+}
 // TODO: Turn port ID into its own type
 pub struct Channel {
     pub putter_id: PortIdLocal, // can put on it, so from the connector's point of view, this is an output
     pub getter_id: PortIdLocal, // vice versa: can get on it, so an input for the connector
+}
@@ \ No newline at end of file @@

src/runtime2/runtime.rs

➞

Show inline comments

 use std::sync::Arc;
 use std::collections::{HashMap, VecDeque};
 use std::collections::hash_map::{Entry};
 use crate::{Polarity, PortId};
 use crate::common::Id;
 use crate::protocol::*;
 use crate::protocol::eval::*;
 use super::messages::*;
 #[derive(Debug)]
 pub enum AddComponentError {
     ModuleDoesNotExist,
     ConnectorDoesNotExist,
     InvalidArgumentType(usize), // value is index of (first) invalid argument
+}
 pub(crate) struct PortDesc {
     id: u32,
     peer_id: u32,
     owning_connector_id: Option<u32>,
     is_getter: bool, // otherwise one can only call `put`
+}
 pub(crate) struct ConnectorDesc {
     id: u32,
     in_sync: bool,
     pub(crate) branches: Vec<BranchDesc>, // first one is always non-speculative one
     spec_branches_active: VecDeque<u32>, // branches that can be run immediately
     spec_branches_pending_receive: HashMap<PortId, Vec<u32>>, // from port_id to branch index
     spec_branches_done: Vec<u32>,
     last_checked_done: u32,
     global_inbox: ConnectorInbox,
     global_outbox: ConnectorOutbox,
+}
 impl ConnectorDesc {
     /// Creates a new connector description. Implicit assumption is that there
     /// is one (non-sync) branch that can be immediately executed.
     fn new(id: u32, component_state: ComponentState, owned_ports: Vec<u32>) -> Self {
         Self{
             id,
             in_sync: false,
             branches: vec![BranchDesc::new_non_sync(component_state, owned_ports)],
             spec_branches_active: VecDeque::new(),
             spec_branches_pending_receive: HashMap::new(),
             spec_branches_done: Vec::new(),
             last_checked_done: 0,
             global_inbox: ConnectorInbox::new(),
             global_outbox: ConnectorOutbox::new(),
+        }
+    }
+}
 #[derive(Debug, PartialEq, Eq)]
 pub(crate) enum BranchState {
     RunningNonSync, // regular running non-speculative branch
     RunningSync, // regular running speculative branch
     BranchPoint, // branch which ended up being a branching point
     ReachedEndSync, // branch that successfully reached the end-sync point, is a possible local solution
     Failed, // branch that became inconsistent
     Finished, // branch (necessarily non-sync) that reached end of code
+}
 #[derive(Debug, Clone)]
 struct BranchPortDesc {
     last_registered_index: Option<u32>, // if putter, then last sent branch ID, if getter, then last received branch ID
     num_times_fired: u32, // number of puts/gets on this port
+}
 struct BranchContext {
     just_called_did_put: bool,
     pending_channel: Option<(Value, Value)>,
+}
 pub(crate) struct BranchDesc {
     index: u32,
     parent_index: Option<u32>,
     pub(crate) code_state: ComponentState,
     pub(crate) branch_state: BranchState,
     owned_ports: Vec<u32>,
     message_inbox: HashMap<(PortId, u32), ValueGroup>, // from (port id, 1-based recv index) to received value
     port_mapping: HashMap<PortId, BranchPortDesc>,
     branch_context: BranchContext,
+}
 impl BranchDesc {
     /// Creates the first non-sync branch of a connector
     fn new_non_sync(component_state: ComponentState, owned_ports: Vec<u32>) -> Self {
         Self{
             index: 0,
             parent_index: None,
             code_state: component_state,
             branch_state: BranchState::RunningNonSync,
             owned_ports,
             message_inbox: HashMap::new(),
             port_mapping: HashMap::new(),
             branch_context: BranchContext{
                 just_called_did_put: false,
                 pending_channel: None,
+            }
+        }
+    }
     /// Creates a sync branch based on the supplied branch. This supplied branch
     /// is the branching point for the new one, i.e. the parent in the branching
     /// tree.
     fn new_sync_from(index: u32, branch_state: &BranchDesc) -> Self {
         // We expect that the given branche's context is not halfway handling a
         // `put(...)` or a `channel x -> y` statement.
         debug_assert!(!branch_state.branch_context.just_called_did_put);
         debug_assert!(branch_state.branch_context.pending_channel.is_none());
         Self{
             index,
             parent_index: Some(branch_state.index),
             code_state: branch_state.code_state.clone(),
             branch_state: BranchState::RunningSync,
             owned_ports: branch_state.owned_ports.clone(),
             message_inbox: branch_state.message_inbox.clone(),
             port_mapping: branch_state.port_mapping.clone(),
             branch_context: BranchContext{
                 just_called_did_put: false,
                 pending_channel: None,
+            }
+        }
+    }
+}
 #[derive(PartialEq, Eq)]
 enum Scheduling {
     Immediate,
     Later,
     NotNow,
+}
 #[derive(Clone, Copy, Eq, PartialEq, Debug)]
 enum ProposedBranchConstraint {
     SilentPort(u32), // port id
     BranchNumber(u32), // branch id
     PortMapping(u32, u32), // particular port's mapped branch number
+}
 // Local solution of the connector
 #[derive(Clone)]
 struct ProposedConnectorSolution {
     final_branch_id: u32,
     all_branch_ids: Vec<u32>, // the final branch ID and, recursively, all parents
     port_mapping: HashMap<u32, Option<u32>>, // port IDs of the connector, mapped to their branch IDs (None for silent ports)
+}
 #[derive(Clone)]
 struct ProposedSolution {
     connector_mapping: HashMap<u32, ProposedConnectorSolution>, // from connector ID to branch ID
     connector_constraints: HashMap<u32, Vec<ProposedBranchConstraint>>, // from connector ID to encountered branch numbers
     remaining_connectors: Vec<u32>, // connectors that still need to be visited
+}
 // TODO: @performance, use freelists+ids instead of HashMaps
 pub struct Runtime {
     protocol: Arc<ProtocolDescription>,
     pub(crate) ports: HashMap<u32, PortDesc>,
     port_counter: u32,
     pub(crate) connectors: HashMap<u32, ConnectorDesc>,
     connector_counter: u32,
     connectors_active: VecDeque<u32>,
+}
 impl Runtime {
     pub fn new(pd: Arc<ProtocolDescription>) -> Self {
         Self{
             protocol: pd,
             ports: HashMap::new(),
             port_counter: 0,
             connectors: HashMap::new(),
             connector_counter: 0,
             connectors_active: VecDeque::new(),
+        }
+    }
     /// Creates a new channel that is not owned by any connector and returns its
     /// endpoints. The returned values are of the (putter port, getter port)
     /// respectively.
     pub fn add_channel(&mut self) -> (Value, Value) {
         let (put_id, get_id) = Self::add_owned_channel(&mut self.ports, &mut self.port_counter, None);
         return (
             port_value_from_id(None, put_id, true),
             port_value_from_id(None, get_id, false)
         );
+    }
     pub fn add_component(&mut self, module: &str, procedure: &str, values: ValueGroup) -> Result<(), AddComponentError> {
         use AddComponentError as ACE;
         use crate::runtime::error::AddComponentError as OldACE;
         // TODO: Remove the responsibility of adding a component from the PD
         // Lookup module and the component
         // TODO: Remove this error enum translation. Note that for now this
         //  function forces port-only arguments
         let port_polarities = match self.protocol.component_polarities(module.as_bytes(), procedure.as_bytes()) {
             Ok(polarities) => polarities,
             Err(reason) => match reason {
                 OldACE::NonPortTypeParameters => return Err(ACE::InvalidArgumentType(0)),
                 OldACE::NoSuchModule => return Err(ACE::ModuleDoesNotExist),
                 OldACE::NoSuchComponent => return Err(ACE::ModuleDoesNotExist),
                 _ => unreachable!(),
+            }
         };
         // Make sure supplied values (and types) are correct. At the same time
         // modify the port IDs such that they contain the ID of the connector
         // we're about the create.
         let component_id = self.generate_connector_id();
         let mut ports = Vec::with_capacity(values.values.len());
         for (value_idx, value) in values.values.iter().enumerate() {
             let polarity = &port_polarities[value_idx];
             match value {
                 Value::Input(port_id) => {
                     if *polarity != Polarity::Getter {
                         return Err(ACE::InvalidArgumentType(value_idx))
+                    }
                     ports.push(PortId(Id{
                         connector_id: component_id,
                         u32_suffix: port_id.0.u32_suffix,
                     }));
                 },
                 Value::Output(port_id) => {
                     if *polarity != Polarity::Putter {
                         return Err(ACE::InvalidArgumentType(value_idx))
+                    }
                     ports.push(PortId(Id{
                         connector_id: component_id,
                         u32_suffix: port_id.0.u32_suffix
                     }));
                 },
                 _ => return Err(ACE::InvalidArgumentType(value_idx))
+            }
+        }
         // Instantiate the component, and mark the ports as being owned by the
         // newly instantiated component
         let component_state = self.protocol.new_component(module.as_bytes(), procedure.as_bytes(), &ports);
         let ports = ports.into_iter().map(|v| v.0.u32_suffix).collect();
         for port in &ports {
             let desc = self.ports.get_mut(port).unwrap();
             desc.owning_connector_id = Some(component_id);
+        }
         self.connectors.insert(component_id, ConnectorDesc::new(component_id, component_state, ports));
         self.connectors_active.push_back(component_id);
         Ok(())
+    }
     pub fn run(&mut self) {
         // Go through all active connectors
         while !self.connectors_active.is_empty() {
             // Run a single connector until it indicates we can run another
             // connector
             let next_id = self.connectors_active.pop_front().unwrap();
             let mut scheduling = Scheduling::Immediate;
             while scheduling == Scheduling::Immediate {
                 scheduling = self.run_connector(next_id);
+            }
             match scheduling {
                 Scheduling::Immediate => unreachable!(),
                 Scheduling::Later => self.connectors_active.push_back(next_id),
                 Scheduling::NotNow => {},
+            }
             // Deal with any outgoing messages and potential solutions
             self.empty_connector_outbox(next_id);
             self.check_connector_new_solutions(next_id);
+        }
+    }
     /// Runs a connector for as long as sensible, then returns `true` if the
     /// connector should be run again in the future, and return `false` if the
     /// connector has terminated. Note that a terminated connector still
     /// requires cleanup.
     fn run_connector(&mut self, connector_id: u32) -> Scheduling {
         let desc = self.connectors.get_mut(&connector_id).unwrap();
         if desc.in_sync {
             return self.run_connector_sync_mode(connector_id);
         } else {
             return self.run_connector_regular_mode(connector_id);
+        }
+    }
     #[inline]
     fn run_connector_sync_mode(&mut self, connector_id: u32) -> Scheduling {
         // Retrieve connector and branch that is supposed to be run
         let desc = self.connectors.get_mut(&connector_id).unwrap();
         debug_assert!(desc.in_sync);
         debug_assert!(!desc.spec_branches_active.is_empty());
         let branch_index = desc.spec_branches_active.pop_front().unwrap();
         let branch = &mut desc.branches[branch_index as usize];
         debug_assert_eq!(branch_index, branch.index);
         // Run this particular branch to a next blocking point
         let mut run_context = Context{
             inbox: &branch.message_inbox,
             port_mapping: &branch.port_mapping,
             branch_ctx: &mut branch.branch_context,
         };
         let run_result = branch.code_state.run(&mut run_context, &self.protocol);
         match run_result {
             RunResult::BranchInconsistent => {
                 // Speculative branch became inconsistent. So we don't
                 // run it again
                 branch.branch_state = BranchState::Failed;
             },
             RunResult::BranchMissingPortState(port_id) => {
                 // Branch called `fires()` on a port that did not have a
                 // value assigned yet. So branch and keep running.
                 debug_assert!(branch.owned_ports.contains(&port_id.0.u32_suffix));
                 debug_assert!(branch.port_mapping.get(&port_id).is_none());
                 let mut copied_branch = Self::duplicate_branch(desc, branch_index);
                 let copied_index = copied_branch.index;
                 copied_branch.port_mapping.insert(port_id, BranchPortDesc{
                     last_registered_index: None,
                     num_times_fired: 1,
                 });
                 let branch = &mut desc.branches[branch_index as usize]; // need to reborrow
                 branch.port_mapping.insert(port_id, BranchPortDesc{
                     last_registered_index: None,
                     num_times_fired: 0,
                 });
                 // Run both again
                 desc.branches.push(copied_branch);
                 desc.spec_branches_active.push_back(branch_index);
                 desc.spec_branches_active.push_back(copied_index);
                 return Scheduling::Immediate;
             },
             RunResult::BranchMissingPortValue(port_id) => {
                 // Branch just performed a `get()` on a port that did
                 // not yet receive a value.
                 // First check if a port value is assigned to the
                 // current branch. If so, check if it is consistent.
                 debug_assert!(branch.owned_ports.contains(&port_id.0.u32_suffix));
                 let mut insert_in_pending_receive = false;
                 match branch.port_mapping.entry(port_id) {
                     Entry::Vacant(entry) => {
                         // No entry yet, so force to firing
                         entry.insert(BranchPortDesc{
                             last_registered_index: None,
                             num_times_fired: 1,
                         });
                         branch.branch_state = BranchState::BranchPoint;
                         insert_in_pending_receive = true;
                     },
                     Entry::Occupied(entry) => {
                         // Have an entry, check if it is consistent
                         let entry = entry.get();
                         if entry.num_times_fired == 0 {
                             // Inconsistent
                             branch.branch_state = BranchState::Failed;
                         } else {
                             // Perfectly fine, add to queue
                             debug_assert!(entry.last_registered_index.is_none());
                             assert_eq!(entry.num_times_fired, 1, "temp: keeping fires() for now");
                             branch.branch_state = BranchState::BranchPoint;
                             insert_in_pending_receive = true;
+                        }
+                    }
+                }
                 if insert_in_pending_receive {
                     // Perform the insert
                     match desc.spec_branches_pending_receive.entry(port_id) {
                         Entry::Vacant(entry) => {
                             entry.insert(vec![branch_index]);
+                        }
                         Entry::Occupied(mut entry) => {
                             let entry = entry.get_mut();
                             debug_assert!(!entry.contains(&branch_index));
                             entry.push(branch_index);
+                        }
+                    }
                     // But also check immediately if we don't have a
                     // previously received message. If so, we
                     // immediately branch and accept the message
                     if let Some(messages) = desc.global_inbox.find_matching_message(port_id.0.u32_suffix, None) {
                         for message in messages {
                             let mut new_branch = Self::duplicate_branch(desc, branch_index);
                             let new_branch_idx = new_branch.index;
                             let new_port_desc = new_branch.port_mapping.get_mut(&port_id).unwrap();
                             new_port_desc.last_registered_index = Some(message.peer_cur_branch_id);
                             new_branch.message_inbox.insert((port_id, 1), message.message.clone());
                             desc.branches.push(new_branch);
                             desc.spec_branches_active.push_back(new_branch_idx);
+                        }
                         if !messages.is_empty() {
                             return Scheduling::Immediate;
+                        }
+                    }
+                }
             },
             RunResult::BranchAtSyncEnd => {
                 // Check the branch for any ports that were not used and
                 // insert them in the port mapping as not having fired.
                 for port_id in branch.owned_ports.iter().copied() {
                     let port_id = PortId(Id{ connector_id: desc.id, u32_suffix: port_id });
                     if let Entry::Vacant(entry) = branch.port_mapping.entry(port_id) {
                         entry.insert(BranchPortDesc {
                             last_registered_index: None,
                             num_times_fired: 0
                         });
+                    }
+                }
                 // Mark the branch as being done
                 branch.branch_state = BranchState::ReachedEndSync;
                 desc.spec_branches_done.push(branch_index);
             },
             RunResult::BranchPut(port_id, value_group) => {
                 debug_assert!(branch.owned_ports.contains(&port_id.0.u32_suffix));
                 debug_assert_eq!(value_group.values.len(), 1); // can only send one value
                 // Branch just performed a `put()`. Check if we have
                 // assigned the port value and if so, if it is
                 // consistent.
                 let mut can_put = true;
                 branch.branch_context.just_called_did_put = true;
                 match branch.port_mapping.entry(port_id) {
                     Entry::Vacant(entry) => {
                         // No entry yet
                         entry.insert(BranchPortDesc{
                             last_registered_index: Some(branch.index),
                             num_times_fired: 1,
                         });
                     },
                     Entry::Occupied(mut entry) => {
                         // Pre-existing entry
                         let entry = entry.get_mut();
                         if entry.num_times_fired == 0 {
                             // This is 'fine' in the sense that we have
                             // a normal inconsistency in the branch.
                             branch.branch_state = BranchState::Failed;
                             can_put = false;
                         } else if entry.last_registered_index.is_none() {
                             // A put() that follows a fires()
                             entry.last_registered_index = Some(branch.index);
                         } else {
                             // This should be fine in the future. But
                             // for now we throw an error as it doesn't
                             // mesh well with the 'fires()' concept.
                             todo!("throw an error of some sort, then fail all related")
+                        }
+                    }
+                }
                 if can_put {
                     // Actually put the message in the outbox
                     let port_desc = self.ports.get(&port_id.0.u32_suffix).unwrap();
                     debug_assert_eq!(port_desc.owning_connector_id.unwrap(), connector_id);
                     let peer_id = port_desc.peer_id;
                     let peer_desc = self.ports.get(&peer_id).unwrap();
                     debug_assert!(peer_desc.owning_connector_id.is_some());
                     let peer_id = PortId(Id{
                         connector_id: peer_desc.owning_connector_id.unwrap(),
                         u32_suffix: peer_id
                     });
                     // For now this is the one and only time we're going
                     // to send a message. So for now we can't send a
                     // branch ID.
                     desc.global_outbox.insert_message(BufferedMessage{
                         sending_port: port_id,
                         receiving_port: peer_id,
                         peer_prev_branch_id: None,
                         peer_cur_branch_id: branch_index,
                         message: value_group,
                     });
                     // Finally, because we were able to put the message,
                     // we can run the branch again
                     desc.spec_branches_active.push_back(branch_index);
                     return Scheduling::Immediate;
+                }
             },
             _ => unreachable!("got result '{:?}' from running component in sync mode", run_result),
+        }
         // Did not return that we need to immediately schedule again, so
         // determine if we want to do so based on the current number of active
         // speculative branches
         if desc.spec_branches_active.is_empty() {
             return Scheduling::NotNow;
         } else {
             return Scheduling::Later;
+        }
+    }
     #[inline]
     fn run_connector_regular_mode(&mut self, connector_id: u32) -> Scheduling {
         // Retrieve the connector and the branch (which is always the first one,
         // since we assume we're not running in sync-mode).
         let desc = self.connectors.get_mut(&connector_id).unwrap();
         debug_assert!(!desc.in_sync);
         debug_assert!(desc.spec_branches_active.is_empty());
         debug_assert_eq!(desc.branches.len(), 1);
         let branch = &mut desc.branches[0];
         // Run this branch to its blocking point
         let mut run_context = Context{
             inbox: &branch.message_inbox,
             port_mapping: &branch.port_mapping,
             branch_ctx: &mut branch.branch_context,
         };
         let run_result = branch.code_state.run(&mut run_context, &self.protocol);
         match run_result {
             RunResult::ComponentTerminated => {
                 branch.branch_state = BranchState::Finished;
                 return Scheduling::NotNow
             },
             RunResult::ComponentAtSyncStart => {
                 // Prepare for sync execution
                 Self::prepare_branch_for_sync(desc);
                 return Scheduling::Immediate;
             },
             RunResult::NewComponent(definition_id, monomorph_idx, arguments) => {
                 // Find all references to ports in the provided arguments, the
                 // ownership of these ports will be transferred to the connector
                 // we're about to create.
                 let mut ports = Vec::with_capacity(arguments.values.len());
                 find_ports_in_value_group(&arguments, &mut ports);
                 // Generate a new connector with its own state
                 let new_component_id = self.generate_connector_id();
                 let new_component_state = ComponentState {
                     prompt: Prompt::new(&self.protocol.types, &self.protocol.heap, definition_id, monomorph_idx, arguments)
                 };
                 for port_id in &ports {
                     let port = self.ports.get_mut(&port_id.0.u32_suffix).unwrap();
                     debug_assert_eq!(port.owning_connector_id.unwrap(), connector_id);
                     port.owning_connector_id = Some(new_component_id)
+                }
                 // Finally push the new connector into the registry
                 let ports = ports.into_iter().map(|v| v.0.u32_suffix).collect();
                 self.connectors.insert(new_component_id, ConnectorDesc::new(new_component_id, new_component_state, ports));
                 self.connectors_active.push_back(new_component_id);
                 return Scheduling::Immediate;
             },
             RunResult::NewChannel => {
                 // Prepare channel
                 debug_assert!(run_context.branch_ctx.pending_channel.is_none());
                 let (put_id, get_id) = Self::add_owned_channel(&mut self.ports, &mut self.port_counter, Some(connector_id));
                 run_context.branch_ctx.pending_channel = Some((
                     port_value_from_id(Some(connector_id), put_id, true),
                     port_value_from_id(Some(connector_id), get_id, false)
                 ));
                 return Scheduling::Immediate;
             },
             _ => unreachable!("got result '{:?}' from running component in non-sync mode", run_result),
+        }
+    }
     /// Puts all the messages that are currently in the outbox of a particular
     /// connector into the inbox of the receivers. If possible then branches
     /// will be created that receive those messages.
     fn empty_connector_outbox(&mut self, connector_index: u32) {
         loop {
             let connector = self.connectors.get_mut(&connector_index).unwrap();
             let message_to_send = connector.global_outbox.take_next_message_to_send();
             if message_to_send.is_none() {
                 return;
+            }
             // We have a message to send
             let message_to_send = message_to_send.unwrap();
             // Lookup the target connector
             let target_port = message_to_send.receiving_port;
             let port_desc = self.ports.get(&target_port.0.u32_suffix).unwrap();
             debug_assert_eq!(port_desc.owning_connector_id.unwrap(), target_port.0.connector_id);
             let target_connector_id = port_desc.owning_connector_id.unwrap();
             let target_connector = self.connectors.get_mut(&target_connector_id).unwrap();
             // In any case, always put the message in the global inbox
             target_connector.global_inbox.insert_message(message_to_send.clone());
             // Check if there are any branches that are waiting on
             // receives
             if let Some(branch_indices) = target_connector.spec_branches_pending_receive.get(&target_port) {
                 // Check each of the branches for a port mapping that
                 // matches the one on the message header
                 for branch_index in branch_indices {
                     let branch = &mut target_connector.branches[*branch_index as usize];
                     debug_assert_eq!(branch.branch_state, BranchState::BranchPoint);
                     let mut can_branch = false;
                     if let Some(port_desc) = branch.port_mapping.get(&message_to_send.receiving_port) {
                         if port_desc.last_registered_index == message_to_send.peer_prev_branch_id && port_desc.num_times_fired == 1 {
                             can_branch = true;
+                        }
+                    }
                     if can_branch {
                         // Put the message inside a clone of the currently
                         // waiting branch
                         let mut new_branch = Self::duplicate_branch(target_connector, *branch_index);
                         let new_branch_idx = new_branch.index;
                         let new_port_desc = &mut new_branch.port_mapping.get_mut(&message_to_send.receiving_port).unwrap();
                         new_port_desc.last_registered_index = Some(message_to_send.peer_cur_branch_id);
                         new_branch.message_inbox.insert((message_to_send.receiving_port, 1), message_to_send.message.clone());
                         // And queue the branch for further execution
                         target_connector.branches.push(new_branch);
                         target_connector.spec_branches_active.push_back(new_branch_idx);
                         if !self.connectors_active.contains(&target_connector.id) {
                             self.connectors_active.push_back(target_connector.id);
+                        }
+                    }
+                }
+            }
+        }
+    }
     /// Checks a connector for the submitted solutions. After all neighbouring
     /// connectors have been checked all of their "last checked solution" index
     /// will be incremented.
     fn check_connector_new_solutions(&mut self, connector_id: u32) {
         // Take connector and start processing its solutions
         loop {
             let connector = self.connectors.get_mut(&connector_id).unwrap();
             if connector.last_checked_done == connector.spec_branches_done.len() as u32 {
                 // Nothing to do
                 return;
+            }
             // We have a new solution
             let start_branch_index = connector.spec_branches_done[connector.last_checked_done as usize];
             connector.last_checked_done += 1;
             // Check the connector+branch combination to see if a global
             // solution has already been found
             if let Some(global_solution) = self.check_connector_solution(connector_id, start_branch_index) {
                 // Found a global solution, apply it to all the connectors that
                 // participate
                 for (connector_id, local_solution) in global_solution.connector_mapping {
                     self.commit_connector_solution(connector_id, local_solution.final_branch_id);
+                }
+            }
+        }
+    }
     fn check_connector_solution(&mut self, first_connector_index: u32, first_branch_index: u32) -> Option<ProposedSolution> {
         // Take the connector and branch of interest
         let first_connector = self.connectors.get(&first_connector_index).unwrap();
         let first_branch = &first_connector.branches[first_branch_index as usize];
         debug_assert_eq!(first_branch.branch_state, BranchState::ReachedEndSync);
         // Setup the first solution
         let mut first_solution = ProposedSolution{
             connector_mapping: HashMap::new(),
             connector_constraints: HashMap::new(),
             remaining_connectors: Vec::new(),
         };
         let mut first_local_solution = ProposedConnectorSolution{
             final_branch_id: first_branch.index,
             all_branch_ids: Vec::new(),
             port_mapping: first_branch.port_mapping
                 .iter()
                 .map(|(port_id, port_info)| {
                     (port_id.0.u32_suffix, port_info.last_registered_index)
                 })
                 .collect(),
         };
         self.determine_branch_ids(first_connector, first_branch.index, &mut first_local_solution.all_branch_ids);
         first_solution.connector_mapping.insert(first_connector.id, first_local_solution);
         for (port_id, port_mapping) in first_branch.port_mapping.iter() {
             let port_desc = self.ports.get(&port_id.0.u32_suffix).unwrap();
             let peer_port_id = port_desc.peer_id;
             let peer_port_desc = self.ports.get(&peer_port_id).unwrap();
             let peer_connector_id = peer_port_desc.owning_connector_id.unwrap();
             let constraint = match port_mapping.last_registered_index {
                 Some(branch_id) => ProposedBranchConstraint::BranchNumber(branch_id),
                 None => ProposedBranchConstraint::SilentPort(peer_port_id),
             };
             match first_solution.connector_constraints.entry(peer_connector_id) {
                 Entry::Vacant(entry) => {
                     // Not yet encountered
                     entry.insert(vec![constraint]);
                     first_solution.remaining_connectors.push(peer_connector_id);
                 },
                 Entry::Occupied(mut entry) => {
                     // Already encountered
                     let entry = entry.get_mut();
                     if !entry.contains(&constraint) {
                         entry.push(constraint);
+                    }
+                }
+            }
+        }
         // Setup storage for all possible solutions
         let mut all_solutions = Vec::new();
         all_solutions.push(first_solution);
         while !all_solutions.is_empty() {
             let mut cur_solution = all_solutions.pop().unwrap();
             if cur_solution.remaining_connectors.is_empty() {
                 // All connectors have been visited, so commit the solution
                 debug_assert!(cur_solution.connector_constraints.is_empty());
                 return Some(cur_solution);
             } else {
                 // Not all connectors have been visited yet, so take one of the
                 // connectors and visit it.
                 let target_connector = cur_solution.remaining_connectors.pop().unwrap();
                 self.merge_solution_with_connector(&mut cur_solution, &mut all_solutions, target_connector);
+            }
+        }
         // No satisfying solution found
         return None;
+    }
     fn merge_solution_with_connector(&self, cur_solution: &mut ProposedSolution, all_solutions: &mut Vec<ProposedSolution>, target_connector: u32) {
         debug_assert!(!cur_solution.connector_mapping.contains_key(&target_connector)); // not yet visited
         debug_assert!(cur_solution.connector_constraints.contains_key(&target_connector)); // but we encountered a reference to it
         let branch_constraints = cur_solution.connector_constraints.get(&target_connector).unwrap();
         let cur_connector = self.connectors.get(&target_connector).unwrap();
         // Make sure all propositions are unique
         for i in 0..branch_constraints.len() {
             let proposition_i = branch_constraints[i];
             for j in 0..i {
                 let proposition_j = branch_constraints[j];
                 debug_assert_ne!(proposition_i, proposition_j);
+            }
+        }
         // Go through the current connector's branches that have finished
         'branch_loop: for finished_branch_idx in cur_connector.spec_branches_done.iter().copied() {
             let finished_branch = &cur_connector.branches[finished_branch_idx as usize];
             // Construct a list of all the parent branch numbers
             let mut parent_branch_ids = Vec::new();
             self.determine_branch_ids(cur_connector, finished_branch_idx, &mut parent_branch_ids);
             // Go through all constraints and make sure they are satisfied by
             // the current branch
             let mut all_constraints_satisfied = true;
             for constraint in branch_constraints {
                 match constraint {
                     ProposedBranchConstraint::SilentPort(port_id) => {
                         // Specified should have remained silent
                         let port_id = PortId(Id{
                             connector_id: target_connector,
                             u32_suffix: *port_id,
                         });
                         debug_assert!(finished_branch.port_mapping.contains_key(&port_id));
                         let mapped_port = finished_branch.port_mapping.get(&port_id).unwrap();
                         all_constraints_satisfied = all_constraints_satisfied && mapped_port.num_times_fired == 0;
                     },
                     ProposedBranchConstraint::BranchNumber(branch_id) => {
                         // Branch number should have appeared in the
                         // predecessor branches.
                         all_constraints_satisfied = all_constraints_satisfied && parent_branch_ids.contains(branch_id);
                     },
                     ProposedBranchConstraint::PortMapping(port_id, branch_id) => {
                         // Port should map to a particular branch number
                         let port_id = PortId(Id{
                             connector_id: target_connector,
                             u32_suffix: *port_id,
                         });
                         debug_assert!(finished_branch.port_mapping.contains_key(&port_id));
                         let mapped_port = finished_branch.port_mapping.get(&port_id).unwrap();
                         all_constraints_satisfied = all_constraints_satisfied && mapped_port.last_registered_index == Some(*branch_id);
+                    }
+                }
                 if !all_constraints_satisfied {
                     break;
+                }
+            }
             if !all_constraints_satisfied {
                 continue;
+            }
             // If here, then all constraints on the finished branch are
             // satisfied. But the finished branch also puts constraints on the
             // other connectors. So either:
             // 1. Add them to the list of constraints a peer connector should
             //  adhere to.
             // 2. Make sure that the provided connector solution matches the
             //  constraints imposed by the currently considered finished branch
             //
             // To make our lives a bit easier we already insert our proposed
             // local solution into a prepared global solution. This makes
             // looking up remote ports easier (since the channel might have its
             // two ends owned by the same connector).
             let mut new_solution = cur_solution.clone();
             debug_assert!(!new_solution.remaining_connectors.contains(&target_connector));
             new_solution.connector_constraints.remove(&target_connector);
             new_solution.connector_mapping.insert(target_connector, ProposedConnectorSolution{
                 final_branch_id: finished_branch.index,
                 all_branch_ids: parent_branch_ids,
                 port_mapping: finished_branch.port_mapping
                     .iter()
                     .map(|(port_id, port_desc)| {
                         (port_id.0.u32_suffix, port_desc.last_registered_index)
                     })
                     .collect(),
             });
             for (local_port_id, port_desc) in &finished_branch.port_mapping {
                 // Retrieve port of peer
                 let port_info = self.ports.get(&local_port_id.0.u32_suffix).unwrap();
                 let peer_port_id = port_info.peer_id;
                 let peer_port_info = self.ports.get(&peer_port_id).unwrap();
                 let peer_connector_id = peer_port_info.owning_connector_id.unwrap();
                 // If the connector was not present in the new global solution
                 // yet, add it now, as it simplifies the following logic
                 if !new_solution.connector_mapping.contains_key(&peer_connector_id) && !new_solution.remaining_connectors.contains(&peer_connector_id) {
                     new_solution.connector_constraints.insert(peer_connector_id, Vec::new());
                     new_solution.remaining_connectors.push(peer_connector_id);
+                }
                 if new_solution.remaining_connectors.contains(&peer_connector_id) {
                     // Constraint applies to a connector that has not yet been
                     // visited
                     debug_assert!(new_solution.connector_constraints.contains_key(&peer_connector_id));
                     debug_assert_ne!(peer_connector_id, target_connector);
                     let new_constraint = if port_desc.num_times_fired == 0 {
                         ProposedBranchConstraint::SilentPort(peer_port_id)
                     } else if peer_port_info.is_getter {
                         // Peer port is a getter, so we want its port to map to
                         // the branch number in our port mapping.
                         debug_assert!(port_desc.last_registered_index.is_some());
                         ProposedBranchConstraint::PortMapping(peer_port_id, port_desc.last_registered_index.unwrap())
                     } else {
                         // Peer port is a putter, so we want to restrict the
                         // solution's run to contain the branch ID we received.
                         ProposedBranchConstraint::BranchNumber(port_desc.last_registered_index.unwrap())
                     };
                     let peer_constraints = new_solution.connector_constraints.get_mut(&peer_connector_id).unwrap();
                     if !peer_constraints.contains(&new_constraint) {
                         peer_constraints.push(new_constraint);
+                    }
                 } else {
                     // Constraint applies to an already visited connector
                     let peer_solution = new_solution.connector_mapping.get(&peer_connector_id).unwrap();
                     if port_desc.num_times_fired == 0 {
                         let peer_mapped_id = peer_solution.port_mapping.get(&peer_port_id).unwrap();
                         if peer_mapped_id.is_some() {
                             all_constraints_satisfied = false;
                             break;
+                        }
                     } else if peer_port_info.is_getter {
                         // Peer is getter, so its port should be mapped to one
                         // of our branch IDs. To simplify lookup we look at the
                         // last message we sent to the getter.
                         debug_assert!(port_desc.last_registered_index.is_some());
                         let peer_port = peer_solution.port_mapping.get(&peer_port_id)
                             .map_or(None, |v| *v);
                         if port_desc.last_registered_index != peer_port {
                             // No match
                             all_constraints_satisfied = false;
                             break;
+                        }
                     } else {
                         // Peer is putter, so we expect to find our port mapping
                         // to match one of the branch numbers in the peer
                         // connector's local solution
                         debug_assert!(port_desc.last_registered_index.is_some());
                         let expected_branch_id = port_desc.last_registered_index.unwrap();
                         if !peer_solution.all_branch_ids.contains(&expected_branch_id) {
                             all_constraints_satisfied = false;
                             break;
+                        }
+                    }
+                }
+            }
             if !all_constraints_satisfied {
                 // Final checks failed
                 continue 'branch_loop
+            }
             // We're sure that this branch matches the provided solution, so
             // push it onto the list of considered solutions
             all_solutions.push(new_solution);
+        }
+    }
     fn commit_connector_solution(&mut self, connector_id: u32, branch_id: u32) {
         // Retrieve connector and branch
         let connector = self.connectors.get_mut(&connector_id).unwrap();
         debug_assert_ne!(branch_id, 0); // because at 0 we have our initial backed-up non-sync branch
         debug_assert!(connector.in_sync);
         debug_assert!(connector.spec_branches_done.contains(&branch_id));
         // Put the selected solution in front, the branch at index 0 is the
         // "non-sync" branch.
         connector.branches.swap(0, branch_id as usize);
         connector.branches.truncate(1);
         // And reset the connector's state for further execution
         connector.in_sync = false;
         connector.spec_branches_active.clear();
         connector.spec_branches_pending_receive.clear();
         connector.spec_branches_done.clear();
         connector.last_checked_done = 0;
         connector.global_inbox.clear();
         connector.global_outbox.clear();
         // Do the same thing for the final selected branch
         let final_branch = &mut connector.branches[0];
         final_branch.index = 0;
         final_branch.parent_index = None;
         debug_assert_eq!(final_branch.branch_state, BranchState::ReachedEndSync);
         final_branch.branch_state = BranchState::RunningNonSync;
         final_branch.message_inbox.clear();
         final_branch.port_mapping.clear();
         // Might be that the connector was no longer running, if so, put it back
         // in the list of connectors to run
         if !self.connectors_active.contains(&connector_id) {
             self.connectors_active.push_back(connector_id);
+        }
+    }
     fn generate_connector_id(&mut self) -> u32 {
         let id = self.connector_counter;
         self.connector_counter += 1;
         return id;
+    }
     // -------------------------------------------------------------------------
     // Helpers for port management
     // -------------------------------------------------------------------------
     #[inline]
     fn add_owned_channel(ports: &mut HashMap<u32, PortDesc>, port_counter: &mut u32, owning_connector_id: Option<u32>) -> (u32, u32) {
         let get_id = *port_counter;
         let put_id = *port_counter + 1;
         (*port_counter) += 2;
         ports.insert(get_id, PortDesc{
             id: get_id,
             peer_id: put_id,
             owning_connector_id,
             is_getter: true,
         });
         ports.insert(put_id, PortDesc{
             id: put_id,
             peer_id: get_id,
             owning_connector_id,
             is_getter: false,
         });
         return (put_id, get_id);
+    }
     // -------------------------------------------------------------------------
     // Helpers for branch management
     // -------------------------------------------------------------------------
     /// Prepares a speculative branch for further execution from the connector's
     /// non-speculative base branch.
     fn prepare_branch_for_sync(desc: &mut ConnectorDesc) {
         // Ensure only one branch is active, the non-sync branch
         debug_assert!(!desc.in_sync);
         debug_assert_eq!(desc.branches.len(), 1);
         debug_assert!(desc.spec_branches_active.is_empty());
         let new_branch_index = 1;
         // Push first speculative branch as active branch
         let new_branch = BranchDesc::new_sync_from(new_branch_index, &desc.branches[0]);
         desc.branches.push(new_branch);
         desc.spec_branches_active.push_back(new_branch_index);
         desc.in_sync = true;
+    }
     /// Duplicates a particular (speculative) branch and returns it. Due to
     /// borrowing rules in code that uses this helper the returned branch still
     /// needs to be pushed onto the member `branches`.
     fn duplicate_branch(desc: &ConnectorDesc, original_branch_idx: u32) -> BranchDesc {
         let original_branch = &desc.branches[original_branch_idx as usize];
         debug_assert!(desc.in_sync);
         let copied_index = desc.branches.len() as u32;
         let copied_branch = BranchDesc::new_sync_from(copied_index, original_branch);
         return copied_branch;
+    }
     /// Retrieves all parent IDs of a particular branch. These numbers run from
     /// the leaf towards the parent.
     fn determine_branch_ids(&self, desc: &ConnectorDesc, first_branch_index: u32, result: &mut Vec<u32>) {
         let mut next_branch_index = first_branch_index;
         result.clear();
         loop {
             result.push(next_branch_index);
             let branch = &desc.branches[next_branch_index as usize];
             match branch.parent_index {
                 Some(index) => next_branch_index = index,
                 None => return,
+            }
+        }
+    }
+}
 /// Context accessible by the code while being executed by the runtime. When the
 /// code is being executed by the runtime it sometimes needs to interact with
 /// the runtime. This is achieved by the "code throwing an error code", after
 /// which the runtime modifies the appropriate variables and continues executing
 /// the code again.
 struct Context<'a> {
     // Temporary references to branch related storage
     inbox: &'a HashMap<(PortId, u32), ValueGroup>,
     port_mapping: &'a HashMap<PortId, BranchPortDesc>,
     branch_ctx: &'a mut BranchContext,
+}
 impl<'a> crate::protocol::RunContext for Context<'a> {
-    fn did_put(&mut self, port: PortId) -> bool {
+    fn did_put(&mut self, _port: PortId) -> bool {
         // Note that we want "did put" to return false if we have fired zero
         // times, because this implies we did a prevous
         let old_value = self.branch_ctx.just_called_did_put;
         self.branch_ctx.just_called_did_put = false;
         return old_value;
+    }
     fn get(&mut self, port: PortId) -> Option<ValueGroup> {
         let inbox_key = (port, 1);
         match self.inbox.get(&inbox_key) {
             None => None,
             Some(value) => Some(value.clone()),
+        }
+    }
     fn fires(&mut self, port: PortId) -> Option<Value> {
         match self.port_mapping.get(&port) {
             None => None,
             Some(port_info) => Some(Value::Bool(port_info.num_times_fired != 0)),
+        }
+    }
     fn get_channel(&mut self) -> Option<(Value, Value)> {
         self.branch_ctx.pending_channel.take()
+    }
+}
 /// Recursively goes through the value group, attempting to find ports.
 /// Duplicates will only be added once.
 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
                 for prev_port in ports.iter() {
                     if prev_port == port_id {
                         // Already added
                         return;
+                    }
+                }
                 ports.push(*port_id);
             },
             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);
+    }
+}
 fn port_value_from_id(connector_id: Option<u32>, port_id: u32, is_output: bool) -> Value {
     let connector_id = connector_id.unwrap_or(u32::MAX); // TODO: @hack, review entire PortId/ConnectorId/Id system
     if is_output {
         return Value::Output(PortId(Id{
             connector_id,
             u32_suffix: port_id
         }));
     } else {
         return Value::Input(PortId(Id{
             connector_id,
             u32_suffix: port_id,
         }));
+    }
+}
@@ \ No newline at end of file @@

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

0 comments (0 inline, 0 general)