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

            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

            )?;

        }

    }

    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

            ("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)]

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