next: u32,

}

/// Identifier of a component in a session

#[derive(Copy, Clone, Eq, PartialEq, Ord, Hash, PartialOrd, serde::Serialize, serde::Deserialize)]

pub struct ComponentId(Id); // PUB because it can be returned by errors

/// Identifier of a port in a session

#[derive(Copy, Clone, Eq, PartialEq, Ord, Hash, PartialOrd, serde::Serialize, serde::Deserialize)]

#[repr(transparent)]

pub struct PortId(pub(crate) Id);

/// A safely aliasable heap-allocated payload of message bytes

#[derive(Default, Eq, PartialEq, Clone, Ord, PartialOrd)]

pub struct Payload(pub Arc<Vec<u8>>);

#[derive(Debug, Eq, PartialEq, Clone, Hash, Copy, Ord, PartialOrd)]

/// "Orientation" of a port, determining whether they can send or receive messages with `put` and `get` respectively.

#[repr(C)]

#[derive(serde::Serialize, serde::Deserialize)]

pub enum Polarity {

    Putter, // output port (from the perspective of the component)

    Getter, // input port (from the perspective of the component)

}

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

}

pub(crate) struct Branch {

    index: BranchId,

    parent_index: BranchId,

    // Code execution state

    code_state: ComponentState,

    sync_state: SpeculativeState,

    next_branch_in_queue: Option<u32>,

    // Message/port state

    received: HashMap<PortIdLocal, DataMessage>, // TODO: @temporary, remove together with fires()

    ports_delta: Vec<PortOwnershipDelta>,

}

impl Branch {

    /// Constructs a non-sync branch. It is assumed that the code is at the

    /// first instruction

    pub(crate) fn new_initial_branch(component_state: ComponentState) -> Self {

        Branch{

            index: BranchId::new_invalid(),

            parent_index: BranchId::new_invalid(),

            code_state: component_state,

            sync_state: SpeculativeState::RunningNonSync,

            next_branch_in_queue: None,

            received: HashMap::new(),

            ports_delta: Vec::new(),

        }

    }

    /// Constructs a sync branch. The provided branch is assumed to be the

    /// parent of the new branch within the execution tree.

    fn new_sync_branching_from(new_index: u32, parent_branch: &Branch) -> Self {

        debug_assert!(

            (parent_branch.sync_state == SpeculativeState::RunningNonSync && !parent_branch.parent_index.is_valid()) ||

            (parent_branch.sync_state == SpeculativeState::HaltedAtBranchPoint)

        );

        Branch{

            index: BranchId::new(new_index),

            parent_index: parent_branch.index,

            code_state: parent_branch.code_state.clone(),

            sync_state: SpeculativeState::RunningInSync,

            next_branch_in_queue: None,

            received: parent_branch.received.clone(),

            ports_delta: parent_branch.ports_delta.clone(),

        }

    }

    fn commit_to_sync(&mut self) {

        self.index = BranchId::new(0);

        self.parent_index = BranchId::new_invalid();

        self.sync_state = SpeculativeState::RunningNonSync;

        self.next_branch_in_queue = None;

    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 {

        let branch_idx = branch_idx as usize;

        let num_ports = self.owned_ports.len();

        debug_assert!(port_idx < num_ports);

        debug_assert!((branch_idx + 1) * num_ports <= self.port_mapping.len());

        return branch_idx * num_ports + port_idx;

    }

}

struct BranchQueue {

    first: u32,

    last: u32,

}

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

    sync_active: BranchQueue,

    sync_pending_get: BranchQueue,

    sync_finished: BranchQueue,

    sync_finished_last_handled: u32, // TODO: Change to BranchId?

    cur_round: u32,

    // Port/message management

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

    pub inbox: PrivateInbox,

    pub ports: ConnectorPorts,

}

struct ConnectorRunContext<'a> {

    ports: &'a ConnectorPorts,

    scheduler: SchedulerCtx<'a>,

}

impl<'a> RunContext for ConnectorRunContext<'a> {

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

            // Either acquired or released, must be silent

            return false;

        }

        let port_index = self.ports.get_port_index(PortIdLocal::new(port.0.u32_suffix)).unwrap();

        return mapping.is_assigned;

    }

    fn get(&mut self, port: PortId) -> Option<ValueGroup> {

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

            Some(message) => Some(message.message.clone()),

            None => None,

        }

    }

    fn fires(&mut self, port: PortId) -> Option<Value> {

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

            return None

        }

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

        if mapping.is_assigned {

            return Some(Value::Bool(mapping.num_times_fired != 0));

        } else {

            return None;

        }

    }

    fn get_channel(&mut self) -> Option<(Value, Value)> {

    }

}

impl Connector for ConnectorPDL {

    fn handle_message(&mut self, message: Message, ctx: &ConnectorCtx, delta_state: &mut RunDeltaState) {

        use MessageContents as MC;

        match message.contents {

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

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

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

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

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

        }

    }

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

                    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,

        solution.commit_to_sync();

    }

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

    // Executing connector code

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

    /// Runs the connector in synchronous mode. Potential changes to the global

    /// system's state are added to the `RunDeltaState` object by the connector,

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

        let branch = Self::pop_branch_from_queue(&mut self.branches, &mut self.sync_active);

        // Run the branch to the next blocking point

        // Match statement contains `return` statements only if the particular

        // run result behind handled requires an immediate re-run of the

        // connector.

        match run_result {

            RunResult::BranchInconsistent => {

                // Speculative branch became inconsistent

                branch.sync_state = SpeculativeState::Inconsistent;

            },

            RunResult::BranchMissingPortState(port_id) => {

                // Branch called `fires()` on a port that does not yet have an

                // assigned speculative value. So we need to create those

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

                    results.ports.clear();

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

                    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;

        } else {

            return ConnectorScheduling::Later;

        }

    }

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

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

        debug_assert!(branch.sync_state == SpeculativeState::RunningNonSync);

        let mut run_context = ConnectorRunContext{

            ports: &self.ports,

        };

        match run_result {

            RunResult::ComponentTerminated => {

                // Need to wait until all children are terminated

                // TODO: Think about how to do this?

                branch.sync_state = SpeculativeState::Finished;

                return ConnectorScheduling::Exit;

            },

            RunResult::ComponentAtSyncStart => {

                // Prepare for sync execution and reschedule immediately

                self.in_sync = true;

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

                let first_sync_branch_id = first_sync_branch.index;

                self.branches.push(first_sync_branch);

                Self::push_branch_into_queue(&mut self.branches, &mut self.sync_active, first_sync_branch_id);

                return ConnectorScheduling::Later;

            },

            RunResult::NewComponent(definition_id, monomorph_idx, arguments) => {

                // Construction of a new component. Find all references to ports

                // inside of the arguments

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

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

                if !results.ports.is_empty() {

                    // Ports changing ownership

                    if let Err(_) = Self::release_ports_during_non_sync(&mut self.ports, branch, &results.ports) {

                        todo!("fatal error handling");

                    }

                }

                // Add connector for later execution

                let new_connector_state = ComponentState {

                };

                let new_connector_ports = results.ports.clone(); // TODO: Do something with this

                let new_connector_branch = Branch::new_initial_branch(new_connector_state);

                let new_connector = ConnectorPDL::new(new_connector_branch, new_connector_ports);

                results.new_connectors.push(new_connector);

                return ConnectorScheduling::Later;

            },

            RunResult::NewChannel => {

                // Need to prepare a new channel

            },

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

        }

    }

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

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

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

        todo!("unfinished: add port properties during final solution-commit msgs");

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

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

        todo!("unfinished: add port properties during final solution-commit msgs");

        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

// Imports

use std::collections::VecDeque;

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

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

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

use crate::collections::RawVec;

use crate::ProtocolDescription;

use inbox::Message;

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

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

use crate::runtime2::port::Port;

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

    #[inline]

    UserDefined(ConnectorPDL),

    Native(Box<dyn Connector>),

}

impl Connector for ConnectorVariant {

    fn handle_message(&mut self, message: Message, ctx: &ConnectorCtx, delta_state: &mut RunDeltaState) {

        match self {

            ConnectorVariant::UserDefined(c) => c.handle_message(message, ctx, delta_state),

            ConnectorVariant::Native(c) => c.handle_message(message, ctx, delta_state),

        }

    }

        match self {

        }

    }

}

pub(crate) struct ScheduledConnector {

    pub connector: ConnectorVariant, // access by connector

    pub context: ConnectorCtx, // mutable access by scheduler, immutable by connector

    pub public: ConnectorPublic, // accessible by all schedulers and connectors

}

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

// Runtime

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

/// Externally facing runtime.

pub struct Runtime {

    inner: Arc<RuntimeInner>,

}

impl Runtime {

            lock = self.scheduler_notifier.wait(lock).unwrap();

        }

        return lock.pop_front();

    }

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

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

        lock.push_back(key);

        self.scheduler_notifier.notify_one();

    }

    /// Creates a new port pair. Note that these are stored globally like the

    /// connectors are. Ports stored by components belong to those components.

        use port::{PortIdLocal, PortKind};

        let getter_id = self.port_counter.fetch_add(2, Ordering::SeqCst);

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

        let getter_id = PortIdLocal::new(getter_id);

        let getter_port = Port{

            self_id: getter_id,

            peer_id: putter_id,

            kind: PortKind::Getter,

        };

        let putter_port = Port{

            self_id: putter_id,

            peer_id: getter_id,

            kind: PortKind::Putter,

        };

        return (getter_port, putter_port);

    }

    // --- Creating/retrieving/destroying components

    pub(crate) fn create_interface_component(&self, component: ConnectorApplication) -> ConnectorKey {

        // Initialize as sleeping, as it will be scheduled by the programmer.

        let mut lock = self.connectors.write().unwrap();

        let key = lock.create(ConnectorVariant::Native(Box::new(component)), true);

        self.increment_active_components();

        return key;

    }

    /// Creates a new PDL component. The caller MUST make sure to schedule the

    /// connector.

    // TODO: Nicer code, not forcing the caller to schedule, perhaps?

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

        // Create as not sleeping, as we'll schedule it immediately

        let key = {

            let mut lock = self.connectors.write().unwrap();

        };

        // Transfer the ports

        {

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

            let created = lock.get_private(&key);

            match &created.connector {

                ConnectorVariant::UserDefined(connector) => {

                    for port_id in connector.ports.owned_ports.iter().copied() {

                        println!("DEBUG: Transferring port {:?} from {} to {}", port_id, created_by.context.id.0, key.index);

                        let mut port = created_by.context.remove_port(port_id);

                        created.context.add_port(port);

                    }

                },

                ConnectorVariant::Native(_) => unreachable!(),

            }

        }

        self.increment_active_components();

        return key;

    }

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

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

        return lock.get_private(connector_key);

    }

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

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

        return lock.get_public(connector_id);

    }

    pub(crate) fn destroy_component(&self, connector_key: ConnectorKey) {

        let mut lock = self.connectors.write().unwrap();

        lock.destroy(connector_key);

        self.decrement_active_components();

    }

    // --- Managing exit condition

            debug_assert!(!connector.is_null());

            return &mut (**connector);

        }

    }

    /// Creates a new connector. Caller should ensure ports are set up correctly

    /// and the connector is queued for execution if needed.

    fn create(&mut self, connector: ConnectorVariant, initially_sleeping: bool) -> ConnectorKey {

        let mut connector = ScheduledConnector {

            connector,

            context: ConnectorCtx::new(),

            public: ConnectorPublic::new(initially_sleeping),

        };

        let index;

        let key;

        if self.free.is_empty() {

            // No free entries, allocate new entry

            index = self.connectors.len();

            key = ConnectorKey{ index: index as u32 };

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

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

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

use super::connector::find_ports_in_value_group;

use super::inbox::{Message, MessageContents};

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

pub(crate) trait Connector {

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

    fn handle_message(&mut self, message: Message, ctx: &ConnectorCtx, delta_state: &mut RunDeltaState);

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

}

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

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

enum ApplicationJob {

    NewChannel((Port, Port)),

    NewConnector(ConnectorPDL),

    Shutdown,

}

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

        use MessageContents as MC;

        match message.contents {

            MC::Data(_) => unreachable!("data message in API connector"),

            MC::Sync(_) | MC::RequestCommit(_) | MC::ConfirmCommit(_) => {

                // Handling sync in API

            },

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

            MC::Ping => {},

        }

    }

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

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

            match job {

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

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

                    delta_state.new_ports.reserve(2);

                    delta_state.new_ports.push(endpoint_a);

                    delta_state.new_ports.push(endpoint_b);

                }

                ApplicationJob::NewConnector(connector) => {

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

                    delta_state.new_connectors.push(connector);

impl ApplicationInterface {

    fn new(sync_done: SyncDone, job_queue: JobQueue, runtime: Arc<RuntimeInner>) -> Self {

        return Self{

            sync_done, job_queue, runtime,

            connector_id: ConnectorId::new_invalid(),

            owned_ports: Vec::new(),

        }

    }

    /// Creates a new channel.

    pub fn create_channel(&mut self) -> Channel {

        debug_assert_eq!(getter_port.kind, PortKind::Getter);

        let getter_id = getter_port.self_id;

        let putter_id = putter_port.self_id;

        {

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

            lock.push_back(ApplicationJob::NewChannel((getter_port, putter_port)));

        }

        // Add to owned ports for error checking while creating a connector

        self.owned_ports.reserve(2);

        self.owned_ports.push(putter_id);

    // TODO: Unsure about this, maybe remove, then also remove all struct

    //  instances where I call this

    pub fn new_invalid() -> Self {

        Self{ index: u32::MAX }

    }

    pub fn is_valid(&self) -> bool {

        return self.index != u32::MAX;

    }

}

pub enum PortKind {

    Putter,

    Getter,

}

/// Represents a port inside of the runtime. May be without owner if it is

/// created by the application interfacing with the runtime, instead of being

/// created by a connector.

pub struct Port {

    pub self_id: PortIdLocal,

    pub peer_id: PortIdLocal,

    pub kind: PortKind,

    fn port_id_to_index(&self, id: PortIdLocal) -> usize {

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

            if port.self_id == id {

                return idx;

            }

        }

        panic!("port {:?}, not owned by connector", id);

    }

}

pub(crate) struct SchedulerCtx<'a> {

    pub(crate) runtime: &'a RuntimeInner

}

pub(crate) struct Scheduler {

    runtime: Arc<RuntimeInner>,

    scheduler_id: u32,

}

impl Scheduler {

    pub fn new(runtime: Arc<RuntimeInner>, scheduler_id: u32) -> Self {