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

Changeset - fb814548c7d5

Parent rev.

Child rev.

[Not reviewed]

0 14 0

mh - 3 years ago 2022-04-12 09:30:48
contact@maxhenger.nl

Add tcp component to standard library

14 files changed with 62 insertions and 39 deletions:

src/protocol/ast.rs

src/protocol/eval/executor.rs

src/protocol/parser/mod.rs

src/protocol/parser/pass_definitions.rs

src/protocol/parser/pass_validation_linking.rs

src/runtime2/component/component.rs

src/runtime2/component/component_internet.rs

src/runtime2/component/component_pdl.rs

src/runtime2/component/component_random.rs

src/runtime2/poll/mod.rs

src/runtime2/runtime.rs

src/runtime2/stdlib/internet.rs

src/runtime2/tests/mod.rs

std/std.internet.pdl

0 comments (0 inline, 0 general)

src/protocol/ast.rs

➞

Show inline comments

@@ @@ -897,384 +897,385 @@ impl Definition { @@
         match self {
             Definition::Struct(def) => def.defined_in,
             Definition::Enum(def) => def.defined_in,
             Definition::Union(def) => def.defined_in,
             Definition::Procedure(def) => def.defined_in,
+        }
+    }
     pub fn identifier(&self) -> &Identifier {
         match self {
             Definition::Struct(def) => &def.identifier,
             Definition::Enum(def) => &def.identifier,
             Definition::Union(def) => &def.identifier,
             Definition::Procedure(def) => &def.identifier,
+        }
+    }
     pub fn poly_vars(&self) -> &Vec<Identifier> {
         match self {
             Definition::Struct(def) => &def.poly_vars,
             Definition::Enum(def) => &def.poly_vars,
             Definition::Union(def) => &def.poly_vars,
             Definition::Procedure(def) => &def.poly_vars,
+        }
+    }
+}
 #[derive(Debug, Clone)]
 pub struct StructFieldDefinition {
     pub span: InputSpan,
     pub field: Identifier,
     pub parser_type: ParserType,
+}
 #[derive(Debug, Clone)]
 pub struct StructDefinition {
     pub this: StructDefinitionId,
     pub defined_in: RootId,
     // Symbol scanning
     pub identifier: Identifier,
     pub poly_vars: Vec<Identifier>,
     // Parsing
     pub fields: Vec<StructFieldDefinition>
+}
 impl StructDefinition {
     pub(crate) fn new_empty(
         this: StructDefinitionId, defined_in: RootId,
         identifier: Identifier, poly_vars: Vec<Identifier>
     ) -> Self {
         Self{ this, defined_in, identifier, poly_vars, fields: Vec::new() }
+    }
+}
 #[derive(Debug, Clone, Copy)]
 pub enum EnumVariantValue {
     None,
     Integer(i64),
+}
 #[derive(Debug, Clone)]
 pub struct EnumVariantDefinition {
     pub identifier: Identifier,
     pub value: EnumVariantValue,
+}
 #[derive(Debug, Clone)]
 pub struct EnumDefinition {
     pub this: EnumDefinitionId,
     pub defined_in: RootId,
     // Symbol scanning
     pub identifier: Identifier,
     pub poly_vars: Vec<Identifier>,
     // Parsing
     pub variants: Vec<EnumVariantDefinition>,
+}
 impl EnumDefinition {
     pub(crate) fn new_empty(
         this: EnumDefinitionId, defined_in: RootId,
         identifier: Identifier, poly_vars: Vec<Identifier>
     ) -> Self {
         Self{ this, defined_in, identifier, poly_vars, variants: Vec::new() }
+    }
+}
 #[derive(Debug, Clone)]
 pub struct UnionVariantDefinition {
     pub span: InputSpan,
     pub identifier: Identifier,
     pub value: Vec<ParserType>, // if empty, then union variant does not contain any embedded types
+}
 #[derive(Debug, Clone)]
 pub struct UnionDefinition {
     pub this: UnionDefinitionId,
     pub defined_in: RootId,
     // Phase 1: symbol scanning
     pub identifier: Identifier,
     pub poly_vars: Vec<Identifier>,
     // Phase 2: parsing
     pub variants: Vec<UnionVariantDefinition>,
+}
 impl UnionDefinition {
     pub(crate) fn new_empty(
         this: UnionDefinitionId, defined_in: RootId,
         identifier: Identifier, poly_vars: Vec<Identifier>
     ) -> Self {
         Self{ this, defined_in, identifier, poly_vars, variants: Vec::new() }
+    }
+}
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum ProcedureKind {
     Function, // with return type
     Primitive, // without return type
     Composite,
+}
 /// Monomorphed instantiation of a procedure (or the sole instantiation of a
 /// non-polymorphic procedure).
 #[derive(Debug)]
 pub struct ProcedureDefinitionMonomorph {
     pub argument_types: Vec<TypeId>,
     pub expr_info: Vec<ExpressionInfo>
+}
 impl ProcedureDefinitionMonomorph {
     pub(crate) fn new_invalid() -> Self {
         return Self{
             argument_types: Vec::new(),
             expr_info: Vec::new(),
+        }
+    }
+}
 #[derive(Debug, Clone, Copy)]
 pub struct ExpressionInfo {
     pub type_id: TypeId,
     pub variant: ExpressionInfoVariant,
+}
 impl ExpressionInfo {
     pub(crate) fn new_invalid() -> Self {
         return Self{
             type_id: TypeId::new_invalid(),
             variant: ExpressionInfoVariant::Generic,
+        }
+    }
+}
 #[derive(Debug, Clone, Copy)]
 pub enum ExpressionInfoVariant {
     Generic,
     Procedure(TypeId, u32), // procedure TypeID and its monomorph index
     Select(i32), // index
+}
 impl ExpressionInfoVariant {
     pub(crate) fn as_select(&self) -> i32 {
         match self {
             ExpressionInfoVariant::Select(v) => *v,
             _ => unreachable!(),
+        }
+    }
     pub(crate) fn as_procedure(&self) -> (TypeId, u32) {
         match self {
             ExpressionInfoVariant::Procedure(type_id, monomorph_index) => (*type_id, *monomorph_index),
             _ => unreachable!(),
+        }
+    }
+}
 #[derive(Debug)]
 pub enum ProcedureSource {
     FuncUserDefined,
     CompUserDefined,
     // Builtin functions, available to user
     FuncGet,
     FuncPut,
     FuncFires,
     FuncCreate,
     FuncLength,
     FuncAssert,
     FuncPrint,
     // Buitlin functions, not available to user
     FuncSelectStart,
     FuncSelectRegisterCasePort,
     FuncSelectWait,
     // Builtin components, available to user
     CompRandomU32, // TODO: Remove, temporary thing
     CompTcpClient,
+}
 impl ProcedureSource {
     pub(crate) fn is_builtin(&self) -> bool {
         match self {
             ProcedureSource::FuncUserDefined | ProcedureSource::CompUserDefined => false,
             _ => true,
+        }
+    }
+}
 /// Generic storage for functions, primitive components and composite
 /// components.
 // Note that we will have function definitions for builtin functions as well. In
 // that case the span, the identifier span and the body are all invalid.
 #[derive(Debug)]
 pub struct ProcedureDefinition {
     pub this: ProcedureDefinitionId,
     pub defined_in: RootId,
     // Symbol scanning
     pub kind: ProcedureKind,
     pub identifier: Identifier,
     pub poly_vars: Vec<Identifier>,
     // Parser
     pub source: ProcedureSource,
     pub return_type: Option<ParserType>, // present on functions, not components
     pub parameters: Vec<VariableId>,
     pub scope: ScopeId,
     pub body: BlockStatementId,
     // Monomorphization of typed procedures
     pub monomorphs: Vec<ProcedureDefinitionMonomorph>,
+}
 impl ProcedureDefinition {
     pub(crate) fn new_empty(
         this: ProcedureDefinitionId, defined_in: RootId,
         kind: ProcedureKind, identifier: Identifier, poly_vars: Vec<Identifier>
     ) -> Self {
         Self {
             this, defined_in,
             kind, identifier, poly_vars,
             source: ProcedureSource::FuncUserDefined,
             return_type: None,
             parameters: Vec::new(),
             scope: ScopeId::new_invalid(),
             body: BlockStatementId::new_invalid(),
             monomorphs: Vec::new(),
+        }
+    }
+}
 #[derive(Debug, Clone)]
 pub enum Statement {
     Block(BlockStatement),
     EndBlock(EndBlockStatement),
     Local(LocalStatement),
     Labeled(LabeledStatement),
     If(IfStatement),
     EndIf(EndIfStatement),
     While(WhileStatement),
     EndWhile(EndWhileStatement),
     Break(BreakStatement),
     Continue(ContinueStatement),
     Synchronous(SynchronousStatement),
     EndSynchronous(EndSynchronousStatement),
     Fork(ForkStatement),
     EndFork(EndForkStatement),
     Select(SelectStatement),
     EndSelect(EndSelectStatement),
     Return(ReturnStatement),
     Goto(GotoStatement),
     New(NewStatement),
     Expression(ExpressionStatement),
+}
 impl Statement {
     pub fn as_new(&self) -> &NewStatement {
         match self {
             Statement::New(result) => result,
             _ => panic!("Unable to cast `Statement` to `NewStatement`"),
+        }
+    }
     pub fn span(&self) -> InputSpan {
         match self {
             Statement::Block(v) => v.span,
             Statement::Local(v) => v.span(),
             Statement::Labeled(v) => v.label.span,
             Statement::If(v) => v.span,
             Statement::While(v) => v.span,
             Statement::Break(v) => v.span,
             Statement::Continue(v) => v.span,
             Statement::Synchronous(v) => v.span,
             Statement::Fork(v) => v.span,
             Statement::Select(v) => v.span,
             Statement::Return(v) => v.span,
             Statement::Goto(v) => v.span,
             Statement::New(v) => v.span,
             Statement::Expression(v) => v.span,
             Statement::EndBlock(_)
             | Statement::EndIf(_)
             | Statement::EndWhile(_)
             | Statement::EndSynchronous(_)
             | Statement::EndFork(_)
             | Statement::EndSelect(_) => unreachable!(),
+        }
+    }
     pub fn link_next(&mut self, next: StatementId) {
         match self {
             Statement::Block(stmt) => stmt.next = next,
             Statement::EndBlock(stmt) => stmt.next = next,
             Statement::Local(stmt) => match stmt {
                 LocalStatement::Channel(stmt) => stmt.next = next,
                 LocalStatement::Memory(stmt) => stmt.next = next,
             },
             Statement::EndIf(stmt) => stmt.next = next,
             Statement::EndWhile(stmt) => stmt.next = next,
             Statement::EndSynchronous(stmt) => stmt.next = next,
             Statement::EndFork(stmt) => stmt.next = next,
             Statement::EndSelect(stmt) => stmt.next = next,
             Statement::New(stmt) => stmt.next = next,
             Statement::Expression(stmt) => stmt.next = next,
             Statement::Return(_)
             | Statement::Break(_)
             | Statement::Continue(_)
             | Statement::Synchronous(_)
             | Statement::Fork(_)
             | Statement::Select(_)
             | Statement::Goto(_)
             | Statement::While(_)
             | Statement::Labeled(_)
             | Statement::If(_) => unreachable!(),
+        }
+    }
+}
 #[derive(Debug, Clone)]
 pub struct BlockStatement {
     pub this: BlockStatementId,
     // Phase 1: parser
     pub span: InputSpan, // of the complete block
     pub statements: Vec<StatementId>,
     pub end_block: EndBlockStatementId,
     // Phase 2: linker
     pub scope: ScopeId,
     pub next: StatementId,
+}
 #[derive(Debug, Clone)]
 pub struct EndBlockStatement {
     pub this: EndBlockStatementId,
     // Parser
     pub start_block: BlockStatementId,
     // Validation/Linking
     pub next: StatementId,
+}
 #[derive(Debug, Clone)]
 pub enum LocalStatement {
     Memory(MemoryStatement),
     Channel(ChannelStatement),
+}
 impl LocalStatement {
     pub fn this(&self) -> LocalStatementId {
         match self {
             LocalStatement::Memory(stmt) => stmt.this.upcast(),
             LocalStatement::Channel(stmt) => stmt.this.upcast(),
+        }
+    }
     pub fn span(&self) -> InputSpan {
         match self {
             LocalStatement::Channel(v) => v.span,
             LocalStatement::Memory(v) => v.span,
+        }
+    }
+}
 #[derive(Debug, Clone)]
 pub struct MemoryStatement {
     pub this: MemoryStatementId,
     // Phase 1: parser
     pub span: InputSpan,
     pub variable: VariableId,
     pub initial_expr: AssignmentExpressionId,
     // Phase 2: linker
     pub next: StatementId,
+}
 /// ChannelStatement is the declaration of an input and output port associated
@@ @@ -1663,325 +1664,326 @@ pub struct AssignmentExpression { @@
     pub operation: AssignmentOperator,
     pub right: ExpressionId,
     // Validator/Linker
     pub parent: ExpressionParent,
     // Typing
     pub type_index: i32,
+}
 #[derive(Debug, Clone)]
 pub struct BindingExpression {
     pub this: BindingExpressionId,
     // Parsing
     pub operator_span: InputSpan,
     pub full_span: InputSpan,
     pub bound_to: ExpressionId,
     pub bound_from: ExpressionId,
     // Validator/Linker
     pub parent: ExpressionParent,
     // Typing
     pub type_index: i32,
+}
 #[derive(Debug, Clone)]
 pub struct ConditionalExpression {
     pub this: ConditionalExpressionId,
     // Parsing
     pub operator_span: InputSpan,
     pub full_span: InputSpan,
     pub test: ExpressionId,
     pub true_expression: ExpressionId,
     pub false_expression: ExpressionId,
     // Validator/Linking
     pub parent: ExpressionParent,
     // Typing
     pub type_index: i32,
+}
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum BinaryOperator {
     Concatenate,
     LogicalOr,
     LogicalAnd,
     BitwiseOr,
     BitwiseXor,
     BitwiseAnd,
     Equality,
     Inequality,
     LessThan,
     GreaterThan,
     LessThanEqual,
     GreaterThanEqual,
     ShiftLeft,
     ShiftRight,
     Add,
     Subtract,
     Multiply,
     Divide,
     Remainder,
+}
 #[derive(Debug, Clone)]
 pub struct BinaryExpression {
     pub this: BinaryExpressionId,
     // Parsing
     pub operator_span: InputSpan,
     pub full_span: InputSpan,
     pub left: ExpressionId,
     pub operation: BinaryOperator,
     pub right: ExpressionId,
     // Validator/Linker
     pub parent: ExpressionParent,
     // Typing
     pub type_index: i32,
+}
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum UnaryOperator {
     Positive,
     Negative,
     BitwiseNot,
     LogicalNot,
+}
 #[derive(Debug, Clone)]
 pub struct UnaryExpression {
     pub this: UnaryExpressionId,
     // Parsing
     pub operator_span: InputSpan,
     pub full_span: InputSpan,
     pub operation: UnaryOperator,
     pub expression: ExpressionId,
     // Validator/Linker
     pub parent: ExpressionParent,
     // Typing
     pub type_index: i32,
+}
 #[derive(Debug, Clone)]
 pub struct IndexingExpression {
     pub this: IndexingExpressionId,
     // Parsing
     pub operator_span: InputSpan,
     pub full_span: InputSpan,
     pub subject: ExpressionId,
     pub index: ExpressionId,
     // Validator/Linker
     pub parent: ExpressionParent,
     // Typing
     pub type_index: i32,
+}
 #[derive(Debug, Clone)]
 pub struct SlicingExpression {
     pub this: SlicingExpressionId,
     // Parsing
     pub slicing_span: InputSpan, // from '[' to ']'
     pub full_span: InputSpan, // includes subject
     pub subject: ExpressionId,
     pub from_index: ExpressionId,
     pub to_index: ExpressionId,
     // Validator/Linker
     pub parent: ExpressionParent,
     // Typing
     pub type_index: i32,
+}
 #[derive(Debug, Clone)]
 pub enum SelectKind {
     StructField(Identifier),
     TupleMember(u64), // u64 is overkill, but space is taken up by `StructField` variant anyway
+}
 #[derive(Debug, Clone)]
 pub struct SelectExpression {
     pub this: SelectExpressionId,
     // Parsing
     pub operator_span: InputSpan, // of the '.'
     pub full_span: InputSpan, // includes subject and field
     pub subject: ExpressionId,
     pub kind: SelectKind,
     // Validator/Linker
     pub parent: ExpressionParent,
     // Typing
     pub type_index: i32,
+}
 #[derive(Debug, Clone)]
 pub struct CastExpression {
     pub this: CastExpressionId,
     // Parsing
     pub cast_span: InputSpan, // of the "cast" keyword,
     pub full_span: InputSpan, // includes the cast subject
     pub to_type: ParserType,
     pub subject: ExpressionId,
     // Validator/linker
     pub parent: ExpressionParent,
     // Typing
     pub type_index: i32,
+}
 #[derive(Debug, Clone)]
 pub struct CallExpression {
     pub this: CallExpressionId,
     // Parsing
     pub func_span: InputSpan, // of the function name
     pub full_span: InputSpan, // includes the arguments and parentheses
     pub parser_type: ParserType, // of the function call, not the return type
     pub method: Method,
     pub arguments: Vec<ExpressionId>,
     pub procedure: ProcedureDefinitionId,
     // Validator/Linker
     pub parent: ExpressionParent,
     // Typing
     pub type_index: i32,
+}
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum Method {
     // Builtin function, accessible by programmer
     Get,
     Put,
     Fires,
     Create,
     Length,
     Assert,
     Print,
     // Builtin function, not accessible by programmer
     SelectStart, // SelectStart(total_num_cases, total_num_ports)
     SelectRegisterCasePort, // SelectRegisterCasePort(case_index, port_index, port_id)
     SelectWait, // SelectWait() -> u32
     // Builtin component,
     ComponentRandomU32,
     ComponentTcpClient,
     // User-defined
     UserFunction,
     UserComponent,
+}
 impl Method {
     pub(crate) fn is_public_builtin(&self) -> bool {
         use Method::*;
         match self {
             Get | Put | Fires | Create | Length | Assert | Print => true,
             ComponentRandomU32 => true,
+            ComponentRandomU32 | ComponentTcpClient => true,
             _ => false,
+        }
+    }
     pub(crate) fn is_user_defined(&self) -> bool {
         use Method::*;
         match self {
             UserFunction | UserComponent => true,
             _ => false,
+        }
+    }
+}
 #[derive(Debug, Clone)]
 pub struct LiteralExpression {
     pub this: LiteralExpressionId,
     // Parsing
     pub span: InputSpan,
     pub value: Literal,
     // Validator/Linker
     pub parent: ExpressionParent,
     // Typing
     pub type_index: i32,
+}
 #[derive(Debug, Clone)]
 pub enum Literal {
     Null, // message
     True,
     False,
     Character(char),
     String(StringRef<'static>),
     Integer(LiteralInteger),
     Struct(LiteralStruct),
     Enum(LiteralEnum),
     Union(LiteralUnion),
     Array(Vec<ExpressionId>),
     Tuple(Vec<ExpressionId>),
+}
 impl Literal {
     pub(crate) fn as_struct(&self) -> &LiteralStruct {
         if let Literal::Struct(literal) = self{
             literal
         } else {
             unreachable!("Attempted to obtain {:?} as Literal::Struct", self)
+        }
+    }
     pub(crate) fn as_enum(&self) -> &LiteralEnum {
         if let Literal::Enum(literal) = self {
             literal
         } else {
             unreachable!("Attempted to obtain {:?} as Literal::Enum", self)
+        }
+    }
     pub(crate) fn as_union(&self) -> &LiteralUnion {
         if let Literal::Union(literal) = self {
             literal
         } else {
             unreachable!("Attempted to obtain {:?} as Literal::Union", self)
+        }
+    }
+}
 #[derive(Debug, Clone)]
 pub struct LiteralInteger {
     pub(crate) unsigned_value: u64,
     pub(crate) negated: bool, // for constant expression evaluation, TODO: @Int
+}
 #[derive(Debug, Clone)]
 pub struct LiteralStructField {
     // Phase 1: parser
     pub(crate) identifier: Identifier,
     pub(crate) value: ExpressionId,
     // Phase 2: linker
     pub(crate) field_idx: usize, // in struct definition
+}
 #[derive(Debug, Clone)]
 pub struct LiteralStruct {
     // Phase 1: parser
     pub(crate) parser_type: ParserType,
     pub(crate) fields: Vec<LiteralStructField>,
     pub(crate) definition: DefinitionId,
+}
 #[derive(Debug, Clone)]
 pub struct LiteralEnum {
     // Phase 1: parser
     pub(crate) parser_type: ParserType,
     pub(crate) variant: Identifier,
     pub(crate) definition: DefinitionId,
     // Phase 2: linker
     pub(crate) variant_idx: usize, // as present in the type table
+}
 #[derive(Debug, Clone)]
 pub struct LiteralUnion {
     // Phase 1: parser
     pub(crate) parser_type: ParserType,
     pub(crate) variant: Identifier,
     pub(crate) values: Vec<ExpressionId>,
     pub(crate) definition: DefinitionId,
     // Phase 2: linker
     pub(crate) variant_idx: usize, // as present in type table
+}
 #[derive(Debug, Clone)]
 pub struct VariableExpression {
     pub this: VariableExpressionId,
     // Parsing
     pub identifier: Identifier,
     // Validator/Linker
     pub declaration: Option<VariableId>,
     pub used_as_binding_target: bool,
     pub parent: ExpressionParent,
     // Typing
     pub type_index: i32,
+}
@@ \ No newline at end of file @@

src/protocol/eval/executor.rs

➞

Show inline comments

@@ @@ -566,385 +566,385 @@ impl Prompt { @@
                                     );
                                     Value::Array(heap_pos)
+                                }
                                 Literal::Tuple(lit_value) => {
                                     let heap_pos = transfer_expression_values_front_into_heap(
                                         cur_frame, &mut self.store, lit_value.len()
                                     );
                                     Value::Tuple(heap_pos)
+                                }
                             };
                             cur_frame.expr_values.push_back(value);
                         },
                         Expression::Cast(expr) => {
                             let mono_data = &heap[cur_frame.definition].monomorphs[cur_frame.monomorph_index];
                             let type_id = mono_data.expr_info[expr.type_index as usize].type_id;
                             let concrete_type = &types.get_monomorph(type_id).concrete_type;
                             // Typechecking reduced this to two cases: either we
                             // have casting noop (same types), or we're casting
                             // between integer/bool/char types.
                             let subject = cur_frame.expr_values.pop_back().unwrap();
                             match apply_casting(&mut self.store, concrete_type, &subject) {
                                 Ok(value) => cur_frame.expr_values.push_back(value),
                                 Err(msg) => {
                                     return Err(EvalError::new_error_at_expr(self, modules, heap, expr.this.upcast(), msg));
+                                }
+                            }
                             self.store.drop_value(subject.get_heap_pos());
+                        }
                         Expression::Call(expr) => {
                             // If we're dealing with a builtin we don't do any
                             // fancy shenanigans at all, just push the result.
                             match expr.method {
                                 Method::Get => {
                                     let value = cur_frame.expr_values.pop_front().unwrap();
                                     let value = self.store.maybe_read_ref(&value).clone();
                                     let port_id = if let Value::Input(port_id) = value {
                                         port_id
                                     } else {
                                         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 = 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 => {
                                     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();
                                     return Ok(EvalContinuation::SelectStart(num_cases, num_ports));
                                 },
                                 Method::SelectRegisterCasePort => {
                                     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();
                                     return Ok(EvalContinuation::SelectRegisterPort(case_index, port_index, port_value));
                                 },
                                 Method::SelectWait => {
                                     match ctx.performed_select_wait() {
                                         Some(select_index) => {
                                             cur_frame.expr_values.push_back(Value::UInt32(select_index));
                                         },
                                         None => {
                                             cur_frame.expr_stack.push_back(ExprInstruction::EvalExpr(expr.this.upcast()));
                                             return Ok(EvalContinuation::SelectWait)
                                         },
+                                    }
                                 },
                                 Method::ComponentRandomU32 => {
+                                Method::ComponentRandomU32 | Method::ComponentTcpClient => {
                                     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::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))
                             };
                             cur_frame.expr_values.push_back(ref_value);
+                        }
+                    }
+                }
+            }
+        }
         debug_log!("Frame [{:?}] at {:?}", cur_frame.definition, cur_frame.position);
         if debug_enabled!() {
             debug_log!("Expression value stack (size = {}):", cur_frame.expr_values.len());
             for (_stack_idx, _stack_val) in cur_frame.expr_values.iter().enumerate() {
                 debug_log!("  [{:03}] {:?}", _stack_idx, _stack_val);
+            }
             debug_log!("Stack (size = {}):", self.store.stack.len());
             for (_stack_idx, _stack_val) in self.store.stack.iter().enumerate() {
                 debug_log!("  [{:03}] {:?}", _stack_idx, _stack_val);
+            }
             debug_log!("Heap:");
             for (_heap_idx, _heap_region) in self.store.heap_regions.iter().enumerate() {
                 let _is_free = self.store.free_regions.iter().any(|idx| *idx as usize == _heap_idx);
                 debug_log!("  [{:03}] in_use: {}, len: {}, vals: {:?}", _heap_idx, !_is_free, _heap_region.values.len(), &_heap_region.values);
+            }
+        }
         // No (more) expressions to evaluate. So evaluate statement (that may
         // depend on the result on the last evaluated expression(s))
         let stmt = &heap[cur_frame.position];
         let return_value = match stmt {
             Statement::Block(stmt) => {
                 debug_assert!(stmt.statements.is_empty() || stmt.next == stmt.statements[0]);
                 cur_frame.position = stmt.next;
                 Ok(EvalContinuation::Stepping)
             },
             Statement::EndBlock(stmt) => {
                 let block = &heap[stmt.start_block];
                 let scope = &heap[block.scope];
                 self.store.clear_stack(scope.first_unique_id_in_scope as usize);
                 cur_frame.position = stmt.next;
                 Ok(EvalContinuation::Stepping)
             },
             Statement::Local(stmt) => {
                 match stmt {
                     LocalStatement::Memory(stmt) => {
                         dbg_code!({
                             let variable = &heap[stmt.variable];
                             debug_assert!(match self.store.read_ref(ValueId::Stack(variable.unique_id_in_scope as u32)) {
                                 Value::Unassigned => false,
                                 _ => true,
                             });
                         });
                         cur_frame.position = stmt.next;
                         Ok(EvalContinuation::Stepping)
                     },
                     LocalStatement::Channel(stmt) => {
                         // Need to create a new channel by requesting it from
                         // the runtime.
                         match ctx.created_channel() {
                             None => {
                                 // No channel is pending. So request one
                                     Ok(EvalContinuation::NewChannel)
                             },
                             Some((put_port, get_port)) => {
                                 self.store.write(ValueId::Stack(heap[stmt.from].unique_id_in_scope as u32), put_port);
                                 self.store.write(ValueId::Stack(heap[stmt.to].unique_id_in_scope as u32), get_port);
                                 cur_frame.position = stmt.next;
                                 Ok(EvalContinuation::Stepping)
+                            }
+                        }
+                    }
+                }
             },
             Statement::Labeled(stmt) => {
                 cur_frame.position = stmt.body;
                 Ok(EvalContinuation::Stepping)
             },
             Statement::If(stmt) => {
                 debug_assert_eq!(cur_frame.expr_values.len(), 1, "expected one expr value for if statement");
                 let test_value = cur_frame.expr_values.pop_back().unwrap();
                 let test_value = self.store.maybe_read_ref(&test_value).as_bool();
                 if test_value {
                     cur_frame.position = stmt.true_case.body;
                 } else if let Some(false_body) = stmt.false_case {
                     cur_frame.position = false_body.body;
                 } else {
                     // Not true, and no false body
                     cur_frame.position = stmt.end_if.upcast();
+                }
                 Ok(EvalContinuation::Stepping)
             },
             Statement::EndIf(stmt) => {
                 cur_frame.position = stmt.next;
                 let if_stmt = &heap[stmt.start_if];
                 debug_assert_eq!(
                     heap[if_stmt.true_case.scope].first_unique_id_in_scope,
                     heap[if_stmt.false_case.unwrap_or(if_stmt.true_case).scope].first_unique_id_in_scope,
                 );
                 let scope = &heap[if_stmt.true_case.scope];
                 self.store.clear_stack(scope.first_unique_id_in_scope as usize);
                 Ok(EvalContinuation::Stepping)
             },
             Statement::While(stmt) => {
                 debug_assert_eq!(cur_frame.expr_values.len(), 1, "expected one expr value for while statement");
                 let test_value = cur_frame.expr_values.pop_back().unwrap();
                 let test_value = self.store.maybe_read_ref(&test_value).as_bool();
                 if test_value {
                     cur_frame.position = stmt.body;
                 } else {
                     cur_frame.position = stmt.end_while.upcast();
+                }
                 Ok(EvalContinuation::Stepping)
             },
             Statement::EndWhile(stmt) => {
                 cur_frame.position = stmt.next;
                 let start_while = &heap[stmt.start_while];
                 let scope = &heap[start_while.scope];
                 self.store.clear_stack(scope.first_unique_id_in_scope as usize);
                 Ok(EvalContinuation::Stepping)
             },
             Statement::Break(stmt) => {
                 cur_frame.position = stmt.target.upcast();
                 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];

src/protocol/parser/mod.rs

➞

Show inline comments

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

src/protocol/parser/pass_definitions.rs

➞

Show inline comments

@@ @@ -188,384 +188,385 @@ impl PassDefinitions { @@
         Ok(())
+    }
     fn visit_union_definition(
         &mut self, module: &Module, iter: &mut TokenIter, ctx: &mut PassCtx
     ) -> Result<(), ParseError> {
         consume_exact_ident(&module.source, iter, KW_UNION)?;
         let (ident_text, _) = consume_ident(&module.source, iter)?;
         // Retrieve preallocated DefinitionId
         let module_scope = SymbolScope::Module(module.root_id);
         let definition_id = ctx.symbols.get_symbol_by_name_defined_in_scope(module_scope, ident_text)
             .unwrap().variant.as_definition().definition_id;
         self.cur_definition = definition_id;
         // Parse union definition
         consume_polymorphic_vars_spilled(&module.source, iter, ctx)?;
         let mut variants_section = self.union_variants.start_section();
         consume_comma_separated(
             TokenKind::OpenCurly, TokenKind::CloseCurly, &module.source, iter, ctx,
             |source, iter, ctx| {
                 let identifier = consume_ident_interned(source, iter, ctx)?;
                 let mut close_pos = identifier.span.end;
                 let mut types_section = self.parser_types.start_section();
                 let has_embedded = maybe_consume_comma_separated(
                     TokenKind::OpenParen, TokenKind::CloseParen, source, iter, ctx,
                     |source, iter, ctx| {
                         let poly_vars = ctx.heap[definition_id].poly_vars();
                         self.type_parser.consume_parser_type(
                             iter, &ctx.heap, source, &ctx.symbols, poly_vars, definition_id,
                             module_scope, false, false, None
+                        )
                     },
                     &mut types_section, "an embedded type", Some(&mut close_pos)
                 )?;
                 let value = if has_embedded {
                     types_section.into_vec()
                 } else {
                     types_section.forget();
                     Vec::new()
                 };
                 Ok(UnionVariantDefinition{
                     span: InputSpan::from_positions(identifier.span.begin, close_pos),
                     identifier,
                     value
                 })
             },
             &mut variants_section, "a union variant", "a list of union variants", None
         )?;
         // Transfer to AST
         let union_def = ctx.heap[definition_id].as_union_mut();
         union_def.variants = variants_section.into_vec();
         Ok(())
+    }
     fn visit_function_definition(
         &mut self, module: &Module, iter: &mut TokenIter, ctx: &mut PassCtx
     ) -> Result<(), ParseError> {
         // Retrieve function name
         consume_exact_ident(&module.source, iter, KW_FUNCTION)?;
         let (ident_text, _) = consume_ident(&module.source, iter)?;
         // Retrieve preallocated DefinitionId
         let module_scope = SymbolScope::Module(module.root_id);
         let definition_id = ctx.symbols.get_symbol_by_name_defined_in_scope(module_scope, ident_text)
             .unwrap().variant.as_definition().definition_id;
         self.cur_definition = definition_id;
         let allow_compiler_types = module.is_compiler_file;
         consume_polymorphic_vars_spilled(&module.source, iter, ctx)?;
         // Parse function's argument list
         let mut parameter_section = self.variables.start_section();
         consume_parameter_list(
             &mut self.type_parser, &module.source, iter, ctx, &mut parameter_section,
             module_scope, definition_id, allow_compiler_types
         )?;
         let parameters = parameter_section.into_vec();
         // Consume return types
         consume_token(&module.source, iter, TokenKind::ArrowRight)?;
         let poly_vars = ctx.heap[definition_id].poly_vars();
         let parser_type = self.type_parser.consume_parser_type(
             iter, &ctx.heap, &module.source, &ctx.symbols, poly_vars, definition_id,
             module_scope, false, allow_compiler_types, None
         )?;
         // Consume body
         let (body_id, source) = self.consume_procedure_body(module, iter, ctx, definition_id, ProcedureKind::Function)?;
         let scope_id = ctx.heap.alloc_scope(|this| Scope::new(this, ScopeAssociation::Definition(definition_id)));
         // Assign everything in the preallocated AST node
         let function = ctx.heap[definition_id].as_procedure_mut();
         function.source = source;
         function.return_type = Some(parser_type);
         function.parameters = parameters;
         function.scope = scope_id;
         function.body = body_id;
         Ok(())
+    }
     fn visit_component_definition(
         &mut self, module: &Module, iter: &mut TokenIter, ctx: &mut PassCtx
     ) -> Result<(), ParseError> {
         // Consume component variant and name
         let (_variant_text, _) = consume_any_ident(&module.source, iter)?;
         debug_assert!(_variant_text == KW_PRIMITIVE || _variant_text == KW_COMPOSITE);
         let (ident_text, _) = consume_ident(&module.source, iter)?;
         // Retrieve preallocated definition
         let module_scope = SymbolScope::Module(module.root_id);
         let definition_id = ctx.symbols.get_symbol_by_name_defined_in_scope(module_scope, ident_text)
             .unwrap().variant.as_definition().definition_id;
         self.cur_definition = definition_id;
         let allow_compiler_types = module.is_compiler_file;
         consume_polymorphic_vars_spilled(&module.source, iter, ctx)?;
         // Parse component's argument list
         let mut parameter_section = self.variables.start_section();
         consume_parameter_list(
             &mut self.type_parser, &module.source, iter, ctx, &mut parameter_section,
             module_scope, definition_id, allow_compiler_types
         )?;
         let parameters = parameter_section.into_vec();
         // Consume body
         let procedure_kind = ctx.heap[definition_id].as_procedure().kind;
         let (body_id, source) = self.consume_procedure_body(module, iter, ctx, definition_id, procedure_kind)?;
         let scope_id = ctx.heap.alloc_scope(|this| Scope::new(this, ScopeAssociation::Definition(definition_id)));
         // Assign everything in the AST node
         let component = ctx.heap[definition_id].as_procedure_mut();
         debug_assert!(component.return_type.is_none());
         component.source = source;
         component.parameters = parameters;
         component.scope = scope_id;
         component.body = body_id;
         Ok(())
+    }
     /// Consumes a procedure's body: either a user-defined procedure, which we
     /// parse as normal, or a builtin function, where we'll make sure we expect
     /// the particular builtin.
     ///
     /// We expect that the procedure's name is already stored in the
     /// preallocated AST node.
     fn consume_procedure_body(
         &mut self, module: &Module, iter: &mut TokenIter, ctx: &mut PassCtx, definition_id: DefinitionId, kind: ProcedureKind
     ) -> Result<(BlockStatementId, ProcedureSource), ParseError> {
         if iter.next() == Some(TokenKind::OpenCurly) && iter.peek() == Some(TokenKind::Pragma) {
             // Consume the placeholder "{ #builtin }" tokens
             iter.consume(); // opening curly brace
             let (pragma, pragma_span) = consume_pragma(&module.source, iter)?;
             if pragma != b"#builtin" {
                 return Err(ParseError::new_error_str_at_span(
                     &module.source, pragma_span,
                     "expected a '#builtin' pragma, or a function body"
                 ));
+            }
             if iter.next() != Some(TokenKind::CloseCurly) {
                 // Just to keep the compiler writers in line ;)
                 panic!("compiler error: when using the #builtin pragma, wrap it in curly braces");
+            }
             iter.consume();
             // Retrieve module and procedure name
             assert!(module.name.is_some(), "compiler error: builtin procedure body in unnamed module");
             let (_, module_name) = module.name.as_ref().unwrap();
             let module_name = module_name.as_str();
             let definition = ctx.heap[definition_id].as_procedure();
             let procedure_name = definition.identifier.value.as_str();
             let source = match (module_name, procedure_name) {
                 ("std.global", "get") => ProcedureSource::FuncGet,
                 ("std.global", "put") => ProcedureSource::FuncPut,
                 ("std.global", "fires") => ProcedureSource::FuncFires,
                 ("std.global", "create") => ProcedureSource::FuncCreate,
                 ("std.global", "length") => ProcedureSource::FuncLength,
                 ("std.global", "assert") => ProcedureSource::FuncAssert,
                 ("std.global", "print") => ProcedureSource::FuncPrint,
                 ("std.random", "random_u32") => ProcedureSource::CompRandomU32,
                 ("std.internet", "tcp_client") => ProcedureSource::CompTcpClient,
                 _ => panic!(
                     "compiler error: unknown builtin procedure '{}' in module '{}'",
                     procedure_name, module_name
                 ),
             };
             return Ok((BlockStatementId::new_invalid(), source));
         } else {
             let body_id = self.consume_block_statement(module, iter, ctx)?;
             let source = match kind {
                 ProcedureKind::Function =>
                     ProcedureSource::FuncUserDefined,
                 ProcedureKind::Primitive | ProcedureKind::Composite =>
                     ProcedureSource::CompUserDefined,
             };
             return Ok((body_id, source))
+        }
+    }
     /// Consumes a statement and returns a boolean indicating whether it was a
     /// block or not.
     fn consume_statement(&mut self, module: &Module, iter: &mut TokenIter, ctx: &mut PassCtx) -> Result<StatementId, ParseError> {
         let next = iter.next().expect("consume_statement has a next token");
         if next == TokenKind::OpenCurly {
             let id = self.consume_block_statement(module, iter, ctx)?;
             return Ok(id.upcast());
         } else if next == TokenKind::Ident {
             let ident = peek_ident(&module.source, iter).unwrap();
             if ident == KW_STMT_IF {
                 // Consume if statement and place end-if statement directly
                 // after it.
                 let id = self.consume_if_statement(module, iter, ctx)?;
                 return Ok(id.upcast());
             } else if ident == KW_STMT_WHILE {
                 let id = self.consume_while_statement(module, iter, ctx)?;
                 return Ok(id.upcast());
             } else if ident == KW_STMT_BREAK {
                 let id = self.consume_break_statement(module, iter, ctx)?;
                 return Ok(id.upcast());
             } else if ident == KW_STMT_CONTINUE {
                 let id = self.consume_continue_statement(module, iter, ctx)?;
                 return Ok(id.upcast());
             } else if ident == KW_STMT_SYNC {
                 let id = self.consume_synchronous_statement(module, iter, ctx)?;
                 return Ok(id.upcast());
             } else if ident == KW_STMT_FORK {
                 let id = self.consume_fork_statement(module, iter, ctx)?;
                 let end_fork = ctx.heap.alloc_end_fork_statement(|this| EndForkStatement {
                     this,
                     start_fork: id,
                     next: StatementId::new_invalid(),
                 });
                 let fork_stmt = &mut ctx.heap[id];
                 fork_stmt.end_fork = end_fork;
                 return Ok(id.upcast());
             } else if ident == KW_STMT_SELECT {
                 let id = self.consume_select_statement(module, iter, ctx)?;
                 return Ok(id.upcast());
             } else if ident == KW_STMT_RETURN {
                 let id = self.consume_return_statement(module, iter, ctx)?;
                 return Ok(id.upcast());
             } else if ident == KW_STMT_GOTO {
                 let id = self.consume_goto_statement(module, iter, ctx)?;
                 return Ok(id.upcast());
             } else if ident == KW_STMT_NEW {
                 let id = self.consume_new_statement(module, iter, ctx)?;
                 return Ok(id.upcast());
             } else if ident == KW_STMT_CHANNEL {
                 let id = self.consume_channel_statement(module, iter, ctx)?;
                 return Ok(id.upcast().upcast());
             } else if iter.peek() == Some(TokenKind::Colon) {
                 let id = self.consume_labeled_statement(module, iter, ctx)?;
                 return Ok(id.upcast());
             } else {
                 // Two fallback possibilities: the first one is a memory
                 // declaration, the other one is to parse it as a normal
                 // expression. This is a bit ugly.
                 if let Some(memory_stmt_id) = self.maybe_consume_memory_statement_without_semicolon(module, iter, ctx)? {
                     consume_token(&module.source, iter, TokenKind::SemiColon)?;
                     return Ok(memory_stmt_id.upcast().upcast());
                 } else {
                     let id = self.consume_expression_statement(module, iter, ctx)?;
                     return Ok(id.upcast());
+                }
+            }
         } else if next == TokenKind::OpenParen {
             // Same as above: memory statement or normal expression
             if let Some(memory_stmt_id) = self.maybe_consume_memory_statement_without_semicolon(module, iter, ctx)? {
                 consume_token(&module.source, iter, TokenKind::SemiColon)?;
                 return Ok(memory_stmt_id.upcast().upcast());
             } else {
                 let id = self.consume_expression_statement(module, iter, ctx)?;
                 return Ok(id.upcast());
+            }
         } else {
             let id = self.consume_expression_statement(module, iter, ctx)?;
             return Ok(id.upcast());
+        }
+    }
     fn consume_block_statement(
         &mut self, module: &Module, iter: &mut TokenIter, ctx: &mut PassCtx
     ) -> Result<BlockStatementId, ParseError> {
         let open_curly_span = consume_token(&module.source, iter, TokenKind::OpenCurly)?;
         let mut stmt_section = self.statements.start_section();
         let mut next = iter.next();
         while next != Some(TokenKind::CloseCurly) {
             if next.is_none() {
                 return Err(ParseError::new_error_str_at_pos(
                     &module.source, iter.last_valid_pos(), "expected a statement or '}'"
                 ));
+            }
             let stmt_id = self.consume_statement(module, iter, ctx)?;
             stmt_section.push(stmt_id);
             next = iter.next();
+        }
         let statements = stmt_section.into_vec();
         let mut block_span = consume_token(&module.source, iter, TokenKind::CloseCurly)?;
         block_span.begin = open_curly_span.begin;
         let block_id = ctx.heap.alloc_block_statement(|this| BlockStatement{
             this,
             span: block_span,
             statements,
             end_block: EndBlockStatementId::new_invalid(),
             scope: ScopeId::new_invalid(),
             next: StatementId::new_invalid(),
         });
         let scope_id = ctx.heap.alloc_scope(|this| Scope::new(this, ScopeAssociation::Block(block_id)));
         let end_block_id = ctx.heap.alloc_end_block_statement(|this| EndBlockStatement{
             this, start_block: block_id, next: StatementId::new_invalid()
         });
         let block_stmt = &mut ctx.heap[block_id];
         block_stmt.end_block = end_block_id;
         block_stmt.scope = scope_id;
         Ok(block_id)
+    }
     fn consume_if_statement(
         &mut self, module: &Module, iter: &mut TokenIter, ctx: &mut PassCtx
     ) -> Result<IfStatementId, ParseError> {
         let if_span = consume_exact_ident(&module.source, iter, KW_STMT_IF)?;
         consume_token(&module.source, iter, TokenKind::OpenParen)?;
         let test = self.consume_expression(module, iter, ctx)?;
         consume_token(&module.source, iter, TokenKind::CloseParen)?;
         // Consume bodies of if-statement
         let true_body = IfStatementCase{
             body: self.consume_statement(module, iter, ctx)?,
             scope: ScopeId::new_invalid(),
         };
         let false_body = if has_ident(&module.source, iter, KW_STMT_ELSE) {
             iter.consume();
             let false_body = IfStatementCase{
                 body: self.consume_statement(module, iter, ctx)?,
                 scope: ScopeId::new_invalid(),
             };
             Some(false_body)
         } else {
             None
         };
         // Construct AST elements
         let if_stmt_id = ctx.heap.alloc_if_statement(|this| IfStatement{
             this,
             span: if_span,
             test,
             true_case: true_body,
             false_case: false_body,
             end_if: EndIfStatementId::new_invalid(),
         });
         let end_if_stmt_id = ctx.heap.alloc_end_if_statement(|this| EndIfStatement{
             this,
             start_if: if_stmt_id,
             next: StatementId::new_invalid(),
         });
         let true_scope_id = ctx.heap.alloc_scope(|this| Scope::new(this, ScopeAssociation::If(if_stmt_id, true)));
         let false_scope_id = if false_body.is_some() {
             Some(ctx.heap.alloc_scope(|this| Scope::new(this, ScopeAssociation::If(if_stmt_id, false))))
         } else {
@@ @@ -1469,384 +1470,385 @@ impl PassDefinitions { @@
                 literal_id.upcast()
             } else {
                 // Start by consuming one expression, then check for a comma
                 let expr_id = self.consume_expression(module, iter, ctx)?;
                 if Some(TokenKind::Comma) == iter.next() && Some(TokenKind::CloseParen) != iter.peek() {
                     // Must be an N-tuple
                     iter.consume(); // the comma
                     let mut scoped_section = self.expressions.start_section();
                     scoped_section.push(expr_id);
                     let mut close_paren_pos = open_paren_pos;
                     consume_comma_separated_until(
                         TokenKind::CloseParen, &module.source, iter, ctx,
                         |_source, iter, ctx| self.consume_expression(module, iter, ctx),
                         &mut scoped_section, "an expression", Some(&mut close_paren_pos)
                     )?;
                     debug_assert!(scoped_section.len() > 1); // peeked token wasn't CloseParen, must be expression
                     let literal_id = ctx.heap.alloc_literal_expression(|this| LiteralExpression{
                         this,
                         span: InputSpan::from_positions(open_paren_pos, close_paren_pos),
                         value: Literal::Tuple(scoped_section.into_vec()),
                         parent: ExpressionParent::None,
                         type_index: -1,
                     });
                     literal_id.upcast()
                 } else {
                     // Assume we're dealing with a normal expression
                     consume_token(&module.source, iter, TokenKind::CloseParen)?;
                     expr_id
+                }
             };
             result
         } else if next == Some(TokenKind::OpenCurly) {
             // Array literal
             let (start_pos, mut end_pos) = iter.next_positions();
             let mut scoped_section = self.expressions.start_section();
             consume_comma_separated(
                 TokenKind::OpenCurly, TokenKind::CloseCurly, &module.source, iter, ctx,
                 |_source, iter, ctx| self.consume_expression(module, iter, ctx),
                 &mut scoped_section, "an expression", "a list of expressions", Some(&mut end_pos)
             )?;
             ctx.heap.alloc_literal_expression(|this| LiteralExpression{
                 this,
                 span: InputSpan::from_positions(start_pos, end_pos),
                 value: Literal::Array(scoped_section.into_vec()),
                 parent: ExpressionParent::None,
                 type_index: -1,
             }).upcast()
         } else if next == Some(TokenKind::Integer) {
             let (literal, span) = consume_integer_literal(&module.source, iter, &mut self.buffer)?;
             ctx.heap.alloc_literal_expression(|this| LiteralExpression{
                 this, span,
                 value: Literal::Integer(LiteralInteger{ unsigned_value: literal, negated: false }),
                 parent: ExpressionParent::None,
                 type_index: -1,
             }).upcast()
         } else if next == Some(TokenKind::String) {
             let span = consume_string_literal(&module.source, iter, &mut self.buffer)?;
             let interned = ctx.pool.intern(self.buffer.as_bytes());
             ctx.heap.alloc_literal_expression(|this| LiteralExpression{
                 this, span,
                 value: Literal::String(interned),
                 parent: ExpressionParent::None,
                 type_index: -1,
             }).upcast()
         } else if next == Some(TokenKind::Character) {
             let (character, span) = consume_character_literal(&module.source, iter)?;
             ctx.heap.alloc_literal_expression(|this| LiteralExpression{
                 this, span,
                 value: Literal::Character(character),
                 parent: ExpressionParent::None,
                 type_index: -1,
             }).upcast()
         } else if next == Some(TokenKind::Ident) {
             // May be a variable, a type instantiation or a function call. If we
             // have a single identifier that we cannot find in the type table
             // then we're going to assume that we're dealing with a variable.
             let ident_span = iter.next_span();
             let ident_text = module.source.section_at_span(ident_span);
             let symbol = ctx.symbols.get_symbol_by_name(SymbolScope::Module(module.root_id), ident_text);
             if symbol.is_some() {
                 // The first bit looked like a symbol, so we're going to follow
                 // that all the way through, assume we arrive at some kind of
                 // function call or type instantiation
                 use ParserTypeVariant as PTV;
                 let symbol_scope = SymbolScope::Definition(self.cur_definition);
                 let poly_vars = ctx.heap[self.cur_definition].poly_vars();
                 let parser_type = self.type_parser.consume_parser_type(
                     iter, &ctx.heap, &module.source, &ctx.symbols, poly_vars, self.cur_definition,
                     symbol_scope, true, false, None
                 )?;
                 debug_assert!(!parser_type.elements.is_empty());
                 match parser_type.elements[0].variant {
                     PTV::Definition(target_definition_id, _) => {
                         let definition = &ctx.heap[target_definition_id];
                         match definition {
                             Definition::Struct(_) => {
                                 // Struct literal
                                 let mut last_token = iter.last_valid_pos();
                                 let mut struct_fields = Vec::new();
                                 consume_comma_separated(
                                     TokenKind::OpenCurly, TokenKind::CloseCurly, &module.source, iter, ctx,
                                     |source, iter, ctx| {
                                         let identifier = consume_ident_interned(source, iter, ctx)?;
                                         consume_token(source, iter, TokenKind::Colon)?;
                                         let value = self.consume_expression(module, iter, ctx)?;
                                         Ok(LiteralStructField{ identifier, value, field_idx: 0 })
                                     },
                                     &mut struct_fields, "a struct field", "a list of struct fields", Some(&mut last_token)
                                 )?;
                                 ctx.heap.alloc_literal_expression(|this| LiteralExpression{
                                     this,
                                     span: InputSpan::from_positions(ident_span.begin, last_token),
                                     value: Literal::Struct(LiteralStruct{
                                         parser_type,
                                         fields: struct_fields,
                                         definition: target_definition_id,
                                     }),
                                     parent: ExpressionParent::None,
                                     type_index: -1,
                                 }).upcast()
                             },
                             Definition::Enum(_) => {
                                 // Enum literal: consume the variant
                                 consume_token(&module.source, iter, TokenKind::ColonColon)?;
                                 let variant = consume_ident_interned(&module.source, iter, ctx)?;
                                 ctx.heap.alloc_literal_expression(|this| LiteralExpression{
                                     this,
                                     span: InputSpan::from_positions(ident_span.begin, variant.span.end),
                                     value: Literal::Enum(LiteralEnum{
                                         parser_type,
                                         variant,
                                         definition: target_definition_id,
                                         variant_idx: 0
                                     }),
                                     parent: ExpressionParent::None,
                                     type_index: -1,
                                 }).upcast()
                             },
                             Definition::Union(_) => {
                                 // Union literal: consume the variant
                                 consume_token(&module.source, iter, TokenKind::ColonColon)?;
                                 let variant = consume_ident_interned(&module.source, iter, ctx)?;
                                 // Consume any possible embedded values
                                 let mut end_pos = variant.span.end;
                                 let values = if Some(TokenKind::OpenParen) == iter.next() {
                                     self.consume_expression_list(module, iter, ctx, Some(&mut end_pos))?
                                 } else {
                                     Vec::new()
                                 };
                                 ctx.heap.alloc_literal_expression(|this| LiteralExpression{
                                     this,
                                     span: InputSpan::from_positions(ident_span.begin, end_pos),
                                     value: Literal::Union(LiteralUnion{
                                         parser_type, variant, values,
                                         definition: target_definition_id,
                                         variant_idx: 0,
                                     }),
                                     parent: ExpressionParent::None,
                                     type_index: -1,
                                 }).upcast()
                             },
                             Definition::Procedure(proc_def) => {
                                 // Check whether it is a builtin function
                                 // TODO: Once we start generating bytecode this is unnecessary
                                 let procedure_id = proc_def.this;
                                 let method = match proc_def.source {
                                     ProcedureSource::FuncUserDefined => Method::UserFunction,
                                     ProcedureSource::CompUserDefined => Method::UserComponent,
                                     ProcedureSource::FuncGet => Method::Get,
                                     ProcedureSource::FuncPut => Method::Put,
                                     ProcedureSource::FuncFires => Method::Fires,
                                     ProcedureSource::FuncCreate => Method::Create,
                                     ProcedureSource::FuncLength => Method::Length,
                                     ProcedureSource::FuncAssert => Method::Assert,
                                     ProcedureSource::FuncPrint => Method::Print,
                                     ProcedureSource::CompRandomU32 => Method::ComponentRandomU32,
                                     ProcedureSource::CompTcpClient => Method::ComponentTcpClient,
                                     _ => todo!("other procedure sources"),
                                 };
                                 // Function call: consume the arguments
                                 let func_span = parser_type.full_span;
                                 let mut full_span = func_span;
                                 let arguments = self.consume_expression_list(
                                     module, iter, ctx, Some(&mut full_span.end)
                                 )?;
                                 ctx.heap.alloc_call_expression(|this| CallExpression{
                                     this, func_span, full_span, parser_type, method, arguments,
                                     procedure: procedure_id,
                                     parent: ExpressionParent::None,
                                     type_index: -1,
                                 }).upcast()
+                            }
+                        }
                     },
                     _ => {
                         return Err(ParseError::new_error_str_at_span(
                             &module.source, parser_type.full_span, "unexpected type in expression"
                         ))
+                    }
+                }
             } else {
                 // Check for builtin keywords or builtin functions
                 if ident_text == KW_LIT_NULL || ident_text == KW_LIT_TRUE || ident_text == KW_LIT_FALSE {
                     iter.consume();
                     // Parse builtin literal
                     let value = match ident_text {
                         KW_LIT_NULL => Literal::Null,
                         KW_LIT_TRUE => Literal::True,
                         KW_LIT_FALSE => Literal::False,
                         _ => unreachable!(),
                     };
                     ctx.heap.alloc_literal_expression(|this| LiteralExpression {
                         this,
                         span: ident_span,
                         value,
                         parent: ExpressionParent::None,
                         type_index: -1,
                     }).upcast()
                 } else if ident_text == KW_LET {
                     // Binding expression
                     let operator_span = iter.next_span();
                     iter.consume();
                     let bound_to = self.consume_prefix_expression(module, iter, ctx)?;
                     consume_token(&module.source, iter, TokenKind::Equal)?;
                     let bound_from = self.consume_prefix_expression(module, iter, ctx)?;
                     let full_span = InputSpan::from_positions(
                         operator_span.begin, ctx.heap[bound_from].full_span().end,
                     );
                     ctx.heap.alloc_binding_expression(|this| BindingExpression{
                         this, operator_span, full_span, bound_to, bound_from,
                         parent: ExpressionParent::None,
                         type_index: -1,
                     }).upcast()
                 } else if ident_text == KW_CAST {
                     // Casting expression
                     iter.consume();
                     let to_type = if Some(TokenKind::OpenAngle) == iter.next() {
                         let angle_start_pos = iter.next_start_position();
                         iter.consume();
                         let definition_id = self.cur_definition;
                         let poly_vars = ctx.heap[definition_id].poly_vars();
                         self.type_parser.consume_parser_type(
                             iter, &ctx.heap, &module.source, &ctx.symbols,
                             poly_vars, definition_id, SymbolScope::Module(module.root_id),
                             true, false, Some(angle_start_pos)
                         )?
                     } else {
                         // Automatic casting with inferred target type
                         ParserType{
                             elements: vec![ParserTypeElement{
                                 element_span: ident_span,
                                 variant: ParserTypeVariant::Inferred,
                             }],
                             full_span: ident_span
+                        }
                     };
                     consume_token(&module.source, iter, TokenKind::OpenParen)?;
                     let subject = self.consume_expression(module, iter, ctx)?;
                     let mut full_span = iter.next_span();
                     full_span.begin = to_type.full_span.begin;
                     consume_token(&module.source, iter, TokenKind::CloseParen)?;
                     ctx.heap.alloc_cast_expression(|this| CastExpression{
                         this,
                         cast_span: to_type.full_span,
                         full_span, to_type, subject,
                         parent: ExpressionParent::None,
                         type_index: -1,
                     }).upcast()
                 } else {
                     // Not a builtin literal, but also not a known type. So we
                     // assume it is a variable expression. Although if we do,
                     // then if a programmer mistyped a struct/function name the
                     // error messages will be rather cryptic. For polymorphic
                     // arguments we can't really do anything at all (because it
                     // uses the '<' token). In the other cases we try to provide
                     // a better error message.
                     iter.consume();
                     let next = iter.next();
                     if Some(TokenKind::ColonColon) == next {
                         return Err(ParseError::new_error_str_at_span(&module.source, ident_span, "unknown identifier"));
                     } else if Some(TokenKind::OpenParen) == next {
                         return Err(ParseError::new_error_str_at_span(
                             &module.source, ident_span,
                             "unknown identifier, did you mistype a union variant's, component's, or function's name?"
                         ));
                     } else if Some(TokenKind::OpenCurly) == next {
                         return Err(ParseError::new_error_str_at_span(
                             &module.source, ident_span,
                             "unknown identifier, did you mistype a struct type's name?"
                         ))
+                    }
                     let ident_text = ctx.pool.intern(ident_text);
                     let identifier = Identifier { span: ident_span, value: ident_text };
                     ctx.heap.alloc_variable_expression(|this| VariableExpression {
                         this,
                         identifier,
                         declaration: None,
                         used_as_binding_target: false,
                         parent: ExpressionParent::None,
                         type_index: -1,
                     }).upcast()
+                }
+            }
         } else {
             return Err(ParseError::new_error_str_at_pos(
                 &module.source, iter.last_valid_pos(), "expected an expression"
             ));
         };
         Ok(result)
+    }
     //--------------------------------------------------------------------------
     // Expression Utilities
     //--------------------------------------------------------------------------
     #[inline]
     fn consume_generic_binary_expression<
         M: Fn(Option<TokenKind>) -> Option<BinaryOperator>,
         F: Fn(&mut PassDefinitions, &Module, &mut TokenIter, &mut PassCtx) -> Result<ExpressionId, ParseError>
     >(
         &mut self, module: &Module, iter: &mut TokenIter, ctx: &mut PassCtx, match_fn: M, higher_precedence_fn: F
     ) -> Result<ExpressionId, ParseError> {
         let mut result = higher_precedence_fn(self, module, iter, ctx)?;
         while let Some(operation) = match_fn(iter.next()) {
             let operator_span = iter.next_span();
             iter.consume();
             let left = result;
             let right = higher_precedence_fn(self, module, iter, ctx)?;
             let full_span = InputSpan::from_positions(
                 ctx.heap[left].full_span().begin,
                 ctx.heap[right].full_span().end,
             );
             result = ctx.heap.alloc_binary_expression(|this| BinaryExpression{
                 this, operator_span, full_span, left, operation, right,
                 parent: ExpressionParent::None,
                 type_index: -1,
             }).upcast();
+        }
         Ok(result)
+    }
     #[inline]
     fn consume_expression_list(
         &mut self, module: &Module, iter: &mut TokenIter, ctx: &mut PassCtx, end_pos: Option<&mut InputPosition>
     ) -> Result<Vec<ExpressionId>, ParseError> {
         let mut section = self.expressions.start_section();
         consume_comma_separated(
             TokenKind::OpenParen, TokenKind::CloseParen, &module.source, iter, ctx,
             |_source, iter, ctx| self.consume_expression(module, iter, ctx),
             &mut section, "an expression", "a list of expressions", end_pos
         )?;
         Ok(section.into_vec())
+    }

src/protocol/parser/pass_validation_linking.rs

➞

Show inline comments

@@ @@ -971,385 +971,386 @@ impl Visitor for PassValidationLinking { @@
+                }
                 // Need to traverse fields expressions in struct and evaluate
                 // the poly args
                 let mut expr_section = self.expression_buffer.start_section();
                 for field in &literal.fields {
                     expr_section.push(field.value);
+                }
                 for expr_idx in 0..expr_section.len() {
                     let expr_id = expr_section[expr_idx];
                     self.expr_parent = ExpressionParent::Expression(upcast_id, expr_idx as u32);
                     self.visit_expr(ctx, expr_id)?;
+                }
                 expr_section.forget();
             },
             Literal::Enum(literal) => {
                 // Make sure the variant exists
                 let type_definition = ctx.types.get_base_definition(&literal.definition).unwrap();
                 let enum_definition = type_definition.definition.as_enum();
                 let variant_idx = enum_definition.variants.iter().position(|v| {
                     v.identifier == literal.variant
                 });
                 if variant_idx.is_none() {
                     let literal = ctx.heap[id].value.as_enum();
                     let ast_definition = ctx.heap[literal.definition].as_enum();
                     return Err(ParseError::new_error_at_span(
                         &ctx.module().source, literal.parser_type.full_span, format!(
                             "the variant '{}' does not exist on the enum '{}'",
                             literal.variant.value.as_str(), ast_definition.identifier.value.as_str()
+                        )
                     ));
+                }
                 literal.variant_idx = variant_idx.unwrap();
             },
             Literal::Union(literal) => {
                 // Make sure the variant exists
                 let type_definition = ctx.types.get_base_definition(&literal.definition).unwrap();
                 let union_definition = type_definition.definition.as_union();
                 let variant_idx = union_definition.variants.iter().position(|v| {
                     v.identifier == literal.variant
                 });
                 if variant_idx.is_none() {
                     let literal = ctx.heap[id].value.as_union();
                     let ast_definition = ctx.heap[literal.definition].as_union();
                     return Err(ParseError::new_error_at_span(
                         &ctx.module().source, literal.parser_type.full_span, format!(
                             "the variant '{}' does not exist on the union '{}'",
                             literal.variant.value.as_str(), ast_definition.identifier.value.as_str()
+                        )
                     ));
+                }
                 literal.variant_idx = variant_idx.unwrap();
                 // Make sure the number of specified values matches the expected
                 // number of embedded values in the union variant.
                 let union_variant = &union_definition.variants[literal.variant_idx];
                 if union_variant.embedded.len() != literal.values.len() {
                     let literal = ctx.heap[id].value.as_union();
                     let ast_definition = ctx.heap[literal.definition].as_union();
                     return Err(ParseError::new_error_at_span(
                         &ctx.module().source, literal.parser_type.full_span, format!(
                             "The variant '{}' of union '{}' expects {} embedded values, but {} were specified",
                             literal.variant.value.as_str(), ast_definition.identifier.value.as_str(),
                             union_variant.embedded.len(), literal.values.len()
                         ),
                     ))
+                }
                 // Traverse embedded values of union (if any) and evaluate the
                 // polymorphic arguments
                 let upcast_id = id.upcast();
                 let mut expr_section = self.expression_buffer.start_section();
                 for value in &literal.values {
                     expr_section.push(*value);
+                }
                 for expr_idx in 0..expr_section.len() {
                     let expr_id = expr_section[expr_idx];
                     self.expr_parent = ExpressionParent::Expression(upcast_id, expr_idx as u32);
                     self.visit_expr(ctx, expr_id)?;
+                }
                 expr_section.forget();
             },
             Literal::Array(literal) | Literal::Tuple(literal) => {
                 // Visit all expressions in the array
                 let upcast_id = id.upcast();
                 let expr_section = self.expression_buffer.start_section_initialized(literal);
                 for expr_idx in 0..expr_section.len() {
                     let expr_id = expr_section[expr_idx];
                     self.expr_parent = ExpressionParent::Expression(upcast_id, expr_idx as u32);
                     self.visit_expr(ctx, expr_id)?;
+                }
                 expr_section.forget();
+            }
+        }
         self.expr_parent = old_expr_parent;
         Ok(())
+    }
     fn visit_cast_expr(&mut self, ctx: &mut Ctx, id: CastExpressionId) -> VisitorResult {
         let cast_expr = &mut ctx.heap[id];
         if let Some(span) = self.must_be_assignable {
             return Err(ParseError::new_error_str_at_span(
                 &ctx.module().source, span, "cannot assign to the result from a cast expression"
             ))
+        }
         let upcast_id = id.upcast();
         let old_expr_parent = self.expr_parent;
         cast_expr.parent = old_expr_parent;
         // Recurse into the thing that we're casting
         self.expr_parent = ExpressionParent::Expression(upcast_id, 0);
         let subject_id = cast_expr.subject;
         self.visit_expr(ctx, subject_id)?;
         self.expr_parent = old_expr_parent;
         Ok(())
+    }
     fn visit_call_expr(&mut self, ctx: &mut Ctx, id: CallExpressionId) -> VisitorResult {
         let call_expr = &ctx.heap[id];
         if let Some(span) = self.must_be_assignable {
             return Err(ParseError::new_error_str_at_span(
                 &ctx.module().source, span, "cannot assign to the result from a call expression"
             ))
+        }
         // Check whether the method is allowed to be called within the code's
         // context (in sync, definition type, etc.)
         let mut expecting_wrapping_new_stmt = false;
         let mut expecting_primitive_def = false;
         let mut expecting_wrapping_sync_stmt = false;
         let mut expecting_no_select_stmt = false;
         match call_expr.method {
             Method::Get => {
                 expecting_primitive_def = true;
                 expecting_wrapping_sync_stmt = true;
                 if !self.in_select_guard.is_invalid() {
                     // In a select guard. Take the argument (i.e. the port we're
                     // retrieving from) and add it to the list of involved ports
                     // of the guard
                     if call_expr.arguments.len() == 1 {
                         // We're checking the number of arguments later, for now
                         // assume it is correct.
                         let argument = call_expr.arguments[0];
                         let select_stmt = &mut ctx.heap[self.in_select_guard];
                         let select_case = &mut select_stmt.cases[self.in_select_arm as usize];
                         select_case.involved_ports.push((id, argument));
+                    }
+                }
             },
             Method::Put => {
                 expecting_primitive_def = true;
                 expecting_wrapping_sync_stmt = true;
                 expecting_no_select_stmt = true;
             },
             Method::Fires => {
                 expecting_primitive_def = true;
                 expecting_wrapping_sync_stmt = true;
             },
             Method::Create => {},
             Method::Length => {},
             Method::Assert => {
                 expecting_wrapping_sync_stmt = true;
                 expecting_no_select_stmt = true;
                 if self.proc_kind == ProcedureKind::Function {
                     let call_span = call_expr.func_span;
                     return Err(ParseError::new_error_str_at_span(
                         &ctx.module().source, call_span,
                         "assert statement may only occur in components"
                     ));
+                }
             },
             Method::Print => {},
             Method::SelectStart
             | Method::SelectRegisterCasePort
             | Method::SelectWait => unreachable!(), // not usable by programmer directly
             Method::ComponentRandomU32 => {
             Method::ComponentRandomU32
             | Method::ComponentTcpClient => {
                 expecting_wrapping_new_stmt = true;
             },
             Method::UserFunction => {}
             Method::UserComponent => {
                 expecting_wrapping_new_stmt = true;
             },
+        }
         let call_expr = &mut ctx.heap[id];
         fn get_span_and_name<'a>(ctx: &'a Ctx, id: CallExpressionId) -> (InputSpan, String) {
             let call = &ctx.heap[id];
             let span = call.func_span;
             let name = String::from_utf8_lossy(ctx.module().source.section_at_span(span)).to_string();
             return (span, name);
+        }
         if expecting_primitive_def {
             if self.proc_kind != ProcedureKind::Primitive {
                 let (call_span, func_name) = get_span_and_name(ctx, id);
                 return Err(ParseError::new_error_at_span(
                     &ctx.module().source, call_span,
                     format!("a call to '{}' may only occur in primitive component definitions", func_name)
                 ));
+            }
+        }
         if expecting_wrapping_sync_stmt {
             if self.in_sync.is_invalid() {
                 let (call_span, func_name) = get_span_and_name(ctx, id);
                 return Err(ParseError::new_error_at_span(
                     &ctx.module().source, call_span,
                     format!("a call to '{}' may only occur inside synchronous blocks", func_name)
                 ))
+            }
+        }
         if expecting_no_select_stmt {
             if !self.in_select_guard.is_invalid() {
                 let (call_span, func_name) = get_span_and_name(ctx, id);
                 return Err(ParseError::new_error_at_span(
                     &ctx.module().source, call_span,
                     format!("a call to '{}' may not occur in a select statement's guard", func_name)
                 ));
+            }
+        }
         if expecting_wrapping_new_stmt {
             if !self.expr_parent.is_new() {
                 let call_span = call_expr.func_span;
                 return Err(ParseError::new_error_str_at_span(
                     &ctx.module().source, call_span,
                     "cannot call a component, it can only be instantiated by using 'new'"
                 ));
+            }
         } else {
             if self.expr_parent.is_new() {
                 let call_span = call_expr.func_span;
                 return Err(ParseError::new_error_str_at_span(
                     &ctx.module().source, call_span,
                     "only components can be instantiated, this is a function"
                 ));
+            }
+        }
         // Check the number of arguments
         let call_definition = ctx.types.get_base_definition(&call_expr.procedure.upcast()).unwrap();
         let num_expected_args = match &call_definition.definition {
             DefinedTypeVariant::Procedure(definition) => definition.arguments.len(),
             _ => unreachable!(),
         };
         let num_provided_args = call_expr.arguments.len();
         if num_provided_args != num_expected_args {
             let argument_text = if num_expected_args == 1 { "argument" } else { "arguments" };
             let call_span = call_expr.full_span;
             return Err(ParseError::new_error_at_span(
                 &ctx.module().source, call_span, format!(
                     "expected {} {}, but {} were provided",
                     num_expected_args, argument_text, num_provided_args
+                )
             ));
+        }
         // Recurse into all of the arguments and set the expression's parent
         let upcast_id = id.upcast();
         let section = self.expression_buffer.start_section_initialized(&call_expr.arguments);
         let old_expr_parent = self.expr_parent;
         call_expr.parent = old_expr_parent;
         for arg_expr_idx in 0..section.len() {
             let arg_expr_id = section[arg_expr_idx];
             self.expr_parent = ExpressionParent::Expression(upcast_id, arg_expr_idx as u32);
             self.visit_expr(ctx, arg_expr_id)?;
+        }
         section.forget();
         self.expr_parent = old_expr_parent;
         Ok(())
+    }
     fn visit_variable_expr(&mut self, ctx: &mut Ctx, id: VariableExpressionId) -> VisitorResult {
         let var_expr = &ctx.heap[id];
         // Check if declaration was already resolved (this occurs for the
         // variable expr that is on the LHS of the assignment expr that is
         // associated with a variable declaration)
         let mut variable_id = var_expr.declaration;
         let mut is_binding_target = false;
         // Otherwise try to find it
         if variable_id.is_none() {
             variable_id = self.find_variable(ctx, self.relative_pos_in_parent, &var_expr.identifier);
+        }
         // Otherwise try to see if is a variable introduced by a binding expr
         let variable_id = if let Some(variable_id) = variable_id {
             variable_id
         } else {
             if self.in_binding_expr.is_invalid() || !self.in_binding_expr_lhs {
                 return Err(ParseError::new_error_str_at_span(
                     &ctx.module().source, var_expr.identifier.span, "unresolved variable"
                 ));
+            }
             // This is a binding variable, but it may only appear in very
             // specific locations.
             let is_valid_binding = match self.expr_parent {
                 ExpressionParent::Expression(expr_id, idx) => {
                     match &ctx.heap[expr_id] {
                         Expression::Binding(_binding_expr) => {
                             // Nested binding is disallowed, and because of
                             // the check above we know we're directly at the
                             // LHS of the binding expression
                             debug_assert_eq!(_binding_expr.this, self.in_binding_expr);
                             debug_assert_eq!(idx, 0);
                             true
+                        }
                         Expression::Literal(_lit_expr) => {
                             // Only struct, unions, tuples and arrays can
                             // have subexpressions, so we're always fine
                             dbg_code!({
                                 match _lit_expr.value {
                                     Literal::Struct(_) | Literal::Union(_) | Literal::Array(_) | Literal::Tuple(_) => {},
                                     _ => unreachable!(),
+                                }
                             });
                             true
                         },
                         _ => false,
+                    }
                 },
                 _ => {
                     false
+                }
             };
             if !is_valid_binding {
                 let binding_expr = &ctx.heap[self.in_binding_expr];
                 return Err(ParseError::new_error_str_at_span(
                     &ctx.module().source, var_expr.identifier.span,
                     "illegal location for binding variable: binding variables may only be nested under a binding expression, or a struct, union or array literal"
                 ).with_info_at_span(
                     &ctx.module().source, binding_expr.operator_span, format!(
                         "'{}' was interpreted as a binding variable because the variable is not declared and it is nested under this binding expression",
                         var_expr.identifier.value.as_str()
+                    )
                 ));
+            }
             // By now we know that this is a valid binding expression. Given
             // that a binding expression must be nested under an if/while
             // statement, we now add the variable to the scope associated with
             // that statement.
             let bound_identifier = var_expr.identifier.clone();
             let bound_variable_id = ctx.heap.alloc_variable(|this| Variable {
                 this,
                 kind: VariableKind::Binding,
                 parser_type: ParserType {
                     elements: vec![ParserTypeElement {
                         element_span: bound_identifier.span,
                         variant: ParserTypeVariant::Inferred
                     }],
                     full_span: bound_identifier.span
                 },
                 identifier: bound_identifier,
                 relative_pos_in_parent: 0,
                 unique_id_in_scope: -1,
             });

src/runtime2/component/component.rs

➞

Show inline comments

 use crate::protocol::eval::{Prompt, EvalError, ValueGroup, PortId as EvalPortId};
 use crate::protocol::*;
 use crate::runtime2::*;
 use crate::runtime2::communication::*;
 use super::{CompCtx, CompPDL};
+use super::{CompCtx, CompPDL, CompId};
 use super::component_context::*;
 use super::component_random::*;
 use super::component_internet::*;
 use super::control_layer::*;
 use super::consensus::*;
 pub enum CompScheduling {
     Immediate,
     Requeue,
     Sleep,
     Exit,
+}
 /// Generic representation of a component (as viewed by a scheduler).
 pub(crate) trait Component {
     /// Called upon the creation of the component.
     fn on_creation(&mut self, sched_ctx: &SchedulerCtx);
+    fn on_creation(&mut self, comp_id: CompId, sched_ctx: &SchedulerCtx);
     /// Called if the component is created by another component and the messages
     /// are being transferred between the two.
     fn adopt_message(&mut self, comp_ctx: &mut CompCtx, message: DataMessage);
     /// Called if the component receives a new message. The component is
     /// responsible for deciding where that messages goes.
     fn handle_message(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx, message: Message);
     /// Called if the component's routine should be executed. The return value
     /// can be used to indicate when the routine should be run again.
     fn run(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx) -> Result<CompScheduling, EvalError>;
+}
 /// Representation of the generic operating mode of a component.
 #[derive(Debug, Copy, Clone, PartialEq, Eq)]
 pub(crate) enum CompMode {
     NonSync, // not in sync mode
     Sync, // in sync mode, can interact with other components
     SyncEnd, // awaiting a solution, i.e. encountered the end of the sync block
     BlockedGet, // blocked because we need to receive a message on a particular port
     BlockedPut, // component is blocked because the port is blocked
     BlockedSelect, // waiting on message to complete the select statement
     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
+}
 impl CompMode {
     pub(crate) fn is_in_sync_block(&self) -> bool {
         use CompMode::*;
         match self {
             Sync | SyncEnd | BlockedGet | BlockedPut | BlockedSelect => true,
             NonSync | StartExit | BusyExit | Exit => false,
+        }
+    }
+}
 /// Component execution state: the execution mode along with some descriptive
 /// fields. Fields are public for ergonomic reasons, use member functions when
 /// appropriate.
 pub(crate) struct CompExecState {
     pub mode: CompMode,
     pub mode_port: PortId, // valid if blocked on a port (put/get)
     pub mode_value: ValueGroup, // valid if blocked on a put
+}
 impl CompExecState {
     pub(crate) fn new() -> Self {
         return Self{
             mode: CompMode::NonSync,
             mode_port: PortId::new_invalid(),
             mode_value: ValueGroup::default(),
+        }
+    }
     pub(crate) fn set_as_blocked_get(&mut self, port: PortId) {
         self.mode = CompMode::BlockedGet;
         self.mode_port = port;
         debug_assert!(self.mode_value.values.is_empty());
+    }
     pub(crate) fn is_blocked_on_get(&self, port: PortId) -> bool {
         return
             self.mode == CompMode::BlockedGet &&
             self.mode_port == port;
+    }
     pub(crate) fn set_as_blocked_put(&mut self, port: PortId, value: ValueGroup) {
         self.mode = CompMode::BlockedPut;
         self.mode_port = port;
         self.mode_value = value;
+    }
     pub(crate) fn is_blocked_on_put(&self, port: PortId) -> bool {
         return
             self.mode == CompMode::BlockedPut &&
             self.mode_port == port;
+    }
+}
 /// Creates a new component based on its definition. Meaning that if it is a
 /// user-defined component then we set up the PDL code state. Otherwise we
 /// construct a custom component. This does NOT take care of port and message
 /// management.
 pub(crate) fn create_component(
     protocol: &ProtocolDescription,
     definition_id: ProcedureDefinitionId, type_id: TypeId,
     arguments: ValueGroup, num_ports: usize
 ) -> Box<dyn Component> {
     let definition = &protocol.heap[definition_id];
     debug_assert!(definition.kind == ProcedureKind::Primitive || definition.kind == ProcedureKind::Composite);
     if definition.source.is_builtin() {
         // Builtin component
         let component = match definition.source {
+        let component: Box<dyn Component> = match definition.source {
             ProcedureSource::CompRandomU32 => Box::new(ComponentRandomU32::new(arguments)),
             ProcedureSource::CompTcpClient => Box::new(ComponentTcpClient::new(arguments)),
             _ => unreachable!(),
         };
         return component;
     } else {
         // User-defined component
         let prompt = Prompt::new(
             &protocol.types, &protocol.heap,
             definition_id, type_id, arguments
         );
         let component = CompPDL::new(prompt, num_ports);
         return Box::new(component);
+    }
+}
 // -----------------------------------------------------------------------------
 // Generic component messaging utilities (for sending and receiving)
 // -----------------------------------------------------------------------------
 /// Default handling of sending a data message. In case the port is blocked then
 /// the `ExecState` will become blocked as well. Note that
 /// `default_handle_control_message` will ensure that the port becomes
 /// unblocked if so instructed by the receiving component. The returned
 /// scheduling value must be used.
 #[must_use]
 pub(crate) fn default_send_data_message(
     exec_state: &mut CompExecState, transmitting_port_id: PortId, value: ValueGroup,
     sched_ctx: &SchedulerCtx, consensus: &mut Consensus, comp_ctx: &mut CompCtx
 ) -> CompScheduling {
     debug_assert_eq!(exec_state.mode, CompMode::Sync);
     // TODO: Handle closed ports
     let port_handle = comp_ctx.get_port_handle(transmitting_port_id);
     let port_info = comp_ctx.get_port(port_handle);
     debug_assert_eq!(port_info.kind, PortKind::Putter);
     if port_info.state.is_blocked() {
         // Port is blocked, so we cannot send
         exec_state.set_as_blocked_put(transmitting_port_id, value);
         return CompScheduling::Sleep;
     } else {
         // Port is not blocked, so send to the peer
         let peer_handle = comp_ctx.get_peer_handle(port_info.peer_comp_id);
         let peer_info = comp_ctx.get_peer(peer_handle);
         let annotated_message = consensus.annotate_data_message(comp_ctx, port_info, value);
         peer_info.handle.send_message(&sched_ctx.runtime, Message::Data(annotated_message), true);
         return CompScheduling::Immediate;
+    }
+}
 pub(crate) enum IncomingData {
     PlacedInSlot,
     SlotFull(DataMessage),
+}
 /// Default handling of receiving a data message. In case there is no room for
 /// the message it is returned from this function. Note that this function is
 /// different from PDL code performing a `get` on a port; this is the case where
 /// the message first arrives at the component.
 // NOTE: This is supposed to be a somewhat temporary implementation. It would be
 //  nicest if the sending component can figure out it cannot send any more data.
 #[must_use]
 pub(crate) fn default_handle_incoming_data_message(
     exec_state: &mut CompExecState, port_value_slot: &mut Option<DataMessage>,
     comp_ctx: &mut CompCtx, incoming_message: DataMessage,
     sched_ctx: &SchedulerCtx, control: &mut ControlLayer
 ) -> IncomingData {
     let target_port_id = incoming_message.data_header.target_port;
     if port_value_slot.is_none() {
         // We can put the value in the slot
         *port_value_slot = Some(incoming_message);
         // Check if we're blocked on receiving this message.
         dbg_code!({
             // Our port cannot have been blocked itself, because we're able to
             // directly insert the message into its slot.
             let port_handle = comp_ctx.get_port_handle(target_port_id);
             assert!(!comp_ctx.get_port(port_handle).state.is_blocked());
         });
         if exec_state.is_blocked_on_get(target_port_id) {
             // Return to normal operation
             exec_state.mode = CompMode::Sync;
             exec_state.mode_port = PortId::new_invalid();
             debug_assert!(exec_state.mode_value.values.is_empty());
+        }
         return IncomingData::PlacedInSlot
     } else {
         // Slot is already full, so if the port was previously opened, it will
         // now become closed
         let port_handle = comp_ctx.get_port_handle(target_port_id);
         let port_info = comp_ctx.get_port_mut(port_handle);
         debug_assert!(port_info.state == PortState::Open || port_info.state.is_blocked()); // i.e. not closed, but will go off if more states are added in the future
         if port_info.state == PortState::Open {
             comp_ctx.set_port_state(port_handle, PortState::BlockedDueToFullBuffers);
             let (peer_handle, message) =
                 control.initiate_port_blocking(comp_ctx, port_handle);
             let peer = comp_ctx.get_peer(peer_handle);
             peer.handle.send_message(&sched_ctx.runtime, Message::Control(message), true);
+        }
         return IncomingData::SlotFull(incoming_message)
+    }
+}
 /// Handles control messages in the default way. Note that this function may
 /// take a lot of actions in the name of the caller: pending messages may be
 /// sent, ports may become blocked/unblocked, etc. So the execution
 /// (`CompExecState`), control (`ControlLayer`) and consensus (`Consensus`)
 /// state may all change.
 pub(crate) fn default_handle_control_message(
     exec_state: &mut CompExecState, control: &mut ControlLayer, consensus: &mut Consensus,
     message: ControlMessage, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx
 ) {
     match message.content {
         ControlMessageContent::Ack => {
             default_handle_ack(control, message.id, sched_ctx, comp_ctx);
         },
         ControlMessageContent::BlockPort(port_id) => {
             // One of our messages was accepted, but the port should be
             // blocked.
             let port_handle = comp_ctx.get_port_handle(port_id);
             let port_info = comp_ctx.get_port(port_handle);
             debug_assert_eq!(port_info.kind, PortKind::Putter);
             if port_info.state == PortState::Open {
                 // only when open: we don't do this when closed, and we we don't do this if we're blocked due to peer changes
                 comp_ctx.set_port_state(port_handle, PortState::BlockedDueToFullBuffers);
+            }
         },
         ControlMessageContent::ClosePort(port_id) => {
             // Request to close the port. We immediately comply and remove
             // the component handle as well
             let port_handle = comp_ctx.get_port_handle(port_id);
             let peer_comp_id = comp_ctx.get_port(port_handle).peer_comp_id;
             let peer_handle = comp_ctx.get_peer_handle(peer_comp_id);
             // One exception to sending an `Ack` is if we just closed the
             // port ourselves, meaning that the `ClosePort` messages got
             // sent to one another.
             if let Some(control_id) = control.has_close_port_entry(port_handle, comp_ctx) {
                 default_handle_ack(control, control_id, sched_ctx, comp_ctx);
             } else {
                 default_send_ack(message.id, peer_handle, sched_ctx, comp_ctx);
                 comp_ctx.remove_peer(sched_ctx, port_handle, peer_comp_id, false); // do not remove if closed
                 comp_ctx.set_port_state(port_handle, PortState::Closed); // now set to closed
+            }
         },
         ControlMessageContent::UnblockPort(port_id) => {
             // We were previously blocked (or already closed)
             let port_handle = comp_ctx.get_port_handle(port_id);
             let port_info = comp_ctx.get_port(port_handle);
             debug_assert_eq!(port_info.kind, PortKind::Putter);
             if port_info.state == PortState::BlockedDueToFullBuffers {
                 default_handle_unblock_put(exec_state, consensus, port_handle, sched_ctx, comp_ctx);
+            }
         },
         ControlMessageContent::PortPeerChangedBlock(port_id) => {
             // The peer of our port has just changed. So we are asked to
             // temporarily block the port (while our original recipient is
             // potentially rerouting some of the in-flight messages) and
             // Ack. Then we wait for the `unblock` call.
             debug_assert_eq!(message.target_port_id, Some(port_id));
             let port_handle = comp_ctx.get_port_handle(port_id);
             comp_ctx.set_port_state(port_handle, PortState::BlockedDueToPeerChange);
             let port_info = comp_ctx.get_port(port_handle);
             let peer_handle = comp_ctx.get_peer_handle(port_info.peer_comp_id);
             default_send_ack(message.id, peer_handle, sched_ctx, comp_ctx);
         },
         ControlMessageContent::PortPeerChangedUnblock(new_port_id, new_comp_id) => {
             let port_handle = comp_ctx.get_port_handle(message.target_port_id.unwrap());
             let port_info = comp_ctx.get_port(port_handle);
             debug_assert!(port_info.state == PortState::BlockedDueToPeerChange);
             let old_peer_id = port_info.peer_comp_id;
             comp_ctx.remove_peer(sched_ctx, port_handle, old_peer_id, false);
             let port_info = comp_ctx.get_port_mut(port_handle);
             port_info.peer_comp_id = new_comp_id;
             port_info.peer_port_id = new_port_id;
             comp_ctx.add_peer(port_handle, sched_ctx, new_comp_id, None);
             default_handle_unblock_put(exec_state, consensus, port_handle, sched_ctx, comp_ctx);
+        }
+    }
+}
 /// Handles a component initiating the exiting procedure, and closing all of its

src/runtime2/component/component_internet.rs

➞

Show inline comments

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

src/runtime2/component/component_pdl.rs

➞

Show inline comments

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

src/runtime2/component/component_random.rs

➞

Show inline comments

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

src/runtime2/poll/mod.rs

➞

Show inline comments

@@ @@ -106,219 +106,219 @@ impl Poller { @@
 #[cfg(unix)]
 impl Drop for Poller {
     fn drop(&mut self) {
         unsafe{ libc::close(self.handle); }
+    }
+}
 #[inline]
 fn syscall_result(result: c_int) -> io::Result<c_int> {
     if result < 0 {
         return Err(io::Error::last_os_error());
     } else {
         return Ok(result);
+    }
+}
 #[cfg(not(unix))]
 struct Poller {
     // Not implemented for OS's other than unix
+}
 // -----------------------------------------------------------------------------
 // Polling Thread
 // -----------------------------------------------------------------------------
 enum PollCmd {
     Register(CompHandle, UserData),
     Unregister(FileDescriptor, UserData),
     Shutdown,
+}
 pub struct PollingThread {
     poller: Arc<Poller>,
     runtime: Arc<RuntimeInner>,
     queue: QueueDynMpsc<PollCmd>,
     logging_enabled: bool,
+}
 impl PollingThread {
     pub(crate) fn new(runtime: Arc<RuntimeInner>, logging_enabled: bool) -> Result<(PollingThreadHandle, PollingClientFactory), RtError> {
         let poller = Poller::new()
             .map_err(|e| rt_error!("failed to create poller, because: {}", e))?;
         let poller = Arc::new(poller);
         let queue = QueueDynMpsc::new(64);
         let queue_producers = queue.producer_factory();
         let mut thread_data = PollingThread{
             poller: poller.clone(),
             runtime: runtime.clone(),
             queue,
             logging_enabled,
         };
         let thread_handle = thread::spawn(move || { thread_data.run() });
         let thread_handle = PollingThreadHandle{
             queue: Some(queue_producers.producer()),
             handle: Some(thread_handle),
         };
         let client_factory = PollingClientFactory{
             poller,
             generation_counter: Arc::new(AtomicU32::new(0)),
             queue_factory: queue_producers,
         };
         return Ok((thread_handle, client_factory));
+    }
     pub(crate) fn run(&mut self) {
         use crate::runtime2::scheduler::SchedulerCtx;
         use crate::runtime2::communication::Message;
         const NUM_EVENTS: usize = 256;
         const EPOLL_DURATION: time::Duration = time::Duration::from_millis(250);
         // @performance: Lot of improvements possible here, a HashMap is likely
         // a horrible way to do this.
         let mut events = Vec::with_capacity(NUM_EVENTS);
         let mut lookup = HashMap::with_capacity(64);
         self.log("Starting polling thread");
         loop {
             // Retrieve events first (because the PollingClient will first
             // register at epoll, and then push a command into the queue).
             self.poller.wait(&mut events, EPOLL_DURATION).unwrap();
             // Then handle everything in the command queue.
             while let Some(command) = self.queue.pop() {
                 match command {
                     PollCmd::Register(handle, user_data) => {
                         self.log(&format!("Registering component {:?} as {}", handle.id(), user_data.0));
                         let key = Self::user_data_as_key(user_data);
                         debug_assert!(!lookup.contains_key(&key));
                         lookup.insert(key, handle);
                     },
                     PollCmd::Unregister(_file_descriptor, user_data) => {
                         let key = Self::user_data_as_key(user_data);
                         debug_assert!(lookup.contains_key(&key));
                         let mut handle = lookup.remove(&key).unwrap();
                         self.log(&format!("Unregistering component {:?} as {}", handle.id(), user_data.0));
                         if let Some(key) = handle.decrement_users() {
                             self.runtime.destroy_component(key);
+                        }
                     },
                     PollCmd::Shutdown => {
                         // The contract is that all scheduler threads shutdown
                         // before the polling thread. This happens when all
                         // components are removed.
                         self.log("Received shutdown signal");
                         debug_assert!(lookup.is_empty());
                         return;
+                    }
+                }
+            }
             // Now process all of the events. Because we might have had a
             // `Register` command followed by an `Unregister` command (e.g. a
             // component has died), we might get events that are not associated
             // with an entry in the lookup.
             for event in events.drain(..) {
                 let key = event.u64;
                 if let Some(handle) = lookup.get(&key) {
                     let events = event.events;
                     self.log(&format!("Sending poll to {:?} (event: {:x})", handle.id(), events));
                     handle.send_message(&self.runtime, Message::Poll, true);
+                }
+            }
+        }
+    }
     #[inline]
     fn user_data_as_key(data: UserData) -> u64 {
         return data.0;
+    }
     fn log(&self, message: &str) {
         if self.logging_enabled {
             println!("[polling] {}", message);
+        }
+    }
+}
 // bit convoluted, but it works
 pub(crate) struct PollingThreadHandle {
     // requires Option, because:
     queue: Option<QueueDynProducer<PollCmd>>, // destructor needs to be called
     handle: Option<thread::JoinHandle<()>>, // we need to call `join`
+}
 impl PollingThreadHandle {
     pub(crate) fn shutdown(&mut self) -> thread::Result<()> {
         debug_assert!(self.handle.is_some(), "polling thread already destroyed");
         self.queue.take().unwrap().push(PollCmd::Shutdown);
         return self.handle.take().unwrap().join();
+    }
+}
 impl Drop for PollingThreadHandle {
     fn drop(&mut self) {
         debug_assert!(self.queue.is_none() && self.handle.is_none());
+    }
+}
 // oh my god, now I'm writing factory objects. I'm not feeling too well
 pub(crate) struct PollingClientFactory {
     poller: Arc<Poller>,
     generation_counter: Arc<AtomicU32>,
     queue_factory: QueueDynProducerFactory<PollCmd>,
+}
 impl PollingClientFactory {
     pub(crate) fn client(&self) -> PollingClient {
         return PollingClient{
             poller: self.poller.clone(),
             generation_counter: self.generation_counter.clone(),
             queue: self.queue_factory.producer(),
         };
+    }
+}
 pub(crate) struct PollTicket(FileDescriptor, u64);
 /// A structure that allows the owner to register components at the polling
 /// thread. Because of assumptions in the communication queue all of these
 /// clients should be dropped before stopping the polling thread.
 pub(crate) struct PollingClient {
     poller: Arc<Poller>,
     generation_counter: Arc<AtomicU32>,
     queue: QueueDynProducer<PollCmd>,
+}
 impl PollingClient {
     fn register<F: AsFileDescriptor>(&self, entity: F, handle: CompHandle, read: bool, write: bool) -> Result<PollTicket, RtError> {
+    pub(crate) fn register<F: AsFileDescriptor>(&self, entity: &F, handle: CompHandle, read: bool, write: bool) -> Result<PollTicket, RtError> {
         let generation = self.generation_counter.fetch_add(1, Ordering::Relaxed);
         let user_data = user_data_for_component(handle.id().0, generation);
         self.queue.push(PollCmd::Register(handle, user_data));
         let file_descriptor = entity.as_file_descriptor();
         self.poller.register(file_descriptor, user_data, read, write)
             .map_err(|e| rt_error!("failed to register for polling, because: {}", e))?;
         return Ok(PollTicket(file_descriptor, user_data.0));
+    }
     fn unregister(&self, ticket: PollTicket) -> Result<(), RtError> {
+    pub(crate) fn unregister(&self, ticket: PollTicket) -> Result<(), RtError> {
         let file_descriptor = ticket.0;
         let user_data = UserData(ticket.1);
         self.queue.push(PollCmd::Unregister(file_descriptor, user_data));
         self.poller.unregister(file_descriptor)
             .map_err(|e| rt_error!("failed to unregister polling, because: {}", e))?;
         return Ok(());
+    }
+}
 #[inline]
 fn user_data_for_component(component_id: u32, generation: u32) -> UserData {
     return UserData((generation as u64) << 32 | (component_id as u64));
+}
@@ \ No newline at end of file @@

src/runtime2/runtime.rs

➞

Show inline comments

@@ @@ -117,222 +117,225 @@ impl CompHandle { @@
     /// Returns the `CompKey` to the component if it should be destroyed
     pub(crate) fn decrement_users(&mut self) -> Option<CompKey> {
         dbg_code!(assert!(!self.decremented, "illegal to 'decrement_users' twice"));
         let old_count = self.num_handles.fetch_sub(1, Ordering::AcqRel);
         let new_count = old_count - 1;
         dbg_code!(self.decremented = true);
         if new_count == 0 {
             return Some(unsafe{ self.id.upgrade() });
+        }
         return None;
+    }
+}
 impl Clone for CompHandle {
     fn clone(&self) -> Self {
         dbg_code!(assert!(!self.decremented, "illegal to clone after 'decrement_users'"));
         self.increment_users();
         return CompHandle{
             target: self.target,
             id: self.id,
             #[cfg(debug_assertions)] decremented: false,
         };
+    }
+}
 impl std::ops::Deref for CompHandle {
     type Target = CompPublic;
     fn deref(&self) -> &Self::Target {
         dbg_code!(assert!(!self.decremented)); // cannot access if control is relinquished
         return unsafe{ &*self.target };
+    }
+}
 impl Drop for CompHandle {
     fn drop(&mut self) {
         dbg_code!(assert!(self.decremented, "need call to 'decrement_users' before dropping"));
+    }
+}
 // -----------------------------------------------------------------------------
 // Runtime
 // -----------------------------------------------------------------------------
 pub struct Runtime {
     pub(crate) inner: Arc<RuntimeInner>,
     scheduler_threads: Vec<thread::JoinHandle<()>>,
     polling_handle: PollingThreadHandle,
+}
 impl Runtime {
     // TODO: debug_logging should be removed at some point
     pub fn new(num_threads: u32, debug_logging: bool, protocol_description: ProtocolDescription) -> Result<Runtime, RtError> {
         if num_threads == 0 {
             return Err(rt_error!("need at least one thread to create the runtime"));
+        }
         let runtime_inner = Arc::new(RuntimeInner {
             protocol: protocol_description,
             components: ComponentStore::new(128),
             work_queue: Mutex::new(VecDeque::with_capacity(128)),
             work_condvar: Condvar::new(),
             active_elements: AtomicU32::new(1),
         });
         let (polling_handle, polling_clients) = rt_error_try!(
             PollingThread::new(runtime_inner.clone(), debug_logging),
             "failed to build polling thread"
         );
         let mut scheduler_threads = Vec::with_capacity(num_threads as usize);
         for thread_index in 0..num_threads {
             let mut scheduler = Scheduler::new(
                 runtime_inner.clone(), polling_clients.client(),
                 thread_index, debug_logging
             );
             let thread_handle = thread::spawn(move || {
                 scheduler.run();
             });
             scheduler_threads.push(thread_handle);
+        }
         return Ok(Runtime{
             inner: runtime_inner,
             scheduler_threads,
             polling_handle,
         });
+    }
     pub fn create_component(&self, module_name: &[u8], routine_name: &[u8]) -> Result<(), ComponentCreationError> {
         use crate::protocol::eval::ValueGroup;
         let prompt = self.inner.protocol.new_component(
             module_name, routine_name,
             ValueGroup::new_stack(Vec::new())
         )?;
         let reserved = self.inner.start_create_pdl_component();
         let ctx = CompCtx::new(&reserved);
         let component = Box::new(CompPDL::new(prompt, 0));
         let (key, _) = self.inner.finish_create_pdl_component(reserved, component, ctx, false);
         self.inner.enqueue_work(key);
         return Ok(())
+    }
+}
 impl Drop for Runtime {
     fn drop(&mut self) {
         self.inner.decrement_active_components();
         for handle in self.scheduler_threads.drain(..) {
             handle.join().expect("join scheduler thread");
+        }
         self.polling_handle.shutdown().expect("shutdown polling thread");
+    }
+}
 /// Memory that is maintained by "the runtime". In practice it is maintained by
 /// multiple schedulers, and this serves as the common interface to that memory.
 pub(crate) struct RuntimeInner {
     pub protocol: ProtocolDescription,
     components: ComponentStore<RuntimeComp>,
     work_queue: Mutex<VecDeque<CompKey>>,
     work_condvar: Condvar,
     active_elements: AtomicU32, // active components and APIs (i.e. component creators)
+}
 impl RuntimeInner {
     // Scheduling and retrieving work
     pub(crate) fn take_work(&self) -> Option<CompKey> {
         let mut lock = self.work_queue.lock().unwrap();
         while lock.is_empty() && self.active_elements.load(Ordering::Acquire) != 0 {
             lock = self.work_condvar.wait(lock).unwrap();
+        }
         // We have work, or the schedulers should exit.
         return lock.pop_front();
+    }
     pub(crate) fn enqueue_work(&self, key: CompKey) {
         let mut lock = self.work_queue.lock().unwrap();
         lock.push_back(key);
         self.work_condvar.notify_one();
+    }
     // Creating/destroying components
     pub(crate) fn start_create_pdl_component(&self) -> CompReserved {
         self.increment_active_components();
         let reservation = self.components.reserve();
         return CompReserved{ reservation };
+    }
     pub(crate) fn finish_create_pdl_component(
         &self, reserved: CompReserved,
         component: Box<dyn Component>, mut context: CompCtx, initially_sleeping: bool,
     ) -> (CompKey, &mut RuntimeComp) {
         let inbox_queue = QueueDynMpsc::new(16);
         let inbox_producer = inbox_queue.producer();
         let _id = reserved.id();
         context.id = reserved.id();
         let component = RuntimeComp {
             public: CompPublic{
                 sleeping: AtomicBool::new(initially_sleeping),
                 num_handles: AtomicU32::new(1), // the component itself acts like a handle
                 inbox: inbox_producer,
             },
             component,
             ctx: context,
             inbox: inbox_queue,
             exiting: false,
         };
         let index = self.components.submit(reserved.reservation, component);
         debug_assert_eq!(index, _id.0);
         let component = self.components.get_mut(index);
         return (CompKey(index), component);
+    }
     pub(crate) fn get_component(&self, key: CompKey) -> &mut RuntimeComp {
         let component = self.components.get_mut(key.0);
         return component;
+    }
     pub(crate) fn get_component_public(&self, id: CompId) -> CompHandle {
         let component = self.components.get(id.0);
         return CompHandle::new(id, &component.public);
+    }
     /// Will remove a component and its memory from the runtime. May only be
     /// called if the necessary conditions for destruction have been met.
     pub(crate) fn destroy_component(&self, key: CompKey) {
         dbg_code!({
             let component = self.get_component(key);
             debug_assert!(component.exiting);
             debug_assert_eq!(component.public.num_handles.load(Ordering::Acquire), 0);
         });
         self.decrement_active_components();
         self.components.destroy(key.0);
+    }
     // Tracking number of active interfaces and the active components
     #[inline]
     fn increment_active_components(&self) {
         let _old_val = self.active_elements.fetch_add(1, Ordering::AcqRel);
         debug_assert!(_old_val > 0); // can only create a component from a API/component, so can never be 0.
+    }
     fn decrement_active_components(&self) {
         let old_val = self.active_elements.fetch_sub(1, Ordering::AcqRel);
         debug_assert!(old_val > 0); // something wrong with incr/decr logic
         let new_val = old_val - 1;
         if new_val == 0 {
             // Just to be sure, in case the last thing that gets destroyed is an
             // API instead of a thread.
             let _lock = self.work_queue.lock();
             self.work_condvar.notify_all();
+        }
+    }
+}

src/runtime2/stdlib/internet.rs

➞

Show inline comments

 use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
 use std::mem::size_of;
 use std::io::{Error as IoError, ErrorKind as IoErrorKind};
 use libc::{
     c_int,
     sockaddr_in, sockaddr_in6, in_addr, in6_addr,
     socket, bind, listen, accept, connect, close,
 };
 use mio::{event, Interest, Registry, Token};
 use crate::runtime2::poll::{AsFileDescriptor, FileDescriptor};
 #[derive(Debug)]
 pub enum SocketError {
     Opening,
     Modifying,
     Binding,
     Listening,
     Connecting,
     Accepted,
     Accepting,
+}
 enum SocketState {
     Opened,
     Listening,
+}
 /// TCP connection
 pub struct SocketTcpClient {
     socket_handle: libc::c_int,
     is_blocking: bool,
+}
 impl SocketTcpClient {
     pub fn new(ip: IpAddr, port: u16) -> Result<Self, SocketError> {
         const BLOCKING: bool = false;
         let socket_handle = create_and_connect_socket(
             libc::SOCK_STREAM, libc::IPPROTO_TCP, ip, port
         )?;
         if !set_socket_blocking(socket_handle, BLOCKING) {
             unsafe{ libc::close(socket_handle); }
             return Err(SocketError::Modifying);
+        }
         return Ok(SocketTcpClient{
             socket_handle,
             is_blocking: BLOCKING,
         })
+    }
     pub fn send(&self, message: &[u8]) -> Result<usize, IoError> {
         let result = unsafe{
             let message_pointer = message.as_ptr().cast();
             libc::send(self.socket_handle, message_pointer, message.len() as libc::size_t, 0)
         };
         if result < 0 {
             return Err(IoError::last_os_error());
+        }
         return Ok(result as usize);
+    }
     /// Receives data from the TCP socket. Returns the number of bytes received.
     /// More bytes may be present even thought `used < buffer.len()`.
     pub fn receive(&self, buffer: &mut [u8]) -> Result<usize, IoError> {
         let result = unsafe {
             let message_pointer = buffer.as_mut_ptr().cast();
             libc::recv(self.socket_handle, message_pointer, buffer.len(), 0)
         };
         if result < 0 {
             return Err(IoError::last_os_error());
+        }
         return Ok(result as usize);
+    }
+}
 impl Drop for SocketTcpClient {
     fn drop(&mut self) {
         debug_assert!(self.socket_handle >= 0);
         unsafe{ close(self.socket_handle) };
+    }
+}
 impl AsFileDescriptor for SocketTcpClient {
     fn as_file_descriptor(&self) -> FileDescriptor {
         return self.socket_handle;
+    }
+}
 /// Raw socket receiver. Essentially a listener that accepts a single connection
 struct SocketRawRx {
     listen_handle: c_int,
     accepted_handle: c_int,
+}
 impl SocketRawRx {
     pub fn new(ip: Option<Ipv4Addr>, port: u16) -> Result<Self, SocketError> {
         let ip = ip.unwrap_or(Ipv4Addr::UNSPECIFIED); // unspecified is the same as INADDR_ANY
         let address = unsafe{ in_addr{
             s_addr: std::mem::transmute(ip.octets()),
         }};
         let socket_address = sockaddr_in{
             sin_family: libc::AF_INET as libc::sa_family_t,
             sin_port: htons(port),
             sin_addr: address,
             sin_zero: [0; 8],
         };
         unsafe {
             let socket_handle = create_and_bind_socket(libc::SOCK_RAW, 0, IpAddr::V4(ip), port)?;
             let result = listen(socket_handle, 3);
             if result < 0 { return Err(SocketError::Listening); }
             return Ok(SocketRawRx{
                 listen_handle: socket_handle,
                 accepted_handle: -1,
             });
+        }
+    }
     // pub fn try_accept(&mut self, timeout_ms: u32) -> Result<(), SocketError> {
     //     if self.accepted_handle >= 0 {
     //         // Already accepted a connection
     //         return Err(SocketError::Accepted);
     //     }
     //
     //     let mut socket_address = sockaddr_in{
     //         sin_family: 0,
     //         sin_port: 0,
     //         sin_addr: in_addr{ s_addr: 0 },
     //         sin_zero: [0; 8]
     //     };
     //     let mut size = size_of::<sockaddr_in>() as u32;
     //     unsafe {
     //         let result = accept(self.listen_handle, &mut socket_address as *mut _, &mut size as *mut _);
     //         if result < 0 {
     //             return Err(SocketError::Accepting);
     //         }
     //     }
     //
     //     return Ok(());
     // }
+}
 impl Drop for SocketRawRx {
     fn drop(&mut self) {
         if self.accepted_handle >= 0 {
             unsafe {
                 close(self.accepted_handle);
+            }
+        }
         if self.listen_handle >= 0 {
             unsafe {
                 close(self.listen_handle);
+            }
+        }
+    }
+}
 // The following is essentially stolen from `mio`'s io_source.rs file.
 #[cfg(unix)]
 trait AsRawFileDescriptor {
     fn as_raw_file_descriptor(&self) -> c_int;
+}
 impl AsRawFileDescriptor for SocketTcpClient {
     fn as_raw_file_descriptor(&self) -> c_int {
         return self.socket_handle;
+    }
+}
 /// Performs the `socket` and `bind` calls.
 fn create_and_bind_socket(socket_type: libc::c_int, protocol: libc::c_int, ip: IpAddr, port: u16) -> Result<libc::c_int, SocketError> {
     let family = socket_family_from_ip(ip);
     unsafe {
         let socket_handle = socket(family, socket_type, protocol);
         if socket_handle < 0 {
             return Err(SocketError::Opening);
+        }
         let result = match ip {
             IpAddr::V4(ip) => {
                 let (socket_address, address_size) = create_sockaddr_in_v4(ip, port);
                 let socket_pointer = &socket_address as *const sockaddr_in;
                 bind(socket_handle, socket_pointer.cast(), address_size)
             },
             IpAddr::V6(ip) => {
                 let (socket_address, address_size) = create_sockaddr_in_v6(ip, port);
                 let socket_pointer= &socket_address as *const sockaddr_in6;
                 bind(socket_handle, socket_pointer.cast(), address_size)
+            }
         };
         if result < 0 {
             close(socket_handle);
             return Err(SocketError::Binding);
+        }
         return Ok(socket_handle);
+    }
+}
 /// Performs the `socket` and `connect` calls
 fn create_and_connect_socket(socket_type: libc::c_int, protocol: libc::c_int, ip: IpAddr, port: u16) -> Result<libc::c_int, SocketError> {
     let family = socket_family_from_ip(ip);
     unsafe {
         let socket_handle = socket(family, socket_type, protocol);
         if socket_handle < 0 {
             return Err(SocketError::Opening);
+        }
         let result = match ip {
             IpAddr::V4(ip) => {
                 let (socket_address, address_size) = create_sockaddr_in_v4(ip, port);
                 let socket_pointer = &socket_address as *const sockaddr_in;
                 connect(socket_handle, socket_pointer.cast(), address_size)
             },
             IpAddr::V6(ip) => {
                 let (socket_address, address_size) = create_sockaddr_in_v6(ip, port);
                 let socket_pointer= &socket_address as *const sockaddr_in6;
                 connect(socket_handle, socket_pointer.cast(), address_size)
+            }
         };
         if result < 0 {
             close(socket_handle);
             return Err(SocketError::Connecting);
+        }
         return Ok(socket_handle);
+    }
+}
 #[inline]
 fn create_sockaddr_in_v4(ip: Ipv4Addr, port: u16) -> (sockaddr_in, libc::socklen_t) {
     let address = unsafe{
         in_addr{
             s_addr: std::mem::transmute(ip.octets())
+        }
     };
     let socket_address = sockaddr_in{
         sin_family: libc::AF_INET as libc::sa_family_t,
         sin_port: htons(port),
         sin_addr: address,
         sin_zero: [0; 8]
     };
     let address_size = size_of::<sockaddr_in>();
     return (socket_address, address_size as _);
+}
 #[inline]
 fn create_sockaddr_in_v6(ip: Ipv6Addr, port: u16) -> (sockaddr_in6, libc::socklen_t) {
     // flow label is advised to be, according to RFC6437 a (somewhat
     // secure) random number taken from a uniform distribution
     let flow_info = rand::random();
     let address = unsafe{
         in6_addr{
             s6_addr: ip.octets()
+        }
     };
     let socket_address = sockaddr_in6{
         sin6_family: libc::AF_INET6 as libc::sa_family_t,
         sin6_port: htons(port),
         sin6_flowinfo: flow_info,
         sin6_addr: address,
         sin6_scope_id: 0, // incorrect in case of loopback address
     };
     let address_size = size_of::<sockaddr_in6>();
     return (socket_address, address_size as _);
+}
 #[inline]
 fn set_socket_blocking(handle: libc::c_int, blocking: bool) -> bool {
     if handle < 0 {
         return false;
+    }
     unsafe{
         let mut flags = libc::fcntl(handle, libc::F_GETFL, 0);
         if flags < 0 {
             return false;
+        }
         if blocking {
             flags &= !libc::O_NONBLOCK;
         } else {
             flags |= libc::O_NONBLOCK;
+        }
         let result = libc::fcntl(handle, libc::F_SETFL, flags);
         if result < 0 {
             return false;
+        }
+    }
     return true;
+}
 #[inline]
 fn socket_family_from_ip(ip: IpAddr) -> libc::c_int {
     return match ip {
         IpAddr::V4(_) => libc::AF_INET,
         IpAddr::V6(_) => libc::AF_INET6,
     };
+}
 #[inline]
 fn htons(port: u16) -> u16 {
     return port.to_be();
+}
 mod tests {
     use std::net::*;
     use super::*;
     // #[test] @nocommit Remove this
     // fn test_inet_thingo() {
     //     const SIZE: usize = 1024;
     //
     //     let s = SocketTcpClient::new(IpAddr::V4(Ipv4Addr::new(142, 250, 179, 163)), 80).expect("connect");
     //     s.send(b"GET / HTTP/1.1\r\n\r\n").expect("sending");
     //     let mut total = Vec::<u8>::new();
     //     let mut buffer = [0; SIZE];
     //     let mut received = SIZE;
     //
     //     while received > 0 {
     //         received = s.receive(&mut buffer).expect("receiving");
     //         println!("DEBUG: Received {} bytes", received);
     //         total.extend_from_slice(&buffer[..received]);
     //     }
     //     let as_str = String::from_utf8_lossy(total.as_slice());
     //     println!("Yay! Got {} bytes:\n{}", as_str.len(), as_str);
     // }
+}
@@ \ No newline at end of file @@

src/runtime2/tests/mod.rs

➞

Show inline comments

@@ @@ -57,193 +57,193 @@ fn test_component_communication() { @@
+            }
             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).unwrap();
     create_component(&rt, "", "constructor", no_args());
+}
 #[test]
 fn test_intermediate_messenger() {
     let pd = ProtocolDescription::parse(b"
     primitive receiver<T>(in<T> rx, u32 num) {
         auto index = 0;
         while (index < num) {
             sync { auto v = get(rx); }
             index += 1;
+        }
+    }
     primitive middleman<T>(in<T> rx, out<T> tx, u32 num) {
         auto index = 0;
         while (index < num) {
             sync { put(tx, get(rx)); }
             index += 1;
+        }
+    }
     primitive sender<T>(out<T> tx, u32 num) {
         auto index = 0;
         while (index < num) {
             sync put(tx, 1337);
             index += 1;
+        }
+    }
     composite constructor_template<T>() {
         auto num = 0;
         channel<T> tx_a -> rx_a;
         channel tx_b -> rx_b;
         new sender(tx_a, 3);
         new middleman(rx_a, tx_b, 3);
         new receiver(rx_b, 3);
+    }
     composite constructor() {
         new constructor_template<u16>();
         new constructor_template<u32>();
         new constructor_template<u64>();
         new constructor_template<s16>();
         new constructor_template<s32>();
         new constructor_template<s64>();
+    }
     ").expect("compilation");
     let rt = Runtime::new(3, true, pd).unwrap();
     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!\"); }
         return ();
+    }
     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\");
                     auto _ = infinite_assert(v, num_from_a);
                     num_from_a += 1;
+                }
                 auto v = get(in_b) -> {
                     print(\"got something from B\");
                     auto _ = 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 = 1;
         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(3, true, pd).unwrap();
     create_component(&rt, "", "constructor", no_args());
+}
 #[test]
 fn test_unguarded_select() {
     let pd = ProtocolDescription::parse(b"
     primitive constructor_outside_select() {
         u32 index = 0;
         while (index < 5) {
             sync select { auto v = () -> print(\"hello\"); }
             index += 1;
+        }
+    }
     primitive constructor_inside_select() {
         u32 index = 0;
         while (index < 5) {
             sync select { auto v = () -> index += 1; }
+        }
+    }
     ").expect("compilation");
     let rt = Runtime::new(3, false, pd).unwrap();
     create_component(&rt, "", "constructor_outside_select", no_args());
     create_component(&rt, "", "constructor_inside_select", no_args());
+}
 #[test]
 fn test_empty_select() {
     let pd = ProtocolDescription::parse(b"
     primitive constructor() {
         u32 index = 0;
         while (index < 5) {
             sync select {}
             index += 1;
+        }
+    }
     ").expect("compilation");
     let rt = Runtime::new(3, false, pd).unwrap();
     create_component(&rt, "", "constructor", no_args());
+}
 #[test]
 fn test_random_u32_temporary_thingo() {
     let pd = ProtocolDescription::parse(b"
     import std.random::random_u32;
     primitive random_taker(in<u32> generator, u32 num_values) {
         auto i = 0;
         while (i < num_values) {
             sync {
                 auto a = get(generator);
+            }
             i += 1;
+        }
+    }
     composite constructor() {
         channel tx -> rx;
         auto num_values = 25;
         new random_u32(tx, 1, 100, num_values);
         new random_taker(rx, num_values);
+    }
     ").expect("compilation");
     let rt = Runtime::new(1, true, pd).unwrap();
     create_component(&rt, "", "constructor", no_args());
+}
@@ \ No newline at end of file @@
+}

std/std.internet.pdl

➞

Show inline comments

 #module std.internet
 union Cmd {
     Send(u8[]),
     Receive,
     Finish,
     Shutdown,
+}
 primitive tcp_client(u8[] ip, u16 port, in<Cmd> cmds, out<u8[]> rx) {
     #builtin
+}
@@ \ No newline at end of file @@

0 comments (0 inline, 0 general)