InSync,

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

trait BranchListItem {

    #[inline] fn get_id(&self) -> BranchId;

    #[inline] fn set_next_id(&mut self, id: BranchId);

    #[inline] fn get_next_id(&self) -> BranchId;

}

impl BranchListItem for Branch {

    #[inline] fn get_id(&self) -> BranchId { return self.id; }

    #[inline] fn set_next_id(&mut self, id: BranchId) { self.next_in_queue = id; }

    #[inline] fn get_next_id(&self) -> BranchId { return self.next_in_queue; }

}

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

// ExecTree

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

        return pop_from_queue(&mut self.queues[kind.as_index()], &mut self.branches);

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

    pub fn iter_queue(&self, kind: QueueKind, start_after: Option<BranchId>) -> BranchQueueIter<'_, Branch> {

        // Make sure branch is in correct queue while in debug mode

        debug_assert!(start_after

            .map(|branch_id| self.iter_queue(kind, None).any(|v| v.id == branch_id))

            .unwrap_or(true));

        return iter_queue(queue, &self.branches, start_after);

/// Iterator over branches in a `ExecTree` queue.

pub(crate) struct BranchQueueIter<'a, B: BranchListItem> {

    branches: &'a [B],

impl<'a, B: BranchListItem> Iterator for BranchQueueIter<'a, B> {

    type Item = &'a B;

        self.index = branch.get_next_id().index as usize;

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

}

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

// FakeTree

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

/// Generic fake branch. This is supposed to be used in conjunction with the

/// fake tree. The purpose is to have a branching-like tree to use in

/// combination with a consensus algorithm in places where we don't have PDL

/// code.

pub(crate) struct FakeBranch {

    pub id: BranchId,

    pub parent_id: BranchId,

    pub sync_state: SpeculativeState,

    pub awaiting_port: PortIdLocal,

    pub next_in_queue: BranchId,

    pub inbox: HashMap<PortIdLocal, ValueGroup>,

}

impl BranchListItem for FakeBranch {

    #[inline] fn get_id(&self) -> BranchId { return self.id; }

    #[inline] fn set_next_id(&mut self, id: BranchId) { self.next_in_queue = id; }

    #[inline] fn get_next_id(&self) -> BranchId { return self.next_in_queue; }

}

impl FakeBranch {

    fn new_root(index: u32) -> FakeBranch {

        debug_assert!(index == 0);

        return FakeBranch{

            id: BranchId::new_invalid(),

            parent_id: BranchId::new_invalid(),

            sync_state: SpeculativeState::RunningInSync,

            awaiting_port: PortIdLocal::new_invalid(),

            next_in_queue: BranchId::new_invalid(),

            inbox: HashMap::new(),

        }

    }

    fn new_branching(index: u32, parent_branch: &FakeBranch) -> FakeBranch {

        return FakeBranch {

            id: BranchId::new(index),

            parent_id: parent_branch.id,

            sync_state: SpeculativeState::RunningInSync,

            awaiting_port: parent_branch.awaiting_port,

            next_in_queue: BranchId::new_invalid(),

            inbox: parent_branch.inbox.clone(),

        }

    }

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

        debug_assert!(target_port.is_valid());

        debug_assert!(self.awaiting_port == target_port);

        self.awaiting_port = PortIdLocal::new_invalid();

        self.inbox.insert(target_port, contents);

    }

}

/// A little helper for native components that don't have a set of branches that

/// are actually executing code, but just have to manage the idea of branches

/// due to them performing the equivalent of a branching `get` call.

pub(crate) struct FakeTree {

    pub branches: Vec<FakeBranch>,

    queues: [BranchQueue; NUM_QUEUES],

}

impl FakeTree {

    pub fn new() -> Self {

        return Self {

            branches: Vec::new(),

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

        }

    }

    fn is_in_sync(&self) -> bool {

        return !self.branches.is_empty();

    }

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

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

    }

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

        debug_assert_ne!(kind, QueueKind::FinishedSync);

        return pop_from_queue(&mut self.queues[kind.as_index()], &mut self.branches);

    }

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

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

    }

    pub fn iter_queue(&self, kind: QueueKind, start_after: Option<BranchId>) -> BranchQueueIter<'_, FakeBranch> {

        debug_assert!(start_after

            .map(|branch_id| self.iter_queue(kind, None).any(|v| v.id == branch_id))

            .unwrap_or(true)

        );

        return iter_queue(&self.queues[kind.as_index()], &self.branches, start_after);

    }

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

        debug_assert!(!self.is_in_sync());

        // Create the first branch

        let sync_branch = FakeBranch::new_root(0);

        let sync_branch_id = sync_branch.id;

        self.branches.push(sync_branch);

        return sync_branch_id;

    }

    pub fn fork_branch(&mut self, parent_branch_id: BranchId) -> BranchId {

        debug_assert!(self.is_in_sync());

        let parent_branch = &self[parent_branch_id];

        let new_branch = FakeBranch::new_branching(self.branches.len() as u32, parent_branch);

        let new_branch_id = new_branch.id;

        self.branches.push(new_branch);

        return new_branch_id;

    }

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

        debug_assert!(self.is_in_sync());

        debug_assert!(self.iter_queue(QueueKind::FinishedSync, None).any(|v| v.id == BranchId));

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

        let mut iter = self.branches.drain(branch_id.index..);

        let result = iter.next().unwrap();

        for queue_index in 0..NUM_QUEUES {

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

        }

        return result;

    }

}

impl Index<BranchId> for FakeTree {

    type Output = FakeBranch;

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

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

    }

}

impl IndexMut<BranchId> for FakeTree {

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

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

    }

}

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

// Shared logic

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

fn pop_from_queue<B: BranchListItem>(queue: &mut BranchQueue, branches: &mut [B]) -> Option<BranchId> {

    if queue.is_empty() {

        return None;

    } else {

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

        queue.first = first_branch.get_next_id();

        first_branch.set_next_id(BranchId::new_invalid());

        if !queue.first.is_valid() {

            queue.last = BranchId::new_invalid();

        }

        return Some(first_branch.get_id());

    }

}

fn push_into_queue<B: BranchListItem>(queue: &mut BranchQueue, branches: &mut [B], branch_id: BranchId) {

    debug_assert!(!branches[branch_id.index as usize].get_next_id().is_valid());

    if queue.is_empty() {

        queue.first = branch_id;

        queue.last = branch_id;

    } else {

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

        last_branch.set_next_id(branch_id);

        queue.last = branch_id;

    }

}

fn iter_queue<'a, B: BranchListItem>(queue: &BranchQueue, branches: &'a [B], start_after: Option<BranchId>) -> BranchQueueIter<'a, B> {

    let index = match start_after {

        Some(branch_id) => {

            // Assuming caller is correct and that the branch is in the queue

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

            first_branch.get_next_id().index as usize

        },

        None => {

            queue.first.index as usize

        }

    };

    return BranchQueueIter{ branches, index };

                    let mut any_message_received = false;

                            any_message_received = true;

                    if any_message_received {

                } else {

use crate::runtime2::branch::BranchQueueIter;

    pub fn handle_new_data_message(&mut self, potential_receivers: BranchQueueIter<'_, >, message: &DataMessage, ctx: &mut ComponentCtx, target_ids: &mut Vec<BranchId>) -> bool {

use crate::runtime2::branch::{FakeTree, QueueKind, SpeculativeState};

use crate::runtime2::consensus::{Consensus, Consistency};

use crate::runtime2::inbox::{DataContent, DataMessage, SyncMessage};

    SyncRound(Vec<ApplicationSyncAction>),

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

// ConnectorApplication

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

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

//  is an approximate copy-pasta from the regular component logic.

    // Communicating about new jobs and setting up sync rounds

    is_in_sync: bool,

    // Handling current sync round

    sync_desc: Vec<ApplicationSyncAction>,

    exec_tree: FakeTree,

    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 {

            return self.run_in_sync_mode(sched_ctx, comp_ctx);

        } else {

            return self.run_in_deterministic_mode(sched_ctx, comp_ctx);

        }

    }

            job_queue: job_queue.clone(),

            is_in_sync: false,

            sync_desc: Vec::new(),

    fn handle_new_messages(&mut self, comp_ctx: &mut ComponentCtx) {

    }

    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.

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

        let mut branches = Vec::new(); // TODO: @Remove

        if !self.consensus.handle_new_data_message(&self.tree, &message, ctx, &mut branches) {

            // Old message, so drop it

            return;

        }

        for branch_id in 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.as_message().unwrap().clone());

            self.consensus.notify_of_received_message(receiving_branch_id, &message.sync_header, &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: SyncMessage, ctx: &mut ComponentCtx) {

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

            self.collapse_sync_to_solution_branch(solution_branch_id, ctx);

        }

    }

    fn run_in_sync_mode(&mut self, _sched_ctx: SchedulerCtx, comp_ctx: &mut ComponentCtx) -> ConnectorScheduling {

        debug_assert!(self.is_in_sync);

        self.handle_new_messages(comp_ctx);

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

        if branch_id.is_none() {

            return ConnectorScheduling::NotNow;

        }

        let branch_id = branch_id.unwrap();

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

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

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

            // Performed last instruction, so this branch is officially at the

            // end of the synchronous interaction.

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

            if consistency == Consistency::Valid {

                branch.sync_state = SpeculativeState::ReachedSyncEnd;

                self.exec_tree.push_into_queue(QueueKind::FinishedSync, branch_id);

            } else {

                branch.sync_state = SpeculativeState::Inconsistent;

            }

        } else {

            // We still have instructions to perform

            let cur_instruction = &self.sync_desc[instruction_idx];

            self.branch_extra[branch_id.index as usize] += 1;

            match &cur_instruction {

                ApplicationSyncAction::Put(port_id, content) => {

                    let port_id = *port_id;

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

                    if consistency == Consistency::Valid {

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

                        let message = Message::Data(DataMessage {

                            sync_header,

                            data_header,

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

                        });

                        comp_ctx.submit_message(message);

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

                        return ConnectorScheduling::Immediate;

                    } else {

                        branch.sync_state = SpeculativeState::Inconsistent;

                },

                ApplicationSyncAction::Get(port_id) => {

                    let port_id = *port_id;

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

                    if consistency == Consistency::Valid {

                        branch.sync_state = SpeculativeState::HaltedAtBranchPoint;

                        branch.awaiting_port = port_id;

                        self.exec_tree.push_into_queue(QueueKind::AwaitingMessage, branch_id);

                        let mut any_message_received = false;

                        for message in comp_ctx.get_read_data_messages(port_id) {

                            if self.consensus.branch_can_receive(branch_id, &message.sync_header, &message.data_header, &message.content) {

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

                                // fork-and-receive dance

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

                                let branch = &mut self.exec_tree[receiving_branch_id];

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

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

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

                                any_message_received = true;

                            }

                        }

                        if any_message_received {

                            return ConnectorScheduling::Immediate;

                        }

                    } else {

                        branch.sync_state = SpeculativeState::Inconsistent;

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

            return ConnectorScheduling::NotNow;

        } else {

            return ConnectorScheduling::Later;

        }

    }

    fn run_in_deterministic_mode(&mut self, _sched_ctx: SchedulerCtx, comp_ctx: &mut ComponentCtx) -> ConnectorScheduling {

        debug_assert!(!self.is_in_sync);

        // In non-sync mode the application component doesn't really do anything

        // except performing jobs submitted from the API. This is the only

        // case where we expect to be woken up.

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

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

            match job {

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

                    comp_ctx.push_port(endpoint_a);

                    comp_ctx.push_port(endpoint_b);

                }

                ApplicationJob::NewConnector(connector, initial_ports) => {

                    comp_ctx.push_component(connector, initial_ports);

                },

                ApplicationJob::SyncRound(mut description) => {

                    // Entering sync mode

                    if description.is_empty() {

                        // To simplify logic we always have one instruction

                        description.push(ApplicationSyncAction::Noop);

                    }

                    self.sync_desc = description;

                    self.is_in_sync = true;

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

                    let first_branch_id = self.exec_tree.start_sync();

                    self.exec_tree.push_into_queue(QueueKind::Runnable, first_branch_id);

                    self.consensus.start_sync(ctx);

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

                    return ConnectorScheduling::Immediate;

                },

                ApplicationJob::Shutdown => {

                    debug_assert!(queue.is_empty());

                    return ConnectorScheduling::Exit;

                }

            }

        }

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

// ApplicationInterface

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

#[derive(Debug)]

pub enum ChannelCreationError {

    InSync,

}

#[derive(Debug)]

pub enum ApplicationStartSyncError {

    AlreadyInSync,

    NoSyncActions,

    IncorrectPortKind,

    UnownedPort,

}

#[derive(Debug)]

pub enum ApplicationEndSyncError {

    NotInSync,

}

pub enum ApplicationSyncAction {

    Put(PortIdLocal, ValueGroup),

    Get(PortIdLocal),

}

    is_in_sync: bool,

    owned_ports: Vec<(PortKind, PortIdLocal)>,

            is_in_sync: false,

    /// Creates a new channel. Can only fail if the application interface is

    /// currently in sync mode.

    pub fn create_channel(&mut self) -> Result<Channel, ChannelCreationError> {

        if self.is_in_sync {

            return Err(ChannelCreationError::InSync);

        }

        self.owned_ports.push((PortKind::Putter, putter_id));

        self.owned_ports.push((PortKind::Getter, getter_id));

        return Ok(Channel{ putter_id, getter_id });

        if self.is_in_sync {

            return Err(ComponentCreationError::InSync);

        }

    /// Queues up a description of a synchronous round to run. Will not actually

    /// run the synchronous behaviour in blocking fashion. The results *must* be

    /// retrieved using `try_wait` or `wait` for the interface to be considered

    /// in non-sync mode.

    pub fn perform_sync_round(&mut self, actions: Vec<ApplicationSyncAction>) -> Result<(), ApplicationStartSyncError> {

        if self.is_in_sync {

            return Err(ApplicationStartSyncError::AlreadyInSync);

        }

        // Check the action ports for consistency

        for action in &actions {

            let (port_id, expected_kind) = match action {

                ApplicationSyncAction::Put(port_id, _) => (*port_id, PortKind::Putter),

                ApplicationSyncAction::Get(port_id) => (*port_id, PortKind::Getter),

            };

            match self.find_port_by_id(port_id) {

                Some(port_kind) => {

                    if port_kind != expected_kind {

                        return Err(ApplicationStartSyncError::IncorrectPortKind)

                    }

                },

                None => {

                    return Err(ApplicationStartSyncError::UnownedPort);

                }

            }

        }

        // Everything is consistent, go into sync mode and send the actions off

        // to the component that will actually perform the sync round

        self.is_in_sync = true;

        {

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

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

            *lock = false;

        }

        {

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

            lock.push_back(ApplicationJob::SyncRound(actions));

        }

        self.wake_up_connector_with_ping();

        return Ok(())

    pub fn wait(&self) -> Result<Vec<ValueGroup>, ApplicationEndSyncError> {

        if !self.is_in_sync {

            return Err(ApplicationEndSyncError::NotInSync);

        }

    fn find_port_by_id(&self, port_id: PortIdLocal) -> Option<PortKind> {

        return self.owned_ports.iter()

            .find(|(_, owned_id)| *owned_id == port_id)

            .map(|(port_kind, _)| *port_kind);

    }