// To share some logic between the FakeTree and ExecTree implementation

trait BranchListItem {

}

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

const NUM_QUEUES: usize = 3;

pub(crate) enum QueueKind {

    Runnable,

    AwaitingMessage,

        return &mut self.branches[0];

    }

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

    }

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

    /// using the provided branch as the final sync result.

    pub fn end_sync(&mut self, branch_id: BranchId) {

        debug_assert!(self.is_in_sync());

        // Swap indicated branch into the first position

        self.branches.swap(0, branch_id.index as usize);

    }

}

/// Iterator over the parents of an `ExecTree` branch.

pub(crate) struct BranchParentIter<'a> {

    branches: &'a [Branch],

}

impl FakeBranch {

        return FakeBranch{

            parent_id: BranchId::new_invalid(),

            sync_state: SpeculativeState::RunningInSync,

            awaiting_port: PortIdLocal::new_invalid(),

impl FakeTree {

    pub fn new() -> Self {

        return Self {

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

        }

    }

        push_into_queue(&mut self.queues[kind.as_index()], &mut self.branches, id);

    }

    }

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

    }

    pub fn end_sync(&mut self, branch_id: BranchId) -> FakeBranch {

        debug_assert!(self.is_in_sync());

        // Take out the succeeding branch, then just clear all fake branches.

        for queue_index in 0..NUM_QUEUES {

            self.queues[queue_index] = BranchQueue::new();

        queue.first = branch_id;

        queue.last = branch_id;

    } else {

        last_branch.set_next_id(branch_id);

        queue.last = branch_id;

    }

}

pub(crate) struct ConnectorPDL {

    tree: ExecTree,

    consensus: Consensus,

}

struct ConnectorRunContext<'a> {

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

        self.handle_new_messages(comp_ctx);

        if self.tree.is_in_sync() {

            let scheduling = self.run_in_sync_mode(sched_ctx, comp_ctx);

                    self.collapse_sync_to_solution_branch(solution_branch_id, comp_ctx);

                    return ConnectorScheduling::Immediate;

                }

        } else {

            let scheduling = self.run_in_deterministic_mode(sched_ctx, comp_ctx);

            return scheduling;

        Self{

            tree: ExecTree::new(initial),

            consensus: Consensus::new(),

        }

    }

    pub fn handle_new_data_message(&mut self, message: DataMessage, ctx: &mut ComponentCtx) {

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

        // there is one that can receive this message.

            // Old message, so drop it

            return;

        }

            // 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.as_message().unwrap().clone());

            // And prepare the branch for running

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

                    // a message that targets this branch, so check now.

                    let mut any_message_received = false;

                    for message in comp_ctx.get_read_data_messages(port_id) {

                            // This branch can receive the message, so we do the

                            // fork-and-receive dance

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

                            branch.insert_message(port_id, message.content.as_message().unwrap().clone());

                            self.consensus.notify_of_new_branch(branch_id, receiving_branch_id);

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

                            any_message_received = true;

            RunResult::ComponentAtSyncStart => {

                comp_ctx.notify_sync_start();

                let sync_branch_id = self.tree.start_sync();

                self.consensus.start_sync(comp_ctx);

                self.consensus.notify_of_new_branch(BranchId::new_invalid(), sync_branch_id);

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

        }

        ctx.notify_sync_end(&[]);

    }

}

use crate::collections::VecSet;

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

use super::ConnectorId;

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

use super::inbox::{

    Message, PortAnnotation,

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

//  performed on the same port to prevent repeated lookups

// TODO: A lot of stuff should be batched. Like checking all the sync headers

pub(crate) struct Consensus {

    // --- State that is cleared after each round

    // Local component's state

    highest_connector_id: ConnectorId,

    branch_annotations: Vec<BranchAnnotation>,

    // Gathered state from communication

    encountered_ports: VecSet<PortIdLocal>, // to determine if we should send "port remains silent" messages.

    solution_combiner: SolutionCombiner,

        return Self {

            highest_connector_id: ConnectorId::new_invalid(),

            branch_annotations: Vec::new(),

            encountered_ports: VecSet::new(),

            solution_combiner: SolutionCombiner::new(),

            peers: Vec::new(),

    pub fn start_sync(&mut self, ctx: &ComponentCtx) {

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

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

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

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

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

    }

        // Turn the port mapping into a local solution

        let mut target_mapping = Vec::with_capacity(source_mapping.len());

        for port in source_mapping {

        let local_solution = LocalSolution{

            component: ctx.id,

            port_mapping: target_mapping,

        };

        let solution_branch = self.send_or_store_local_solution(local_solution, ctx);

        return solution_branch;

    }

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

        debug_assert!(self.is_in_sync());

        // Clear out internal storage to defaults

        self.highest_connector_id = ConnectorId::new_invalid();

        self.branch_annotations.clear();

        self.encountered_ports.clear();

        self.solution_combiner.clear();

        return (self.create_sync_header(ctx), data_header);

    }

        return self.handle_received_sync_header(&message.sync_header, ctx)

    }

        }

    }

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

        for mapping in &mut branch.port_mapping {

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

                // Check for sent ports

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

                if !self.workspace_ports.is_empty() {

                    todo!("handle received ports");

                    self.workspace_ports.clear();

    /// Matches the mapping between the branch and the data message. If they

    /// match then the branch can receive the message.

                return false;

            }

        }

            // No port can receive a "silent" notification.

            return false;

        }

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

            // If we own the port, then we have an entry in the

            // annotation, check if the current mapping matches

            for current in &annotation.port_mapping {

    // --- Internal helpers

    fn handle_received_sync_header(&mut self, sync_header: &SyncHeader, ctx: &mut ComponentCtx) -> bool {

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

        if !self.handle_peer(sync_header) {

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

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

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

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

pub(crate) trait Connector {

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

}

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

enum ApplicationJob {

/// The connector which an application can directly interface with. Once may set

/// up the next synchronous round, and retrieve the data afterwards.

// TODO: Strong candidate for logic reduction in handling put/get. A lot of code

pub struct ConnectorApplication {

    // Communicating about new jobs and setting up sync rounds

    sync_done: SyncDone,

    is_in_sync: bool,

    // Handling current sync round

    sync_desc: Vec<ApplicationSyncAction>,

    consensus: Consensus,

    branch_extra: Vec<usize>, // instruction counter per branch

}

impl Connector for ConnectorApplication {

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

        if self.is_in_sync {

        } else {

            return self.run_in_deterministic_mode(sched_ctx, comp_ctx);

        }

impl ConnectorApplication {

    pub(crate) fn new(runtime: Arc<RuntimeInner>) -> (Self, ApplicationInterface) {

        let job_queue = Arc::new(Mutex::new(VecDeque::with_capacity(32)));

        let connector = ConnectorApplication {

            job_queue: job_queue.clone(),

            is_in_sync: false,

            sync_desc: Vec::new(),

        };

        let interface = ApplicationInterface::new(sync_done, job_queue, runtime);

        }

    }

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

        // there is one that can receive this message.

            // Old message, so drop it

            return;

        }

            // 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.as_message().unwrap().clone());

            // And prepare the branch for running

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

        }

    }

        if let Some(solution_branch_id) = self.consensus.handle_new_sync_message(message, ctx) {

            self.collapse_sync_to_solution_branch(solution_branch_id, ctx);

        }

        self.handle_new_messages(comp_ctx);

        if branch_id.is_none() {

            return ConnectorScheduling::NotNow;

        }

        let branch_id = branch_id.unwrap();

        let mut instruction_idx = self.branch_extra[branch_id.index as usize];

        if instruction_idx >= self.sync_desc.len() {

            let consistency = self.consensus.notify_of_finished_branch(branch_id);

            if consistency == Consistency::Valid {

                branch.sync_state = SpeculativeState::ReachedSyncEnd;

            } else {

                branch.sync_state = SpeculativeState::Inconsistent;

            }

                    let port_id = *port_id;

                    let consistency = self.consensus.notify_of_speculative_mapping(branch_id, port_id, true);

                    if consistency == Consistency::Valid {

                        let message = Message::Data(DataMessage {

                            sync_header,

                            data_header,

                            content: DataContent::Message(content.clone()),

                        });

                        comp_ctx.submit_message(message);

                        return ConnectorScheduling::Immediate;

                    } else {

                        branch.sync_state = SpeculativeState::Inconsistent;

                    if consistency == Consistency::Valid {

                        branch.sync_state = SpeculativeState::HaltedAtBranchPoint;

                        branch.awaiting_port = port_id;

                        let mut any_message_received = false;

                        for message in comp_ctx.get_read_data_messages(port_id) {

                                // This branch can receive the message, so we do the

                                // fork-and-receive dance

                                debug_assert!(receiving_branch_id.index as usize == self.branch_extra.len());

                                self.branch_extra.push(instruction_idx + 1);

                                branch.insert_message(port_id, message.content.as_message().unwrap().clone());

                                self.consensus.notify_of_new_branch(branch_id, receiving_branch_id);

                                any_message_received = true;

                            }

            }

        }

            return ConnectorScheduling::NotNow;

        } else {

            return ConnectorScheduling::Later;

                },

                ApplicationJob::SyncRound(mut description) => {

                    // Entering sync mode

                    self.sync_desc = description;

                    self.is_in_sync = true;

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

                    self.branch_extra.push(0); // set first branch to first instruction

                    return ConnectorScheduling::Immediate;

        return ConnectorScheduling::NotNow;

    }

}

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

        let mut initial_ports = Vec::new();

        find_ports_in_value_group(&arguments, &mut initial_ports);

        for initial_port in &initial_ports {

                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 (is_done, _) = &*self.sync_done;

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

        }

        {

        }

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

    }

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

use super::*;

use crate::{PortId, ProtocolDescription};

use crate::common::Id;

    let thing = TestTimer::new("put_and_get");

    run_test_in_runtime(CODE, |api| {

        api.create_connector("", "putter", ValueGroup::new_stack(vec![

            Value::Output(PortId(Id{ connector_id: 0, u32_suffix: channel.putter_id.index })),

        api.create_connector("", "constructor", ValueGroup::new_stack(vec![

            Value::UInt32(5),

            Value::UInt32(NUM_LOOPS),

    });

}

        api.create_connector("", "constructor", ValueGroup::new_stack(vec![

            Value::UInt32(5),

            Value::UInt32(NUM_LOOPS)

    });

}