[package]

name = "reowolf_rs"

version = "1.2.0"

authors = [

	"Max Henger <henger@cwi.nl>",

	"Christopher Esterhuyse <esterhuy@cwi.nl>",

	"Hans-Dieter Hiep <hdh@cwi.nl>"

]

edition = "2021"

[features]

default=["internet"]

internet=["libc"]

[dependencies]

# randomness

rand = "0.8.4"

rand_pcg = "0.3.1"

[lib]

crate-type = [

	"rlib", # for use as a Rust dependency.

]

1. Port IDs are decided locally. So a peer may have an ID that is outdated for the intended recipient.

2. Ports can move from owner to owner. So a peer might have a component ID that is outdated for the intended recipient.

3. Ports may be closed.

4. Message intended for specific ports may end up at an intermediate component that is passing that message along.

However, there are some properties that we can take advantage of:

1. When a component sends a message, it knows for certain what its own component ID and port ID is. So a transmitting port always sends the correct source component/port ID.

2. When a component receives a message, it knows for certain what its own component ID and port ID is. So once a receiving port receives a properly annotated message from a transmitted port, the receiving end can be certain about the component IDs and port IDs that make up the channel.

Note that although the message transmitter may not be certain about its final recipient, the components along the way *are* aware of the routing that is necessary to make the message arrive at the intended target. Combing back to the case where we have a creator `C`, new component `N` and peer `P`. Then `P` will send a message intended for `N`, but arriving at `C`. Here `C` can change the target port and component to `N` (as it is in the process of transferring that port, so knows both its original and new port ID). Once the message arrives and is accepted by the recipient then it is certain about the component and port IDs.

## Sending Sync Messages

Sync messages have the purpose of making sure that consensus is reached about the interactions that took place in all of the participating components' sync blocks. The previous runtime featured speculative execution and a branching execution model: a component could exhibit multiple behaviours, and at the end all components decide which combination of local behaviours constitute a satisfying single global behaviour (if one exists at all). Without speculative execution the model can be a lot simpler.

We'll only shortly discuss the mechanisms that are present in the synchronization algorithm. A component has a local counter, that is reset for each synchronous interaction, that is used when transmitting data messages. Such a message will be annotated with the counter at `N`, after which the component sends the next message with annotation `N+1`. At the same time the component will keep track of a local mapping from port ID to port annotation, we'll call this the port mapping. Likewise, when a component receives a data message it will assign the received annotation in its own port mapping. If two components assign the same annotation to the ports that constitute a channel, then there is an agreeable interaction there.

And so at the end of the synchronous round a component will somehow broadcast its port mapping. Note from the discussion above that a transmitting port's annotation is only associated with that transmitting port, since a transmitting port can only truly ever know its own component/port ID. While the receiving port's annotation knows about the peer's component/port ID as well. And so a component can broadcast `(component ID, port ID, mapping)` for each of its transmitting ports, while it can broadcast `(own component ID, own port ID, peer component ID, peer port ID, mapping)` for each receiving port. Then a recipient of these mappings can match them up and make sure that the mappings agree.

Note that this broadcasting of synchronous messages is essentially a component-to-component operation. However these messages must still be sent over ports anyway (and any port that was used to transmit messages to a particular receiving component will do). There are two reasons:

1. The sync message may need to be rerouted (e.g. a sender quickly fires both a data message and a subsequent sync message while the receiving port is being transferred to a new component), but needs to arrive at the same target as the data message. This is essentially restating that a transmitter never knows about the component ID of the recipient.

2. The sync message must not be taken into account by the recipient if it has not accepted any messages from the sender yet. Ofcourse this can be achieved in various ways but a simple way to achieve this is to send the sync message over ports.

## Annotating Data Messages

These port mappings are also sent along when sending data messages. We will not go into details but here the mapping makes sure that messages arrive in the right order, and certain kinds of deadlock or inconsistent protocol behaviour may be detected. This port mapping is checked for consistency by the recipient and, when consistent, the target port is updated with its new mapping.

As we'll send along this mapping we will only consider the ports that are shared between the two components. But in the most general case the transmitting ports of the component do not have knowledge about the peer component. And so the sent port mapping will have to contain the annotation for *all* transmitting ports. Receiving port mappings only have to be sent along if they received a message, and here we can indeed apply filtering. Likewise, if the recipient of a port mapping has not yet received anything on its receiving port, then it cannot be sure about the identity of the sender.

This leads to problems both for ensuring the correct ordering of the messages. For finding consensus it is not. Suppose that a port `a` sends a message to a port `b`. Port `b` does not accept it. Upon trying to find consensus we see that `a` will submit an entry in its port mapping, and `b` does not submit anything at all. Hence no solution can be found, as desired.

For the message ordering we require from the receiver that it confirms that, for all of the shared channels, it has the same mapping as the sender sends along. Suppose a component `A` has ports `a_put` and `b_put`, while a component `B` has ports `a_get` and `b_get` (where `a_put -> a_get` and `b_put -> b_get`). Suppose component `A` sends on `a_put` and `b_put` consecutively. And component `B` only receives from `b_get`. Then since `a_get` did not receive from `a_put` (hence did not learn that component/port ID pair of `a_put` is associated with `a_get`), the component `B` cannot figure out that `a_get` should precede a `b_get`. Likewise component `A` has no way to know that `a_put` and `b_put` are sending to the same component, hence it cannot indicate to component `B` that `a_get` should precede `b_get`.

There are some simple assumptions we can make that makes the problem a little bit easier to think about. Suppose again `a_put -> a_get` and `b_put -> b_get`. Suppose `a_put` is used first, where we send along the mapping of `a_put` and `b_put`. Then we send along `b_put`, again sending along the mapping. Then it is possible for the receiving component to accept the wrong message first (e.g. `b_get`, therefore learning about `b`), but it will be impossible to get from `a_get` later, since that one requires `b_put` (of which we learned that it matches `b_get`) to not have sent any messages.

Without adding any extra overhead (e.g. some kind of discovery round per synchronous interaction), we can take three approaches:

1. We simply don't care. It is impossible for a round where messages are received out of order to complete. Hence we temporarily allow a component to take the wrong actions, therefore wasting some CPU time, and to crash/error afterward.

2. We remove the entire concept of ordering of channels at a single component. Channels are always independent entities. This way we truly do not have to care. All we care about is that the messages that have been sent over a channel arrive at the other side.

3. We slightly modify the algorithm to detect these problems. This can be done in reasonable fashion, albeit a bit "hacky". For each channel there is a slot to receive messages. Messages wait there until the receiver performs a `get` in the PDL code. So far we've only considered learning about the component/port IDs that constitute a channel the moment they're received with a `get`. The algorithm could be changed to already learn about the peer component/port ID the moment the message arrives in the slot.

We'll go with the last option in the current implementation. We return to the problematic example above. Note that messages between components are sent in ordered fashion, and `a_put` happens before `b_put`. Then component `B` will first learn that `a_put` is the peer of `a_get`, then it performs the first `get` on the message from `b_put` to `b_get`. This message is annotated with a port mapping that `a_put` has been used before. We're now able to detect at component `B` that we cannot accept `b_get` before `a_get`.

Concluding:

- Every data message that is transmitted needs to contain the port mapping of all `put`ting ports (annotating them appropriately if they have not yet been used). We also need to include the port mapping of all `get`ting ports that have a pending/received message. The port mapping for `put`ting ports will only include their own ID, the port mapping for `get`ting ports will include the IDs of their peer as well.

                    },

                    Literal::Union(literal) => {

                        for value_expr_id in &literal.values {

                            self.expr_stack.push_back(ExprInstruction::PushValToFront);

                            self.serialize_expression(heap, *value_expr_id);

                        }

                    },

                    Literal::Array(value_expr_ids) => {

                        for value_expr_id in value_expr_ids {

                            self.expr_stack.push_back(ExprInstruction::PushValToFront);

                            self.serialize_expression(heap, *value_expr_id);

                        }

                    },

                    Literal::Tuple(value_expr_ids) => {

                        for value_expr_id in value_expr_ids {

                            self.expr_stack.push_back(ExprInstruction::PushValToFront);

                            self.serialize_expression(heap, *value_expr_id);

                        }

                    }

                }

            },

            Expression::Cast(expr) => {

                self.serialize_expression(heap, expr.subject);

            }

            Expression::Call(expr) => {

                for arg_expr_id in &expr.arguments {

                    self.expr_stack.push_back(ExprInstruction::PushValToFront);

                    self.serialize_expression(heap, *arg_expr_id);

                }

            },

            Expression::Variable(_expr) => {

                // No subexpressions

            }

        }

    }

}

pub type EvalResult = Result<EvalContinuation, EvalError>;

#[derive(Debug)]

pub enum EvalContinuation {

    // Returned in both sync and non-sync modes

    Stepping,

    // Returned only in sync mode

    BranchInconsistent,

    SyncBlockEnd,

    NewFork,

    BlockFires(PortId),

    SelectStart(u32, u32), // (num_cases, num_ports_total)

    SelectWait, // wait until select can continue

    // Returned only in non-sync mode

    ComponentTerminated,

    SyncBlockStart,

    NewComponent(ProcedureDefinitionId, TypeId, ValueGroup),

    NewChannel,

}

// Note: cloning is fine, methinks. cloning all values and the heap regions then

// we end up with valid "pointers" to heap regions.

#[derive(Debug, Clone)]

pub struct Prompt {

    pub(crate) frames: Vec<Frame>,

    pub(crate) store: Store,

}

impl Prompt {

    pub fn new(types: &TypeTable, heap: &Heap, def: ProcedureDefinitionId, type_id: TypeId, args: ValueGroup) -> Self {

        let mut prompt = Self{

            frames: Vec::new(),

            store: Store::new(),

        };

        // Maybe do typechecking in the future?

        let monomorph_index = types.get_monomorph(type_id).variant.as_procedure().monomorph_index;

        let new_frame = Frame::new(heap, def, type_id, monomorph_index);

        let max_stack_size = new_frame.max_stack_size;

        prompt.frames.push(new_frame);

        args.into_store(&mut prompt.store);

        prompt.store.reserve_stack(max_stack_size);

        prompt

    }

    /// Big 'ol function right here. Didn't want to break it up unnecessarily.

    /// It consists of, in sequence: executing any expressions that should be

    /// executed before the next statement can be evaluated, then a section that

    /// performs debug printing, and finally a section that takes the next

    /// statement and executes it. If the statement requires any expressions to

    /// be evaluated, then they will be added such that the next time `step` is

    /// called, all of these expressions are indeed evaluated.

    pub(crate) fn step(&mut self, types: &TypeTable, heap: &Heap, modules: &[Module], ctx: &mut impl RunContext) -> EvalResult {

        // Helper function to transfer multiple values from the expression value

        // array into a heap region (e.g. constructing arrays or structs).

        fn transfer_expression_values_front_into_heap(cur_frame: &mut Frame, store: &mut Store, num_values: usize) -> HeapPos {

            let heap_pos = store.alloc_heap();

            // Do the transformation first (because Rust...)

                            };

                            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));

                                        }

                                    }

                                },

                                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]);

                                    }

                                },

                                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 = self.store.maybe_read_ref(&value);

                                    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

                                    if is_literal_string {

                                        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();

                                },

                                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::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;

        let mut call_id_section = self.call_expr_buffer.start_section();

        let mut expr_id_section = self.expression_buffer.start_section();

        for case in select_stmt.cases.iter() {

            total_num_ports += case.involved_ports.len();

            for (call_id, expr_id) in case.involved_ports.iter().copied() {

                call_id_section.push(call_id);

                expr_id_section.push(expr_id);

            }

        }

        // Transform all of the call expressions by takings its argument (the

        // port from which we `get`) and turning it into a temporary variable.

        let mut transformed_stmts = Vec::with_capacity(total_num_ports); // TODO: Recompute this preallocated length, put assert at the end

        let mut locals = Vec::with_capacity(total_num_ports);

        for port_var_idx in 0..call_id_section.len() {

            let get_call_expr_id = call_id_section[port_var_idx];

            let port_expr_id = expr_id_section[port_var_idx];

            let port_type_index = ctx.heap[port_expr_id].type_index();

            let port_type_ref = TypeIdReference::IndirectSameAsExpr(port_type_index);

            // Move the port expression such that it gets assigned to a temporary variable

            let variable_id = create_ast_variable(ctx, outer_scope_id);

            let variable_decl_stmt_id = create_ast_variable_declaration_stmt(ctx, self.current_procedure_id, variable_id, port_type_ref, port_expr_id);

            // Replace the original port expression in the call with a reference

            // to the replacement variable

            let variable_expr_id = create_ast_variable_expr(ctx, self.current_procedure_id, variable_id, port_type_ref);

            let call_expr = &mut ctx.heap[get_call_expr_id];

            call_expr.arguments[0] = variable_expr_id.upcast();

            transformed_stmts.push(variable_decl_stmt_id.upcast().upcast());

            locals.push((variable_id, port_type_ref));

        }

        // Insert runtime calls that facilitate the semantics of the select

        // block.

        // Create the call that indicates the start of the select block

        {

            let num_cases_expression_id = create_ast_literal_integer_expr(ctx, self.current_procedure_id, total_num_cases as u64, ctx.arch.uint32_type_id);

            let num_ports_expression_id = create_ast_literal_integer_expr(ctx, self.current_procedure_id, total_num_ports as u64, ctx.arch.uint32_type_id);

            let arguments = vec![

                num_cases_expression_id.upcast(),

                num_ports_expression_id.upcast()

            ];

            let call_statement_id = create_ast_expression_stmt(ctx, call_expression_id.upcast());

            transformed_stmts.push(call_statement_id.upcast());

        }

        // Create calls for each select case that will register the ports that

        // we are waiting on at the runtime.

        {

            let mut total_port_index = 0;

            for case_index in 0..total_num_cases {

                let case = &ctx.heap[id].cases[case_index];

                let case_num_ports = case.involved_ports.len();

                for case_port_index in 0..case_num_ports {

                    // Arguments to runtime call

                    let (port_variable_id, port_variable_type) = locals[total_port_index]; // so far this variable contains the temporary variables for the port expressions

                    let case_index_expr_id = create_ast_literal_integer_expr(ctx, self.current_procedure_id, case_index as u64, ctx.arch.uint32_type_id);

                    let port_index_expr_id = create_ast_literal_integer_expr(ctx, self.current_procedure_id, case_port_index as u64, ctx.arch.uint32_type_id);

                    let port_variable_expr_id = create_ast_variable_expr(ctx, self.current_procedure_id, port_variable_id, port_variable_type);

                    let runtime_call_arguments = vec![

                        case_index_expr_id.upcast(),

                        port_index_expr_id.upcast(),

                        port_variable_expr_id.upcast()

                    ];

                    // Create runtime call, then store it

                    let runtime_call_stmt_id = create_ast_expression_stmt(ctx, runtime_call_expr_id.upcast());

                    transformed_stmts.push(runtime_call_stmt_id.upcast());

                    total_port_index += 1;

                }

            }

        }

        // Create the variable that will hold the result of a completed select

        // block. Then create the runtime call that will produce this result

        let select_variable_id = create_ast_variable(ctx, outer_scope_id);

        let select_variable_type = TypeIdReference::DirectTypeId(ctx.arch.uint32_type_id);

        locals.push((select_variable_id, select_variable_type));

        {

            let variable_stmt_id = create_ast_variable_declaration_stmt(ctx, self.current_procedure_id, select_variable_id, select_variable_type, runtime_call_expr_id.upcast());

            transformed_stmts.push(variable_stmt_id.upcast().upcast());

        }

        call_id_section.forget();

        expr_id_section.forget();

        // Now we transform each of the select block case's guard and code into

        // a chained if-else statement.

        let relative_pos = transformed_stmts.len() as i32;

        if total_num_cases > 0 {

            let (if_stmt_id, end_if_stmt_id, scope_id) = transform_select_case_code(ctx, self.current_procedure_id, id, 0, select_variable_id, select_variable_type);

            link_existing_child_to_new_parent_scope(ctx, &mut self.scope_buffer, outer_scope_id, scope_id, relative_pos);

            let first_end_if_stmt = &mut ctx.heap[end_if_stmt_id];

            first_end_if_stmt.next = outer_end_block_id.upcast();

            let mut last_if_stmt_id = if_stmt_id;

            let mut last_end_if_stmt_id = end_if_stmt_id;

            let mut last_parent_scope_id = outer_scope_id;

            let mut last_relative_pos = transformed_stmts.len() as i32 + 1;

            transformed_stmts.push(last_if_stmt_id.upcast());

            for case_index in 1..total_num_cases {

                let (if_stmt_id, end_if_stmt_id, scope_id) = transform_select_case_code(ctx, self.current_procedure_id, id, case_index, select_variable_id, select_variable_type);

                let false_case_scope_id = ctx.heap.alloc_scope(|this| Scope::new(this, ScopeAssociation::If(last_if_stmt_id, false)));

                link_existing_child_to_new_parent_scope(ctx, &mut self.scope_buffer, false_case_scope_id, scope_id, 0);

                link_new_child_to_existing_parent_scope(ctx, &mut self.scope_buffer, last_parent_scope_id, false_case_scope_id, last_relative_pos);

                set_ast_if_statement_false_body(ctx, last_if_stmt_id, last_end_if_stmt_id, IfStatementCase{ body: if_stmt_id.upcast(), scope: false_case_scope_id });

                let end_if_stmt = &mut ctx.heap[end_if_stmt_id];

                end_if_stmt.next = last_end_if_stmt_id.upcast();

                last_if_stmt_id = if_stmt_id;

                last_end_if_stmt_id = end_if_stmt_id;

                last_parent_scope_id = false_case_scope_id;

                last_relative_pos = 0;

            }

        }

        // Final steps: set the statements of the replacement block statement,

        // link all of those statements together, and update the scopes.

        let first_stmt_id = transformed_stmts[0];

        let mut last_stmt_id = transformed_stmts[0];

        for stmt_id in transformed_stmts.iter().skip(1).copied() {

            set_ast_statement_next(ctx, last_stmt_id, stmt_id);

            last_stmt_id = stmt_id;

        }

        let outer_block_stmt = &mut ctx.heap[outer_block_id];

        outer_block_stmt.next = first_stmt_id;

        outer_block_stmt.statements = transformed_stmts;

        return Ok(())

    }

}

// -----------------------------------------------------------------------------

// Utilities to create compiler-generated AST nodes

// -----------------------------------------------------------------------------

#[derive(Clone, Copy)]

enum TypeIdReference {

    DirectTypeId(TypeId),

    IndirectSameAsExpr(i32), // by type index

}

fn create_ast_variable(ctx: &mut Ctx, scope_id: ScopeId) -> VariableId {

    let variable_id = ctx.heap.alloc_variable(|this| Variable{

        this,

        kind: VariableKind::Local,

        parser_type: ParserType{

            elements: Vec::new(),

            full_span: InputSpan::new(),

        },

        identifier: Identifier::new_empty(InputSpan::new()),

        relative_pos_in_parent: -1,

        unique_id_in_scope: -1,

    });

    let scope = &mut ctx.heap[scope_id];

    scope.variables.push(variable_id);

    return variable_id;

}

fn create_ast_variable_expr(ctx: &mut Ctx, containing_procedure_id: ProcedureDefinitionId, variable_id: VariableId, variable_type_id: TypeIdReference) -> VariableExpressionId {

    let variable_type_index = add_new_procedure_expression_type(ctx, containing_procedure_id, variable_type_id);

    return ctx.heap.alloc_variable_expression(|this| VariableExpression{

        this,

        identifier: Identifier::new_empty(InputSpan::new()),

        declaration: Some(variable_id),

        used_as_binding_target: false,

        parent: ExpressionParent::None,

        type_index: variable_type_index,

    });

}

    let call_type_id = match method {

        Method::SelectStart => ctx.arch.void_type_id,

        Method::SelectRegisterCasePort => ctx.arch.void_type_id,

        Method::SelectWait => ctx.arch.uint32_type_id, // TODO: Not pretty, this. Pretty error prone

        _ => unreachable!(), // if this goes of, add the appropriate method here.

    };

    let expression_ids = buffer.start_section_initialized(&arguments);

    let call_type_index = add_new_procedure_expression_type(ctx, containing_procedure_id, TypeIdReference::DirectTypeId(call_type_id));

    let call_expression_id = ctx.heap.alloc_call_expression(|this| CallExpression{

        func_span: InputSpan::new(),

        this,

        parser_type: ParserType{

            elements: Vec::new(),

            full_span: InputSpan::new(),

        },

        method,

        arguments,

        procedure: ProcedureDefinitionId::new_invalid(),

        parent: ExpressionParent::None,

        type_index: call_type_index,

    });

    for argument_index in 0..expression_ids.len() {

        let argument_id = expression_ids[argument_index];

        let argument_expr = &mut ctx.heap[argument_id];

        *argument_expr.parent_mut() = ExpressionParent::Expression(call_expression_id.upcast(), argument_index as u32);

    }

    return call_expression_id;

}

fn create_ast_literal_integer_expr(ctx: &mut Ctx, containing_procedure_id: ProcedureDefinitionId, unsigned_value: u64, type_id: TypeId) -> LiteralExpressionId {

    let literal_type_index = add_new_procedure_expression_type(ctx, containing_procedure_id, TypeIdReference::DirectTypeId(type_id));

    return ctx.heap.alloc_literal_expression(|this| LiteralExpression{

        this,

        span: InputSpan::new(),

        value: Literal::Integer(LiteralInteger{

            unsigned_value,

            negated: false,

        }),

        parent: ExpressionParent::None,

        type_index: literal_type_index,

    });

}

fn create_ast_equality_comparison_expr(

    ctx: &mut Ctx, containing_procedure_id: ProcedureDefinitionId,

    variable_id: VariableId, variable_type: TypeIdReference, value: u64

) -> BinaryExpressionId {

    let var_expr_id = create_ast_variable_expr(ctx, containing_procedure_id, variable_id, variable_type);

    let int_expr_id = create_ast_literal_integer_expr(ctx, containing_procedure_id, value, ctx.arch.uint32_type_id);

    let cmp_type_index = add_new_procedure_expression_type(ctx, containing_procedure_id, TypeIdReference::DirectTypeId(ctx.arch.bool_type_id));

    let cmp_expr_id = ctx.heap.alloc_binary_expression(|this| BinaryExpression{

        this,

        operator_span: InputSpan::new(),

        full_span: InputSpan::new(),

        left: var_expr_id.upcast(),

        operation: BinaryOperator::Equality,

        right: int_expr_id.upcast(),

        let run_result = Self::run_prompt(&mut branch.code_state, &sched_ctx.runtime.protocol_description, &mut run_context);

        if let Err(eval_error) = run_result {

            self.eval_error = Some(eval_error);

            self.mode = Mode::SyncError;

            if let Some(conclusion) = self.consensus.notify_of_fatal_branch(branch_id, comp_ctx) {

                // We can exit immediately

                return self.enter_non_sync_mode(conclusion, comp_ctx);

            } else {

                // Current branch failed. But we may have other things that are

                // running.

                return ConnectorScheduling::Immediate;

            }

        }

        let run_result = run_result.unwrap();

        // Handle the returned result. Note that this match statement contains

        // explicit returns in case the run result requires that the component's

        // code is ran again immediately

        match run_result {

            EvalContinuation::BranchInconsistent => {

                // Branch became inconsistent

                branch.sync_state = SpeculativeState::Inconsistent;

            },

            EvalContinuation::BlockFires(port_id) => {

                // Branch called `fires()` on a port that has not been used yet.

                let port_id = PortIdLocal::new(port_id.id);

                // Create two forks, one that assumes the port will fire, and

                // one that assumes the port remains silent

                branch.sync_state = SpeculativeState::HaltedAtBranchPoint;

                let firing_branch_id = self.tree.fork_branch(branch_id);

                let silent_branch_id = self.tree.fork_branch(branch_id);

                self.consensus.notify_of_new_branch(branch_id, firing_branch_id);

                let _result = self.consensus.notify_of_speculative_mapping(firing_branch_id, port_id, true, comp_ctx);

                debug_assert_eq!(_result, Consistency::Valid);

                self.consensus.notify_of_new_branch(branch_id, silent_branch_id);

                let _result = self.consensus.notify_of_speculative_mapping(silent_branch_id, port_id, false, comp_ctx);

                debug_assert_eq!(_result, Consistency::Valid);

                // Somewhat important: we push the firing one first, such that

                // that branch is ran again immediately.

                self.tree.push_into_queue(QueueKind::Runnable, firing_branch_id);

                self.tree.push_into_queue(QueueKind::Runnable, silent_branch_id);

                return ConnectorScheduling::Immediate;

            },

                // Branch performed a `get()` on a port that does not have a

                // received message on that port.

                let port_id = PortIdLocal::new(port_id.id);

                branch.sync_state = SpeculativeState::HaltedAtBranchPoint;

                branch.awaiting_port = port_id;

                self.tree.push_into_queue(QueueKind::AwaitingMessage, branch_id);

                // Note: we only know that a branch is waiting on a message when

                // it reaches the `get` call. But we might have already received

                // a message that targets this branch, so check now.

                let mut any_message_received = false;

                for message in comp_ctx.get_read_data_messages(port_id) {

                    if self.consensus.branch_can_receive(branch_id, &message) {

                        // This branch can receive the message, so we do the

                        // fork-and-receive dance

                        let receiving_branch_id = self.tree.fork_branch(branch_id);

                        let branch = &mut self.tree[receiving_branch_id];

                        branch.awaiting_port = PortIdLocal::new_invalid();

                        branch.prepared = PreparedStatement::PerformedGet(message.content.clone());

                        self.consensus.notify_of_new_branch(branch_id, receiving_branch_id);

                        self.consensus.notify_of_received_message(receiving_branch_id, &message, comp_ctx);

                        self.tree.push_into_queue(QueueKind::Runnable, receiving_branch_id);

                        any_message_received = true;

                    }

                }

                if any_message_received {

                    return ConnectorScheduling::Immediate;

                }

            }

            EvalContinuation::SyncBlockEnd => {

                let consistency = self.consensus.notify_of_finished_branch(branch_id);

                if consistency == Consistency::Valid {

                    branch.sync_state = SpeculativeState::ReachedSyncEnd;

                    self.tree.push_into_queue(QueueKind::FinishedSync, branch_id);

                } else {

                    branch.sync_state = SpeculativeState::Inconsistent;

                }

            },

            EvalContinuation::NewFork => {

                // Like the `NewChannel` result. This means we're setting up

                // a branch and putting a marker inside the RunContext for the

                // next time we run the PDL code

                let left_id = branch_id;

                let right_id = self.tree.fork_branch(left_id);

                self.consensus.notify_of_new_branch(left_id, right_id);

                self.tree.push_into_queue(QueueKind::Runnable, left_id);

                self.tree.push_into_queue(QueueKind::Runnable, right_id);

                let left_branch = &mut self.tree[left_id];

                left_branch.prepared = PreparedStatement::ForkedExecution(true);

                let right_branch = &mut self.tree[right_id];

                right_branch.prepared = PreparedStatement::ForkedExecution(false);

            }

                // Branch is attempting to send data

                let port_id = PortIdLocal::new(port_id.id);