}

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;

}

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

#[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 {

            NonSync | StartExit | BusyExit | Exit => false,

        }

    }

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

        use CompMode::*;

        match self {

            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,

        }

    }

}

/// Component execution state: the execution mode along with some descriptive

/// fields. Fields are public for ergonomic reasons, use member functions when

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

    }

        self.mode = CompMode::BlockedPut;

        self.mode_port = port;

        self.mode_value = value;

    }

        return

            self.mode == CompMode::BlockedPut &&

            self.mode_port == port;

    }

}

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

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

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

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

        // Port is blocked, so we cannot send

        return Ok(CompScheduling::Sleep);

    } else {

    }

}

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

    comp_ctx.get_port_mut(port_handle).received_message_for_sync = true;

    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

        // now become closed

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

                let port_has_had_message = port_info.received_message_for_sync;

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

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

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

        },

        ControlMessageContent::PortPeerChangedBlock => {

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

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

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

        },

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

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

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

            let port_info = comp_ctx.get_port_mut(port_handle);

            port_info.peer_port_id = new_port_id;

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

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

        }

    }

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

}

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

    debug_assert_eq!(exec_state.mode, CompMode::Sync); // busy in sync, trying to send

    let sending_port_handle = comp_ctx.get_port_handle(sending_port_id);

    let sending_port_info = comp_ctx.get_port_mut(sending_port_handle);

    sending_port_info.last_instruction = sending_port_instruction;

    let mut transmit_ports = Vec::new();

    find_ports_in_value_group(&value, &mut transmit_ports);

    debug_assert!(!transmit_ports.is_empty()); // requisite for calling this function

    // Set up the final Ack that triggers us to send our final message

    let unblock_put_control_id = control.add_unblock_put_with_ports_entry();

    for (_, port_id) in &transmit_ports {

        let transmit_port_handle = comp_ctx.get_port_handle(*port_id);

        let transmit_port_info = comp_ctx.get_port_mut(transmit_port_handle);

        let peer_comp_id = transmit_port_info.peer_comp_id;

        let peer_port_id = transmit_port_info.peer_port_id;

        // Note: we checked earlier that we are currently in sync mode. Now we

        // will check if we've already used the port we're about to transmit.

        if !transmit_port_info.last_instruction.is_none() {

            return Err((

                sending_port_instruction,

                String::from("Cannot transmit one of the ports in this message, as it is used in this sync round")

            ));

        }

        if transmit_port_info.state.is_set(PortStateFlag::Transmitted) {

            return Err((

                sending_port_instruction,

                String::from("Cannot transmit one of the ports in this message, as that port is already transmitted")

            ));

        }

        // Set the flag for transmission

        transmit_port_info.state.set(PortStateFlag::Transmitted);

        // Block the peer of the port

        let message = control.create_port_transfer_message(unblock_put_control_id, comp_ctx.id, peer_port_id);

        println!("DEBUG: Port transfer message\nControl ID: {:?}\nMessage: {:?}", unblock_put_control_id, message);

        let peer_handle = comp_ctx.get_peer_handle(peer_comp_id);

        let peer_info = comp_ctx.get_peer(peer_handle);

        peer_info.handle.send_message_logged(sched_ctx, message, true);

    }

    // We've set up the protocol, once all the PPC's are blocked we are supposed

    // to transfer the message to the recipient. So store it temporarily

    return Ok(());

}

/// Performs the transmission of a data message that contains ports. These were

/// all stored in the component's execution state by the

fn perform_send_message_with_ports(

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

    inbox_main: &mut InboxMain, inbox_backup: &mut InboxBackup

) {

    // Find all ports again

    let mut transmit_ports = Vec::new();

    find_ports_in_value_group(&exec_state.mode_value, &mut transmit_ports);

    let port_handle = comp_ctx.get_port_handle(exec_state.mode_port);

    let port_info = comp_ctx.get_port(port_handle);

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

    // Annotate the data message

    let message_value = exec_state.mode_value.take();

    let mut annotated_message = consensus.annotate_data_message(comp_ctx, port_info, message_value);

    // And further enhance the message by adding data about the ports that are

    // being transferred

    for (port_locations, transmit_port_id) in transmit_ports {

        let transmit_port_handle = comp_ctx.get_port_handle(transmit_port_id);

        let transmit_port_info = comp_ctx.get_port(transmit_port_handle);

        let transmit_messages = take_port_messages(comp_ctx, transmit_port_id, inbox_main, inbox_backup);

        let mut transmit_port_state = transmit_port_info.state;

        debug_assert!(transmit_port_state.is_set(PortStateFlag::Transmitted));

        transmit_port_state.clear(PortStateFlag::Transmitted);

        annotated_message.ports.push(TransmittedPort{

            locations: port_locations,

            messages: transmit_messages,

            peer_comp: transmit_port_info.peer_comp_id,

            peer_port: transmit_port_info.peer_port_id,

            kind: transmit_port_info.kind,

            state: transmit_port_state

        })

    }

    // And finally, send the message to the peer

    let peer_info = comp_ctx.get_peer(peer_handle);

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

}

/// Handles an `Ack` for the control layer.

fn default_handle_ack(

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

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

    inbox_main: &mut InboxMain, inbox_backup: &mut InboxBackup

) {

    // Since an `Ack` may cause another one, handle them in a loop

    let mut to_ack = control_id;

    loop {

        let (action, new_to_ack) = control.handle_ack(to_ack, sched_ctx, comp_ctx);

        match action {

            AckAction::SendMessage(target_comp, message) => {

                // FIX @NoDirectHandle

                let mut handle = sched_ctx.runtime.get_component_public(target_comp);

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

                let _should_remove = handle.decrement_users();

                debug_assert!(_should_remove.is_none());

            },

            AckAction::ScheduleComponent(to_schedule) => {

                // FIX @NoDirectHandle

                let mut handle = sched_ctx.runtime.get_component_public(to_schedule);

                // Note that the component is intentionally not

                // sleeping, so we just wake it up

                debug_assert!(!handle.sleeping.load(std::sync::atomic::Ordering::Acquire));

                let key = unsafe { to_schedule.upgrade() };

                sched_ctx.runtime.enqueue_work(key);

                let _should_remove = handle.decrement_users();

                debug_assert!(_should_remove.is_none());

            },

            AckAction::UnblockPutWithPorts => {

                println!("DEBUG: Unblocking put with ports");

            },

            AckAction::None => {}

        }

        match new_to_ack {

            Some(new_to_ack) => to_ack = new_to_ack,

            None => break,

        }

    }

}

/// Little helper for sending the most common kind of `Ack`

fn default_send_ack(

    causer_of_ack_id: ControlId, peer_handle: LocalPeerHandle,

    sched_ctx: &SchedulerCtx, comp_ctx: &CompCtx

) {

    let peer_info = comp_ctx.get_peer(peer_handle);

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

        id: causer_of_ack_id,

        sender_comp_id: comp_ctx.id,

        target_port_id: None,

        content: ControlMessageContent::Ack

    }), true);

}

/// Handles the unblocking of a putter port. In case there is a pending message

fn default_handle_recently_unblocked_port(

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

    port_handle: LocalPortHandle, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx,

) {

    let port_info = comp_ctx.get_port_mut(port_handle);

    let port_id = port_info.self_id;

        // Return to the regular execution mode

        exec_state.mode = CompMode::Sync;

        exec_state.mode_port = PortId::new_invalid();

        // Annotate the message that we're going to send

        let port_info = comp_ctx.get_port(port_handle); // for immutable access

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

        let to_send = exec_state.mode_value.take();

        let to_send = consensus.annotate_data_message(comp_ctx, port_info, to_send);

        // Retrieve peer to send the message

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

        let peer_info = comp_ctx.get_peer(peer_handle);

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

        exec_state.mode = CompMode::Sync; // because we're blocked on a `put`, we must've started in the sync state.

        exec_state.mode_port = PortId::new_invalid();

    }

}

#[inline]

pub(crate) fn port_id_from_eval(port_id: EvalPortId) -> PortId {

    return PortId(port_id.id);

}

#[inline]

pub(crate) fn port_id_to_eval(port_id: PortId) -> EvalPortId {

    return EvalPortId{ id: port_id.0 };

}

// TODO: Optimize double vec

type EncounteredPorts = Vec<(Vec<ValueId>, PortId)>;

/// Recursively goes through the value group, attempting to find ports.

/// Duplicates will only be added once.

pub(crate) fn find_ports_in_value_group(value_group: &ValueGroup, ports: &mut EncounteredPorts) {

    // Helper to check a value for a port and recurse if needed.

    fn find_port_in_value(group: &ValueGroup, value: &Value, value_location: ValueId, ports: &mut EncounteredPorts) {

        match value {

            Value::Input(port_id) | Value::Output(port_id) => {

                // This is an actual port

                let cur_port = PortId(port_id.id);

                for prev_port in ports.iter_mut() {

                    if prev_port.1 == cur_port {

                        // Already added

                        prev_port.0.push(value_location);

                        return;

                    }

                }

                ports.push((vec![value_location], cur_port));

            },

            Value::Array(heap_pos) |

            Value::Message(heap_pos) |

            Value::String(heap_pos) |

            Value::Struct(heap_pos) |

            Value::Union(_, heap_pos) => {

                // Reference to some dynamic thing which might contain ports,

                // so recurse

                let heap_region = &group.regions[*heap_pos as usize];

                for (value_index, embedded_value) in heap_region.iter().enumerate() {

                    let value_location = ValueId::Heap(*heap_pos, value_index as u32);

                    find_port_in_value(group, embedded_value, value_location, ports);

                }

            },

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

        let slot = &mut self.inbox_main[0];

        if slot.is_none() {

            *slot = 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);

                component::default_handle_sync_decision(sched_ctx, &mut self.exec_state, comp_ctx, decision, &mut self.consensus);

            },

            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, &mut self.inbox_main, &mut self.inbox_backup

                ) {

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

                }

            },

            Message::Poll => {

                sched_ctx.info("Received polling event");

            },

        }

    }

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

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

        match self.exec_state.mode {

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

                            self.exec_state.set_as_start_exit(ExitReason::Termination);

                        } else {

                            // Reset for a new request

                            self.sync_state = SyncState::AwaitingCmd;

                            component::default_handle_sync_start(

                                &mut self.exec_state, &mut self.inbox_main, sched_ctx, comp_ctx, &mut self.consensus

                            );

                        }

                        return CompScheduling::Immediate;

                    },

                    SocketState::Error => {

                        // Could potentially send an error message to the

                        // connected component.

                        self.exec_state.set_as_start_exit(ExitReason::ErrorNonSync);

                        return CompScheduling::Immediate;

                    }

                }

            },

            CompMode::Sync => {

                // When in sync mode: wait for a command to come in

                match self.sync_state {

                    SyncState::AwaitingCmd => {

                        match component::default_attempt_get(

                            &mut self.exec_state, self.pdl_input_port_id, PortInstruction::NoSource,

                            &mut self.inbox_main, &mut self.inbox_backup, sched_ctx, comp_ctx,

                            &mut self.control, &mut self.consensus

                        ) {

                            GetResult::Received(message) => {

                                let (tag_value, embedded_heap_pos) = message.content.values[0].as_union();

                                if tag_value == self.input_union_send_tag_value {

                                    // Retrieve bytes from the message

                                    self.byte_buffer.clear();

                                    let union_content = &message.content.regions[embedded_heap_pos as usize];

                                    debug_assert_eq!(union_content.len(), 1);

                                    let array_heap_pos = union_content[0].as_array();

                                    let array_values = &message.content.regions[array_heap_pos as usize];

                                    self.byte_buffer.reserve(array_values.len());

                                    for value in array_values {

                                        self.byte_buffer.push(value.as_uint8());

            self.inbox_backup.push(message);

        }

    }

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

        sched_ctx.debug(&format!("handling message: {:?}", message));

        if let Some(new_target) = self.control.should_reroute(&mut message) {

            let mut target = sched_ctx.runtime.get_component_public(new_target); // TODO: @NoDirectHandle

            sched_ctx.debug(&format!("rerouting to: {:?}", new_target));

            target.send_message_logged(sched_ctx, message, false); // not waking up: we schedule once we've received all PortPeerChanged Acks

            let _should_remove = target.decrement_users();

            debug_assert!(_should_remove.is_none());

            return;

        }

        sched_ctx.debug("handling message myself");

        match message {

            Message::Data(message) => {

                self.handle_incoming_data_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,