use std::fs::File;

use std::io::Read;

use clap::{App, Arg};

use reowolf_rs as rw;

fn main() {

    let app = App::new("rwc")

        .author("Henger, M.")

        .version(env!("CARGO_PKG_VERSION"))

        .about("Reowolf compiler")

        .arg(

            Arg::new("input")

                .long("input")

                .short('i')

                .help("input files")

                .required(true)

                .takes_value(true)

                .multiple_occurrences(true)

        )

        .arg(

            Arg::new("threads")

                .long("threads")

                .short('t')

                .help("number of runtime threads")

                .default_value("1")

                .takes_value(true)

        )

        .arg(

            Arg::new("loglevel")

                .long("log_level")

                .short('l')

                .help("set log level ('none', 'info', 'debug' or 'all')")

                .default_value("all")

                .takes_value(true)

        )

        .arg(

            Arg::new("stdlib")

                .long("stdlib")

                .short('s')

                .help("standard library directory (overrides default)")

                .takes_value(true)

        );

    // Retrieve arguments and convert

    let app = app.get_matches();

    let input_files = app.values_of("input");

    if input_files.is_none() {

        println!("ERROR: Expected at least one input file");

        return;

    }

    let num_threads = app.value_of("threads").unwrap();

    let num_threads = match num_threads.parse::<i32>() {

        Ok(num_threads) => {

            if num_threads < 0 || num_threads > 255 {

                println!("ERROR: Number of threads must be a number between 0 and 256");

                return;

            }

            num_threads as u32

        },

        Err(err) => {

            println!("ERROR: Failed to parse number of threads\nbecause: {}", err);

            return;

        }

    };

    let log_level = app.value_of("loglevel").unwrap();

    let log_level = match log_level {

        "none" | "None" => rw::runtime2::LogLevel::None,

        "debug" | "Debug" => rw::runtime2::LogLevel::Debug,

        "info" | "Info" | "all" | "All" => rw::runtime2::LogLevel::Info,

        _ => {

            println!("ERROR: Unexpected log level");

            return;

        }

    };

    let standard_library_dir = app.value_of("stdlib")

        .map(|v| v.to_string());

    // Add input files to file buffer

    let input_files = input_files.unwrap();

    assert!(input_files.len() > 0); // because arg is required

    let mut builder = rw::ProtocolDescriptionBuilder::new(standard_library_dir)

        .expect("create protocol description builder");

    let mut file_buffer = Vec::with_capacity(4096);

    for input_file in input_files {

        print!("Adding file: {} ... ", input_file);

        let mut file = match File::open(input_file) {

            Ok(file) => file,

            Err(err) => {

                println!("FAILED (to open file)\nbecause:\n{}", err);

                return;

            }

        };

        file_buffer.clear();

        if let Err(err) = file.read_to_end(&mut file_buffer) {

            println!("FAILED (to read file)\nbecause:\n{}", err);

            return;

        }

        if let Err(err) = builder.add(input_file.to_string(), file_buffer.clone()) {

            println!("FAILED (to tokenize file)\nbecause:\n{}", err);

        }

        println!("Success");

    }

    // Compile the program

    print!("Compiling program ... ");

    let protocol_description = match builder.compile() {

        Ok(pd) => pd,

        Err(err) => {

            println!("FAILED\nbecause:\n{}", err);

            return;

        }

    };

    println!("Success");

    // Start runtime

    print!("Startup of runtime ... ");

    let runtime = rw::runtime2::Runtime::new(num_threads, log_level, protocol_description);

    if let Err(err) = &runtime {

        println!("FAILED\nbecause:\n{}", err);

    }

    println!("Success");

    // Make sure there is a nameless module with a main component

    print!("Creating main component ... ");

    let runtime = runtime.unwrap();

    if let Err(err) = runtime.create_component(b"", b"main") {

        use rw::ComponentCreationError as CCE;

        let reason = match err {

            CCE::ModuleDoesntExist => "Input files did not contain a nameless module (that should contain the 'main' component)",

            CCE::DefinitionDoesntExist => "Input files did not contain a component called 'main'",

            CCE::DefinitionNotComponent => "Input file contained a 'main' function, but not a 'main' component",

            _ => "Unexpected error"

        };

        println!("FAILED\nbecause:\n{} (raw error: {:?})", reason, err);

        return;

    }

    println!("Success");

    println!("Now running until all components have exited");

    println!("--------------------------------------------\n\n");

}

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

   The runtime. This is the code that keeps track of the components, decides when they can run, where their messages should end up, and bring various control algorithms (running behind the scene) to completion.

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.

Once such builtin features are encountered in the interpreter (e.g. the creation of a channel), a signal will be emitted to the runtime. A rough outline of the architecture, and handling these signals, is discussed in this document.

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

- A set of scheduler threads. They attempt to retrieve work from the work queue. This work is in the form of component IDs, which they use to retrieve the associated component and run its PDL code. They do this by invoking the interpreter on the component's current execution and memory state. Once a runtime signal (as mentioned above) is emitted by the interpreter, the scheduler thread will deal with it appropriately.

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

It is instructive at this point to roughly explain how components are stored in memory. The components memory is roughly divided into two regions. There is the publicly accessible memory of a component and a private memory region. The public part is accessible by all scheduler threads. So all of the memory in the public memory region is somehow behind a lock, or some kind of locking mechanism (we will include the concept of atomics into "some kind of locking mechanism" for now). Hence the aforementioned `sleeping` boolean lives in this region. Conversely, the private memory region of a component is only accessed by the scheduler thread that is running the component. So here we store things like the memory state and execution state of the component.

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

   // We are currently executing, so our sleeping flag MUST be `false`

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

   atomic::store(this_component.sleeping, true); // we force the flag to true

   // 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):

1. Data messages: these are the messages that are sent by `put` calls and received with `get` calls. As will be explained later in this document, more information is attached to data messages than the values given as argument to the `put` call. These messages will arrive in the target component's inbox. When the target component performs a call to `get` they're pulled out of the inbox and transferred to the component's memory state, such that the PDL code can read the transferred values.

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.

This has some implications for channels. Certainly a component that sends a value must have it stored somewhere temporarily, and the component receiving it needs to keep it around as well. But the channel is not an entity that you'll find in memory. Rather there is one component owning one port, and when a message is `put` through it, it will arrive at another component owning the peer port; there is no memory sahred between components that will store a message flowing through a channel.

A multi-machine runtime also requires the runtime to embrace the asynchronous nature of components. `put`s are non-blocking and can be performed one after the other before the peer has performed a corresponding `get`. The language does not contain the concept of a "lock" such that two components can agree on who owns a shared bit of memory. Rather each component is executed in its own thread of execution, and for multiple components to coordinate their actions they must use the messaging facilities. In order to make this coordination-through-messages somewhat simple to reason about one of the design drivers of the runtime was to ensure that each message sent in a specific order from one component to another will arrive in that same order at the target component.

And so we have a multi-machine runtime where components running in their own thread can only coordinate through messages. As a result an ever-important consideration in designing internal (control) algorithms is something called "message crossing". Two components may decide to initiate a protocol at the same time, hence send each other the exact same protocol-initiating message (e.g. we have components `A` and `B`, and a protocol that requires an initiator to send a `Request` message, and then wait for a response in terms of a `Response` message, then we may have `A` and `B` both sending each other `Request` at the same time).

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

We'll now discuss several of the control algorithms. These control algorithms may be initiated by the scheduler threads when certain runtime signals are emitted by the interpreter. The control algorithms are brought to completion by sending messages. We'll talk about these messages as if they're sent from component to another component (this is for the sake of clarity: in reality they're sent by one scheduler thread to the memory location reserved for the target component's inbox). Because messages may be relayed one or more times before they arrive at the intended receiver (we'll introduce this concept soon), most messages include their intended target port in some kind of message header. This is true for all data messages, and most control messages. Only when a component is certain about the identity of the receiving component can it send messages without a target port in a header.

### Changing Port Peers due to Component Creation

Components, when they're not in sync mode, may decide to create new components. Ports may be used as the arguments to these newly created components. The rule we place on such a kind of port transfer is that the component that is creating the new component fully relinquishes ownership of the transferred port, and after the new component is created, the new component owns that port. As an annotated example:

```

comp creator(in<u32> one_port) {

   channel another_port -> and_a_final_one;

   sync {

      auto value = get(one_port); // legal, just receiving an integer

      put(another_port, 1337); // legal, sending a value over an owned

   }

   // perform component creation

   new some_other_component(one_port, and_a_final_one); // transferring two ports

   sync get(one_port); // illegal! Port was transferred

   sync put(another_port, 1338); // still legal, we still own this port

   sync get(and_a_final_one); // also illegal, port was transferred.

}

```

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.

- Similarly to the above: a peer of a transferred port needs to be aware at some point that its peer port has changed ownership.

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

Here we see the asynchronous nature of the runtime rear its head. Because the transferring of ports does not just happen to the receiving end of a port (in which case we transfer already received messages, hence messages only arrive at their correct destination eventually). It may happen to the transmitting end of a port as well. What this means for the receiver is that it is never sure which component is its peer until it has recevied a data message that is annotated with the origin of the message. At that moment in time the peer of the port is known, but only until the end of the synchronous round. Because after the synchronous round it is perfectly possible for the port to be passed around again.

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

1. The component that is creating the new component (we'll call the creator the instantiator component, and the created one the new component) temporarily has access to the private memory of the new component. Reason being is that a component is always created on the same machine as the instantiator component. And so the first step it takes is to create new port IDs (that make sense for the newly created component, instead of for the instantiator component) and map the old port IDs to the new ones.

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.

7. The peers, upon receiving the `PortPeerChanged_Unblock` message, will update the metadata of their ports such that they point to the new component's ports. They will also unblock the port such that messages can be sent again.

With this control algorithm, all peers are now aware of the new port's position. We've also maintained message ordering for the message sent to the new component. Although it was mentioned in point (4), we'll mention it here to be extra clear: creating a new component will be blocked until all of the transferred ports are unblocked. If we don't do this a data/control message may end up at the wrong component.

Likewise we see the asynchronous nature of ports: the peers are eventually consistent. This is why we stressed earlier that almost all messages have their targeted port in their message header. This is needed such that a component like the instantiator discussed above knows when to relay messages. In this process the relaying component will also update the target port ID in the header to the new port ID.

### Shutting Down Components

A component will require a bit of memory to run. So when we're done executing a component (either because it has crashes, or because its program has terminated) we would like to release this memory again. Earlier we mentioned that components send messages by accessing an inbox in the public memory region of a component. This memory will, ofcourse, be freed as well. So we need to make sure that when a component shuts down, all of its peers will somehow be notified that they can never send messages to that terminated component again.

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.

To reiterate: we have to be careful and annotate the `ClosePort` message with the target port. The terminating component will delay sending a `ClosePort` message if the port is blocked, but it may be that we have the `ClosePort` message crossing with a `PortPeerChanged_Block` message. Which implies that our `ClosePort` message will be relayed by the peer component.

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

We'll enforce a second rule upon transferring ports. Namely that ports transferred in a synchronous round may not have been used in `get` or `put` operations. I'm certain that it is possible to come up with a set of rules that will make this possible. But the protocol for transferring components over channels is a lot easier if we disallow this. For this reason we'll introduce a field in the metadata for each port that registers when the port was last used. If the relinquishing component attempts to transfer a port that has been used within the same sync round, then it will crash.

Like before we want to ensure that all messages intended for the transferred port arrive in the correct order at the adopting component.

And so the control protocol for transmitting ports proceeds as following:

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.

3. The peers, upon receiving the `PortPeerChanged_Block` message, will proceed as they would in the case where ports were transferred to a new component: they'll block the port and send an `Ack`.

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.

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.

For this reason, it does not make a lot of sense to deal with component failure through the consensus algorithm. Dealing with the failure through the consensus algorithm only makes sense if we can find the synchronous region that we would have discovered if we were able to fully execute the sync block of each participating component. As explained above: we can't, and so we'll opt to deal with failure on a peer-by-peer basis.

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.

## Sync Algorithm

A description of the synchronous algorithm is present in different documents. We will mention here that central to the consensus algorithm is that two components agree on the interactions that took place over a specific channel. In order for this to happen we'll send along a lot of metadata when trying to reach consensus, but here we're just concerned with attempting to match up the two ends of a channel. 

A port is identified by a `(component ID, port ID)` pair, and channel is a pair of those identifying pairs. So to match up the two ends of a channel we would have to find a consistent pair of ports that agree on who their peers are. However, we're dealing with the problem of eventual consistency: `put`ting ports never know who their peer is, because the sent message might be relayed. However, `get`ting ports *will* know who their peer is for the duration of a single synchronous round once they've received a single message.

This is the trick we will apply in the consensus algorithm. If a channel did not see any messages passing through it, then the components that own those ports will not have to reach consensus because they will not be part of the same synchronous region. However if a message did go through the channel then the components join the same synchronous region, and they'll have to form some sort of consensus on what interaction took place on that channel.

And so the `put`ting component will only submit its own `(component ID, port ID, metadata_for_sync_round)` triplet. The `get`ting port will submit information containing `(self component ID, self port ID, peer component ID, peer port ID, metadata_for_sync_round)`. The consensus algorithm can now figure out which two ports belong to the same channel.

# Known Issues

The current implementation of Reowolf has the following known issues:

- Cannot create uninitialized variables that are later known to be initialized. This is not a problem for the regular types (perhaps a bit tedious), but is a problem for channels/ports. That is to say: if a component needs a temporary variable for a port, then it must create a complete channel. e.g.

  ```

  comp send(out<u32> tx1, out<u32> tx2, in<bool> which) {

    channel unused -> temporary;

    while (true) sync {

      if (get(which)) {

        temporary = tx1;

      } else {

        temporary = tx2;

      }

      put(temporary, 1);

    }

  }

  ```

  Another solution would be to use an empty array and to put a port inside of that. Hacks galore!

- Reserved memory for ports will grow without bounds: Ports can be given away from one component to another by creating a component, or by sending a message containing them. The component sending those ports cannot remove them from its own memory if there are still other references to the transferred port in its memory. This is because we want to throw a reasonable error if that transferred port is used by the original owner. Hence we need to keep some information about that transferred port in the sending component's memory. The solution is to have reference counting for the ports, but this is not implemented.

- An extra to the above statements: when transferring ports to a new component, the memory that remembers the state of that port is removed from the component that is creating the new one. Hence using old references to that port within the creating component's PDL code results in a crash.

- Some control algorithms are not robust under multithreading. Mainly error handling when in sync mode (because there needs to be a revision where we keep track of which components are still reachable by another component). And complicated scenarios where ports are transferred.

- There is an assertion in the interpreter that makes sure that there are no values left on the expression stack when a statement has completed. This is not true when you have an expression statement! If you want to remove this assertion make sure to clear the stack (using the method on the `Store`).

- The TCP listener component should probably do a `shutdown` before a `close` on the socket handle. It should also set the `SO_REUSEADDR` option.

- The way in which putting ports are ordered to block if the corresponding getter port's main inbox is full is rather silly. This led to the introduction of the "backup inbox" as it is found in the runtime's code. There is a design decision to make here, but the current implementation is a bit silly. There are two options: (a) have an atomic boolean indicating if the message slot for an inbox is full, or (b) do away with the "main inbox" alltogether, and have an unbounded message queue.

- For practical use in components whose code supports an arbitrary number of peers (i.e. their code contains an array of ports that is used for communication and changes in size during the component's lifetime), the `select` statement somehow needs to support waiting on any one of those ports.

- The compiler currently prevents one from using `sync` blocks (or corresponding `get` and/or `put` operations) in functions. They can only be used within components. When writing large programs this it makes it rather hard to re-use code: all code that interacts with other components can only be written within a sync block. I would advise creating `sync func`, `nonsync` func and regular `func`tions. Where:

  - `sync func`tions can only be called from within `sync` functions. They may not open new sync blocks, but may perform calls to `get`/`put`. These are useful to encapsulate sequences of `put`/`get` calls together with some common message-modifying code.

  - `nonsync func`tions (or `async func`tions) may only be called outside of sync blocks, and may open new sync blocks themselves. They are useful to encapsulate a single interaction with other components. One may also create new components here.

  - regular `func`tions. Are as useful as in any other language, but here we disallow calling `nonsync func`tions or `sync func`tions.

- The `Ack` messages that are sent in response to `PeerPortChanged_Block` messages should contain the sending components `(component ID, port ID)` pair in case the `PeerPortChanged_Block` message is relayed. When such an `Ack` message is received, the peer of the port must be updated before transferring the port to the new owner.

- The compiler currently accepts a select arm's guard that is formulated as `auto a = get(get(rx))`. This should be disallowed.

use std::fmt;

use std::fmt::{Debug, Display, Formatter};

use std::ops::{Index, IndexMut};

use super::arena::{Arena, Id};

use crate::collections::StringRef;

use crate::protocol::input_source::InputSpan;

use crate::protocol::TypeId;

/// Helper macro that defines a type alias for a AST element ID. In this case 

/// only used to alias the `Id<T>` types.

macro_rules! define_aliased_ast_id {

    // Variant where we just defined the alias, without any indexing

    ($name:ident, $parent:ty) => {

        pub type $name = $parent;

    };

    // Variant where we define the type, and the Index and IndexMut traits

    (

        $name:ident, $parent:ty, 

        index($indexed_type:ty, $indexed_arena:ident)

    ) => {

        define_aliased_ast_id!($name, $parent);

        impl Index<$name> for Heap {

            type Output = $indexed_type;

            fn index(&self, index: $name) -> &Self::Output {

                &self.$indexed_arena[index]

            }

        }

        impl IndexMut<$name> for Heap {

            fn index_mut(&mut self, index: $name) -> &mut Self::Output {

                &mut self.$indexed_arena[index]

            }

        }

    };

    // Variant where we define type, Index(Mut) traits and an allocation function

    (

        $name:ident, $parent:ty,

        index($indexed_type:ty, $indexed_arena:ident),

        alloc($fn_name:ident)

    ) => {

        define_aliased_ast_id!($name, $parent, index($indexed_type, $indexed_arena));

        impl Heap {

            pub fn $fn_name(&mut self, f: impl FnOnce($name) -> $indexed_type) -> $name {

                self.$indexed_arena.alloc_with_id(|id| f(id))

            }

        }

    };

}

/// Helper macro that defines a wrapper type for a particular variant of an AST

/// element ID. Only used to define single-wrapping IDs.

macro_rules! define_new_ast_id {

    // Variant where we just defined the new type, without any indexing

    ($name:ident, $parent:ty) => {

        #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]

        pub struct $name (pub(crate) $parent);

        #[allow(dead_code)]

        impl $name {

            pub(crate) fn new_invalid() -> Self     { Self(<$parent>::new_invalid()) }

            pub(crate) fn is_invalid(&self) -> bool { self.0.is_invalid() }

            pub fn upcast(self) -> $parent          { self.0 }

        }

    };

    // Variant where we define the type, and the Index and IndexMut traits

    (

        $name:ident, $parent:ty, 

        index($indexed_type:ty, $wrapper_type:path, $indexed_arena:ident)

    ) => {

        define_new_ast_id!($name, $parent);

        impl Index<$name> for Heap {

            type Output = $indexed_type;

            fn index(&self, index: $name) -> &Self::Output {

                if let $wrapper_type(v) = &self.$indexed_arena[index.0] {

                    v

                } else {

                    unreachable!()

                }

            }

        }

        impl IndexMut<$name> for Heap {

            fn index_mut(&mut self, index: $name) -> &mut Self::Output {

                if let $wrapper_type(v) = &mut self.$indexed_arena[index.0] {

                    v

                } else {

                    unreachable!()

                }

            }

        }

    };

    // Variant where we define the type, the Index and IndexMut traits, and an allocation function

    (

        $name:ident, $parent:ty, 

        index($indexed_type:ty, $wrapper_type:path, $indexed_arena:ident),

        alloc($fn_name:ident)

    ) => {

        define_new_ast_id!($name, $parent, index($indexed_type, $wrapper_type, $indexed_arena));

        impl Heap {

            pub fn $fn_name(&mut self, f: impl FnOnce($name) -> $indexed_type) -> $name {

                $name(

                    self.$indexed_arena.alloc_with_id(|id| {

                        $wrapper_type(f($name(id)))

                    })

                )

            }

        }

    }

}

define_aliased_ast_id!(RootId, Id<Root>, index(Root, protocol_descriptions), alloc(alloc_protocol_description));

define_aliased_ast_id!(PragmaId, Id<Pragma>, index(Pragma, pragmas), alloc(alloc_pragma));

define_aliased_ast_id!(ImportId, Id<Import>, index(Import, imports), alloc(alloc_import));

define_aliased_ast_id!(VariableId, Id<Variable>, index(Variable, variables), alloc(alloc_variable));

define_aliased_ast_id!(DefinitionId, Id<Definition>, index(Definition, definitions));

define_new_ast_id!(StructDefinitionId, DefinitionId, index(StructDefinition, Definition::Struct, definitions), alloc(alloc_struct_definition));

define_new_ast_id!(EnumDefinitionId, DefinitionId, index(EnumDefinition, Definition::Enum, definitions), alloc(alloc_enum_definition));

define_new_ast_id!(UnionDefinitionId, DefinitionId, index(UnionDefinition, Definition::Union, definitions), alloc(alloc_union_definition));

define_new_ast_id!(ProcedureDefinitionId, DefinitionId, index(ProcedureDefinition, Definition::Procedure, definitions), alloc(alloc_procedure_definition));

define_aliased_ast_id!(StatementId, Id<Statement>, index(Statement, statements));

define_new_ast_id!(BlockStatementId, StatementId, index(BlockStatement, Statement::Block, statements), alloc(alloc_block_statement));

define_new_ast_id!(EndBlockStatementId, StatementId, index(EndBlockStatement, Statement::EndBlock, statements), alloc(alloc_end_block_statement));

define_new_ast_id!(LocalStatementId, StatementId, index(LocalStatement, Statement::Local, statements));

define_new_ast_id!(MemoryStatementId, LocalStatementId);

define_new_ast_id!(ChannelStatementId, LocalStatementId);

define_new_ast_id!(LabeledStatementId, StatementId, index(LabeledStatement, Statement::Labeled, statements), alloc(alloc_labeled_statement));

define_new_ast_id!(IfStatementId, StatementId, index(IfStatement, Statement::If, statements), alloc(alloc_if_statement));

define_new_ast_id!(EndIfStatementId, StatementId, index(EndIfStatement, Statement::EndIf, statements), alloc(alloc_end_if_statement));

define_new_ast_id!(WhileStatementId, StatementId, index(WhileStatement, Statement::While, statements), alloc(alloc_while_statement));

define_new_ast_id!(EndWhileStatementId, StatementId, index(EndWhileStatement, Statement::EndWhile, statements), alloc(alloc_end_while_statement));

define_new_ast_id!(BreakStatementId, StatementId, index(BreakStatement, Statement::Break, statements), alloc(alloc_break_statement));

define_new_ast_id!(ContinueStatementId, StatementId, index(ContinueStatement, Statement::Continue, statements), alloc(alloc_continue_statement));

define_new_ast_id!(SynchronousStatementId, StatementId, index(SynchronousStatement, Statement::Synchronous, statements), alloc(alloc_synchronous_statement));

define_new_ast_id!(EndSynchronousStatementId, StatementId, index(EndSynchronousStatement, Statement::EndSynchronous, statements), alloc(alloc_end_synchronous_statement));

define_new_ast_id!(ForkStatementId, StatementId, index(ForkStatement, Statement::Fork, statements), alloc(alloc_fork_statement));

define_new_ast_id!(EndForkStatementId, StatementId, index(EndForkStatement, Statement::EndFork, statements), alloc(alloc_end_fork_statement));

define_new_ast_id!(SelectStatementId, StatementId, index(SelectStatement, Statement::Select, statements), alloc(alloc_select_statement));

define_new_ast_id!(EndSelectStatementId, StatementId, index(EndSelectStatement, Statement::EndSelect, statements), alloc(alloc_end_select_statement));

define_new_ast_id!(ReturnStatementId, StatementId, index(ReturnStatement, Statement::Return, statements), alloc(alloc_return_statement));

define_new_ast_id!(GotoStatementId, StatementId, index(GotoStatement, Statement::Goto, statements), alloc(alloc_goto_statement));

define_new_ast_id!(NewStatementId, StatementId, index(NewStatement, Statement::New, statements), alloc(alloc_new_statement));

define_new_ast_id!(ExpressionStatementId, StatementId, index(ExpressionStatement, Statement::Expression, statements), alloc(alloc_expression_statement));

define_aliased_ast_id!(ExpressionId, Id<Expression>, index(Expression, expressions));

define_new_ast_id!(AssignmentExpressionId, ExpressionId, index(AssignmentExpression, Expression::Assignment, expressions), alloc(alloc_assignment_expression));

define_new_ast_id!(BindingExpressionId, ExpressionId, index(BindingExpression, Expression::Binding, expressions), alloc(alloc_binding_expression));

define_new_ast_id!(ConditionalExpressionId, ExpressionId, index(ConditionalExpression, Expression::Conditional, expressions), alloc(alloc_conditional_expression));

define_new_ast_id!(BinaryExpressionId, ExpressionId, index(BinaryExpression, Expression::Binary, expressions), alloc(alloc_binary_expression));

define_new_ast_id!(UnaryExpressionId, ExpressionId, index(UnaryExpression, Expression::Unary, expressions), alloc(alloc_unary_expression));

define_new_ast_id!(IndexingExpressionId, ExpressionId, index(IndexingExpression, Expression::Indexing, expressions), alloc(alloc_indexing_expression));

define_new_ast_id!(SlicingExpressionId, ExpressionId, index(SlicingExpression, Expression::Slicing, expressions), alloc(alloc_slicing_expression));

define_new_ast_id!(SelectExpressionId, ExpressionId, index(SelectExpression, Expression::Select, expressions), alloc(alloc_select_expression));

define_new_ast_id!(LiteralExpressionId, ExpressionId, index(LiteralExpression, Expression::Literal, expressions), alloc(alloc_literal_expression));

define_new_ast_id!(CastExpressionId, ExpressionId, index(CastExpression, Expression::Cast, expressions), alloc(alloc_cast_expression));

define_new_ast_id!(CallExpressionId, ExpressionId, index(CallExpression, Expression::Call, expressions), alloc(alloc_call_expression));

define_new_ast_id!(VariableExpressionId, ExpressionId, index(VariableExpression, Expression::Variable, expressions), alloc(alloc_variable_expression));

define_aliased_ast_id!(ScopeId, Id<Scope>, index(Scope, scopes), alloc(alloc_scope));

#[derive(Debug)]

pub struct Heap {

    // Root arena, contains the entry point for different modules. Each root

    // contains lists of IDs that correspond to the other arenas.

    pub(crate) protocol_descriptions: Arena<Root>,

    // Contents of a file, these are the elements the `Root` elements refer to

    pragmas: Arena<Pragma>,

    pub(crate) imports: Arena<Import>,

    pub(crate) variables: Arena<Variable>,

    pub(crate) definitions: Arena<Definition>,

    pub(crate) statements: Arena<Statement>,

    pub(crate) expressions: Arena<Expression>,

    pub(crate) scopes: Arena<Scope>,

}

impl Heap {

    pub fn new() -> Heap {

        Heap {

            // string_alloc: StringAllocator::new(),

            protocol_descriptions: Arena::new(),

            pragmas: Arena::new(),

            imports: Arena::new(),

            variables: Arena::new(),

            definitions: Arena::new(),

            statements: Arena::new(),

            expressions: Arena::new(),

            scopes: Arena::new(),

        }

    }

    pub fn alloc_memory_statement(

        &mut self,

        f: impl FnOnce(MemoryStatementId) -> MemoryStatement,

    ) -> MemoryStatementId {

        MemoryStatementId(LocalStatementId(self.statements.alloc_with_id(|id| {

            Statement::Local(LocalStatement::Memory(

                f(MemoryStatementId(LocalStatementId(id)))

            ))

        })))

    }

    pub fn alloc_channel_statement(

        &mut self,

        f: impl FnOnce(ChannelStatementId) -> ChannelStatement,

    ) -> ChannelStatementId {

        ChannelStatementId(LocalStatementId(self.statements.alloc_with_id(|id| {

            Statement::Local(LocalStatement::Channel(

                f(ChannelStatementId(LocalStatementId(id)))

            ))

        })))

    }

}

impl Index<MemoryStatementId> for Heap {

    type Output = MemoryStatement;

    fn index(&self, index: MemoryStatementId) -> &Self::Output {

        match &self.statements[index.0.0] {

            Statement::Local(LocalStatement::Memory(v)) => v,

            _ => unreachable!(),

        }

    }

}

impl Index<ChannelStatementId> for Heap {

    type Output = ChannelStatement;

    fn index(&self, index: ChannelStatementId) -> &Self::Output {

        match &self.statements[index.0.0] {

            Statement::Local(LocalStatement::Channel(v)) => v,

            _ => unreachable!(),

        }

    }

}

#[derive(Debug, Clone)]

pub struct Root {

    pub this: RootId,

    // Phase 1: parser

    // pub position: InputPosition,

    pub pragmas: Vec<PragmaId>,

    pub imports: Vec<ImportId>,

    pub definitions: Vec<DefinitionId>,

}

impl Root {

    pub fn get_definition_by_ident(&self, h: &Heap, id: &[u8]) -> Option<DefinitionId> {

        for &def in self.definitions.iter() {

            if h[def].identifier().value.as_bytes() == id {

                return Some(def);

            }

        }

        None

    }

}

#[derive(Debug, Clone)]

pub enum Pragma {

    Version(PragmaVersion),

    Module(PragmaModule),

}

impl Pragma {

    pub(crate) fn as_module(&self) -> &PragmaModule {

        match self {

            Pragma::Module(pragma) => pragma,

            _ => unreachable!("Tried to obtain {:?} as PragmaModule", self),

        }

    }

}

#[derive(Debug, Clone)]

pub struct PragmaVersion {

    pub this: PragmaId,

    // Phase 1: parser

    pub span: InputSpan, // of full pragma

    pub version: u64,

}

#[derive(Debug, Clone)]

pub struct PragmaModule {

    pub this: PragmaId,

    // Phase 1: parser

    pub span: InputSpan, // of full pragma

    pub value: Identifier,

}

#[derive(Debug, Clone)]

pub enum Import {

    Module(ImportModule),

    Symbols(ImportSymbols)

}

impl Import {

    pub(crate) fn span(&self) -> InputSpan {

        match self {

            Import::Module(v) => v.span,

            Import::Symbols(v) => v.span,

        }

    }

    pub(crate) fn as_module(&self) -> &ImportModule {

        match self {

            Import::Module(m) => m,

            _ => unreachable!("Unable to cast 'Import' to 'ImportModule'")

        }

    }

    pub(crate) fn as_symbols(&self) -> &ImportSymbols {

        match self {

            Import::Symbols(m) => m,

            _ => unreachable!("Unable to cast 'Import' to 'ImportSymbols'")

        }

    }

    pub(crate) fn as_symbols_mut(&mut self) -> &mut ImportSymbols {

        match self {

            Import::Symbols(m) => m,

            _ => unreachable!("Unable to cast 'Import' to 'ImportSymbols'")

        }

    }

}

#[derive(Debug, Clone)]

pub struct ImportModule {

    pub this: ImportId,

    // Phase 1: parser

    pub span: InputSpan,

    pub module: Identifier,

    pub alias: Identifier,

    pub module_id: RootId,

}

#[derive(Debug, Clone)]

pub struct AliasedSymbol {

    pub name: Identifier,

    pub alias: Option<Identifier>,

    pub definition_id: DefinitionId,

}

#[derive(Debug, Clone)]

pub struct ImportSymbols {

    pub this: ImportId,

    // Phase 1: parser

    pub span: InputSpan,

    pub module: Identifier,

    pub module_id: RootId,

    pub symbols: Vec<AliasedSymbol>,

}

#[derive(Debug, Clone)]

pub struct Identifier {

    pub span: InputSpan,

    pub value: StringRef<'static>,

}

impl Identifier {

    pub(crate) const fn new_empty(span: InputSpan) -> Identifier {

        return Identifier{

            span,

            value: StringRef::new_empty(),

        };

    }

}

impl PartialEq for Identifier {

    fn eq(&self, other: &Self) -> bool {

        return self.value == other.value

    }

}

impl Display for Identifier {