#[derive(Copy, Clone)]

struct BranchQueue {

    first: BranchId,

    last: BranchId,

}

impl BranchQueue {

    #[inline]

    fn new() -> Self {

        Self{

            first: BranchId::new_invalid(),

            last: BranchId::new_invalid()

        }

    }

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

        }

    }

}

/// Execution tree of branches. Tries to keep the extra information stored

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

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

    /// Returns if tree is in speculative mode

    pub fn is_in_sync(&self) -> bool {

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

    }

    /// Returns true if the particular queue is empty

    pub fn queue_is_empty(&self, kind: QueueKind) -> bool {

        return self.queues[kind.as_index()].is_empty();

    }

    /// Pops a branch (ID) from a queue.

    pub fn pop_from_queue(&mut self, kind: QueueKind) -> Option<BranchId> {

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

        if queue.is_empty() {

            return None;

        } else {

            let first_branch = &mut self.branches[queue.first.index as usize];

            queue.first = first_branch.next_in_queue;

            first_branch.next_in_queue = BranchId::new_invalid();

            if !queue.first.is_valid() {

                queue.last = BranchId::new_invalid();

            }

            return Some(first_branch.id);

        }

    }

    /// Pushes a branch (ID) into a queue.

    pub fn push_into_queue(&mut self, kind: QueueKind, id: BranchId) {

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

        if queue.is_empty() {

            queue.first = id;

            queue.last = id;

        } else {

            let last_branch = &mut self.branches[queue.last.index as usize];

            last_branch.next_in_queue = id;

            queue.last = id;

        }

    }

    /// Returns the non-sync branch (TODO: better name?)

    pub fn base_branch_mut(&mut self) -> &mut Branch {

        debug_assert!(!self.is_in_sync());

        return &mut self.branches[0];

    }

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

        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 {

        return BranchParentIter{

            branches: self.branches.as_slice(),

            index: branch_id.index as usize,

        }

    }

    // --- Preparing and finishing a speculative round

    /// Starts a synchronous round by cloning the non-sync branch and marking it

    /// as the root of the speculative tree. The id of this root sync branch is

    /// returned.

    pub fn start_sync(&mut self) -> BranchId {

        debug_assert!(!self.is_in_sync());

        let sync_branch = Branch::new_sync(1, &self.branches[0]);

        let sync_branch_id = sync_branch.id;

/// connector.rs

///

/// Represents a component. A component (and the scheduler that is running it)

/// has many properties that are not easy to subdivide into aspects that are

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

use super::inbox::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 {

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

        ConnectorPublic{

            inbox: PublicInbox::new(),

            sleeping: AtomicBool::new(initialize_as_sleeping),

        }

    }

}

#[derive(Eq, PartialEq)]

pub(crate) enum ConnectorScheduling {

    Immediate,      // Run again, immediately

    Later,          // Schedule for running, at some later point in time

    NotNow,         // Do not reschedule for running

}

pub(crate) struct ConnectorPDL {

    tree: ExecTree,

    consensus: Consensus,

}

impl RunContext for ConnectorRunContext{

    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 run(&mut self, sched_ctx: SchedulerCtx, comp_ctx: &mut ComponentCtxFancy) -> ConnectorScheduling {

    }

}

impl ConnectorPDL {

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

        Self{

            tree: ExecTree::new(initial),

            consensus: Consensus::new(),

        }

    }

    // --- Handling messages

    pub fn handle_new_messages(&mut self, ctx: &mut ComponentCtxFancy) {

        while let Some(message) = ctx.read_next_message() {

            match message {

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

                MessageFancy::Sync(message) => 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) {

        self.consensus.handle_received_sync_header(&message.sync_header, ctx);

        self.consensus.handle_received_data_header(&self.tree, &message.data_header, &mut ctx.workspace_branches);

        for branch_id in ctx.workspace_branches.drain(..) {

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

        }

    }

    pub fn handle_new_sync_message(&mut self, message: SyncMessageFancy, ctx: &mut ComponentCtxFancy) {

        self.consensus.handle_received_sync_header(&message.sync_header, ctx);

        todo!("handle content of message?");

    }

    // --- Running code

        // Check if we have any branch that needs running

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

        let branch_id = self.tree.pop_from_queue(QueueKind::Runnable);

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

            },

            RunResult::BranchMissingPortState(port_id) => {

                // Branch called `fires()` on a port that has not been used yet.

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

                // Create two forks, one that assumes the port will fire, and

                // one that assumes the port remains silent

                branch.sync_state = SpeculativeState::HaltedAtBranchPoint;

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

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

                self.consensus.notify_of_new_branch(branch_id, firing_branch_id);

                let _result = self.consensus.notify_of_speculative_mapping(firing_branch_id, port_id, true);

                debug_assert_eq!(_result, Consistency::Valid);

                self.consensus.notify_of_new_branch(branch_id, silent_branch_id);

                    // `put()` is valid.

                    let (sync_header, data_header) = self.consensus.handle_message_to_send(branch_id, port_id, &content, comp_ctx);

                    comp_ctx.submit_message(MessageFancy::Data(DataMessageFancy{

                        sync_header, data_header, content

                    }));

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

                    return ConnectorScheduling::Immediate;

                } else {

                    branch.sync_state = SpeculativeState::Inconsistent;

                }

            },

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

        }

        // If here then the run result did not require a particular action. We

        // return whether we have more active branches to run or not.

        if self.tree.queue_is_empty(QueueKind::Runnable) {

            return ConnectorScheduling::NotNow;

        } else {

            return ConnectorScheduling::Later;

        }

    }

        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 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 current_ports = comp_ctx.notify_sync_start();

                let sync_branch_id = self.tree.start_sync();

                self.consensus.start_sync(current_ports);

                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 crate::protocol::eval::ValueGroup;

use crate::runtime2::branch::{BranchId, ExecTree, QueueKind};

use crate::runtime2::ConnectorId;

use crate::runtime2::inbox2::{DataHeader, SyncHeader};

use crate::runtime2::port::{Port, PortIdLocal};

use crate::runtime2::scheduler::ComponentCtxFancy;

use super::inbox2::PortAnnotation;

struct BranchAnnotation {

    port_mapping: Vec<PortAnnotation>,

}

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

///

/// The type itself serves as an experiment to see how code should be organized.

// TODO: Flatten all datastructures

// TODO: Have a "branch+port position hint" in case multiple operations are

//  performed on the same port to prevent repeated lookups

pub(crate) struct Consensus {

    highest_connector_id: ConnectorId,

    branch_annotations: Vec<BranchAnnotation>,

    workspace_ports: Vec<PortIdLocal>,

}

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

pub(crate) enum Consistency {

    Valid,

    Inconsistent,

}

impl Consensus {

    pub fn new() -> Self {

        return Self {

            highest_connector_id: ConnectorId::new_invalid(),

            branch_annotations: Vec::new(),

        }

    }

    // --- Controlling sync round and branches

    /// Returns whether the consensus algorithm is running in sync mode

    pub fn is_in_sync(&self) -> bool {

        return !self.branch_annotations.is_empty();

    }

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

    pub fn start_sync(&mut self, ports: &[Port]) {

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

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

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

            port_mapping: ports.iter()

                .map(|v| PortAnnotation{

                    port_id: v.self_id,

                    registered_id: None,

                    expected_firing: None,

                })

                .collect(),

        });

    }

    /// Notifies the consensus algorithm that a new branch has appeared. Must be

    /// called for each forked branch in the execution tree.

    pub fn notify_of_new_branch(&mut self, parent_branch_id: BranchId, new_branch_id: BranchId) {

        // If called correctly. Then each time we are notified the new branch's

        // index is the length in `branch_annotations`.

        debug_assert!(self.branch_annotations.len() == new_branch_id.index as usize);

        let parent_branch_annotations = &self.branch_annotations[parent_branch_id.index as usize];

        let new_branch_annotations = BranchAnnotation{

            port_mapping: parent_branch_annotations.port_mapping.clone(),

        };

        self.branch_annotations.push(new_branch_annotations);

    }

    /// Notifies the consensus algorithm that a branch has reached the end of

    /// the sync block. A final check for consistency will be performed that the

    pub fn notify_of_finished_branch(&self, branch_id: BranchId) -> Consistency {

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

        for mapping in &branch.port_mapping {

            match mapping.expected_firing {

                Some(expected) => {

                    if expected != mapping.registered_id.is_some() {

                        // Inconsistent speculative state and actual state

                        debug_assert!(mapping.registered_id.is_none()); // because if we did fire on a silent port, we should've caught that earlier

                        return Consistency::Inconsistent;

                    }

                },

                None => {},

            }

        }

        return Consistency::Valid;

    }

    /// Notifies the consensus algorithm that a particular branch has assumed

    /// a speculative value for its port mapping.

    pub fn notify_of_speculative_mapping(&mut self, branch_id: BranchId, port_id: PortIdLocal, does_fire: bool) -> Consistency {

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

        for mapping in &mut branch.port_mapping {

            if mapping.port_id == port_id {

                match mapping.expected_firing {

                    None => {

                        // Not yet mapped, perform speculative mapping

                        mapping.expected_firing = Some(does_fire);

                        return Consistency::Valid;

                    },

                    Some(current) => {

                        // Already mapped

                        if current == does_fire {

                            return Consistency::Valid;

                        } else {

                            return Consistency::Inconsistent;

                        }

                    }

                }

            }

        }

        unreachable!("notify_of_speculative_mapping called with unowned port");

    }

    pub fn end_sync(&mut self, branch_id: BranchId, final_ports: &mut Vec<PortIdLocal>) {

    }

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

        }

        // Check for ports that are begin sent

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

        }

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

            }

        }

        return (sync_header, data_header);

    }

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

        todo!("should check IDs and maybe send sync messages");

    }

    /// Checks data header and consults the stored port mapping and the

    /// execution tree to see which branches may receive the data message's

    /// contents.

    ///

    /// This function is generally called for freshly received messages that

    /// should be matched against previously halted branches.

    pub fn handle_received_data_header(&mut self, exec_tree: &ExecTree, data_header: &DataHeader, target_ids: &mut Vec<BranchId>) {

            if branch.awaiting_port == data_header.target_port {

                // Found a branch awaiting the message, but we need to make sure

                // the mapping is correct

                if self.branch_can_receive(branch.id, data_header) {

                    target_ids.push(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

                mapping.registered_id = Some(data_header.new_mapping);

                // Check for sent ports

                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 received ports");

                    self.workspace_ports.clear();