[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.

]

# Synchronous Communication and Component Orchestration

## Introduction

The Reowolf runtime consists of a system that allows multiple components to run within their own thread of execution. These components are able to exchange messages with one another. Components are capable of creating other components, and of creating channels. We may visualise the runtime as a cloud of all kinds of components, connected by the communication channels between them, hence a kind of communication graph.

With this version of the runtime there were several main drivers. For performance reasons we want:

- As little centralized information as possible (because centralization of information implies synchronization to access it).

- As much parallelism as possible (when information *must* be exchanged, then make sure as little components are affected as possible).

To keep the complexity of the runtime to a reasonable minimum, the following requirements have to be met as well:

- A component may terminate, therefore potentially not being able to participate in synchronous communication or receive messages. Experimentation showed that the system that ensures that termination is broadcast to all peers should be kept simple (earlier systems depended on a state-machine with assumptions about the order in which messages were exchanged, which greatly complicates the messaging subsystem of the runtime).

- Messages should arrive in order (with some exceptions). As we'll see later in this document we have different types of messages. Reasoning about a component's operational state becomes much simpler if we can assume that the transmission of messages between components is ordered.

As we will see there are several types of messages that can be exchanged. Among them we have:

- Data messages: these messages contain the data that is "transmitted from and to PDL code". For each `put` a data message is annotated by the runtime and sent along to the receiving component, which will then hopefully retrieve the data with a `get`. These messages are conceptually sent over channels.

- Sync messages: these messages are sent between components to communicate their consensus-state. These messages are not necessarily associated with channels.

- Control messages: these messages are sent between components to ensure that the entire runtime is reliably facilitating data exchange. That is: they ensure that the language is working as intended. As an example: sending a port to a different component requires a bit of bookkeeping to ensure that all involved components are aware of the port exchange.

The remainder of this document tries to describe the various technical aspects of synchronous communication and component orchestration.

## Brief Description of Schedulers

Each component conceptually has its own thread of execution. It is executing a program with at any given point in time a particular memory state. In reality there are a limited number of threads that execute components. Making sure that components are scheduled correctly is based on the fact that components are generally executing programs that are blocked at some point: a message needs to be received, or a port is blocked so we cannot send any information. At that point a component is considered "sleeping". Should another component, scheduled on a particular thread, send a message to this sleeping component, then it is "woken up" by putting it into the execution queue.

The job of the scheduler is then to: execute a component scheduled for execution, wait until a component is scheduled, or shut down in case there are no more components to execute.

The details of the execution queue (currently rather simplistically implemented) is not of interest. What is of interest is that a component can only be scheduled once.

## Creation of Channels

Within PDL code it is possible to create new channels. And so a component will always (that is to say: for now) create both endpoints of the channel, hence own both endpoints of the channel upon creation. Identifiers for these ports are generated locally (we don't want to resolve to having to synchronize on some kind of global port ID counter).

As these IDs are generated locally there is no real technical challenge, but note that ports at different components may have the same port ID.

## Creation of Components

Within PDL code it is possible to create components. Upon their creation they can be given endpoints of channels. Hence at component creation we are changing the configuration of the communication graph. All of the relevant components need to be informed about the port changing ownership.

Here we run into a plethora of problems. The other endpoint might have been given away to another created component. The other endpoint may have already been used in communication, such that we already have messages underway for the port we're trying to give to a newly created component. We may have that the local port ID assigned by the creating component is not the same as the local port ID that the newly created component may want to assign to it. We may have that this port has been passed along multiple times already, etc.

We cannot help that messages have already arrived, or are in transit, for the transferred port. But we can make some assumptions that simplify the transfer of ports. As a first one, we have that the creating component decides when the created component is scheduled for execution. We'll choose not to execute it initially, such that we can be sure that it will not send messages over its ports the moment is created. To further simplify the problem, we have assumed that messages arrive in order. So although messages might still be underway for the transferred ports, if we ask the sender to stop sending, and the sender blocks the port and acknowledges that it has received this command. Then the moment the creator receives the acknowledgement it is certain that it has received all messages intended for the transferred ports. We'll also disallow the creation of components during synchronous interactions.

And so here we have our first control protocol. If a port is transferred then we might have:

1. That the peer port is transferred to the new component as well. All is fine and we can take care of the exchange immediately.

2. That the peer port stays with the creating component. Here all is fine as well, everything is running in a single thread of execution so we diligently do our bookkeeping on the data associated with the port and the channel and we can transfer the port.

3. The peer port is already owned by a different component. Here we need to have a slightly more complicated protocol

In this last case we take the following actions, `C` for creating component, `N` for newly created component, and `P` for the peer component that holds the other port of the same channel.

1. `C` transfers the port to the newly created component `N`, and ask it to come up with a new ID for that port. The port had an ID that was decided by its old owner `C`, and now has one that is agreeable with component `N`.

2. `C` sends a control message `PeerChangeBlockPort` to the peer `P`.

3. `P` receives the `PeerChangeBlockPort` message. It causes the peer port to be temporarily blocked. `P` may still continue executing its code, but the moment it wishes to send something over this port it is forced to block its execution. In response `P` sends an `Acknowledge` message back to `C`.

4. `C` waits for the `Acknowledge` message of `C`. Since the `Acknowledge` message was sent after the last data message that `P` sent to the port under consideration, and because `P` has blocked that port, we are certain that we received all messages. We transfer these messages to `N`.

5. Note that there may be multiple ports being transferred from `C` to `N`, each with a potentially different peer. And so here `C` waits until steps 2 through 4 are completed for all of the transferred ports.

6. Once `C` has received all of the `Acknowledge` messages it was waiting for, it will send a `PeerChangeUnblockPort` message to each peer `P`. This message contains the new port ID, such that `P` can unblock its port, and continue sending messages over this channel, but now correctly arriving at `N`. Additionally, `C` will now schedule the new component `N`.

There is a bit of extra overhead here with the `PeerChangeBlockPort` -> `Acknowledge` -> `PeerChangeUnblockPort` sequence with respect to other possible schemes. But this one allows `P` to continue executing its code as long as it does not use its blocked port. It also ensures that messages arrive in order: they're all collected by `C`, given to `N`, and only then may `P` continue creating messages to `N`, hence arriving after the earlier messages have been handed off to `N`.

As we'll later introduce, ports can end up being closed. When a port is closed the above procedure is not initiated. A port cannot be reopened, and once a port is closed its peer is also notified that the channel cannot be used anymore. As such, it is not needed (and perhaps impossible, if the memory backing the peer component has already been freed) to notify that the port will change peer.

## Managing the Lifetime of Components

Components are created by other components or by the API. Note that the API may for intents and purposes be viewed as a component. It may own and create channels, and it may create components. Component start, like a function, executing at the top of their program listing and may end in two ways. Either they encounter an unrecoverable error, or they neatly reach the end of their execution.

In either case we want to free up the memory that was occupied by the component during its execution. And we want to make sure that any kind of pending message/action is correctly handled. Apart from that the component contains somekind of message inbox, whose memory is associated with that component. Hence we want to make sure that all peers are aware that they're no longer able to send messages to a terminated component.

A small interlude before we continue: trying to take care of unrecoverable errors that occur during a sync round by incorporating the appropriate logic into the consensus algorithm proved to be rather hard. It caused a large increase in the number of states a component could be in, and as a result made the code much harder to reason about. That is: not so much communicating that an error had ocurred (that needs to occur in every synchronous algorithm), but keeping track of which messages can be sent to which component during the consensus algorithm.

For this reason there are two systems that make sure that components stay alive as long as needed. Firstly components will have a reference counter. For simplicity the component also holds a reference to itself. The component will remove this self-reference when it terminates. Each channel causes two components to also hold references to eachother. *If* a consensus algorithm is implemented such that a central components ends up communicating to all participating parties (using channels or not using channels), then we can make sure that it can reach all participating components by incrementing their reference counts (note that this is not yet properly implemented in the runtime). 

Through this mechanism the consensus algorithm can be greatly simplified. If a component encounters a critical error and is already participated in a sync round, then it can notify the other participants, but remain reachable to all other components until the round ends (and the reference counts are decreased again).

A second system is needed to ensure that a component actually exits (because all the peers hold a reference, and we need all of those references to drop to 0 to truly remove the component from the runtime). And so when a component exits it will send `ClosePort` messages to all of its peers. These peers will `Acknowledge` those messages and close the respective ports, meaning that they will no longer allow sending messages over those ports, that will be a fatal error. However, messages that were sent to the exiting component before receiving the `ClosePort` message might still be underway. And so the terminating component will wait for all of the `Acknowledge` messages for all of its channels such that it knows that it has received all data messages. The component will respond to these intermediate messages with a `DataMessageFailed` message, meaning that the message has been received the moment the component was already terminated, hence the sender should consider this a failed message transmission.

Bringing these systems together: apart from data messages there might still be control messages in transit, or the exiting component might still have some control/sync work to do. And so we need to modify something we said earlier: instead of a component removing its self-reference the moment it terminates, we do this the moment we have received all the `Acknowledge` messages we were expecting. This way if a component is busy creating another component, we're certain that the appropriate protocols are observed. So:

Concluding everything described above, two separate mechanisms will act in conjunction:

- A. The exiting component `E` waiting until it has finished notifying all peers `P` of its termination:

  1. Component `E` sends a `ClosePort` message to each peer `P` for each of the shared channels (except when the port is already closed because `P` is also in the process of shutting down).

  2. The peer `P` receives the `ClosePort` message and closes the port as a result. This is a change of port state that will cause any subsequent `put` operations on that port to become a fatal error for component `P`. In response the peer `P` sends an `Acknowledge` message to component `E`, unless component `P` exited itself (that is: sending a `ClosePort` message to `E` before the `ClosePort` message from `P` arrived). After closing the port the component `P` will remove the reference to `E`.

  3. Component `E` waits until all of its pending control operations are finished (i.e. waiting for the `Acknowledge` messages following `ClosePort` messages, `PeerChangeBlockPort` messages, etc.). Once all of these are finished, note that we can no longer participate in any future control actions: component `E` will not create channels/components itself. Since all of its ports are closed, the peers `P` will also not send any data or control messages.

  4. Component `E` now checks its inbox for any remaining messages. It will respond to any data messages that arrived after `E` sending `ClosePort` and before `E` receiving `Acknowledge` with a `DataMessageFailed` message (except for ports that were closed before the `Acknowledge`). Then it removes the reference to itself, therefore decrementing the reference counter for the component by 1.

- B. The reference counting mechanism. Any sync round the exiting component `E` is participating in will conceptually hold a reference to `E`. The component `E` will always respond to sync messages as if it were alive (albeit trying to indicate to everyone that it is actually exiting). The component that removes the last reference to the component `E` (which may be `E` itself, but also a peer `P`) will truly remove the associated memory from the runtime. 

**Note**: We did not consider machine termination. That is to say: once we reach the runtime maturity where communication occurs over different machines, then we have to consider that machines encounter fatal errors. However these can only be resolved by embedding the possibility of failure inside the protocol over which these machines communicate. 

## Sending Data Messages

So far we've discussed the following properties associated with sending data messages:

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.

- Every arriving data message will immediately be used to identify the sender as the peer of the corresponding `get`ter port. Since messages between components arrive in order this allows us to detect when the `put`s are in a different order at the sender as the `get`s at the receiver.

use std::collections::VecDeque;

use super::value::*;

use super::store::*;

use super::error::*;

use crate::protocol::*;

use crate::protocol::ast::*;

use crate::protocol::type_table::*;

macro_rules! debug_enabled { () => { false }; }

macro_rules! debug_log {

    ($format:literal) => {

        enabled_debug_print!(false, "exec", $format);

    };

    ($format:literal, $($args:expr),*) => {

        enabled_debug_print!(false, "exec", $format, $($args),*);

    };

}

#[derive(Debug, Clone)]

pub(crate) enum ExprInstruction {

    EvalExpr(ExpressionId),

    PushValToFront,

}

#[derive(Debug, Clone)]

pub(crate) struct Frame {

    pub(crate) definition: ProcedureDefinitionId,

    pub(crate) monomorph_index: usize,

    pub(crate) position: StatementId,

    pub(crate) expr_stack: VecDeque<ExprInstruction>, // hack for expression evaluation, evaluated by popping from back

    pub(crate) expr_values: VecDeque<Value>, // hack for expression results, evaluated by popping from front/back

    pub(crate) max_stack_size: u32,

}

impl Frame {

    /// Creates a new execution frame. Does not modify the stack in any way.

    pub fn new(heap: &Heap, definition_id: ProcedureDefinitionId, _monomorph_type_id: TypeId, monomorph_index: u32) -> Self {

        let definition = &heap[definition_id];

        let outer_scope_id = definition.scope;

        let first_statement_id = definition.body;

        // Another not-so-pretty thing that has to be replaced somewhere in the

        // future...

        fn determine_max_stack_size(heap: &Heap, scope_id: ScopeId, max_size: &mut u32) {

            let scope = &heap[scope_id];

            // Check current block

            let cur_size = scope.next_unique_id_in_scope as u32;

            if cur_size > *max_size { *max_size = cur_size; }

            // And child blocks

            for child_scope in &scope.nested {

                determine_max_stack_size(heap, *child_scope, max_size);

            }

        }

        let mut max_stack_size = 0;

        determine_max_stack_size(heap, outer_scope_id, &mut max_stack_size);

        Frame{

            definition: definition_id,

            monomorph_index: monomorph_index as usize,

            position: first_statement_id.upcast(),

            expr_stack: VecDeque::with_capacity(128),

            expr_values: VecDeque::with_capacity(128),

            max_stack_size,

        }

    }

    /// Prepares a single expression for execution. This involves walking the

    /// expression tree and putting them in the `expr_stack` such that

    /// continuously popping from its back will evaluate the expression. The

    /// results of each expression will be stored by pushing onto `expr_values`.

    pub fn prepare_single_expression(&mut self, heap: &Heap, expr_id: ExpressionId) {

        debug_assert!(self.expr_stack.is_empty());

        self.expr_values.clear(); // May not be empty if last expression result(s) were discarded

        self.serialize_expression(heap, expr_id);

    }

    /// Prepares multiple expressions for execution (i.e. evaluating all

    /// function arguments or all elements of an array/union literal). Per

    /// expression this works the same as `prepare_single_expression`. However

    /// after each expression is evaluated we insert a `PushValToFront`

    /// instruction

    pub fn prepare_multiple_expressions(&mut self, heap: &Heap, expr_ids: &[ExpressionId]) {

        debug_assert!(self.expr_stack.is_empty());

        self.expr_values.clear();

        for expr_id in expr_ids {

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

            self.serialize_expression(heap, *expr_id);

        }

    }

    /// Performs depth-first serialization of expression tree. Let's not care

    /// about performance for a temporary runtime implementation

    fn serialize_expression(&mut self, heap: &Heap, id: ExpressionId) {

        self.expr_stack.push_back(ExprInstruction::EvalExpr(id));

        match &heap[id] {

            Expression::Assignment(expr) => {

                self.serialize_expression(heap, expr.left);

                self.serialize_expression(heap, expr.right);

            },

            Expression::Binding(expr) => {

                self.serialize_expression(heap, expr.bound_to);

                self.serialize_expression(heap, expr.bound_from);

            },

            Expression::Conditional(expr) => {

                self.serialize_expression(heap, expr.test);

            },

            Expression::Binary(expr) => {

                self.serialize_expression(heap, expr.left);

                self.serialize_expression(heap, expr.right);

            },

            Expression::Unary(expr) => {

                self.serialize_expression(heap, expr.expression);

            },

            Expression::Indexing(expr) => {

                self.serialize_expression(heap, expr.index);

                self.serialize_expression(heap, expr.subject);

            },

            Expression::Slicing(expr) => {

                self.serialize_expression(heap, expr.from_index);

                self.serialize_expression(heap, expr.to_index);

                self.serialize_expression(heap, expr.subject);

            },

            Expression::Select(expr) => {

                self.serialize_expression(heap, expr.subject);

            },

            Expression::Literal(expr) => {

                // Here we only care about literals that have subexpressions

                match &expr.value {

                    Literal::Null | Literal::True | Literal::False |

                    Literal::Character(_) | Literal::Bytestring(_) | Literal::String(_) |

                    Literal::Integer(_) | Literal::Enum(_) => {

                        // No subexpressions

                    },

                    Literal::Struct(literal) => {

                        // Note: fields expressions are evaluated in programmer-

                        // specified order. But struct construction expects them

                        // in type-defined order. I might want to come back to

                        // this.

                        let mut _num_pushed = 0;

                        for want_field_idx in 0..literal.fields.len() {

                            for field in &literal.fields {

                                if field.field_idx == want_field_idx {

                                    _num_pushed += 1;

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

                                    self.serialize_expression(heap, field.value);

                                }

                            }

                        }

                        debug_assert_eq!(_num_pushed, literal.fields.len())

                    },

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

            for val_idx in 0..num_values {

                cur_frame.expr_values[val_idx] = store.read_take_ownership(cur_frame.expr_values[val_idx].clone());

            }

            // And now transfer to the heap region

            let values = &mut store.heap_regions[heap_pos as usize].values;

            debug_assert!(values.is_empty());

            values.reserve(num_values);

            for _ in 0..num_values {

                values.push(cur_frame.expr_values.pop_front().unwrap());

            }

            heap_pos

        }

        // Helper function to make sure that an index into an aray is valid.

        fn array_inclusive_index_is_invalid(store: &Store, array_heap_pos: u32, idx: i64) -> bool {

            let array_len = store.heap_regions[array_heap_pos as usize].values.len();

            return idx < 0 || idx >= array_len as i64;

        }

        fn array_exclusive_index_is_invalid(store: &Store, array_heap_pos: u32, idx: i64) -> bool {

            let array_len = store.heap_regions[array_heap_pos as usize].values.len();

            return idx < 0 || idx > array_len as i64;

        }

        fn construct_array_error(prompt: &Prompt, modules: &[Module], heap: &Heap, expr_id: ExpressionId, heap_pos: u32, idx: i64) -> EvalError {

            let array_len = prompt.store.heap_regions[heap_pos as usize].values.len();

            return EvalError::new_error_at_expr(

                prompt, modules, heap, expr_id,

                format!("index {} is out of bounds: array length is {}", idx, array_len)

            )

        }

        // Checking if we're at the end of execution

        let cur_frame = self.frames.last_mut().unwrap();

        if cur_frame.position.is_invalid() {

            if heap[cur_frame.definition].kind == ProcedureKind::Function {

                todo!("End of function without return, return an evaluation error");

            }

            return Ok(EvalContinuation::ComponentTerminated);

        }

        debug_log!("Taking step in '{}'", heap[cur_frame.definition].identifier.value.as_str());

        // Execute all pending expressions

        while !cur_frame.expr_stack.is_empty() {

            let next = cur_frame.expr_stack.pop_back().unwrap();

            debug_log!("Expr stack: {:?}", next);

            match next {

                ExprInstruction::PushValToFront => {

                    cur_frame.expr_values.rotate_right(1);

                },

                ExprInstruction::EvalExpr(expr_id) => {

                    let expr = &heap[expr_id];

                    match expr {

                        Expression::Assignment(expr) => {

                            let to = cur_frame.expr_values.pop_back().unwrap().as_ref();

                            let rhs = cur_frame.expr_values.pop_back().unwrap();

                            // Note: although not pretty, the assignment operator takes ownership

                            // of the right-hand side value when possible. So we do not drop the

                            // rhs's optionally owned heap data.

                            let rhs = self.store.read_take_ownership(rhs);

                            apply_assignment_operator(&mut self.store, to, expr.operation, rhs);

                        },

                        Expression::Binding(_expr) => {

                            let bind_to = cur_frame.expr_values.pop_back().unwrap();

                            let bind_from = cur_frame.expr_values.pop_back().unwrap();

                            let bind_to_heap_pos = bind_to.get_heap_pos();

                            let bind_from_heap_pos = bind_from.get_heap_pos();

                            let result = apply_binding_operator(&mut self.store, bind_to, bind_from);

                            self.store.drop_value(bind_to_heap_pos);

                            self.store.drop_value(bind_from_heap_pos);

                            cur_frame.expr_values.push_back(Value::Bool(result));

                        },

                        Expression::Conditional(expr) => {

                            // Evaluate testing expression, then extend the

                            // expression stack with the appropriate expression

                            let test_result = cur_frame.expr_values.pop_back().unwrap().as_bool();

                            if test_result {

                                cur_frame.serialize_expression(heap, expr.true_expression);

                            } else {

                                cur_frame.serialize_expression(heap, expr.false_expression);

                            }

                        },

                        Expression::Binary(expr) => {

                            let lhs = cur_frame.expr_values.pop_back().unwrap();

                            let rhs = cur_frame.expr_values.pop_back().unwrap();

                            let result = apply_binary_operator(&mut self.store, &lhs, expr.operation, &rhs);

                            cur_frame.expr_values.push_back(result);

                            self.store.drop_value(lhs.get_heap_pos());

                            self.store.drop_value(rhs.get_heap_pos());

                        },

                        Expression::Unary(expr) => {

                            let val = cur_frame.expr_values.pop_back().unwrap();

                            let result = apply_unary_operator(&mut self.store, expr.operation, &val);

                            cur_frame.expr_values.push_back(result);

                            self.store.drop_value(val.get_heap_pos());

                        },

                        Expression::Indexing(_expr) => {

                            // Evaluate index. Never heap allocated so we do

                            // not have to drop it.

                            let index = cur_frame.expr_values.pop_back().unwrap();

                            let index = self.store.maybe_read_ref(&index);

                            debug_assert!(index.is_integer());

                            let index = if index.is_signed_integer() {

                                index.as_signed_integer() as i64

                            } else {

                                index.as_unsigned_integer() as i64

                            };

                            let subject = cur_frame.expr_values.pop_back().unwrap();

                            let (deallocate_heap_pos, value_to_push) = match subject {

                                Value::Ref(value_ref) => {

                                    // Our expression stack value is a reference to something that

                                    // exists in the normal stack/heap. We don't want to deallocate

                                    // this thing. Rather we want to return a reference to it.

                                    let subject = self.store.read_ref(value_ref);

                                    let subject_heap_pos = match subject {

                                        Value::String(v) => *v,

                                        Value::Array(v) => *v,

                                        Value::Message(v) => *v,

                                        _ => unreachable!(),

                                    };

                                    if array_inclusive_index_is_invalid(&self.store, subject_heap_pos, index) {

                                        return Err(construct_array_error(self, modules, heap, expr_id, subject_heap_pos, index));

                                    }

                                    (None, Value::Ref(ValueId::Heap(subject_heap_pos, index as u32)))

                                },

                                _ => {

                                    // Our value lives on the expression stack, hence we need to

                                    // clone whatever we're referring to. Then drop the subject.

                                    let subject_heap_pos = match &subject {

                                        Value::String(v) => *v,

                                        Value::Array(v) => *v,

                                        Value::Message(v) => *v,

                                        _ => unreachable!(),

                                    };

                                    if array_inclusive_index_is_invalid(&self.store, subject_heap_pos, index) {

                                        return Err(construct_array_error(self, modules, heap, expr_id, subject_heap_pos, index));

                                    }

                                    let subject_indexed = Value::Ref(ValueId::Heap(subject_heap_pos, index as u32));

                                    (Some(subject_heap_pos), self.store.clone_value(subject_indexed))

                                },

                            };

                            cur_frame.expr_values.push_back(value_to_push);

                            self.store.drop_value(deallocate_heap_pos);

                        },

                        Expression::Slicing(expr) => {

                            // Evaluate indices

                            let from_index = cur_frame.expr_values.pop_back().unwrap();

                            let from_index = self.store.maybe_read_ref(&from_index);

                            let to_index = cur_frame.expr_values.pop_back().unwrap();

                            let to_index = self.store.maybe_read_ref(&to_index);

                            debug_assert!(from_index.is_integer() && to_index.is_integer());

                            let from_index = if from_index.is_signed_integer() {

                                from_index.as_signed_integer()

                            } else {

                                from_index.as_unsigned_integer() as i64

                            };

                            let to_index = if to_index.is_signed_integer() {

                                to_index.as_signed_integer()

                            } else {

                                to_index.as_unsigned_integer() as i64

                            };

                            // Dereference subject if needed

                            let subject = cur_frame.expr_values.pop_back().unwrap();

                            let deref_subject = self.store.maybe_read_ref(&subject);

                            // Slicing needs to produce a copy anyway (with the

                            // current evaluator implementation)

                            enum ValueKind{ Array, String, Message }

                            let (value_kind, array_heap_pos) = match deref_subject {

                                Value::Array(v) => (ValueKind::Array, *v),

                                Value::String(v) => (ValueKind::String, *v),

                                Value::Message(v) => (ValueKind::Message, *v),

                                _ => unreachable!()

                            };

                            if array_inclusive_index_is_invalid(&self.store, array_heap_pos, from_index) {

                                return Err(construct_array_error(self, modules, heap, expr.from_index, array_heap_pos, from_index));

                            }

                            if array_exclusive_index_is_invalid(&self.store, array_heap_pos, to_index) {

                                return Err(construct_array_error(self, modules, heap, expr.to_index, array_heap_pos, to_index));

                            }

                            // Again: would love to push directly, but rust...

                            let new_heap_pos = self.store.alloc_heap();

                            debug_assert!(self.store.heap_regions[new_heap_pos as usize].values.is_empty());

                            if to_index > from_index {

                                let from_index = from_index as usize;

                                let to_index = to_index as usize;

                                let mut values = Vec::with_capacity(to_index - from_index);

                                for idx in from_index..to_index {

                                    let value = self.store.heap_regions[array_heap_pos as usize].values[idx].clone();

                                    values.push(self.store.clone_value(value));

                                }

                                self.store.heap_regions[new_heap_pos as usize].values = values;

                            } // else: empty range

                            cur_frame.expr_values.push_back(match value_kind {

                                ValueKind::Array => Value::Array(new_heap_pos),

                                ValueKind::String => Value::String(new_heap_pos),

                                ValueKind::Message => Value::Message(new_heap_pos),

                            });

                            // Dropping the original subject, because we don't

                            // want to drop something on the stack

                            self.store.drop_value(subject.get_heap_pos());

                        },

                        Expression::Select(expr) => {

                            let subject= cur_frame.expr_values.pop_back().unwrap();

                            let mono_data = &heap[cur_frame.definition].monomorphs[cur_frame.monomorph_index];

                            let field_idx = mono_data.expr_info[expr.type_index as usize].variant.as_select() as u32;

                            // Note: same as above: clone if value lives on expr stack, simply

                            // refer to it if it already lives on the stack/heap.

                            let (deallocate_heap_pos, value_to_push) = match subject {

                                Value::Ref(value_ref) => {

                                    let subject = self.store.read_ref(value_ref);

                                    let subject_heap_pos = match expr.kind {

                                        SelectKind::StructField(_) => subject.as_struct(),

                                        SelectKind::TupleMember(_) => subject.as_tuple(),

                                    };

                                    (None, Value::Ref(ValueId::Heap(subject_heap_pos, field_idx)))

                                },

                                _ => {

                                    let subject_heap_pos = match expr.kind {

                                        SelectKind::StructField(_) => subject.as_struct(),

                                        SelectKind::TupleMember(_) => subject.as_tuple(),

                                    };

                                    let subject_indexed = Value::Ref(ValueId::Heap(subject_heap_pos, field_idx));

                                    (Some(subject_heap_pos), self.store.clone_value(subject_indexed))

                                },

                            };

                            cur_frame.expr_values.push_back(value_to_push);

                            self.store.drop_value(deallocate_heap_pos);

                        },

                        Expression::Literal(expr) => {

                            let value = match &expr.value {

                                Literal::Null => Value::Null,

                                Literal::True => Value::Bool(true),

                                Literal::False => Value::Bool(false),

                                Literal::Character(lit_value) => Value::Char(*lit_value),

                                Literal::Bytestring(lit_value) => {

                                    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.reserve(lit_value.len());

                                    for byte in lit_value {

                                        values.push(Value::UInt8(*byte));

                                    }

                                    Value::Array(heap_pos)

                                }

                                Literal::String(lit_value) => {

                                    let heap_pos = self.store.alloc_heap();

                                    let values = &mut self.store.heap_regions[heap_pos as usize].values;

                                    let value = lit_value.as_str();

                                    debug_assert!(values.is_empty());

                                    values.reserve(value.len());

                                    for character in value.as_bytes() {

                                        debug_assert!(character.is_ascii());

                                        values.push(Value::Char(*character as char));

                                    }

                                    Value::String(heap_pos)

                                }

                                Literal::Integer(lit_value) => {

                                    use ConcreteTypePart as CTP;

                                    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;

                                    debug_assert_eq!(concrete_type.parts.len(), 1);

                                    match concrete_type.parts[0] {

                                        CTP::UInt8  => Value::UInt8(lit_value.unsigned_value as u8),

                                        CTP::UInt16 => Value::UInt16(lit_value.unsigned_value as u16),

                                        CTP::UInt32 => Value::UInt32(lit_value.unsigned_value as u32),

                                        CTP::UInt64 => Value::UInt64(lit_value.unsigned_value as u64),

                                        CTP::SInt8  => Value::SInt8(lit_value.unsigned_value as i8),

                                        CTP::SInt16 => Value::SInt16(lit_value.unsigned_value as i16),

                                        CTP::SInt32 => Value::SInt32(lit_value.unsigned_value as i32),

                                        CTP::SInt64 => Value::SInt64(lit_value.unsigned_value as i64),

                                        _ => unreachable!("got concrete type {:?} for integer literal at expr {:?}", concrete_type, expr_id),

                                    }

                                }

                                Literal::Struct(lit_value) => {

                                    let heap_pos = transfer_expression_values_front_into_heap(

                                        cur_frame, &mut self.store, lit_value.fields.len()

                                    );

                                    Value::Struct(heap_pos)

                                }

                                Literal::Enum(lit_value) => {

                                    Value::Enum(lit_value.variant_idx as i64)

                                }

                                Literal::Union(lit_value) => {

                                    let heap_pos = transfer_expression_values_front_into_heap(

                                        cur_frame, &mut self.store, lit_value.values.len()

                                    );

                                    Value::Union(lit_value.variant_idx as i64, heap_pos)

                                }

                                Literal::Array(lit_value) => {

                                    let heap_pos = transfer_expression_values_front_into_heap(

                                        cur_frame, &mut self.store, lit_value.len()

                                    );

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

                                        },