let module_root = module_root.unwrap();

        let root = &self.heap[module_root];

        if definition_id.is_none() {

            return Err(ComponentCreationError::DefinitionDoesntExist);

        }

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

        // the connector.

        return Ok(ComponentState{

        });

    }

    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 {

                    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;

    }

}

pub(crate) enum SpeculativeState {

    // Non-synchronous variants

    RunningNonSync,         // regular execution of code

    }

}

struct PortOwnershipDelta {

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

    port_id: PortIdLocal,

    #[inline]

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

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

                return Some(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]

    sync_active: BranchQueue,

    sync_pending_get: BranchQueue,

    sync_finished: BranchQueue,

    cur_round: u32,

    // Port/message management

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

    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.

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

                        }

            return scheduling;

        } else {

            return scheduling;

        }

    }

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

                            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;

                            }

                    };

                    match new_solution.add_or_check_constraint(peer_connector_id, constraint) {

                    }

                }

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

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

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

                    results.ports.clear();

                    return ConnectorScheduling::Immediate

                } else {

                    branch.sync_state = SpeculativeState::Inconsistent;

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

        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

        }

    }

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

                    }

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

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

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

            }

                SyncBranchConstraint::SilentPort(port.peer_id)

            };

                return None;

            }

        }

    // state changes and try to apply them.

    pub outbox: Vec<MessageContents>,

    pub new_connectors: Vec<ConnectorPDL>,

    // Workspaces

    pub ports: Vec<PortIdLocal>,

}

        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

                // This is an actual port

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

                for prev_port in ports.iter() {

                        // Already added

                        return;

                    }

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.

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

/// may hold this

pub(crate) struct ConnectorId(pub u32);

impl ConnectorId {

        unsafe {

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

            debug_assert!(!connector.is_null());

        }

    }

        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 {

                router: Router::new(),

            };

            if lock.free.is_empty() {

                let connector = Box::into_raw(Box::new(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 {

**/

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

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

                    }

    }

}

pub struct SolutionMessage {

    pub comparison_number: u64,

    pub connector_origin: 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.

        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;

            }

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

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

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

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

use global_store::{ConnectorVariant, GlobalStore};

use scheduler::Scheduler;

use native::{ConnectorApplication, ApplicationInterface};

    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

        {

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

            for _ in 0..num_threads {

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

                    scheduler.run();

                });

    /// created.

    pub fn create_interface(&self) -> ApplicationInterface {

        let (connector, mut interface) = ConnectorApplication::new(self.inner.clone());

        interface.set_connector_id(connector_key.downcast());

        // Note that we're not scheduling. That is done by the interface in case

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

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

use crate::protocol::ComponentCreationError;

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

use crate::ProtocolDescription;

use crate::runtime2::connector::{Branch, find_ports_in_value_group};

use crate::runtime2::inbox::MessageContents;

use crate::runtime2::port::{Port, PortKind};

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

use super::RuntimeInner;

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

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

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

pub trait Connector {

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

enum ApplicationJob {

    NewConnector(ConnectorPDL),

}

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

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

            match job {

                ApplicationJob::NewConnector(connector) => {

                    delta_state.new_connectors.push(connector);

                }

    job_queue: JobQueue,

    runtime: Arc<RuntimeInner>,

    connector_id: ConnectorId,

}

impl ApplicationInterface {

        return Self{

            sync_done, job_queue, runtime,

            connector_id: ConnectorId::new_invalid(),

        let putter_id = PortIdLocal::new(getter_id + 1);

        let getter_id = PortIdLocal::new(getter_id);

            self_id: getter_id,

            peer_id: putter_id,

            kind: PortKind::Getter,

            peer_connector: self.connector_id,

            self_id: putter_id,

            peer_id: getter_id,

            kind: PortKind::Putter,

            peer_connector: self.connector_id,

        return Channel{ putter_id, getter_id };

    }

            match self.owned_ports.iter().position(|v| v == port_to_remove) {

                Some(index_to_remove) => {

                    // We own the port, so continue

                },

                None => {

                    // We don't own the port

        // Send ping message to wake up connector

        let connector = self.runtime.global_store.connectors.get_shared(self.connector_id);

        let should_wake_up = connector.sleeping

            .compare_exchange(true, false, Ordering::SeqCst, Ordering::Acquire)

            .is_ok();

use super::global_store::ConnectorId;

pub(crate) struct PortIdLocal {

    pub index: u32,

}

    }

}

pub enum PortKind {

    Putter,

    Getter,

use std::sync::Arc;

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

use std::time::Duration;

use std::thread;

use crate::runtime2::global_store::ConnectorVariant;

use crate::runtime2::inbox::MessageContents;

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

use super::RuntimeInner;

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

/// Contains fields that are mostly managed by the scheduler, but may be

/// accessed by the connector

        Self{

            id: ConnectorId::new_invalid(),

            port_counter,

        }

    }

            while cur_schedule == ConnectorScheduling::Immediate {

                // Check all the message that are in the shared inbox

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

                    }

                        match content.content {

                            ControlMessageVariant::ChangePortPeer(port_id, new_target_connector_id) => {

                                // Need to change port target

                                // connector we just changed. This is because the `ack` will

                                // clear the rerouting entry of the `ack`-receiver.

                                self.send_message_and_wake_up_if_sleeping(

                                    Message{

                                        sending_connector: connector_key.downcast(),

                                        receiving_port: PortIdLocal::new_invalid(),

                                scheduled.router.handle_ack(content.id);

                            }

                        }

                    }

                }

                            let message = Message{

                                sending_connector: connector_id,

                                receiving_port: PortIdLocal::new_invalid(),

                            };

                            self.send_message_and_wake_up_if_sleeping(*to_visit, message);

                        }

            }

        }