// Did not add it ourselves

                if encountered.push(presence.owner_a) {

                    // Not yet sent a message

                    let message = SyncCompMessage{

                        sync_header: self.create_sync_header(ctx),

                        target_component_id: presence.owner_a,

                        content: SyncCompContent::GlobalFailure,

                    };

                    ctx.submit_message(Message::SyncComp(message)).unwrap(); // unwrap: sending to component instead of through channel

                }

            } else if let Some(owner_b) = presence.owner_b {

                if owner_b != ctx.id {

                    if encountered.push(owner_b) {

                        let message = SyncCompMessage{

                            sync_header: self.create_sync_header(ctx),

                            target_component_id: owner_b,

                            content: SyncCompContent::GlobalFailure,

                        };

                        ctx.submit_message(Message::SyncComp(message)).unwrap();

                    }

                }

            }

        }

        self.conclusion = Some(RoundConclusion::Failure);

        if encountered.is_empty() {

            // We don't have to wait on Acks

            return Some(RoundConclusion::Failure);

        } else {

            self.ack_remaining = encountered.len() as u32;

            return None;

        }

    }

    fn initiate_sync_failure(&mut self, ctx: &mut ComponentCtx) -> Option<RoundConclusion> {

        debug_assert!(self.is_in_sync());

        // Notify leader of our channels and the fact that we just failed

        let channel_mapping = &self.branch_annotations[0].channel_mapping;

        let mut channel_presence = Vec::with_capacity(channel_mapping.len());

        for mapping in channel_mapping {

            let port = ctx.get_port_by_channel_id(mapping.channel_id).unwrap();

            channel_presence.push(LocalChannelPresence{

                channel_id: mapping.channel_id,

                is_closed: port.state == PortState::Closed,

            });

        }

        let maybe_already = self.send_to_leader_or_handle_as_leader(SyncCompContent::Presence(ComponentPresence{

#[cfg(test)] mod tests;

mod connector;

// 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 connector::{ConnectorPDL, ConnectorPublic, ConnectorScheduling};

use scheduler::{Scheduler, ComponentCtx, SchedulerCtx, ControlMessageHandler};

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

use inbox::Message;

use port::{ChannelId, Port, PortState};

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

    pub fn downcast(&self) -> ConnectorId {

    }

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

    }

}

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

/// may hold this

impl ConnectorId {

    // TODO: Like the other `new_invalid`, maybe remove

    #[inline]

    pub fn new_invalid() -> ConnectorId {

    }

    #[inline]

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

    }

}

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

//  `router` of `ScheduledConnector` back into `Connector`. The reason I don't

//  want `Box<dyn Connector>` everywhere is because of the v-table overhead. But

//  to truly design this properly I need some benchmarks.

pub(crate) enum ConnectorVariant {

    UserDefined(ConnectorPDL),

    Native(Box<dyn Connector>),

}

impl Connector for ConnectorVariant {

    fn run(&mut self, scheduler_ctx: SchedulerCtx, comp_ctx: &mut ComponentCtx) -> ConnectorScheduling {

        match self {

            ConnectorVariant::UserDefined(c) => c.run(scheduler_ctx, comp_ctx),

            ConnectorVariant::Native(c) => c.run(scheduler_ctx, comp_ctx),

        }

    }

}

pub(crate) struct ScheduledConnector {

    pub connector: ConnectorVariant, // access by connector

    pub ctx: ComponentCtx,

    #[inline]

    fn signal_for_shutdown(&self) {

        debug_assert_eq!(self.active_interfaces.load(Ordering::Acquire), 0);

        debug_assert_eq!(self.active_connectors.load(Ordering::Acquire), 0);

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

        let should_signal = self.should_exit

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

            .is_ok();

        if should_signal {

            self.scheduler_notifier.notify_all();

        }

    }

}

// TODO: Come back to this at some point

unsafe impl Send for RuntimeInner {}

unsafe impl Sync for RuntimeInner {}

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

// ConnectorStore

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

struct ConnectorStore {

    // Freelist storage of connectors. Storage should be pointer-stable as

    // someone might be mutating the vector while we're executing one of the

    // connectors.

    free: Vec<usize>,

}

impl ConnectorStore {

    fn with_capacity(capacity: usize) -> Self {

        Self {

            free: Vec::with_capacity(capacity),

        }

    }

    /// Retrieves public part of connector - accessible by many threads at once.

    fn get_public(&self, id: ConnectorId) -> &'static ConnectorPublic {

        unsafe {

        }

    }

    /// Retrieves private part of connector - accessible by one thread at a

    /// time.

    fn get_private(&self, key: &ConnectorKey) -> &'static mut ScheduledConnector {

        unsafe {

        }

    }

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

            ctx: ComponentCtx::new_empty(),

            public: ConnectorPublic::new(initially_sleeping),

            router: ControlMessageHandler::new(),

            shutting_down: false,

        };

        let index;

        let key;

        if self.free.is_empty() {

            // No free entries, allocate new entry

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

        } else {

            // Free spot available

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

            unsafe {

            }

        }

        println!("DEBUG [ global store  ] Created component at {}", key.index);

        return key;

    }

    /// Destroys a connector. Caller should make sure it is not scheduled for

    /// execution. Otherwise one experiences "bad stuff" (tm).

    fn destroy(&mut self, key: ConnectorKey) {

        unsafe {

            std::ptr::drop_in_place(*target);

            // Note: but not deallocating!

        }

        println!("DEBUG [ global store  ] Destroyed component at {}", key.index);

        self.free.push(key.index as usize);

    }

}

impl Drop for ConnectorStore {

    fn drop(&mut self) {

        // Everything in the freelist already had its destructor called, so only

        // has to be deallocated

        for free_idx in self.free.iter().copied() {

            unsafe {

                let layout = std::alloc::Layout::for_value(&**memory);

                std::alloc::dealloc(*memory as *mut u8, layout);

                // mark as null for the remainder

                *memory = std::ptr::null_mut();

            }

        }

        // With the deallocated stuff marked as null, clear the remainder that

        // is not null

            unsafe {

                if !memory.is_null() {

                    let _ = Box::from_raw(memory); // take care of deallocation, bit dirty, but meh

                }

            }

        }

    }

}

                            .iter()

                            .position(|v| v.self_id == port_change.port.self_id)

                            .unwrap();

                        scheduled.ctx.ports.remove(index);

                    }

                }

            }

        }

        // Finally, check if we just entered or just left a sync region

        if scheduled.ctx.changed_in_sync {

            if scheduled.ctx.is_in_sync {

                // Just entered sync region

            } else {

                // Just left sync region. So prepare inbox for the next sync

                // round

                scheduled.ctx.inbox.clear_read_messages();

            }

            scheduled.ctx.changed_in_sync = false; // reset flag

        }

    }

    fn try_go_to_sleep(&self, connector_key: ConnectorKey, connector: &mut ScheduledConnector) {

        debug_assert_eq!(connector.public.sleeping.load(Ordering::Acquire), false);

        // This is the running connector, and only the running connector may

        // decide it wants to sleep again.

        connector.public.sleeping.store(true, Ordering::Release);

        // But due to reordering we might have received messages from peers who

        // did not consider us sleeping. If so, then we wake ourselves again.

        if !connector.public.inbox.is_empty() {

            // Try to wake ourselves up (needed because someone might be trying

            // the exact same atomic compare-and-swap at this point in time)

            let should_wake_up_again = connector.public.sleeping

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

                .is_ok();

            if should_wake_up_again {

                self.runtime.push_work(connector_key)

            }

        }

    }

    // TODO: Remove, this is debugging stuff

    fn debug(&self, message: &str) {

        println!("DEBUG [thrd:{:02} conn:  ]: {}", self.scheduler_id, message);

    }

    fn debug_conn(&self, conn: ConnectorId, message: &str) {

    }

}

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

// ComponentCtx

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

enum ComponentStateChange {

    CreatedComponent(ConnectorPDL, Vec<PortIdLocal>),

    CreatedPort(Port),

    ChangedPort(ComponentPortChange),

}

#[derive(Clone)]

pub(crate) struct ComponentPortChange {

    pub is_acquired: bool, // otherwise: released

    pub port: Port,

}

/// The component context (better name may be invented). This was created

/// because part of the component's state is managed by the scheduler, and part

/// of it by the component itself. When the component starts a sync block or

/// exits a sync block the partially managed state by both component and

/// scheduler need to be exchanged.

            workspace_branches: Vec::new(),

        };

    }

    /// Notify the runtime that the component has created a new component. May

    /// only be called outside of a sync block.

    pub(crate) fn push_component(&mut self, component: ConnectorPDL, initial_ports: Vec<PortIdLocal>) {

        debug_assert!(!self.is_in_sync);

        self.state_changes.push_back(ComponentStateChange::CreatedComponent(component, initial_ports));

    }

    /// Notify the runtime that the component has created a new port. May only

    /// be called outside of a sync block (for ports received during a sync

    /// block, pass them when calling `notify_sync_end`).

    pub(crate) fn push_port(&mut self, port: Port) {

        debug_assert!(!self.is_in_sync);

        self.state_changes.push_back(ComponentStateChange::CreatedPort(port))

    }

    /// Notify the runtime of an error. Note that this will not perform any

    /// special action beyond printing the error. The component is responsible

    /// for waiting until it is appropriate to shut down (i.e. being outside

    /// of a sync region) and returning the `Exit` scheduling code.

    pub(crate) fn push_error(&mut self, error: EvalError) {

    }

    #[inline]

    pub(crate) fn get_ports(&self) -> &[Port] {

        return self.ports.as_slice();

    }

    pub(crate) fn get_port_by_id(&self, id: PortIdLocal) -> Option<&Port> {

        return self.ports.iter().find(|v| v.self_id == id);

    }

    pub(crate) fn get_port_by_channel_id(&self, id: ChannelId) -> Option<&Port> {

        return self.ports.iter().find(|v| v.channel_id == id);

    }

    fn get_port_mut_by_id(&mut self, id: PortIdLocal) -> Option<&mut Port> {

        return self.ports.iter_mut().find(|v| v.self_id == id);

    }

    /// Notify that component will enter a sync block. Note that after calling

    /// this function you must allow the scheduler to pick up the changes in the

    /// context by exiting your code-executing loop, and to continue executing

    /// code the next time the scheduler picks up the component.

    pub(crate) fn notify_sync_start(&mut self) {