#[derive(Debug, Copy, Clone, PartialEq, Eq)]

pub struct PortId(pub u32);

impl PortId {

    /// This value is not significant, it is chosen to make debugging easier: a

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

}

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

// Data messages

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

#[derive(Debug)]

pub struct DataMessage {

    pub data_header: MessageDataHeader,

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

// Control messages

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

#[derive(Debug)]

pub struct ControlMessage {

    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)

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

#[derive(Debug)]

pub struct MessageSyncHeader {

    pub sync_round: u32,

    pub sending_id: CompId,

    pub highest_id: CompId,

    port_instruction: PortInstruction, value: ValueGroup,

    sched_ctx: &SchedulerCtx, consensus: &mut Consensus, 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);

        // 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 port_info.state.is_blocked() {

        // Port is blocked, so we cannot send

        exec_state.set_as_blocked_put(transmitting_port_id, value);

        return Ok(CompScheduling::Sleep);

    } else {

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

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

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

/// instruction we're attempting on this port.

pub(crate) fn default_attempt_get(

    exec_state: &mut CompExecState, target_port: PortId, target_port_instruction: PortInstruction,

    inbox_main: &mut InboxMainRef, inbox_backup: &mut InboxBackup, sched_ctx: &SchedulerCtx,

    comp_ctx: &mut CompCtx, control: &mut ControlLayer, consensus: &mut Consensus

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

        ));

    }

    if let Some(message) = &inbox_main[port_index] {

        if consensus.try_receive_data_message(sched_ctx, comp_ctx, message) {

            // We're allowed to receive this message

            let message = inbox_main[port_index].take().unwrap();

            debug_assert_eq!(target_port, message.data_header.target_port);

/// will become unblocked).

pub(crate) fn default_handle_received_data_message(

    targeted_port: PortId, port_instruction: PortInstruction,

    inbox_main: &mut InboxMainRef, inbox_backup: &mut InboxBackup,

    comp_ctx: &mut CompCtx, sched_ctx: &SchedulerCtx, control: &mut ControlLayer

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

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

}

/// Handles control messages in the default way. Note that this function may

/// take a lot of actions in the name of the caller: pending messages may be

/// sent, ports may become blocked/unblocked, etc. So the execution

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

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

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

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

                if exec_state.mode.is_in_sync_block() {

                    let closed_during_sync_round = content.closed_in_sync_round && port_was_used;

                    let closed_before_sync_round = !content.closed_in_sync_round && !port_has_had_message;

                    if closed_during_sync_round || closed_before_sync_round {

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

}

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

) {

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

    }

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

    }

    return CompScheduling::Immediate; // to check if we can shut down immediately

}

/// Handles a component waiting until all peers are notified that it is quitting

            exec_state.mode.is_in_sync_block() && decision == SyncRoundDecision::Failure

        )

    );

    sched_ctx.info(&format!("Handling decision {:?} (in mode: {:?})", decision, exec_state.mode));

    consensus.notify_sync_decision(decision);

    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 {

        // We may get failure both in all possible cases. But we should only

        // modify the execution state if we're not already in exit mode

        if !exec_state.mode.is_busy_exiting() {

            sched_ctx.error("failed synchronous round, initiating exit");

            exec_state.set_as_start_exit(ExitReason::ErrorNonSync);

        }

) {

    let peer_info = comp_ctx.get_peer(peer_handle);

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

        id: causer_of_ack_id,

        sender_comp_id: comp_ctx.id,

        target_port_id: None,

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

        let to_send = consensus.annotate_data_message(comp_ctx, port_info, to_send);

        // Retrieve peer to send the message

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

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

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

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

use crate::protocol::ExpressionId;

/// Helper struct to remember when the last operation on the port took place.

#[derive(Debug, PartialEq, Copy, Clone)]

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

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

        }

    }

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

            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.

    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,

        });

        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 {

        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;

    }

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

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

        peer.num_associated_ports -= 1;

        if peer.num_associated_ports == 0 {

            let mut peer = self.peers.remove(peer_index);

            if let Some(key) = peer.handle.decrement_users() {

                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 {

        return self.must_get_port_index(port_handle);

    }

    pub(crate) fn get_peer_handle(&self, peer_id: CompId) -> LocalPeerHandle {

        return LocalPeerHandle(peer_id);

    }

    #[inline]

    pub(crate) fn num_ports(&self) -> usize {

        return self.ports.len();

    }

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

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

            }

        }

        unreachable!()

    }

                return CompScheduling::Requeue;

            },

            EC::SelectRegisterPort(expr_id, case_index, port_index, port_id) => {

                debug_assert_eq!(self.exec_state.mode, CompMode::Sync);

                let port_id = port_id_from_eval(port_id);

                let port_handle = comp_ctx.get_port_handle(port_id);

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

                        &self.prompt, &protocol.modules, &protocol.heap, expr_id,

                        String::from("Cannot have the one port appear in the same guard twice")

                    )));

                } else if port_is_closed {

                    // Port is closed

                    let peer_id = comp_ctx.get_port(port_handle).peer_comp_id;

                creator_port.kind, creator_port.state

            );

            let created_port = created_ctx.get_port(created_port_handle);

            let created_port_id = created_port.self_id;

            let port_id_pair = PortPair {

                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)

            let arg_value = if let Some(heap_pos) = port_reference.heap_pos {

                &mut arg_iter.group.regions[heap_pos][port_reference.index]

            } else {

                &mut arg_iter.group.values[port_reference.index]

            };

            match arg_value {

            },

            ControlContent::ScheduleComponent(to_schedule) => {

                // If all change-of-peers are `Ack`d, then we're ready to

                // schedule the component!

                return (AckAction::ScheduleComponent(to_schedule), None);

            },

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

            }

        }

    }

    pub(crate) fn has_acks_remaining(&self) -> bool {

        return !self.entries.is_empty();

    }

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

                schedule_entry_id,

            }),

        });

        // increment counter on schedule entry

        let entry_index = self.get_entry_index_by_id(schedule_entry_id).unwrap();

        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

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

    pub(crate) fn has_close_port_entry(&self, port_handle: LocalPortHandle, comp_ctx: &CompCtx) -> Option<ControlId> {

        let port = comp_ctx.get_port(port_handle);

        let port_id = port.self_id;

        for entry in self.entries.iter() {

            if let ControlContent::ClosedPort(entry_port_id) = &entry.content {

                }

            }

        }

        return None;

    }

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

            content: ControlContent::ClosedPort(port.self_id),