pub fn with_capacity(capacity: usize) -> Self {

        // Could be done a bit more efficiently

        let mut result = Self::new();

        return result;

    }

    }

    pub fn push(&mut self, item: T) {

        unsafe {

            let target = self.base.add(self.len);

            std::ptr::write(target, item);

                    dealloc(old_base, old_layout);

                }

                self.cap = new_cap;

            }

        } // else: still enough space

            debug_assert!(!self.base.is_null());

            let (_, layout) = self.current_layout();

            unsafe {

                if cfg!(debug_assertions) {

                    self.base = ptr::null_mut();

                }

            let mut breadcrumb = self.type_loop_breadcrumbs[breadcrumb_idx].clone();

            let poly_type = self.lookup.get(&breadcrumb.definition_id).unwrap();

            let resolve_result = match &poly_type.definition {

                DTV::Enum(_) => {

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::port::{Port, PortKind};

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

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

/// yet receive anything from any branch).

#[derive(Clone, Copy, PartialEq, Eq)]

    pub index: u32,

}

    port_id: PortIdLocal,

}

enum PortOwnershipError {

    UsedInInteraction(PortIdLocal),

    AlreadyGivenAway(PortIdLocal)

    // Contains P*B entries, where P is the number of ports and B is the number

    // of branches. One can find the appropriate mapping of port p at branch b

    // at linear index `b*P+p`.

}

impl ConnectorPorts {

        self.port_mapping.reserve(num_ports);

        for offset in 0..num_ports {

            let parent_port = &self.port_mapping[parent_base_idx + offset];

        }

    }

}

impl ConnectorPublic {

        ConnectorPublic{

            inbox: PublicInbox::new(),

        }

    }

}

}

impl Connector for ConnectorPDL {

        use MessageContents as MC;

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

        }

    }

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

    #[inline]

    }

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

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

                // compatible.

                self.submit_sync_solution(new_solution, ctx, results);

                // Consider the next branch

                    break;

                }

            }

        }

    }

        }

    }

        // Make sure this is the message we actually committed to. As long as

        // we're running on a single machine this is fine.

        // TODO: Take care of nefarious peers

    /// where it is the caller's responsibility to immediately take care of

    /// those changes. The return value indicates when (and if) the connector

    /// needs to be scheduled again.

        debug_assert!(self.in_sync);

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

                let local_port_index = self.ports.get_port_index(local_port_id).unwrap();

                debug_assert!(self.ports.owned_ports.contains(&local_port_id));

                silent_port.mark_speculative(0);

                // Run both branches again

                self.branches.push(firing_branch);

                return ConnectorScheduling::Immediate;

            },

                if is_valid_get {

                    // Mark as a branching point for future messages

                    branch.sync_state = SpeculativeState::HaltedAtBranchPoint;

                    // But if some messages can be immediately applied, do so

                    // now.

                        did_have_messages = true;

                        // For each message prepare a new branch to execute

                        let new_branch_index = self.branches.len() as u32;

                        let port_mapping = self.ports.get_port_mut(new_branch_index, local_port_index);

                        port_mapping.last_registered_branch_id = message.sender_cur_branch_id;

                        // Schedule the new branch

                        debug_assert!(new_branch.sync_state == SpeculativeState::RunningInSync);

                        self.branches.push(new_branch);

                    }

                    if did_have_messages {

                    }

                }

                branch.sync_state = SpeculativeState::ReachedSyncEnd;

            },

            RunResult::BranchPut(port_id, value_group) => {

                // Branch performed a `put` on a particualar port.

                    // ownership over them in this branch

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

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

                    results.ports.clear();

                    results.outbox.push(MessageContents::Data(message));

    }

    /// Runs the connector in non-synchronous mode.

        debug_assert!(!self.in_sync);

        debug_assert!(self.sync_active.is_empty() && self.sync_pending_get.is_empty() && self.sync_finished.is_empty());

        debug_assert!(self.branches.len() == 1);

                // Prepare for sync execution and reschedule immediately

                self.in_sync = true;

                let first_sync_branch = Branch::new_sync_branching_from(1, branch);

                self.branches.push(first_sync_branch);

                return ConnectorScheduling::Later;

            },

            }

            prev_index = next_index;

        }

        // If here, then we didn't find the element

            // TODO: Maybe another package for random?

            let comparison_number: u64 = unsafe {

                let mut random_array = [0u8; 8];

                std::mem::transmute(random_array)

            };

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

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

use crate::ProtocolDescription;

/// The registry containing all connectors. The idea here is that when someone

/// owns a `ConnectorKey`, then one has unique access to that connector.

    pub(crate) fn create_interface(&self, connector: ConnectorApplication) -> ConnectorKey {

        // Connector interface does not own any initial ports, and cannot be

        // created by another connector

        return key;

    }

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

    pub(crate) fn create_pdl(&self, created_by: &mut ScheduledConnector, connector: ConnectorPDL) -> ConnectorKey {

        let new_connector = self.get_mut(&key);

        // Transferring ownership of ports (and crashing if there is a

    }

    pub(crate) fn destroy(&self, key: ConnectorKey) {

        unsafe {

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

    }

    /// Creates a connector but does not set its initial ports

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

        let index;

        {

            let connector = ScheduledConnector {

                connector,

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

                router: Router::new(),

            };

            if lock.free.is_empty() {

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

                index = lock.connectors.len();

                lock.connectors.push(connector);

            } else {

                index = lock.free.pop().unwrap();

/// TODO: @docs

/// TODO: @Optimize, very lazy implementation of concurrent datastructures.

///     This includes the `should_exit` and `did_exit` pair!

    pub connector_queue: MpmcQueue<ConnectorKey>,

    pub connectors: ConnectorStore,

    pub should_exit: AtomicBool,    // signal threads to exit

}

impl GlobalStore {

        Self{

            connector_queue: MpmcQueue::with_capacity(256),

            connectors: ConnectorStore::with_capacity(256),

    }

}

    // "Normal" messages, intended for a PDL protocol. These need to stick

    // around during an entire sync-block (to handle `put`s for which the

    // corresponding `get`s have not yet been reached).

    len_read: usize,

}

    /// Will insert the message into the inbox. Only exception is when the tuple

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

    /// nothing is inserted..

            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;

            }

        }

    }

    /// Retrieves all previously read messages that satisfy the provided

    /// This function should only be used to check if already-received messages

    /// could be received by a newly encountered `get` call in a connector's

    /// PDL code.

        return InboxMessageIter{

            messages: &self.messages,

            next_index: 0,

    /// Retrieves the next unread message. Should only be called by the

    /// inbox-reader.

        if self.len_read == self.messages.len() {

            return None;

        }

        self.len_read += 1;

        return Some(to_return);

    }

    /// Simply empties the inbox

        self.messages.clear();

        self.len_read = 0;

    }

}

/// Iterator over previously received messages in the inbox.

    next_index: usize,

    max_index: usize,

    match_port_id: PortIdLocal,

    fn next(&mut self) -> Option<Self::Item> {

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

        while self.next_index < self.max_index {

                // Found a match

                break;

            }

            return None;

        }

        self.next_index += 1;

        return Some(message);

    }

use std::collections::HashMap;

use crate::PortId;

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

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

// Imports

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

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

use crate::ProtocolDescription;

use native::{ConnectorApplication, ApplicationInterface};

pub struct Runtime {

    inner: Arc<RuntimeInner>,

}

pub(crate) struct RuntimeInner {

    pub(crate) protocol_description: ProtocolDescription,

}

// TODO: Come back to this at some point

unsafe impl Sync for RuntimeInner {}

impl Runtime {

        // Setup global state

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

        let runtime_inner = Arc::new(RuntimeInner{

            global_store: GlobalStore::new(),

            protocol_description,

            schedulers: Mutex::new(Vec::new()),

        });

        // Launch threads

        {

                let cloned_runtime_inner = runtime_inner.clone();

                        scheduler.run();

                schedulers.push(thread);

            }

        self.inner.global_store.should_exit.store(true, Ordering::Release);

        let mut schedulers = self.inner.schedulers.lock().unwrap();

        for scheduler in schedulers.drain(..) {

        }

    }

}

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

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

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

use crate::ProtocolDescription;

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

use super::inbox::Message;

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

    /// Handle a new message (preprocessed by the scheduler). You probably only

    /// want to handle `Data`, `Sync`, and `Solution` messages. The others are

    /// intended for the scheduler itself.

    /// Should run the connector's behaviour up until the next blocking point.

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

}

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

enum ApplicationJob {

    NewChannel((Port, Port)),

impl ConnectorApplication {

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

        let sync_done = Arc::new(( Mutex::new(false), Condvar::new() ));

        let connector = ConnectorApplication { sync_done: sync_done.clone(), job_queue: job_queue.clone() };

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

}

impl Connector for ConnectorApplication {

    }

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

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

            match job {

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

                    delta_state.new_ports.reserve(2);

                    delta_state.new_ports.push(endpoint_a);

                    delta_state.new_ports.push(endpoint_b);

                }

                ApplicationJob::NewConnector(connector) => {

                    delta_state.new_connectors.push(connector);

                }

            }

}

impl ApplicationInterface {

        return Self{

            sync_done, job_queue, runtime,

            connector_id: ConnectorId::new_invalid(),

        {

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

        }