// conceptually handled by particular data structures. That is to say: the code

// that we run governs: running PDL code, keeping track of ports, instantiating

// new components and transports (i.e. interacting with the runtime), running

// a consensus algorithm, etc. But on the other hand, our data is rather

// simple: we have a speculative execution tree, a set of ports that we own,

// and a bit of code that we should run.

//

// So currently the code is organized as following:

// - The scheduler that is running the component is the authoritative source on

//     ports during *non-sync* mode. The consensus algorithm is the

//     authoritative source during *sync* mode. They retrieve each other's

//     state during the transitions. Hence port data exists duplicated between

//     these two datastructures.

// - The execution tree is where executed branches reside. But the execution

//     tree is only aware of the tree shape itself (and keeps track of some

//     queues of branches that are in a particular state), and tends to store

//     the PDL program state. The consensus algorithm is also somewhat aware

//     of the execution tree, but only in terms of what is needed to complete

//     a sync round (for now, that means the port mapping in each branch).

//     Hence once more we have properties conceptually associated with branches

//     in two places.

// - TODO: Write about handling messages, consensus wrapping data

// - TODO: Write about way information is exchanged between PDL/component and scheduler through ctx

use std::sync::atomic::AtomicBool;

use crate::{PortId, ProtocolDescription};

use crate::protocol::eval::{EvalContinuation, EvalError, Prompt, Value, ValueGroup};

use super::branch::{BranchId, ExecTree, QueueKind, SpeculativeState, PreparedStatement};

use super::consensus::{Consensus, Consistency, RoundConclusion, find_ports_in_value_group};

use super::native::Connector;

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

use super::scheduler::{ComponentCtx, SchedulerCtx, MessageTicket};

pub(crate) struct ConnectorPublic {

    pub inbox: PublicInbox,

    pub sleeping: AtomicBool,

}

impl ConnectorPublic {

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

        ConnectorPublic{

            inbox: PublicInbox::new(),

            sleeping: AtomicBool::new(initialize_as_sleeping),

        }

    }

}

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

enum Mode {

    NonSync,    // running non-sync code

    Sync,       // running sync code (in potentially multiple branches)

    SyncError,  // encountered an unrecoverable error in sync mode

    Error,      // encountered an error in non-sync mode (or finished handling the sync mode error).

            }

            let cur_component = &mut self.local[component_index];

            let cur_solution = &mut cur_component.solutions[solution_index];

            match cur_solution.matches.iter_mut()

                .find(|v| v.target_id == new_component_match.target_id)

            {

                Some(other_match) => {

                    // Already have an entry

                    debug_assert_eq!(other_match.target_index, new_component_match.target_index);

                    other_match.match_indices.extend(&new_component_match.match_indices);

                },

                None => {

                    // Create a new entry

                    cur_solution.matches.push(new_component_match);

                }

            }

        }

        return self.check_for_global_solution(component_index, solution_index);

    }

    fn add_presence_and_check_for_global_failure(&mut self, component_id: ConnectorId, channels: &[LocalChannelPresence]) -> bool {

            let mut found = false;

            for existing in &mut self.presence {

                if existing.id == entry.channel_id {

                    // Same entry. We only update if we have the second

                    // component coming in it owns one end of the channel, or if

                    // a component is telling us that the channel is (now)

                    // closed.

                    if entry.is_closed {

                        existing.state = PresenceState::Closed;

                    } else if component_id != existing.owner_a && existing.state != PresenceState::Closed {

                        existing.state = PresenceState::BothPresent;

                    }

                    if existing.owner_a != component_id {

                        existing.owner_b = Some(component_id);

                    }

                    found = true;

                    break;

                }

            }

            if !found {

use std::sync::Mutex;

use std::collections::VecDeque;

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

use crate::runtime2::consensus::{ComponentPresence, SolutionCombiner};

use crate::runtime2::port::ChannelId;

use super::ConnectorId;

use super::consensus::{GlobalSolution, LocalSolution};

use super::port::PortIdLocal;

// TODO: Remove Debug derive from all types

#[derive(Debug, Copy, Clone)]

pub(crate) struct ChannelAnnotation {

    pub channel_id: ChannelId,

    pub registered_id: Option<BranchMarker>,

    pub expected_firing: Option<bool>,

}

/// Marker for a branch in a port mapping. A marker is, like a branch ID, a

/// unique identifier for a branch, but differs in that a branch only has one

/// branch ID, but might have multiple associated markers (i.e. one branch

/// performing a `put` three times will generate three markers.

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

pub(crate) struct BranchMarker{

    marker: u32,

}

impl BranchMarker {

    #[inline]

    pub(crate) fn new(marker: u32) -> Self {

    messages: Mutex<VecDeque<Message>>,

}

impl PublicInbox {

    pub fn new() -> Self {

        Self{

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

        }

    }

    pub(crate) fn insert_message(&self, message: Message) {

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

        lock.push_back(message);

    }

    pub(crate) fn take_message(&self) -> Option<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();

    }

}

mod inbox;

#[cfg(test)] mod tests;

mod connector;

// 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 connector::{ConnectorPDL, ConnectorPublic, ConnectorScheduling};

use scheduler::{Scheduler, ComponentCtx, SchedulerCtx, ControlMessageHandler};

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

use inbox::Message;

use port::{ChannelId, Port, PortState};

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

    pub generation: u32,

}

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{

            index: self.index,

            generation: self.generation,

        };

    }

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

            generation: id.generation,

        };

        let channel_id = ChannelId::new(getter_id);

        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,

            channel_id,

            kind: PortKind::Getter,

            state: PortState::Open,

            peer_connector: creating_connector,

        };

        let putter_port = Port{

            self_id: putter_id,

            peer_id: getter_id,

            channel_id,

            kind: PortKind::Putter,

            state: PortState::Open,

            peer_connector: creating_connector,

        };

        return (getter_port, putter_port);

    }

        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 mut lock = self.connectors.write().unwrap();

        self.decrement_active_components();

    }

    // --- Managing exit condition

    #[inline]

    pub(crate) fn increment_active_interfaces(&self) {

        let _old_num = self.active_interfaces.fetch_add(1, Ordering::SeqCst);

        debug_assert_ne!(_old_num, 0); // once it hits 0, it stays zero

    }

    pub(crate) fn decrement_active_interfaces(&self) {

        let old_num = self.active_interfaces.fetch_sub(1, Ordering::SeqCst);

        debug_assert!(old_num > 0);

        if old_num == 1 { // such that active interfaces is now 0

            let num_connectors = self.active_connectors.load(Ordering::Acquire);

            if num_connectors == 0 {

                self.signal_for_shutdown();

            }

        }

    }

    #[inline]

    fn increment_active_components(&self) {

        debug_assert_eq!(self.active_connectors.load(Ordering::Acquire), 0);

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

        let should_signal = self.should_exit

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

            .is_ok();

        if should_signal {

            self.scheduler_notifier.notify_all();

        }

    }

}

// TODO: Come back to this at some point

unsafe impl Send for RuntimeInner {}

unsafe impl Sync for RuntimeInner {}

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

// ConnectorStore

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

struct StoreEntry {

    connector: ScheduledConnector,

    generation: std::sync::atomic::AtomicU32,

}

struct ConnectorStore {

    // Freelist storage of connectors. Storage should be pointer-stable as

    // someone might be mutating the vector while we're executing one of the

    // connectors.

    entries: RawVec<*mut StoreEntry>,

    free: Vec<usize>,

}

impl ConnectorStore {

    fn with_capacity(capacity: usize) -> Self {

        Self {

            entries: RawVec::with_capacity(capacity),

            free: Vec::with_capacity(capacity),

        }

    }

        unsafe {

        }

    }

    /// Creates a new connector. Caller should ensure ports are set up correctly

    /// and the connector is queued for execution if needed.

    fn create(&mut self, connector: ConnectorVariant, initially_sleeping: bool) -> ConnectorKey {

        let mut connector = ScheduledConnector {

            connector,

            ctx: ComponentCtx::new_empty(),

            public: ConnectorPublic::new(initially_sleeping),

            router: ControlMessageHandler::new(),

            shutting_down: false,

        };

        let index;

        let key;

        if self.free.is_empty() {

            // No free entries, allocate new entry

            index = self.entries.len();

            key = ConnectorKey{

                index: index as u32, generation: 0

            };

            connector.ctx.id = key.downcast();

            let connector = Box::into_raw(Box::new(StoreEntry{

                connector,

                generation: AtomicU32::new(0),

            }));

            self.entries.push(connector);

        } else {

            // Free spot available

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

            unsafe {

                let target = &mut **self.entries.get_mut(index);

                std::ptr::write(&mut target.connector as *mut _, connector);

                let generation = target.generation.load(Ordering::Acquire);

                key = ConnectorKey{ index: index as u32, generation };

                target.connector.ctx.id = key.downcast();

            }

        }

        println!("DEBUG [ global store  ] Created component at {}", key.index);

        return key;

    }

    /// Destroys a connector. Caller should make sure it is not scheduled for

    /// execution. Otherwise one experiences "bad stuff" (tm).

    fn destroy(&mut self, key: ConnectorKey) {

        unsafe {

            let target = self.entries.get_mut(key.index as usize);

            (**target).generation.fetch_add(1, Ordering::SeqCst);

            std::ptr::drop_in_place(*target);

            // Note: but not deallocating!

        }

        println!("DEBUG [ global store  ] Destroyed component at {}", key.index);

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

    }

use std::collections::VecDeque;

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

use crate::protocol::ComponentCreationError;

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

use crate::runtime2::consensus::RoundConclusion;

use super::branch::{BranchId, FakeTree, QueueKind, SpeculativeState};

use super::scheduler::{SchedulerCtx, ComponentCtx, MessageTicket};

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

use super::consensus::{Consensus, Consistency, find_ports_in_value_group};

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

use super::inbox::{

    Message, DataMessage,

    SyncCompMessage, SyncPortMessage,

    ControlContent, ControlMessage

};

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

pub(crate) trait Connector {

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

    /// One should generally request and handle new messages from the component

    /// context. Then perform any logic the component has to do, and in the

    /// process perhaps queue up some state changes using the same context.

    fn run(&mut self, sched_ctx: SchedulerCtx, comp_ctx: &mut ComponentCtx) -> ConnectorScheduling;

}

pub(crate) struct FinishedSync {

    // In the order of the `get` calls

    success: bool,

    inbox: Vec<ValueGroup>,

    pub fn wait(&mut self) -> Result<Vec<ValueGroup>, ApplicationEndSyncError> {

        if !self.is_in_sync {

            return Err(ApplicationEndSyncError::NotInSync);

        }

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

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

        lock = condition.wait_while(lock, |v| v.is_none()).unwrap(); // wait while not done

        self.is_in_sync = false;

        let result = lock.take().unwrap();

        if result.success {

            return Ok(result.inbox);

        } else {

            return Err(ApplicationEndSyncError::Failure);

        }

    }

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

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

        self.connector_id = id;

    }

    fn wake_up_connector_with_ping(&self) {

            id: 0,

            sending_component_id: self.connector_id,

            content: ControlContent::Ping,

    }

    fn find_port_by_id(&self, port_id: PortIdLocal) -> Option<PortKind> {

        return self.owned_ports.iter()

            .find(|(_, owned_id)| *owned_id == port_id)

            .map(|(port_kind, _)| *port_kind);

    }

}

impl Drop for ApplicationInterface {

    fn drop(&mut self) {

        {

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

            lock.push_back(ApplicationJob::Shutdown);

        }

        self.wake_up_connector_with_ping();

        self.runtime.decrement_active_interfaces();

    }

}

use std::collections::VecDeque;

use std::sync::Arc;

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

use crate::protocol::eval::EvalError;

use crate::runtime2::port::ChannelId;

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

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

use super::native::Connector;

use super::branch::{BranchId};

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

use super::inbox::{

    ControlMessage, ControlContent,

    SyncControlMessage, SyncControlContent,

};

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

            }

            // 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 let ConnectorScheduling::Immediate = cur_schedule {

                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.

                    self.debug_conn(connector_id, &format!("Shutting down, {} Acks remaining", scheduled.router.num_pending_acks()));

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

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

                        // currently running connector

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

                    self.debug_conn(connector_id, &format!("Finished running (new scheduling is {:?})", new_schedule));

                    // 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.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 to {:?} [ exit ] \n --- {:?}", port.peer_connector, message));

                        }

                    }

                    // Any messages still in the public inbox should be handled

                    scheduled.ctx.inbox.clear_read_messages();

                    while let Some(ticket) = scheduled.ctx.get_next_message_ticket_even_if_not_in_sync() {

                        let message = scheduled.ctx.take_message_using_ticket(ticket);

                        self.handle_message_while_shutting_down(message, scheduled);

                    }

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

                        // All ports (if any) already closed

                        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.id;

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

            // Check if the message has to be rerouted because we have moved the

            // target port to another component.

            self.debug_conn(connector_id, &format!("Handling message\n --- {:#?}", message));

            if let Some(target_port) = message.target_port() {

                if let Some(other_component_id) = scheduled.router.should_reroute(target_port) {

                    self.debug_conn(connector_id, " ... Rerouting the message");

                    // We insert directly into the private inbox. Since we have

                    // a reroute entry the component can not yet be running.

                    if let Message::Control(_) = &message {