fn get_channel(&mut self) -> Option<(Value, Value)> {

        todo!()

    }

}

impl Connector for ConnectorPDL {

    fn run(&mut self, sched_ctx: SchedulerCtx, comp_ctx: &mut ComponentCtxFancy) -> ConnectorScheduling {

        todo!()

    }

}

impl ConnectorPDL {

        Self{

            tree: ExecTree::new(initial),

            consensus: Consensus::new(),

        }

    }

    // --- Handling messages

    pub fn handle_new_messages(&mut self, ctx: &mut ComponentCtxFancy) {

        while let Some(message) = ctx.read_next_message() {

            match message {

                MessageFancy::Data(message) => handle_new_data_message(message, ctx),

        // there is one that can receive this message.

        debug_assert!(ctx.workspace_branches.is_empty());

        self.consensus.handle_received_sync_header(&message.sync_header, ctx);

        self.consensus.handle_received_data_header(&self.tree, &message.data_header, &mut ctx.workspace_branches);

        for branch_id in ctx.workspace_branches.drain(..) {

            // This branch can receive, so fork and given it the message

            let receiving_branch_id = self.tree.fork_branch(branch_id);

            self.consensus.notify_of_new_branch(branch_id, receiving_branch_id);

            let receiving_branch = &mut self.tree[receiving_branch_id];

            receiving_branch.insert_message(message.data_header.target_port, message.content.clone());

            // And prepare the branch for running

            self.tree.push_into_queue(QueueKind::Runnable, receiving_branch_id);

        }

    }

    pub fn handle_new_sync_message(&mut self, message: SyncMessageFancy, ctx: &mut ComponentCtxFancy) {

        self.consensus.handle_received_sync_header(&message.sync_header, ctx);

        todo!("handle content of message?");

    }

    // --- Running code

                // we are in non-sync mode the scheduler will handle and check

                // port ownership transfer.

                debug_assert!(comp_ctx.workspace_ports.is_empty());

                find_ports_in_value_group(&arguments, &mut comp_ctx.workspace_ports);

                let new_state = ComponentState {

                    prompt: Prompt::new(

                        &sched_ctx.runtime.protocol_description.types,

                        &sched_ctx.runtime.protocol_description.heap,

                        definition_id, monomorph_idx, arguments

                    ),

                };

                return ConnectorScheduling::Later;

            },

            RunResult::NewChannel => {

                let (getter, putter) = sched_ctx.runtime.create_channel(comp_ctx.id);

                debug_assert!(getter.kind == PortKind::Getter && putter.kind == PortKind::Putter);

                branch.prepared_channel = Some((

                    Value::Input(PortId::new(putter.self_id.index)),

                    Value::Output(PortId::new(getter.self_id.index)),

                ));

                comp_ctx.push_port(putter);

        let sync_header = SyncHeader{

            sending_component_id: ctx.id,

            highest_component_id: self.highest_connector_id,

        };

        // TODO: Handle multiple firings. Right now we just assign the current

        //  branch to the `None` value because we know we can only send once.

        debug_assert!(branch.port_mapping.iter().find(|v| v.port_id == source_port_id).unwrap().registered_id.is_none());

        let port_info = ctx.get_port_by_id(source_port_id).unwrap();

        let data_header = DataHeader{

            expected_mapping: branch.port_mapping.clone(),

            target_port: port_info.peer_id,

            new_mapping: branch_id

        };

        for mapping in &mut branch.port_mapping {

            if mapping.port_id == source_port_id {

                mapping.expected_firing = Some(true);

                mapping.registered_id = Some(branch_id);

            }

        }

        return (sync_header, data_header);

threads, and they're taken out by a single thread. These messages may contain

control messages and may be filtered or redirected by the scheduler.

The `PrivateInbox` is a temporary storage for all messages that are received

within a certain sync-round.

**/

use std::collections::VecDeque;

use std::sync::Mutex;

use super::ConnectorId;

use crate::protocol::eval::ValueGroup;

use super::connector::BranchId;

use super::port::PortIdLocal;

/// A message that has been delivered (after being imbued with the receiving

/// port by the scheduler) to a connector.

// TODO: Remove Debug on messages

#[derive(Debug, Clone)]

pub struct DataMessage {

    pub sending_port: PortIdLocal,

    pub sender_prev_branch_id: BranchId,

    pub sender_cur_branch_id: BranchId,

    pub message: ValueGroup,

pub struct Message {

    pub sending_connector: ConnectorId,

    pub receiving_port: PortIdLocal, // may be invalid (in case of messages targeted at the connector)

    pub contents: MessageContents,

}

/// The public inbox of a connector. The thread running the connector that owns

/// this inbox may retrieved from it. Non-owning threads may only put new

/// messages inside of it.

// TODO: @Optimize, lazy concurrency. Probably ringbuffer with read/write heads.

//  Should behave as a MPSC queue.

pub struct PublicInbox {

}

impl PublicInbox {

    pub fn new() -> Self {

        Self{

            messages: Mutex::new(VecDeque::new()),

        }

    }

        let mut lock = self.messages.lock().unwrap();

        lock.push_back(message);

    }

        let mut lock = self.messages.lock().unwrap();

        return lock.pop_front();

    }

    pub fn is_empty(&self) -> bool {

        let lock = self.messages.lock().unwrap();

        return lock.is_empty();

    }

}

use crate::protocol::eval::ValueGroup;

use crate::runtime2::branch::BranchId;

use crate::runtime2::ConnectorId;

use crate::runtime2::port::PortIdLocal;

pub(crate) struct PortAnnotation {

    pub port_id: PortIdLocal,

    pub registered_id: Option<BranchId>,

    pub expected_firing: Option<bool>,

}

/// The header added by the synchronization algorithm to all.

pub(crate) struct SyncHeader {

    pub sending_component_id: ConnectorId,

    pub highest_component_id: ConnectorId,

}

/// The header added to data messages

pub(crate) struct DataHeader {

    pub expected_mapping: Vec<PortAnnotation>,

    pub target_port: PortIdLocal,

    pub new_mapping: BranchId,

}

/// A data message is a message that is intended for the receiver's PDL code,

/// but will also be handled by the consensus algrorithm

pub(crate) struct DataMessageFancy {

    pub sync_header: SyncHeader,

    pub data_header: DataHeader,

    pub content: ValueGroup,

}

pub(crate) enum SyncContent {

}

/// A sync message is a message that is intended only for the consensus

/// algorithm.

pub(crate) struct SyncMessageFancy {

    pub sync_header: SyncHeader,

    pub content: SyncContent,

}

/// A control message is a message intended for the scheduler that is executing

/// a component.

pub(crate) struct ControlMessageFancy {

    pub id: u32, // generic identifier, used to match request to response

    pub content: ControlContent,

}

pub(crate) enum ControlContent {

    PortPeerChanged(PortIdLocal, ConnectorId),

    CloseChannel(PortIdLocal),

    Ack,

    Ping,

}

/// Combination of data message and control messages.

pub(crate) enum MessageFancy {

    Data(DataMessageFancy),

    Sync(SyncMessageFancy),

    Control(ControlMessageFancy),

}

// Imports

use std::collections::VecDeque;

use std::sync::{Arc, Condvar, Mutex, RwLock};

use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};

use std::thread::{self, JoinHandle};

use crate::collections::RawVec;

use crate::ProtocolDescription;

use inbox::Message;

use scheduler::{Scheduler, ControlMessageHandler};

use native::{Connector, ConnectorApplication, ApplicationInterface};

use crate::runtime2::port::{Port, PortState};

use crate::runtime2::scheduler::{ComponentCtxFancy, SchedulerCtx};

/// A kind of token that, once obtained, allows mutable access to a connector.

/// We're trying to use move semantics as much as possible: the owner of this

/// key is the only one that may execute the connector's code.

pub(crate) struct ConnectorKey {

    pub index: u32, // of connector

}

impl ConnectorKey {

    /// Downcasts the `ConnectorKey` type, which can be used to obtain mutable

            self_id: putter_id,

            peer_id: getter_id,

            kind: PortKind::Putter,

            state: PortState::Open,

            peer_connector: creating_connector,

        };

        return (getter_port, putter_port);

    }

    /// Sends a message to a particular connector. If the connector happened to

    /// be sleeping then it will be scheduled for execution.

        let target = self.get_component_public(target_id);

        target.inbox.insert_message(message);

        let should_wake_up = target.sleeping

            .compare_exchange(true, false, Ordering::SeqCst, Ordering::Acquire)

            .is_ok();

        if should_wake_up {

            let key = unsafe{ ConnectorKey::from_id(target_id) };

            self.push_work(key);

        }

    }

use std::collections::VecDeque;

use std::sync::Arc;

use std::sync::atomic::Ordering;

use super::{ScheduledConnector, RuntimeInner, ConnectorId, ConnectorKey, ConnectorVariant};

use super::port::{Port, PortState, PortIdLocal};

use super::native::Connector;

// Because it contains pointers we're going to do a copy by value on this one

#[derive(Clone, Copy)]

pub(crate) struct SchedulerCtx<'a> {

    pub(crate) runtime: &'a RuntimeInner

}

pub(crate) struct Scheduler {

    runtime: Arc<RuntimeInner>,

    scheduler_id: u32,

}

                },

                ConnectorScheduling::Exit => {

                    // Prepare for exit. Set the shutdown flag and broadcast

                    // messages to notify peers of closing channels

                    scheduled.shutting_down = true;

                    for port in &scheduled.ctx_fancy.ports {

                        if port.state != PortState::Closed {

                            let message = scheduled.router.prepare_closing_channel(

                                port.self_id, port.peer_id,

                                connector_id

                            );

                            self.debug_conn(connector_id, &format!("Sending message [ exit ] \n --- {:?}", message));

                        }

                    }

                    if scheduled.router.num_pending_acks() == 0 {

                        self.runtime.destroy_component(connector_key);

                        continue 'thread_loop;

                    }

                    self.try_go_to_sleep(connector_key, scheduled);

                }

            }

        }

        while let Some(message) = scheduled.public.inbox.take_message() {

            // Check for rerouting

            self.debug_conn(connector_id, &format!("Handling message from conn({}) at port({})\n --- {:?}", message.sending_connector.0, message.receiving_port.index, message));

            if let Some(other_connector_id) = scheduled.router.should_reroute(message.sending_connector, message.receiving_port) {

                self.debug_conn(connector_id, &format!(" ... Rerouting to connector {}", other_connector_id.0));

                self.runtime.send_message(other_connector_id, message);

                continue;

            }

            // Handle special messages here, messages for the component

            // will be added to the inbox.

            self.debug_conn(connector_id, " ... Handling message myself");

                            // Need to change port target

                            let port = scheduled.ctx_fancy.get_port_mut_by_id(port_id).unwrap();

                            port.peer_connector = new_target_connector_id;

                            // Note: for simplicity we program the scheduler to always finish

                            // running a connector with an empty outbox. If this ever changes

                            // then accepting the "port peer changed" message implies we need

                            // to change the recipient of the message in the outbox.

                            debug_assert!(scheduled.ctx_fancy.outbox.is_empty());

                            // And respond with an Ack

                            self.debug_conn(connector_id, &format!("Sending message [pp ack]\n --- {:?}", ack_message));

                        },

                            // Mark the port as being closed

                            let port = scheduled.ctx_fancy.get_port_mut_by_id(port_id).unwrap();

                            port.state = PortState::Closed;

                            // Send an Ack

                            self.debug_conn(connector_id, &format!("Sending message [cc ack] \n --- {:?}", ack_message));

                        },

                            scheduled.router.handle_ack(content.id);

                    }

                },

                _ => {

                    // All other cases have to be handled by the component

                    scheduled.ctx_fancy.inbox_messages.push(message);

                }

            }

        }

    }

    /// Handles changes to the context that were made by the component. This is

    /// the way (due to Rust's borrowing rules) that we bubble up changes in the

    /// component's state that the scheduler needs to know about (e.g. a message

    fn handle_changes_in_context(&mut self, scheduled: &mut ScheduledConnector) {

        let connector_id = scheduled.ctx_fancy.id;

        // Handling any messages that were sent

        while let Some(mut message) = scheduled.ctx_fancy.outbox.pop_front() {

            self.debug_conn(connector_id, &format!("Sending message [outbox] \n --- {:?}", message));

                    }

                },

                }

            };

        }

        while let Some(state_change) = scheduled.ctx_fancy.state_changes.pop_front() {

            match state_change {

                    // Add the new connector to the global registry

                    let new_key = self.runtime.create_pdl_component(component, false);

                    let new_connector = self.runtime.get_component_private(&new_key);

                    // Transfer ports

                        // Transfer messages associated with the transferred port

                        let mut message_idx = 0;

                        while message_idx < scheduled.ctx_fancy.inbox_messages.len() {

                            let message = &scheduled.ctx_fancy.inbox_messages[message_idx];

                            if message.receiving_port == *port_id {

                                // Need to transfer this message

                                let taken_message = scheduled.ctx_fancy.inbox_messages.remove(message_idx);

                                new_connector.ctx_fancy.inbox_messages.push(taken_message);

                            } else {

                                message_idx += 1;

                            }

                        }

                            .position(|v| v.self_id == *port_id)

                            .unwrap();

                        let port = scheduled.ctx_fancy.ports.remove(port_index);

                        new_connector.ctx_fancy.ports.push(port.clone());

                        // Notify the peer that the port has changed

                        let reroute_message = scheduled.router.prepare_reroute(

                            port.self_id, port.peer_id, scheduled.ctx_fancy.id,

                            port.peer_connector, new_connector.ctx_fancy.id

                        );

                        self.debug_conn(connector_id, &format!("Sending message [newcon]\n --- {:?}", reroute_message));

                    }

                    // Schedule new connector to run

                    self.runtime.push_work(new_key);

                },

                ComponentStateChange::CreatedPort(port) => {

                    scheduled.ctx_fancy.ports.push(port);

                },

                ComponentStateChange::ChangedPort(port_change) => {

                    if port_change.is_acquired {

                        scheduled.ctx_fancy.ports.push(port_change.port);

                    } else {

    }

    fn debug_conn(&self, conn: ConnectorId, message: &str) {

        println!("DEBUG [thrd:{:02} conn:{:02}]: {}", self.scheduler_id, conn.0, message);

    }

}

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

// ComponentCtx

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

enum ComponentStateChange {

    CreatedPort(Port),

    ChangedPort(ComponentPortChange),

}

#[derive(Clone)]

pub(crate) struct ComponentPortChange {

    pub is_acquired: bool, // otherwise: released

    pub port: Port,

}

/// The component context (better name may be invented). This was created

/// because part of the component's state is managed by the scheduler, and part

/// exits a sync block the partially managed state by both component and

/// scheduler need to be exchanged.

pub(crate) struct ComponentCtxFancy {

    // Mostly managed by the scheduler

    pub(crate) id: ConnectorId,

    ports: Vec<Port>,

    inbox_messages: Vec<MessageFancy>, // never control or ping messages

    inbox_len_read: usize,

    // Submitted by the component

    is_in_sync: bool,

    changed_in_sync: bool,

    outbox: VecDeque<MessageFancy>,

    // Workspaces that may be used by components to (generally) prevent

    // allocations. Be a good scout and leave it empty after you've used it.

    pub workspace_ports: Vec<PortIdLocal>,

    pub workspace_branches: Vec<BranchId>,

}

impl ComponentCtxFancy {

    pub(crate) fn new_empty() -> Self {

        return Self{

            id: ConnectorId::new_invalid(),

            ports: Vec::new(),

            inbox_messages: Vec::new(),

            inbox_len_read: 0,

            is_in_sync: false,

            changed_in_sync: false,

            outbox: VecDeque::new(),

            state_changes: VecDeque::new(),

        };

    }

    /// Notify the runtime that the component has created a new component. May

    /// only be called outside of a sync block.

        debug_assert!(!self.is_in_sync);

    }

    /// Notify the runtime that the component has created a new port. May only

    /// be called outside of a sync block (for ports received during a sync

    /// block, pass them when calling `notify_sync_end`).

    pub(crate) fn push_port(&mut self, port: Port) {

        debug_assert!(!self.is_in_sync);

        self.state_changes.push_back(ComponentStateChange::CreatedPort(port))

    }

    pub(crate) fn get_port_by_id(&self, id: PortIdLocal) -> Option<&Port> {

        return self.ports.iter().find(|v| v.self_id == id);

            max_index: self.inbox_len_read,

            match_port_id

        };

    }

    /// Retrieves the next unread message from the inbox `None` if there are no

    /// (new) messages to read.

    // TODO: Fix the clone of the data message, entirely unnecessary

    pub(crate) fn read_next_message(&mut self) -> Option<MessageFancy> {

        if !self.is_in_sync { return None; }

        if self.inbox_len_read == self.inbox_messages.len() { return None; }

        let message = &self.inbox_messages[self.inbox_len_read];

        }

    }

}

pub(crate) struct MessagesIter<'a> {

    messages: &'a [MessageFancy],

    next_index: usize,

    max_index: usize,

    match_port_id: PortIdLocal,

}

impl<'a> Iterator for MessagesIter<'a> {

    pub fn new() -> Self {

        ControlMessageHandler {

            id_counter: 0,

            active: Vec::new(),

        }

    }

    /// Prepares a message indicating that a channel has closed, we keep a local

    /// entry to match against the (hopefully) returned `Ack` message.

    pub fn prepare_closing_channel(

        &mut self, self_port_id: PortIdLocal, peer_port_id: PortIdLocal,

        self_connector_id: ConnectorId

        let id = self.take_id();

        self.active.push(ControlEntry{

            id,

            variant: ControlVariant::ClosedChannel(ControlClosedChannel{

                source_port: self_port_id,

                target_port: peer_port_id,

            }),

        });

        };

    }

    /// Prepares rerouting messages due to changed ownership of a port. The

    /// control message returned by this function must be sent to the

    /// transferred port's peer connector.

    pub fn prepare_reroute(

        &mut self,

        port_id: PortIdLocal, peer_port_id: PortIdLocal,

        self_connector_id: ConnectorId, peer_connector_id: ConnectorId,

        new_owner_connector_id: ConnectorId

        let id = self.take_id();

        self.active.push(ControlEntry{

            id,

            variant: ControlVariant::ChangedPort(ControlChangedPort{

                target_port: port_id,

                source_connector: peer_connector_id,

                target_connector: new_owner_connector_id,

            }),

        });

        };

    }

    /// Returns true if the supplied message should be rerouted. If so then this

    /// function returns the connector that should retrieve this message.

    pub fn should_reroute(&self, sending_connector: ConnectorId, target_port: PortIdLocal) -> Option<ConnectorId> {

        for entry in &self.active {

            if let ControlVariant::ChangedPort(entry) = &entry.variant {

                if entry.target_port == target_port {

                    // Need to reroute this message

                    return Some(entry.target_connector);

                }