let length_value = length_value.as_signed_integer();

                                        if length_value < 0 {

                                            return Err(EvalError::new_error_at_expr(

                                                self, modules, heap, expr_id,

                                                format!("got length '{}', can only create a message with a non-negative length", length_value)

                                            ));

                                        }

                                        length_value as u64

                                    } else {

                                        debug_assert!(length_value.is_unsigned_integer());

                                        length_value.as_unsigned_integer()

                                    };

                                    let heap_pos = self.store.alloc_heap();

                                    let values = &mut self.store.heap_regions[heap_pos as usize].values;

                                    debug_assert!(values.is_empty());

                                    values.resize(length as usize, Value::UInt8(0));

                                    cur_frame.expr_values.push_back(Value::Message(heap_pos));

                                },

                                Method::Length => {

                                    let value = cur_frame.expr_values.pop_front().unwrap();

                                    let value_heap_pos = value.get_heap_pos();

                                    let value = self.store.maybe_read_ref(&value);

                                    let heap_pos = match value {

                                        Value::Array(pos) => *pos,

                                        Value::String(pos) => *pos,

                                        _ => unreachable!("length(...) on {:?}", value),

                                    };

                                    let len = self.store.heap_regions[heap_pos as usize].values.len();

                                    // TODO: @PtrInt

                                    cur_frame.expr_values.push_back(Value::UInt32(len as u32));

                                    self.store.drop_value(value_heap_pos);

                                },

                                Method::Assert => {

                                    let value = cur_frame.expr_values.pop_front().unwrap();

                                    let value = self.store.maybe_read_ref(&value).clone();

                                    if !value.as_bool() {

                                        return Ok(EvalContinuation::BranchInconsistent)

                                    }

                                },

                                Method::Print => {

                                    // Convert the runtime-variant of a string

                                    // into an actual string.

                                    let value = cur_frame.expr_values.pop_front().unwrap();

                                    let value = self.store.maybe_read_ref(&value);

                                    let value_heap_pos = value.as_string();

                                    let elements = &self.store.heap_regions[value_heap_pos as usize].values;

                                    let mut message = String::with_capacity(elements.len());

                                    for element in elements {

                                        message.push(element.as_char());

                                    }

                                    // Drop the heap-allocated value from the

                                    // store

                                    if is_literal_string {

                                        self.store.drop_heap_pos(value_heap_pos);

                                    }

                                    println!("{}", message);

                                },

                                Method::SelectStart => {

                                    let num_cases = self.store.maybe_read_ref(&cur_frame.expr_values.pop_front().unwrap()).as_uint32();

                                    let num_ports = self.store.maybe_read_ref(&cur_frame.expr_values.pop_front().unwrap()).as_uint32();

                                    return Ok(EvalContinuation::SelectStart(num_cases, num_ports));

                                },

                                Method::SelectRegisterCasePort => {

                                    let case_index = self.store.maybe_read_ref(&cur_frame.expr_values.pop_front().unwrap()).as_uint32();

                                    let port_index = self.store.maybe_read_ref(&cur_frame.expr_values.pop_front().unwrap()).as_uint32();

                                    let port_value = self.store.maybe_read_ref(&cur_frame.expr_values.pop_front().unwrap()).as_port_id();

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

                                },

                                Method::SelectWait => {

                                    match ctx.performed_select_wait() {

                                        Some(select_index) => {

                                            cur_frame.expr_values.push_back(Value::UInt32(select_index));

                                        },

                                        None => {

                                            cur_frame.expr_stack.push_back(ExprInstruction::EvalExpr(expr.this.upcast()));

                                            return Ok(EvalContinuation::SelectWait)

                                        },

                                    }

                                },

                                Method::ComponentRandomU32 | Method::ComponentTcpClient => {

                                    debug_assert_eq!(heap[expr.procedure].parameters.len(), cur_frame.expr_values.len());

                                    debug_assert_eq!(heap[cur_frame.position].as_new().expression, expr.this);

                                },

                                Method::UserComponent => {

                                    // This is actually handled by the evaluation

                                    // of the statement.

        for port_var_idx in 0..call_id_section.len() {

            let get_call_expr_id = call_id_section[port_var_idx];

            let port_expr_id = expr_id_section[port_var_idx];

            let port_type_index = ctx.heap[port_expr_id].type_index();

            let port_type_ref = TypeIdReference::IndirectSameAsExpr(port_type_index);

            // Move the port expression such that it gets assigned to a temporary variable

            let variable_id = create_ast_variable(ctx, outer_scope_id);

            let variable_decl_stmt_id = create_ast_variable_declaration_stmt(ctx, self.current_procedure_id, variable_id, port_type_ref, port_expr_id);

            // Replace the original port expression in the call with a reference

            // to the replacement variable

            let variable_expr_id = create_ast_variable_expr(ctx, self.current_procedure_id, variable_id, port_type_ref);

            let call_expr = &mut ctx.heap[get_call_expr_id];

            call_expr.arguments[0] = variable_expr_id.upcast();

            transformed_stmts.push(variable_decl_stmt_id.upcast().upcast());

            locals.push((variable_id, port_type_ref));

        }

        // Insert runtime calls that facilitate the semantics of the select

        // block.

        // Create the call that indicates the start of the select block

        {

            let num_cases_expression_id = create_ast_literal_integer_expr(ctx, self.current_procedure_id, total_num_cases as u64, ctx.arch.uint32_type_id);

            let num_ports_expression_id = create_ast_literal_integer_expr(ctx, self.current_procedure_id, total_num_ports as u64, ctx.arch.uint32_type_id);

            let arguments = vec![

                num_cases_expression_id.upcast(),

                num_ports_expression_id.upcast()

            ];

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

            let call_statement_id = create_ast_expression_stmt(ctx, call_expression_id.upcast());

            transformed_stmts.push(call_statement_id.upcast());

        }

        // Create calls for each select case that will register the ports that

        // we are waiting on at the runtime.

        {

            let mut total_port_index = 0;

            for case_index in 0..total_num_cases {

                let case = &ctx.heap[id].cases[case_index];

                let case_num_ports = case.involved_ports.len();

                for case_port_index in 0..case_num_ports {

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

                    let (port_variable_id, port_variable_type) = locals[total_port_index]; // so far this variable contains the temporary variables for the port expressions

                    let case_index_expr_id = create_ast_literal_integer_expr(ctx, self.current_procedure_id, case_index as u64, ctx.arch.uint32_type_id);

                    let port_index_expr_id = create_ast_literal_integer_expr(ctx, self.current_procedure_id, case_port_index as u64, ctx.arch.uint32_type_id);

                    let port_variable_expr_id = create_ast_variable_expr(ctx, self.current_procedure_id, port_variable_id, port_variable_type);

                    let runtime_call_arguments = vec![

                        case_index_expr_id.upcast(),

                        port_index_expr_id.upcast(),

                        port_variable_expr_id.upcast()

                    ];

                    // Create runtime call, then store it

                    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_stmt_id = create_ast_expression_stmt(ctx, runtime_call_expr_id.upcast());

                    transformed_stmts.push(runtime_call_stmt_id.upcast());

                    total_port_index += 1;

                }

            }

        }

        // Create the variable that will hold the result of a completed select

        // block. Then create the runtime call that will produce this result

        let select_variable_id = create_ast_variable(ctx, outer_scope_id);

        let select_variable_type = TypeIdReference::DirectTypeId(ctx.arch.uint32_type_id);

        locals.push((select_variable_id, select_variable_type));

        {

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

            let variable_stmt_id = create_ast_variable_declaration_stmt(ctx, self.current_procedure_id, select_variable_id, select_variable_type, runtime_call_expr_id.upcast());

            transformed_stmts.push(variable_stmt_id.upcast().upcast());

        }

        call_id_section.forget();

        expr_id_section.forget();

        // Now we transform each of the select block case's guard and code into

        // a chained if-else statement.

        let relative_pos = transformed_stmts.len() as i32;

        if total_num_cases > 0 {

            let (if_stmt_id, end_if_stmt_id, scope_id) = transform_select_case_code(ctx, self.current_procedure_id, id, 0, select_variable_id, select_variable_type);

            link_existing_child_to_new_parent_scope(ctx, &mut self.scope_buffer, outer_scope_id, scope_id, relative_pos);

            let first_end_if_stmt = &mut ctx.heap[end_if_stmt_id];

            first_end_if_stmt.next = outer_end_block_id.upcast();

            let mut last_if_stmt_id = if_stmt_id;

            let mut last_end_if_stmt_id = end_if_stmt_id;

            let mut last_parent_scope_id = outer_scope_id;

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

        }

    }

}

/// Generic representation of a component (as viewed by a scheduler).

pub(crate) trait Component {

    /// Called upon the creation of the component. Note that the scheduler

    /// context is officially running another component (the component that is

    /// creating the new component).

    fn on_creation(&mut self, comp_id: CompId, sched_ctx: &SchedulerCtx);

    /// Called when a component crashes or wishes to exit. So is not called

    /// right before destruction, other components may still hold a handle to

    /// the component and send it messages!

    fn on_shutdown(&mut self, sched_ctx: &SchedulerCtx);

    /// Called if the component is created by another component and the messages

    /// are being transferred between the two.

    fn adopt_message(&mut self, comp_ctx: &mut CompCtx, message: DataMessage);

    /// Called if the component receives a new message. The component is

    /// responsible for deciding where that messages goes.

    fn handle_message(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx, message: Message);

    /// Called if the component's routine should be executed. The return value

    /// can be used to indicate when the routine should be run again.

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

}

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

pub(crate) enum CompMode {

    NonSync, // not in sync mode

    Sync, // in sync mode, can interact with other components

    SyncEnd, // awaiting a solution, i.e. encountered the end of the sync block

    BlockedGet, // blocked because we need to receive a message on a particular port

    BlockedPut, // component is blocked because the port is blocked

    BlockedSelect, // waiting on message to complete the select statement

    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

    Exit, // exiting: shutdown process started, now waiting until the reference count drops to 0

}

impl CompMode {

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

        use CompMode::*;

        match self {

            Sync | SyncEnd | BlockedGet | BlockedPut | BlockedSelect => true,

            NonSync | StartExit | BusyExit | Exit => false,

        }

    }

    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,

/// appropriate.

pub(crate) struct CompExecState {

    pub mode: CompMode,

    pub mode_port: PortId, // valid if blocked on a port (put/get)

    pub mode_value: ValueGroup, // valid if blocked on a put

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

}

impl CompExecState {

    pub(crate) fn new() -> Self {

        return Self{

            mode: CompMode::NonSync,

            mode_port: PortId::new_invalid(),

            mode_value: ValueGroup::default(),

            exit_reason: ExitReason::Termination,

        }

    }

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

        self.mode = CompMode::StartExit;

        self.exit_reason = reason;

    }

    pub(crate) fn set_as_blocked_get(&mut self, port: PortId) {

        self.mode = CompMode::BlockedGet;

        self.mode_port = port;

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

    }

    pub(crate) fn is_blocked_on_get(&self, port: PortId) -> bool {

        return

            self.mode == CompMode::BlockedGet &&

            self.mode_port == port;

    }

    pub(crate) fn set_as_blocked_put(&mut self, port: PortId, value: ValueGroup) {

        self.mode = CompMode::BlockedPut;

        self.mode_port = port;

        self.mode_value = value;

    }

    pub(crate) fn is_blocked_on_put(&self, port: PortId) -> bool {

        return

            self.mode == CompMode::BlockedPut &&

            self.mode_port == port;

    }

}

/// Creates a new component based on its definition. Meaning that if it is a

/// user-defined component then we set up the PDL code state. Otherwise we

/// construct a custom component. This does NOT take care of port and message

/// management.

pub(crate) fn create_component(

    protocol: &ProtocolDescription,

    definition_id: ProcedureDefinitionId, type_id: TypeId,

    arguments: ValueGroup, num_ports: usize

) -> Box<dyn Component> {

    let definition = &protocol.heap[definition_id];

    debug_assert!(definition.kind == ProcedureKind::Primitive || definition.kind == ProcedureKind::Composite);

    if definition.source.is_builtin() {

        // Builtin component

        let component: Box<dyn Component> = match definition.source {

            ProcedureSource::CompRandomU32 => Box::new(ComponentRandomU32::new(arguments)),

            ProcedureSource::CompTcpClient => Box::new(ComponentTcpClient::new(arguments)),

            _ => unreachable!(),

        };

        return component;

    } else {

        // User-defined component

        let prompt = Prompt::new(

            &protocol.types, &protocol.heap,

            definition_id, type_id, arguments

        );

        let component = CompPDL::new(prompt, num_ports);

        return Box::new(component);

    }

}

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

// 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, comp_ctx: &mut CompCtx

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

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

    return Ok(());

}

/// Handles control messages in the default way. Note that this function may

/// take a lot of actions in the name of the caller: pending messages may be

/// sent, ports may become blocked/unblocked, etc. So the execution

/// (`CompExecState`), control (`ControlLayer`) and consensus (`Consensus`)

/// state may all change.

pub(crate) fn default_handle_control_message(

    exec_state: &mut CompExecState, control: &mut ControlLayer, consensus: &mut Consensus,

    message: ControlMessage, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx

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

    match message.content {

        ControlMessageContent::Ack => {

            default_handle_ack(control, message.id, sched_ctx, comp_ctx);

        },

        ControlMessageContent::BlockPort(port_id) => {

            // One of our messages was accepted, but the port should be

            // blocked.

            let port_handle = comp_ctx.get_port_handle(port_id);

            let port_info = comp_ctx.get_port(port_handle);

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

            if port_info.state == PortState::Open {

                // only when open: we don't do this when closed, and we we don't do this if we're blocked due to peer changes

                comp_ctx.set_port_state(port_handle, PortState::BlockedDueToFullBuffers);

            }

        },

        ControlMessageContent::ClosePort(content) => {

            // Request to close the port. We immediately comply and remove

            // the component handle as well

            let port_handle = comp_ctx.get_port_handle(content.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 port_info = comp_ctx.get_port(port_handle);

            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

                default_handle_ack(control, control_id, sched_ctx, comp_ctx);

            } else {

                // Respond to the message

                let last_instruction = port_info.last_instruction;

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

                comp_ctx.remove_peer(sched_ctx, port_handle, peer_comp_id, false); // do not remove if closed

                }

            }

        },

        ControlMessageContent::UnblockPort(port_id) => {

            // We were previously blocked (or already closed)

            let port_handle = comp_ctx.get_port_handle(port_id);

            let port_info = comp_ctx.get_port(port_handle);

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

            if port_info.state == PortState::BlockedDueToFullBuffers {

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

            }

        },

        ControlMessageContent::PortPeerChangedBlock(port_id) => {

            // The peer of our port has just changed. So we are asked to

            // temporarily block the port (while our original recipient is

            // potentially rerouting some of the in-flight messages) and

            // Ack. Then we wait for the `unblock` call.

            debug_assert_eq!(message.target_port_id, Some(port_id));

            let port_handle = comp_ctx.get_port_handle(port_id);

            comp_ctx.set_port_state(port_handle, PortState::BlockedDueToPeerChange);

            let port_info = comp_ctx.get_port(port_handle);

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

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

        },

        ControlMessageContent::PortPeerChangedUnblock(new_port_id, new_comp_id) => {

            let port_handle = comp_ctx.get_port_handle(message.target_port_id.unwrap());

            let port_info = comp_ctx.get_port(port_handle);

            debug_assert!(port_info.state == PortState::BlockedDueToPeerChange);

            let old_peer_id = port_info.peer_comp_id;

            comp_ctx.remove_peer(sched_ctx, port_handle, old_peer_id, false);

            let port_info = comp_ctx.get_port_mut(port_handle);

            port_info.peer_comp_id = new_comp_id;

            port_info.peer_port_id = new_port_id;

            comp_ctx.add_peer(port_handle, sched_ctx, new_comp_id, None);

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

        }

    }

    return Ok(());

}

/// Handles a component initiating the exiting procedure, and closing all of its

/// ports. Should only be called once per component (which is ensured by

/// checking and modifying the mode in the execution state).

#[must_use]

pub(crate) fn default_handle_start_exit(

    exec_state: &mut CompExecState, control: &mut ControlLayer,

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

) -> CompScheduling {

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

    sched_ctx.log("Component starting exit");

    exec_state.mode = CompMode::BusyExit;

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

    }

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

    for port_index in 0..comp_ctx.num_ports() {

        let port = comp_ctx.get_port_by_index_mut(port_index);

        if port.state == PortState::Closed {

            // Already closed, or in the process of being closed

            continue;

        }

        // Mark as closed

        let port_id = port.self_id;

        port.state = PortState::Closed;

        // Notify peer of closing

        let port_handle = comp_ctx.get_port_handle(port_id);

        let (peer, message) = control.initiate_port_closing(port_handle, exit_inside_sync, comp_ctx);

        let peer_info = comp_ctx.get_peer(peer);

        peer_info.handle.send_message(&sched_ctx.runtime, Message::Control(message), true);

    }

    return CompScheduling::Immediate; // to check if we can shut down immediately

}

/// Handles a component waiting until all peers are notified that it is quitting

/// (i.e. after calling `default_handle_start_exit`).

#[must_use]

pub(crate) fn default_handle_busy_exit(

    exec_state: &mut CompExecState, control: &ControlLayer,

    sched_ctx: &SchedulerCtx

) -> CompScheduling {

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

    if control.has_acks_remaining() {

        sched_ctx.log("Component busy exiting, still has `Ack`s remaining");

        return CompScheduling::Sleep;

    } else {

        sched_ctx.log("Component busy exiting, now shutting down");

        exec_state.mode = CompMode::Exit;

        return CompScheduling::Exit;

    }

}

/// Handles a potential synchronous round decision. If there was a decision then

/// the `Some(success)` value indicates whether the round succeeded or not.

/// Might also end up changing the `ExecState`.

///

/// Might be called in two cases:

/// 1. The component is in regular execution mode, at the end of a sync round,

use crate::runtime2::scheduler::*;

use crate::runtime2::runtime::*;

use crate::runtime2::communication::*;

use crate::protocol::ExpressionId;

/// Helper struct to remember when the last operation on the port took place.

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

pub enum PortInstruction {

    None,

    NoSource,

    SourceLocation(ExpressionId),

}

#[derive(Debug)]

pub struct Port {

    pub self_id: PortId,

    pub peer_comp_id: CompId, // eventually consistent

    pub peer_port_id: PortId, // eventually consistent

    pub kind: PortKind,

    pub state: PortState,

    #[cfg(debug_assertions)] pub(crate) associated_with_peer: bool,

}

pub struct Peer {

    pub id: CompId,

    pub num_associated_ports: u32,

    pub(crate) handle: CompHandle,

}

/// Port and peer management structure. Will keep a local reference counter to

/// the ports associate with peers, additionally manages the atomic reference

/// counter associated with the peers' component handles.

pub struct CompCtx {

    pub id: CompId,

    ports: Vec<Port>,

    peers: Vec<Peer>,

    port_id_counter: u32,

}

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

pub struct LocalPortHandle(PortId);

#[derive(Copy, Clone)]

pub struct LocalPeerHandle(CompId);

impl CompCtx {

    /// Creates a new component context based on a reserved entry in the

    /// component store. This reservation is used such that we already know our

    /// assigned ID.

    pub(crate) fn new(reservation: &CompReserved) -> Self {

        return Self{

            id: reservation.id(),

            ports: Vec::new(),

            peers: Vec::new(),

            port_id_counter: 0,

        }

    }

    /// Creates a new channel that is fully owned by the component associated

    /// with this context.

    pub(crate) fn create_channel(&mut self) -> Channel {

        let putter_id = PortId(self.take_port_id());

        let getter_id = PortId(self.take_port_id());

        self.ports.push(Port{

            self_id: putter_id,

            peer_port_id: getter_id,

            kind: PortKind::Putter,

            state: PortState::Open,

            peer_comp_id: self.id,

            last_instruction: PortInstruction::None,

            close_at_sync_end: false,

            #[cfg(debug_assertions)] associated_with_peer: false,

        });

        self.ports.push(Port{

            self_id: getter_id,

            peer_port_id: putter_id,

            kind: PortKind::Getter,

            state: PortState::Open,

            peer_comp_id: self.id,

            last_instruction: PortInstruction::None,

            close_at_sync_end: false,

            #[cfg(debug_assertions)] associated_with_peer: false,

        });

        return Channel{ putter_id, getter_id };

    }

    /// Adds a new port. Make sure to call `add_peer` afterwards.

    pub(crate) fn add_port(&mut self, peer_comp_id: CompId, peer_port_id: PortId, kind: PortKind, state: PortState) -> LocalPortHandle {

        let self_id = PortId(self.take_port_id());

        self.ports.push(Port{

            self_id, peer_comp_id, peer_port_id, kind, state,

            last_instruction: PortInstruction::None,

            close_at_sync_end: false,

            #[cfg(debug_assertions)] associated_with_peer: false,

        });

        return LocalPortHandle(self_id);

    }

    /// Removes a port. Make sure you called `remove_peer` first.

    pub(crate) fn remove_port(&mut self, port_handle: LocalPortHandle) -> Port {

        let port_index = self.must_get_port_index(port_handle);

        let port = self.ports.remove(port_index);

        dbg_code!(assert!(!port.associated_with_peer));

        return port;

    }

    /// Adds a new peer. This must be called for every port, no matter the

    /// component the channel is connected to. If a `CompHandle` is supplied,

    /// then it will be used to add the peer. Otherwise it will be retrieved

    /// from the runtime using its ID.

    pub(crate) fn add_peer(&mut self, port_handle: LocalPortHandle, sched_ctx: &SchedulerCtx, peer_comp_id: CompId, handle: Option<&CompHandle>) {

        let self_id = self.id;

        let port = self.get_port_mut(port_handle);

        debug_assert_eq!(port.peer_comp_id, peer_comp_id);

        dbg_code!(assert!(!port.associated_with_peer));

        if !Self::requires_peer_reference(port, self_id, false) {

            return;

        }

        dbg_code!(port.associated_with_peer = true);

        match self.get_peer_index_by_id(peer_comp_id) {

            Some(peer_index) => {

                let peer = &mut self.peers[peer_index];

                peer.num_associated_ports += 1;

            },

            None => {

                let handle = match handle {

                    Some(handle) => handle.clone(),

                    None => sched_ctx.runtime.get_component_public(peer_comp_id)

                };

                self.peers.push(Peer{

                    id: peer_comp_id,

                    num_associated_ports: 1,

                    handle,

                });

            }

        }

    }

    /// Removes a peer associated with a port.

    pub(crate) fn remove_peer(&mut self, sched_ctx: &SchedulerCtx, port_handle: LocalPortHandle, peer_id: CompId, also_remove_if_closed: bool) {

impl Component for ComponentTcpClient {

    fn on_creation(&mut self, id: CompId, sched_ctx: &SchedulerCtx) {

        // Retrieve type information for messages we're going to receive

        let pd = &sched_ctx.runtime.protocol;

        let cmd_type = pd.find_type(b"std.internet", b"Cmd")

            .expect("'Cmd' type in the 'std.internet' module");

        let cmd_type = cmd_type

            .as_union();

        self.input_union_send_tag_value = cmd_type.get_variant_tag_value(b"Send").unwrap();

        self.input_union_receive_tag_value = cmd_type.get_variant_tag_value(b"Receive").unwrap();

        self.input_union_finish_tag_value = cmd_type.get_variant_tag_value(b"Finish").unwrap();

        self.input_union_shutdown_tag_value = cmd_type.get_variant_tag_value(b"Shutdown").unwrap();

        // Register socket for async events

        if let SocketState::Connected(socket) = &self.socket_state {

            let self_handle = sched_ctx.runtime.get_component_public(id);

            let poll_ticket = sched_ctx.polling.register(socket, self_handle, true, true)

                .expect("registering tcp component");

            debug_assert!(self.poll_ticket.is_none());

            self.poll_ticket = Some(poll_ticket);

        }

    }

    fn on_shutdown(&mut self, sched_ctx: &SchedulerCtx) {

        if let Some(poll_ticket) = self.poll_ticket.take() {

            sched_ctx.polling.unregister(poll_ticket)

                .expect("unregistering tcp component");

        }

    }

    fn adopt_message(&mut self, _comp_ctx: &mut CompCtx, message: DataMessage) {

        if self.inbox_main.is_none() {

            self.inbox_main = Some(message);

        } else {

            self.inbox_backup.push(message);

        }

    }

    fn handle_message(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx, message: Message) {

        match message {

            Message::Data(message) => {

                self.handle_incoming_data_message(sched_ctx, comp_ctx, message);

            },

            Message::Sync(message) => {

                let decision = self.consensus.receive_sync_message(sched_ctx, comp_ctx, message);

            },

            Message::Control(message) => {

                if let Err(location_and_message) = component::default_handle_control_message(

                    &mut self.exec_state, &mut self.control, &mut self.consensus,

                    message, sched_ctx, comp_ctx

                ) {

                    component::default_handle_error_for_builtin(&mut self.exec_state, sched_ctx, location_and_message);

                }

            },

            Message::Poll => {

                sched_ctx.log("Received polling event");

            },

        }

    }

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

        sched_ctx.log(&format!("Running component ComponentTcpClient (mode: {:?}, sync state: {:?})", self.exec_state.mode, self.sync_state));

        match self.exec_state.mode {

            CompMode::BlockedSelect => {

                // Not possible: we never enter this state

                unreachable!();

            },

            CompMode::NonSync => {

                // When in non-sync mode

                match &mut self.socket_state {

                    SocketState::Connected(_socket) => {

                        if self.sync_state == SyncState::FinishSyncThenQuit {

                            // Previous request was to let the component shut down