//  entirety. Find some way to interface with the parameter's types.

    pub(crate) fn new_component_v2(

        &self, module_name: &[u8], identifier: &[u8], arguments: ValueGroup

    ) -> Result<ComponentState, ComponentCreationError> {

        // Find the module in which the definition can be found

        let module_root = self.lookup_module_root(module_name);

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

        if !definition.is_component() {

            return Err(ComponentCreationError::DefinitionNotComponent);

        }

        // Make sure that the types of the provided value group matches that of

        // the expected types.

        // - for each argument try to make sure the types match

        for arg_idx in 0..arguments.values.len() {

            let expected_type = &expr_data.arg_types[arg_idx];

            let provided_value = &arguments.values[arg_idx];

            if !self.verify_same_type(expected_type, 0, &arguments, provided_value) {

                return Err(ComponentCreationError::InvalidArgumentType(arg_idx));

            }

        }

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

                Some(name) => if name.as_bytes() == module_name {

                    return Some(module.root_id);

                },

                None => if module_name.is_empty() {

                    return Some(module.root_id);

                }

            }

        }

        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;

                        }

                    }

                } else {

                    return false;

                }

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

                    let heap_pos = *heap_pos;

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

                        if !self.verify_same_type(expected, expected_idx + 1, arguments, element) {

                            return false;

                        }

                    }

                    return true;

                } else {

                    return false;

                }

            },

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

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

                return false;

            },

        }

    }

}

// TODO: @temp Should just become a concrete thing that is passed in

pub trait RunContext {

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

    fn get(&mut self, port: PortId) -> Option<ValueGroup>; // None if still waiting on message

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

/// yet receive anything from any branch).

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

pub(crate) struct BranchId {

    pub index: u32,

}

    fn new(index: u32) -> Self {

        debug_assert!(index != 0);

        Self{ index }

    }

    #[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

    RunningInSync,          // running within a sync block

    HaltedAtBranchPoint,    // at a branching point (at a `get` call)

    ReachedSyncEnd,         // reached end of sync block, branch represents a local solution

    Inconsistent,           // branch can never represent a local solution, so halted

}

        self.is_assigned = true;

        self.num_times_fired = num_times_fired;

    }

    #[inline]

    fn mark_definitive(&mut self, branch_id: BranchId, num_times_fired: u32) {

        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 {

    UsedInInteraction(PortIdLocal),

    AlreadyGivenAway(PortIdLocal)

}

/// Contains a description of the port mapping during a particular sync session.

/// TODO: Extend documentation

        debug_assert!(self.port_mapping.len() == self.owned_ports.len()); // in non-sync mode

        let port_index = self.get_port_index(port_id).unwrap();

        self.owned_ports.remove(port_index);

        self.port_mapping.remove(port_index);

    }

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

    }

    /// Retrieves the ID associated with the port at the provided index

    #[inline]

    fn get_port_id(&self, port_index: usize) -> PortIdLocal {

        return self.owned_ports[port_index];

    }

    #[inline]

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

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

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

    }

    // Function for internal use: retrieve index in flattened port mapping array

    // based on branch/port index.

    #[inline]

    fn mapped_index(&self, branch_idx: u32, port_idx: usize) -> usize {

}

// TODO: Maybe prevent false sharing by aligning `public` to next cache line.

// TODO: Do this outside of the connector, create a wrapping struct

pub(crate) struct ConnectorPDL {

    // State and properties of connector itself

    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,

}

struct TempCtx {}

impl RunContext for TempCtx {

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

        todo!()

    }

        match message {

            MC::Data(message) => self.handle_data_message(message),

            MC::Sync(message) => self.handle_sync_message(message, ctx, delta_state),

            MC::RequestCommit(message) => self.handle_request_commit_message(message, ctx, delta_state),

            MC::ConfirmCommit(message) => self.handle_confirm_commit_message(message, ctx, delta_state),

            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;

                if self.sync_finished_last_handled == 0 {

                    next_id = self.sync_finished.first;

                } else {

                    let last_handled = &self.branches[self.sync_finished_last_handled as usize];

                    debug_assert!(last_handled.next_branch_in_queue.is_some()); // because "last handled" != "last in queue"

                    next_id = last_handled.next_branch_in_queue.unwrap();

                }

                loop {

                    let branch_id = BranchId::new(next_id);

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

                    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;

                    }

                    match branch_next {

                        Some(id) => next_id = id,

                        None => break,

                    }

                }

                self.sync_finished_last_handled = next_id;

            }

            return scheduling;

        } else {

            return scheduling;

        }

    }

}

impl ConnectorPDL {

    /// Constructs a representation of a connector. The assumption is that the

    /// initial branch is at the first instruction of the connector's code,

    /// hence is in a non-sync state.

    pub fn new(initial_branch: Branch, owned_ports: Vec<PortIdLocal>) -> Self {

        Self{

            in_sync: false,

    // Handling connector messages

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

    #[inline]

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

        self.inbox.insert_message(message);

    }

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

            let constraints_index = message.constraints

                .iter()

                .position(|v| v.connector_id == ctx.id)

                .unwrap();

            let constraints = &message.constraints[constraints_index].constraints;

            debug_assert!(!constraints.is_empty());

                                continue 'branch_loop;

                            }

                        },

                        SyncBranchConstraint::PortMapping(port_id, expected_branch_id) => {

                            let port_index = self.ports.get_port_index(*port_id);

                            if port_index.is_none() {

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

                            }

                        },

                    }

                }

                // If here, then all of the external constraints were satisfied

                // for the current branch. But the branch itself also imposes

                // constraints. So while building up the new solution, make sure

                // that those are satisfied as well.

                // TODO: Code below can probably be merged with initial solution

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

                    let constraint = if mapping.num_times_fired == 0 {

                        SyncBranchConstraint::SilentPort(peer_port_id)

                    } else {

                        if peer_is_getter {

                            SyncBranchConstraint::PortMapping(peer_port_id, mapping.last_registered_branch_id)

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

                // Consider the next branch

                if branch_index == self.sync_finished_last_handled {

                    // At the end of the previously handled solutions

                    break;

                }

                debug_assert!(branch.next_branch_in_queue.is_some()); // because we cannot be at the end of the queue

                branch_index = branch.next_branch_in_queue.unwrap();

            }

        }

    }

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

            self.committed_to = Some((message.connector_origin, message.comparison_number));

            if message.to_visit.is_empty() {

                // Visited all of the connectors, so every connector can now

                        // Valid if speculatively firing

                        port_mapping.num_times_fired == 1

                    }

                } else {

                    // Not yet assigned, do so now

                    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,

                    };

                    // If the message contains any ports then we release our

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

        }

        // Not immediately scheduling, so schedule again if there are more

        // branches to run

        if self.sync_active.is_empty() {

            return ConnectorScheduling::NotNow;

        }

    }

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

    // Internal helpers

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

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

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

    ) {

        debug_assert!(to_delete.is_valid()); // we're deleting a valid item

        debug_assert!(queue.first != 0 && queue.last != 0); // queue isn't empty to begin with

        // Retrieve branch and its next element

            // We must own the port, or something is wrong with our code

            todo!("Set up some kind of message router");

            debug_assert!(ports.get_port_index(*port_id).is_some());

            ports.remove_port(*port_id);

        }

        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

                    // already and not used to put/get data.

                    let port_mapping = ports.get_port(branch.index.index, port_index);

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

                        }

                    }

                    branch.ports_delta.push(PortOwnershipDelta{

                        acquired: false,

                        port_id: *port_id,

                    });

                },

                    debug_assert!(to_delete_index != -1);

                    branch.ports_delta.remove(to_delete_index as usize);

                }

            }

        }

        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 {

                        // Somehow already received this port.

                        // TODO: @security

                        todo!("take care of nefarious peers");

                    } else {

                        // Sending ports to ourselves

                        debug_assert!(ports.get_port_index(*port_id).is_some());

        };

        let mut sync_message = SyncMessage::new(initial_local_solution, approximate_peers);

        // Turn local port mapping into constraints on other connectors

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

                return None;

            }

        }

        // - constraints on other components due to owned ports

        for port_index in 0..self.ports.num_ports() {

            let port = ctx.get_port(port_id);

            let constraint = if port_mapping.is_assigned {

                if port.kind == PortKind::Getter {

                    SyncBranchConstraint::BranchNumber(port_mapping.last_registered_branch_id)

                } else {

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

                }

            } else {

                SyncBranchConstraint::SilentPort(port.peer_id)

            };

                return None;

            }

        }

        return Some(sync_message);

    }

    fn submit_sync_solution(&mut self, partial_solution: SyncMessage, ctx: &ConnectorCtx, results: &mut RunDeltaState) {

        if partial_solution.to_visit.is_empty() {

            // Solution is completely consistent. So ask everyone to commit

            // TODO: Maybe another package for random?

            let comparison_number: u64 = unsafe {

    }

}

/// A data structure passed to a connector whose code is being executed that is

/// used to queue up various state changes that have to be applied after

/// running, e.g. the messages the have to be transferred to other connectors.

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

}

/// Recursively goes through the value group, attempting to find ports.

/// Duplicates will only be added once.

pub(crate) fn find_ports_in_value_group(value_group: &ValueGroup, ports: &mut Vec<PortIdLocal>) {

    // Helper to check a value for a port and recurse if needed.

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

            },

            Value::Array(heap_pos) |

            Value::Message(heap_pos) |

            Value::String(heap_pos) |

            Value::Struct(heap_pos) |

            Value::Union(_, heap_pos) => {

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 {

    pub index: u32, // of connector

}

impl ConnectorKey {

    /// Downcasts the `ConnectorKey` type, which can be used to obtain mutable

    /// access, to a "regular ID" which can be used to obtain immutable access.

    }

    /// Turns the `ConnectorId` into a `ConnectorKey`, marked as unsafe as it

    /// bypasses the type-enforced `ConnectorKey`/`ConnectorId` system

    #[inline]

    pub unsafe fn from_id(id: ConnectorId) -> 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 {

        return ConnectorId(u32::MAX);

    }

    #[inline]

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

        return self.0 != u32::MAX;

                free: Vec::with_capacity(capacity),

            }),

        };

    }

    /// Retrieves the shared members of the connector.

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