libc = { version = "^0.2", optional = true } # raw sockets, epoll access

## Handling Fatal Component Errors

Components may, during their execution, encounter errors that prevent them from continuing executing their code. For the purposes of this chapter we may consider these to occur during two particular phases of their execution:

1. The error occurred outside of a sync block. Or equivalently (from the point of view of the runtime): the error ocurred inside a sync block, but the component has not interacted with other components through `put`/`get` calls.

2. The error occurred inside of a sync block. The component can have performed any number of `put`/`get` calls. But for the sake of discussion we will only discuss the case where we perform:

   1. One `put` in the synchronous round.

   2. One `get` in the synchronous round.

As a preliminary remark: note that encountering an error is nothing special: the component can simply print an error to `stdout` and stop executing. The handling of the error by peers is of importance! If an interaction is made impossible because a peer has stopped executing, then the component that wishes to perform that interaction should error out itself!

### Handling Errors Outside of a Sync Block

If a component `E` encounters a critical error outside of a sync block. Then we can be sure that if it had a lat synchronous round, that it succeeded. However, there might be future synchronous rounds for component `E`, likewise a peer component `C` might have already put a message in `E`'s inbox.

The requirement for the outside-sync error of `E` is that any future sync interactions by `C` will fail (but, if `C` has no future interactions, it shouldn't fail either!). 

Note that `E` cannot perform `put`/`get` requests, because we're assuming `E` is outside of a sync block. Hence the only possible failing interaction is that `C` has performed a `put`, or is attempting a `get`. In the case the `C` `put`s to `E`, then `E` might not have figured out the identity of `C` yet (see earlier remarks on the eventual consistency of peer detection). Hence `C` is responsible for ensuring its own correct shutdown due to a failing `put`. Likewise for a `get`: `C` cannot receive from `E` if it is failing. So if `C` is waiting on a message to arrive, or if it will call `get` in the future, then `C` must fail as well.

In this case it is sufficient for `E` to send around a `ClosePort` message. As detailed in another chapter of this document. However, a particular race condition might occur. We have assumed that `E` is not in a sync block. But `C` is not aware of this fact. `C` might not be able to distinguish between the following three cases:

1. Regular shutdown: Components `C` and `E` are not in a sync round.

   - `E` broadcasts `ClosePort`.

   - `C` receives `ClosePort`.

2. Shutdown within a sync round, `ClosePort` leads `Solution`: A leader component `L`, peer component `C` and failing component `E`. Assume that all are/were busy in a synchronous round with one another.

   - `L` broadcasts `Solution` for the current sync round.

   - `E` receives `Solution`, finishes round. 

   - `E` encounters an error, so sends `ClosePort` to `C`.

   - `C` receives `ClosePort` from `E`.

   - `C` receives `Solution` from `L`.

3. Shutdown within a sync round, `Solution` leads `ClosePort`: Same components `L`, `C` and `E`.

   - `L` broadcasts `Solution` for the current sync round.

   - `E` receives `Solution` finishes round.

   - `E` encounters an error, so sends `ClosePort` to `C`.

   - `C` receives `Solution` from `L`.

   - `C` receives `ClosePort` from `E`.

In all described cases `E` encounters an error after finishing a sync round. But from the point of view of `C` it is unsure whether the `ClosePort` message pertains to the current synchronous round or not. In case 1 and 3 nothing is out of the ordinary. But in case 2 we have that `C` is at a particular point in time aware of the `ClosePort` from `E`, but not yet of the `Solution` from `L`. `C` should not fail the sync round, as it is completed, but it is unaware of this fact.

As a rather simple solution, since components that are participating with one another in a sync round move in lock-step at the end of the sync block, we send a boolean along with the `ClosePort`, e.g. `ClosePort(nonsync)`. This boolean indicates whether `E` was inside or outside of a sync block during it encountering an error. Now `C` can distinguish between the three cases: in all cases it agrees that `E` was not in a sync block (and hence: the sync round in cases 2 and 3 can be completed).

### Handling Errors Inside of a Sync Block

If `E` is inside of a sync block. Then it has interacted with other components. Our requirement now is that the sync round fails (and ofcourse, that all of the peers are notified that `E` will no longer be present in the runtime). There are two things that are complicating this type of failure:

1. Suppose that in the successful case of the synchronous interaction, there are a large number of components interacting with one another. Now it might be that `E` fails very early in its sync block, such that it cannot interact with several components. This lack of interaction might cause the single sync block to break up into several smaller sync blocks. Each of these separated regions is supposed to fail.

2. Within a particular synchronous interaction we might have that the leader `L` has a reference to the component `E` without it being a direct peer. There is a reference counting system in place that makes sure that `L` can always send messages to `E`. But we still need to make sure that those references stay alive for as long as needed. 

Suppose a synchronous region is (partially) established, and the component `E` encounters a critical error. The two points given above imply that two processes need to be initiated. For the first error-handling process, we simply use the same scheme as described in the case where `E` is not in a synchronous region. However now we broadcast `ClosePort(sync)` instead of `ClosePort(nonsync)` messages. Consider the following two cases: 

1. Component `C` is not part of the same synchronous region as `E`. And component `C` has tried `put`ting to `E`. If `C` receives a `ClosePort(sync)`, then it knows that its interaction should fail. Note: it might be that `E` wasn't planning on `get`ting from `C` in the sync round in which `E` failed, but much later. In that case it still makes sense for `C` to fail; it would have failed in the future. A small inconsistency here (within the current infinitely-deadlocking implementation) is that if `E` would *never* `get` from `C`, then `C` would deadlock instead of crash (one could argue that this implies that deadlocking should lead to crashing through a timeout mechanism).

2. Component `C` is not part of the same synchronous region as `E`. And if `E` wouldn't have crashed, then it would've `put` a message to `C`. In this case it is still proper for `C` to crash. The reasoning works the same as above.

So that is to say that this `ClosePort(sync)` causes instant failure of `C` if it has used the closed port in a round without consensus, or if it uses that port in the future. Note that this `ClosePort(sync)` system causes cascading failures throughout the disjoint synchronous regions. This is as intended: once one component's PDL program can no longer be executed, we cannot depend on the discovery of all the peers that constitute the intended synchronous region. So instead we rely on a peer-to-peer mechanism to make sure that every component is notified of failure.

However, while these cascading peer-to-peer `ClosePort(sync)` messages are happily shared around, we still have a leader component somewhere, and components that have not yet been notified of the failure.

    BlockGet(ExpressionId, PortId),

    Put(ExpressionId, PortId, ValueGroup),

    SelectRegisterPort(ExpressionId, u32, u32, PortId), // (call_expr_id, case_index, port_index_in_case, port_id)

                                            return Ok(EvalContinuation::BlockGet(expr_id, port_id));

                                        return Ok(EvalContinuation::Put(expr_id, port_id, value_group));

                                    let is_literal_string = value.get_heap_pos().is_some();

                                    return Ok(EvalContinuation::SelectRegisterPort(expr_id, case_index, port_index, port_value));

            let call_expression_id = create_ast_call_expr(ctx, InputSpan::new(), self.current_procedure_id, Method::SelectStart, &mut self.expression_buffer, arguments);

                    // Retrieve original span in case of error reporting for the

                    // inserted function call

                    let original_get_call_id = ctx.heap[id].cases[case_index].involved_ports[case_port_index].0;

                    let original_get_call_span = ctx.heap[original_get_call_id].full_span;

                    let runtime_call_expr_id = create_ast_call_expr(ctx, original_get_call_span, self.current_procedure_id, Method::SelectRegisterCasePort, &mut self.expression_buffer, runtime_call_arguments);

            let runtime_call_expr_id = create_ast_call_expr(ctx, InputSpan::new(), self.current_procedure_id, Method::SelectWait, &mut self.expression_buffer, Vec::new());

/// Creates an AST call expression. The provided expression span is useful for

/// the cases where we perform compiler-builtin function calls that, when they

/// fail, should provide an error pointing at a specific point in the source

/// code. The `containing_procedure_id` is the procedure whose instructions are

/// going to contain this new call expression.

fn create_ast_call_expr(ctx: &mut Ctx, span: InputSpan, containing_procedure_id: ProcedureDefinitionId, method: Method, buffer: &mut ScopedBuffer<ExpressionId>, arguments: Vec<ExpressionId>) -> CallExpressionId {

        full_span: span,

            EvalContinuation::BlockGet(_expr_id, port_id) => {

            EvalContinuation::Put(_expr_id, port_id, content) => {

    pub failed: bool,

    pub failed: bool,

/// Content of a control message. If the content refers to a port then the

/// `target_port_id` field is the one that it refers to.

    BlockPort,

    UnblockPort,

    ClosePort(ControlMessageClosePort),

    PortPeerChangedBlock,

    PortPeerChangedUnblock(PortId, CompId), // contains (new_port_id, new_component_id)

}

#[derive(Copy, Clone, Debug)]

pub struct ControlMessageClosePort {

    pub closed_in_sync_round: bool, // needed to ensure correct handling of errors

use std::fmt::{Display as FmtDisplay, Result as FmtResult, Formatter};

/// Potential error emitted by a component

pub enum CompError {

    /// Error originating from the code executor. Hence has an associated

    /// source location.

    Executor(EvalError),

    /// Error originating from a component, but not necessarily associated with

    /// a location in the source.

    Component(String), // TODO: Maybe a different embedded value in the future?

    /// Pure runtime error. Not necessarily originating from the component

    /// itself. Should be treated as a very severe runtime-compromising error.

    Runtime(RtError),

}

impl FmtDisplay for CompError {

    fn fmt(&self, f: &mut Formatter<'_>) -> FmtResult {

        match self {

            CompError::Executor(v) => v.fmt(f),

            CompError::Component(v) => v.fmt(f),

            CompError::Runtime(v) => v.fmt(f),

        }

    }

}

    fn run(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx) -> CompScheduling;

/// Representation of the generic operating mode of a component. Although not

/// every state may be used by every kind of (builtin) component, this allows

/// writing standard handlers for particular events in a component's lifetime.

    StartExit, // temporary state: if encountered then we start the shutdown process.

    BusyExit, // temporary state: waiting for Acks for all the closed ports, potentially waiting for sync round to finish

    pub(crate) fn is_busy_exiting(&self) -> bool {

        use CompMode::*;

        match self {

            NonSync | Sync | SyncEnd | BlockedGet | BlockedPut | BlockedSelect => false,

            StartExit | BusyExit => true,

            Exit => false,

        }

    }

}

#[derive(Debug)]

pub(crate) enum ExitReason {

    Termination, // regular termination of component

    ErrorInSync,

    ErrorNonSync,

}

impl ExitReason {

    pub(crate) fn is_in_sync(&self) -> bool {

        use ExitReason::*;

        match self {

            Termination | ErrorNonSync => false,

            ErrorInSync => true,

        }

    }

    pub(crate) fn is_error(&self) -> bool {

        use ExitReason::*;

        match self {

            Termination => false,

            ErrorInSync | ErrorNonSync => true,

        }

    }

    pub exit_reason: ExitReason, // valid if in StartExit/BusyExit/Exit mode

            exit_reason: ExitReason::Termination,

    pub(crate) fn set_as_start_exit(&mut self, reason: ExitReason) {

        self.mode = CompMode::StartExit;

        self.exit_reason = reason;

    }

// TODO: Replace when implementing port sending. Should probably be incorporated

//  into CompCtx (and rename CompCtx into CompComms)

pub(crate) type InboxMain = Vec<Option<DataMessage>>;

pub(crate) type InboxMainRef = [Option<DataMessage>];

pub(crate) type InboxBackup = Vec<DataMessage>;

    exec_state: &mut CompExecState, transmitting_port_id: PortId,

    port_instruction: PortInstruction, value: ValueGroup,

) -> Result<CompScheduling, (PortInstruction, String)> {

    let port_info = comp_ctx.get_port_mut(port_handle);

    port_info.last_instruction = port_instruction;

    if port_info.state.is_closed() {

        // Note: normally peer is eventually consistent, but if it has shut down

        // then we can be sure it is consistent (I think?)

        return Err((

            port_info.last_instruction,

            format!("Cannot send on this port, as the peer (id:{}) has shut down", port_info.peer_comp_id.0)

        ))

    } else if port_info.state.is_blocked() {

        return Ok(CompScheduling::Sleep);

        peer_info.handle.send_message_logged(sched_ctx, Message::Data(annotated_message), true);

        return Ok(CompScheduling::Immediate);

    exec_state: &mut CompExecState, inbox_main: &mut InboxMain,

    let port_handle = comp_ctx.get_port_handle(incoming_message.data_header.target_port);

    let port_index = comp_ctx.get_port_index(port_handle);

    comp_ctx.get_port_mut(port_handle).received_message_for_sync = true;

    let port_value_slot = &mut inbox_main[port_index];

        if port_info.state.is_open() {

            port_info.state.set(PortStateFlag::BlockedDueToFullBuffers);

            peer.handle.send_message_logged(sched_ctx, Message::Control(message), true);

pub(crate) enum GetResult {

    Received(DataMessage),

    NoMessage,

    Error((PortInstruction, String)),

}

/// Default attempt at trying to receive from a port (i.e. through a `get`, or

/// the equivalent operation for a builtin component). `target_port` is the port

/// we're trying to receive from, and the `target_port_instruction` is the

/// instruction we're attempting on this port.

pub(crate) fn default_attempt_get(

    exec_state: &mut CompExecState, target_port: PortId, target_port_instruction: PortInstruction,

    inbox_main: &mut InboxMainRef, inbox_backup: &mut InboxBackup, sched_ctx: &SchedulerCtx,

    comp_ctx: &mut CompCtx, control: &mut ControlLayer, consensus: &mut Consensus

) -> GetResult {

    let port_handle = comp_ctx.get_port_handle(target_port);

    let port_index = comp_ctx.get_port_index(port_handle);

    let port_info = comp_ctx.get_port_mut(port_handle);

    port_info.last_instruction = target_port_instruction;

    if port_info.state.is_closed() {

        let peer_id = port_info.peer_comp_id;

        return GetResult::Error((

            target_port_instruction,

            format!("Cannot get from this port, as the peer component (id:{}) closed the port", peer_id.0)

        ));

    }

    if let Some(message) = &inbox_main[port_index] {

        if consensus.try_receive_data_message(sched_ctx, comp_ctx, message) {

            // We're allowed to receive this message

            let message = inbox_main[port_index].take().unwrap();

            debug_assert_eq!(target_port, message.data_header.target_port);

            // Note: we can still run into an unrecoverable error when actually

            // receiving this message

            match default_handle_received_data_message(

                target_port, target_port_instruction, inbox_main, inbox_backup,

                comp_ctx, sched_ctx, control,

            ) {

                Ok(()) => return GetResult::Received(message),

                Err(location_and_message) => return GetResult::Error(location_and_message)

            }

        } else {

            // We're not allowed to receive this message. This means that the

            // receiver is attempting to receive something out of order with

            // respect to the sender.

            return GetResult::Error((target_port_instruction, String::from(

                "Cannot get from this port, as this causes a deadlock. This happens if you `get` in a different order as another component `put`s"

            )));

        }

    } else {

        // We don't have a message waiting for us and the port is not blocked.

        // So enter the BlockedGet state

        exec_state.set_as_blocked_get(target_port);

        return GetResult::NoMessage;

    }

}

    targeted_port: PortId, port_instruction: PortInstruction,

    inbox_main: &mut InboxMainRef, inbox_backup: &mut InboxBackup,

) -> Result<(), (PortInstruction, String)> {

    let port_handle = comp_ctx.get_port_handle(targeted_port);

    let port_index = comp_ctx.get_port_index(port_handle);

    let slot = &mut inbox_main[port_index];

    // Modify last-known location where port instruction was retrieved

    debug_assert_ne!(port_info.last_instruction, PortInstruction::None); // set by caller

    debug_assert!(port_info.state.is_open()); // checked by caller

    // Check if there are any more messages in the backup buffer

            debug_assert!(comp_ctx.get_port(port_handle).state.is_blocked()); // since we're removing another message from the backup

            return Ok(());

    if port_info.state.is_set(PortStateFlag::BlockedDueToFullBuffers) {

        let port_info = comp_ctx.get_port_mut(port_handle);

        port_info.state.clear(PortStateFlag::BlockedDueToFullBuffers);

        peer_info.handle.send_message_logged(sched_ctx, Message::Control(message), true);

    return Ok(());

) -> Result<(), (PortInstruction, String)> {

        ControlMessageContent::BlockPort => {

            let port_to_block = message.target_port_id.unwrap();

            let port_handle = comp_ctx.get_port_handle(port_to_block);

            let port_info = comp_ctx.get_port_mut(port_handle);

            port_info.state.set(PortStateFlag::BlockedDueToFullBuffers);

        ControlMessageContent::ClosePort(content) => {

            let port_to_close = message.target_port_id.unwrap();

            let port_handle = comp_ctx.get_port_handle(port_to_close);

            // We're closing the port, so we will always update the peer of the

            // port (in case of error messages)

            let port_info = comp_ctx.get_port_mut(port_handle);

            port_info.peer_comp_id = message.sender_comp_id;

            port_info.close_at_sync_end = true; // might be redundant (we might set it closed now)

            let peer_comp_id = port_info.peer_comp_id;

                // The two components (sender and this component) are closing

                // the channel at the same time. So we don't care about the

                // content of the `ClosePort` message.

                // Respond to the message

                let port_info = comp_ctx.get_port(port_handle);

                let last_instruction = port_info.last_instruction;

                let port_has_had_message = port_info.received_message_for_sync;

                comp_ctx.change_port_peer(sched_ctx, port_handle, None);

                // Handle any possible error conditions (which boil down to: the

                // port has been used, but the peer has died). If not in sync

                // mode then we close the port immediately.

                // Note that `port_was_used` does not mean that any messages

                // were actually received. It might also mean that e.g. the

                // component attempted a `get`, but there were no messages, so

                // now it is in the `BlockedGet` state.

                let port_was_used = last_instruction != PortInstruction::None;

                if exec_state.mode.is_in_sync_block() {

                    let closed_during_sync_round = content.closed_in_sync_round && port_was_used;

                    let closed_before_sync_round = !content.closed_in_sync_round && !port_has_had_message;

                    if closed_during_sync_round || closed_before_sync_round {

                        return Err((

                            last_instruction,

                            format!("Peer component (id:{}) shut down, so communication cannot (have) succeed(ed)", peer_comp_id.0)

                        ));

                    }

                } else {

                    let port_info = comp_ctx.get_port_mut(port_handle);

                    port_info.state.set(PortStateFlag::Closed);

                }

        ControlMessageContent::UnblockPort => {

            let port_to_unblock = message.target_port_id.unwrap();

            let port_handle = comp_ctx.get_port_handle(port_to_unblock);

            let port_info = comp_ctx.get_port_mut(port_handle);

            debug_assert!(port_info.state.is_set(PortStateFlag::BlockedDueToFullBuffers));

            port_info.state.clear(PortStateFlag::BlockedDueToFullBuffers);

            default_handle_recently_unblocked_port(exec_state, consensus, port_handle, sched_ctx, comp_ctx);

        ControlMessageContent::PortPeerChangedBlock => {

            let port_to_change = message.target_port_id.unwrap();

            let port_handle = comp_ctx.get_port_handle(port_to_change);

            let port_info = comp_ctx.get_port_mut(port_handle);

            let peer_comp_id = port_info.peer_comp_id;

            port_info.state.set(PortStateFlag::BlockedDueToPeerChange);

            let peer_handle = comp_ctx.get_peer_handle(peer_comp_id);

            let port_to_change = message.target_port_id.unwrap();

            let port_handle = comp_ctx.get_port_handle(port_to_change);

            debug_assert!(port_info.state.is_set(PortStateFlag::BlockedDueToPeerChange));

            port_info.state.clear(PortStateFlag::BlockedDueToPeerChange);

            comp_ctx.change_port_peer(sched_ctx, port_handle, Some(new_comp_id));

            default_handle_recently_unblocked_port(exec_state, consensus, port_handle, sched_ctx, comp_ctx);

    return Ok(());

}

/// Handles a component entering the synchronous block. Will ensure that the

/// `Consensus` and the `ComponentCtx` are initialized properly.

pub(crate) fn default_handle_sync_start(

    exec_state: &mut CompExecState, inbox_main: &mut InboxMainRef,

    sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, consensus: &mut Consensus

) {

    sched_ctx.info("Component starting sync mode");

    // If any messages are present for this sync round, set the appropriate flag

    // and notify the consensus handler of the present messages

    consensus.notify_sync_start(comp_ctx);

    for (port_index, message) in inbox_main.iter().enumerate() {

        if let Some(message) = message {

            consensus.handle_incoming_data_message(comp_ctx, message);

            let port_info = comp_ctx.get_port_by_index_mut(port_index);

            port_info.received_message_for_sync = true;

        }

    }

    // Modify execution state

    debug_assert_eq!(exec_state.mode, CompMode::NonSync);

    exec_state.mode = CompMode::Sync;

}

/// Handles a component that has reached the end of the sync block. This does

/// not necessarily mean that the component will go into the `NonSync` mode, as

/// it might have to wait for the leader to finish the round for everyone (see

/// `default_handle_sync_decision`)

pub(crate) fn default_handle_sync_end(

    exec_state: &mut CompExecState, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx,

    consensus: &mut Consensus

) {

    sched_ctx.info("Component ending sync mode (but possibly waiting for a solution)");

    debug_assert_eq!(exec_state.mode, CompMode::Sync);

    let decision = consensus.notify_sync_end_success(sched_ctx, comp_ctx);

    exec_state.mode = CompMode::SyncEnd;

    default_handle_sync_decision(sched_ctx, exec_state, comp_ctx, decision, consensus);

    sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, consensus: &mut Consensus

    sched_ctx.info(&format!("Component starting exit (reason: {:?})", exec_state.exit_reason));

    let exit_inside_sync = exec_state.exit_reason.is_in_sync();

    // If exiting while inside sync mode, report to the leader of the current

    // round that we've failed.

    if exit_inside_sync {

        let decision = consensus.notify_sync_end_failure(sched_ctx, comp_ctx);

        default_handle_sync_decision(sched_ctx, exec_state, comp_ctx, decision, consensus);

    }

    // Iterating over ports by index to work around borrowing rules

        if port.state.is_closed() || port.close_at_sync_end {