use std::ptr;

use std::sync::{Arc, Barrier, RwLock, RwLockReadGuard};

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

use crate::collections::{MpmcQueue, RawVec};

use super::connector::{ConnectorPDL, ConnectorPublic};

use super::port::{PortIdLocal, Port, PortKind, PortOwnership, Channel};

use super::inbox::PublicInbox;

use super::scheduler::Router;

use crate::ProtocolDescription;

use crate::runtime2::connector::{ConnectorScheduling, RunDeltaState};

use crate::runtime2::inbox::{DataMessage, MessageContents, SyncMessage};

use crate::runtime2::native::Connector;

use crate::runtime2::scheduler::ConnectorCtx;

/// 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(Copy, Clone)]

pub(crate) 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.

pub enum ConnectorVariant {

    UserDefined(ConnectorPDL),

    Native(Box<dyn Connector>),

}

impl Connector for ConnectorVariant {

    fn handle_message(&mut self, message: MessageContents, ctx: &ConnectorCtx, delta_state: &mut RunDeltaState) {

        match self {

            ConnectorVariant::UserDefined(c) => c.handle_message(message, ctx, delta_state),

            ConnectorVariant::Native(c) => c.handle_message(message, ctx, delta_state),

        }

    }

    fn run(&mut self, protocol_description: &ProtocolDescription, ctx: &ConnectorCtx, delta_state: &mut RunDeltaState) -> ConnectorScheduling {

        match self {

            ConnectorVariant::UserDefined(c) => c.run(protocol_description, ctx, delta_state),

            ConnectorVariant::Native(c) => c.run(protocol_description, ctx, delta_state),

        }

    }

}

pub struct ScheduledConnector {

    pub connector: ConnectorVariant, // access by connector

    pub context: ConnectorCtx, // mutable access by scheduler, immutable by connector

    pub public: ConnectorPublic, // accessible by all schedulers and connectors

    pub router: Router,

}

/// The registry containing all connectors. The idea here is that when someone

/// owns a `ConnectorKey`, then one has unique access to that connector.

/// Otherwise one has shared access.

///

/// This datastructure is built to be wrapped in a RwLock.

pub(crate) struct ConnectorStore {

    pub(crate) port_counter: Arc<AtomicU32>,

    inner: RwLock<ConnectorStoreInner>,

}

struct ConnectorStoreInner {

    connectors: RawVec<*mut ScheduledConnector>,

    free: Vec<usize>,

}

impl ConnectorStore {

    fn with_capacity(capacity: usize) -> Self {

        return Self{

            port_counter: Arc::new(AtomicU32::new(0)),

            inner: RwLock::new(ConnectorStoreInner {

                connectors: RawVec::with_capacity(capacity),

                free: Vec::with_capacity(capacity),

            }),

        };

    }

    /// Retrieves the shared members of the connector.

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

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

        unsafe {

            let connector = lock.connectors.get(connector_id.0 as usize);

            debug_assert!(!connector.is_null());

            return &*connector.public;

        }

    }

    /// Retrieves a particular connector. Only the thread that pulled the

    /// associated key out of the execution queue should (be able to) call this.

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

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

        unsafe {

            let connector = lock.connectors.get_mut(key.index as usize);

            debug_assert!(!connector.is_null());

            return *connector as &mut _;

        }

    }

    /// Create a new connector, returning the key that can be used to retrieve

        // Creation of the connector in the global store, requires a lock

        {

            let lock = self.inner.write().unwrap();

            let connector = ScheduledConnector {

                connector,

                context: ConnectorCtx::new(self.port_counter.clone()),

                public: ConnectorPublic::new(),

                router: Router::new(),

            };

            let index;

            if lock.free.is_empty() {

                let connector = Box::into_raw(Box::new(connector));

                unsafe {

                    // Cheating a bit here. Anyway, move to heap, store in list

                    index = lock.connectors.len();

                    lock.connectors.push(connector);

                }

            } else {

                index = lock.free.pop().unwrap();

                unsafe {

                    let target = lock.connectors.get_mut(index);

                    debug_assert!(!target.is_null());

                    ptr::write(*target, connector);

                }

            }

        }

        // Setting of new connector's ID

        let key = ConnectorKey{ index: index as u32 };

        let new_connector = self.get_mut(&key);

        new_connector.context.id = key.downcast();

        // Transferring ownership of ports (and crashing if there is a

        // programmer's mistake in port management)

        match &new_connector.connector {

            ConnectorVariant::UserDefined(connector) => {

                for port_id in &connector.ports.owned_ports {

                    let mut port = created_by.context.remove_port(*port_id);

                    new_connector.context.add_port(port);

                }

            },

            ConnectorVariant::Native(_) => {}, // no initial ports (yet!)

        }

        return key;

    }

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

        let lock = self.inner.write().unwrap();

        unsafe {

            let connector = lock.connectors.get_mut(key.index as usize);

            ptr::drop_in_place(*connector);

            // Note: but not deallocating!

        }

        lock.free.push(key.index as usize);

    }

}

impl Drop for ConnectorStore {

    fn drop(&mut self) {

        let lock = self.inner.write().unwrap();

        for idx in 0..lock.connectors.len() {

            unsafe {

                let memory = *lock.connectors.get_mut(idx);

                let _ = Box::from_raw(memory); // takes care of deallocation

            }

        }

    }

}

/// Global store of connectors, ports and queues that are used by the sceduler

/// threads. The global store has the appearance of a thread-safe datatype, but

/// one needs to be careful using it.

///

/// TODO: @docs

/// TODO: @Optimize, very lazy implementation of concurrent datastructures.

///     This includes the `should_exit` and `did_exit` pair!

pub struct GlobalStore {

    pub connector_queue: MpmcQueue<ConnectorKey>,

    pub connectors: ConnectorStore,

    pub should_exit: AtomicBool,    // signal threads to exit

}

impl GlobalStore {

    pub fn new() -> Self {

        Self{

            connector_queue: MpmcQueue::with_capacity(256),

            connectors: ConnectorStore::with_capacity(256),

            should_exit: AtomicBool::new(false),

        }

    }

}

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

use std::cell::Cell;

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

use crate::protocol::ComponentCreationError;

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

use crate::ProtocolDescription;

use crate::runtime2::connector::{Branch, find_ports_in_value_group};

use crate::runtime2::global_store::{ConnectorKey, GlobalStore};

use crate::runtime2::inbox::MessageContents;

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

use crate::runtime2::scheduler::ConnectorCtx;

use super::RuntimeInner;

use super::global_store::{ConnectorVariant, ConnectorId};

use super::port::{Channel, PortIdLocal};

use super::connector::{ConnectorPDL, ConnectorScheduling, RunDeltaState};

use super::inbox::{Message, DataMessage, SyncMessage};

/// Generic connector interface from the scheduler's point of view.

pub trait Connector {

    /// Handle a new message (preprocessed by the scheduler). You probably only

    /// want to handle `Data`, `Sync`, and `Solution` messages. The others are

    /// intended for the scheduler itself.

    fn handle_message(&mut self, message: MessageContents, ctx: &ConnectorCtx, delta_state: &mut RunDeltaState);

    /// Should run the connector's behaviour up until the next blocking point.

    fn run(&mut self, protocol_description: &ProtocolDescription, ctx: &ConnectorCtx, delta_state: &mut RunDeltaState) -> ConnectorScheduling;

}

type SyncDone = Arc<(Mutex<bool>, Condvar)>;

type JobQueue = Arc<Mutex<Vec<ApplicationJob>>>;

enum ApplicationJob {

    NewConnector(ConnectorPDL),

}

/// The connector which an application can directly interface with. Once may set

/// up the next synchronous round, and retrieve the data afterwards.

pub struct ConnectorApplication {

    sync_done: SyncDone,

    job_queue: JobQueue,

}

impl ConnectorApplication {

    pub(crate) fn new(runtime: Arc<RuntimeInner>) -> (Self, ApplicationInterface) {

        let sync_done = Arc::new(( Mutex::new(false), Condvar::new() ));

        let job_queue = Arc::new(Mutex::new(Vec::with_capacity(32)));

        let connector = ConnectorApplication { sync_done: sync_done.clone(), job_queue: job_queue.clone() };

        let interface = ApplicationInterface::new(sync_done, job_queue, runtime);

        return (connector, interface);

    }

}

impl Connector for ConnectorApplication {

    fn handle_message(&mut self, message: MessageContents, ctx: &ConnectorCtx, delta_state: &mut RunDeltaState) {

        todo!("handling messages in ConnectorApplication (API for runtime)")

    }

    fn run(&mut self, protocol_description: &ProtocolDescription, ctx: &ConnectorCtx, delta_state: &mut RunDeltaState) -> ConnectorScheduling {

        let mut queue = self.job_queue.lock().unwrap();

        while let Some(job) = queue.pop() {

            match job {

                ApplicationJob::NewConnector(connector) => {

                    delta_state.new_connectors.push(connector);

                }

            }

        }

        return ConnectorScheduling::NotNow;

    }

}

/// The interface to a `ApplicationConnector`. This allows setting up the

/// interactions the `ApplicationConnector` performs within a synchronous round.

pub struct ApplicationInterface {

    sync_done: SyncDone,

    job_queue: JobQueue,

    runtime: Arc<RuntimeInner>,

    connector_id: ConnectorId,

}

impl ApplicationInterface {

    pub(crate) fn new(sync_done: SyncDone, job_queue: JobQueue, runtime: Arc<RuntimeInner1>) -> Self {

        return Self{

            sync_done, job_queue, runtime,

            connector_id: ConnectorId::new_invalid(),

            owned_ports: Vec::new(),

        }

    }

    /// Creates a new channel.

    pub fn create_channel(&mut self) -> Channel {

        // TODO: Duplicated logic in scheduler

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

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

        let getter_id = PortIdLocal::new(getter_id);

            self_id: getter_id,

            peer_id: putter_id,

            kind: PortKind::Getter,

            peer_connector: self.connector_id,

        });

    }

    /// Creates a new connector. Note that it is not scheduled immediately, but

    /// depends on the `ApplicationConnector` to run, followed by the created

    /// connector being scheduled.

    // TODO: Optimize by yanking out scheduler logic for common use.

    pub fn create_connector(&mut self, module: &str, routine: &str, arguments: ValueGroup) -> Result<(), ComponentCreationError> {

        // Retrieve ports and make sure that we own the ones that are currently

        // specified. This is also checked by the scheduler, but that is done

        // asynchronously.

        let mut initial_ports = Vec::new();

        find_ports_in_value_group(&arguments, &mut initial_ports);

        for port_to_remove in &initial_ports {

            match self.owned_ports.iter().position(|v| v == port_to_remove) {

                Some(index_to_remove) => {

                    // We own the port, so continue

                    self.owned_ports.remove(index_to_remove)

                },

                None => {

                    // We don't own the port

                    return Err(ComponentCreationError::UnownedPort);

                }

            }

        }

        let state = self.runtime.protocol_description.new_component_v2(module.as_bytes(), routine.as_bytes(), arguments)?;

        let connector = ConnectorPDL::new(Branch::new_initial_branch(state), initial_ports);

        // Put on job queue

        {

            let mut queue = self.job_queue.lock().unwrap();

            queue.push(ApplicationJob::NewConnector(connector));

        }

        // Send ping message to wake up connector

        let connector = self.runtime.global_store.connectors.get_shared(self.connector_id);

        connector.inbox.insert_message(Message::Ping);

        let should_wake_up = connector.sleeping

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

            .is_ok();

        if should_wake_up {

            let key = unsafe{ ConnectorKey::from_id(self.connector_id) };

            self.runtime.global_store.connector_queue.push_back(key);

        }

        return Ok(());

    }

    /// Check if the next sync-round is finished.

    pub fn try_wait(&self) -> bool {

        let (is_done, _) = &*self.sync_done;

        let lock = is_done.lock().unwrap();

        return *lock;

    }

    /// Wait until the next sync-round is finished

    pub fn wait(&self) {

        let (is_done, condition) = &*self.sync_done;

        let lock = is_done.lock().unwrap();

        condition.wait_while(lock, |v| !*v); // wait while not done

    }

    /// Called by runtime to set associated connector's ID.

    pub(crate) fn set_connector_id(&mut self, id: ConnectorId) {

        self.connector_id = id;

    }

}

use super::global_store::ConnectorId;

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

pub(crate) struct PortIdLocal {

    pub index: u32,

}

impl PortIdLocal {

    pub fn new(id: u32) -> Self {

        Self{ index: id }

    }

    // TODO: Unsure about this, maybe remove, then also remove all struct

    //  instances where I call this

    pub fn new_invalid() -> Self {

        Self{ index: u32::MAX }

    }

    pub fn is_valid(&self) -> bool {

        return self.index != u32::MAX;

    }

}

pub enum PortKind {

    Putter,

    Getter,

}

/// Represents a port inside of the runtime. May be without owner if it is

/// created by the application interfacing with the runtime, instead of being

/// created by a connector.

pub struct Port {

    pub self_id: PortIdLocal,

    pub peer_id: PortIdLocal,

    pub kind: PortKind,

    pub peer_connector: ConnectorId, // might be temporarily inconsistent while peer port is sent around in non-sync phase.

}

// TODO: Turn port ID into its own type

pub struct Channel {

    pub putter_id: PortIdLocal, // can put on it, so from the connector's point of view, this is an output

    pub getter_id: PortIdLocal, // vice versa: can get on it, so an input for the connector

}

use std::sync::Arc;

use std::sync::Condvar;

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

use std::time::Duration;

use std::thread;

use crate::ProtocolDescription;

use crate::runtime2::global_store::ConnectorVariant;

use crate::runtime2::inbox::MessageContents;

use crate::runtime2::native::Connector;

use crate::runtime2::port::{Channel, PortKind, PortOwnership};

use super::RuntimeInner;

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

use super::inbox::{Message, DataMessage, ControlMessage, ControlMessageVariant};

use super::connector::{ConnectorPDL, ConnectorPublic, ConnectorScheduling, RunDeltaState};

use super::global_store::{ConnectorKey, ConnectorId, GlobalStore};

/// Contains fields that are mostly managed by the scheduler, but may be

/// accessed by the connector

pub(crate) struct ConnectorCtx {

    pub(crate) id: ConnectorId,

    port_counter: Arc<AtomicU32>,

    pub(crate) ports: Vec<Port>,

}

impl ConnectorCtx {

    pub(crate) fn new(port_counter: Arc<AtomicU32>) -> ConnectorCtx {

        Self{

            id: ConnectorId::new_invalid(),

            port_counter,

        }

    }

    /// Creates a (putter, getter) port pair belonging to the same channel. The

    /// port will be implicitly owned by the connector.

    pub(crate) fn create_channel(&mut self) -> Channel {

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

        self.ports.push(Port{

            self_id: getter_id,

            peer_id: putter_id,

            kind: PortKind::Getter,

            peer_connector: self.id,

        });

        self.ports.push(Port{

            self_id: putter_id,

            peer_id: getter_id,

            kind: PortKind::Putter,

            peer_connector: self.id,

        });

        return Channel{ getter_id, putter_id };

    }

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

        debug_assert!(!self.ports.iter().any(|v| v.self_id == port.self_id));

        self.ports.push(port);

    }

    pub(crate) fn remove_port(&mut self, id: PortIdLocal) -> Port {

        let index = self.port_id_to_index(id);

        return self.ports.remove(index);

    }

    pub(crate) fn get_port(&self, id: PortIdLocal) -> &Port {

        let index = self.port_id_to_index(id);

        return &self.ports[index];

    }

    pub(crate) fn get_port_mut(&mut self, id: PortIdLocal) -> &mut Port {

        let index = self.port_id_to_index(id);

        return &mut self.ports[index];

    }

    fn port_id_to_index(&self, id: PortIdLocal) -> usize {

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

            if port.self_id == id {

                return idx;

            }

        }

        panic!("port {:?}, not owned by connector", id);

    }

}

pub(crate) struct Scheduler {

    runtime: Arc<RuntimeInner>,

}

// Thinking aloud: actual ports should be accessible by connector, but managed

// by the scheduler (to handle rerouting messages). We could just give a read-

// only context, instead of an extra call on the "Connector" trait.

impl Scheduler {

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

        return Self{ runtime };

    }

    pub fn run(&mut self) {

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

        // connector that we run

        let mut delta_state = RunDeltaState::new();

        'thread_loop: loop {

            // Retrieve a unit of work

            let connector_key = self.runtime.global_store.connector_queue.pop_front();

            if connector_key.is_none() {

                // TODO: @Performance, needs condition or something, and most

                //  def' not sleeping

                thread::sleep(Duration::new(1, 0));

                if self.runtime.global_store.should_exit.load(Ordering::Acquire) {

                    // Thread exits!

                    break 'thread_loop;

                }

                continue 'thread_loop;

            }

            // We have something to do

            let connector_key = connector_key.unwrap();

            let scheduled = self.runtime.global_store.connectors.get_mut(&connector_key);

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

            // connector.

            let mut cur_schedule = ConnectorScheduling::Immediate;

            while cur_schedule == ConnectorScheduling::Immediate {

                // Check all the message that are in the shared inbox

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

                    match message.contents {

                        MessageContents::Data(content) => {

                            // Check if we need to reroute, or can just put it

                            // in the private inbox of the connector

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

                                self.send_message_and_wake_up_if_sleeping(other_connector_id, Message::Data(content));

                            } else {

                                scheduled.connector.insert_data_message(content);

                            }

                        }

                        MessageContents::Sync(content) => {

                            scheduled.connector.insert_sync_message(content, &scheduled.context, &mut delta_state);

                        }

                        MessageContents::Solution(content) => {

                            // TODO: Handle solution message

                        },

                        MessageContents::Control(content) => {

                            match content.content {

                                ControlMessageVariant::ChangePortPeer(port_id, new_target_connector_id) => {

                                    // Need to change port target

                                    let port = scheduled.context.get_port_mut(port_id);

                                    port.peer_connector = new_target_connector_id;

                                    debug_assert!(delta_state.outbox.is_empty());

                                    // And respond with an Ack

                                    // Note: after this code has been reached, we may not have any

                                    // messages in the outbox that send to the port whose owning

                                    // connector we just changed. This is because the `ack` will

                                    // clear the rerouting entry of the `ack`-receiver.

                                    self.send_message_and_wake_up_if_sleeping(

                                        content.sender,

                                        Message{

                                            sending_connector: connector_key.downcast(),

                                            receiving_port: PortIdLocal::new_invalid(),

                                            contents: MessageContents::Control(ControlMessage{

                                                id: content.id,

                                                content: ControlMessageVariant::Ack,

                                            }),

                                        }

                                    );

                                },

                                ControlMessageVariant::Ack => {

                                    scheduled.router.handle_ack(content.id);

                                }

                            }

                        }

                        Message::Ping => {},

                    }

                }

                // Actually run the connector

                let new_schedule = scheduled.connector.run(

                    &self.runtime.protocol_description, &scheduled.context, &mut delta_state

                );

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

                // send and connectors to instantiate.

                self.handle_delta_state(&connector_key, &mut scheduled.context, &mut delta_state);

                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.global_store.connector_queue.push_back(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`.

                    debug_assert_eq!(scheduled.public.sleeping.load(Ordering::Acquire), false);

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

                    // We might have received a message in the meantime from a

                    // thread that did not see the sleeping flag set to `true`,

                    // so:

                    if !scheduled.public.inbox.is_empty() {

                        let should_reschedule_self = scheduled.public.sleeping

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

                            .is_ok();

                        if should_reschedule_self {

                            self.runtime.global_store.connector_queue.push_back(connector_key);

                        }

                    }

                }

            }

        }

    }

    fn handle_delta_state(&mut self, connector_key: &ConnectorKey, context: &mut ConnectorCtx, delta_state: &mut RunDeltaState) {

        // Handling any messages that were sent

        let connector_id = connector_key.downcast();

        if !delta_state.outbox.is_empty() {

            for mut message in delta_state.outbox.drain(..) {

                // Based on the message contents, decide where the message

                // should be sent to. This might end up modifying the message.

                let (peer_connector, peer_port) = match &mut message {

                    MessageContents::Data(contents) => {

                        let port = context.get_port(contents.sending_port);

                        (port.peer_connector, port.peer_id)

                    },

                    MessageContents::Sync(contents) => {

                        let connector = contents.to_visit.pop().unwrap();

                        (connector, PortIdLocal::new_invalid())

                    },

                    MessageContents::RequestCommit(contents)=> {

                        let connector = contents.to_visit.pop().unwrap();

                        (connector, PortIdLocal::new_invalid())

                    },

                    MessageContents::ConfirmCommit(contents) => {

                        for to_visit in &contents.to_visit {

                            let message = Message{

                                sending_connector: connector_id,

                                receiving_port: PortIdLocal::new_invalid(),

                                contents: contents.clone(),

                            };

                            self.send_message_and_wake_up_if_sleeping(*to_visit, message);

                        }

                        (ConnectorId::new_invalid(), PortIdLocal::new_invalid())

                    },

                    MessageContents::Control(_) | MessageContents::Ping => {

                        // Never generated by the user's code

                        unreachable!();

                    }

                };

                // TODO: Maybe clean this up, perhaps special case for

                //  ConfirmCommit can be handled differently.

                if peer_connector.is_valid() {

                    let message = Message {

                        sending_connector: connector_id,

                        receiving_port: peer_port,

                        contents: message,

                    };

                    self.send_message_and_wake_up_if_sleeping(peer_connector, message);

                }

            }

        }

        // Handling any new connectors that were scheduled

        // TODO: Pool outgoing messages to reduce atomic access

        if !delta_state.new_connectors.is_empty() {

            let cur_connector = self.runtime.global_store.connectors.get_mut(connector_key);

            for new_connector in delta_state.new_connectors.drain(..) {

                // Add to global registry to obtain key

                let new_key = self.runtime.global_store.connectors.create(cur_connector, ConnectorVariant::UserDefined(new_connector));

                let new_connector = self.runtime.global_store.connectors.get_mut(&new_key);

                // Call above changed ownership of ports, but we still have to

                // let the other end of the channel know that the port has

                // changed location.

                for port in &new_connector.context.ports {

                    let reroute_message = cur_connector.router.prepare_reroute(

                        port.self_id, port.peer_id, cur_connector.context.id,

                        port.peer_connector, new_connector.context.id

                    );

                    self.send_message_and_wake_up_if_sleeping(peer_connector_id, reroute_message);

                }

                // Schedule new connector to run

                self.runtime.global_store.connector_queue.push_back(new_key);

            }

        }

    }

    pub fn send_message_and_wake_up_if_sleeping(&self, connector_id: ConnectorId, message: Message) {

        let connector = self.runtime.global_store.connectors.get_shared(connector_id);

        connector.inbox.insert_message(message);

        let should_wake_up = connector.sleeping

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

            .is_ok();

        if should_wake_up {

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

            self.runtime.global_store.connector_queue.push_back(key);

        }

    }

}

/// Represents a rerouting entry due to a moved port

// TODO: Optimize

struct ReroutedTraffic {

    id: u32,                        // ID of control message

    port: PortIdLocal,              // targeted port

    source_connector: ConnectorId,  // connector we expect messages from

    target_connector: ConnectorId,  // connector they should be rerouted to

}

pub(crate) struct Router {

    id_counter: u32,

    active: Vec<ReroutedTraffic>,

}

impl Router {

    pub fn new() -> Self {

        Router{

            id_counter: 0,

            active: Vec::new(),

        }

    }

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

    ) -> Message {

        let id = self.id_counter;

        self.id_counter.overflowing_add(1);

        self.active.push(ReroutedTraffic{

            id,

            port: port_id,

            source_connector: peer_connector_id,

            target_connector: new_owner_connector_id,

        });

        return Message::Control(ControlMessage{

            id,

            content: ControlMessageVariant::ChangePortPeer(peer_port_id, 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, sending_port: PortIdLocal) -> Option<ConnectorId> {

        for reroute in &self.active {

            if reroute.source_connector == sending_connector &&

                reroute.port == sending_port {

                // Need to reroute this message

                return Some(reroute.target_connector);

            }

        }

        return None;