return;

### The cases in which peers crash in response

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

Because the crashing component is potentially unaware of the component IDs it will end up notifying that it has failed, we can not design the crash-handling algorithm in such a way such that the crashing component notifies the peers of when they have to crash. We'll do the opposite: the crashing component simply crashes and somehow attempts to notify the peers. Those peers themselves decide whether they have to crash in response to such a notification.

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, like the other control algorithms. the failing component will wait for the port that is shutting down to become unblocked before it will send the `ClosePort` message).

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. However, the handling component itself might've performed `put` operations already. So now that the handling component receives a `ClosePort` message, it realizes that those earlier `put` operations can never be acknowledged. For this reason a component stores when it last used a port in the metadata associated with a port. When, in this second case, a `ClosePort` message comes in while the port has been used already, the handling component should crash as well.

Next up is the third case, where both the crashing component and the handling component were both in the same synchronous round. Like before we mark the port as closed and future use will cause a crash. Like the second case, if the handling component has already used a port (which in this case may also be having received a message from the crashing component), then it should crash as well.

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 (assumed to be running somewhere in a thread of execution different from the two components). 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.

### Distinguishing the crashing cases

So far we've pretended like we could already determine the relation between the crashing component's synchronous round and the handling component's synchronous round. But in order to do this we need to add a bit of extra information to the `ClosePort` message.

The simplest case is to determine if the two components are both in the same synchronous round (case three, as described above). The crashing component annotates the `ClosePort` message with whether it was in a synchronous round or not. Then if both components are in a synchronous round (as checking by the handling component), and the about-to-be-closed port at the handling component was used in that round, or will be used in that round, then the handling component should crash.

Equally simple: the handling component can figure out itself if it is in a synchronous round (case one, as described above). If not: then the port is marked closed and future use causes crashes.

The last two cases require a bit more work: how do we distinguish the edge case where the handling component's round will complete in the future, from the case where it should crash. To distinguish the edge case we need the handling component to know if the last interaction the crashing component handled was the one in the handling component's current synchronous round.

For this reason we keep track of the synchronous round number. That is to say: there is a counter that increments each time a synchronous round completes for a component. We have a field in the metadata for a port that registers this round number. If a component performs a `put` operation, then it stores its own round number in that port's metadata, and sends this round number along with the message. If a component performs a `get` operation, then it stores the *received* round number in the port's metadata.

When a component closes a port, it will also send along the last registered round number in the `ClosePort` message. If the handling component receives a `ClosePort` message, and the last registered round number in the port's metadata matches the round number in the `ClosePort` message, and the crashing component was not in a synchronous round, then the crashing component crashed after the handling component's sync round. Hence: the handling component can complete its sync round.

To conclude: if we receive a `ClosePort` message, then we always mark the port as closed. If the handling and the crashing component were in a synchronous round, and the closed port was used in that synchronous round, then the handling component crashes as well. If the handling component *is* in a synchronous round but the crashing component *is not* in a synchronous round, the port of the handling component is used in the synchronous round and the port's last registered round number does not match the round number in the `ClosePort` message, then the handling component crashes as well.

## Component Nomenclature

Earlier versions of the Reowolf runtime featured the distinction between primitive and composite components. This was put into the language from a design perspective. Primitive components could do nitty-gritty protocol execution: perform `put`/`get` operations, and entering into sync blocks. Conversely, composite components were tasked with setting up a network of interconnected components: creating channels and handing off the appropriate ports to the instantiated components.

Once the runtime was capable of sending ports over channels, it became apparent that this distinction no longer made sense. Because if only primitive components can send/receive ports, and cannot create new components, then the programmer is limited to using those received ports directly in the primitive's code! And so the split between primitive and composite components was removed: only the concept of a "component" is left.

- The TCP listener and TCP sender components have not been tested extensively in a multi-threaded setup.

- The work queue in the runtime is still a mutex-locked queue. The `QueueMpsc` type should be extended to be a multiple-producer multiple-consumer queue. This type should then replace the mutex-locked work queue.

    pub fn maybe_span(&self) -> Option<InputSpan> {

            Statement::Block(v) => Some(v.span),

            Statement::Local(v) => Some(v.span()),

            Statement::Labeled(v) => Some(v.label.span),

            Statement::If(v) => Some(v.span),

            Statement::While(v) => Some(v.span),

            Statement::Break(v) => Some(v.span),

            Statement::Continue(v) => Some(v.span),

            Statement::Synchronous(v) => Some(v.span),

            Statement::Fork(v) => Some(v.span),

            Statement::Select(v) => Some(v.span),

            Statement::Return(v) => Some(v.span),

            Statement::Goto(v) => Some(v.span),

            Statement::New(v) => Some(v.span),

            Statement::Expression(v) => Some(v.span),

            | Statement::EndSelect(_) => None,

    pub fn span(&self) -> InputSpan {

        return self.maybe_span().unwrap();

    }

    pub line: Option<u32>,

        let line_str = match self.line {

            Some(line_number) => line_number.to_string(),

            None => String::from("??"),

        };

            write!(f, "{} {}:{}", func_or_comp, &self.procedure, line_str)

            write!(f, "{} {}:{}:{}", func_or_comp, &self.module_name, &self.procedure, line_str)

            let statement_span = statement.maybe_span();

                line: statement_span.map(|v| v.begin.line),

#[derive(Debug)]

        let stmt = &mut ctx.heap[id];

        stmt.in_sync = self.in_sync;

    pub registered_round: Option<u32>,

    port_info.last_registered_round = Some(consensus.round_number());

        // Transfer messages to main/backup inbox. Make sure to modify the

        // target port to our own

        for message in received_port.messages.iter_mut() {

            message.data_header.target_port = new_port.self_id;

        }

        let _new_inbox_index = inbox_main.len();

        if !received_port.messages.is_empty() {

            inbox_main.push(Some(received_port.messages.remove(0)));

            inbox_backup.extend(received_port.messages.drain(..));

        } else {

            inbox_main.push(None);

        }

        debug_assert_eq!(_new_inbox_index, comp_ctx.get_port_index(new_port_handle));

    let port_info = comp_ctx.get_port_mut(port_handle);

    port_info.last_registered_round = Some(message.sync_header.sync_round);

                let last_registered_round = port_info.last_registered_round;

                    let round_has_succeeded = !content.closed_in_sync_round && last_registered_round == content.registered_round;

                    let closed_during_sync_round = content.closed_in_sync_round;

                    let closed_before_sync_round = !closed_during_sync_round && !round_has_succeeded;

                    if (closed_during_sync_round || closed_before_sync_round) && port_was_used {

    pub last_registered_round: Option<u32>,

            last_registered_round: None,

            last_registered_round: None,

            last_registered_round: None,

    #[inline]

    pub(crate) fn round_number(&self) -> u32 {

        return self.round_index;

    }

        // debug_assert_eq!(self.mode, Mode::SyncBusy);

                    registered_round: port.last_registered_round,

        if !set_socket_reuse_address(socket_handle) {

            unsafe{ libc::close(socket_handle); }

            return Err(SocketError::Modifying);

        }

#[inline]

fn set_socket_reuse_address(handle: libc::c_int) -> bool {

    if handle < 0 {

        return false;

    }

    unsafe {

        let enable: libc::c_int = 1;

        let enable_ptr: *const _ = &enable;

        let result = libc::setsockopt(

            handle, libc::SOL_SOCKET, libc::SO_REUSEADDR,

            enable_ptr.cast(), size_of::<libc::c_int>() as libc::socklen_t

        );

        if result < 0 {

            return false;

        }

        let result = libc::setsockopt(

            handle, libc::SOL_SOCKET, libc::SO_REUSEPORT,

            enable_ptr.cast(), size_of::<libc::c_int>() as libc::socklen_t

        );

        if result < 0 {

            return false;

        }

    }

    return true;

}

    let runtime = Runtime::new(NUM_THREADS, LOG_LEVEL, protocol)

// Previously the following example caused the inadvertent synchronization of

// all participating components

func lots_of_unneccesary_work(u64 min_val, u64 max_val) -> u64 {

    u64 sum = 0;

    u64 index = min_val;

    while (index <= max_val) {

        sum += index;

        index += 1;

    }

    return sum;

}

comp data_producer(out<u64> tx, u64 min_val, u64 max_val, string send_text) {

    while (true) {

        sync {

            auto value = lots_of_unneccesary_work(min_val, max_val);

            print(send_text);

            put(tx, value);

        }

    }

}

comp data_receiver_v1(in<u64> rx_a, in<u64> rx_b, in<u64> rx_c, u32 num_rounds) {

    u32 counter = 0;

    auto rxs = { rx_a, rx_b, rx_c };

    while (counter < num_rounds) {

        auto num_peers = length(rxs);

        auto peer_index = 0;

        while (peer_index < num_peers) {

            sync {

                auto result = get(rxs[peer_index]);

                print("received message (V1)");

                peer_index += 1;

            }

        }

        counter += 1;

    }

}

// The reason was that a synchronous interaction checked *all* ports for a valid

// interaction. So for the `round_robin_receiver` we have that it communicates

// with one peer per round, but it still requires the other peers to agree that

// they didn't send anything at all! Note that this already implies that all

// running components need to synchronize. We could fix this by writing:

comp data_receiver_v2(in<u64> rx_a, in<u64> rx_b, in<u64> rx_c, u32 num_rounds) {

    u32 counter = 0;

    auto rxs = { rx_a, rx_b, rx_c };

    while (counter < num_rounds) {

        auto num_peers = length(rxs);

        auto peer_index = 0;

        sync {

            while (peer_index < num_peers) {

                auto result = get(rxs[peer_index]);

                print("received message (V2)");

                peer_index += 1;

            }

        }

        counter += 1;

    }

}

// But this is not the intended behaviour. We want the producer components to

// be able to run independently of one another. This requires a change in the

// semantics of the language! We no longer have that each peer is automatically

// dragged into the synchronous round. Instead, once the first message of the

// peer is received through a `get` call, will we join each other's synchronous

// rounds.

//

// With such a change to the runtime, we now have that the first version (

// written above) produces the intended behaviour: the consumer accepts one

// value and synchronizes with its sender. Then goes to the next round and 4

// synchronizes with the next sender.

//

// But what we would really like to do is to synchronize with any of the peers

// that happens to have its work ready for consumption. And so the 'select'

// statement is introduced into the language. This statement can be used to

// describe a set of possible behaviours we could execute. Each behaviour will

// have an associated set of ports. When those associated set of ports have a

// message ready to be read, then the corresponding behaviour will execute. So

// to complete the example above, we have:

comp data_receiver_v3(in<u64> rx_a, in<u64> rx_b, in<u64> rx_c, u32 num_rounds) {

    u32 counter = 0;

    auto rxs = { rx_a, rx_b, rx_c };

    u32 received_from_a = 0;

    u32 received_from_b_or_c = 0;

    u32 received_from_a_or_c = 0;

    u64 sum_received_from_c = 0;

    while (counter < num_rounds*3) {

        sync {

            select {

                auto value = get(rx_a) -> {

                    received_from_a += 1;

                    received_from_a_or_c += 1;

                }

                auto value = get(rx_b) -> {

                    received_from_b_or_c += 1;

                }

                auto value = get(rx_c) -> {

                    received_from_a_or_c += 1;

                    received_from_b_or_c += 1;

                    sum_received_from_c += value;

                }

            }

            print("received message (V3)");

        }

        counter += 1;

    }

}

comp main() {

    u32 version = 3;

    u32 num_rounds = 3;

    channel tx_a -> rx_a;

    channel tx_b -> rx_b;

    channel tx_c -> rx_c;

    new data_producer(tx_a, 0xBEEF, 0xCAFE, "sent from A");

    new data_producer(tx_b, 0xBEEF, 0xCAFE, "sent from B");

    new data_producer(tx_c, 0xBEEF, 0xCAFE, "sent from C");

    if (version == 1) {

        new data_receiver_v1(rx_a, rx_b, rx_c, num_rounds);

    } else if (version == 2) {

        new data_receiver_v2(rx_a, rx_b, rx_c, num_rounds);

    } else if (version == 3) {

        new data_receiver_v3(rx_a, rx_b, rx_c, num_rounds);

    } else {

        print("ERROR: invalid version in source");

    }

}

// Although in an unstable state, there is an initial implementation for error

// handling. Roughly speaking: if a component has failed then it cannot complete

// any current or future synchronous rounds anymore. Hence, apart from some edge

// cases, any received message by a peer should cause a failure at that peer as

// well. We may have a look at the various places where a component with respect

// to a peer that is receiving its messages

enum ErrorLocation {

    BeforeSync,

    DuringSyncBeforeFirstInteraction,

    DuringSyncBeforeSecondInteraction,

    DuringSyncAfterInteractions,

    AfterSync,

}

func error_location_to_string(ErrorLocation loc) -> string {

    if (let ErrorLocation::BeforeSync = loc) {

        return "before sync";

    } else if (let ErrorLocation::DuringSyncBeforeFirstInteraction = loc) {

        return "during sync before first interaction";

    } else if (let ErrorLocation::DuringSyncBeforeSecondInteraction = loc) {

        return "during sync before second interaction";

    } else if (let ErrorLocation::DuringSyncAfterInteractions = loc) {

        return "during sync after interactions";

    } else { return "after sync"; }

}

func crash() -> u8 {

    return {}[0]; // access index 1 of an empty array

}

comp sender_and_crasher(out<u32> value, ErrorLocation loc) {

    print("sender: will crash " @ error_location_to_string(loc));

    if (loc == ErrorLocation::BeforeSync) { crash(); }

    sync {

        if (loc == ErrorLocation::DuringSyncBeforeFirstInteraction) { crash(); }

        print("sender: sending first value");

        put(value, 0);

        if (loc == ErrorLocation::DuringSyncBeforeSecondInteraction) { crash(); }

        print("sender: sending second value");

        put(value, 1);

        if (loc == ErrorLocation::DuringSyncAfterInteractions) { crash(); }

    }

    if (loc == ErrorLocation::AfterSync) { crash(); }

}

comp receiver(in<u32> value) {

    sync {

        auto a = get(value);

        auto b = get(value);

    }

}

// Note that when we run the example with the error location before sync, or

// during sync, that the receiver always crashes. However the location where it

// will crash is somewhat random! Due to the asynchronous nature of the runtime

// a sender of messages will always just `put` the value onto the port and

// continue execution. So even though the sender component might already be done

// with its sync round, the receiver officially still has to receive its first

// message. In any case, a neat error message should be displayed in the

// console.

//

// Note especially, given the asynchronous nature of the runtime, that the

// receiver should figure out when the peer component has crashed, but it can

// still finish the current synchronous round. This might happen if the peer

// component crashes *just* after the synchronous round. There may be a case

// where the peer receives the information that the peer crashed *before* it

// receives the information that the synchronous round has succeeded.

comp main() {

    channel tx -> rx;

    new sender_and_crasher(tx, ErrorLocation::AfterSync);

    new receiver(rx);

}

// Since this release transmitting ports is possible. This means that we can

// send ports through ports. In fact, we can send ports that may send ports that

// may send ports. But don't be fooled by the apparent complexity. The inner

// type `T` of a port like `in<T>` simply states that that is the message type.

// Should the type `T` contain one or more ports, then we kick off a bit of code

// that takes care of the transfer of the port. Should the port inside of `T`

// itself, after being received, send a port, then we simply kick off that same

// procedure again.

//

// In the simplest case, we have someone transmitting the receiving end of a

// channel to another component, which then uses that receiving end to receive a

// value.

comp port_sender(out<in<u32>> tx, in<u32> to_transmit) {

    sync put(tx, to_transmit);

}

comp port_receiver_and_value_getter(in<in<u32>> rx, u32 expected_value) {

    u32 got_value = 0;

    sync {

        auto port = get(rx);

        got_value = get(port);

    }

    if (expected_value == got_value) {

        print("got the expected value :)");

    } else {

        print("got a different value :(");

    }

}

comp value_sender(out<u32> tx, u32 to_send) {

    sync put(tx, to_send);

}

comp main() {

    u32 value = 1337_2392;

    channel port_tx -> port_rx;

    channel value_tx -> value_rx;

    new port_sender(port_tx, value_rx);

    new port_receiver_and_value_getter(port_rx, value);

    new value_sender(value_tx, value);

}

// Ofcourse we may do something a little more complicated than this. Suppose

// that we don't just send one port, but send a series of ports. i.e. we use

// an `Option` union type, to ...

union Option<T> {

    Some(T),

    None,

}

// ... turn an array of ports that we're going to transmit into a series of

// messages containing ports, each sent to a specific component.

comp port_sender(out<Option<in<u32>>>[] txs, in<u32>[] to_transmit) {

    auto num_peers = length(txs);

    auto num_ports = length(to_transmit);

    auto num_per_peer = num_ports / num_peers;

    auto num_remaining = num_ports - (num_per_peer * num_peers);

    auto peer_index = 0;

    auto port_index = 0;

    while (peer_index < num_peers) {

        auto peer_port = txs[peer_index];

        auto counter = 0;

        // Distribute part of the ports to one of the peers.

        sync {

            // Sending the main batch of ports for the peer

            while (counter < num_per_peer) {

                put(peer_port, Option::Some(to_transmit[port_index]));

                port_index += 1;

                counter += 1;

            }

            // Sending the remainder of ports, one per peer until they're gone

            if (num_remaining > 0) {

                put(peer_port, Option::Some(to_transmit[port_index]));

                port_index += 1;

                num_remaining -= 1;

            }

            // Finish the custom protocol by sending nothing, which indicates to

            // the peer that it has received all the ports we have to hand out.

            put(peer_port, Option::None);

        }

        peer_index += 1;

    }

}

// And here we have the component which will receive on that port. We can design

// the synchronous regions any we want. In this case when we receive ports we

// just synchronize `port_sender`, but the moment we receive messages we

// synchronize with everyone.

comp port_receiver(in<Option<in<u32>>> port_rxs, out<u32> sum_tx) {

    // Receive all ports

    auto value_rxs = {};

    sync {

        while (true) {

            auto maybe_port = get(port_rxs);

            if (let Option::Some(certainly_a_port) = maybe_port) {

                value_rxs @= { certainly_a_port };

            } else {

                break;

            }

        }

    }

    // Receive all values

    auto received_sum = 0;

    sync {

        auto port_index = 0;

        auto num_ports = length(value_rxs);

        while (port_index < num_ports) {

            auto value = get(value_rxs[port_index]);

            received_sum += value;

            port_index += 1;

        }

    }

    // And send the sum

    sync put(sum_tx, received_sum);

}

// Now we need something to send the values, we'll make something incredibly

// simple. Namely:

comp value_sender(out<u32> tx, u32 value_to_send) {

    sync put(tx, value_to_send);

}

comp sum_collector(in<u32>[] partial_sum_rx, out<u32> total_sum_tx) {

    auto sum = 0;

    auto index = 0;

    while (index < length(partial_sum_rx)) {

        sync sum += get(partial_sum_rx[index]);

        index += 1;

    }