UnownedPort,

use crate::runtime2::inbox::{PrivateInbox, PublicInbox, OutgoingMessage, DataMessage, SyncMessage};

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

            received: HashMap::new(),

            received: parent_branch.received.clone(),

pub(crate) struct ConnectorPDL {

    sync_finished_last_handled: u32,

impl ConnectorPDL {

            sync_finished_last_handled: 0, // none at all

    pub fn insert_sync_message(&mut self, message: SyncMessage, results: &mut RunDeltaState) {

    }

    pub fn run(&mut self, pd: &ProtocolDescription, results: &mut RunDeltaState) -> ConnectorScheduling {

        if self.in_sync {

            let scheduling = self.run_in_speculative_mode(pd, results);

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

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

                }

                // Transform branch into proposed global solution

            }

        } else {

            let scheduling = self.run_in_deterministic_mode(pd, results);

            return scheduling;

        }

    }

                    let mut did_have_messages = false;

                    for message in messages {

                        did_have_messages = true;

                        // For each message prepare a new branch to execute

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

                        let mut new_branch = Branch::new_sync_branching_from(new_branch_index, branch);

                        self.ports.prepare_sync_branch(branch.index.index, new_branch_index);

                        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;

                        debug_assert!(port_mapping.is_assigned && port_mapping.num_times_fired == 1);

                        new_branch.received.insert(local_port_id, message.clone());

                        // If the message contains any ports then they will now

                        // be owned by the new branch

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

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

                        Self::acquire_ports_during_sync(&mut self.ports, &mut new_branch, &results.ports);

                        results.ports.clear();

                        // Schedule the new branch

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

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

                        self.branches.push(new_branch);

                    }

                    if did_have_messages {

                        // If we did create any new branches, then we can run

                        // them immediately.

                // assigned and map them to non-firing.

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

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

    fn push_branch_into_queue(

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

    ) {

    // Helpers for generating and handling sync messages (and the solutions that

    // are described by those sync messages)

    fn generate_initial_solution_for_branch(&self, branch_id: BranchId,)

    pub new_connectors: Vec<ConnectorPDL>,

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

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

// TODO: Change this, I hate this. But I also don't want to put `public` and

//  `router` of `ScheduledConnector` back into `Connector`.

pub enum ConnectorVariant {

    UserDefined(ConnectorPDL),

    Native(Box<dyn Connector>),

}

    pub connector: ConnectorVariant,

    pub(crate) fn create(&self, connector: ConnectorVariant) -> ConnectorKey {

    pub(crate) fn create_channel(&self, creating_connector: ConnectorId) -> Channel {

        fn reserve_port(lock: &mut std::sync::RwLockWriteGuard<'_, PortStoreInner>, kind: PortKind, creating_connector: ConnectorId) -> u32 {

                    ownership: PortOwnership::Owned,

                port.ownership = PortOwnership::Owned;

        return Channel{

            putter_id: PortIdLocal::new(putter_id),

            getter_id: PortIdLocal::new(getter_id),

        }

pub enum SyncBranchConstraint {

    SilentPort(PortIdLocal),

    BranchNumber(u32),

    PortMapping(PortIdLocal, u32),

}

pub struct SyncConnectorSolution {

    connector_id: ConnectorId,

    terminating_branch_id: BranchId,

    execution_branch_ids: Vec<BranchId>, // ends with terminating branch ID

}

pub struct SyncConnectorConstraints {

    connector_id: ConnectorId,

    constraints: Vec<SyncBranchConstraint>,

}

pub struct SyncMessage {

    connector_solutions: Vec<SyncConnectorSolution>,

    connector_constraints: Vec<SyncConnectorConstraints>,

    connectors_to_visit: Vec<u32>,

}

    Data(DataMessage),          // data message, handled by connector

    Sync(SyncMessage),          // sync message, handled by both connector/scheduler

    Control(ControlMessage),    // control message, handled by scheduler

    Ping,                       // ping message, intentionally waking up a connector (used for native connectors)

mod native;

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

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

use global_store::{ConnectorVariant, GlobalStore};

use connector::{Branch, ConnectorPDL, find_ports_in_value_group};

use native::{ConnectorApplication, ApplicationInterface};

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

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

//  - Running schedulers: schedulers need to be signaled that they should exit, then wait until all are done

//  - User-owned interfaces: As long as these are owned user may still decide to create new connectors.

    inner: Arc<RuntimeInner>,

}

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

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

        let runtime_inner = Arc::new(RuntimeInner{

            global_store: GlobalStore::new(),

            protocol_description,

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

        });

        {

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

            for _ in 0..num_threads {

                let mut scheduler = Scheduler::new(runtime_inner.clone());

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

                    scheduler.run();

                });

                schedulers.push(thread);

            }

            let mut lock = runtime_inner.schedulers.lock().unwrap();

            *lock = schedulers;

        // Return runtime

        return Runtime{ inner: runtime_inner };

    /// Returns a new interface through which channels and connectors can be

    /// created.

    pub fn create_interface(&self) -> ApplicationInterface {

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

        let connector = Box::new(connector);

        let connector_key = self.global_store.connectors.create(ConnectorVariant::Native(connector));

        interface.set_connector_id(connector_key.downcast());

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

        // it is actually needed.

        return interface;

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

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

        for scheduler in schedulers.drain(..) {

            scheduler.join();

        }

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

use std::cell::Cell;

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::global_store::ConnectorKey;

use super::RuntimeInner;

use super::global_store::{ConnectorVariant, ConnectorId};

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

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

use super::inbox::{Message, DataMessage, SyncMessage};

pub trait Connector {

    fn insert_data_message(&mut self, message: DataMessage);

    fn insert_sync_message(&mut self, message: SyncMessage, delta_state: &mut RunDeltaState);

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

}

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

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

enum ApplicationJob {

    NewConnector(ConnectorPDL),

}

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

/// up the next synchronous round, and retrieve the data afterwards.

pub struct ConnectorApplication {

    sync_done: SyncDone,

    job_queue: JobQueue,

}

impl ConnectorApplication {

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

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

        let job_queue = Arc::new(Mutex::new(Vec::with_capacity(32)));

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

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

        return (connector, interface);

    }

}

impl Connector for ConnectorApplication {

    fn insert_sync_message(&mut self, message: SyncMessage, delta_state: &mut RunDeltaState) {

        todo!("handling sync messages in ApplicationConnector");

    }

    fn insert_data_message(&mut self, message: DataMessage)  {

        todo!("handling messages in ApplicationConnector");

    }

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

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

                }

            }

        }

        return ConnectorScheduling::NotNow;

    }

}

/// The interface to a `ApplicationConnector`. This allows setting up the

/// interactions the `ApplicationConnector` performs within a synchronous round.

pub struct ApplicationInterface {

    sync_done: SyncDone,

    job_queue: JobQueue,

    runtime: Arc<RuntimeInner>,

    connector_id: ConnectorId,

    owned_ports: Vec<PortIdLocal>,

}

impl ApplicationInterface {

    pub(crate) fn new(sync_done: SyncDone, job_queue: JobQueue, runtime: Arc<RuntimeInner1>) -> 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 {

        let channel = self.runtime.global_store.ports.create_channel(self.connector_id);

        self.owned_ports.push(channel.putter_id);

        self.owned_ports.push(channel.getter_id);

        return channel;

    }

    /// Creates a new connector. Note that it is not scheduled immediately, but

    /// depends on the `ApplicationConnector` to run, followed by the created

    /// connector being scheduled.

    // TODO: Optimize by yanking out scheduler logic for common use.

    pub fn create_connector(&mut self, module: &str, routine: &str, arguments: ValueGroup) -> Result<(), ComponentCreationError> {

        // Retrieve ports and make sure that we own the ones that are currently

        // specified. This is also checked by the scheduler, but that is done

        // asynchronously.

        let mut initial_ports = Vec::new();

        find_ports_in_value_group(&arguments, &mut initial_ports);

        for port_to_remove in &initial_ports {

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

                Some(index_to_remove) => {

                    // We own the port, so continue

                    self.owned_ports.remove(index_to_remove)

                },

                None => {

                    // We don't own the port

                    return Err(ComponentCreationError::UnownedPort);

                }

            }

        }

        let state = self.runtime.protocol_description.new_component_v2(module.as_bytes(), routine.as_bytes(), arguments)?;

        let connector = ConnectorPDL::new(0, Branch::new_initial_branch(state), initial_ports);

        // Put on job queue

        {

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

            queue.push(ApplicationJob::NewConnector(connector));

        }

        // Send ping message to wake up connector

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

        connector.inbox.insert_message(Message::Ping);

        let should_wake_up = connector.sleeping

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

            .is_ok();

        if should_wake_up {

            let key = unsafe{ ConnectorKey::from_id(self.connector_id) };

            self.runtime.global_store.connector_queue.push_back(key);

        }

        return Ok(());

    }

    /// Check if the next sync-round is finished.

    pub fn try_wait(&self) -> bool {

        let (is_done, _) = &*self.sync_done;

        let lock = is_done.lock().unwrap();

        return *lock;

    }

    /// Wait until the next sync-round is finished

    pub fn wait(&self) {

        let (is_done, condition) = &*self.sync_done;

        let lock = is_done.lock().unwrap();

        condition.wait_while(lock, |v| !*v); // wait while not done

    }

    /// Called by runtime to set associated connector's ID.

    pub(crate) fn set_connector_id(&mut self, id: ConnectorId) {

        self.connector_id = id;

    }

}

    pub putter_id: PortIdLocal, // can put on it, so from the connector's point of view, this is an output

    pub getter_id: PortIdLocal, // vice versa: can get on it, so an input for the connector

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

use super::RuntimeInner;

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

    runtime: Arc<RuntimeInner>,

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

        return Self{ runtime };

            let connector_key = self.runtime.global_store.connector_queue.pop_front();

                if self.runtime.global_store.should_exit.load(Ordering::Acquire) {

            let scheduled = self.runtime.global_store.connectors.get_mut(&connector_key);

                                    let port = self.runtime.global_store.ports.get(&connector_key, port_id);

                        },

                        Message::Ping => {},

                // TODO: Revise

                match &mut scheduled.connector {

                    ConnectorVariant::UserDefined(connector) => {

                        if connector.is_in_sync_mode() {

                            // In synchronous mode, so we can expect messages being sent,

                            // but we never expect the creation of connectors

                            new_schedule = connector.run_in_speculative_mode(&self.runtime.protocol_description, &mut delta_state);

                            debug_assert!(delta_state.new_connectors.is_empty());

                        } else {

                            // In regular running mode (not in a sync block) we cannot send

                            // messages but we can create new connectors

                            new_schedule = connector.run_in_deterministic_mode(&self.runtime.protocol_description, &mut delta_state);

                            debug_assert!(delta_state.outbox.is_empty());

                        }

                    },

                    ConnectorVariant::Native(connector) => {

                        new_schedule = connector.run(&self.runtime.protocol_description);

                    },

                    self.runtime.global_store.connector_queue.push_back(connector_key);

                            self.runtime.global_store.connector_queue.push_back(connector_key);

                    let sending_port = self.runtime.global_store.ports.get(&connector_key, message.sending_port);

            let cur_connector = self.runtime.global_store.connectors.get_mut(connector_key);

                let new_key = self.runtime.global_store.connectors.create(ConnectorVariant::UserDefined(new_connector));

                let new_connector = self.runtime.global_store.connectors.get_mut(&new_key);

                        let mut port = self.runtime.global_store.ports.get(connector_key, *port_id);