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

                        },