use std::collections::HashMap;

use std::ops::{Index, IndexMut};

use crate::protocol::ComponentState;

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

/// Generic branch ID. A component will always have one branch: the

/// non-speculative branch. This branch has ID 0. Hence in a speculative context

/// we use this fact to let branch ID 0 denote the ID being invalid.

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

pub struct BranchId {

    pub index: u32

}

impl BranchId {

    #[inline]

    pub(crate) fn new_invalid() -> Self {

            sync_state: SpeculativeState::RunningNonSync,

            awaiting_port: PortIdLocal::new_invalid(),

            next_in_queue: BranchId::new_invalid(),

            inbox: HashMap::new(),

            prepared_channel: None,

        }

    }

    /// Constructs a sync branch. The provided branch is assumed to be the

    /// parent of the new branch within the execution tree.

    fn new_sync(new_index: u32, parent_branch: &Branch) -> Self {

        debug_assert!(

        debug_assert!(parent_branch.prepared_channel.is_none());

        Branch {

            id: BranchId::new(new_index),

            code_state: parent_branch.code_state.clone(),

            sync_state: SpeculativeState::RunningInSync,

            awaiting_port: parent_branch.awaiting_port,

            next_in_queue: BranchId::new_invalid(),

            inbox: parent_branch.inbox.clone(),

            prepared_channel: None,

        }

    }

    /// Inserts a message into the branch for retrieval by a corresponding

    /// `get(port)` call.

    pub(crate) fn insert_message(&mut self, target_port: PortIdLocal, contents: ValueGroup) {

        }

    }

    #[inline]

    fn is_empty(&self) -> bool {

        debug_assert!(self.first.is_valid() == self.last.is_valid());

        return !self.first.is_valid();

    }

}

const NUM_QUEUES: usize = 3;

pub(crate) enum QueueKind {

    Runnable,

    AwaitingMessage,

    FinishedSync,

}

impl QueueKind {

    fn as_index(&self) -> usize {

        return match self {

            QueueKind::Runnable => 0,

            QueueKind::AwaitingMessage => 1,

            QueueKind::FinishedSync => 2,

/// herein to a minimum. So the execution tree is aware of the branches, their

/// execution state and the way they're dependent on each other, but the

/// execution tree should not be aware of e.g. sync algorithms.

///

/// Note that the tree keeps track of multiple lists of branches. Each list

/// contains branches that ended up in a particular execution state. The lists

/// are described by the various `BranchQueue` instances and the `next_in_queue`

/// field in each branch.

pub(crate) struct ExecTree {

    // All branches. the `parent_id` field in each branch implies the shape of

    // the tree. Branches are index stable throughout a sync round.

    pub branches: Vec<Branch>,

}

impl ExecTree {

    /// Constructs a new execution tree with a single non-sync branch.

    pub fn new(component: ComponentState) -> Self {

        return Self {

            branches: vec![Branch::new_non_sync(component)],

            queues: [BranchQueue::new(); 3]

        }

    }

    // --- Generic branch (queue) management

    /// branch. Just make sure it actually is in the provided queue.

    pub fn iter_queue(&self, kind: QueueKind, start_at: Option<BranchId>) -> BranchQueueIter {

        let queue = &self.queues[kind.as_index()];

        let index = match start_at {

            Some(branch_id) => {

                debug_assert!(self.iter_queue(kind, None).any(|v| v.id == branch_id));

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

                branch.next_in_queue.index as usize

            },

            None => {

            }

        };

        return BranchQueueIter {

            branches: self.branches.as_slice(),

            index,

        }

    }

    /// Returns an iterator that starts with the provided branch, and then

    /// continues to visit all of the branch's parents.

    pub fn iter_parents(&self, branch_id: BranchId) -> BranchParentIter {

        debug_assert!(index.is_valid());

        return &self.branches[index.index as usize];

    }

}

impl IndexMut<BranchId> for ExecTree {

    fn index_mut(&mut self, index: BranchId) -> &mut Self::Output {

        debug_assert!(index.is_valid());

        return &mut self.branches[index.index as usize];

    }

}

    branches: &'a [Branch],

    index: usize,

}

impl<'a> Iterator for BranchQueueIter<'a> {

    type Item = &'a Branch;

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

        if self.index == 0 {

            // i.e. the invalid branch index

            return None;

        }

        let branch = &self.branches[self.index];

        self.index = branch.next_in_queue.index as usize;

        return Some(branch);

    }

}

    branches: &'a [Branch],

    index: usize,

}

impl<'a> Iterator for BranchParentIter<'a> {

    type Item = &'a Branch;

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

        if self.index == 0 {

            return None;

        }

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

use crate::common::ComponentState;

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

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

use crate::runtime2::consensus::find_ports_in_value_group;

use crate::runtime2::port::PortKind;

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

use super::inbox2::{DataMessageFancy, MessageFancy, SyncMessageFancy, PublicInbox};

use super::native::Connector;

use super::port::PortIdLocal;

use super::scheduler::{ComponentCtxFancy, SchedulerCtx};

pub(crate) struct ConnectorPublic {

    pub inbox: PublicInbox,

    pub sleeping: AtomicBool,

}

impl ConnectorPublic {

            match message {

                MessageFancy::Data(message) => self.handle_new_data_message(message, ctx),

                MessageFancy::Sync(message) => self.handle_new_sync_message(message, ctx),

                MessageFancy::Control(_) => unreachable!("control message in component"),

            }

        }

    }

    pub fn handle_new_data_message(&mut self, message: DataMessageFancy, ctx: &mut ComponentCtxFancy) {

        // Go through all branches that are awaiting new messages and see if

        // there is one that can receive this message.

        debug_assert!(ctx.workspace_branches.is_empty());

            // This branch can receive, so fork and given it the message

            let receiving_branch_id = self.tree.fork_branch(branch_id);

            self.consensus.notify_of_new_branch(branch_id, receiving_branch_id);

            let receiving_branch = &mut self.tree[receiving_branch_id];

            receiving_branch.insert_message(message.data_header.target_port, message.content.clone());

            self.consensus.notify_of_received_message(branch_id, &message.data_header, &message.content);

            // And prepare the branch for running

            self.tree.push_into_queue(QueueKind::Runnable, receiving_branch_id);

        }

    }

        if branch_id.is_none() {

            return ConnectorScheduling::NotNow;

        }

        // Retrieve the branch and run it

        let branch_id = branch_id.unwrap();

        let branch = &mut self.tree[branch_id];

        let mut run_context = ConnectorRunContext{

            branch_id,

            consensus: &self.consensus,

            received: &branch.inbox,

            prepared_channel: branch.prepared_channel.take(),

        };

        let run_result = branch.code_state.run(&mut run_context, &sched_ctx.runtime.protocol_description);

        // Handle the returned result. Note that this match statement contains

        // explicit returns in case the run result requires that the component's

        // code is ran again immediately

        match run_result {

            RunResult::BranchInconsistent => {

                // Branch became inconsistent

                branch.sync_state = SpeculativeState::Inconsistent;

            },

    }

    pub fn run_in_deterministic_mode(&mut self, sched_ctx: SchedulerCtx, comp_ctx: &mut ComponentCtxFancy) -> ConnectorScheduling {

        debug_assert!(!self.tree.is_in_sync() && !self.consensus.is_in_sync());

        let branch = self.tree.base_branch_mut();

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

        let mut run_context = ConnectorRunContext{

            branch_id: branch.id,

            consensus: &self.consensus,

            received: &branch.inbox,

            prepared_channel: branch.prepared_channel.take(),

        };

        let run_result = branch.code_state.run(&mut run_context, &sched_ctx.runtime.protocol_description);

        match run_result {

            RunResult::ComponentTerminated => {

                branch.sync_state = SpeculativeState::Finished;

                return ConnectorScheduling::Exit;

            },

            RunResult::ComponentAtSyncStart => {

                let sync_branch_id = self.tree.start_sync();

                self.tree.push_into_queue(QueueKind::Runnable, sync_branch_id);

                return ConnectorScheduling::Immediate;

            },

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

                // Note: we're relinquishing ownership of ports. But because

                // we are in non-sync mode the scheduler will handle and check

                // port ownership transfer.

                debug_assert!(comp_ctx.workspace_ports.is_empty());

                find_ports_in_value_group(&arguments, &mut comp_ctx.workspace_ports);

                let new_state = ComponentState {

use super::ConnectorId;

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

use super::inbox2::{

    DataHeader, DataMessageFancy, MessageFancy,

    SyncContent, SyncHeader, SyncMessageFancy, PortAnnotation

};

use super::scheduler::ComponentCtxFancy;

struct BranchAnnotation {

    port_mapping: Vec<PortAnnotation>,

}

pub(crate) struct LocalSolution {

    component: ConnectorId,

    final_branch_id: BranchId,

    port_mapping: Vec<(ChannelId, BranchId)>,

}

pub(crate) struct GlobalSolution {

    component_branches: Vec<(ConnectorId, BranchId)>,

    channel_mapping: Vec<(ChannelId, BranchId)>, // TODO: This can go, is debugging info

}

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

// Consensus

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

/// The consensus algorithm. Currently only implemented to find the component

/// with the highest ID within the sync region and letting it handle all the

/// local solutions.

    // Local component's state

    highest_connector_id: ConnectorId,

    branch_annotations: Vec<BranchAnnotation>,

    last_finished_handled: Option<BranchId>,

    // Gathered state (in case we are currently the leader of the distributed

    // consensus protocol)

    encountered_peers: VecSet<ConnectorId>,

    solution_combiner: SolutionCombiner,

    // Workspaces

    workspace_ports: Vec<PortIdLocal>,

}

pub(crate) enum Consistency {

    Valid,

    Inconsistent,

}

impl Consensus {

    pub fn new() -> Self {

        return Self {

            highest_connector_id: ConnectorId::new_invalid(),

            branch_annotations: Vec::new(),

            last_finished_handled: None,

            encountered_peers: VecSet::new(),

    }

    /// TODO: Remove this once multi-fire is in place

    pub fn get_annotation(&self, branch_id: BranchId, port_id: PortIdLocal) -> &PortAnnotation {

        let branch = &self.branch_annotations[branch_id.index as usize];

        let port = branch.port_mapping.iter().find(|v| v.port_id == port_id).unwrap();

        return port;

    }

    /// Sets up the consensus algorithm for a new synchronous round. The

    /// provided ports should be the ports the component owns at the start of

    /// the sync round.

        debug_assert!(!self.highest_connector_id.is_valid());

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

        debug_assert!(self.last_finished_handled.is_none());

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

        debug_assert!(self.solution_combiner.local.is_empty());

        // We'll use the first "branch" (the non-sync one) to store our ports,

        // this allows cloning if we created a new branch.

        self.branch_annotations.push(BranchAnnotation{

                .map(|v| PortAnnotation{

                    port_id: v.self_id,

                    registered_id: None,

                    expected_firing: None,

                })

                .collect(),

        });

        self.highest_connector_id = ctx.id;

    }

        self.branch_annotations.clear();

        self.last_finished_handled = None;

        self.encountered_peers.clear();

        self.solution_combiner.clear();

    }

    // --- Handling messages

    /// Prepares a message for sending. Caller should have made sure that

    /// sending the message is consistent with the speculative state.

    pub fn handle_message_to_send(&mut self, branch_id: BranchId, source_port_id: PortIdLocal, content: &ValueGroup, ctx: &mut ComponentCtxFancy) -> (SyncHeader, DataHeader) {

        debug_assert!(self.is_in_sync());

        let branch = &mut self.branch_annotations[branch_id.index as usize];

        if cfg!(debug_assertions) {

            let port = branch.port_mapping.iter()

                .find(|v| v.port_id == source_port_id)

                .unwrap();

            debug_assert!(port.expected_firing == None || port.expected_firing == Some(true));

        }

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

        find_ports_in_value_group(content, &mut self.workspace_ports);

        if !self.workspace_ports.is_empty() {

            todo!("handle sending ports");

            self.workspace_ports.clear();

        }

        // TODO: Handle multiple firings. Right now we just assign the current

        //  branch to the `None` value because we know we can only send once.

        debug_assert!(branch.port_mapping.iter().find(|v| v.port_id == source_port_id).unwrap().registered_id.is_none());

        let port_info = ctx.get_port_by_id(source_port_id).unwrap();

        let data_header = DataHeader{

            expected_mapping: branch.port_mapping.clone(),

            sending_port: port_info.peer_id,

            target_port: port_info.peer_id,

            new_mapping: branch_id

        };

        for mapping in &mut branch.port_mapping {

            if mapping.port_id == source_port_id {

                mapping.expected_firing = Some(true);

                mapping.registered_id = Some(branch_id);

            SyncContent::Notification => {

                // We were just interested in the header

                return None;

            },

            SyncContent::LocalSolution(solution) => {

                // We might be the leader, or earlier messages caused us to not

                // be the leader anymore.

                return self.send_or_store_local_solution(solution, ctx);

            },

            SyncContent::GlobalSolution(solution) => {

                // Take branch of interest and return it.

                let (_, branch_id) = solution.component_branches.iter()

                    .unwrap();

                return Some(*branch_id);

            }

        }

    }

    pub fn notify_of_received_message(&mut self, branch_id: BranchId, data_header: &DataHeader, content: &ValueGroup) {

        debug_assert!(self.branch_can_receive(branch_id, data_header));

        let branch = &mut self.branch_annotations[branch_id.index as usize];

        for mapping in &mut branch.port_mapping {

            if mapping.port_id == data_header.target_port {

                // Found the port in which the message should be inserted

    }

    fn handle_received_sync_header(&mut self, sync_header: &SyncHeader, ctx: &mut ComponentCtxFancy) {

        debug_assert!(sync_header.sending_component_id != ctx.id); // not sending to ourselves

        self.encountered_peers.push(sync_header.sending_component_id);

        if sync_header.highest_component_id > self.highest_connector_id {

            // Sender has higher component ID. So should be the target of our

            // messages. We should also let all of our peers know

            self.highest_connector_id = sync_header.highest_component_id;

            for encountered_id in self.encountered_peers.iter() {

                    // Don't need to send it to this one

                    continue

                }

                let message = SyncMessageFancy{

                    sync_header: self.create_sync_header(ctx),

                    content: SyncContent::Notification,

                };

                ctx.submit_message(MessageFancy::Sync(message));

            }

            // But also send our locally combined solution

            self.forward_local_solutions(ctx);

        } else if sync_header.highest_component_id < self.highest_connector_id {

            // Sender has lower leader ID, so it should know about our higher

            // one.

            let message = SyncMessageFancy{

                sync_header: self.create_sync_header(ctx),

                if !matching_channels.is_empty() {

                    // We share some ports

                    component_peers.push(ComponentPeer{

                        target_id: other_component.component,

                        target_index: other_index,

                        involved_channels: matching_channels,

                    });

                }

            }

            let mut num_ports_in_peers = 0;

                num_ports_in_peers += peer.involved_channels.len();

            }

            if num_ports_in_peers == cur_ports.len() {

                // Newly added component has all required peers present

                self.local[component_index].all_peers_present = true;

            }

            // Add the found component pairing entries to the solution entries

            // for the two involved components

            for component_match in component_peers {

                // Check the other component for having all peers present

        total_num_channels /= 2;

        let mut final_mapping = Vec::with_capacity(total_num_channels);

        let mut total_num_checked = 0;

        for (component_index, solution_index) in check_stack.iter().copied() {

            let component = &self.local[component_index];

            let solution = &component.solutions[solution_index];

            for (channel_id, branch_id) in solution.channel_mapping.iter().copied() {

                match final_mapping.iter().find(|(v, _)| *v == channel_id) {

                    Some((_, encountered_branch_id)) => {

                        total_num_checked += 1;

                    },

                    None => {

                        final_mapping.push((channel_id, branch_id));

                    }

                }

            }

        }

        debug_assert_eq!(total_num_checked, total_num_channels);

        return Some(GlobalSolution{

//  Should behave as a MPSC queue.

pub struct PublicInbox {

    messages: Mutex<VecDeque<MessageFancy>>,

}

impl PublicInbox {

    pub fn new() -> Self {

        Self{

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

        }

    }

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

        lock.push_back(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();

    }

}

    }

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

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

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 super::{ConnectorKey, ConnectorId, RuntimeInner};

use super::scheduler::{SchedulerCtx, ComponentCtxFancy};

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

use super::consensus::find_ports_in_value_group;

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

use super::inbox2::{MessageFancy, ControlContent, ControlMessageFancy};

/// 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 ComponentCtxFancy) -> ConnectorScheduling;

}

        // 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 initial_port in &initial_ports {

            if !self.owned_ports.iter().any(|v| v == initial_port) {

                return Err(ComponentCreationError::UnownedPort);

            }

        }

        // We own all ports, so remove them on this side

        for initial_port in &initial_ports {

            self.owned_ports.remove(position);

        }

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

        let connector = ConnectorPDL::new(state);

        // Put on job queue

        {

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

            queue.push_back(ApplicationJob::NewConnector(connector, initial_ports));

        }

use std::collections::VecDeque;

use std::sync::Arc;

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

use crate::runtime2::inbox2::ControlContent;

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

use super::native::Connector;

use super::branch::{BranchId};

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

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

}

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

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

            }

            match message {

                MessageFancy::Control(message) => {

                    match message.content {

                        ControlContent::PortPeerChanged(port_id, new_target_connector_id) => {

                            // Need to change port target

                            let port = scheduled.ctx_fancy.get_port_mut_by_id(port_id).unwrap();

                            port.peer_connector = new_target_connector_id;

                            // Note: for simplicity we program the scheduler to always finish

                            // running a connector with an empty outbox. If this ever changes

                            // then accepting the "port peer changed" message implies we need

                            // to change the recipient of the message in the outbox.

            }

        }

    }

    /// Handles changes to the context that were made by the component. This is

    /// the way (due to Rust's borrowing rules) that we bubble up changes in the

    /// component's state that the scheduler needs to know about (e.g. a message

    /// that the component wants to send, a port that has been added).

    fn handle_changes_in_context(&mut self, scheduled: &mut ScheduledConnector) {

        let connector_id = scheduled.ctx_fancy.id;

        // Handling any messages that were sent

            self.debug_conn(connector_id, &format!("Sending message [outbox] \n --- {:?}", message));

            let target_component_id = match &message {

                MessageFancy::Data(content) => {

                    // Data messages are always sent to a particular port, and

                    // may end up being rerouted.

                    let port_desc = scheduled.ctx_fancy.get_port_by_id(content.data_header.sending_port).unwrap();

                    debug_assert_eq!(port_desc.peer_id, content.data_header.target_port);

                    if port_desc.state == PortState::Closed {

                        todo!("handle sending over a closed port")

                    }

                },

                MessageFancy::Control(_) => {

                    unreachable!("component sending control messages directly");

                }

            };

            self.runtime.send_message(target_component_id, message);

        }

        while let Some(state_change) = scheduled.ctx_fancy.state_changes.pop_front() {

            match state_change {

                ComponentStateChange::CreatedComponent(component, initial_ports) => {

                    let new_key = self.runtime.create_pdl_component(component, false);

                    let new_connector = self.runtime.get_component_private(&new_key);

                    for port_id in initial_ports {

                        // Transfer messages associated with the transferred port

                        let mut message_idx = 0;

                        while message_idx < scheduled.ctx_fancy.inbox_messages.len() {

                            let message = &scheduled.ctx_fancy.inbox_messages[message_idx];

                                // Need to transfer this message

                            } else {

                                message_idx += 1;

                            }

                        }

                        // Transfer the port itself

                        let port_index = scheduled.ctx_fancy.ports.iter()

                            .unwrap();

                        let port = scheduled.ctx_fancy.ports.remove(port_index);

                        new_connector.ctx_fancy.ports.push(port.clone());

                        // Notify the peer that the port has changed

                        let reroute_message = scheduled.router.prepare_reroute(

                            port.self_id, port.peer_id, scheduled.ctx_fancy.id,

                            port.peer_connector, new_connector.ctx_fancy.id

                        );

                        self.debug_conn(connector_id, &format!("Sending message [newcon]\n --- {:?}", reroute_message));

                        self.runtime.send_message(port.peer_connector, MessageFancy::Control(reroute_message));

            // Try to wake ourselves up (needed because someone might be trying

            // the exact same atomic compare-and-swap at this point in time)

            let should_wake_up_again = connector.public.sleeping

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

                .is_ok();

            if should_wake_up_again {

                self.runtime.push_work(connector_key)

            }

        }

    }