/// very large port number is more likely to shine a light on bugs.

    pub fn new_invalid() -> Self {

        return Self(u32::MAX);

    }

}

pub struct Channel {

    pub putter_id: PortId,

    pub getter_id: PortId,

}

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

    pub(crate) id: ControlId,

    pub sender_comp_id: CompId,

    pub target_port_id: Option<PortId>,

    pub content: ControlMessageContent,

}

#[derive(Copy, Clone, Debug)]

pub enum ControlMessageContent {

    Ack,

    ClosePort(ControlMessageClosePort),

}

#[derive(Copy, Clone, Debug)]

pub struct ControlMessageClosePort {

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

}

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

// Messages (generic)

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

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

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

        ))

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

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

        }

) -> GetResult {

    let port_handle = comp_ctx.get_port_handle(target_port);

    let port_index = comp_ctx.get_port_index(port_handle);

    let port_info = comp_ctx.get_port_mut(port_handle);

    port_info.last_instruction = target_port_instruction;

        let peer_id = port_info.peer_comp_id;

        return GetResult::Error((

            target_port_instruction,

            format!("Cannot get from this port, as the peer component (id:{}) closed the port", peer_id.0)

        ));

    }

    let port_handle = comp_ctx.get_port_handle(targeted_port);

    let port_index = comp_ctx.get_port_index(port_handle);

    let slot = &mut inbox_main[port_index];

    debug_assert!(slot.is_none()); // because we've just received from it

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

    debug_assert_ne!(port_info.last_instruction, PortInstruction::None); // set 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);

            *slot = Some(message);

            return Ok(());

        }

    }

    // Did not have any more messages, so if we were blocked, then we need to

    // unblock the port now (and inform the peer of this unblocking)

        let (peer_handle, message) = control.cancel_port_blocking(comp_ctx, port_handle);

        let peer_info = comp_ctx.get_peer(peer_handle);

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

    }

    return Ok(());

    message: ControlMessage, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx

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

    match message.content {

        ControlMessageContent::Ack => {

            default_handle_ack(control, message.id, sched_ctx, comp_ctx);

        },

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

            // blocked.

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

        },

        ControlMessageContent::ClosePort(content) => {

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

            // the component handle as well

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

                // 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(control, control_id, sched_ctx, comp_ctx);

            } else {

                // Respond to the message

                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.remove_peer(sched_ctx, port_handle, peer_comp_id, false); // do not remove if closed

                // Handle any possible error conditions (which boil down to: the

                        return Err((

                            last_instruction,

                            format!("Peer component (id:{}) shut down, so previous communication cannot have succeeded", peer_comp_id.0)

                        ));

                    }

                } else {

                }

            }

        },

            // We were previously blocked (or already closed)

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

        },

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

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

        },

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

            let port_info = comp_ctx.get_port(port_handle);

            let old_peer_id = port_info.peer_comp_id;

            comp_ctx.remove_peer(sched_ctx, port_handle, old_peer_id, false);

            let port_info = comp_ctx.get_port_mut(port_handle);

            port_info.peer_comp_id = new_comp_id;

            port_info.peer_port_id = new_port_id;

            comp_ctx.add_peer(port_handle, sched_ctx, new_comp_id, None);

        }

    }

    return Ok(());

}

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

    }

    // Iterating over ports by index to work around borrowing rules

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

        let port = comp_ctx.get_port_by_index_mut(port_index);

            // Already closed, or in the process of being closed

            continue;

        }

        // Mark as closed

        let port_id = port.self_id;

        // Notify peer of closing

        let port_handle = comp_ctx.get_port_handle(port_id);

        let (peer, message) = control.initiate_port_closing(port_handle, exit_inside_sync, comp_ctx);

        let peer_info = comp_ctx.get_peer(peer);

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

    if success {

        // We cannot get a success message if the component has encountered an

        // error.

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

            let port_info = comp_ctx.get_port_by_index_mut(port_index);

            if port_info.close_at_sync_end {

            }

        }

        debug_assert_eq!(exec_state.mode, CompMode::SyncEnd);

        exec_state.mode = CompMode::NonSync;

        return Some(true);

    } else {

        content: ControlMessageContent::Ack

    }), true);

}

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

/// on that port then it will be sent.

    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;

    if exec_state.is_blocked_on_put(port_id) {

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

use crate::runtime2::scheduler::*;

use crate::runtime2::runtime::*;

use crate::runtime2::communication::*;

use crate::protocol::ExpressionId;

pub enum PortInstruction {

    None,

    NoSource,

    SourceLocation(ExpressionId),

}

#[derive(Debug)]

pub struct Port {

    // Identifiers

    pub self_id: PortId,

    pub peer_comp_id: CompId, // eventually consistent

    pub peer_port_id: PortId, // eventually consistent

    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 struct Peer {

    pub id: CompId,

    pub num_associated_ports: u32,

    pub(crate) handle: CompHandle,

        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,

            peer_comp_id: self.id,

            last_instruction: PortInstruction::None,

            close_at_sync_end: false,

            received_message_for_sync: false,

        });

        self.ports.push(Port{

            self_id: getter_id,

            peer_port_id: putter_id,

            kind: PortKind::Getter,

            peer_comp_id: self.id,

            last_instruction: PortInstruction::None,

            close_at_sync_end: false,

            received_message_for_sync: false,

        });

        return Channel{ putter_id, getter_id };

    }

    /// Adds a new port. Make sure to call `add_peer` afterwards.

        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,

        });

        return LocalPortHandle(self_id);

    }

    /// Removes a port. Make sure you called `remove_peer` first.

    pub(crate) fn remove_port(&mut self, port_handle: LocalPortHandle) -> Port {

                debug_assert_ne!(key.downgrade(), self.id); // should be upheld by the code that shuts down a component

                sched_ctx.runtime.destroy_component(key);

            }

        }

    }

    pub(crate) fn get_port_handle(&self, port_id: PortId) -> LocalPortHandle {

        return LocalPortHandle(port_id);

    }

    // should perhaps be revised, used in main inbox

    pub(crate) fn get_port_index(&self, port_handle: LocalPortHandle) -> usize {

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

    // Local utilities

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

    #[inline]

    fn requires_peer_reference(port: &Port, self_id: CompId, required_if_closed: bool) -> bool {

    }

    fn must_get_port_index(&self, handle: LocalPortHandle) -> usize {

        for (index, port) in self.ports.iter().enumerate() {

            if port.self_id == handle.0 {

                return index;

                // Note: we register the "last_instruction" here already. This

                // way if we get a `ClosePort` message, the condition to fail

                // the synchronous round is satisfied.

                let port_info = comp_ctx.get_port_mut(port_handle);

                port_info.last_instruction = PortInstruction::SourceLocation(expr_id);

                // Register port as part of select guard

                if let Err(_err) = self.select_state.register_select_case_port(comp_ctx, case_index, port_index, port_id) {

                    // Failure occurs if a port is used twice in the same guard

                    let protocol = &sched_ctx.runtime.protocol;

                    self.handle_component_error(sched_ctx, CompError::Executor(EvalError::new_error_at_expr(

                creator_handle: creator_port_handle,

                creator_id: creator_port_id,

                created_handle: created_port_handle,

                created_id: created_port_id,

            };

                closed_port_id_pairs.push(port_id_pair)

            } else {

                opened_port_id_pairs.push(port_id_pair);

            }

            // Modify value in arguments (bit dirty, but double vec in ValueGroup causes lifetime issues)

            ControlContent::ClosedPort(closed_port) => {

                // If a closed port is Ack'd, then we remove the reference to

                // that component.

                let port_handle = comp_ctx.get_port_handle(closed_port);

                let port_info = comp_ctx.get_port(port_handle);

                let port_peer_comp_id = port_info.peer_comp_id;

                comp_ctx.remove_peer(sched_ctx, port_handle, port_peer_comp_id, true); // remove if closed

                return (AckAction::None, None);

            }

        }

    }

        self.entries[entry_index].ack_countdown += 1;

        return Message::Control(ControlMessage{

            id: entry_id,

            sender_comp_id: creator_comp_id,

            target_port_id: Some(source_port_id),

        })

    }

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

    // Blocking, unblocking, and closing ports

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

    /// Initiates the control message procedures for closing a port. Caller must

    /// make sure that the port state has already been set to `Closed`.

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

        let port = comp_ctx.get_port(port_handle);

        let peer_port_id = port.peer_port_id;

        // Construct the port-closing entry

        let entry_id = self.take_id();

        self.entries.push(ControlEntry{

            id: entry_id,

            ack_countdown: 1,

            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

        let peer_port_id = port_info.peer_port_id;

        let peer_comp_id = port_info.peer_comp_id;

        let peer_handle = comp_ctx.get_peer_handle(peer_comp_id);

        return (

            peer_handle,

            ControlMessage{

                id: ControlId::new_invalid(),

                sender_comp_id: comp_ctx.id,

            }

        );

    }

    /// Generates a messages used to indicate to a peer that a port should be

    /// unblocked again.

    pub(crate) fn cancel_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 initiating the unblocking

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

        return (

            peer_handle,

            ControlMessage{

                id: ControlId::new_invalid(),

                sender_comp_id: comp_ctx.id,

                target_port_id: Some(port_info.peer_port_id),

            }

        );

    }

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

    // Internal utilities

#[cfg(test)] mod tests;

pub use runtime::Runtime;

pub(crate) use error::RtError;

pub(crate) use scheduler::SchedulerCtx;

pub(crate) use communication::{

    Message, ControlMessage, SyncMessage, DataMessage,

    SyncRoundDecision

};