# Runtime Design

## General Architecture

Roughly speaking the project consists of the following parts:

1. 

   The compiler itself. This transforms the PDL source code into an executable format.

2. 

   The interpreter. This takes the executable format and executes it. It is a very unoptimized AST walker based on stack frames. Generally speaking the bottommost frame in the stack contains the code and memory associated with a component.

    Once the interpreter hits a point where it needs to interact with the runtime (generally in order to communicate with another component) it will halt and emit a signal to the runtime.

3. 

We'll not go into points 1 and 2 in this document. One may simply assume that at the language level there is support for all of the things that are implemented in the runtime: sync blocks, channels, ports, component creation, sending and receiving messages, etc.

## Runtime Architecture

The runtime's code essentially consists of:

- The machinery that keeps track of the components. That is to say: there is some memory reserved for each of the components. And using some kind of component ID we can look up this memory in a registry. If the runtime finds that there are no more user-controlled components running (i.e. the ones that a programmer uses to create more components) and there are no more regular components running, then the runtime shuts down.

- A bit of shared memory for all of the OS threads that will be managed by the runtime. This mainly consists of a work queue. This work queue contains the identities of the components that are scheduled for execution.

- An auxilliary polling thread. Not of great importance, so a short description suffices: although most components react to one another, some components (e.g. a TCP socket) might have nothing to do until the OS instructs it to do something. This polling thread ensures that once there is something to do, the component is put back onto the work queue.

As long as the runtime doesn't shut down there will be `T` threads executing `N` components. A component can either be running (by being executed by a scheduler thread), scheduled for execution (its ID is in the work queue), or sleeping. All of these states are exclusive. Maintaining the exclusivity of these states is of great importance! We never want to end up in a place where two threads are both modifying the code/memory state of the same component!

A component will start its lifecycle as being put into the work queue (not exactly true, but this will be dealt with later) by the component that created it. At some point a scheduler will pick up the newly created component's ID from the work queue and start executing its code. Once running (and we'll exclude fatal errors for now) the component may at some point reach the end of its code and terminate, or it may encounter a place in its code where it blocks and needs to wait for external input (e.g. it performed a `get`, but a message has not arrived yet).

Once the execution of a component is blocked, it will attempt to go to sleep. Meaning: it will be in a state where it is not running, but also not scheduled for execution. A component may then be woken up by a different component, or the polling thread, by sending the sleeping a component a message. To prevent components from scheduling a sleeping component multiple times, the memory of a component contains an atomic `sleeping` boolean.

Returning to the idea of a component wishing to enter the "sleeping" state. The procedure in pseudocode is written as:

```

func go_to_sleep(this_component) {

   assert(atomic::load(this_component.sleeping) == false);

   // Note that while we were trying to go to sleep, someone has sent us a new

   // message, but it did not see yet that we stored `false` in the sleeping

   // flag, so we need to check ourselves.

   if messages_in_inbox() {

      // We try to set the flag back to false, but another scheduler thread may

      // have already done this

      let swap_success = atomic::cas(this_component.sleeping, true, false);

      if swap_success {

         put_in_work_queue(this_component.id);

      }

   }

}

```

Similarly, each time we try to send a component a message, we must do the following:

```

func send_message_to(target_component, message_data) {

   put_message_in_inbox_locked(target_component.inbox, message_data);

   let swap_success = atomic::cas(target_component.sleeping, true, false);

   if swap_success {

      put_in_work_queue(target_component.id);

   }

}

```

Note that, because we cannot predict how the OS threads are scheduled, we can also not predict the way in which our own schedulers (which are running on OS threads) will schedule the execution of the components. Hence as a mental model, one may assume that each component is running in its own thread. The system above ensures that if a component has something to do (because it has received a message), it will eventually end up being executed by a scheduler. With the code for the "sleeping" state we'll ensure that a component can only be executed by one scheduler at a time.

## General Messaging

With this rough architecture, components can send each other messages. One will find three kinds of messages in the runtime (okay, four, but the last one is just to make the OS polling thread work):

2. Control messages: to facilitate certain operations permitted within PDL, the scheduler thread may decide to send messages to other components. These messages are called control messages. We'll encounter them later in this document when describing component creation, transmitting ports, and closing channels. Different from data messages, which may linger in the component's inbox for a while, control messages are handled by the receiving component immediately. This is important for various control algorithms.

3. Sync messages: to facilitate the consensus algorithm, there will be messages initiated by the scheduler thread as well. That is to say: when the component sends data messages there will be information attached to the data message that facilitates a successful sync round. But apart from that when the components are all done executing their code they must somehow reach consensus. This is done through these sync messages.

Note that already the concept of a channel, each having its own little slot (or limited buffer), becomes a superfluous design decision for the runtime. This is because the scheduler threads themselves also need to be able to send messages to other components. Hence we'll need something more generic than a per-channel buffer, namely a generic message inbox.

As we'll later also see, the concept of a directional channel *may* be a useful tool within PDL code (and I'm not arguing against such a concept), but as we'll see throughout this document the control messages can (conceptually) flow both ways over the channel.

## Runtime Design Drivers

Most design choices in the runtime were based on the fact that the Reowolf language should facilitate easier programming over the internet. So although the entire system currently only considers components running on a single machine, those components were conceptually regarded as living on different machines. In other words: components were conceptually considered not to have shared memory they can both access.

Yet another result is that we decided to design the runtime without any globally unique component and/or port IDs. Certainly: on a single machine a component IDs allows one to retrieve a component's memory. But when sending a message to a component living on another machine, it may well be that we're sending it to a through a port that has the same port ID as ours, and targeting a component that has the same ID as ours.

## Control Algorithms

### Changing Port Peers due to Component Creation

```

comp creator(in<u32> one_port) {

   channel another_port -> and_a_final_one;

   sync {

   }

   // perform component creation

}

```

We have several runtime properties to uphold when we're transferring ports:

- No globally unique port IDs, so the new component is allowed to choose new port IDs for the ports it is adopting ownership of.

- The peers of the transferred ports may be unaware that a new component is created. In fact those peers may have already transferred messages to the instantiating component! As a design decision (the one that we find makes sense) any incoming, but unread, messages for a port are transferred along to the new component.

- Together with the requirement that peers need to be aware of the transferred ports, we also need to maintain ordering in the sent messages that intend to arrive at that transferred port at some point in time.

For all of the requirements above, the internal control algorithm to transfer a port to a new component is as following:

2. The component transfers all of the metadata associated with the port, and transfers all of the messages that are targeted at those transferred ports to the new component.

3. For each transferred port the instantiator sends a `PortPeerChanged_Block` control message to the peers. This message instructs the peer that the port should be temporarily blocked. Any component that tries to send a message through that port enters a blocked state that can only be lifted if the corresponding `PortPeerChanged_Unblock` control message is sent. At the same time the instantiator sets up a special bit of code that will relay all incoming messages from that peer to the new component. We've mentioned earlier that all messages will have a target port. So when messages arrive at the instantiator component that need to be relayed, the instantiator component will modify the target port to the new component's chosen port ID.

4. Once a peer has received a `PortPeerChanged_Block`, it will, as stated above, stop sending messages over that channel. Not only data messages, but control messages as well. This also means that if the other component cannot start transferring ports itself. In any case, it will respond with an `Ack`nowledgement back to the instantiator component.

5. The instantiator component waits until it has received an `Ack` for all of the `PortPeerChanged_Block` message it has sent. This is done such that we're sure that we've received all of the messages that are actually intended for the new component (because while the new component is being created the peer may still be sending messages intended for the new component, but sent to the instantiator component). As a result, the new component will have all of the data messages in the inbox in the order in which they were sent, therefore maintaining the runtime property of message ordering.

6. When all of the `Ack`s are received, the instantiator component will remove the bit of code that relays all of the messages and will schedule the new component for execution. At this point the instantiator component will no longer access the private memory of the new component. Since the instantiator component is aware of the new component's ID and the new port IDs for all of the transferred ports, it will send `PortPeerChanged_Unblock` messages to all of the peer components. This message will also contain the new component's ID and its port ID.

### Shutting Down Components

In order to do so we have another control protocol. We'll extend this protocol when we discuss encountering crashing components, but we'll introduce a simpler variant here. The protocol is relatively simple. For each of the ports that are not yet closed and are owned by the terminating component we will:

1. Make sure that the port is not blocked. If the port is blocked then the component blocks until the associated port is becomes unblocked. If the port is already closed then we do not execute the other steps in this control algorithm.

2. The port will send a `ClosePort` message to the peer of the port that is closing. Note that this `ClosePort` message will have a target port. If it happens to be that the terminating component will receive a `PortPeerChanged_Block` message for that port in the near future, we're certain that the `ClosePort` message will at least arrive at the correct peer (since the target port will be used to relay that message to the correct receiver).

3. The peer of the port, upon receiving a `ClosePort` message, will mark the port as being closed in its metadata. From that point onwards, any attempt to `put` or `get` on that port will result in the peer component crashing. In response to the `ClosePort` message, the peer component will send an `Ack`.  There is one exception, and that is when the peer component itself already initiated a `ClosePort` control algorithm for that port. In that case the incoming `ClosePort` message is treated like an `Ack`.

4. The terminating component will wait until all of the `Ack`s have arrived (or crossing `ClosePort` messages, as stated in point (3)). Once they do, they will instruct the runtime to remove the component from memory.

### Transferring Ports through Data Messages

The PDL code allows for ports to be transferred through ports. As a simple example, consider the following code:

```

struct Pair {

   in<bool> command,

   out<u32> response,

}

comp some_component(

   in<u32> to_transmit,

   out<in<u32>> tx_one_port,

   out<Pair> tx_two_ports

) {

   // Transmitting a port directly

   sync put(tx_one_port, to_transmit);

   // Transmitting multiple ports at a time using a data structure

   channel command_tx -> command_rx;

   channel response_tx -> response_rx;

   sync {

      let message = Pair{ 

         command: command_rx,

         response: response_tx,

      };

      put(tx_two_ports, message);

   }

   // Sending a command and receiving a response

   sync {

      put(command_tx, true);

      auto response = get(response_rx);

   }

}

```

To facilitate this, we'll follow roughly the same procedure as when we're transferring ports to a newly created component. But we have one complication: we do not have direct access to the private memory of the component we're sending the ports to (we'll call this component the "adopting component", and the sending component the "relinquishing component"). And so we'll have to follow a control protocol that is slightly different.

Note that it is perfectly okay to send closed ports. The adopting component will receive this component together with the information that the port is closed. In this way, if the adopting component attempts a `put` or `get` on that received component, it will crash.

1. The relinquishing component will first make sure that none of the ports are blocked. If the ports are blocked then it will sleep until the ports become unblocked. As stated above the relinquishing component will also make sure that the ports were not previously used within the synchronous round.

2. The relinquishing component will send `PortPeerChanged_Block` message to all of the peers of the ports that will be transferred. However, in this case it will not relay any messages to the new component, they will still pile up in the relinquishing component's inbox.

4. The relinquishing component will wait until all of the expected `Ack` message are received. Once they are received the component will wait until the port the message will travel through becomes unblocked (that is: the port that is used to transfer the ports to the adopting component).

5. The relinquishing component will send the data message containing the transferred ports to the adopting component. It will annotate this message with a list containing `(tranferred port ID, peer component ID, peer port ID)` triples. Note that since those peer ports are blocked, they will not be transferred in the meantime. This is essential for the next step.

6. The adopting component will receive the annotated data message containing the transferred ports. For each transferred port it will decide upon a new port ID.

7. The adopting component will, for each adopted port, send out a `PortPeerChanged_Unblock` message to the blocked peer ports. This message will be annotated with the `(adopting component ID, new port ID)` pairs. Such that the peers all know where the peers can be found.

## Dealing with Crashing Components

A component may at any point during its execution be triggered to crash. This may be because of something simple like an out-of-bounds array access. But as described above using closed ports may lead to such an event as well. In such a case we not only need to go through the `ClosePort` control protocol, to make sure that we can remove the crashing component's memory from the runtime, but we'll also have to make sure that all of the peers are aware that *their* peer has crashed. Here we'll make a design decision: if a peer component crashes during a synchronous round and there were interactions with that component, then that interacting component should crash as well. The exact reasons will be introduced later, but it comes down to the fact that we need to do something about the fact that the synchronous round will never be able to complete.

We'll talk ourselves through the case of a component crashing before coming up with the control algorithm to deal with components crashing.

We'll first consider that a component may crash inside our outside of a synchronous block. From the point of view of the peer component, we'll have four cases to consider:

1. The peer component is not in a synchronous block. 

2. The crashing component died before the peer component entered the synchronous block.

3. The crashing component died during the same synchronous block as the peer component.

4. The crashing component died after reaching consensus on the synchronous block that the peer component is currently still in.

Before discussing these cases, it is important to remember that the entire runtime has components running in their own thread of execution. We may have that the crashing component is unaware of its peers (due to the fact that peer ports might change ownership at any point in time). We'll discuss the consensus algorithm in more detail later within the documentation. For now it is important to note that the components will discover the synchronous region they are part of while the PDL code is executing. So if a component crashes within a synchronous region before the end of the sync block is reached, it may be possible that it will not discover the full synchronous region it would be part of.

We'll go back to the four cases we've discusses above. We'll change our point of view: we're now considering a component (the "handling component") that has to deal with the failure of a peer (the "crashing component"). We'll introduce a small part of our solution a-priori: like a component shutting down, a failing component will simply end its life by broadcasting `ClosePort` message over all of its owned ports that are not closed (and the failing component will also wait until all of those ports are not blocked).

In the first case, we're dealing with a failing component while the handling component is not in a synchronous block. This means that if there was a previous synchronous block, that it has succeeded. We might still have data messages in our inbox that were sent by the failing component. But in this case it is rather easy to deal with this: we mark the ports as closed, and if we end up using them in the next synchronous block, then we will crash ourselves.

In the second case we have that the peer component died before we ourselves have entered the synchronous block. This case is somewhat equivalent to the case we described above. The crashing component cannot have sent the handling component any messages. So we mark the port as closed, potentially failing in the future if they end up being used.

The fourth case is where the failing component crashes *after* the handling component finished its sync round. This is an edge cases dealing with the following situation: both the handling as the crashing component have submitted their local solution to the consensus algorithm. The crashing component receives a global solution, finishes the sync round, and then crashes, therefore sending the `ClosePort` message to the handling component. The handling component, due to the asynchronous nature of the runtime, receives the `ClosePort` message before the global solution has a chance to reach the handling component. In this case, however, the handling component should be able to finish the synchronous round, and it shouldn't crash.

Here is where we arrive at the protocol for dealing with component crashes. To let the handling component deal with this last case, we'll let the crashing component send the `ClosePort` messages together with a boolean indicating if it crashed inside, or outside of a synchronous block.

In this way we modify the control algorithm for terminating components. Now we're able to deal correctly with crashing components.

# Previous Consensus Algorithm

## Introduction

The previous consensus algorithm (the one within Reowolf 1.0 and 1.1) had support for speculative execution. This means that the user may (directly or indirectly) fork the execution of a component. That particular execution (which may be a forked execution already) then becomes >1 executions. At some point a component will have to choose which particular execution will be committed to memory. This is one reason for the existence of a `sync` block: a block of code wherein one may perform forking, and at the end a component will have to choose the execution that is committed to memory.

With speculative execution we may have multiple components that are all forking their execution and sending/receiving messages. So we do not end up with one component that has to choose its final execution, but all components choosing their final execution. Note that one component's execution may apply restrictions on the validity of another component's execution. As an example, suppose the following components and their executions:

- Component A: Has two executions:

    - Execution A1: Component A has sent a message to component B.

    - Execution A2: Component A has received a message from component B.

- Component B: Has three executions:

    - Execution B1: Component B has received a message from component A, then sends a message back to component A.

    - Execution B2: Component B has received a message from component A.

    - Execution B3: Component B has sent two messages to component A.

Without delving into too much detail, one may see that the only valid solution to this problem is the combination of `A1` and `B2`. So the components cannot just pick any execution, but must pick an execution that is agreeable with the chosen executions of the components it has interacted with.

## Component Execution Tree, and Execution Traces

Components execute PDL code, which may contain calls to `fork`, `put`, and `get`. A `fork` explicitly forks the execution of the code. A `put` sends a message to a particular component, and a `get` receives a message from a component and forks (as explained later).

As the component enters a `sync` block, it has only one possible execution. But as stated above there are reasons for the execution to split up. These individual executions may themselves split up later, thereby forming a so-called "execution tree":

```

                             +-----+       +------+

                             | put |------>| sync |

+-------+      +------+----->+-----+       | end  |

| sync  |      | fork |                    +------+

| start |----->+------+----->+-----+

+-------+                    | get |------>+------+

                             +-----+       | sync |

                                   |       | end  |

                                   |       +------+

                                   |

                                   +------>+------+

                                   |       | sync |

                                   |       | end  |

                                   |       +------+

                                   |

                                   +--> ...

```

```

primitive some_component(out<u32> tx, in<u32> rx) {

  sync {

    fork {

      put(tx, 5);

    } or fork {

      get(rx, 1);

    }

}

```

Note that the branching points can in the *general case* only be discovered at runtime. Any code may have control flow points like `if` statements, or `while` loops. Consider the following code:

```

primitive some_component(out<u32> tx, bool which_way) {

  sync {

    if (which_way) {

      fork {

        put(tx, 1);

      } or fork {

        put(tx, 2);

      }

    } else {

      put(tx, 3);

    }

  }

}

```

Depending on the value of `which_way` we produce two different execution trees (of which we can determine all traces). The compiler cannot decide at compile time which execution tree will be generated.

Note that the `get` branching points have an arbitrary number of forked executions arising from them. We'll call them "waiting points". In the *general case* we cannot figure out how many forked executions arise from a `get` branching point. The reason being might be illustrated by the following simple example:

```

primitive sender(out<u32> tx, u32 num_forks) {

  sync {

    auto fork_counter = 1;

    while (fork_counter < num_forks) {

      fork {

        put(tx, fork_counter); 

      } or fork { } // empty case

    }

    put(tx, num_forks);

  }

}

primitive receiver(in<u32> rx) {

  u32[] values = {};

  sync {

    bool keep_going = true;

    while (keep_going) {

      auto new_value = get(rx);

      values @= { new_value }; // append

      fork { 

        keep_going = false; 

      } or fork { }

    }

  }

}

```

If the sender is connected to the receiver, then the sender will send anywhere between `1` and `num_forks` messages (distributed over `num_forks` forks), depending on a user-supplied parameter (which we cannot figure out at compile-time). The isolated receiver can generate an infinite number of forked executions. We can analyze that the receiver will at most have `num_forks + 1` forked executions arising from its `get` branching point (the `num_forks` branches that do receive, and one final fork that is infinitely waiting on another message), but the compiler cannot.

For this reason a `get` branching point needs to be kept around for the entire duration of the sync block. The runtime will always need to have a copy of the component's memory and execution state the moment it encountered a `get` instruction, because it might just be that another component (in perhaps a new fork, which we cannot predict) will send it another message, such that it needs to produce a new forked execution.

So before treading into more detail, here we've found that in the general case:

- Speculative execution implies that we end up with an execution tree.

- The set of traces produced by a component in its sync block can practically only be discovered at runtime.

## Towards a Consensus Algorithm

The key to the consensus problem is somehow discovering the ways in which the components have influenced the memory state of their peers. If we have a complete trace for each component, for which all peer components agree on the way they have influenced that complete trace, then we've found a solution to the consensus problem. Hence we can subdivide the consensus problem into four parts:

1. Keeping track of the messages that influence the memory state of components.

2. Keeping track of the peers that influence the memory state of components.

3. Finding a set of interactions between components on which all involved components agree, i.e. each `put` should have a corresponding `get` at the peer.

4. Somehow having a communication protocol that finds these agreeable interactions.

We'll not consider the last point, as this is essentially a gossip protocol, and the appropriate gossip protocol varies with the requirements of the user (e.g. robust to failure, memory efficient, runtime efficiency, message complexity). We define some terms to make the following discussion easier:

- "sync region": The group of components that have interacted with one another and should agree on the global consensus solution.

- "local solution": A complete trace of a component. For the component this is a valid local solution, but might not be part of a global solution.

- "global solution": A set of traces, one for each of the components in the sync region, that all agree on the interactions that took place between the components in the sync region.

We'll now work incrementally to the final consensus algorithm, making it a bit of a story in order to explain the reasoning and intuition behind the consensus algorithm.

However, as we've outlined above, we cannot know exactly which messages we're going to receive. We'll have to discover these messages while executing a component. The next best thing is to keep track of the values of the messages that we've received in a complete trace. Once we have complete traces for all of the interacting components, we can check that the received value corresponds to a sent value. e.g.

```

primitive sender(out<u32> tx) {

  sync {

    fork {

      put(tx, 1);   

    } or fork {

      put(tx, 2);

    }

  }

}

primitive receiver(in<u32> rx) {

  u32 value = 0; 

  sync {

    value = get(rx);

  }

}

```

Where `tx` is part of the same channel as `rx`. In this case we'll have two traces for each of the components, resulting in two valid global consensus solutions. In one solution the message `1` was transferred, in another the message `2` was transferred. There are two problems with this solution: firstly it doesn't take the identity of the channel into account. And secondly it doesn't take the effects of previous messages into account.

To illustrate the first problem, consider:

```

primitive sender(out<u32> tx_a, out<u32> tx_b) {

  u32 some_variable_in_memory = 0;

  sync {

    fork {

      put(tx_a, 1);

      some_variable_in_memory = 1;

    } or fork {

      put(tx_b, 1);

      some_variable_in_memory = 2;

    }

  }

}

primitive receiver(in<u32> rx_a, in<u32> rx_b) {

  u32 value = 0; 

  sync {

    value = get(rx_a);

  }

}

```

Here the fact that the sender has the solutions `1` and `1` does not help the receiver figure out which of those corresponds to its own solution of `1`.

To illustrate the second problem, consider:

```

primitive sender(out<u32> tx) {

  sync {

    fork {

      put(tx, 1);

      put(tx, 2);

    } or fork {

      put(tx, 2);

    }

  }

}

primitive receiver(in<u32> rx) {

  u32 value = 0; 

  sync {

    value = get(rx);

  }

}

```

Now we'll have `sender` contributing the solutions `1, 2` and `2`. While the receiver will generate the solutions `1`, `2` and `2`. The reason there are three solutions for the receiver is because it cannot figure out that the message `2` from the sender depended on the first message `1` from the sender having arrived.

So we need to change the algorithm. Instead of just tracking which messages were sent, each component needs to have a mapping from port identities to sent messages (internally the runtime will generate port/channel IDs, but for the sake of this discussion we'll use the postfixes to the port names in the PDL code to indicate to which channel they belong, e.g. the `tx_a` out-port is part of the same channel `a` as the `rx_a` in-port). Secondly, if we sent a message, we need to transmit in which way it depends on previously received messages by sending along the sender's port mapping. The receiver, upon `get`ting a message, checks the port mapping to see if there is any of its own executions that can accept that message.

We're already calling this information the "port mapping" (because we'll truly turn it into a mapping later), but for now the sent mapping is a list of pairs containing `(port ID, sent value`).

This changes the way we can interpret the execution tree: each node is not only associated with the performed operation (`fork`, `put` or `get`), but also associated with a particular port mapping that indicates the influence of other components that allowed it to reach that exection node. We modify the port mapping per node in the following way:

- For a `put`: we transmit the current port mapping, and the transmitted value in the message. We then update the mapping from the sending port: that particular port ID now maps to the recently transmitted value.

- For a `get`: we receive the transmitted port mapping and value. Note that this `get` might be a particular statement executed by multiple different forked executions (each with their own port mapping). And so for a `get` to succeed, we need the shared channels between the sender and the receiver to agree on their port mapping. If such a `get` succeeds, then it forks into a new execution where the receiving port now maps to the received value.

So, for a slightly more complicated example, combining the two previously examples:

```

primitive initiator(out<u32> tx_a, out<u32> tx_b, in<u32> rx_c) {

  u32 value = 0;

  sync {

    put(tx_a, 1); // operation i1

    fork {

      put(tx_a, 1); // operation i2

      value = get(rx_c); // operation i3

    } or fork {

      put(tx_b, 2); // operation i4

      value = get(rx_c); // operation i5

    }

  }

}

primitive responder(in<u32> rx_a, in<u32> rx_b, out<u32> tx_c) {

  sync {

    auto value_1 = get(rx_a); // operation r1

    auto value_2 = 0;

    fork {

      value_2 = get(rx_a); // operation r2

    } or fork {

      value_2 = get(rx_b); // operation r3

    }

    put(tx_c, value1 + value2); // operation r4

}

```

```

For the initiator:

sync  --> put (i1) --> fork +--> put (i2) -----> get (i3) -+-----> sync end {A}

start     [(a,1)]           |    [(a,1),(a,1)]   [(a,1),(a,1),(c,2)]

                            |                              |

                            |                              |

                            |                              +-> ...

                            |

                            +--> put (i4) -----> get (i5) -+-----> sync end {B}

                                 [(a,1),(b,2)]   [(a,1),(b,2),(c,3)]

                                                           |

                                                           |

                                                           +-> ...

For the responder:

sync  --> get (r1) -+--> fork +--> get (r2) -----> put (r4) ----> sync end {C}

start     [(a,1)]   |         |    [(a,1),(a,1)]   [(a,1),(a,1),(c,2)]

                    |         |

                    +-> ...   +--> get (r3) -----> put (r4) ----> sync end {D}

                                   [(a,1),(b,2)]   [(a,1),(b,2),(c,3)]

```

We can see that the `put` operation at `i2` does not end up being received at `r1`. The reason being that at `r1` the responder expects to not have received anything on `rx_a` yet. The message that the initiator sends contains the annotation `[(a,1),(a,1)]`, meaning: I have previously sent `[(a,1)]`, and am now sending `[(a,1)]`. The only operation that can receive this operation is at `r2`, because that expects the mapping `[(a,1)]`! 

Similarly: when the responder `put`s at `r4`, this happens in two branches. The branch that ends up being trace `C` expects the initiator to be in the state `[(a,1),(a,1)]` (hence can only be received at operation `i3`, resulting in trace `A` of the initiator). The branch that ends up being trace `D` expects the initiator to be in the state `[(a,1),(b,2)]` (hence can only be received at operation `i5`, resulting in trace `B` of the initiator).

And ofcourse, when both components are finished, they can compare the mappings in both of them and conclude that the traces `A` and `C` are compatible, since their port mappings are compatible. Similarly traces `B` and `D` are compatible. So there are two global solutions to the consensus problem.

For the sake of simplicity, we've only considered two components. But there may be many more components involved in a single synchronous region. In this case we'll have to clarify when receiving a message is valid. When a message is sent to another component, the receiving component first filters the port mapping (both for the mapping stored in the execution tree and the mapping sent over the channel) such that only the channels shared between the two components are left. If those two mappings are equal, then the message can be received in that branch. 

The same process needs to be applied when we seek a global solution. In rather silly pseudocode, but the simplest way to explain this process, is to have the following algorithm seeking the global solution:

```

all_components = [ component_1, component_2, ..., component_N ]

global_solutions = []

// Nested loop through all components

for all complete traces in component_1:

  for all complete traces in component_2:

    ...

      ...

        for all complete traces in component_N:

          let success = true;

          let all_traces = [trace_of_1, trace_of_2, ..., trace_of_N];

          // Looping through every pair of traces

          for index_a in 0..N:

            for index_b in index_a + 1..N:

              let component_a = all_components[index_a]

              let trace_a = all_traces[index_a]

              let component_b = all_components[index_b]

              let trace_b = all_traces[index_b]

              trace_a = filter_on_shared_channels(trace_a, component_a, component_b)

              trace_b = filter_on_shared_channels(trace_b, component_a, component_b) 

              if trace_a != trace_b:

                success = false;

                break

            if !success:

              break

          if success:

            global_solutions = append(global_solutions, all_traces)

```

We'll now apply the last bit of trickery to this algorithm. Firstly keeping track of the sent message values may be prohibitively expensive. Suppose some kind of streaming data processor that receives gigabytes of data in a single synchronous round. It would be an unwise design decision to store all of that data in the port mapping. So what we'll do instead is assinging each node in the execution tree a unique number (unique only for that execution tree, different execution trees might contain the same number). With that unique number we'll redefine the port mapping to consist of a list of `(port ID, branch number)` pairs.

The reason this is a valid trick is because of the arguments made earlier regarding control flow being influenced by received messages. If we know how a component was influenced by external influences, then the control flow path it takes is deterministic, hence the content of the sent messages will be deterministic. Locally a component `A` may only describe the way it was influenced by its peer `B`, but `B` also records how it was influenced by its peers `C` and `D`. So transitively `A` will also know the indirect mutual influences between it and `C` and `D`.

- A `get` operation never receives a message: in that case we keep blocking indefinitely, we'll never get a complete trace, hence are prevented from finding a local solution.

- A `get` operation receives a message containing a compatible port mapping: in this case both components agree on the order of operations that took place for the message to be transferred.

- A `put` operation is never received: again, we're fine. The `put`ting component might submit a local solution for a complete trace. But the mapping for that never-received message will also never appear in the supposed receiver's port mapping. Hence the two components will never agree on the control flow paths they took.

## The Consensus Algorithm

With the story above, we may describe the complete consensus finding algorithm as following.

Each component will locally construct an execution tree. Branches appear whenever it encounters a `fork` (producing a set of new branches), a `put` operation (this will create a single new branch) or a `get` operation (which will create new branches if the sent message's port mapping is compatible with the port mapping stored at that `get`'s branch). Whenever a new branch is created it is assigned a locally unique identifier.

The port mapping at each branch consists of a mapping `{port ID -> branch number}`. This port mapping starts out empty at the start of a sync block. When a component `put`s a value through a channel, it will:

- Generate the `put` branch and assign it a unique ID.

- Send the message that is annotated with the ancestor branch's port mapping. Together with the (`sending port's ID`, `newly assigned branch ID)` pair.

- Update the port mapping of the `put` branch: it will copy the ancestor branch's port mapping, and update it with the `(send port's ID, newly assigned branch ID)` pair. In case a previous entry is present for that specific `port ID`, then it is overwritten.

When a component encounters a `fork`, it will simply copy the ancestor branch's port mapping for each new branch. Each new branch will receive a new unique ID. When a component performs a `get`, it will block until it receives a message annotated with a port mapping. If that happens, then it will:

- Compare the port mapping in the message with the branch's port mapping. If one of the shared channels do not agree on the port mapping, then the message is not valid for that `get` operation. Note that the message *must* be kept around, because in the future there may be a `get` operation that *is* valid for that port mapping.

- If the port mapping for the shared channels do agree, then we:

  - Generate a new fork originating from the `get` branch and assign it a unique ID.

  - Update the port mapping of the `get` branch: copy the ancestor branch's port mapping, and update it with the `(sending port ID, peer's branch number)` pair.

## Reasons for not implementing this Consensus Algorithm

There are a lot of practical issues with this consensus algorithm:

1. The fact that a `get` operation never knows when it will receive a new message requires us to keep a complete copy of the component's memory and execution state at that point. Hence for each `get` operation we're incurring a rather large memory overhead.

2. The fact that we never know if a received message can be discarded because it cannot be received by any of the `get` operations in the component's code. There may be another message coming in that causes a fork with a `get` operation that *can* receive this message. Hence we need to keep around *all* of the messages received in a synchronous round.

4. Considering the previous points: the simplicity (especially when running over the internet) for a nefarious actor to incur computational overhead for the receiver. All it has to do is to keep sending messages to the receiver with an acceptable port mapping, but to never offer the consensus algorithm a valid local solution. Each accepted message will spawn a new fork at the receiver.

# Current Consensus Algorithm and Communication Rules

## Introduction 

For all of the reasons described in the previous consensus algorithm, speculative execution was removed from the runtime. This greatly simplifies the consensus algorithm. In fact, there isn't much need for a consensus algorithm anymore, apart from some edge cases. Hence the `fork` statement is gone from the language.

Most of this document will describe a different problem that wasn't discussed in the document describing the previous consensus algorithm: the introduction of a `select` statement and the way in which components are allowed into a synchronous region.

## Reducing the Previous Consensus Algorithm

Without speculative execution, there is no more execution tree: there is only one control flow path a component can take through its PDL code. That execution may not form a complete trace (i.e. reach the end of the sync block) because it encounters a blocking operation. But to reiterate: if there is a complete trace, then it can only have taken one control flow path through its code.

However, there is still are reasons for incorporating parts of the old consensus algorithm. That is: there is still a port mapping, `put` operations update this port mapping, and `get` operations (which will not fork anymore, but simply block until a message is present) do the same thing. 

There are two reasons for this. Firstly there is the design choice of enforcing strict ordering on the way channels are used between components. If there are two channels between components, then we may have the following code:

```

comp sender_variant_a(out<u32> tx_a, out<u32> tx_b) {

    sync {

        put(tx_a, 1);

        put(tx_b, 2);

    }

}

comp sender_variant_b(out<u32> tx_a, out<u32> tx_b) {

    sync {

        put(tx_b, 2);

        put(tx_b, 1);

    }

}

comp receiver(in<u32> rx_a, in<u32> rx_b) {

    sync {

        auto va = get(rx_a);

        auto vb = get(rx_b);

    }

}

```

If we wouldn't send along the port mapping. Then both `sender_variant_a` as `sender_variant_b` would be valid peers for the `receiver` component. This is because if put operations are still asynchronous (that is: they send the message and continue executing, not waiting for the corresponding `get` to complete), then both messages will arrive at the `receiver` component. The receiver component can retrieve these messages in any order. However, if we *do* send along the port mapping, then only `sender_variant_a` can be a valid peer of `receiver`. Note: this is a design choice.

We could reduce this design choice by only sending along the port mapping of the port over which is a message is sent. This would imply that messages over the same channel have to be received in the correct order, but messages over different channels can be received in any desired order. We've kept this strict ordering in place by sending along the full port mapping.

The second reason for still including the port mapping has to do with the fact that, by design, `put` operations are not acknowledged. That is to say: the `put` operation is asynchronous, and does not prevent the sending component from continuing its execution. For a `put` operation to be valid, the message *must* be received by the peer. Without any kind of port mapping we have no way of detecting the following mismatch:

```

comp sender(out<u32> tx) {

    sync {

        put(tx, 0);

        put(tx, 1);

    }

}

comp receiver(in<u32> rx) {

    sync get(rx);

}

```

However, there are much simpler ways of designing a consensus algorithm without speculative execution: one may annotate the port with the number of message sent, one may (as stated above) only send along the port mapping for a single port, instead of all shared channels, etc.

That being said, the current implementation still uses branch IDs (differently interpreted in this case: a simple operation counter). There is still a port mapping from `port ID` to `branch ID`, and finding the global solution still proceeds in the same way (except now there is only one complete trace per component).

## Synchronous Regions

There were some aspects of the consensus algorithm that were specifically left out in the previous document. Among them is when to consider a component part of the synchronous region. In version 1.0 of the Reowolf runtime each peer of each component was considered part of the synchronous region. The idea behind this decision was that the fact that a channel remains unused in a synchronous round should be seen as part of the consensus algorithm: if a component did not `get` on a port, then the peer must agree that it did not `put` on a port.

So when a complete trace was found (that is: a component reached the end of its sync block), a component went over all its ports and sent a special message over the unused/silent channels asking the peers to join the synchronous round with their local solutions, where those local solutions are required not to have sent/received anything over those silent channels.

The problem with this approach may be illustrated with a simple example: suppose a set of requesting servers that are connected to a database server of some sorts. In the normal situation it would be perfectly normal for multiple servers to be storing database resources at the same time. These can all be handled in sequence by the database server, but the requesting servers do not necessarily have to wait for one another. Some pseudocode for this example:

```

union Request { 

    StoreData(u8[]),

    /* more things here in a more complicated example */ 

}

comp requester(out<Request> tx_cmd) {

    // Setup code here

    // Perhaps some code that requires retrieving database resources

    while (!should_exit) {

        sync {

            u8[] data = { 1, 2, 3 } /* something more complicated */

            sync {

                put(tx_cmd, Request::StoreData(data))

            }

        }

    }

}

comp database(in<Request>[] rx_cmds) {

    // Note the array of input ports: there are multiple requesters

    // Lot of complicated stuff

    while (!should_exit && length(rx_cmds) > 0) {

        // Here is the conundrum (and a second one, mentioned later in this document):

        auto command = get(rx_cmds[0]); // take a message from the first requester

        if (let Request::StoreData(to_store) = command) {

            store_the_data(to_store);        

        } else {

            // other commands, in a more complicated example

        }

    }

}

```

In this case, the code seems reasonable, but will *always* fail if there are >1 requesters at the database component. Because once the database reaches the end of its sync block, it will have a mapping for `rx_cmds[0]`, but the remaining ports were all silent. So the consensus algorithm asks the peer `requester` components if their channels were silent. But they aren't! Each requester can only send data in its sync block.

So one might be inclined (in fact, its the only way to solve this in the unmodified language) to write the requester as:

```

comp requester(out<Request> tx_cmd) {

    // Setup code here

    // Perhaps some code that requires retrieving database resources

    while (!should_exit) {

        sync {

            fork {

                // We either do nothing

            } or fork {

                // Or we communicate with the database server 

                u8[] data = { 1, 2, 3 } /* something more complicated */

                sync {

                    put(tx_cmd, Request::StoreData(data))

                }

            }

        }

        some_code_that_may_take_a_really_long_time();

    }

}

```

In some sense the problem is solved. Only one `requester` component will have its request going through, the other ones have silent channels, and the `database` component is capable of receiving that message. But there are two further problems here: Firstly there is no way to give any guarantees about fairness: the component that can communicate to the `database` component the fastest will probably have their traces completed first, hence will be the first ones submitted to the consensus algorithm, hence will likely be the one picked first. If this happens over the internet then the slow connections will almost never be able to submit their information. Secondly we have that none of the components can run at their own pace. Each component *always* has to wait until all of the other ones have joined the synchronous region. There is no way in which a single `requester` can do some other work on its own unfettered by the execution speeds of its peers.

And so the rule for joining a synchronous region was changed. Now it is simply: if a component performs a `get` and receives a message from another component, then they both become part of the same synchronous region (and perhaps this region is already larger because the components are part of larger separate synchronous regions). If a message is in the inbox of the component, but there is no `get` operation performed, then the message is kept around for perhaps a later synchronous round. Note that this works together with the consensus algorithm in such a way that joining a consensus region happens upon the first accepted message on a channel. Later messages within that synchronous round now *must* arrive due to the consensus algorithm.

## Select Statement

The example above hints at a second problem (figured out a long time ago by our unix overlords): there is no way within PDL code to say that we can perform a variety of behaviours triggered by the arrival of messages. For our `database` component above, we see that it can have multiple requesters connected to it, but there is no way to indicate that we can have valid behaviour for any one of them. So we introduce the `select` statement. It is a statement with a set of arms. Each arm has a guard, where we indicate which ports need to have incoming messages, and a block of code, that is executed when all of the indicated ports actually have received a message.

The select statement may only appear within a sync block. The arm's guard is formulated in terms of an expression or a variable declaration (that may contain `get` calls, but not `put` calls). The corresponding block of code is executed when all of the `get` calls in the guard have a message ready for them. In case multiple arm guards are satisfied then a random arms is chosen by the runtime for execution.

So the select statement takes the form:

```

select {

    auto value = get(rx_a) + get(rx_b) 

        -> do_something_with_the_value(value),

    auto value = get(rx_a)

        -> do_something_with_the_value(value),

}

```

This code is transformed by the compiler into something akin to:

```

while (still_waiting_for_a_message) {

    if (value_present(rx_a) && value_present(rx_b)) {

        auto value = get(rx_a) + get(rx_b);

        do_something_with_the_value(value);

        break;

    } else if (value_present(rx_a)) {

        auto value = get(rx_a);

        do_something_with_the_value(value);

        break;

    }

}

```

The combination of the `select` statement and the way we introduce components into the synchronous region permits components to run independently of another when their protocol admits it.

The rule where components only enter a synchronous region when the `get` a message that is present in the inbox still applies here. If, in the example above, the arm requiring messages on channel `b` executes, then only the peer component of this channel joins the synchronous region. The message that came over channel `a` will still be present in the inbox for later interactions or sync blocks.

## The Current Consensus Algorithm

Because the current consensus algorithm is a scaled down version of the previous one, we'll be a bit more concise: Each component has a local counter, this produces number we (still) call branch branch numbers. The counter is incremented each time a `get` or a `put` call is performed. We'll use this counter to annotate ports in the port mapping each time a `put` call is performed. The port mapping is sent along with messages sent through `put` calls. This message will arrive in the inbox of the receiver. When the peer performs a `get` call on the receiving end of the channel, it will check if the received port mapping matches the local port mapping. If it doesn't, we can immediately let the component crash. If it does, then the message is received and the component continues execution. Once a `get` call has completed, the sender is incorporated into the synchronous region, if it wasn't already.