impl ConnectorPublic {

    pub fn new(initialize_as_sleeping: bool) -> Self {

        ConnectorPublic{

            inbox: PublicInbox::new(),

            sleeping: AtomicBool::new(initialize_as_sleeping),

        }

    }

}

#[derive(Eq, PartialEq)]

pub(crate) enum ConnectorScheduling {

    Immediate,      // Run again, immediately

    Later,          // Schedule for running, at some later point in time

    NotNow,         // Do not reschedule for running

    Exit,          // Connector has exited

}

pub(crate) struct ConnectorPDL {

    tree: ExecTree,

    consensus: Consensus,

}

struct ConnectorRunContext {}

impl RunContext for ConnectorRunContext{

    fn did_put(&mut self, port: PortId) -> bool {

        todo!()

    }

    fn get(&mut self, port: PortId) -> Option<ValueGroup> {

        todo!()

    }

    fn fires(&mut self, port: PortId) -> Option<Value> {

        todo!()

    }

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

                MessageFancy::Sync(message) => handle_new_sync_message(message, ctx),

                MessageFancy::Control(_) => unreachable!("control message in component"),

            }

        }

    }

    pub fn handle_new_data_message(&mut self, message: DataMessageFancy, ctx: &mut ComponentCtxFancy) {

        // Go through all branches that are awaiting new messages and see if

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

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

        // Check if we have any branch that needs running

        debug_assert!(self.tree.is_in_sync() && self.consensus.is_in_sync());

        let branch_id = self.tree.pop_from_queue(QueueKind::Runnable);

        if branch_id.is_none() {

            return ConnectorScheduling::NotNow;

        }

        // Retrieve the branch and run it

        let branch_id = branch_id.unwrap();

        let branch = &mut self.tree[branch_id];

        let mut run_context = ConnectorRunContext{};

        let run_result = branch.code_state.run(&mut run_context, &sched_ctx.runtime.protocol_description);

        // Handle the returned result. Note that this match statement contains

        // explicit returns in case the run result requires that the component's

        // code is ran again immediately

        match run_result {

            RunResult::BranchInconsistent => {

                // Branch became inconsistent

                branch.sync_state = SpeculativeState::Inconsistent;

            },

            RunResult::BranchMissingPortState(port_id) => {

                // Branch called `fires()` on a port that has not been used yet.

                let port_id = PortIdLocal::new(port_id.0.u32_suffix);

                // Create two forks, one that assumes the port will fire, and

                // one that assumes the port remains silent

                branch.sync_state = SpeculativeState::HaltedAtBranchPoint;

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

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

                self.consensus.notify_of_new_branch(branch_id, firing_branch_id);

                let _result = self.consensus.notify_of_speculative_mapping(firing_branch_id, port_id, true);

        }

        // If here then the run result did not require a particular action. We

        // return whether we have more active branches to run or not.

        if self.tree.queue_is_empty(QueueKind::Runnable) {

            return ConnectorScheduling::NotNow;

        } else {

            return ConnectorScheduling::Later;

        }

    }

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

        debug_assert!(!self.tree.is_in_sync() && !self.consensus.is_in_sync());

        let branch = self.tree.base_branch_mut();

        debug_assert!(branch.sync_state == SpeculativeState::RunningNonSync);

        let mut run_context = ConnectorRunContext{};

        let run_result = branch.code_state.run(&mut run_context, &sched_ctx.runtime.protocol_description);

        match run_result {

            RunResult::ComponentTerminated => {

                branch.sync_state = SpeculativeState::Finished;

                return ConnectorScheduling::Exit;

            },

            RunResult::ComponentAtSyncStart => {

                let current_ports = comp_ctx.notify_sync_start();

                let sync_branch_id = self.tree.start_sync();

                self.consensus.start_sync(current_ports);

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

                return ConnectorScheduling::Immediate;

            },

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

                // Note: we're relinquishing ownership of ports. But because

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

                comp_ctx.push_port(getter);

                return ConnectorScheduling::Immediate;

            },

            _ => unreachable!("unexpected run result '{:?}' while running in non-sync mode", run_result),

        }

    }

}

                            return Consistency::Inconsistent;

                        }

                    }

                }

            }

        }

        unreachable!("notify_of_speculative_mapping called with unowned port");

    }

    pub fn end_sync(&mut self, branch_id: BranchId, final_ports: &mut Vec<PortIdLocal>) {

        todo!("write");

    }

    // --- Handling messages

    /// Prepares a message for sending. Caller should have made sure that

    /// sending the message is consistent with the speculative state.

    pub fn handle_message_to_send(&mut self, branch_id: BranchId, source_port_id: PortIdLocal, content: &ValueGroup, ctx: &mut ComponentCtxFancy) -> (SyncHeader, DataHeader) {

        debug_assert!(self.is_in_sync());

        let branch = &mut self.branch_annotations[branch_id.index as usize];

        if cfg!(debug_assertions) {

            let port = branch.port_mapping.iter()

                .find(|v| v.port_id == source_port_id)

                .unwrap();

            debug_assert!(port.expected_firing == None || port.expected_firing == Some(true));

        }

        // Check for ports that are begin sent

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

        find_ports_in_value_group(content, &mut self.workspace_ports);

        if !self.workspace_ports.is_empty() {

            todo!("handle sending ports");

            self.workspace_ports.clear();

        }

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

    }

    pub fn handle_received_sync_header(&mut self, sync_header: &SyncHeader, ctx: &mut ComponentCtxFancy) {

        todo!("should check IDs and maybe send sync messages");

    }

    /// Checks data header and consults the stored port mapping and the

    /// execution tree to see which branches may receive the data message's

    /// contents.

    ///

    /// This function is generally called for freshly received messages that

    /// should be matched against previously halted branches.

    pub fn handle_received_data_header(&mut self, exec_tree: &ExecTree, data_header: &DataHeader, target_ids: &mut Vec<BranchId>) {

        for branch in exec_tree.iter_queue(QueueKind::AwaitingMessage) {

            if branch.awaiting_port == data_header.target_port {

                // Found a branch awaiting the message, but we need to make sure

                // the mapping is correct

                if self.branch_can_receive(branch.id, data_header) {

                    target_ids.push(branch.id);

                }

            }

        }

    }

    pub fn notify_of_received_message(&mut self, branch_id: BranchId, data_header: &DataHeader, content: &ValueGroup) {

        debug_assert!(self.branch_can_receive(branch_id, data_header));

        let branch = &mut self.branch_annotations[branch_id.index as usize];

        for mapping in &mut branch.port_mapping {

            if mapping.port_id == data_header.target_port {

                // Found the port in which the message should be inserted

                mapping.registered_id = Some(data_header.new_mapping);

                // Check for sent ports

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

                find_ports_in_value_group(content, &mut self.workspace_ports);

                if !self.workspace_ports.is_empty() {

/**

inbox.rs

Contains various types of inboxes and message types for the connectors. There

are two kinds of inboxes:

The `PublicInbox` is a simple message queue. Messages are put in by various

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,

}

#[derive(Debug, Clone)]

pub enum SyncBranchConstraint {

    SilentPort(PortIdLocal),

    BranchNumber(BranchId),

    PortMapping(PortIdLocal, BranchId),

}

#[derive(Debug, Clone)]

pub struct SyncConnectorSolution {

    pub connector_id: ConnectorId,

    pub terminating_branch_id: BranchId,

    pub execution_branch_ids: Vec<BranchId>, // no particular ordering of IDs enforced

    pub final_port_mapping: Vec<(PortIdLocal, BranchId)>

}

#[derive(Debug, Clone)]

pub struct SyncConnectorConstraints {

    pub connector_id: ConnectorId,

    pub constraints: Vec<SyncBranchConstraint>,

}

#[derive(Debug, Clone)]

pub struct SyncMessage {

    pub local_solutions: Vec<SyncConnectorSolution>,

    pub constraints: Vec<SyncConnectorConstraints>,

    pub to_visit: Vec<ConnectorId>,

}

// TODO: Shouldn't really be here, right?

impl SyncMessage {

    /// Creates a new sync message. Assumes that it is created by a connector

    /// that has just encountered a new local solution.

    pub(crate) fn new(initial_solution: SyncConnectorSolution, approximate_peers: usize) -> Self {

        let mut local_solutions = Vec::with_capacity(approximate_peers);

#[derive(Debug, Clone)]

pub struct SolutionMessage {

    pub comparison_number: u64,

    pub connector_origin: ConnectorId,

    pub local_solutions: Vec<(ConnectorId, BranchId)>,

    pub to_visit: Vec<ConnectorId>,

}

/// A control message. These might be sent by the scheduler to notify eachother

/// of asynchronous state changes.

#[derive(Debug, Clone)]

pub struct ControlMessage {

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

    pub content: ControlMessageVariant,

}

#[derive(Debug, Clone)]

pub enum ControlMessageVariant {

    ChangePortPeer(PortIdLocal, ConnectorId), // specified port has a new peer, sent to owner of said port

    CloseChannel(PortIdLocal), // close the port associated with this

    Ack, // acknowledgement of previous control message, matching occurs through control message ID.

}

/// Generic message contents.

#[derive(Debug, Clone)]

pub enum MessageContents {

    Data(DataMessage),              // data message, handled by connector

    Sync(SyncMessage),              // sync message, handled by both connector/scheduler

    RequestCommit(SolutionMessage), // solution message, requesting participants to commit

    ConfirmCommit(SolutionMessage), // solution message, confirming a solution everyone committed to

    Control(ControlMessage),        // control message, handled by scheduler

    Ping,                           // ping message, intentionally waking up a connector (used for native connectors)

}

#[derive(Debug)]

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

}

// Structure of module

mod runtime;

mod messages;

mod connector;

mod branch;

mod native;

mod port;

mod scheduler;

mod inbox;

mod consensus;

mod inbox2;

#[cfg(test)] mod tests;

mod connector2;

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

    /// access, to a "regular ID" which can be used to obtain immutable access.

    #[inline]

    pub fn downcast(&self) -> ConnectorId {

        return ConnectorId(self.index);

    }

    /// Turns the `ConnectorId` into a `ConnectorKey`, marked as unsafe as it

    /// bypasses the type-enforced `ConnectorKey`/`ConnectorId` system

    #[inline]

    pub unsafe fn from_id(id: ConnectorId) -> ConnectorKey {

        return ConnectorKey{ index: id.0 };

    }

}

/// A kind of token that allows shared access to a connector. Multiple threads

/// may hold this

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

pub struct ConnectorId(pub u32);

impl ConnectorId {

    // TODO: Like the other `new_invalid`, maybe remove

    #[inline]

    pub fn new_invalid() -> ConnectorId {

        return ConnectorId(u32::MAX);

    }

    #[inline]

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

        return self.0 != u32::MAX;

    }

}

// TODO: Change this, I hate this. But I also don't want to put `public` and

//  `router` of `ScheduledConnector` back into `Connector`. The reason I don't

//  want `Box<dyn Connector>` everywhere is because of the v-table overhead. But

//  to truly design this properly I need some benchmarks.

    /// will be returned. If there is no more work, then `None` will be

    /// returned.

    pub(crate) fn wait_for_work(&self) -> Option<ConnectorKey> {

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

        while lock.is_empty() && !self.should_exit.load(Ordering::Acquire) {

            lock = self.scheduler_notifier.wait(lock).unwrap();

        }

        return lock.pop_front();

    }

    pub(crate) fn push_work(&self, key: ConnectorKey) {

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

        lock.push_back(key);

        self.scheduler_notifier.notify_one();

    }

    // --- Creating/using ports

    /// Creates a new port pair. Note that these are stored globally like the

    /// connectors are. Ports stored by components belong to those components.

    pub(crate) fn create_channel(&self, creating_connector: ConnectorId) -> (Port, Port) {

        use port::{PortIdLocal, PortKind};

        let getter_id = self.port_counter.fetch_add(2, Ordering::SeqCst);

        let putter_id = PortIdLocal::new(getter_id + 1);

        let getter_id = PortIdLocal::new(getter_id);

        let getter_port = Port{

            self_id: getter_id,

            peer_id: putter_id,

            kind: PortKind::Getter,

            state: PortState::Open,

            peer_connector: creating_connector,

        };

        let putter_port = Port{

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

        }

    }

    // --- Creating/retrieving/destroying components

    /// Creates an initially sleeping application connector.

    fn create_interface_component(&self, component: ConnectorApplication) -> ConnectorKey {

        // Initialize as sleeping, as it will be scheduled by the programmer.

        let mut lock = self.connectors.write().unwrap();

        let key = lock.create(ConnectorVariant::Native(Box::new(component)), true);

        self.increment_active_components();

        return key;

    }

    /// Creates a new PDL component. This function just creates the component.

    /// If you create it initially awake, then you must add it to the work

    /// queue. Other aspects of correctness (i.e. setting initial ports) are

    /// relinquished to the caller!

    pub(crate) fn create_pdl_component(&self, connector: ConnectorPDL, initially_sleeping: bool) -> ConnectorKey {

        // Create as not sleeping, as we'll schedule it immediately

        let key = {

            let mut lock = self.connectors.write().unwrap();

            lock.create(ConnectorVariant::UserDefined(connector), initially_sleeping)

        };

        self.increment_active_components();

        return key;

    }

    #[inline]

    pub(crate) fn get_component_private(&self, connector_key: &ConnectorKey) -> &'static mut ScheduledConnector {

        let lock = self.connectors.read().unwrap();

        return lock.get_private(connector_key);

    }

    #[inline]

    pub(crate) fn get_component_public(&self, connector_id: ConnectorId) -> &'static ConnectorPublic {

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,

}

impl Scheduler {

    pub fn new(runtime: Arc<RuntimeInner>, scheduler_id: u32) -> Self {

        return Self{ runtime, scheduler_id };

    }

    pub fn run(&mut self) {

        // Setup global storage and workspaces that are reused for every

        // connector that we run

        'thread_loop: loop {

            // Retrieve a unit of work

            self.debug("Waiting for work");

            let connector_key = self.runtime.wait_for_work();

            if connector_key.is_none() {

                // We should exit

                self.debug(" ... No more work, quitting");

                break 'thread_loop;

            }

            // We have something to do

            let connector_key = connector_key.unwrap();

            let connector_id = connector_key.downcast();

            self.debug_conn(connector_id, &format!(" ... Got work, running {}", connector_key.index));

            let scheduled = self.runtime.get_component_private(&connector_key);

            // Keep running until we should no longer immediately schedule the

            // connector.

            let mut cur_schedule = ConnectorScheduling::Immediate;

            while cur_schedule == ConnectorScheduling::Immediate {

                self.handle_inbox_messages(scheduled);

                // Run the main behaviour of the connector, depending on its

                // current state.

                if scheduled.shutting_down {

                    // Nothing to do. But we're stil waiting for all our pending

                    // control messages to be answered.

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

                        // We're actually done, we can safely destroy the

                        // currently running connector

                        self.runtime.destroy_component(connector_key);

                        continue 'thread_loop;

                    } else {

                        cur_schedule = ConnectorScheduling::NotNow;

                    }

                } else {

                    self.debug_conn(connector_id, "Running ...");

                    let scheduler_ctx = SchedulerCtx{ runtime: &*self.runtime };

                    let new_schedule = scheduled.connector.run(scheduler_ctx, &mut scheduled.ctx_fancy);

                    self.debug_conn(connector_id, "Finished running");

                    // Handle all of the output from the current run: messages to

                    // send and connectors to instantiate.

                    self.handle_changes_in_context(scheduled);

                    cur_schedule = new_schedule;

                }

            }

            // If here then the connector does not require immediate execution.

            // So enqueue it if requested, and otherwise put it in a sleeping

            // state.

            match cur_schedule {

                ConnectorScheduling::Immediate => unreachable!(),

                ConnectorScheduling::Later => {

                    // Simply queue it again later

                    self.runtime.push_work(connector_key);

                },

                ConnectorScheduling::NotNow => {

                    // Need to sleep, note that we are the only ones which are

                    // allows to set the sleeping state to `true`, and since

                    // we're running it must currently be `false`.

                    self.try_go_to_sleep(connector_key, scheduled);

                },

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

                }

            }

        }

    }

    /// Receiving messages from the public inbox and handling them or storing

    /// them in the component's private inbox

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

        let connector_id = scheduled.ctx_fancy.id;

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