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

}

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

        // now become closed

        let port_info = comp_ctx.get_port_mut(port_handle);

        if port_info.state.is_open() {

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

            let (peer_handle, message) =

                control.initiate_port_blocking(comp_ctx, port_handle);

            let peer = comp_ctx.get_peer(peer_handle);

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

        }

        return IncomingData::SlotFull(incoming_message)

    }

}

pub(crate) enum GetResult {

    Received(DataMessage),

    NoMessage,

    Error((PortInstruction, String)),

}

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

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

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

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

    inbox_main: &mut InboxMain, inbox_backup: &mut InboxBackup

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

    match message.content {

        ControlMessageContent::Ack => {

            default_handle_ack(exec_state, control, message.id, sched_ctx, comp_ctx, consensus, inbox_main, inbox_backup)?;

        },

        ControlMessageContent::BlockPort => {

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

            // blocked.

            let port_to_block = message.target_port_id.unwrap();

            let port_handle = comp_ctx.get_port_handle(port_to_block);

            let port_info = comp_ctx.get_port_mut(port_handle);

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

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

        },

        ControlMessageContent::ClosePort(content) => {

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

            // 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_registered_round = port_info.last_registered_round;

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

                    let round_has_succeeded = !content.closed_in_sync_round && last_registered_round == content.registered_round;

                    let closed_during_sync_round = content.closed_in_sync_round;

                    if (closed_during_sync_round || closed_before_sync_round) && port_was_used {

                        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(

                exec_state, control, consensus, port_handle, sched_ctx,

                comp_ctx, inbox_main, inbox_backup

            )?;

        },

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

            default_handle_recently_unblocked_port(

                exec_state, control, consensus, port_handle, sched_ctx,

                comp_ctx, inbox_main, inbox_backup

            )?;

        }

    }

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

        }

    }

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

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

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

        let port_info = comp_ctx.get_port_by_index_mut(port_index);

        if port_info.state.is_blocked() {

            return CompScheduling::Sleep;

        }

    }

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

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

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

    // lower-level utils

    #[inline]

    pub fn set(&mut self, flag: PortStateFlag) {

        self.flags |= flag as u32;

    }

    #[inline]

    pub fn clear(&mut self, flag: PortStateFlag) {

        self.flags &= !(flag as u32);

    }

    #[inline]

    pub const fn is_set(&self, flag: PortStateFlag) -> bool {

        return (self.flags & (flag as u32)) != 0;

    }

}

impl Debug for PortState {

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

        use PortStateFlag::*;

        let mut s = f.debug_struct("PortState");

        for (flag_name, flag_value) in &[

            ("closed", Closed),

            ("blocked_peer_change", BlockedDueToPeerChange),

            ("blocked_full_buffers", BlockedDueToFullBuffers),

            ("transmitted", Transmitted),

        ] {

            s.field(flag_name, &self.is_set(*flag_value));

        }

        return s.finish();

    }

}

#[derive(Debug)]

pub struct Port {

    // Identifiers

    pub self_id: PortId,

    pub peer_comp_id: CompId, // eventually consistent

    pub peer_port_id: PortId, // eventually consistent

    // Generic operating state

    pub kind: PortKind,

    pub state: PortState,

    // State tracking for error detection and error handling

    pub last_registered_round: Option<u32>,

    pub last_instruction: PortInstruction, // used during sync round to detect port-closed-during-sync errors

    pub close_at_sync_end: bool, // set during sync round when receiving a port-closed-after-sync message

    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::new(),

            peer_comp_id: self.id,

            last_registered_round: None,

            last_instruction: PortInstruction::None,

            close_at_sync_end: false,

            associated_with_peer: false,

        });

        self.ports.push(Port{

            self_id: getter_id,

            peer_port_id: putter_id,

            kind: PortKind::Getter,

            state: PortState::new(),

            peer_comp_id: self.id,

            last_registered_round: None,

            last_instruction: PortInstruction::None,

            close_at_sync_end: false,

            associated_with_peer: false,

        });

        return Channel{ putter_id, getter_id };

    }

    /// Adds a new port. Make sure to call `change_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_registered_round: None,

            last_instruction: PortInstruction::None,

            close_at_sync_end: false,

            associated_with_peer: false,

        });

        return LocalPortHandle(self_id);

    }

    /// Adds a self-reference. Called by the runtime/scheduler

    pub(crate) fn add_self_reference(&mut self, self_handle: CompHandle) {

        debug_assert_eq!(self.id, self_handle.id());

        debug_assert!(self.get_peer_index_by_id(self.id).is_none());

        self.peers.push(Peer{

            id: self.id,

            num_associated_ports: 0,

            handle: self_handle

        });

    }

    /// Removes a self-reference. Called by the runtime/scheduler

    pub(crate) fn remove_self_reference(&mut self) -> Option<CompKey> {

        let self_index = self.get_peer_index_by_id(self.id).unwrap();

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

        let maybe_comp_key = peer.handle.decrement_users();

        self.peers.remove(self_index);

        return maybe_comp_key;

    }

    /// Removes a port. Make sure you called `change_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;

    }

    /// Changes a peer

    pub(crate) fn change_port_peer(&mut self, sched_ctx: &SchedulerCtx, port_handle: LocalPortHandle, new_peer_comp_id: Option<CompId>) {

        // If port is currently associated with a peer, then remove that peer

        let port_index = self.get_port_index(port_handle);

        let port = &mut self.ports[port_index];

        let port_is_closed = port.state.is_closed();

        if port.associated_with_peer {

            // Remove old peer association

            port.associated_with_peer = false;

            let peer_comp_id = port.peer_comp_id;

            let peer_index = self.get_peer_index_by_id(peer_comp_id).unwrap();

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

            peer.num_associated_ports -= 1;