pub struct ControlMessage {

    pub id: ControlId,

    pub sender_comp_id: CompId,

    pub target_port_id: Option<PortId>,

    pub content: ControlMessageContent,

}

#[derive(Copy, Clone)]

pub enum ControlMessageContent {

    Ack,

    BlockPort(PortId),

    UnblockPort(PortId),

    PortPeerChangedBlock(PortId),

    PortPeerChangedUnblock(PortId, CompId),

}

pub enum Message {

    Data(DataMessage),

    Control(ControlMessage),

}

impl Message {

    pub(crate) fn target_port(&self) -> Option<PortId> {

        match self {

use crate::protocol::*;

use crate::protocol::eval::{

    PortId as EvalPortId, Prompt,

    ValueGroup, Value,

    EvalContinuation, EvalResult, EvalError

};

use crate::runtime2::store::QueueDynMpsc;

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

use crate::runtime2::scheduler::SchedulerCtx;

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

use super::control_layer::*;

use super::consensus::Consensus;

pub enum CompScheduling {

    Immediate,

    Requeue,

    Sleep,

    Exit,

}

pub struct CompCtx {

    pub id: CompId,

    }

    pub(crate) fn get_port_index(&self, port_id: PortId) -> Option<usize> {

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

            if port.self_id == port_id {

                return Some(index);

            }

        }

        return None;

    }

        let index = self.get_peer_index(peer_id).unwrap();

        return &self.peers[index];

    }

    fn get_peer_mut(&mut self, peer_id: CompId) -> &mut Peer {

        let index = self.get_peer_index(peer_id).unwrap();

        return &mut self.peers[index];

    }

    pub(crate) fn get_peer_index(&self, peer_id: CompId) -> Option<usize> {

        for (index, peer) in self.peers.iter().enumerate() {

            if peer.id == peer_id {

    pub inbox_backup: Vec<DataMessage>,

}

impl CompPDL {

    pub(crate) fn new(initial_state: Prompt) -> Self {

        return Self{

            mode: Mode::NonSync,

            mode_port: PortId::new_invalid(),

            mode_value: ValueGroup::default(),

            prompt: initial_state,

            control: ControlLayer::default(),

            consensus: Consensus::new(),

            exec_ctx: ExecCtx{

                stmt: ExecStmt::None,

            },

            inbox_main: Vec::new(),

            inbox_backup: Vec::new(),

        }

    }

    pub(crate) fn handle_setup(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx) {

        self.inbox.resize(comp_ctx.ports.len(), None);

    }

            wake_up_if_sleeping(sched_ctx, target_comp_id, &peer.handle);

        }

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

        self.inbox_backup.push(message);

    }

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

        match message.content {

            ControlMessageContent::Ack => {

                let mut to_ack = message.id;

                loop {

                    match action {

                        AckAction::SendMessageAndAck(target_comp, message, new_to_ack) => {

                            // FIX @NoDirectHandle

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

                            handle.inbox.push(Message::Control(message));

                            wake_up_if_sleeping(sched_ctx, target_comp, &handle);

                            to_ack = new_to_ack;

                        },

                        AckAction::ScheduleComponent(to_schedule) => {

                            // FIX @NoDirectHandle

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

                            wake_up_if_sleeping(sched_ctx, to_schedule, &handle);

                }

            },

            ControlMessageContent::BlockPort(port_id) => {

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

                // blocked.

                let port_info = comp_ctx.get_port_mut(port_id);

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

                if port_info.state != PortState::Closed {

                    debug_assert_ne!(port_info.state, PortState::Blocked); // implies unnecessary messages

                    port_info.state = PortState::Blocked;

                }

            },

            ControlMessageContent::UnblockPort(port_id) => {

                // We were previously blocked (or already closed)

                let port_info = comp_ctx.get_port(port_id);

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

                debug_assert!(port_info.state == PortState::Blocked || port_info.state == PortState::Closed);

                if port_info.state == PortState::Blocked {

                    self.unblock_port(sched_ctx, comp_ctx, port_id);

                }

            },

            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

                    find_port_in_value(group, embedded_value, ports);

                }

            },

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

        }

    }

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

    ports.clear();

    for value in &value_group.values {

        find_port_in_value(value_group, value, ports);

    }

}

        return ControlId(u32::MAX);

    }

}

struct ControlEntry {

    id: ControlId,

    ack_countdown: u32,

    content: ControlContent,

}

enum ControlContent {

    PeerChange(ContentPeerChange),

}

struct ContentPeerChange {

    source_port: PortId,

    source_comp: CompId,

    target_port: PortId,

    new_target_comp: CompId,

    schedule_entry_id: ControlId,

}

pub(crate) enum AckAction {

    None,

    SendMessageAndAck(CompId, ControlMessage, ControlId),

    ScheduleComponent(CompId),

}

/// Handling/sending control messages.

pub(crate) struct ControlLayer {

    id_counter: ControlId,

    entries: Vec<ControlEntry>,

}

        let target_port = target_port.unwrap();

        for entry in &self.entries {

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

                if entry.target_port == target_port {

                    return Some(entry.new_target_comp);

                }

            }

        }

        return None;

    }

        let entry_index = self.get_entry_index(entry_id).unwrap();

        let entry = &mut self.entries[entry_index];

        debug_assert!(entry.ack_countdown > 0);

        entry.ack_countdown -= 1;

        if entry.ack_countdown != 0 {

            return AckAction::None;

        }

        // All `Ack`s received, take action based on the kind of entry

        match &entry.content {

            ControlContent::PeerChange(content) => {

                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.source_port,

                        content.new_target_comp

                    )

                };

                let to_ack = content.schedule_entry_id;

                self.entries.remove(entry_index);

                return AckAction::SendMessageAndAck(target_comp_id, message_to_send, to_ack);

            },

            ControlContent::ScheduleComponent(content) => {

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

                // schedule the component!

                return AckAction::ScheduleComponent(content.to_schedule);

            },

            ControlContent::BlockedPort(_) => unreachable!(),

        }

    }

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

    // 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 entry_id = self.take_id();

        self.entries.push(ControlEntry{

            id: entry_id,

            }

        }

        return Message::Control(ControlMessage{

            id: entry_id,

            sender_comp_id: creator_comp_id,

            target_port_id: Some(source_port_id),

            content: ControlMessageContent::PortPeerChangedBlock(source_port_id)

        })

    }

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

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

    pub(crate) fn mark_port_blocked(&mut self, port_id: PortId, comp_ctx: &mut CompCtx) -> (CompId, ControlMessage) {

        // TODO: Feels like this shouldn't be an entry. Hence this class should

        //  be renamed. Lets see where the code ends up being

        let entry_id = self.take_id();

        let port_info = comp_ctx.get_port_mut(port_id);

        debug_assert_eq!(port_info.state, PortState::Open); // prevent unforeseen issues

        port_info.state = PortState::Blocked;

        self.entries.push(ControlEntry{

            id: entry_id,

            ack_countdown: 0,

            content: ControlContent::BlockedPort(ContentBlockedPort{

mod component_pdl;

mod control_layer;

mod consensus;

pub(crate) use component_pdl::{CompPDL, CompCtx, CompScheduling};

use super::communication::Message;

use super::component::{CompCtx, CompPDL};

use super::store::{ComponentStore, QueueDynMpsc, QueueDynProducer};

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

// Component

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

/// Key to a component. Type system somewhat ensures that there can only be one

/// of these. Only with a key one may retrieve privately-accessible memory for

/// a component. Practically just a generational index, like `CompId` is.

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

impl CompKey {

    pub(crate) fn downgrade(&self) -> CompId {

        return CompId(self.0);

    }

}

/// Generational ID of a component

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

impl CompId {

    pub(crate) fn new_invalid() -> CompId {

        return CompId(u32::MAX);

    }

    /// Upgrade component ID to component key. Unsafe because the caller needs

    /// to make sure that only one component key can exist at a time (to ensure

    /// a component can only be scheduled/executed by one thread).

    pub(crate) unsafe fn upgrade(&self) -> CompKey {

        return CompKey(self.0);

    }

}

/// Private fields of a component, may only be modified by a single thread at

/// a time.

pub(crate) struct RuntimeComp {

    pub public: CompPublic,

    pub code: CompPDL,

    pub ctx: CompCtx,

    pub inbox: QueueDynMpsc<Message>

}

/// Should contain everything that is accessible in a thread-safe manner

pub(crate) struct CompPublic {

    pub sleeping: AtomicBool,

    pub num_handles: AtomicU32, // manually modified (!)

    pub inbox: QueueDynProducer<Message>,

}

/// Handle to public part of a component. Would be nice if we could

/// automagically manage the `num_handles` counter. But when it reaches zero we

/// need to manually remove the handle from the runtime. So we just have debug

/// code to make sure this actually happens.

pub(crate) struct CompHandle {

    target: *const CompPublic,

    pub(crate) fn get_component(&self, key: CompKey) -> &mut RuntimeComp {

        let component = self.components.get_mut(key.0);

        return component;

    }

    pub(crate) fn get_component_public(&self, id: CompId) -> CompHandle {

        let component = self.components.get(id.0);

        return CompHandle::new(&component.public);

    }

    pub(crate) fn destroy_component(&self, key: CompKey) {

        self.components.destroy(key.0);

    }

    // Interacting with components

}

    pub fn run(&mut self) {

        let mut scheduler_ctx = SchedulerCtx::new(&*self.runtime);

        'run_loop: loop {

            // Wait until we have something to do (or need to quit)

            let comp_key = self.runtime.take_work();

            if comp_key.is_none() {

                break 'run_loop;

            }

            let comp_key = comp_key.unwrap();

            let component = self.runtime.get_component(comp_key);

            // Run the component until it no longer indicates that it needs to

            // be re-executed immediately.

            let mut new_scheduling = CompScheduling::Immediate;

            while let CompScheduling::Immediate = new_scheduling {

                new_scheduling = component.code.run(&mut scheduler_ctx, &mut component.private.ctx).expect("TODO: Handle error");

            }

            // Handle the new scheduling

            match new_scheduling {

                CompScheduling::Immediate => unreachable!(),

                CompScheduling::Requeue => { self.runtime.enqueue_work(comp_key); },

                CompScheduling::Sleep => { self.mark_component_as_sleeping(comp_key, component); },

            }

        }

    }

    // local utilities

    fn mark_component_as_sleeping(&self, key: CompKey, component: &mut RuntimeComp) {

        debug_assert_eq!(key.downgrade(), component.private.ctx.id); // make sure component matches key

        debug_assert_eq!(component.public.sleeping.load(Ordering::Acquire), false); // we're executing it, so it cannot be sleeping

        component.public.sleeping.store(true, Ordering::Release);

    }

    }

}

                write_head: AtomicU32::new(initial_capacity),

                limit_head: AtomicU32::new(initial_capacity),

                #[cfg(debug_assertions)] dbg: AtomicU32::new(0),

            }),

        };

    }

    #[inline]

    pub fn producer(&self) -> QueueDynProducer<T> {

        return QueueDynProducer::new(self);

    }

    /// Perform an attempted read from the queue. It might be that some producer

    /// is putting something in the queue while this function is executing, and

    /// we don't get the consume it.

    pub fn pop(&mut self) -> Option<T> {

        let data_lock = self.inner.data.lock_shared();

        let cur_read = self.inner.read_head.load(Ordering::Acquire);

        let cur_limit = self.inner.limit_head.load(Ordering::Acquire);

        let buf_size = data_lock.data.cap() as u32;

        if (cur_read + buf_size) & data_lock.compare_mask != cur_limit {

            // Make a bitwise copy of the value and return it. The receiver is

            // responsible for dropping it.