use crate::runtime2::inbox::{Inbox, OutboxMessage};

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

    inbox: HashMap<PortIdLocal, OutboxMessage>, // TODO: @temporary, remove together with fires()

/// Public fields of the connector that can be freely shared between multiple

/// threads. Note that this is not enforced by the compiler. The global store

/// allows retrieving the entire `Connector` as a mutable reference by one

/// thread, and this `ConnectorPublic` by any number of threads.

pub(crate) struct ConnectorPublic {

    pub inbox: Inbox,

}

impl ConnectorPublic {

    pub fn new() -> Self {

        ConnectorPublic{

            inbox: Inbox::new(),

        }

    }

}

// TODO: Maybe prevent false sharing by aligning `public` to next cache line.

    pub ports: ConnectorPorts,

    pub public: ConnectorPublic,

            public: ConnectorPublic::new(),

                let local_port_id = PortIdLocal{ index: port_id.0.u32_suffix };

                    let message = OutboxMessage {

    pub outbox: Vec<OutboxMessage>,

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

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

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

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

    inner: RwLock<ConnectorStoreInner>,

}

struct ConnectorStoreInner {

        return Self{

            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: u32) -> &'static ConnectorPublic {

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

        unsafe {

            let connector = lock.connectors.get(connector_id 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 Connector {

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

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

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

    /// and/or queue it.

    pub(crate) fn create(&self, connector: Connector) -> ConnectorKey {

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

        if lock.free.is_empty() {

                index = lock.connectors.len();

                lock.connectors.push(connector);

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

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

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

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

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

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

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

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

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

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

/// The registry of all ports

pub struct PortStore {

    inner: RwLock<PortStoreInner>,

struct PortStoreInner {

    ports: RawVec<Port>,

    free: Vec<usize>,

}

impl PortStore {

    fn with_capacity(capacity: usize) -> Self {

            inner: RwLock::new(PortStoreInner{

                ports: RawVec::with_capacity(capacity),

                free: Vec::with_capacity(capacity),

            }),

    pub(crate) fn get(&self, key: &ConnectorKey, port_id: PortIdLocal) -> PortRef {

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

        debug_assert!(port_id.is_valid());

        unsafe {

            let port = lock.ports.get_mut(port_id.index as usize);

            let port = &mut *port;

            debug_assert_eq!(port.owning_connector_id, key.index); // race condition (if they are not equal, which should never happen), better than nothing

            return PortRef{ lock, port };

        }

    pub(crate) fn create_channel(&self, creating_connector: Option<u32>) -> Channel {

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

        // Reserves a new port. Doesn't point it to its counterpart

        fn reserve_port(lock: &mut std::sync::RwLockWriteGuard<'_, PortStoreInner>, kind: PortKind, creating_connector: Option<u32>) -> u32 {

            let index;

            let ownership = if creating_connector.is_some() { PortOwnership::Owned } else { PortOwnership::Unowned };

            let connector_id = creating_connector.unwrap_or(0);

            if lock.free.is_empty() {

                index = lock.ports.len() as u32;

                lock.ports.push(Port{

                    self_id: PortIdLocal::new(index),

                    peer_id: PortIdLocal::new_invalid(),

                    kind,

                    ownership,

                    owning_connector: connector_id,

                    peer_connector: connector_id

                });

            } else {

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

                let port = unsafe{ &mut *lock.ports.get_mut(index as usize) };

                port.peer_id = PortIdLocal::new_invalid();

                port.kind = kind;

                port.ownership = ownership;

                port.owning_connector = connector_id;

                port.peer_connector = connector_id;

            }

            return index;

        }

        // Create the ports

        let putter_id = reserve_port(&mut lock, PortKind::Putter, creating_connector);

        let getter_id = reserve_port(&mut lock, PortKind::Getter, creating_connector);

        debug_assert_ne!(putter_id, getter_id);

        // Point them to one another

        unsafe {

            let putter_port = &mut *lock.ports.get_mut(putter_id as usize);

            let getter_port = &mut *lock.ports.get_mut(getter_id as usize);

            putter_port.peer_id = getter_port.self_id;

            getter_port.peer_id = putter_port.self_id;

        }

        return Channel{ putter_id, getter_id }

}

pub struct PortRef<'p> {

    lock: RwLockReadGuard<'p, PortStoreInner>,

    port: &'static mut Port,

}

impl<'p> std::ops::Deref for PortRef<'p> {

    type Target = Port;

    fn deref(&self) -> &Self::Target {

        return self.port;

}

impl<'p> std::ops::DerefMut for PortRef<'p> {

    fn deref_mut(&mut self) -> &mut Self::Target {

        return self.port;

    }

}

impl Drop for PortStore {

    fn drop(&mut self) {

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

        // Very lazy code

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

            if lock.free.contains(&idx) {

                continue;

            }

            unsafe {

                let port = lock.ports.get_mut(idx);

                std::ptr::drop_in_place(port);

            }

        }

}

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

pub struct GlobalStore {

    pub connector_queue: MpmcQueue<ConnectorKey>,

    pub connectors: ConnectorStore,

    pub ports: PortStore,

}

impl GlobalStore {

    pub fn new() -> Self {

        Self{

            connector_queue: MpmcQueue::with_capacity(256),

            connectors: ConnectorStore::with_capacity(256),

            ports: PortStore::with_capacity(256),

        }

use std::collections::VecDeque;

use std::sync::{RwLock, RwLockReadGuard, Mutex};

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

/// A message prepared by a connector. Waiting to be picked up by the runtime to

/// be sent to another connector.

pub struct OutboxMessage {

/// A message inserted into the inbox of a connector by the runtime.

#[derive(Clone)]

pub struct InboxMessage {

    pub sending_port: PortIdLocal,

    pub receiving_port: PortIdLocal,

    pub sender_prev_branch_id: BranchId,

    pub sender_cur_branch_id: BranchId,

    pub message: ValueGroup,

}

/// A message sent between connectors to communicate something about their

/// scheduling state.

pub enum ControlMessage {

    ChangePortPeer(u32, PortIdLocal, u32), // (control message ID, port to change, new peer connector ID)

    Ack(u32), // (control message ID)

}

/// If a connector has no more code to run, and its inbox does not contain any

/// new messages, then it may go into sleep mode.

///

// TODO: @Optimize, this is a temporary lazy implementation

    // "Normal" messages, intended for a PDL protocol. These need to stick

    // around during an entire sync-block (to handle `put`s for which the

    // corresponding `get`s have not yet been reached).

    messages: RwLock<Vec<InboxMessage>>,

    len_read: AtomicUsize,

    // System messages. These are handled by the scheduler and only need to be

    // handled once.

    system_messages: Mutex<VecDeque<ControlMessage>>,

        Self{

            messages: RwLock::new(Vec::new()),

            len_read: AtomicUsize::new(0),

            system_messages: Mutex::new(VecDeque::new()),

        }

    pub fn insert_message(&self, message: InboxMessage) {

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

        for existing in messages.iter() {

            if existing.sender_prev_branch_id == message.sender_prev_branch_id &&

                    existing.sender_cur_branch_id == message.sender_cur_branch_id &&

                    existing.receiving_port == message.receiving_port {

                // Message was already received

                return;

            }

        messages.push(message);

    /// Retrieves all previously read messages that satisfy the provided

    /// speculative conditions. Note that the inbox remains read-locked until

    /// the returned iterator is dropped. Should only be called by the

    /// inbox-reader (i.e. the thread executing a connector's PDL code).

    ///

    /// This function should only be used to check if already-received messages

    /// could be received by a newly encountered `get` call in a connector's

    /// PDL code.

    pub fn get_messages(&self, port_id: PortIdLocal, prev_branch_id: BranchId) -> InboxMessageIter {

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

        return InboxMessageIter{

            lock,

            next_index: 0,

            max_index: self.len_read.load(Ordering::Acquire),

            match_port_id: port_id,

            match_prev_branch_id: prev_branch_id,

        };

    }

    /// Retrieves the next unread message. Should only be called by the

    /// inbox-reader.

    pub fn next_message(&self) -> Option<InboxMessageRef> {

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

        let cur_index = self.len_read.load(Ordering::Acquire);

        if cur_index >= lock.len() {

            return None;

        }

        // TODO: Accept the correctness and simply make it an add, or even

        //  remove the atomic altogether.

        if let Err(_) = self.len_read.compare_exchange(cur_index, cur_index + 1, Ordering::AcqRel, Ordering::Acquire) {

            panic!("multiple readers modifying number of messages read");

        return Some(InboxMessageRef{

            lock,

            index: cur_index,

        });

    pub fn insert_control_message(&self, message: ControlMessage) {

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

        lock.push_back(message);

    }

    pub fn take_control_message(&self) -> Option<ControlMessage> {

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

        return lock.pop_front();

    }

}

/// Reference to a new message

pub struct InboxMessageRef<'i> {

    lock: RwLockReadGuard<'i, Vec<InboxMessage>>,

    index: usize,

}

impl<'i> std::ops::Deref for InboxMessageRef<'i> {

    type Target = InboxMessage;

    fn deref(&self) -> &'i Self::Target {

        return &self.lock[self.index];

/// Iterator over previously received messages in the inbox.

pub struct InboxMessageIter<'i> {

    lock: RwLockReadGuard<'i, Vec<InboxMessage>>,

    next_index: usize,

    max_index: usize,

    match_port_id: PortIdLocal,

    match_prev_branch_id: BranchId,

}

impl<'m: 'i, 'i> Iterator for InboxMessageIter<'i> {

    type Item = &'m InboxMessage;

    fn next(&'m mut self) -> Option<Self::Item> {

        // Loop until match is found or at end of messages

        while self.next_index < self.max_index {

            let cur_message = &self.lock[self.next_index];

            if cur_message.receiving_port == self.match_port_id && cur_message.sender_prev_branch_id == self.match_prev_branch_id {

                // Found a match

                break;

            }

            self.next_index += 1;

        }

        if self.next_index == self.max_index {

            return None;

        }

        let message = &self.lock[self.next_index];

        self.next_index += 1;

        return Some(message);

    }

}

mod port;

#[derive(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,

}

pub enum PortOwnership {

    Unowned, // i.e. held by a native application

    Owned,

    InTransit,

}

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

    // Once created, these values are immutable

    pub self_id: PortIdLocal,

    pub peer_id: PortIdLocal,

    pub kind: PortKind,

    // But this can be changed, but only by the connector that owns it

    pub ownership: PortOwnership,

    pub owning_connector: u32,

    pub peer_connector: u32, // 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: u32, // can put on it, so from the connector's point of view, this is an output

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

}

use super::inbox::InboxMessage;

        let mut delta_state = RunDeltaState::new();

            let connector_key = self.global.connector_queue.pop_front();

            let connector = self.global.connectors.get_mut(&connector_key);

                // TODO: Check inbox for new message

                        // There are message to send

                            let (inbox_message, target_connector_id) = {

                                // Note: retrieving a port incurs a read lock

                                let sending_port = self.global.ports.get(&connector_key, message.sending_port);

                                (

                                    InboxMessage {

                                        sending_port: sending_port.self_id,

                                        receiving_port: sending_port.peer_id,

                                        sender_prev_branch_id: message.sender_prev_branch_id,

                                        sender_cur_branch_id: message.sender_cur_branch_id,

                                        message: message.message,

                                    },

                                    sending_port.peer_connector,

                                )

                            };

                            let target_connector = self.global.connectors.get_shared(target_connector_id);

                            target_connector.inbox.insert_message(inbox_message);

                            // TODO: Check silent state. Queue connector if it was silent

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

                            // Create connector, modify all of the ports that

                            // it now owns, then queue it for execution

                            let connector_key = self.global.connectors.create(connector);

                            self.global.connector_queue.push_back(connector_key);

                        }