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

Changeset - bc4fc1729942

Parent rev.

Child rev.

[Not reviewed]

0 11 0

MH - 3 years ago 2022-02-28 13:26:09
contact@maxhenger.nl

WIP: Handling interaction between PDL/runtime for select stmt

11 files changed with 204 insertions and 81 deletions:

src/macros.rs

src/protocol/eval/executor.rs

src/protocol/eval/value.rs

src/protocol/mod.rs

src/protocol/parser/pass_rewriting.rs

src/protocol/parser/pass_typing.rs

src/protocol/tests/utils.rs

src/runtime/connector.rs

src/runtime2/component/component_pdl.rs

src/runtime2/component/consensus.rs

src/runtime2/tests/mod.rs

0 comments (0 inline, 0 general)

src/macros.rs

➞

Show inline comments

 // Utility for performing debug printing within a particular module. Still
 // requires some extra macros to be defined to be ergonomic.
 macro_rules! enabled_debug_print {
     (false, $name:literal, $format:literal) => {};
     (false, $name:literal, $format:literal, $($args:expr),*) => {};
     (true, $name:literal, $format:literal) => {
         println!("[{}] {}", $name, $format)
     };
     (true, $name:literal, $format:literal, $($args:expr),*) => {
         println!("[{}] {}", $name, format!($format, $($args),*))
     };
+}
 // Utility for inserting code only executed in debug mode. Because writing the
 // conditional cfg is tedious and looks ugly. Still doesn't work for struct
 // fields, though.
 macro_rules! dbg_code {
     ($code:stmt) => {
         #[cfg(debug_assertions)] $code
+    }
+}
 // Given a function name, return type and variant, will generate the all-so
 // common `union_value.as_variant()` method.
 macro_rules! union_cast_method_impl {
 // common `union_value.as_variant()` method. The return value is the reference
 // to the embedded union type.
 macro_rules! union_cast_to_ref_method_impl {
     ($func_name:ident, $ret_type:ty, $variant:path) => {
         fn $func_name(&self) -> &$ret_type {
             match self {
                 $variant(content) => return content,
                 _ => unreachable!(),
+            }
+        }
+    }
+}
 // Another union cast, but now returning a copy of the value
 macro_rules! union_cast_to_value_method_impl {
     ($func_name:ident, $ret_type:ty, $variant:path) => {
         impl Value {
             pub(crate) fn $func_name(&self) -> $ret_type {
                 match self {
                     $variant(v) => *v,
                     _ => unreachable!(),
+                }
+            }
+        }
+    }
+}
@@ \ No newline at end of file @@

src/protocol/eval/executor.rs

➞

Show inline comments

@@ @@ -162,96 +162,99 @@ impl Frame { @@
                         for value_expr_id in &literal.values {
                             self.expr_stack.push_back(ExprInstruction::PushValToFront);
                             self.serialize_expression(heap, *value_expr_id);
+                        }
                     },
                     Literal::Array(value_expr_ids) => {
                         for value_expr_id in value_expr_ids {
                             self.expr_stack.push_back(ExprInstruction::PushValToFront);
                             self.serialize_expression(heap, *value_expr_id);
+                        }
                     },
                     Literal::Tuple(value_expr_ids) => {
                         for value_expr_id in value_expr_ids {
                             self.expr_stack.push_back(ExprInstruction::PushValToFront);
                             self.serialize_expression(heap, *value_expr_id);
+                        }
+                    }
+                }
             },
             Expression::Cast(expr) => {
                 self.serialize_expression(heap, expr.subject);
+            }
             Expression::Call(expr) => {
                 for arg_expr_id in &expr.arguments {
                     self.expr_stack.push_back(ExprInstruction::PushValToFront);
                     self.serialize_expression(heap, *arg_expr_id);
+                }
             },
             Expression::Variable(_expr) => {
                 // No subexpressions
+            }
+        }
+    }
+}
 pub type EvalResult = Result<EvalContinuation, EvalError>;
 #[derive(Debug)]
 pub enum EvalContinuation {
     // Returned in both sync and non-sync modes
     Stepping,
     // Returned only in sync mode
     BranchInconsistent,
     SyncBlockEnd,
     NewFork,
     BlockFires(PortId),
     BlockGet(PortId),
     Put(PortId, ValueGroup),
     SelectStart(u32, u32), // (num_cases, num_ports_total)
     SelectRegisterPort(u32, u32, PortId), // (case_index, port_index_in_case, port_id)
     SelectWait, // wait until select can continue
     // Returned only in non-sync mode
     ComponentTerminated,
     SyncBlockStart,
     NewComponent(ProcedureDefinitionId, TypeId, ValueGroup),
     NewChannel,
+}
 // Note: cloning is fine, methinks. cloning all values and the heap regions then
 // we end up with valid "pointers" to heap regions.
 #[derive(Debug, Clone)]
 pub struct Prompt {
     pub(crate) frames: Vec<Frame>,
     pub(crate) store: Store,
+}
 impl Prompt {
     pub fn new(types: &TypeTable, heap: &Heap, def: ProcedureDefinitionId, type_id: TypeId, args: ValueGroup) -> Self {
         let mut prompt = Self{
             frames: Vec::new(),
             store: Store::new(),
         };
         // Maybe do typechecking in the future?
         let monomorph_index = types.get_monomorph(type_id).variant.as_procedure().monomorph_index;
         let new_frame = Frame::new(heap, def, type_id, monomorph_index);
         let max_stack_size = new_frame.max_stack_size;
         prompt.frames.push(new_frame);
         args.into_store(&mut prompt.store);
         prompt.store.reserve_stack(max_stack_size);
         prompt
+    }
     /// Big 'ol function right here. Didn't want to break it up unnecessarily.
     /// It consists of, in sequence: executing any expressions that should be
     /// executed before the next statement can be evaluated, then a section that
     /// performs debug printing, and finally a section that takes the next
     /// statement and executes it. If the statement requires any expressions to
     /// be evaluated, then they will be added such that the next time `step` is
     /// called, all of these expressions are indeed evaluated.
     pub(crate) fn step(&mut self, types: &TypeTable, heap: &Heap, modules: &[Module], ctx: &mut impl RunContext) -> EvalResult {
         // Helper function to transfer multiple values from the expression value
         // array into a heap region (e.g. constructing arrays or structs).
         fn transfer_expression_values_front_into_heap(cur_frame: &mut Frame, store: &mut Store, num_values: usize) -> HeapPos {
             let heap_pos = store.alloc_heap();
             // Do the transformation first (because Rust...)
             for val_idx in 0..num_values {
@@ @@ -607,187 +610,197 @@ impl Prompt { @@
                                         unreachable!("executor calling 'get' on value {:?}", value)
                                     };
                                     match ctx.performed_get(port_id) {
                                         Some(result) => {
                                             // We have the result. Merge the `ValueGroup` with the
                                             // stack/heap storage.
                                             debug_assert_eq!(result.values.len(), 1);
                                             result.into_stack(&mut cur_frame.expr_values, &mut self.store);
                                         },
                                         None => {
                                             // Don't have the result yet, prepare the expression to
                                             // get run again after we've received a message.
                                             cur_frame.expr_values.push_front(value.clone());
                                             cur_frame.expr_stack.push_back(ExprInstruction::EvalExpr(expr_id));
                                             return Ok(EvalContinuation::BlockGet(port_id));
+                                        }
+                                    }
                                 },
                                 Method::Put => {
                                     let port_value = cur_frame.expr_values.pop_front().unwrap();
                                     let deref_port_value = self.store.maybe_read_ref(&port_value).clone();
                                     let port_id = if let Value::Output(port_id) = deref_port_value {
                                         port_id
                                     } else {
                                         unreachable!("executor calling 'put' on value {:?}", deref_port_value)
                                     };
                                     let msg_value = cur_frame.expr_values.pop_front().unwrap();
                                     let deref_msg_value = self.store.maybe_read_ref(&msg_value).clone();
                                     if ctx.performed_put(port_id) {
                                         // We're fine, deallocate in case the expression value stack
                                         // held an owned value
                                         self.store.drop_value(msg_value.get_heap_pos());
                                     } else {
                                         // Prepare to execute again
                                         cur_frame.expr_values.push_front(msg_value);
                                         cur_frame.expr_values.push_front(port_value);
                                         cur_frame.expr_stack.push_back(ExprInstruction::EvalExpr(expr_id));
                                         let value_group = ValueGroup::from_store(&self.store, &[deref_msg_value]);
                                         return Ok(EvalContinuation::Put(port_id, value_group));
+                                    }
                                 },
                                 Method::Fires => {
                                     let port_value = cur_frame.expr_values.pop_front().unwrap();
                                     let port_value_deref = self.store.maybe_read_ref(&port_value).clone();
                                     let port_id = match port_value_deref {
                                         Value::Input(port_id) => port_id,
                                         Value::Output(port_id) => port_id,
                                         _ => unreachable!("executor calling 'fires' on value {:?}", port_value_deref),
                                     };
                                     let port_id = port_value_deref.as_port_id();
                                     match ctx.fires(port_id) {
                                         None => {
                                             cur_frame.expr_values.push_front(port_value);
                                             cur_frame.expr_stack.push_back(ExprInstruction::EvalExpr(expr_id));
                                             return Ok(EvalContinuation::BlockFires(port_id));
                                         },
                                         Some(value) => {
                                             cur_frame.expr_values.push_back(value);
+                                        }
+                                    }
                                 },
                                 Method::Create => {
                                     let length_value = cur_frame.expr_values.pop_front().unwrap();
                                     let length_value = self.store.maybe_read_ref(&length_value);
                                     let length = if length_value.is_signed_integer() {
                                         let length_value = length_value.as_signed_integer();
                                         if length_value < 0 {
                                             return Err(EvalError::new_error_at_expr(
                                                 self, modules, heap, expr_id,
                                                 format!("got length '{}', can only create a message with a non-negative length", length_value)
                                             ));
+                                        }
                                         length_value as u64
                                     } else {
                                         debug_assert!(length_value.is_unsigned_integer());
                                         length_value.as_unsigned_integer()
                                     };
                                     let heap_pos = self.store.alloc_heap();
                                     let values = &mut self.store.heap_regions[heap_pos as usize].values;
                                     debug_assert!(values.is_empty());
                                     values.resize(length as usize, Value::UInt8(0));
                                     cur_frame.expr_values.push_back(Value::Message(heap_pos));
                                 },
                                 Method::Length => {
                                     let value = cur_frame.expr_values.pop_front().unwrap();
                                     let value_heap_pos = value.get_heap_pos();
                                     let value = self.store.maybe_read_ref(&value);
                                     let heap_pos = match value {
                                         Value::Array(pos) => *pos,
                                         Value::String(pos) => *pos,
                                         _ => unreachable!("length(...) on {:?}", value),
                                     };
                                     let len = self.store.heap_regions[heap_pos as usize].values.len();
                                     // TODO: @PtrInt
                                     cur_frame.expr_values.push_back(Value::UInt32(len as u32));
                                     self.store.drop_value(value_heap_pos);
                                 },
                                 Method::Assert => {
                                     let value = cur_frame.expr_values.pop_front().unwrap();
                                     let value = self.store.maybe_read_ref(&value).clone();
                                     if !value.as_bool() {
                                         return Ok(EvalContinuation::BranchInconsistent)
+                                    }
                                 },
                                 Method::Print => {
                                     // Convert the runtime-variant of a string
                                     // into an actual string.
                                     let value = cur_frame.expr_values.pop_front().unwrap();
                                     let value_heap_pos = value.as_string();
                                     let elements = &self.store.heap_regions[value_heap_pos as usize].values;
                                     let mut message = String::with_capacity(elements.len());
                                     for element in elements {
                                         message.push(element.as_char());
+                                    }
                                     // Drop the heap-allocated value from the
                                     // store
                                     self.store.drop_heap_pos(value_heap_pos);
                                     println!("{}", message);
                                 },
                                 Method::SelectStart => {
                                     todo!("select start");
                                     let num_cases = self.store.maybe_read_ref(&cur_frame.expr_values.pop_front().unwrap()).as_uint32();
                                     let num_ports = self.store.maybe_read_ref(&cur_frame.expr_values.pop_front().unwrap()).as_uint32();
                                     if !ctx.select_start(num_cases, num_ports) {
                                         return Ok(EvalContinuation::SelectStart(num_cases, num_ports))
+                                    }
                                 },
                                 Method::SelectRegisterCasePort => {
                                     todo!("select register");
                                     let case_index = self.store.maybe_read_ref(&cur_frame.expr_values.pop_front().unwrap()).as_uint32();
                                     let port_index = self.store.maybe_read_ref(&cur_frame.expr_values.pop_front().unwrap()).as_uint32();
                                     let port_value = self.store.maybe_read_ref(&cur_frame.expr_values.pop_front().unwrap()).as_port_id();
                                     if !ctx.performed_select_start() {
                                         return Ok(EvalContinuation::SelectRegisterPort(case_index, port_index, port_value));
+                                    }
                                 },
                                 Method::SelectWait => {
                                     todo!("select wait");
                                     match ctx.performed_select_wait() {
                                         Some(select_index) => {
                                             cur_frame.expr_values.push_back(Value::UInt32(select_index));
                                         },
                                         None => return Ok(EvalContinuation::SelectWait),
+                                    }
                                 },
                                 Method::UserComponent => {
                                     // This is actually handled by the evaluation
                                     // of the statement.
                                     debug_assert_eq!(heap[expr.procedure].parameters.len(), cur_frame.expr_values.len());
                                     debug_assert_eq!(heap[cur_frame.position].as_new().expression, expr.this)
                                 },
                                 Method::UserFunction => {
                                     // Push a new frame. Note that all expressions have
                                     // been pushed to the front, so they're in the order
                                     // of the definition.
                                     let num_args = expr.arguments.len();
                                     // Determine stack boundaries
                                     let cur_stack_boundary = self.store.cur_stack_boundary;
                                     let new_stack_boundary = self.store.stack.len();
                                     // Push new boundary and function arguments for new frame
                                     self.store.stack.push(Value::PrevStackBoundary(cur_stack_boundary as isize));
                                     for _ in 0..num_args {
                                         let argument = self.store.read_take_ownership(cur_frame.expr_values.pop_front().unwrap());
                                         self.store.stack.push(argument);
+                                    }
                                     // Determine the monomorph index of the function we're calling
                                     let mono_data = &heap[cur_frame.definition].monomorphs[cur_frame.monomorph_index];
                                     let (type_id, monomorph_index) = mono_data.expr_info[expr.type_index as usize].variant.as_procedure();
                                     // Push the new frame and reserve its stack size
                                     let new_frame = Frame::new(heap, expr.procedure, type_id, monomorph_index);
                                     let new_stack_size = new_frame.max_stack_size;
                                     self.frames.push(new_frame);
                                     self.store.cur_stack_boundary = new_stack_boundary;
                                     self.store.reserve_stack(new_stack_size);
                                     // To simplify the logic a little bit we will now
                                     // return and ask our caller to call us again
                                     return Ok(EvalContinuation::Stepping);
+                                }
+                            }
                         },
                         Expression::Variable(expr) => {
                             let variable = &heap[expr.declaration.unwrap()];
                             let ref_value = if expr.used_as_binding_target {
                                 Value::Binding(variable.unique_id_in_scope as StackPos)
                             } else {
                                 Value::Ref(ValueId::Stack(variable.unique_id_in_scope as StackPos))
                             };
@@ @@ -920,98 +933,102 @@ impl Prompt { @@
                 Ok(EvalContinuation::Stepping)
             },
             Statement::Continue(stmt) => {
                 cur_frame.position = stmt.target.upcast();
                 Ok(EvalContinuation::Stepping)
             },
             Statement::Synchronous(stmt) => {
                 cur_frame.position = stmt.body;
                 Ok(EvalContinuation::SyncBlockStart)
             },
             Statement::EndSynchronous(stmt) => {
                 cur_frame.position = stmt.next;
                 let start_synchronous = &heap[stmt.start_sync];
                 let scope = &heap[start_synchronous.scope];
                 self.store.clear_stack(scope.first_unique_id_in_scope as usize);
                 Ok(EvalContinuation::SyncBlockEnd)
             },
             Statement::Fork(stmt) => {
                 if stmt.right_body.is_none() {
                     // No reason to fork
                     cur_frame.position = stmt.left_body;
                 } else {
                     // Need to fork
                     if let Some(go_left) = ctx.performed_fork() {
                         // Runtime has created a fork
                         if go_left {
                             cur_frame.position = stmt.left_body;
                         } else {
                             cur_frame.position = stmt.right_body.unwrap();
+                        }
                     } else {
                         // Request the runtime to create a fork of the current
                         // branch
                         return Ok(EvalContinuation::NewFork);
+                    }
+                }
                 Ok(EvalContinuation::Stepping)
             },
             Statement::EndFork(stmt) => {
                 cur_frame.position = stmt.next;
                 Ok(EvalContinuation::Stepping)
             },
             Statement::Select(_stmt) => {
                 todo!("implement select evaluation")
             Statement::Select(stmt) => {
                 // This is a trampoline for the statements that were placed by
                 // the AST transformation pass
                 cur_frame.position = stmt.next;
                 Ok(EvalContinuation::Stepping)
             },
             Statement::EndSelect(stmt) => {
                 cur_frame.position = stmt.next;
                 let start_select = &heap[stmt.start_select];
                 if let Some(select_case) = start_select.cases.first() {
                     let scope = &heap[select_case.scope];
                     self.store.clear_stack(scope.first_unique_id_in_scope as usize);
+                }
                 Ok(EvalContinuation::Stepping)
             },
             Statement::Return(_stmt) => {
                 debug_assert_eq!(cur_frame.expr_values.len(), 1, "expected one expr value for return statement");
                 // The preceding frame has executed a call, so is expecting the
                 // return expression on its expression value stack. Note that
                 // we may be returning a reference to something on our stack,
                 // so we need to read that value and clone it.
                 let return_value = cur_frame.expr_values.pop_back().unwrap();
                 let return_value = match return_value {
                     Value::Ref(value_id) => self.store.read_copy(value_id),
                     _ => return_value,
                 };
                 // Pre-emptively pop our stack frame
                 self.frames.pop();
                 // Clean up our section of the stack
                 self.store.clear_stack(0);
                 self.store.stack.truncate(self.store.cur_stack_boundary + 1);
                 let prev_stack_idx = self.store.stack.pop().unwrap().as_stack_boundary();
                 // TODO: Temporary hack for testing, remove at some point
                 if self.frames.is_empty() {
                     debug_assert!(prev_stack_idx == -1);
                     debug_assert!(self.store.stack.len() == 0);
                     self.store.stack.push(return_value);
                     return Ok(EvalContinuation::ComponentTerminated);
+                }
                 debug_assert!(prev_stack_idx >= 0);
                 // Return to original state of stack frame
                 self.store.cur_stack_boundary = prev_stack_idx as usize;
                 let cur_frame = self.frames.last_mut().unwrap();
                 cur_frame.expr_values.push_back(return_value);
                 // We just returned to the previous frame, which might be in
                 // the middle of evaluating expressions for a particular

src/protocol/eval/value.rs

➞

Show inline comments

@@ @@ -20,138 +20,133 @@ pub enum ValueId { @@
+}
 #[derive(Debug, Copy, Clone, PartialEq, Eq)]
 pub struct PortId{
     pub(crate) id: u32
+}
 impl PortId {
     pub fn new(id: u32) -> Self {
         return Self{ id };
+    }
+}
 /// Represents a value stored on the stack or on the heap. Some values contain
 /// a `HeapPos`, implying that they're stored in the store's `Heap`. Clearing
 /// a `Value` with a `HeapPos` from a stack must also clear the associated
 /// region from the `Heap`.
 #[derive(Debug, Clone)]
 pub enum Value {
     // Special types, never encountered during evaluation if the compiler works correctly
     Unassigned,                 // Marker when variables are first declared, immediately followed by assignment
     PrevStackBoundary(isize),   // Marker for stack frame beginning, so we can pop stack values
     Ref(ValueId),               // Reference to a value, used by expressions producing references
     Binding(StackPos),          // Reference to a binding variable (reserved on the stack)
     // Builtin types
     Input(PortId),
     Output(PortId),
     Message(HeapPos),
     Null,
     Bool(bool),
     Char(char),
     String(HeapPos),
     UInt8(u8),
     UInt16(u16),
     UInt32(u32),
     UInt64(u64),
     SInt8(i8),
     SInt16(i16),
     SInt32(i32),
     SInt64(i64),
     Array(HeapPos),
     Tuple(HeapPos),
     // Instances of user-defined types
     Enum(i64),
     Union(i64, HeapPos),
     Struct(HeapPos),
+}
 macro_rules! impl_union_unpack_as_value {
     ($func_name:ident, $variant_name:path, $return_type:ty) => {
         impl Value {
             pub(crate) fn $func_name(&self) -> $return_type {
                 match self {
                     $variant_name(v) => *v,
                     _ => panic!(concat!("called ", stringify!($func_name()), " on {:?}"), self),
+                }
+            }
+        }
+    }
+}
 impl_union_unpack_as_value!(as_stack_boundary, Value::PrevStackBoundary, isize);
 impl_union_unpack_as_value!(as_ref,     Value::Ref,     ValueId);
 impl_union_unpack_as_value!(as_input,   Value::Input,   PortId);
 impl_union_unpack_as_value!(as_output,  Value::Output,  PortId);
 impl_union_unpack_as_value!(as_message, Value::Message, HeapPos);
 impl_union_unpack_as_value!(as_bool,    Value::Bool,    bool);
 impl_union_unpack_as_value!(as_char,    Value::Char,    char);
 impl_union_unpack_as_value!(as_string,  Value::String,  HeapPos);
 impl_union_unpack_as_value!(as_uint8,   Value::UInt8,   u8);
 impl_union_unpack_as_value!(as_uint16,  Value::UInt16,  u16);
 impl_union_unpack_as_value!(as_uint32,  Value::UInt32,  u32);
 impl_union_unpack_as_value!(as_uint64,  Value::UInt64,  u64);
 impl_union_unpack_as_value!(as_sint8,   Value::SInt8,   i8);
 impl_union_unpack_as_value!(as_sint16,  Value::SInt16,  i16);
 impl_union_unpack_as_value!(as_sint32,  Value::SInt32,  i32);
 impl_union_unpack_as_value!(as_sint64,  Value::SInt64,  i64);
 impl_union_unpack_as_value!(as_array,   Value::Array,   HeapPos);
 impl_union_unpack_as_value!(as_tuple,   Value::Tuple,   HeapPos);
 impl_union_unpack_as_value!(as_enum,    Value::Enum,    i64);
 impl_union_unpack_as_value!(as_struct,  Value::Struct,  HeapPos);
 union_cast_to_value_method_impl!(as_stack_boundary, isize, Value::PrevStackBoundary);
 union_cast_to_value_method_impl!(as_ref, ValueId, Value::Ref);
 union_cast_to_value_method_impl!(as_input, PortId, Value::Input);
 union_cast_to_value_method_impl!(as_output, PortId, Value::Output);
 union_cast_to_value_method_impl!(as_message, HeapPos, Value::Message);
 union_cast_to_value_method_impl!(as_bool, bool, Value::Bool);
 union_cast_to_value_method_impl!(as_char, char, Value::Char);
 union_cast_to_value_method_impl!(as_string, HeapPos, Value::String);
 union_cast_to_value_method_impl!(as_uint8, u8, Value::UInt8);
 union_cast_to_value_method_impl!(as_uint16, u16, Value::UInt16);
 union_cast_to_value_method_impl!(as_uint32, u32, Value::UInt32);
 union_cast_to_value_method_impl!(as_uint64, u64, Value::UInt64);
 union_cast_to_value_method_impl!(as_sint8, i8, Value::SInt8);
 union_cast_to_value_method_impl!(as_sint16, i16, Value::SInt16);
 union_cast_to_value_method_impl!(as_sint32, i32, Value::SInt32);
 union_cast_to_value_method_impl!(as_sint64, i64, Value::SInt64);
 union_cast_to_value_method_impl!(as_array, HeapPos, Value::Array);
 union_cast_to_value_method_impl!(as_tuple, HeapPos, Value::Tuple);
 union_cast_to_value_method_impl!(as_enum, i64, Value::Enum);
 union_cast_to_value_method_impl!(as_struct, HeapPos, Value::Struct);
 impl Value {
     pub(crate) fn as_union(&self) -> (i64, HeapPos) {
         match self {
             Value::Union(tag, v) => (*tag, *v),
             _ => panic!("called as_union on {:?}", self),
+        }
+    }
     pub(crate) fn as_port_id(&self) -> PortId {
         match self {
             Value::Input(v) => *v,
             Value::Output(v) => *v,
             _ => unreachable!(),
+        }
+    }
     pub(crate) fn is_integer(&self) -> bool {
         match self {
             Value::UInt8(_) | Value::UInt16(_) | Value::UInt32(_) | Value::UInt64(_) |
             Value::SInt8(_) | Value::SInt16(_) | Value::SInt32(_) | Value::SInt64(_) => true,
             _ => false
+        }
+    }
     pub(crate) fn is_unsigned_integer(&self) -> bool {
         match self {
             Value::UInt8(_) | Value::UInt16(_) | Value::UInt32(_) | Value::UInt64(_) => true,
             _ => false
+        }
+    }
     pub(crate) fn is_signed_integer(&self) -> bool {
         match self {
             Value::SInt8(_) | Value::SInt16(_) | Value::SInt32(_) | Value::SInt64(_) => true,
             _ => false
+        }
+    }
     pub(crate) fn as_unsigned_integer(&self) -> u64 {
         match self {
             Value::UInt8(v)  => *v as u64,
             Value::UInt16(v) => *v as u64,
             Value::UInt32(v) => *v as u64,
             Value::UInt64(v) => *v as u64,
             _ => unreachable!("called as_unsigned_integer on {:?}", self),
+        }
+    }
     pub(crate) fn as_signed_integer(&self) -> i64 {
         match self {
             Value::SInt8(v)  => *v as i64,
             Value::SInt16(v) => *v as i64,
             Value::SInt32(v) => *v as i64,
             Value::SInt64(v) => *v as i64,
             _ => unreachable!("called as_signed_integer on {:?}", self)
+        }
+    }
     /// Returns the heap position associated with the value. If the value
     /// doesn't store anything in the heap then we return `None`.
     pub(crate) fn get_heap_pos(&self) -> Option<HeapPos> {
         match self {
             Value::Message(v) => Some(*v),
             Value::String(v) => Some(*v),

src/protocol/mod.rs

➞

Show inline comments

@@ @@ -170,87 +170,90 @@ impl ProtocolDescription { @@
+                    }
                 } else {
                     return false;
+                }
                 return true;
             },
             CTP::Array => {
                 if let Value::Array(heap_pos) = argument {
                     let heap_pos = *heap_pos;
                     for element in &arguments.regions[heap_pos as usize] {
                         if !self.verify_same_type(expected, expected_idx + 1, arguments, element) {
                             return false;
+                        }
+                    }
                     return true;
                 } else {
                     return false;
+                }
             },
             CTP::Input => if let Value::Input(_) = argument { true } else { false },
             CTP::Output => if let Value::Output(_) = argument { true } else { false },
             CTP::Tuple(_) => todo!("implement full type checking on user-supplied arguments"),
             CTP::Instance(definition_id, _num_embedded) => {
                 let definition = self.types.get_base_definition(definition_id).unwrap();
                 match &definition.definition {
                     DefinedTypeVariant::Enum(definition) => {
                         if let Value::Enum(variant_value) = argument {
                             let is_valid = definition.variants.iter()
                                 .any(|v| v.value == *variant_value);
                             return is_valid;
+                        }
                     },
                     _ => todo!("implement full type checking on user-supplied arguments"),
+                }
                 return false;
             },
+        }
+    }
+}
 pub trait RunContext {
     fn performed_put(&mut self, port: PortId) -> bool;
     fn performed_get(&mut self, port: PortId) -> Option<ValueGroup>; // None if still waiting on message
     fn fires(&mut self, port: PortId) -> Option<Value>; // None if not yet branched
     fn performed_fork(&mut self) -> Option<bool>; // None if not yet forked
     fn created_channel(&mut self) -> Option<(Value, Value)>; // None if not yet prepared
     fn performed_select_start(&mut self) -> bool; // true if performed
     fn performed_select_register_port(&mut self) -> bool; // true if registered
     fn performed_select_wait(&mut self) -> Option<u32>; // None if not yet notified runtime of select blocker
+}
 pub struct ProtocolDescriptionBuilder {
     parser: Parser,
+}
 impl ProtocolDescriptionBuilder {
     pub fn new() -> Self {
         return Self{
             parser: Parser::new(),
+        }
+    }
     pub fn add(&mut self, filename: String, buffer: Vec<u8>) -> Result<(), ParseError> {
         let input = InputSource::new(filename, buffer);
         self.parser.feed(input)?;
         return Ok(())
+    }
     pub fn compile(mut self) -> Result<ProtocolDescription, ParseError> {
         self.parser.parse()?;
         let modules: Vec<Module> = self.parser.modules.into_iter()
             .map(|module| Module{
                 source: module.source,
                 root_id: module.root_id,
                 name: module.name.map(|(_, name)| name)
             })
             .collect();
         return Ok(ProtocolDescription {
             modules,
             heap: self.parser.heap,
             types: self.parser.type_table,
             pool: Mutex::new(self.parser.string_pool),
         });
+    }
+}

src/protocol/parser/pass_rewriting.rs

➞

Show inline comments

@@ @@ -246,103 +246,105 @@ impl Visitor for PassRewriting { @@
                     transformed_stmts.push(runtime_call_stmt_id.upcast());
                     total_port_index += 1;
+                }
+            }
+        }
         // Create the variable that will hold the result of a completed select
         // block. Then create the runtime call that will produce this result
         let select_variable_id = create_ast_variable(ctx, outer_scope_id);
         let select_variable_type = TypeIdReference::DirectTypeId(ctx.arch.uint32_type_id);
         locals.push((select_variable_id, select_variable_type));
+        {
             let runtime_call_expr_id = create_ast_call_expr(ctx, self.current_procedure_id, Method::SelectWait, &mut self.expression_buffer, Vec::new());
             let variable_stmt_id = create_ast_variable_declaration_stmt(ctx, self.current_procedure_id, select_variable_id, select_variable_type, runtime_call_expr_id.upcast());
             transformed_stmts.push(variable_stmt_id.upcast().upcast());
+        }
         call_id_section.forget();
         expr_id_section.forget();
         // Now we transform each of the select block case's guard and code into
         // a chained if-else statement.
         if total_num_cases > 0 {
             let (if_stmt_id, end_if_stmt_id) = transform_select_case_code(ctx, self.current_procedure_id, id, 0, select_variable_id, select_variable_type);
             let first_end_if_stmt = &mut ctx.heap[end_if_stmt_id];
             first_end_if_stmt.next = outer_end_block_id.upcast();
             let mut last_if_stmt_id = if_stmt_id;
             let mut last_end_if_stmt_id = end_if_stmt_id;
             transformed_stmts.push(last_if_stmt_id.upcast());
             for case_index in 1..total_num_cases {
                 let (if_stmt_id, end_if_stmt_id) = transform_select_case_code(ctx, self.current_procedure_id, id, case_index, select_variable_id, select_variable_type);
                 let false_case_scope_id = ctx.heap.alloc_scope(|this| Scope::new(this, ScopeAssociation::If(last_if_stmt_id, false)));
                 set_ast_if_statement_false_body(ctx, last_if_stmt_id, last_end_if_stmt_id, IfStatementCase{ body: if_stmt_id.upcast(), scope: false_case_scope_id });
                 let end_if_stmt = &mut ctx.heap[end_if_stmt_id];
                 end_if_stmt.next = last_end_if_stmt_id.upcast();
                 last_if_stmt_id = if_stmt_id;
                 last_end_if_stmt_id = end_if_stmt_id;
+            }
+        }
         // Final steps: set the statements of the replacement block statement,
         // and link all of those statements together
         let first_stmt_id = transformed_stmts[0];
         let mut last_stmt_id = transformed_stmts[0];
         for stmt_id in transformed_stmts.iter().skip(1).copied() {
             set_ast_statement_next(ctx, last_stmt_id, stmt_id);
             last_stmt_id = stmt_id;
+        }
         let outer_block_stmt = &mut ctx.heap[outer_block_id];
         outer_block_stmt.next = first_stmt_id;
         outer_block_stmt.statements = transformed_stmts;
         return Ok(())
+    }
+}
 // -----------------------------------------------------------------------------
 // Utilities to create compiler-generated AST nodes
 // -----------------------------------------------------------------------------
 #[derive(Clone, Copy)]
 enum TypeIdReference {
     DirectTypeId(TypeId),
     IndirectSameAsExpr(i32), // by type index
+}
 fn create_ast_variable(ctx: &mut Ctx, scope_id: ScopeId) -> VariableId {
     let variable_id = ctx.heap.alloc_variable(|this| Variable{
         this,
         kind: VariableKind::Local,
         parser_type: ParserType{
             elements: Vec::new(),
             full_span: InputSpan::new(),
         },
         identifier: Identifier::new_empty(InputSpan::new()),
         relative_pos_in_parent: -1,
         unique_id_in_scope: -1,
     });
     let scope = &mut ctx.heap[scope_id];
     scope.variables.push(variable_id);
     return variable_id;
+}
 fn create_ast_variable_expr(ctx: &mut Ctx, containing_procedure_id: ProcedureDefinitionId, variable_id: VariableId, variable_type_id: TypeIdReference) -> VariableExpressionId {
     let variable_type_index = add_new_procedure_expression_type(ctx, containing_procedure_id, variable_type_id);
     return ctx.heap.alloc_variable_expression(|this| VariableExpression{
         this,
         identifier: Identifier::new_empty(InputSpan::new()),
         declaration: Some(variable_id),
         used_as_binding_target: false,
         parent: ExpressionParent::None,
         type_index: variable_type_index,
     });
+}
 fn create_ast_call_expr(ctx: &mut Ctx, containing_procedure_id: ProcedureDefinitionId, method: Method, buffer: &mut ScopedBuffer<ExpressionId>, arguments: Vec<ExpressionId>) -> CallExpressionId {
     let call_type_id = match method {

src/protocol/parser/pass_typing.rs

➞

Show inline comments

 struct InferenceNode {
     // filled in during type inference
     expr_type: InferenceType,               // result type from expression
     expr_id: ExpressionId,                  // expression that is evaluated
     inference_rule: InferenceRule,          // rule used to infer node type
     parent_index: Option<InferNodeIndex>,   // parent of inference node
     field_index: i32,                       // index of struct field or tuple member
     poly_data_index: PolyDataIndex,         // index to inference data for polymorphic types
     // filled in once type inference is done
     info_type_id: TypeId,
     info_variant: ExpressionInfoVariant,
+}
 impl InferenceNode {
     #[inline]
     fn as_expression_info(&self) -> ExpressionInfo {
         return ExpressionInfo {
             type_id: self.info_type_id,
             variant: self.info_variant
+        }
+    }
+}
 /// Inferencing rule to apply. Some of these are reasonably generic. Other ones
 /// require so much custom logic that we'll not try to come up with an
 /// abstraction.
 enum InferenceRule {
     Noop,
     MonoTemplate(InferenceRuleTemplate),
     BiEqual(InferenceRuleBiEqual),
     TriEqualArgs(InferenceRuleTriEqualArgs),
     TriEqualAll(InferenceRuleTriEqualAll),
     Concatenate(InferenceRuleTwoArgs),
     IndexingExpr(InferenceRuleIndexingExpr),
     SlicingExpr(InferenceRuleSlicingExpr),
     SelectStructField(InferenceRuleSelectStructField),
     SelectTupleMember(InferenceRuleSelectTupleMember),
     LiteralStruct(InferenceRuleLiteralStruct),
     LiteralEnum,
     LiteralUnion(InferenceRuleLiteralUnion),
     LiteralArray(InferenceRuleLiteralArray),
     LiteralTuple(InferenceRuleLiteralTuple),
     CastExpr(InferenceRuleCastExpr),
     CallExpr(InferenceRuleCallExpr),
     VariableExpr(InferenceRuleVariableExpr),
+}
 impl InferenceRule {
     union_cast_method_impl!(as_mono_template, InferenceRuleTemplate, InferenceRule::MonoTemplate);
     union_cast_method_impl!(as_bi_equal, InferenceRuleBiEqual, InferenceRule::BiEqual);
     union_cast_method_impl!(as_tri_equal_args, InferenceRuleTriEqualArgs, InferenceRule::TriEqualArgs);
     union_cast_method_impl!(as_tri_equal_all, InferenceRuleTriEqualAll, InferenceRule::TriEqualAll);
     union_cast_method_impl!(as_concatenate, InferenceRuleTwoArgs, InferenceRule::Concatenate);
     union_cast_method_impl!(as_indexing_expr, InferenceRuleIndexingExpr, InferenceRule::IndexingExpr);
     union_cast_method_impl!(as_slicing_expr, InferenceRuleSlicingExpr, InferenceRule::SlicingExpr);
     union_cast_method_impl!(as_select_struct_field, InferenceRuleSelectStructField, InferenceRule::SelectStructField);
     union_cast_method_impl!(as_select_tuple_member, InferenceRuleSelectTupleMember, InferenceRule::SelectTupleMember);
     union_cast_method_impl!(as_literal_struct, InferenceRuleLiteralStruct, InferenceRule::LiteralStruct);
     union_cast_method_impl!(as_literal_union, InferenceRuleLiteralUnion, InferenceRule::LiteralUnion);
     union_cast_method_impl!(as_literal_array, InferenceRuleLiteralArray, InferenceRule::LiteralArray);
     union_cast_method_impl!(as_literal_tuple, InferenceRuleLiteralTuple, InferenceRule::LiteralTuple);
     union_cast_method_impl!(as_cast_expr, InferenceRuleCastExpr, InferenceRule::CastExpr);
     union_cast_method_impl!(as_call_expr, InferenceRuleCallExpr, InferenceRule::CallExpr);
     union_cast_method_impl!(as_variable_expr, InferenceRuleVariableExpr, InferenceRule::VariableExpr);
     union_cast_to_ref_method_impl!(as_mono_template, InferenceRuleTemplate, InferenceRule::MonoTemplate);
     union_cast_to_ref_method_impl!(as_bi_equal, InferenceRuleBiEqual, InferenceRule::BiEqual);
     union_cast_to_ref_method_impl!(as_tri_equal_args, InferenceRuleTriEqualArgs, InferenceRule::TriEqualArgs);
     union_cast_to_ref_method_impl!(as_tri_equal_all, InferenceRuleTriEqualAll, InferenceRule::TriEqualAll);
     union_cast_to_ref_method_impl!(as_concatenate, InferenceRuleTwoArgs, InferenceRule::Concatenate);
     union_cast_to_ref_method_impl!(as_indexing_expr, InferenceRuleIndexingExpr, InferenceRule::IndexingExpr);
     union_cast_to_ref_method_impl!(as_slicing_expr, InferenceRuleSlicingExpr, InferenceRule::SlicingExpr);
     union_cast_to_ref_method_impl!(as_select_struct_field, InferenceRuleSelectStructField, InferenceRule::SelectStructField);
     union_cast_to_ref_method_impl!(as_select_tuple_member, InferenceRuleSelectTupleMember, InferenceRule::SelectTupleMember);
     union_cast_to_ref_method_impl!(as_literal_struct, InferenceRuleLiteralStruct, InferenceRule::LiteralStruct);
     union_cast_to_ref_method_impl!(as_literal_union, InferenceRuleLiteralUnion, InferenceRule::LiteralUnion);
     union_cast_to_ref_method_impl!(as_literal_array, InferenceRuleLiteralArray, InferenceRule::LiteralArray);
     union_cast_to_ref_method_impl!(as_literal_tuple, InferenceRuleLiteralTuple, InferenceRule::LiteralTuple);
     union_cast_to_ref_method_impl!(as_cast_expr, InferenceRuleCastExpr, InferenceRule::CastExpr);
     union_cast_to_ref_method_impl!(as_call_expr, InferenceRuleCallExpr, InferenceRule::CallExpr);
     union_cast_to_ref_method_impl!(as_variable_expr, InferenceRuleVariableExpr, InferenceRule::VariableExpr);
+}
 // Note: InferenceRuleTemplate is `Copy`, so don't add dynamically allocated
 // members in the future (or review places where this struct is copied)
 #[derive(Clone, Copy)]
 struct InferenceRuleTemplate {
     template: &'static [InferenceTypePart],
     application: InferenceRuleTemplateApplication,
+}
 impl InferenceRuleTemplate {
     fn new_none() -> Self {
         return Self{
             template: &[],
             application: InferenceRuleTemplateApplication::None,
         };
+    }
     fn new_forced(template: &'static [InferenceTypePart]) -> Self {
         return Self{
             template,
             application: InferenceRuleTemplateApplication::Forced,
         };
+    }
     fn new_template(template: &'static [InferenceTypePart]) -> Self {
         return Self{
             template,
             application: InferenceRuleTemplateApplication::Template,
+        }
+    }
+}
 #[derive(Clone, Copy)]
 enum InferenceRuleTemplateApplication {
     None, // do not apply template, silly, but saves some bytes
     Forced,
     Template,
+}
 /// Type equality applied to 'self' and the argument. An optional template will
 /// be applied to 'self' first. Example: "bitwise not"
 struct InferenceRuleBiEqual {
     template: InferenceRuleTemplate,
     argument_index: InferNodeIndex,
+}
 /// Type equality applied to two arguments. Template can be applied to 'self'

src/protocol/tests/utils.rs

➞

Show inline comments

             for stmt_id in &stmt.statements {
                 if let Some(id) = seek_expr_in_stmt(heap, *stmt_id, f) {
                     return Some(id)
+                }
+            }
             None
         },
         Statement::Labeled(stmt) => {
             seek_expr_in_stmt(heap, stmt.body, f)
         },
         Statement::If(stmt) => {
             None
             .or_else(|| seek_expr_in_expr(heap, stmt.test, f))
             .or_else(|| seek_expr_in_stmt(heap, stmt.true_case.body, f))
             .or_else(|| if let Some(false_body) = stmt.false_case {
                 seek_expr_in_stmt(heap, false_body.body, f)
             } else {
                 None
             })
         },
         Statement::While(stmt) => {
             None
             .or_else(|| seek_expr_in_expr(heap, stmt.test, f))
             .or_else(|| seek_expr_in_stmt(heap, stmt.body, f))
         },
         Statement::Synchronous(stmt) => {
             seek_expr_in_stmt(heap, stmt.body, f)
         },
         Statement::Return(stmt) => {
             for expr_id in &stmt.expressions {
                 if let Some(id) = seek_expr_in_expr(heap, *expr_id, f) {
                     return Some(id);
+                }
+            }
             None
         },
         Statement::New(stmt) => {
             seek_expr_in_expr(heap, stmt.expression.upcast(), f)
         },
         Statement::Expression(stmt) => {
             seek_expr_in_expr(heap, stmt.expression, f)
         },
         _ => None
+    }
+}
 struct FakeRunContext{}
 impl RunContext for FakeRunContext {
     fn performed_put(&mut self, _port: PortId) -> bool {
         unreachable!("'put' called in compiler testing code")
+    }
     fn performed_get(&mut self, _port: PortId) -> Option<ValueGroup> {
         unreachable!("'get' called in compiler testing code")
+    }
     fn fires(&mut self, _port: PortId) -> Option<Value> {
         unreachable!("'fires' called in compiler testing code")
+    }
     fn performed_fork(&mut self) -> Option<bool> {
         unreachable!("'fork' called in compiler testing code")
+    }
     fn created_channel(&mut self) -> Option<(Value, Value)> {
         unreachable!("channel created in compiler testing code")
+    }
     fn performed_put(&mut self, _port: PortId) -> bool { unreachable!() }
     fn performed_get(&mut self, _port: PortId) -> Option<ValueGroup> { unreachable!() }
     fn fires(&mut self, _port: PortId) -> Option<Value> { unreachable!() }
     fn performed_fork(&mut self) -> Option<bool> { unreachable!() }
     fn created_channel(&mut self) -> Option<(Value, Value)> { unreachable!() }
     fn performed_select_start(&mut self) -> bool { unreachable!() }
     fn performed_select_register_port(&mut self) -> bool { unreachable!() }
     fn performed_select_wait(&mut self) -> Option<u32> { unreachable!() }
+}
@@ \ No newline at end of file @@

src/runtime/connector.rs

➞

Show inline comments

@@ @@ -77,96 +77,100 @@ pub(crate) struct ConnectorPDL { @@
     last_finished_handled: Option<BranchId>,
+}
 struct ConnectorRunContext<'a> {
     branch_id: BranchId,
     consensus: &'a Consensus,
     prepared: PreparedStatement,
+}
 impl<'a> RunContext for ConnectorRunContext<'a>{
     fn performed_put(&mut self, _port: PortId) -> bool {
         return match self.prepared.take() {
             PreparedStatement::None => false,
             PreparedStatement::PerformedPut => true,
             taken => unreachable!("prepared statement is '{:?}' during 'performed_put()'", taken)
         };
+    }
     fn performed_get(&mut self, _port: PortId) -> Option<ValueGroup> {
         return match self.prepared.take() {
             PreparedStatement::None => None,
             PreparedStatement::PerformedGet(value) => Some(value),
             taken => unreachable!("prepared statement is '{:?}' during 'performed_get()'", taken),
         };
+    }
     fn fires(&mut self, _port: PortId) -> Option<Value> {
         todo!("Remove fires() now")
         // let port_id = PortIdLocal::new(port.id);
         // let annotation = self.consensus.get_annotation(self.branch_id, port_id);
         // return annotation.expected_firing.map(|v| Value::Bool(v));
+    }
     fn created_channel(&mut self) -> Option<(Value, Value)> {
         return match self.prepared.take() {
             PreparedStatement::None => None,
             PreparedStatement::CreatedChannel(ports) => Some(ports),
             taken => unreachable!("prepared statement is '{:?}' during 'created_channel()'", taken),
         };
+    }
     fn performed_fork(&mut self) -> Option<bool> {
         return match self.prepared.take() {
             PreparedStatement::None => None,
             PreparedStatement::ForkedExecution(path) => Some(path),
             taken => unreachable!("prepared statement is '{:?}' during 'performed_fork()'", taken),
         };
+    }
     fn performed_select_start(&mut self) -> bool { unreachable!() }
     fn performed_select_register_port(&mut self) -> bool { unreachable!() }
     fn performed_select_wait(&mut self) -> Option<u32> { unreachable!() }
+}
 impl Connector for ConnectorPDL {
     fn run(&mut self, sched_ctx: SchedulerCtx, comp_ctx: &mut ComponentCtx) -> ConnectorScheduling {
         if let Some(scheduling) = self.handle_new_messages(comp_ctx) {
             return scheduling;
+        }
         match self.mode {
             Mode::Sync => {
                 // Run in sync mode
                 let scheduling = self.run_in_sync_mode(sched_ctx, comp_ctx);
                 // Handle any new finished branches
                 let mut iter_id = self.last_finished_handled.or(self.tree.get_queue_first(QueueKind::FinishedSync));
                 while let Some(branch_id) = iter_id {
                     iter_id = self.tree.get_queue_next(branch_id);
                     self.last_finished_handled = Some(branch_id);
                     if let Some(round_conclusion) = self.consensus.handle_new_finished_sync_branch(branch_id, comp_ctx) {
                         // Actually found a solution
                         return self.enter_non_sync_mode(round_conclusion, comp_ctx);
+                    }
                     self.last_finished_handled = Some(branch_id);
+                }
                 return scheduling;
             },
             Mode::NonSync => {
                 let scheduling = self.run_in_deterministic_mode(sched_ctx, comp_ctx);
                 return scheduling;
             },
             Mode::SyncError => {
                 let scheduling = self.run_in_sync_mode(sched_ctx, comp_ctx);
                 return scheduling;
             },
             Mode::Error => {
                 // This shouldn't really be called. Because when we reach exit
                 // mode the scheduler should not run the component anymore
                 unreachable!("called component run() during error-mode");
             },
+        }
+    }
+}
 impl ConnectorPDL {
     pub fn new(initial: Prompt) -> Self {

src/runtime2/component/component_pdl.rs

➞

Show inline comments

 use crate::protocol::*;
 use crate::protocol::ast::ProcedureDefinitionId;
 use crate::protocol::eval::{
     PortId as EvalPortId, Prompt,
     ValueGroup, Value,
     EvalContinuation, EvalResult, EvalError
 };
 use crate::runtime2::scheduler::SchedulerCtx;
 use crate::runtime2::communication::*;
 use super::component_context::*;
 use super::control_layer::*;
 use super::consensus::Consensus;
 pub enum CompScheduling {
     Immediate,
     Requeue,
     Sleep,
     Exit,
+}
 pub enum ExecStmt {
     CreatedChannel((Value, Value)),
     PerformedPut,
     PerformedGet(ValueGroup),
     PerformedSelectStart,
     PerformedSelectRegister,
     PerformedSelectWait(u32),
     None,
+}
 impl ExecStmt {
     fn take(&mut self) -> ExecStmt {
         let mut value = ExecStmt::None;
         std::mem::swap(self, &mut value);
         return value;
+    }
     fn is_none(&self) -> bool {
         match self {
             ExecStmt::None => return true,
             _ => return false,
+        }
+    }
+}
 pub struct ExecCtx {
     stmt: ExecStmt,
+}
 impl RunContext for ExecCtx {
     fn performed_put(&mut self, _port: EvalPortId) -> bool {
         match self.stmt.take() {
             ExecStmt::None => return false,
             ExecStmt::PerformedPut => return true,
             _ => unreachable!(),
+        }
+    }
     fn performed_get(&mut self, _port: EvalPortId) -> Option<ValueGroup> {
         match self.stmt.take() {
             ExecStmt::None => return None,
             ExecStmt::PerformedGet(value) => return Some(value),
             _ => unreachable!(),
+        }
+    }
     fn fires(&mut self, _port: EvalPortId) -> Option<Value> {
         todo!("remove fires")
+    }
     fn performed_fork(&mut self) -> Option<bool> {
         todo!("remove fork")
+    }
     fn created_channel(&mut self) -> Option<(Value, Value)> {
         match self.stmt.take() {
             ExecStmt::None => return None,
             ExecStmt::CreatedChannel(ports) => return Some(ports),
             _ => unreachable!(),
+        }
+    }
     fn performed_select_start(&mut self) -> bool {
         match self.stmt.take() {
             ExecStmt::None => return false,
             ExecStmt::PerformedSelectStart => return true,
             _ => unreachable!(),
+        }
+    }
     fn performed_select_register_port(&mut self) -> bool {
         match self.stmt.take() {
             ExecStmt::None => return false,
             ExecStmt::PerformedSelectRegister => return true,
             _ => unreachable!(),
+        }
+    }
     fn performed_select_wait(&mut self) -> Option<u32> {
         match self.stmt.take() {
             ExecStmt::None => return None,
             ExecStmt::PerformedSelectWait(selected_case) => Some(selected_case),
             _ => unreachable!(),
+        }
+    }
+}
 #[derive(Debug, Copy, Clone, PartialEq, Eq)]
 pub(crate) enum Mode {
     NonSync, // not in sync mode
     Sync, // in sync mode, can interact with other components
     SyncEnd, // awaiting a solution, i.e. encountered the end of the sync block
     BlockedGet,
     BlockedPut,
     StartExit, // temporary state: if encountered then we start the shutdown process
     BusyExit, // temporary state: waiting for Acks for all the closed ports
     Exit, // exiting: shutdown process started, now waiting until the reference count drops to 0
+}
 pub(crate) struct CompPDL {
     pub mode: Mode,
     pub mode_port: PortId, // when blocked on a port
     pub mode_value: ValueGroup, // when blocked on a put
     pub prompt: Prompt,
     pub control: ControlLayer,
     pub consensus: Consensus,
     pub sync_counter: u32,
     pub exec_ctx: ExecCtx,
     // TODO: Temporary field, simulates future plans of having one storage place
     //  reserved per port.
     // Should be same length as the number of ports. Corresponding indices imply
     // message is intended for that port.
     pub inbox_main: Vec<Option<DataMessage>>,
     pub inbox_backup: Vec<DataMessage>,
+}
 impl CompPDL {
     pub(crate) fn new(initial_state: Prompt, num_ports: usize) -> Self {
         let mut inbox_main = Vec::new();
         inbox_main.reserve(num_ports);
         for _ in 0..num_ports {
             inbox_main.push(None);
+        }
         return Self{
             mode: Mode::NonSync,
             mode_port: PortId::new_invalid(),
             mode_value: ValueGroup::default(),
             prompt: initial_state,
             control: ControlLayer::default(),
             consensus: Consensus::new(),
             sync_counter: 0,
             exec_ctx: ExecCtx{
@@ @@ -200,183 +227,203 @@ impl CompPDL { @@
                 self.handle_sync_end(sched_ctx, comp_ctx);
                 return Ok(CompScheduling::Immediate);
             },
             EC::BlockGet(port_id) => {
                 debug_assert_eq!(self.mode, Mode::Sync);
                 debug_assert!(self.exec_ctx.stmt.is_none());
                 let port_id = port_id_from_eval(port_id);
                 let port_handle = comp_ctx.get_port_handle(port_id);
                 let port_index = comp_ctx.get_port_index(port_handle);
                 if let Some(message) = &self.inbox_main[port_index] {
                     // Check if we can actually receive the message
                     if self.consensus.try_receive_data_message(sched_ctx, comp_ctx, message) {
                         // Message was received. Make sure any blocked peers and
                         // pending messages are handled.
                         let message = self.inbox_main[port_index].take().unwrap();
                         self.handle_received_data_message(sched_ctx, comp_ctx, port_handle);
                         self.exec_ctx.stmt = ExecStmt::PerformedGet(message.content);
                         return Ok(CompScheduling::Immediate);
                     } else {
                         todo!("handle sync failure due to message deadlock");
                         return Ok(CompScheduling::Sleep);
+                    }
                 } else {
                     // We need to wait
                     self.mode = Mode::BlockedGet;
                     self.mode_port = port_id;
                     return Ok(CompScheduling::Sleep);
+                }
             },
             EC::Put(port_id, value) => {
                 debug_assert_eq!(self.mode, Mode::Sync);
                 let port_id = port_id_from_eval(port_id);
                 let port_handle = comp_ctx.get_port_handle(port_id);
                 let port_info = comp_ctx.get_port(port_handle);
                 if port_info.state.is_blocked() {
                     self.mode = Mode::BlockedPut;
                     self.mode_port = port_id;
                     self.mode_value = value;
                     self.exec_ctx.stmt = ExecStmt::PerformedPut; // prepare for when we become unblocked
                     return Ok(CompScheduling::Sleep);
                 } else {
                     self.send_data_message_and_wake_up(sched_ctx, comp_ctx, port_handle, value);
                     self.exec_ctx.stmt = ExecStmt::PerformedPut;
                     return Ok(CompScheduling::Immediate);
+                }
             },
             EC::SelectStart(num_cases, num_ports) => {
                 debug_assert_eq!(self.mode, Mode::Sync);
                 todo!("finish handling select start")
             },
             EC::SelectRegisterPort(case_index, port_index, port_id) => {
                 debug_assert_eq!(self.mode, Mode::Sync);
                 todo!("finish handling register port")
             },
             EC::SelectWait => {
                 debug_assert_eq!(self.mode, Mode::Sync);
                 self.handle_select_wait(sched_ctx, comp_ctx);
                 todo!("finish handling select wait")
             },
             // Results that can be returned outside of sync mode
             EC::ComponentTerminated => {
                 self.mode = Mode::StartExit; // next call we'll take care of the exit
                 return Ok(CompScheduling::Immediate);
             },
             EC::SyncBlockStart => {
                 debug_assert_eq!(self.mode, Mode::NonSync);
                 self.handle_sync_start(sched_ctx, comp_ctx);
                 return Ok(CompScheduling::Immediate);
             },
             EC::NewComponent(definition_id, type_id, arguments) => {
                 debug_assert_eq!(self.mode, Mode::NonSync);
                 self.create_component_and_transfer_ports(
                     sched_ctx, comp_ctx,
                     definition_id, type_id, arguments
                 );
                 return Ok(CompScheduling::Requeue);
             },
             EC::NewChannel => {
                 debug_assert_eq!(self.mode, Mode::NonSync);
                 debug_assert!(self.exec_ctx.stmt.is_none());
                 let channel = comp_ctx.create_channel();
                 self.exec_ctx.stmt = ExecStmt::CreatedChannel((
                     Value::Output(port_id_to_eval(channel.putter_id)),
                     Value::Input(port_id_to_eval(channel.getter_id))
                 ));
                 self.inbox_main.push(None);
                 self.inbox_main.push(None);
                 return Ok(CompScheduling::Immediate);
+            }
+        }
+    }
     fn execute_prompt(&mut self, sched_ctx: &SchedulerCtx) -> EvalResult {
         let mut step_result = EvalContinuation::Stepping;
         while let EvalContinuation::Stepping = step_result {
             step_result = self.prompt.step(
                 &sched_ctx.runtime.protocol.types, &sched_ctx.runtime.protocol.heap,
                 &sched_ctx.runtime.protocol.modules, &mut self.exec_ctx,
             )?;
+        }
         return Ok(step_result)
+    }
     fn handle_sync_start(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx) {
         sched_ctx.log("Component starting sync mode");
         self.consensus.notify_sync_start(comp_ctx);
         debug_assert_eq!(self.mode, Mode::NonSync);
         self.mode = Mode::Sync;
+    }
     /// Handles end of sync. The conclusion to the sync round might arise
     /// immediately (and be handled immediately), or might come later through
     /// messaging. In any case the component should be scheduled again
     /// immediately
     fn handle_sync_end(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx) {
         sched_ctx.log("Component ending sync mode (now waiting for solution)");
         let decision = self.consensus.notify_sync_end(sched_ctx, comp_ctx);
         self.mode = Mode::SyncEnd;
         self.handle_sync_decision(sched_ctx, comp_ctx, decision);
+    }
     /// Handles decision from the consensus round. This will cause a change in
     /// the internal `Mode`, such that the next call to `run` can take the
     /// appropriate next steps.
     fn handle_sync_decision(&mut self, sched_ctx: &SchedulerCtx, _comp_ctx: &mut CompCtx, decision: SyncRoundDecision) {
         sched_ctx.log(&format!("Handling sync decision: {:?} (in mode {:?})", decision, self.mode));
         let is_success = match decision {
             SyncRoundDecision::None => {
                 // No decision yet
                 return;
             },
             SyncRoundDecision::Solution => true,
             SyncRoundDecision::Failure => false,
         };
         // If here then we've reached a decision
         debug_assert_eq!(self.mode, Mode::SyncEnd);
         if is_success {
             self.mode = Mode::NonSync;
             self.consensus.notify_sync_decision(decision);
         } else {
             self.mode = Mode::StartExit;
+        }
+    }
     /// Handles the moment where the PDL code has notified the runtime of all
     /// the ports it is waiting on.
     fn handle_select_wait(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx) {
         sched_ctx.log("Component waiting for select conclusion");
+    }
     fn handle_component_exit(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx) {
         sched_ctx.log("Component exiting");
         debug_assert_eq!(self.mode, Mode::StartExit);
         self.mode = Mode::BusyExit;
         // Doing this by index, then retrieving the handle is a bit rediculous,
         // but Rust is being Rust with its borrowing rules.
         for port_index in 0..comp_ctx.num_ports() {
             let port = comp_ctx.get_port_by_index_mut(port_index);
             if port.state == PortState::Closed {
                 // Already closed, or in the process of being closed
                 continue;
+            }
             // Mark as closed
             let port_id = port.self_id;
             port.state = PortState::Closed;
             // Notify peer of closing
             let port_handle = comp_ctx.get_port_handle(port_id);
             let (peer, message) = self.control.initiate_port_closing(port_handle, comp_ctx);
             let peer_info = comp_ctx.get_peer(peer);
             peer_info.handle.send_message(sched_ctx, Message::Control(message), true);
+        }
+    }
     // -------------------------------------------------------------------------
     // Handling messages
     // -------------------------------------------------------------------------
     fn send_data_message_and_wake_up(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &CompCtx, source_port_handle: LocalPortHandle, value: ValueGroup) {
         let port_info = comp_ctx.get_port(source_port_handle);
         let peer_handle = comp_ctx.get_peer_handle(port_info.peer_comp_id);
         let peer_info = comp_ctx.get_peer(peer_handle);
         let annotated_message = self.consensus.annotate_data_message(comp_ctx, port_info, value);
         peer_info.handle.send_message(sched_ctx, Message::Data(annotated_message), true);
+    }
     /// Handles a message that came in through the public inbox. This function
     /// will handle putting it in the correct place, and potentially blocking
     /// the port in case too many messages are being received.
     fn handle_incoming_data_message(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, message: DataMessage) {
         // Check if we can insert it directly into the storage associated with
         // the port
         let target_port_id = message.data_header.target_port;
         let port_handle = comp_ctx.get_port_handle(target_port_id);
         let port_index = comp_ctx.get_port_index(port_handle);
         if self.inbox_main[port_index].is_none() {

src/runtime2/component/consensus.rs

➞

Show inline comments

 use crate::protocol::eval::ValueGroup;
 use crate::runtime2::scheduler::*;
 use crate::runtime2::runtime::*;
 use crate::runtime2::communication::*;
 use super::component_context::*;
 pub struct PortAnnotation {
     self_comp_id: CompId,
     self_port_id: PortId,
     peer_comp_id: CompId, // only valid for getter ports
     peer_port_id: PortId, // only valid for getter ports
     mapping: Option<u32>,
+}
 impl PortAnnotation {
     fn new(comp_id: CompId, port_id: PortId) -> Self {
         return Self{
             self_comp_id: comp_id,
             self_port_id: port_id,
             peer_comp_id: CompId::new_invalid(),
             peer_port_id: PortId::new_invalid(),
             mapping: None
+        }
+    }
+}
 #[derive(Debug, Eq, PartialEq)]
 enum Mode {
     NonSync,
     SyncBusy,
     SyncAwaitingSolution,
     SelectBusy,
     SelectWait,
+}
 struct SolutionCombiner {
     solution: SyncPartialSolution,
     matched_channels: usize,
+}
 impl SolutionCombiner {
     fn new() -> Self {
         return Self {
             solution: SyncPartialSolution::default(),
             matched_channels: 0,
+        }
+    }
     #[inline]
     fn has_contributions(&self) -> bool {
         return !self.solution.channel_mapping.is_empty();
+    }
     /// Returns a decision for the current round. If there is no decision (yet)
     /// then `RoundDecision::None` is returned.
     fn get_decision(&self) -> SyncRoundDecision {
         if self.matched_channels == self.solution.channel_mapping.len() {
             debug_assert_ne!(self.solution.decision, SyncRoundDecision::None);
             return self.solution.decision;
+        }
         return SyncRoundDecision::None; // even in case of failure: wait for everyone.
+    }
     fn combine_with_partial_solution(&mut self, partial: SyncPartialSolution) {
         debug_assert_ne!(self.solution.decision, SyncRoundDecision::Solution);
         debug_assert_ne!(partial.decision, SyncRoundDecision::Solution);
         if partial.decision == SyncRoundDecision::Failure {
             self.solution.decision = SyncRoundDecision::Failure;
+        }
         for entry in partial.channel_mapping {
             let channel_index = if entry.getter.is_some() && entry.putter.is_some() {
                 let channel_index = self.solution.channel_mapping.len();
                 self.solution.channel_mapping.push(entry);
                 self.matched_channels += 1;
                 channel_index
             } else if let Some(putter) = entry.putter {
                 self.combine_with_putter_port(putter)

src/runtime2/tests/mod.rs

➞

Show inline comments

@@ @@ -37,49 +37,98 @@ fn test_component_communication() { @@
         u32 outside_index = 0;
         while (outside_index < outside_loops) {
             u32 inside_index = 0;
             sync while (inside_index < inside_loops) {
                 put(o, inside_index);
                 inside_index += 1;
+            }
             outside_index += 1;
+        }
+    }
     primitive receiver(in<u32> i, u32 outside_loops, u32 inside_loops) {
         u32 outside_index = 0;
         while (outside_index < outside_loops) {
             u32 inside_index = 0;
             sync while (inside_index < inside_loops) {
                 auto val = get(i);
                 while (val != inside_index) {} // infinite loop if incorrect value is received
                 inside_index += 1;
+            }
             outside_index += 1;
+        }
+    }
     composite constructor() {
         channel o_orom -> i_orom;
         channel o_mrom -> i_mrom;
         channel o_ormm -> i_ormm;
         channel o_mrmm -> i_mrmm;
         // one round, one message per round
         new sender(o_orom, 1, 1);
         new receiver(i_orom, 1, 1);
         // multiple rounds, one message per round
         new sender(o_mrom, 5, 1);
         new receiver(i_mrom, 5, 1);
         // one round, multiple messages per round
         new sender(o_ormm, 1, 5);
         new receiver(i_ormm, 1, 5);
         // multiple rounds, multiple messages per round
         new sender(o_mrmm, 5, 5);
         new receiver(i_mrmm, 5, 5);
     }").expect("compilation");
     let rt = Runtime::new(3, true, pd);
     create_component(&rt, "", "constructor", no_args());
+}
 #[test]
 fn test_simple_select() {
     let pd = ProtocolDescription::parse(b"
     func infinite_assert<T>(T val, T expected) -> () {
         while (val != expected) { print(\"nope!\"); }
+    }
     primitive receiver(in<u32> in_a, in<u32> in_b, u32 num_sends) {
         auto num_from_a = 0;
         auto num_from_b = 0;
         while (num_from_a + num_from_b < 2 * num_sends) {
             sync select {
                 auto v = get(in_a) -> {
                     print(\"got something from A\");
                     infinite_assert(v, num_from_a);
                     num_from_a += 1;
+                }
                 auto v = get(in_b) -> {
                     print(\"got something from B\");
                     infinite_assert(v, num_from_b);
                     num_from_b +=1;
+                }
+            }
+        }
+    }
     primitive sender(out<u32> tx, u32 num_sends) {
         auto index = 0;
         while (index < num_sends) {
             sync {
                 put(tx, index);
                 index += 1;
+            }
+        }
+    }
     composite constructor() {
         auto num_sends = 3;
         channel tx_a -> rx_a;
         channel tx_b -> rx_b;
         new sender(tx_a, num_sends);
         new receiver(rx_a, rx_b, num_sends);
         new sender(tx_b, num_sends);
+    }
     ").expect("compilation");
     let rt = Runtime::new(1, true, pd);
     create_component(&rt, "", "constructor", no_args());
+}
@@ \ No newline at end of file @@

0 comments (0 inline, 0 general)