use std::collections::HashMap;

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

use crate::{PortId, ProtocolDescription};

use crate::protocol::{ComponentState, RunContext, RunResult};

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

use super::ConnectorId;

use super::native::Connector;

use super::inbox::{

    PrivateInbox, PublicInbox,

    DataMessage, SyncMessage, SolutionMessage, Message, MessageContents,

    SyncBranchConstraint, SyncConnectorSolution

};

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

/// Represents the identifier of a branch (the index within its container). An

/// ID of `0` generally means "no branch" (e.g. no parent, or a port did not

/// yet receive anything from any branch).

// TODO: Remove Debug derive

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

pub struct BranchId {

    pub index: u32,

}

impl BranchId {

    fn new_invalid() -> Self {

        Self{ index: 0 }

    }

    fn new(index: u32) -> Self {

        debug_assert!(index != 0);

        Self{ index }

            inbox: PublicInbox::new(),

            sleeping: AtomicBool::new(initialize_as_sleeping),

        }

    }

}

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

// TODO: Do this outside of the connector, create a wrapping struct

pub(crate) struct ConnectorPDL {

    // State and properties of connector itself

    in_sync: bool,

    // Branch management

    branches: Vec<Branch>, // first branch is always non-speculative one

    sync_active: BranchQueue,

    sync_pending_get: BranchQueue,

    sync_finished: BranchQueue,

    sync_finished_last_handled: u32, // TODO: Change to BranchId?

    cur_round: u32,

    // Port/message management

    pub committed_to: Option<(ConnectorId, u64)>,

    pub inbox: PrivateInbox,

    pub ports: ConnectorPorts,

}

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

    }

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

    }

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

    }

    fn get_channel(&mut self) -> Option<(Value, Value)> {

    }

}

impl Connector for ConnectorPDL {

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

        use MessageContents as MC;

        match message.contents {

            MC::Data(content) => self.handle_data_message(message.receiving_port, content),

            MC::Sync(content) => self.handle_sync_message(content, ctx, delta_state),

            MC::RequestCommit(content) => self.handle_request_commit_message(content, ctx, delta_state),

            MC::ConfirmCommit(content) => self.handle_confirm_commit_message(content, ctx, delta_state),

            MC::Control(_) | MC::Ping => {},

        }

    }

        if self.in_sync {

            // When in speculative mode we might have generated new sync

            // solutions, we need to turn them into proposed solutions here.

            if self.sync_finished_last_handled != self.sync_finished.last {

                // Retrieve first element in queue

                let mut next_id;

                if self.sync_finished_last_handled == 0 {

                    next_id = self.sync_finished.first;

                } else {

                    let last_handled = &self.branches[self.sync_finished_last_handled as usize];

                    debug_assert!(last_handled.next_branch_in_queue.is_some()); // because "last handled" != "last in queue"

                    next_id = last_handled.next_branch_in_queue.unwrap();

                }

                loop {

                    let branch_id = BranchId::new(next_id);

                    let branch = &self.branches[next_id as usize];

                    let branch_next = branch.next_branch_in_queue;

                    // Turn local solution into a message and send it along

                    // TODO: Like `ports` access, also revise the construction of this `key`, should not be needed

                    if let Some(valid_solution) = solution_message {

                    } else {

                        // Branch is actually invalid, but we only just figured

                        // it out. We need to mark it as invalid to prevent

                        // future use

                        Self::remove_branch_from_queue(&mut self.branches, &mut self.sync_finished, branch_id);

                        if branch_id.index == self.sync_finished_last_handled {

                            self.sync_finished_last_handled = self.sync_finished.last;

                        }

                        let branch = &mut self.branches[next_id as usize];

                        branch.sync_state = SpeculativeState::Inconsistent;

                    }

                    match branch_next {

                        Some(id) => next_id = id,

                        None => break,

                    }

                }

                self.sync_finished_last_handled = next_id;

            }

            return scheduling;

        } else {

            return scheduling;

        }

    }

}

impl ConnectorPDL {

    /// Constructs a representation of a connector. The assumption is that the

    /// initial branch is at the first instruction of the connector's code,

    /// hence is in a non-sync state.

    pub fn new(initial_branch: Branch, owned_ports: Vec<PortIdLocal>) -> Self {

        Self{

            in_sync: false,

            branches: vec![initial_branch],

            sync_active: BranchQueue::new(),

            sync_pending_get: BranchQueue::new(),

            sync_finished: BranchQueue::new(),

            sync_finished_last_handled: 0, // none at all

            cur_round: 0,

            committed_to: None,

            inbox: PrivateInbox::new(),

            ports: ConnectorPorts::new(owned_ports),

        }

    }

                self.ports.add_port(port_delta.port_id);

            } else {

                self.ports.remove_port(port_delta.port_id);

            }

        }

        solution.commit_to_sync();

    }

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

    // Executing connector code

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

    /// Runs the connector in synchronous mode. Potential changes to the global

    /// system's state are added to the `RunDeltaState` object by the connector,

    /// where it is the caller's responsibility to immediately take care of

    /// those changes. The return value indicates when (and if) the connector

    /// needs to be scheduled again.

    pub fn run_in_speculative_mode(&mut self, pd: &ProtocolDescription, _context: &ConnectorCtx, results: &mut RunDeltaState) -> ConnectorScheduling {

        debug_assert!(self.in_sync);

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

        let branch = Self::pop_branch_from_queue(&mut self.branches, &mut self.sync_active);

        // Run the branch to the next blocking point

        let run_result = branch.code_state.run(&mut run_context, pd);

        // Match statement contains `return` statements only if the particular

        // run result behind handled requires an immediate re-run of the

        // connector.

        match run_result {

            RunResult::BranchInconsistent => {

                // Speculative branch became inconsistent

                branch.sync_state = SpeculativeState::Inconsistent;

            },

            RunResult::BranchMissingPortState(port_id) => {

                // Branch called `fires()` on a port that does not yet have an

                // assigned speculative value. So we need to create those

                // branches

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

                let local_port_index = self.ports.get_port_index(local_port_id).unwrap();

                debug_assert!(self.ports.owned_ports.contains(&local_port_id));

                // Create two copied branches, one silent and one firing

                branch.sync_state = SpeculativeState::HaltedAtBranchPoint;

                let parent_branch_id = branch.index;

                let parent_branch = &self.branches[parent_branch_id.index as usize];

                    debug_assert!(results.ports.is_empty());

                    find_ports_in_value_group(&message.message, &mut results.ports);

                    Self::release_ports_during_sync(&mut self.ports, branch, &results.ports).unwrap();

                    results.ports.clear();

                    results.outbox.push(MessageContents::Data(message));

                    return ConnectorScheduling::Immediate

                } else {

                    branch.sync_state = SpeculativeState::Inconsistent;

                }

            },

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

        }

        // Not immediately scheduling, so schedule again if there are more

        // branches to run

        if self.sync_active.is_empty() {

            return ConnectorScheduling::NotNow;

        } else {

            return ConnectorScheduling::Later;

        }

    }

    /// Runs the connector in non-synchronous mode.

        debug_assert!(!self.in_sync);

        debug_assert!(self.sync_active.is_empty() && self.sync_pending_get.is_empty() && self.sync_finished.is_empty());

        debug_assert!(self.branches.len() == 1);

        let branch = &mut self.branches[0];

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

        let run_result = branch.code_state.run(&mut run_context, pd);

        match run_result {

            RunResult::ComponentTerminated => {

                // Need to wait until all children are terminated

                // TODO: Think about how to do this?

                branch.sync_state = SpeculativeState::Finished;

                return ConnectorScheduling::Exit;

            },

            RunResult::ComponentAtSyncStart => {

                // Prepare for sync execution and reschedule immediately

                self.in_sync = true;

                let first_sync_branch = Branch::new_sync_branching_from(1, branch);

                let first_sync_branch_id = first_sync_branch.index;

                self.branches.push(first_sync_branch);

                Self::push_branch_into_queue(&mut self.branches, &mut self.sync_active, first_sync_branch_id);

                return ConnectorScheduling::Later;

            },

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

                // Construction of a new component. Find all references to ports

                // inside of the arguments

                debug_assert!(results.ports.is_empty());

                find_ports_in_value_group(&arguments, &mut results.ports);

    /// (prev_branch_id, cur_branch_id, receiving_port_id) already exists, then

    /// nothing is inserted..

    pub(crate) fn insert_message(&mut self, target_port: PortIdLocal, message: DataMessage) {

        for (existing_target_port, existing) in self.messages.iter() {

            if existing.sender_prev_branch_id == message.sender_prev_branch_id &&

                    existing.sender_cur_branch_id == message.sender_cur_branch_id &&

                    *existing_target_port == target_port {

                // Message was already received

                return;

            }

        }

        self.messages.push((target_port, 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(crate) fn get_messages(&self, port_id: PortIdLocal, prev_branch_id: BranchId) -> InboxMessageIter {

            messages: &self.messages,

            next_index: 0,

            max_index: self.len_read,

            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(crate) fn next_message(&mut self) -> Option<&DataMessage> {

        if self.len_read == self.messages.len() {

            return None;

        }

        let (_, to_return) = &self.messages[self.len_read];

        self.len_read += 1;

        return Some(to_return);

    }

    /// Simply empties the inbox

    pub(crate) fn clear(&mut self) {

        self.messages.clear();

        self.len_read = 0;

mod runtime;

mod messages;

mod connector;

mod native;

mod port;

mod scheduler;

mod inbox;

#[cfg(test)] mod tests;

// Imports

use std::collections::VecDeque;

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

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

use std::thread::{self, JoinHandle};

use crate::collections::RawVec;

use crate::ProtocolDescription;

use inbox::Message;

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

use scheduler::{Scheduler, ConnectorCtx, Router};

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

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

    }

}

    should_exit: AtomicBool,

}

impl RuntimeInner {

    // --- Managing the components queued for execution

    /// Wait until there is a connector to run. If there is one, then `Some`

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

    /// returned.

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

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

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

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

        }

        return lock.pop_front();

    }

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

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

        lock.push_back(key);

        self.scheduler_notifier.notify_one();

    }

    // --- Creating/retrieving/destroying components

    pub(crate) 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. The caller MUST make sure to schedule the

    /// connector.

    // TODO: Nicer code, not forcing the caller to schedule, perhaps?

    pub(crate) fn create_pdl_component(&self, created_by: &mut ScheduledConnector, connector: ConnectorPDL) -> 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), true)

        };

        // Transfer the ports

        {

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

        return key;

    }

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

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

        return lock.get_private(connector_key);

    }

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

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

        return lock.get_public(connector_id);

    }

    pub(crate) fn destroy_component(&self, connector_key: ConnectorKey) {

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

        lock.destroy(connector_key);

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

    }

    fn decrement_active_components(&self) {

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

        debug_assert!(old_num > 0);

        if old_num == 0 { // such that we have no more active connectors (for now!)

            let num_interfaces = self.active_interfaces.load(Ordering::Acquire);

            if num_interfaces == 0 {

                self.signal_for_shutdown();

            }

        }

    }

    #[inline]

    fn signal_for_shutdown(&self) {

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

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

        println!("DEBUG: Signaling for shutdown");

        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 {

            println!("DEBUG: Notifying all waiting schedulers");

            self.scheduler_notifier.notify_all();

        }

use std::collections::VecDeque;

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

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

use crate::protocol::ComponentCreationError;

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

use crate::ProtocolDescription;

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

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

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

use super::connector::find_ports_in_value_group;

use super::inbox::{Message, MessageContents};

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

pub(crate) 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: Message, ctx: &ConnectorCtx, delta_state: &mut RunDeltaState);

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

}

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

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

enum ApplicationJob {

    NewChannel((Port, Port)),

    NewConnector(ConnectorPDL),

    Shutdown,

}

/// 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(VecDeque::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: Message, _ctx: &ConnectorCtx, _delta_state: &mut RunDeltaState) {

        use MessageContents as MC;

        match message.contents {

            MC::Data(_) => unreachable!("data message in API connector"),

            MC::Sync(_) | MC::RequestCommit(_) | MC::ConfirmCommit(_) => {

                // Handling sync in API

            },

            MC::Control(_) => {},

            MC::Ping => {},

        }

    }

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

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

            match job {

                ApplicationJob::NewChannel((endpoint_a, endpoint_b)) => {

                    println!("DEBUG: API adopting ports");

                    delta_state.new_ports.reserve(2);

                    delta_state.new_ports.push(endpoint_a);

                    delta_state.new_ports.push(endpoint_b);

                }

                ApplicationJob::NewConnector(connector) => {

                    println!("DEBUG: API creating connector");

                    delta_state.new_connectors.push(connector);

                },

                ApplicationJob::Shutdown => {

                    debug_assert!(queue.is_empty());

                    return ConnectorScheduling::Exit;

                }

            }

        }

        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,

    owned_ports: Vec<PortIdLocal>,

}

impl ApplicationInterface {

    fn new(sync_done: SyncDone, job_queue: JobQueue, runtime: Arc<RuntimeInner>) -> 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 {

        {

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

            lock.push_back(ApplicationJob::NewChannel((getter_port, putter_port)));

        }

        // Add to owned ports for error checking while creating a connector

        self.owned_ports.reserve(2);

        self.owned_ports.push(putter_id);

        self.owned_ports.push(getter_id);

        return Channel{ putter_id, getter_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);

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

    scheduler_id: u32,

}

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

        let scheduler_id = self.scheduler_id;

        let mut delta_state = RunDeltaState::new();

        'thread_loop: loop {

            // Retrieve a unit of work

            println!("DEBUG [{}]: Waiting for work", scheduler_id);

            let connector_key = self.runtime.wait_for_work();