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

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

const NUM_QUEUES: usize = 3;

pub(crate) enum QueueKind {

    Runnable,

    AwaitingMessage,

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

                branch.next_in_queue.index as usize

            },

            None => {

            }

        };

    }

}

    branches: &'a [Branch],

    index: usize,

}

    }

}

    branches: &'a [Branch],

    index: usize,

}

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;

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

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

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

                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;

    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

    workspace_ports: Vec<PortIdLocal>,

}

pub(crate) enum Consistency {

    Valid,

    Inconsistent,

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

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

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

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

            SyncContent::GlobalSolution(solution) => {

                // Take branch of interest and return it.

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

                    .unwrap();

                return Some(*branch_id);

            }

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

            }

            let mut num_ports_in_peers = 0;

                num_ports_in_peers += peer.involved_channels.len();

            }

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

        }

    }

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

        lock.push_back(message);

    }

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

        return lock.pop_front();

    }

/// A kind of token that allows shared access to a connector. Multiple threads

/// may hold this

pub struct ConnectorId(pub u32);

impl ConnectorId {

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

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

        for initial_port in &initial_ports {

            self.owned_ports.remove(position);

        }

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

        let connector_id = scheduled.ctx_fancy.id;

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

            }

            match message {

                MessageFancy::Control(message) => {

                    match message.content {

        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 {

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

        }

    }

    // TODO: Remove, this is debugging stuff

    fn debug(&self, message: &str) {

        println!("DEBUG [thrd:{:02} conn:  ]: {}", self.scheduler_id, message);

        self.state_changes.push_back(ComponentStateChange::CreatedPort(port))

    }

    pub(crate) fn get_port_by_id(&self, id: PortIdLocal) -> Option<&Port> {

        return self.ports.iter().find(|v| v.self_id == id);

    }

    }

    /// Notify that component will enter a sync block. Note that after calling

        debug_assert!(!self.is_in_sync);

        self.is_in_sync = true;

        self.changed_in_sync = true;

    }

    #[inline]

    /// Returns true if the supplied message should be rerouted. If so then this

    /// function returns the connector that should retrieve this message.

        for entry in &self.active {

            if let ControlVariant::ChangedPort(entry) = &entry.variant {

                if entry.target_port == target_port {