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

   }

}

```

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.

    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.

    While(WhileStatement),

    EndWhile(EndWhileStatement),

    Break(BreakStatement),

    Continue(ContinueStatement),

    Synchronous(SynchronousStatement),

    EndSynchronous(EndSynchronousStatement),

    Fork(ForkStatement),

    EndFork(EndForkStatement),

    Select(SelectStatement),

    EndSelect(EndSelectStatement),

    Return(ReturnStatement),

    Goto(GotoStatement),

    New(NewStatement),

    Expression(ExpressionStatement),

}

impl Statement {

    pub fn as_new(&self) -> &NewStatement {

        match self {

            Statement::New(result) => result,

            _ => panic!("Unable to cast `Statement` to `NewStatement`"),

        }

    }

        match self {

            Statement::EndBlock(_)

            | Statement::EndIf(_)

            | Statement::EndWhile(_)

            | Statement::EndSynchronous(_)

            | Statement::EndFork(_)

        }

    }

    pub fn link_next(&mut self, next: StatementId) {

        match self {

            Statement::Block(stmt) => stmt.next = next,

            Statement::EndBlock(stmt) => stmt.next = next,

            Statement::Local(stmt) => match stmt {

                LocalStatement::Channel(stmt) => stmt.next = next,

                LocalStatement::Memory(stmt) => stmt.next = next,

            },

            Statement::EndIf(stmt) => stmt.next = next,

            Statement::EndWhile(stmt) => stmt.next = next,

            Statement::EndSynchronous(stmt) => stmt.next = next,

            Statement::EndFork(stmt) => stmt.next = next,

            Statement::EndSelect(stmt) => stmt.next = next,

            Statement::New(stmt) => stmt.next = next,

            Statement::Expression(stmt) => stmt.next = next,

            Statement::Return(_)

            | Statement::Break(_)

            | Statement::Continue(_)

            | Statement::Synchronous(_)

            | Statement::Fork(_)

            | Statement::Select(_)

            | Statement::Goto(_)

            | Statement::While(_)

            | Statement::Labeled(_)

use std::fmt;

use crate::protocol::{

    ast::*,

    Module,

    input_source::{ErrorStatement, StatementKind}

};

use super::executor::*;

/// Represents a stack frame recorded in an error

#[derive(Debug)]

pub struct EvalFrame {

    pub module_name: String,

    pub procedure: String, // function or component

    pub is_func: bool,

}

impl fmt::Display for EvalFrame {

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

        let func_or_comp = if self.is_func {

            "function "

        } else {

            "component"

        };

        if self.module_name.is_empty() {

        } else {

        }

    }

}

/// Represents an error that ocurred during evaluation. Contains error

/// statements just like in parsing errors. Additionally may display the current

/// execution state.

#[derive(Debug)]

pub struct EvalError {

    pub(crate) statements: Vec<ErrorStatement>,

    pub(crate) frames: Vec<EvalFrame>,

}

impl EvalError {

    pub(crate) fn new_error_at_expr(prompt: &Prompt, modules: &[Module], heap: &Heap, expr_id: ExpressionId, msg: String) -> EvalError {

        // Create frames

        debug_assert!(!prompt.frames.is_empty());

        let mut frames = Vec::with_capacity(prompt.frames.len());

        let mut last_module_source = &modules[0].source;

        for frame in prompt.frames.iter() {

            let definition = &heap[frame.definition];

            let statement = &heap[frame.position];

            // Lookup module name, if it has one

            let module = modules.iter().find(|m| m.root_id == definition.defined_in).unwrap();

            let module_name = if let Some(name) = &module.name {

                name.as_str().to_string()

            } else {

                String::new()

            };

            last_module_source = &module.source;

            frames.push(EvalFrame{

                module_name,

                procedure: definition.identifier.value.as_str().to_string(),

                is_func: definition.kind == ProcedureKind::Function,

            });

        }

        let expr = &heap[expr_id];

        let statements = vec![

            ErrorStatement::from_source_at_span(StatementKind::Error, last_module_source, expr.full_span(), msg)

        ];

        EvalError{ statements, frames }

    }

}

impl fmt::Display for EvalError {

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

        // Display error statement(s)

        self.statements[0].fmt(f)?;

        for statement in self.statements.iter().skip(1) {

            writeln!(f)?;

            statement.fmt(f)?;

        }

 * statement" field. Because it is so common, this file contain two macros that

 * perform that operation.

 *

 * To make storing types for polymorphic procedures simpler and more efficient,

 * we assign to each expression in the procedure a unique ID. This is what the

 * "next expression index" field achieves. Each expression simply takes the

 * current value, and then increments this counter.

 */

use crate::collections::{ScopedBuffer};

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

use crate::protocol::input_source::*;

use crate::protocol::parser::symbol_table::*;

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

use super::visitor::{

    BUFFER_INIT_CAP_SMALL,

    BUFFER_INIT_CAP_LARGE,

    Ctx,

    Visitor,

    VisitorResult

};

use crate::protocol::parser::ModuleCompilationPhase;

struct ControlFlowStatement {

    in_sync: SynchronousStatementId,

    in_while: WhileStatementId,

    in_scope: ScopeId,

    statement: StatementId, // of 'break', 'continue' or 'goto'

}

/// This particular visitor will go through the entire AST in a recursive manner

/// and check if all statements and expressions are legal (e.g. no "return"

/// statements in component definitions), and will link certain AST nodes to

/// their appropriate targets (e.g. goto statements, or function calls).

///

/// This visitor will not perform control-flow analysis (e.g. making sure that

/// each function actually returns) and will also not perform type checking. So

/// the linking of function calls and component instantiations will be checked

/// and linked to the appropriate definitions, but the return types and/or

/// arguments will not be checked for validity.

///

/// The main idea is, because we're visiting nodes in a tree, to do as much as

/// we can while we have the memory in cache.

pub(crate) struct PassValidationLinking {

    // Traversal state, all valid IDs if inside a certain AST element. Otherwise

    // `id.is_invalid()` returns true.

    in_sync: SynchronousStatementId,

        self.expr_parent = ExpressionParent::None;

        // Visit true and false branch. Executor chooses next statement based on

        // test expression, not on if-statement itself. Hence the if statement

        // does not have a static subsequent statement.

        assign_then_erase_next_stmt!(self, ctx, id.upcast());

        let old_scope = self.push_scope(ctx, false, true_case.scope);

        self.visit_stmt(ctx, true_case.body)?;

        self.pop_scope(old_scope);

        assign_then_erase_next_stmt!(self, ctx, end_if_id.upcast());

        if let Some(false_case) = false_case {

            let old_scope = self.push_scope(ctx, false, false_case.scope);

            self.visit_stmt(ctx, false_case.body)?;

            self.pop_scope(old_scope);

            assign_then_erase_next_stmt!(self, ctx, end_if_id.upcast());

        }

        self.prev_stmt = end_if_id.upcast();

        Ok(())

    }

    fn visit_while_stmt(&mut self, ctx: &mut Ctx, id: WhileStatementId) -> VisitorResult {

        let end_while_id = stmt.end_while;

        let test_expr_id = stmt.test;

        let body_stmt_id = stmt.body;

        let scope_id = stmt.scope;

        let old_while = self.in_while;

        self.in_while = id;

        // Visit test expression

        debug_assert_eq!(self.expr_parent, ExpressionParent::None);

        debug_assert!(self.in_test_expr.is_invalid());

        self.in_test_expr = id.upcast();

        self.expr_parent = ExpressionParent::While(id);

        self.visit_expr(ctx, test_expr_id)?;

        self.in_test_expr = StatementId::new_invalid();

        // Link up to body statement

        assign_then_erase_next_stmt!(self, ctx, id.upcast());

        self.expr_parent = ExpressionParent::None;

        let old_scope = self.push_scope(ctx, false, scope_id);

        self.visit_stmt(ctx, body_stmt_id)?;

        self.pop_scope(old_scope);

        self.in_while = old_while;

        // Link final entry in while's block statement back to the while. The

        // executor will go to the end-while statement if the test expression

        // is false, so put that in as the new previous stmt

#[derive(Debug)]

pub struct ControlMessage {

    pub(crate) id: ControlId,

    pub sender_comp_id: CompId,

    pub target_port_id: Option<PortId>,

    pub content: ControlMessageContent,

}

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

#[derive(Copy, Clone, Debug)]

pub enum ControlMessageContent {

    Ack,

    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

}

// -----------------------------------------------------------------------------

// Messages (generic)

// -----------------------------------------------------------------------------

#[derive(Debug)]

pub struct MessageSyncHeader {

    pub sync_round: u32,

    pub sending_id: CompId,

    pub highest_id: CompId,

}

#[derive(Debug)]

pub enum Message {

    Data(DataMessage),

    Sync(SyncMessage),

    Control(ControlMessage),

    Poll,

}

impl Message {

    pub(crate) fn target_port(&self) -> Option<PortId> {

        match self {

    }

}

// -----------------------------------------------------------------------------

// Generic component messaging utilities (for sending and receiving)

// -----------------------------------------------------------------------------

/// Default handling of sending a data message. In case the port is blocked then

/// the `ExecState` will become blocked as well. Note that

/// `default_handle_control_message` will ensure that the port becomes

/// unblocked if so instructed by the receiving component. The returned

/// scheduling value must be used.

#[must_use]

pub(crate) fn default_send_data_message(

    exec_state: &mut CompExecState, transmitting_port_id: PortId,

    port_instruction: PortInstruction, value: ValueGroup,

    sched_ctx: &SchedulerCtx, consensus: &mut Consensus,

    control: &mut ControlLayer, comp_ctx: &mut CompCtx

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

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

    let port_handle = comp_ctx.get_port_handle(transmitting_port_id);

    let port_info = comp_ctx.get_port_mut(port_handle);

    port_info.last_instruction = port_instruction;

    let port_info = comp_ctx.get_port(port_handle);

    debug_assert_eq!(port_info.kind, PortKind::Putter);

    let mut ports = Vec::new();

    find_ports_in_value_group(&value, &mut ports);

    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 !ports.is_empty() {

        start_send_message_with_ports(

            transmitting_port_id, port_instruction, value, exec_state,

            comp_ctx, sched_ctx, control

        )?;

        return Ok(CompScheduling::Sleep);

    } else if port_info.state.is_blocked() {

        // Port is blocked, so we cannot send

        exec_state.set_as_blocked_put_without_ports(transmitting_port_id, value);

        return Ok(CompScheduling::Immediate);

    }

}

pub(crate) enum IncomingData {

    PlacedInSlot,

    SlotFull(DataMessage),

}

/// Default handling of receiving a data message. In case there is no room for

/// the message it is returned from this function. Note that this function is

/// different from PDL code performing a `get` on a port; this is the case where

/// the message first arrives at the component.

// NOTE: This is supposed to be a somewhat temporary implementation. It would be

//  nicest if the sending component can figure out it cannot send any more data.

#[must_use]

pub(crate) fn default_handle_incoming_data_message(

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

    comp_ctx: &mut CompCtx, incoming_message: DataMessage,

    sched_ctx: &SchedulerCtx, control: &mut ControlLayer

) -> IncomingData {

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

    let port_index = comp_ctx.get_port_index(port_handle);

    let port_value_slot = &mut inbox_main[port_index];

    let target_port_id = incoming_message.data_header.target_port;

    if port_value_slot.is_none() {

        // We can put the value in the slot

        *port_value_slot = Some(incoming_message);

        // Check if we're blocked on receiving this message.

        dbg_code!({

            // Our port cannot have been blocked itself, because we're able to

            // directly insert the message into its slot.

            assert!(!comp_ctx.get_port(port_handle).state.is_blocked());

        });

        if exec_state.is_blocked_on_get(target_port_id) {

            // Return to normal operation

            exec_state.mode = CompMode::Sync;

            exec_state.mode_port = PortId::new_invalid();

            debug_assert!(exec_state.mode_value.values.is_empty());

        }

        return IncomingData::PlacedInSlot

    } else {

        // Slot is already full, so if the port was previously opened, it will

        // So enter the BlockedGet state

        exec_state.set_as_blocked_get(target_port);

        return GetResult::NoMessage;

    }

}

/// Default handling that has been received through a `get`. Will check if any

/// more messages are waiting, and if the corresponding port was blocked because

/// of full buffers (hence, will use the control layer to make sure the peer

/// will become unblocked).

pub(crate) fn default_handle_received_data_message(

    targeted_port: PortId, _port_instruction: PortInstruction, message: &mut DataMessage,

    inbox_main: &mut InboxMain, inbox_backup: &mut InboxBackup,

    comp_ctx: &mut CompCtx, sched_ctx: &SchedulerCtx, control: &mut ControlLayer,

    consensus: &mut Consensus

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

    let port_handle = comp_ctx.get_port_handle(targeted_port);

    let port_index = comp_ctx.get_port_index(port_handle);

    debug_assert!(inbox_main[port_index].is_none()); // because we've just received from it

    // If we received any ports, add them to the port tracking and inbox struct.

    // Then notify the peers that they can continue sending to this port, but

    // now at a new address.

    for received_port in &mut message.ports {

        // Create a new port locally

        let mut new_port_state = received_port.state;

        new_port_state.set(PortStateFlag::Received);

        let new_port_handle = comp_ctx.add_port(

            received_port.peer_comp, received_port.peer_port,

            received_port.kind, new_port_state

        );

        comp_ctx.change_port_peer(sched_ctx, new_port_handle, Some(received_port.peer_comp));

        let new_port = comp_ctx.get_port(new_port_handle);

        // Add the port tho the consensus

        consensus.notify_of_new_port(_new_inbox_index, new_port_handle, comp_ctx);

        // Replace all references to the port in the received message

        for message_location in received_port.locations.iter().copied() {

            let value = message.content.get_value_mut(message_location);

            match value {

                Value::Input(_) => {

                    debug_assert_eq!(new_port.kind, PortKind::Getter);

                    *value = Value::Input(port_id_to_eval(new_port.self_id));

                },

                Value::Output(_) => {

                    debug_assert_eq!(new_port.kind, PortKind::Putter);

                    *value = Value::Output(port_id_to_eval(new_port.self_id));

                },

                _ => unreachable!(),

            }

        }

        // Let the peer know that the port can now be used

        let peer_handle = comp_ctx.get_peer_handle(new_port.peer_comp_id);

        let peer_info = comp_ctx.get_peer(peer_handle);

        peer_info.handle.send_message_logged(sched_ctx, Message::Control(ControlMessage{

            id: ControlId::new_invalid(),

            sender_comp_id: comp_ctx.id,

            target_port_id: Some(new_port.peer_port_id),

            content: ControlMessageContent::PortPeerChangedUnblock(new_port.self_id, comp_ctx.id)

        }), true);

    }

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

    for message_index in 0..inbox_backup.len() {

        let message = &inbox_backup[message_index];

        if message.data_header.target_port == targeted_port {

            // One more message, place it in the slot

            let message = inbox_backup.remove(message_index);

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

            inbox_main[port_index] = Some(message);

            return Ok(());

        }

    }

    // Did not have any more messages, so if we were blocked, then we need to

    // unblock the port now (and inform the peer of this unblocking)

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

        let port_info = comp_ctx.get_port_mut(port_handle);

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

        let (peer_handle, message) = control.cancel_port_blocking(comp_ctx, port_handle);

        let peer_info = comp_ctx.get_peer(peer_handle);

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

    }

            // the component handle as well

            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;

            let peer_handle = comp_ctx.get_peer_handle(peer_comp_id);

            // One exception to sending an `Ack` is if we just closed the

            // port ourselves, meaning that the `ClosePort` messages got

            // sent to one another.

            if let Some(control_id) = control.has_close_port_entry(port_handle, comp_ctx) {

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

                default_handle_ack(exec_state, control, control_id, sched_ctx, comp_ctx, consensus, inbox_main, inbox_backup)?;

            } else {

                // Respond to the message

                let port_info = comp_ctx.get_port(port_handle);

                let last_instruction = port_info.last_instruction;

                default_send_ack(message.id, peer_handle, sched_ctx, comp_ctx);

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

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

            // We were previously blocked (or already closed)

            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_eq!(port_info.kind, PortKind::Putter);

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

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

            default_handle_recently_unblocked_port(