#[inline]

    pub fn clear(&mut self) {

        self.inner.clear();

    }

    #[inline]

        return self.inner.iter();

    }

    #[inline]

    pub fn is_empty(&self) -> bool {

        self.inner.is_empty()

    }

    /// Collapses the speculative execution tree back into a deterministic one,

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

        self.branches.truncate(1);

        // Reset all values to non-sync defaults

        let branch = &mut self.branches[0];

        branch.id = BranchId::new_invalid();

        branch.parent_id = BranchId::new_invalid();

        branch.sync_state = SpeculativeState::RunningNonSync;

        branch.next_in_queue = BranchId::new_invalid();

        // Clear out all the queues

        for queue_idx in 0..NUM_QUEUES {

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

        }

    }

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

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

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

use super::branch::{Branch, BranchId, ExecTree, QueueKind, SpeculativeState};

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

use super::native::Connector;

use super::port::PortIdLocal;

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

pub(crate) struct ConnectorPublic {

    pub inbox: PublicInbox,

    consensus: &'a Consensus,

    received: &'a HashMap<PortIdLocal, ValueGroup>,

    scheduler: SchedulerCtx<'a>,

    prepared_channel: Option<(Value, Value)>,

}

    fn did_put(&mut self, port: PortId) -> bool {

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

        let annotation = self.consensus.get_annotation(self.branch_id, port_id);

        return annotation.registered_id.is_some();

    }

impl Connector for ConnectorPDL {

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

        self.handle_new_messages(comp_ctx);

        if self.tree.is_in_sync() {

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

            return scheduling;

        } else {

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

            return scheduling;

        }

    }

    // --- Handling messages

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

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

            match message {

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

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

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

    }

    // --- Running code

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

        // Check if we have any branch that needs running

        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{

            consensus: &self.consensus,

            received: &branch.inbox,

            scheduler: *sched_ctx,

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

                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

                return ConnectorScheduling::Immediate;

            },

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

        }

    }

}

use crate::collections::VecSet;

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

struct BranchAnnotation {

    port_mapping: Vec<PortAnnotation>,

}

pub(crate) struct LocalSolution {

    final_branch_id: BranchId,

    port_mapping: Vec<(ChannelId, BranchId)>,

}

#[derive(Clone)]

pub(crate) struct GlobalSolution {

}

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

// Consensus

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

        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.encountered_peers.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{

            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

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

    }

    /// Generates sync messages for any branches that are at the end of the

    /// sync block. To find these branches, they should've been put in the

    /// "finished" queue in the execution tree.

        debug_assert!(self.is_in_sync());

        let mut last_branch_id = self.last_finished_handled;

        for branch in tree.iter_queue(QueueKind::FinishedSync, last_branch_id) {

            // Turn the port mapping into a local solution

            let source_mapping = &self.branch_annotations[branch.id.index as usize].port_mapping;

            let local_solution = LocalSolution{

                component: ctx.id,

                final_branch_id: branch.id,

                port_mapping: target_mapping,

            };

            last_branch_id = Some(branch.id);

        }

        self.last_finished_handled = last_branch_id;

    }

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

        debug_assert!(self.is_in_sync());

        // TODO: Handle sending and receiving ports

        final_ports.clear();

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

        for port in &branch.port_mapping {

            final_ports.push(port.port_id);

        }

    }

    // --- Handling messages

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

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

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

            },

            SyncContent::GlobalSolution(solution) => {

                // Take branch of interest and return it.

                    .find(|(connector_id, _)| connector_id == ctx.id)

                    .unwrap();

                return Some(*branch_id);

            }

        }

    }

                content: SyncContent::Notification

            };

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

        } // else: exactly equal, so do nothing

    }

        if self.highest_connector_id == ctx.id {

            // We are the leader

            if let Some(global_solution) = self.solution_combiner.add_solution_and_check_for_global_solution(solution) {

                    let message = SyncMessageFancy{

                        sync_header: self.create_sync_header(ctx),

                        target_component_id: connector_id,

                        content: SyncContent::GlobalSolution(global_solution.clone()),

                    };

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

                }

            }

        } else {

            // Someone else is the leader

            let message = SyncMessageFancy{

                sync_header: self.create_sync_header(ctx),

                target_component_id: self.highest_connector_id,

                content: SyncContent::LocalSolution(solution),

            };

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

        }

    }

    #[inline]

    fn create_sync_header(&self, ctx: &ComponentCtxFancy) -> SyncHeader {

        return SyncHeader{

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

// Solution storage and algorithms

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

struct MatchedLocalSolution {

    final_branch_id: BranchId,

    matches: Vec<ComponentMatches>,

}

struct ComponentMatches {

    target_id: ConnectorId,

    target_index: usize,

    /// new local solutions for matching against already stored local solutions

    /// of peer connectors.

    fn add_solution_and_check_for_global_solution(&mut self, solution: LocalSolution) -> Option<GlobalSolution> {

        let component_id = solution.component;

        let solution = MatchedLocalSolution{

            final_branch_id: solution.final_branch_id,

            matches: Vec::new(),

        };

        // Create an entry for the solution for the particular component

        let component_exists = self.local.iter_mut()

            .enumerate()

        };

        // If this is a solution of a component that is new to us, then we check

        // in the stored solutions which other components are peers of the new

        // one.

        if new_component {

            let mut component_peers = Vec::new();

            // Find the matching components

            for (other_index, other_component) in self.local.iter().enumerate() {

                if other_index == component_index {

                    // Don't match against ourselves

                    continue;

                }

                            // We have a shared port

                        }

                    }

                }

                    // We share some ports

                    component_peers.push(ComponentPeer{

                        target_id: other_component.component,

                        target_index: other_index,

                    });

                }

            }

            let mut num_ports_in_peers = 0;

            for peer in component_peers {

                let mut num_ports_in_peers = component_match.involved_channels.len();

                let other_component = &mut self.local[component_match.target_index];

                for existing_peer in &other_component.peers {

                    num_ports_in_peers += existing_peer.involved_channels.len();

                }

                    other_component.all_peers_present = true;

                }

                other_component.peers.push(ComponentPeer{

                    target_id: component_id,

                    target_index: component_index,

            let other_component = &self.local[peer.target_index];

            for (other_solution_index, other_solution) in other_component.solutions.iter().enumerate() {

                // Check the port mappings between the pair of solutions.

                let mut all_matched = true;

                        if cur_port == other_port {

                            if cur_branch == other_branch {

                                // Same port mapping, go to next port

                                break;

                            } else {

                                // Different port mapping, not a match

        }

        // Because of our earlier checking if we can have a solution at

        // all (all components have their peers), and the exit condition of the

        // while loop: if we're here, then we have a global solution

        debug_assert_eq!(check_stack.len(), self.local.len());

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

            let component = &self.local[component_index];

            let solution = &component.solutions[solution_index];

        }

        // Just debugging here, TODO: @remove

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

            let component = &self.local[component_index];

        }

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

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

                    Some((_, encountered_branch_id)) => {

                        debug_assert_eq!(encountered_branch_id, branch_id);

                        total_num_checked += 1;

                    },

                    None => {

                        final_mapping.push((channel_id, branch_id));

                    }

                }

            }

        }

        return Some(GlobalSolution{

        });

    }

    /// Simple test if a solution is at all possible. If this returns true it

    /// does not mean there actually is a solution.

    fn can_have_solution(&self) -> bool {

        let mut solutions = Vec::with_capacity(reserve_len);

        for component in self.local.drain(..) {

            for solution in component.solutions {

                solutions.push(LocalSolution{

                    component: component.component,

                    final_branch_id: solution.final_branch_id,

                });

            }

        }

        return solutions;

    }

}

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

// Generic Helpers

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

#[derive(Debug)]

pub struct Message {

    pub sending_connector: ConnectorId,

    pub receiving_port: PortIdLocal, // may be invalid (in case of messages targeted at the connector)

    pub contents: MessageContents,

}

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

use crate::runtime2::branch::BranchId;

use crate::runtime2::ConnectorId;

use crate::runtime2::consensus::{GlobalSolution, LocalSolution};

use crate::runtime2::port::PortIdLocal;

#[derive(Debug)]

pub(crate) enum MessageFancy {

    Data(DataMessageFancy),

    Sync(SyncMessageFancy),

    Control(ControlMessageFancy),

}

// Structure of module

mod runtime;

mod messages;

mod branch;

mod native;

mod port;

mod scheduler;

mod consensus;

mod inbox2;

#[cfg(test)] mod tests;

mod connector2;

use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};

use std::thread::{self, JoinHandle};

use crate::collections::RawVec;

use crate::ProtocolDescription;

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

use native::{Connector, ConnectorApplication, ApplicationInterface};

/// A kind of token that, once obtained, allows mutable access to a connector.

/// We're trying to use move semantics as much as possible: the owner of this

/// key is the only one that may execute the connector's code.

pub(crate) struct ConnectorKey {

    pub index: u32, // of connector

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

use crate::protocol::ComponentCreationError;

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

use super::{ConnectorKey, ConnectorId, RuntimeInner};

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

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

type SyncDone = Arc<(Mutex<bool>, Condvar)>;

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

enum ApplicationJob {

    NewChannel((Port, Port)),

    Shutdown,

}

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

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

pub struct ConnectorApplication {

impl Connector for ConnectorApplication {

    fn run(&mut self, _sched_ctx: SchedulerCtx, comp_ctx: &mut ComponentCtxFancy) -> ConnectorScheduling {

        // Handle any incoming messages if we're participating in a round

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

            match message {

            }

        }

        // Handle requests coming from the API

        {

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

                match job {

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

                        println!("DEBUG: API adopting ports");

                        comp_ctx.push_port(endpoint_a);

                        comp_ctx.push_port(endpoint_b);

                    }

                        println!("DEBUG: API creating connector");

                    },

                    ApplicationJob::Shutdown => {

                        debug_assert!(queue.is_empty());

                        return ConnectorScheduling::Exit;

                    }

                }

    pub fn create_connector(&mut self, module: &str, routine: &str, arguments: ValueGroup) -> Result<(), ComponentCreationError> {

        // Retrieve ports and make sure that we own the ones that are currently

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

                return Err(ComponentCreationError::UnownedPort);

            }

        }

        }

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

        // Put on job queue

        {

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

        }

        self.wake_up_connector_with_ping();

        return Ok(());

    }

    pub(crate) fn set_connector_id(&mut self, id: ConnectorId) {

        self.connector_id = id;

    }

    fn wake_up_connector_with_ping(&self) {

        let connector = self.runtime.get_component_public(self.connector_id);