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,

        }

    }

}

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

    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)

        ))

        prepare_send_message_with_ports(

            transmitting_port_id, port_instruction, value, exec_state,

            comp_ctx, sched_ctx, control

        )?;

        return Ok(CompScheduling::Sleep);

    } else {

        // Port is not blocked and no ports to transfer: send to the peer

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

        let peer_info = comp_ctx.get_peer(peer_handle);

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

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

        return Ok(CompScheduling::Immediate);

    }

}

pub(crate) enum IncomingData {

    PlacedInSlot,

    SlotFull(DataMessage),

}

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

}

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

                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,

        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

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

    }

    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 {

        // Depending on the mode don't do anything at all, take some special

        // actions, or fall through and run the PDL code.

        match self.exec_state.mode {

            CompMode::NonSync | CompMode::Sync => {

                // continue and run PDL code

            },

                return CompScheduling::Sleep;

            }

            CompMode::StartExit => return component::default_handle_start_exit(

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

            ),

            CompMode::BusyExit => return component::default_handle_busy_exit(

    }

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

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

        match self.exec_state.mode {

                // impossible for this component, no input ports and no select

                // blocks

                unreachable!();

            }

            CompMode::NonSync => {

                // If in non-sync mode then we check if the arguments make sense