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

// Generic types

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

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

    }

}

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

pub enum PortKind {

    Putter,

    Getter,

}

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

pub enum PortState {

    Open,

    Closed,

}

pub struct Channel {

    pub putter_id: PortId,

    pub getter_id: PortId,

}

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

// Data messages

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

#[derive(Debug)]

pub struct DataMessage {

    pub data_header: MessageDataHeader,

    pub sync_header: MessageSyncHeader,

    pub content: ValueGroup,

}

#[derive(Debug)]

pub struct MessageDataHeader {

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

    pub new_mapping: u32,

    pub source_port: PortId,

    pub target_port: PortId,

}

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

                        // Message was received. Make sure any blocked peers and

                        // pending messages are handled.

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

                        self.handle_received_data_message(sched_ctx, comp_ctx, port_handle);

                        self.exec_ctx.stmt = ExecStmt::PerformedGet(message.content);

                        return Ok(CompScheduling::Immediate);

                    } else {

                        todo!("handle sync failure due to message deadlock");

                        return Ok(CompScheduling::Sleep);

                    }

                } else {

                    // We need to wait

                    self.mode = Mode::BlockedGet;

                    self.mode_port = port_id;

                    return Ok(CompScheduling::Sleep);

                }

            },

            EC::Put(port_id, value) => {

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

                let port_id = port_id_from_eval(port_id);

                let port_handle = comp_ctx.get_port_handle(port_id);

                let port_info = comp_ctx.get_port(port_handle);

                }

            },

            // Results that can be returned outside of sync mode

            EC::ComponentTerminated => {

                self.mode = Mode::StartExit; // next call we'll take care of the exit

                return Ok(CompScheduling::Immediate);

            },

            EC::SyncBlockStart => {

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

                self.handle_sync_start(sched_ctx, comp_ctx);

                return Ok(CompScheduling::Immediate);

            },

            EC::NewComponent(definition_id, monomorph_idx, arguments) => {

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

                self.create_component_and_transfer_ports(

                    sched_ctx, comp_ctx,

                    definition_id, monomorph_idx, arguments

                );

                return Ok(CompScheduling::Requeue);

            },

            EC::NewChannel => {

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

                debug_assert!(self.exec_ctx.stmt.is_none());

                let channel = comp_ctx.create_channel();

                self.exec_ctx.stmt = ExecStmt::CreatedChannel((

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

    fn send_data_message_and_wake_up(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &CompCtx, source_port_handle: LocalPortHandle, value: ValueGroup) {

        let port_info = comp_ctx.get_port(source_port_handle);

        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 = self.consensus.annotate_data_message(comp_ctx, port_info, value);

        peer_info.handle.send_message(sched_ctx, Message::Data(annotated_message), true);

    }

    /// Handles a message that came in through the public inbox. This function

    /// will handle putting it in the correct place, and potentially blocking

    /// the port in case too many messages are being received.

    fn handle_incoming_data_message(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, message: DataMessage) {

        // Check if we can insert it directly into the storage associated with

        // the port

        let target_port_id = message.data_header.target_port;

        let port_handle = comp_ctx.get_port_handle(target_port_id);

        let port_index = comp_ctx.get_port_index(port_handle);

        if self.inbox_main[port_index].is_none() {

            self.inbox_main[port_index] = Some(message);

            // After direct insertion, check if this component's execution is 

            // blocked on receiving a message on that port

            if self.mode == Mode::BlockedGet && self.mode_port == target_port_id {

                // We were indeed blocked

                self.mode = Mode::Sync;

                self.mode_port = PortId::new_invalid();

            }

            return;

        }

        // The direct inbox is full, so the port will become (or was already) blocked

        let port_info = comp_ctx.get_port_mut(port_handle);

        if port_info.state == PortState::Open {

            let (peer_handle, message) =

                self.control.initiate_port_blocking(comp_ctx, port_handle);

            let peer = comp_ctx.get_peer(peer_handle);

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

        }

        // But we still need to remember the message, so:

        self.inbox_backup.push(message);

    }

    /// Handles when a message has been handed off from the inbox to the PDL

    /// code. We check to see if there are more messages waiting and, if not,

    /// then we handle the case where the port might have been blocked

    /// previously.

    fn handle_received_data_message(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, port_handle: LocalPortHandle) {

        let port_index = comp_ctx.get_port_index(port_handle);

        debug_assert!(self.inbox_main[port_index].is_none()); // this function should be called after the message is taken out

        // Check for any more messages

        let port_info = comp_ctx.get_port(port_handle);

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

            let message = &self.inbox_backup[message_index];

            if message.data_header.target_port == port_info.self_id {

                // One more message for this port

                let message = self.inbox_backup.remove(message_index);

                self.inbox_main[port_index] = Some(message);

                return;

            }

        }

        // Did not have any more messages. So if we were blocked, then we need

        // to send the "unblock" message.

            comp_ctx.set_port_state(port_handle, PortState::Open);

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

            let peer_info = comp_ctx.get_peer(peer_handle);

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

        }

    }

    fn handle_incoming_control_message(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, message: ControlMessage) {

        // Little local utility to send an Ack

        fn send_control_ack_message(sched_ctx: &SchedulerCtx, comp_ctx: &CompCtx, causer_id: ControlId, peer_handle: LocalPeerHandle) {

            let peer_info = comp_ctx.get_peer(peer_handle);

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

                id: causer_id,

                sender_comp_id: comp_ctx.id,

                target_port_id: None,

                content: ControlMessageContent::Ack,

            }), true);

        }

        // Handle the content of the control message, and optionally Ack it

        match message.content {

            ControlMessageContent::Ack => {

                self.handle_ack(sched_ctx, comp_ctx, message.id);

            },

            ControlMessageContent::BlockPort(port_id) => {

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

                // blocked.

                let port_handle = comp_ctx.get_port_handle(port_id);

                let port_info = comp_ctx.get_port(port_handle);

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

                }

            },

            ControlMessageContent::ClosePort(port_id) => {

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

                // the component handle as well

                let port_handle = comp_ctx.get_port_handle(port_id);

                let peer_comp_id = comp_ctx.get_port(port_handle).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) = self.control.has_close_port_entry(port_handle, comp_ctx) {

                    self.handle_ack(sched_ctx, comp_ctx, control_id);

                } else {

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

                    comp_ctx.remove_peer(sched_ctx, port_handle, peer_comp_id, false); // do not remove if closed

                    comp_ctx.set_port_state(port_handle, PortState::Closed); // now set to closed

                }

            },

            ControlMessageContent::UnblockPort(port_id) => {

                // We were previously blocked (or already closed)

                let port_handle = comp_ctx.get_port_handle(port_id);

                let port_info = comp_ctx.get_port(port_handle);

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

                    self.handle_unblock_port_instruction(sched_ctx, comp_ctx, port_handle);

                }

            },

            ControlMessageContent::PortPeerChangedBlock(port_id) => {

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

                debug_assert_eq!(message.target_port_id, Some(port_id));

                let port_handle = comp_ctx.get_port_handle(port_id);

                let port_info = comp_ctx.get_port(port_handle);

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

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

            },

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

                let port_handle = comp_ctx.get_port_handle(message.target_port_id.unwrap());

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

                self.handle_unblock_port_instruction(sched_ctx, comp_ctx, port_handle);

            }

        }

    }

    fn handle_incoming_sync_message(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, message: SyncMessage) {

        let decision = self.consensus.receive_sync_message(sched_ctx, comp_ctx, message);

        debug_assert!(self.mode == Mode::Sync || self.mode == Mode::SyncEnd);

        self.handle_sync_decision(sched_ctx, comp_ctx, decision);

    }

    /// Little helper that notifies the control layer of an `Ack`, and takes the

    /// appropriate subsequent action

    fn handle_ack(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, control_id: ControlId) {

        let mut to_ack = control_id;

        loop {

                    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,

            }

        }

    }

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

    // Handling ports

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

    /// Unblocks a port, potentially continuing execution of the component, in

    /// response to a message that told us to unblock a previously blocked

    fn handle_unblock_port_instruction(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, port_handle: LocalPortHandle) {

        let port_info = comp_ctx.get_port_mut(port_handle);

        let port_id = port_info.self_id;

        port_info.state = PortState::Open;

        if self.mode == Mode::BlockedPut && port_id == self.mode_port {

            // We were blocked on the port that just became unblocked, so

            // send the message.

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

            let mut replacement = ValueGroup::default();

            std::mem::swap(&mut replacement, &mut self.mode_value);

            self.send_data_message_and_wake_up(sched_ctx, comp_ctx, port_handle, replacement);

            self.mode = Mode::Sync;

            self.mode_port = PortId::new_invalid();

        }

    }

    fn create_component_and_transfer_ports(

        &mut self,

        sched_ctx: &SchedulerCtx, creator_ctx: &mut CompCtx,

        definition_id: DefinitionId, monomorph_index: i32, mut arguments: ValueGroup

    ) {

        struct PortPair{

            creator_handle: LocalPortHandle,

            creator_id: PortId,

            created_handle: LocalPortHandle,

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.

    fn new_invalid() -> Self {

        return ControlId(u32::MAX);

    }

}

struct ControlEntry {

    id: ControlId,

    ack_countdown: u32,

    content: ControlContent,

}

enum ControlContent {

    PeerChange(ContentPeerChange),

    ScheduleComponent(CompId),

    ClosedPort(PortId),

}

struct ContentPeerChange {

    source_port: PortId,

    source_comp: CompId,

    old_target_port: PortId,

    new_target_port: PortId,

    new_target_comp: CompId,

    schedule_entry_id: ControlId,

}

struct ControlClosedPort {

    closed_port: PortId,

    exit_entry_id: Option<ControlId>,

}

pub(crate) enum AckAction {

    None,

    SendMessage(CompId, ControlMessage),

    ScheduleComponent(CompId),

}

/// Handling/sending control messages.

                // rerouted all of the messages, and the sender's port can now

                // be unblocked again.

                let target_comp_id = content.source_comp;

                let message_to_send = ControlMessage{

                    id: ControlId::new_invalid(),

                    sender_comp_id: comp_ctx.id,

                    target_port_id: Some(content.source_port),

                    content: ControlMessageContent::PortPeerChangedUnblock(

                        content.new_target_port,

                        content.new_target_comp

                    )

                };

                let to_ack = content.schedule_entry_id;

                return (

                    AckAction::SendMessage(target_comp_id, message_to_send),

                    Some(to_ack)

                );

            },

            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;

                debug_assert_eq!(port_info.state, PortState::Closed);

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

    }

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

    // Port transfer (due to component creation)

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

    /// 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 peer_port_id = port.peer_port_id;

        debug_assert!(port.state == PortState::Closed);

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

        });

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

            }

        );

    }

    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{

                sender_comp_id: comp_ctx.id,

                target_port_id: Some(port_info.peer_port_id),

                content: ControlMessageContent::BlockPort(peer_port_id),

            }

        );

    }

    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

        debug_assert_eq!(port_info.state, PortState::Open); // contract with caller, the locally stored entry ensures we were blocked before

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

        return (

            peer_handle,

            ControlMessage{

                sender_comp_id: comp_ctx.id,

                target_port_id: Some(port_info.peer_port_id),

                content: ControlMessageContent::UnblockPort(port_info.peer_port_id)

            }

        );

    }

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

    // Internal utilities

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

    fn take_id(&mut self) -> ControlId {

        let id = self.id_counter;

        self.id_counter.0 = self.id_counter.0.wrapping_add(1);

        return id;

    }

    fn get_entry_index_by_id(&self, entry_id: ControlId) -> Option<usize> {

        for (index, entry) in self.entries.iter().enumerate() {

            if entry.id == entry_id {

                return Some(index);

            }

        }

    let reserved = rt.inner.start_create_pdl_component();

    let ctx = CompCtx::new(&reserved);

    let (key, _) = rt.inner.finish_create_pdl_component(reserved, CompPDL::new(prompt, 0), ctx, false);

    rt.inner.enqueue_work(key);

}

fn no_args() -> ValueGroup { ValueGroup::new_stack(Vec::new()) }

#[test]

fn test_component_creation() {

    let pd = ProtocolDescription::parse(b"

    primitive nothing_at_all() {

        s32 a = 5;

        auto b = 5 + a;

    }

    ").expect("compilation");

    let rt = Runtime::new(1, pd);

    for i in 0..20 {

        create_component(&rt, "", "nothing_at_all", no_args());

    }

}

#[test]

    let pd = ProtocolDescription::parse(b"

    primitive sender(out<u32> o) {

        sync put(o, 1);

    }

    primitive receiver(in<u32> i) {

        sync auto a = get(i);

    }

    composite constructor() {

        channel o -> i;

        new sender(o);

        new receiver(i);

    }

    ").expect("compilation");

    let rt = Runtime::new(1, pd);

    create_component(&rt, "", "constructor", no_args());

}