if module_root.is_none() {

            return Err(ComponentCreationError::ModuleDoesntExist);

        }

        let module_root = module_root.unwrap();

        let root = &self.heap[module_root];

        if definition_id.is_none() {

            return Err(ComponentCreationError::DefinitionDoesntExist);

        }

        let definition_id = definition_id.unwrap();

        let definition = &self.heap[definition_id];

            }

        }

        // By now we're sure that all of the arguments are correct. So create

        // the connector.

        return Ok(ComponentState{

        });

    }

    fn lookup_module_root(&self, module_name: &[u8]) -> Option<RootId> {

        for module in self.modules.iter() {

            match &module.name {

        return None;

    }

    fn verify_same_type(&self, expected: &ConcreteType, expected_idx: usize, arguments: &ValueGroup, argument: &Value) -> bool {

        use ConcreteTypePart as CTP;

        match &expected.parts[expected_idx] {

            CTP::Void | CTP::Message | CTP::Slice | CTP::Function(_, _) | CTP::Component(_, _) => unreachable!(),

            CTP::String => {

                // Match outer string type and embedded character types

                if let Value::String(heap_pos) = argument {

                    for element in &arguments.regions[*heap_pos as usize] {

                        if let Value::Char(_) = element {} else {

                            return false;

                    }

                    return true;

                } else {

                    return false;

                }

            },

            CTP::Instance(_definition_id, _num_embedded) => {

                todo!("implement full type checking on user-supplied arguments");

                return false;

            },

        }

    }

use std::collections::HashMap;

use std::sync::atomic::AtomicBool;

use crate::{PortId, ProtocolDescription};

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

use crate::protocol::eval::{Prompt, Value, ValueGroup};

use crate::runtime2::native::Connector;

use crate::runtime2::scheduler::ConnectorCtx;

use super::global_store::ConnectorId;

use super::inbox::{

    SyncBranchConstraint, SyncConnectorSolution

};

use super::port::PortIdLocal;

/// Represents the identifier of a branch (the index within its container). An

/// ID of `0` generally means "no branch" (e.g. no parent, or a port did not

    #[inline]

    pub(crate) fn is_valid(&self) -> bool {

        return self.index != 0;

    }

}

pub(crate) enum SpeculativeState {

    // Non-synchronous variants

    RunningNonSync,         // regular execution of code

    Error,                  // encountered a runtime error

    Finished,               // finished executing connector's code

    // Synchronous variants

        self.is_assigned = true;

        self.last_registered_branch_id = branch_id;

        self.num_times_fired = num_times_fired;

    }

}

struct PortOwnershipDelta {

    acquired: bool, // if false, then released ownership

    port_id: PortIdLocal,

}

enum PortOwnershipError {

    /// Retrieves the index associated with a port id. Note that the port might

    /// not exist (yet) if a speculative branch has just received the port.

    /// TODO: But then again, one cannot use that port, right?

    #[inline]

    fn get_port_index(&self, port_id: PortIdLocal) -> Option<usize> {

        for (idx, port) in self.owned_ports.iter().enumerate() {

                return Some(idx)

            }

        }

        return None

    }

        return &self.port_mapping[mapped_idx];

    }

    #[inline]

    fn get_port_mut(&mut self, branch_idx: u32, port_idx: usize) -> &mut PortAssignment {

        let mapped_idx = self.mapped_index(branch_idx, port_idx);

    }

    #[inline]

    fn num_ports(&self) -> usize {

        return self.owned_ports.len();

    }

    in_sync: bool,

    // Branch management

    branches: Vec<Branch>, // first branch is always non-speculative one

    sync_active: BranchQueue,

    sync_pending_get: BranchQueue,

    sync_finished: BranchQueue,

    cur_round: u32,

    // Port/message management

    pub committed_to: Option<(ConnectorId, u64)>,

    pub inbox: PrivateInbox,

    pub ports: ConnectorPorts,

}

            MC::Control(_) | MC::Ping => {},

        }

    }

    fn run(&mut self, pd: &ProtocolDescription, ctx: &ConnectorCtx, delta_state: &mut RunDeltaState) -> ConnectorScheduling {

        if self.in_sync {

            // When in speculative mode we might have generated new sync

            // solutions, we need to turn them into proposed solutions here.

            if self.sync_finished_last_handled != self.sync_finished.last {

                // Retrieve first element in queue

                let mut next_id;

                    let branch_next = branch.next_branch_in_queue;

                    // Turn local solution into a message and send it along

                    // TODO: Like `ports` access, also revise the construction of this `key`, should not be needed

                    let solution_message = self.generate_initial_solution_for_branch(branch_id, ctx);

                    if let Some(valid_solution) = solution_message {

                    } else {

                        // Branch is actually invalid, but we only just figured

                        // it out. We need to mark it as invalid to prevent

                        // future use

                        Self::remove_branch_from_queue(&mut self.branches, &mut self.sync_finished, branch_id);

                            self.sync_finished_last_handled = self.sync_finished.last;

                        }

                        let branch = &mut self.branches[next_id as usize];

                        branch.sync_state = SpeculativeState::Inconsistent;

                    }

                self.sync_finished_last_handled = next_id;

            }

            return scheduling;

        } else {

            return scheduling;

        }

    }

}

impl ConnectorPDL {

    }

    /// Accepts a synchronous message and combines it with the locally stored

    /// solution(s). Then queue new `Sync`/`Solution` messages when appropriate.

    pub fn handle_sync_message(&mut self, message: SyncMessage, ctx: &ConnectorCtx, results: &mut RunDeltaState) {

        debug_assert!(!message.to_visit.contains(&ctx.id)); // own ID already removed

        // TODO: Optimize, use some kind of temp workspace vector

        let mut execution_path_branch_ids = Vec::new();

        if self.sync_finished_last_handled != 0 {

            // We have some solutions to match against

                                // Nefarious peer

                                continue 'branch_loop;

                            }

                            let port_index = port_index.unwrap();

                            let mapping = self.ports.get_port(branch_index, port_index);

                                // Not the expected interaction on this port, constraint not satisfied

                                continue 'branch_loop;

                            }

                        },

                    }

                }

                        } else {

                            SyncBranchConstraint::BranchNumber(mapping.last_registered_branch_id)

                        }

                    };

                    match new_solution.add_or_check_constraint(peer_connector_id, constraint) {

                    }

                }

                // If here, then the newly generated solution is completely

                // compatible.

                self.submit_sync_solution(new_solution, ctx, results);

    fn handle_request_commit_message(&mut self, mut message: SolutionMessage, ctx: &ConnectorCtx, delta_state: &mut RunDeltaState) {

        let should_propagate_message = match &self.committed_to {

            Some((previous_origin, previous_comparison)) => {

                // Already committed to something. So will commit to this if it

                // takes precedence over the current solution

                message.comparison_number > *previous_comparison ||

            },

            None => {

                // Not yet committed to a solution, so commit to this one

                true

            }

        };

                };

                if is_valid_put {

                    // Put in run results for thread to pick up and transfer to

                    // the correct connector inbox.

                    port_mapping.mark_definitive(branch.index, 1);

                        sending_port: local_port_id,

                        sender_prev_branch_id: BranchId::new_invalid(),

                        sender_cur_branch_id: branch.index,

                        message: value_group,

                    };

                    // ownership over them in this branch

                    debug_assert!(results.ports.is_empty());

                    find_ports_in_value_group(&message.message, &mut results.ports);

                    Self::release_ports_during_sync(&mut self.ports, branch, &results.ports);

                    results.ports.clear();

                    return ConnectorScheduling::Immediate

                } else {

                    branch.sync_state = SpeculativeState::Inconsistent;

                }

            },

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

    // Helpers for management of the branches and their internally stored

    // `next_branch_in_queue` and the `BranchQueue` objects. Essentially forming

    // linked lists inside of the vector of branches.

    /// Pops from front of linked-list branch queue.

        debug_assert!(queue.first != 0);

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

        branch.next_branch_in_queue = None;

            // No more entries in queue

        }

        return branch;

    }

    /// Pushes branch at the end of the linked-list branch queue.

    fn push_branch_into_queue(

        branches: &mut Vec<Branch>, queue: &mut BranchQueue, to_push: BranchId,

    ) {

        debug_assert!(to_push.is_valid());

        let to_push = to_push.index;

            // No branches in the queue at all

            branches[to_push as usize].next_branch_in_queue = None;

        } else {

            // Pre-existing branch in the queue

        }

    }

    /// Removes branch from linked-list queue. Walks through the entire list to

    /// find the element (!). Assumption is that this is not called often.

    fn remove_branch_from_queue(

        return Ok(())

    }

    /// Releasing ownership of ports during a sync-session. Will provide an

    /// error if the port was already used during a sync block.

    fn release_ports_during_sync(ports: &mut ConnectorPorts, branch: &mut Branch, port_ids: &[PortIdLocal]) -> Result<(), PortOwnershipError> {

        debug_assert!(branch.index.is_valid()); // branch in sync mode

        for port_id in port_ids {

            match ports.get_port_index(*port_id) {

                Some(port_index) => {

                    // We (used to) own the port. Make sure it is not given away

                    if port_mapping.is_assigned && port_mapping.num_times_fired != 0 {

                        // Already used

                        return Err(PortOwnershipError::UsedInInteraction(*port_id));

                    }

                    for delta in &branch.ports_delta {

                            // We cannot have acquired this port, because the

                            // call to `ports.get_port_index` returned an index.

                            debug_assert!(!delta.acquired);

                            return Err(PortOwnershipError::AlreadyGivenAway(*port_id));

                        }

                    }

        return Ok(())

    }

    /// Acquiring ports during a sync-session.

    fn acquire_ports_during_sync(ports: &mut ConnectorPorts, branch: &mut Branch, port_ids: &[PortIdLocal]) -> Result<(), PortOwnershipError> {

        debug_assert!(branch.index.is_valid()); // branch in sync mode

        'port_loop: for port_id in port_ids {

            for (delta_idx, delta) in branch.ports_delta.iter().enumerate() {

                if delta.port_id == *port_id {

                    if delta.acquired {

        // - constraints on other components due to transferred ports

        for port_delta in &branch.ports_delta {

            // For transferred ports we always have two constraints: one for the

            // sender and one for the receiver, ensuring it was not used.

            // TODO: This will fail if a port is passed around multiple times.

            //  maybe a special "passed along" entry in `ports_delta`.

                return None;

            }

            // Might need to check if we own the other side of the channel

            let port = ctx.get_port(port_delta.port_id);

            if !sync_message.add_or_check_constraint(port.peer_connector, SyncBranchConstraint::SilentPort(port.peer_id)).unwrap() {

                    SyncBranchConstraint::PortMapping(port.peer_id, port_mapping.last_registered_branch_id)

                }

            } else {

                SyncBranchConstraint::SilentPort(port.peer_id)

            };

                return None;

            }

        }

        return Some(sync_message);

    }

// TODO: Come up with a better name

pub(crate) struct RunDeltaState {

    // Variables that allow the thread running the connector to pick up global

    // state changes and try to apply them.

    pub outbox: Vec<MessageContents>,

    pub new_connectors: Vec<ConnectorPDL>,

    // Workspaces

    pub ports: Vec<PortIdLocal>,

}

impl RunDeltaState {

    /// Constructs a new `RunDeltaState` object with the default amount of

    /// reserved memory

    pub fn new() -> Self {

        RunDeltaState{

            outbox: Vec::with_capacity(64),

            new_connectors: Vec::new(),

            ports: Vec::with_capacity(64),

        }

    }

}

pub(crate) enum ConnectorScheduling {

    Immediate,      // Run again, immediately

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

    NotNow,         // Do not reschedule for running

}

    fn find_port_in_value(group: &ValueGroup, value: &Value, ports: &mut Vec<PortIdLocal>) {

        match value {

            Value::Input(port_id) | Value::Output(port_id) => {

                // This is an actual port

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

                for prev_port in ports.iter() {

                        // Already added

                        return;

                    }

                }

                ports.push(cur_port);

use std::ptr;

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

use crate::collections::{MpmcQueue, RawVec};

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

use super::scheduler::Router;

use crate::ProtocolDescription;

use crate::runtime2::connector::{ConnectorScheduling, RunDeltaState};

use crate::runtime2::scheduler::ConnectorCtx;

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

        return ConnectorKey{ index: id.0 };

    }

}

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

/// may hold this

pub(crate) struct ConnectorId(pub u32);

impl ConnectorId {

    // TODO: Like the other `new_invalid`, maybe remove

    #[inline]

    pub fn new_invalid() -> ConnectorId {

    pub(crate) fn get_shared(&self, connector_id: ConnectorId) -> &'static ConnectorPublic {

        let lock = self.inner.read().unwrap();

        unsafe {

            let connector = lock.connectors.get(connector_id.0 as usize);

            debug_assert!(!connector.is_null());

        }

    }

    /// Retrieves a particular connector. Only the thread that pulled the

    /// associated key out of the execution queue should (be able to) call this.

    pub(crate) fn get_mut(&self, key: &ConnectorKey) -> &'static mut ScheduledConnector {

        let lock = self.inner.read().unwrap();

        unsafe {

            let connector = lock.connectors.get_mut(key.index as usize);

            debug_assert!(!connector.is_null());

        }

    }

    /// Create a new connector, returning the key that can be used to retrieve

    /// and/or queue it. The caller must make sure that the constructed

    /// connector's code is initialized with the same ports as the ports in the

    /// `initial_ports` array.

        // Creation of the connector in the global store, requires a lock

        {

            let lock = self.inner.write().unwrap();

            let connector = ScheduledConnector {

                connector,

                context: ConnectorCtx::new(self.port_counter.clone()),

                public: ConnectorPublic::new(),

                router: Router::new(),

            };

            if lock.free.is_empty() {

                let connector = Box::into_raw(Box::new(connector));

                unsafe {

                    // Cheating a bit here. Anyway, move to heap, store in list

                    index = lock.connectors.len();

                    debug_assert!(!target.is_null());

                    ptr::write(*target, connector);

                }

            }

        }

        let key = ConnectorKey{ index: index as u32 };

        let new_connector = self.get_mut(&key);

        new_connector.context.id = key.downcast();

        return key;

    }

}

impl Drop for ConnectorStore {

    fn drop(&mut self) {

        let lock = self.inner.write().unwrap();

The `PrivateInbox` is a temporary storage for all messages that are received

within a certain sync-round.

**/

use std::collections::VecDeque;

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

use super::global_store::ConnectorId;

/// A message that has been delivered (after being imbued with the receiving

/// port by the scheduler) to a connector.

#[derive(Clone)]

pub struct DataMessage {

            .find(|v| v.connector_id == connector_id)

            .unwrap();

        match constraint {

            SyncBranchConstraint::SilentPort(silent_port_id) => {

                for (port_id, mapped_id) in &entry.final_port_mapping {

                        // If silent, then mapped ID is invalid

                        return Ok(!mapped_id.is_valid())

                    }

                }

                return Err(());

                return Err(());

            },

        }

    }

}

pub struct SolutionMessage {

    pub comparison_number: u64,

    pub connector_origin: ConnectorId,

    pub local_solutions: Vec<(ConnectorId, BranchId)>,

    pub to_visit: Vec<ConnectorId>,

}

/// A control message. These might be sent by the scheduler to notify eachother

/// of asynchronous state changes.

pub struct ControlMessage {

    pub id: u32, // generic identifier, used to match request to response

    pub content: ControlMessageVariant,

}

pub enum ControlMessageVariant {

    ChangePortPeer(PortIdLocal, ConnectorId), // specified port has a new peer, sent to owner of said port

    Ack, // acknowledgement of previous control message, matching occurs through control message ID.

}

/// Generic message contents.

    /// (prev_branch_id, cur_branch_id, receiving_port_id) already exists, then

    /// nothing is inserted..

    pub fn insert_message(&mut self, message: DataMessage) {

        for existing in self.messages.iter() {

            if existing.sender_prev_branch_id == message.sender_prev_branch_id &&

                    existing.sender_cur_branch_id == message.sender_cur_branch_id &&

                // Message was already received

                return;

            }

        }

        self.messages.push(message);

    next_index: usize,

    max_index: usize,

    match_port_id: PortIdLocal,

    match_prev_branch_id: BranchId,

}

        // Loop until match is found or at end of messages

        while self.next_index < self.max_index {

            let cur_message = &self.messages[self.next_index];

            if cur_message.receiving_port == self.match_port_id && cur_message.sender_prev_branch_id == self.match_prev_branch_id {

                // Found a match

                break;

use std::collections::hash_map::Entry;

use std::collections::HashMap;

use crate::PortId;

use crate::protocol::*;

use crate::protocol::eval::*;

/// A message residing in a connector's inbox (waiting to be put into some kind

/// of speculative branch), or a message waiting to be sent.

#[derive(Clone)]

pub struct BufferedMessage {

    pub(crate) sending_port: PortId,

    pub(crate) receiving_port: PortId,

// Imports

use std::sync::{Arc, Mutex};

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

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

use global_store::{ConnectorVariant, GlobalStore};

use scheduler::Scheduler;

use native::{ConnectorApplication, ApplicationInterface};

// Runtime API

// TODO: Exit condition is very dirty. Take into account:

//  - Connector hack with &'static references. May only destroy (unforced) if all connectors are done working

pub(crate) struct RuntimeInner {

    pub(crate) global_store: GlobalStore,

    pub(crate) protocol_description: ProtocolDescription,

    schedulers: Mutex<Vec<JoinHandle<()>>>, // TODO: Revise, make exit condition something like: all interfaces dropped

}

impl Runtime {

    pub fn new(num_threads: usize, protocol_description: ProtocolDescription) -> Runtime {

        // Setup global state

        assert!(num_threads > 0, "need a thread to run connectors");

        let runtime_inner = Arc::new(RuntimeInner{

            global_store: GlobalStore::new(),

        });

        // Launch threads

        {

            let mut schedulers = Vec::with_capacity(num_threads);

            for _ in 0..num_threads {

                let thread = thread::spawn(move || {

                    scheduler.run();

                });

                schedulers.push(thread);

            }

    }