pub enum ControlMessageContent {

    Ack,

    BlockPort,

    UnblockPort,

    ClosePort(ControlMessageClosePort),

    PortPeerChangedBlock,

    PortPeerChangedUnblock(PortId, CompId), // contains (new_port_id, new_component_id)

}

#[derive(Copy, Clone, Debug)]

pub struct ControlMessageClosePort {

    pub closed_in_sync_round: bool, // needed to ensure correct handling of errors

}

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

// Messages (generic)

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

#[derive(Debug)]

pub struct MessageSyncHeader {

    pub sync_round: u32,

    pub sending_id: CompId,

    pub highest_id: CompId,

}

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

    let mut ports = Vec::new();

    find_ports_in_value_group(&value, &mut ports);

    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,

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

        let peer_info = comp_ctx.get_peer(peer_handle);

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

            id: ControlId::new_invalid(),

            sender_comp_id: comp_ctx.id,

            target_port_id: Some(new_port.peer_port_id),

            content: ControlMessageContent::PortPeerChangedUnblock(new_port.self_id, comp_ctx.id)

        }), true);

    }

    // Modify last-known location where port instruction was retrieved

    debug_assert_ne!(port_info.last_instruction, PortInstruction::None); // set by caller

    debug_assert!(port_info.state.is_open()); // checked by caller

    // Check if there are any more messages in the backup buffer

    for message_index in 0..inbox_backup.len() {

        let message = &inbox_backup[message_index];

        if message.data_header.target_port == targeted_port {

            // One more message, place it in the slot

            let message = inbox_backup.remove(message_index);

            debug_assert!(comp_ctx.get_port(port_handle).state.is_blocked()); // since we're removing another message from the backup

            inbox_main[port_index] = Some(message);

            return Ok(());

        }

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

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

}

#[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_instruction: PortInstruction, // used during sync round to detect port-closed-during-sync errors

    pub received_message_for_sync: bool, // used during sync round to detect port-closed-before-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,

}

    /// 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_instruction: PortInstruction::None,

            close_at_sync_end: false,

            received_message_for_sync: 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_instruction: PortInstruction::None,

            close_at_sync_end: false,

            received_message_for_sync: 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_instruction: PortInstruction::None,

            close_at_sync_end: false,

            received_message_for_sync: 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());

impl Consensus {

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

        return Self{

            round_index: 0,

            highest_id: CompId::new_invalid(),

            ports: Vec::new(),

            mapping_counter: 0,

            mode: Mode::NonSync,

            solution: SolutionCombiner::new(),

        }

    }

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

    // Managing sync state

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

    /// Notifies the consensus management that the PDL code has reached the

    /// start of a sync block.

    pub(crate) fn notify_sync_start(&mut self, comp_ctx: &CompCtx) {

        debug_assert_eq!(self.mode, Mode::NonSync);

        self.highest_id = comp_ctx.id;

        self.mapping_counter = 0;

        self.mode = Mode::SyncBusy;

        });

        // Return the messages notifying peer of the closed port

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

        return (

            peer_handle,

            ControlMessage{

                id: entry_id,

                sender_comp_id: comp_ctx.id,

                target_port_id: Some(peer_port_id),

                content: ControlMessageContent::ClosePort(ControlMessageClosePort{

                    closed_in_sync_round: exit_inside_sync,

                }),

            }

        );

    }

    /// Generates the control message used to indicate to a peer that a port

    /// should be blocked (expects the caller to have set the port's state to

    /// blocked).

    pub(crate) fn initiate_port_blocking(&mut self, comp_ctx: &CompCtx, port_handle: LocalPortHandle) -> (LocalPeerHandle, ControlMessage) {

        let port_info = comp_ctx.get_port(port_handle);

        debug_assert_eq!(port_info.kind, PortKind::Getter); // because we're telling the putter to block

        debug_assert!(port_info.state.is_set(PortStateFlag::BlockedDueToFullBuffers)); // contract with caller