#[derive(Debug)]

pub struct PortAnnotationGetter {

    pub self_comp_id: CompId,

    pub self_port_id: PortId,

    pub peer_comp_id: CompId,

    pub peer_port_id: PortId,

}

#[derive(Debug)]

pub struct PortAnnotationPutter {

    pub self_comp_id: CompId,

    pub self_port_id: PortId,

}

#[derive(Debug)]

pub struct MessageDataHeader {

    pub expected_mapping: Vec<(PortAnnotationKind, Option<u32>)>,

    pub new_mapping: u32,

    pub source_port: PortId,

    pub target_port: PortId,

}

#[derive(Debug)]

pub struct TransmittedPort {

}

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

// Sync messages

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

#[derive(Debug)]

pub struct SyncMessage {

    pub sync_header: MessageSyncHeader,

    pub content: SyncMessageContent,

}

#[derive(Debug)]

pub enum SyncLocalSolutionEntry {

    Putter(SyncSolutionPutterPort),

    Getter(SyncSolutionGetterPort),

}

pub type SyncLocalSolution = Vec<SyncLocalSolutionEntry>;

/// Getter port in a solution. Upon receiving a message it is certain about who

/// its peer is.

#[derive(Debug)]

pub struct SyncSolutionGetterPort {

            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

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

pub(crate) fn default_attempt_get(

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

    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;

    if port_info.state.is_closed() {

        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

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

            // Note: we can still run into an unrecoverable error when actually

            // receiving this message

            match default_handle_received_data_message(

                comp_ctx, sched_ctx, control,

            ) {

                Ok(()) => return GetResult::Received(message),

                Err(location_and_message) => return GetResult::Error(location_and_message)

            }

        } else {

            // We're not allowed to receive this message. This means that the

            // receiver is attempting to receive something out of order with

            // respect to the sender.

            return GetResult::Error((target_port_instruction, String::from(

                "Cannot get from this port, as this causes a deadlock. This happens if you `get` in a different order as another component `put`s"

            )));

        }

    } else {

        // We don't have a message waiting for us and the port is not blocked.

        // So enter the BlockedGet state

        exec_state.set_as_blocked_get(target_port);

        return GetResult::NoMessage;

    }

}

/// Default handling that has been received through a `get`. Will check if any

/// more messages are waiting, and if the corresponding port was blocked because

/// of full buffers (hence, will use the control layer to make sure the peer

/// will become unblocked).

pub(crate) fn default_handle_received_data_message(

    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

    let port_info = comp_ctx.get_port(port_handle);

    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

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

    if port_info.state.is_set(PortStateFlag::BlockedDueToFullBuffers) {

        let port_info = comp_ctx.get_port_mut(port_handle);

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

        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,

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

    match message.content {

        ControlMessageContent::Ack => {

        },

        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.

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

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

pub(crate) fn default_handle_exit(_exec_state: &CompExecState) -> CompScheduling {

    debug_assert_eq!(_exec_state.mode, CompMode::Exit);

    return CompScheduling::Exit;

}

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

// Internal messaging/state utilities

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

/// Sends a message without any transmitted ports. Does not check if sending

/// is actually valid.

fn send_message_without_ports(

    sending_port_handle: LocalPortHandle, value: ValueGroup,

    comp_ctx: &CompCtx, sched_ctx: &SchedulerCtx, consensus: &mut Consensus,

) {

    let port_info = comp_ctx.get_port(sending_port_handle);

    debug_assert!(port_info.state.can_send());

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

}

    exec_state: &mut CompExecState, comp_ctx: &mut CompCtx, sched_ctx: &SchedulerCtx,

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

    debug_assert_eq!(exec_state.mode, CompMode::Sync); // busy in sync, trying to send

    let sending_port_info = comp_ctx.get_port_mut(sending_port_handle);

    sending_port_info.last_instruction = sending_port_instruction;

    let mut transmit_ports = Vec::new();

    find_ports_in_value_group(&value, &mut transmit_ports);

    debug_assert!(!transmit_ports.is_empty()); // requisite for calling this function

    // Set up the final Ack that triggers us to send our final message

    let unblock_put_control_id = control.add_unblock_put_with_ports_entry();

    for (_, port_id) in &transmit_ports {

        let transmit_port_handle = comp_ctx.get_port_handle(*port_id);

        let transmit_port_info = comp_ctx.get_port_mut(transmit_port_handle);

        // Note: we checked earlier that we are currently in sync mode. Now we

        // will check if we've already used the port we're about to transmit.

        if !transmit_port_info.last_instruction.is_none() {

            return Err((

                sending_port_instruction,

            ));

        }

        if transmit_port_info.state.is_set(PortStateFlag::Transmitted) {

            return Err((

                sending_port_instruction,

            ));

        }

        transmit_port_info.state.set(PortStateFlag::Transmitted);

    return Ok(());

}

/// Handles an `Ack` for the control layer.

fn default_handle_ack(

) {

    // Since an `Ack` may cause another one, handle them in a loop

    let mut to_ack = control_id;

    loop {

        let (action, new_to_ack) = control.handle_ack(to_ack, sched_ctx, comp_ctx);

        match action {

            AckAction::SendMessage(target_comp, message) => {

                // FIX @NoDirectHandle

                let mut handle = sched_ctx.runtime.get_component_public(target_comp);

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

                let _should_remove = handle.decrement_users();

                debug_assert!(_should_remove.is_none());

            },

            AckAction::ScheduleComponent(to_schedule) => {

                // FIX @NoDirectHandle

                let mut handle = sched_ctx.runtime.get_component_public(to_schedule);

                // Note that the component is intentionally not

                // sleeping, so we just wake it up

                debug_assert!(!handle.sleeping.load(std::sync::atomic::Ordering::Acquire));

                let key = unsafe { to_schedule.upgrade() };

                sched_ctx.runtime.enqueue_work(key);

                let _should_remove = handle.decrement_users();

                debug_assert!(_should_remove.is_none());

            },

            AckAction::None => {}

        }

        match new_to_ack {

            Some(new_to_ack) => to_ack = new_to_ack,

            None => break,

        }

    }

}

/// Little helper for sending the most common kind of `Ack`

fn default_send_ack(

    causer_of_ack_id: ControlId, peer_handle: LocalPeerHandle,

    sched_ctx: &SchedulerCtx, comp_ctx: &CompCtx

) {

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

}

                ports.push((vec![value_location], cur_port));

            },

            Value::Array(heap_pos) |

            Value::Message(heap_pos) |

            Value::String(heap_pos) |

            Value::Struct(heap_pos) |

            Value::Union(_, heap_pos) => {

                // Reference to some dynamic thing which might contain ports,

                // so recurse

                let heap_region = &group.regions[*heap_pos as usize];

                for (value_index, embedded_value) in heap_region.iter().enumerate() {

                    let value_location = ValueId::Heap(*heap_pos, value_index as u32);

                    find_port_in_value(group, embedded_value, value_location, ports);

                }

            },

            _ => {}, // values we don't care about

        }

    }

    // Clear the ports, then scan all the available values

    ports.clear();

    for (value_index, value) in &value_group.values.iter().enumerate() {

        find_port_in_value(value_group, value, ValueId::Stack(value_index as u32), ports);

    }

}

    pub fn is_none(&self) -> bool {

        match self {

            PortInstruction::None => return true,

            _ => return false,

        }

    }

}

/// Directionality of a port

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

pub enum PortKind {

    Putter,

    Getter,

}

/// Bitflags for port

// TODO: Incorporate remaining flags from `Port` struct

#[repr(u32)]

#[derive(Debug, Copy, Clone)]

pub enum PortStateFlag {

    Closed = 0x01, // If not closed, then the port is open

    BlockedDueToPeerChange = 0x02, // busy changing peers, hence use of port is temporarily blocked

    BlockedDueToFullBuffers = 0x04,

    Transmitted = 0x08, // Transmitted, so cannot be used anymore

}

#[derive(Copy, Clone)]

pub struct PortState {

    flags: u32

}

impl PortState {

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

        return PortState{ flags: 0 }

    }

    // high-level

    #[inline]

    pub fn is_open(&self) -> bool {

        return !self.is_closed();

    }

    #[inline]

    pub fn can_send(&self) -> bool {

        return

            !self.is_set(PortStateFlag::Closed) &&

    }

    #[inline]

    pub fn is_closed(&self) -> bool {

        return self.is_set(PortStateFlag::Closed);

    }

    #[inline]

    pub fn is_blocked(&self) -> bool {

        return

            self.is_set(PortStateFlag::BlockedDueToPeerChange) ||

            self.is_set(PortStateFlag::BlockedDueToFullBuffers);

    }

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

        ] {

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

}

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

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

use crate::runtime2::component::*;

use super::component_context::*;

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

pub(crate) struct ControlId(u32);

impl ControlId {

    /// Like other invalid IDs, this one doesn't care any significance, but is

    /// just set at u32::MAX to hopefully bring out bugs sooner.

        return ControlId(u32::MAX);

    }

}

struct ControlEntry {

    id: ControlId,

    ack_countdown: u32,

    content: ControlContent,

}

enum ControlContent {

    PeerChange(ContentPeerChange),

    ScheduleComponent(CompId),

    ClosedPort(PortId),

    UnblockPutWithPorts

}

struct ContentPeerChange {

    source_port: PortId,

    source_comp: CompId,

    old_target_port: PortId,

    new_target_port: PortId,

    new_target_comp: CompId,

    schedule_entry_id: ControlId,

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

                debug_assert!(comp_ctx.get_port(port_handle).state.is_closed());

                comp_ctx.change_port_peer(sched_ctx, port_handle, None);

                return (AckAction::None, None);

            },

            ControlContent::UnblockPutWithPorts => {

                return (AckAction::UnblockPutWithPorts, None);

            }

        }

    }

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

        return !self.entries.is_empty();

    }

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

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

    /// Adds an entry that, when completely ack'd, will schedule a component.

    pub(crate) fn add_schedule_entry(&mut self, to_schedule_id: CompId) -> ControlId {

        let entry_id = self.take_id();

        self.entries.push(ControlEntry{

            id: entry_id,

            ack_countdown: 0, // incremented by calls to `add_reroute_entry`

            content: ControlContent::ScheduleComponent(to_schedule_id),

        });

        return entry_id;

    }

    /// Adds an entry that returns the similarly named Ack action

    pub(crate) fn add_unblock_put_with_ports_entry(&mut self) -> ControlId {

        let entry_id = self.take_id();

        self.entries.push(ControlEntry{

            id: entry_id,

            ack_countdown: 1, // incremented by calls to `add_reroute_entry`

            content: ControlContent::UnblockPutWithPorts,

        });

        return entry_id;

    }

    pub(crate) fn add_reroute_entry(

        &mut self, creator_comp_id: CompId,

        source_port_id: PortId, source_comp_id: CompId,

        old_target_port_id: PortId, new_target_port_id: PortId, new_comp_id: CompId,

        schedule_entry_id: ControlId,

    ) -> Message {

        let entry_id = self.take_id();

        self.entries.push(ControlEntry{

            id: entry_id,

            ack_countdown: 1,

            content: ControlContent::PeerChange(ContentPeerChange{

                source_port: source_port_id,

                source_comp: source_comp_id,

                old_target_port: old_target_port_id,

                new_target_port: new_target_port_id,

                new_target_comp: new_comp_id,

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

            content: ControlMessageContent::PortPeerChangedBlock

        })

    }

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

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

                if *entry_port_id == port_id {

                    return Some(entry.id);

                }

            }

        }

        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;

        debug_assert!(port.state.is_closed());