if branch.awaiting_port == data_header.target_port {

                // Found a branch awaiting the message, but we need to make sure

                // the mapping is correct

                if self.branch_can_receive(branch.id, data_header, content) {

                    target_ids.push(branch.id);

                }

            }

        }

    }

    fn handle_received_sync_header(&mut self, sync_header: &SyncHeader, ctx: &mut ComponentCtx) {

        debug_assert!(sync_header.sending_component_id != ctx.id); // not sending to ourselves

        self.encountered_peers.push(sync_header.sending_component_id);

        if sync_header.highest_component_id > self.highest_connector_id {

            // Sender has higher component ID. So should be the target of our

            // messages. We should also let all of our peers know

            self.highest_connector_id = sync_header.highest_component_id;

            for encountered_id in self.encountered_peers.iter() {

                if *encountered_id == sync_header.sending_component_id {

                    // Don't need to send it to this one

                    continue

                }

                let message = SyncMessage {

                    sync_header: self.create_sync_header(ctx),

                    target_component_id: *encountered_id,

                    content: SyncContent::Notification,

                };

                ctx.submit_message(Message::Sync(message));

            }

            // But also send our locally combined solution

            self.forward_local_solutions(ctx);

        } else if sync_header.highest_component_id < self.highest_connector_id {

            // Sender has lower leader ID, so it should know about our higher

            // one.

            let message = SyncMessage {

                sync_header: self.create_sync_header(ctx),

                target_component_id: sync_header.sending_component_id,

                content: SyncContent::Notification

            };

            ctx.submit_message(Message::Sync(message));

        } // else: exactly equal, so do nothing

    }

    fn send_or_store_local_solution(&mut self, solution: LocalSolution, ctx: &mut ComponentCtx) -> Option<BranchId> {

        if self.highest_connector_id == ctx.id {

            // We are the leader

            if let Some(global_solution) = self.solution_combiner.add_solution_and_check_for_global_solution(solution) {

                let mut my_final_branch_id = BranchId::new_invalid();

                for (connector_id, branch_id) in global_solution.component_branches.iter().copied() {

                    if connector_id == ctx.id {

                        // This is our solution branch

                        my_final_branch_id = branch_id;

                        continue;

                    }

                    let message = SyncMessage {

                        sync_header: self.create_sync_header(ctx),

                        target_component_id: connector_id,

                        content: SyncContent::GlobalSolution(global_solution.clone()),

                    };

                    ctx.submit_message(Message::Sync(message));

                }

                debug_assert!(my_final_branch_id.is_valid());

                return Some(my_final_branch_id);

            } else {

                return None;

            }

        } else {

            // Someone else is the leader

            let message = SyncMessage {

                sync_header: self.create_sync_header(ctx),

                target_component_id: self.highest_connector_id,

                content: SyncContent::LocalSolution(solution),

            };

            ctx.submit_message(Message::Sync(message));

            return None;

        }

    }

    #[inline]

    fn create_sync_header(&self, ctx: &ComponentCtx) -> SyncHeader {

        return SyncHeader{

            sending_component_id: ctx.id,

            highest_component_id: self.highest_connector_id,

        }

    }

    fn forward_local_solutions(&mut self, ctx: &mut ComponentCtx) {

        debug_assert_ne!(self.highest_connector_id, ctx.id);

        for local_solution in self.solution_combiner.drain() {

    /// Creates an initially sleeping application connector.

    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. This function just creates the component.

    /// If you create it initially awake, then you must add it to the work

    /// queue. Other aspects of correctness (i.e. setting initial ports) are

    /// relinquished to the caller!

    pub(crate) fn create_pdl_component(&self, connector: ConnectorPDL, initially_sleeping: bool) -> ConnectorKey {

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

        let key = {

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

            lock.create(ConnectorVariant::UserDefined(connector), initially_sleeping)

        };

        self.increment_active_components();

        return key;

    }

    #[inline]

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

    }

    #[inline]

    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

    #[inline]

    pub(crate) fn increment_active_interfaces(&self) {

        let _old_num = self.active_interfaces.fetch_add(1, Ordering::SeqCst);

        debug_assert_ne!(_old_num, 0); // once it hits 0, it stays zero

    }

    pub(crate) fn decrement_active_interfaces(&self) {

        let old_num = self.active_interfaces.fetch_sub(1, Ordering::SeqCst);

        debug_assert!(old_num > 0);

        if old_num == 1 { // such that active interfaces is now 0

            let num_connectors = self.active_connectors.load(Ordering::Acquire);

            if num_connectors == 0 {

                self.signal_for_shutdown();

            }

        }

    }

    #[inline]

    fn increment_active_components(&self) {

        let _old_num = self.active_connectors.fetch_add(1, Ordering::SeqCst);

    }

    fn decrement_active_components(&self) {

        let old_num = self.active_connectors.fetch_sub(1, Ordering::SeqCst);

        debug_assert!(old_num > 0);

        if old_num == 1 { // such that we have no more active connectors (for now!)

            let num_interfaces = self.active_interfaces.load(Ordering::Acquire);

            if num_interfaces == 0 {

                self.signal_for_shutdown();

            }

        }

    }

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

    connectors: RawVec<*mut ScheduledConnector>,

    free: Vec<usize>,

}

impl ConnectorStore {

    fn with_capacity(capacity: usize) -> Self {

        Self {

            connectors: RawVec::with_capacity(capacity),

            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 {

            debug_assert!(!self.free.contains(&(id.0 as usize)));

            let connector = self.connectors.get(id.0 as usize);

            debug_assert!(!connector.is_null());

            return &(**connector).public;

        }

    }

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

    /// time.

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

        unsafe {

            debug_assert!(!self.free.contains(&(key.index as usize)));

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

            debug_assert!(!connector.is_null());

            return &mut (**connector);

        }

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

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

enum ApplicationJob {

    NewChannel((Port, Port)),

    NewConnector(ConnectorPDL, Vec<PortIdLocal>),

    Shutdown,

}

/// 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(VecDeque::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 run(&mut self, _sched_ctx: SchedulerCtx, comp_ctx: &mut ComponentCtx) -> ConnectorScheduling {

        // Handle any incoming messages if we're participating in a round

        while let Some(message) = comp_ctx.read_next_message() {

            match message {

                Message::Data(_) => todo!("data message in API connector"),

                Message::Sync(_)  => todo!("sync message in API connector"),

                Message::Control(_) => todo!("impossible control message"),

            }

        }

        // Handle requests coming from the API

        {

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

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

                match job {

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

                        comp_ctx.push_port(endpoint_a);

                        comp_ctx.push_port(endpoint_b);

                    }

                    ApplicationJob::NewConnector(connector, initial_ports) => {

                        comp_ctx.push_component(connector, initial_ports);

                    },

                    ApplicationJob::Shutdown => {

                        debug_assert!(queue.is_empty());

                        return ConnectorScheduling::Exit;

                    }

                }

            }

        }

        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 {

    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 {

        let (getter_port, putter_port) = self.runtime.create_channel(self.connector_id);

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

        }

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

        let connector = ConnectorPDL::new(state);

        // Put on job queue

        {

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

            queue.push_back(ApplicationJob::NewConnector(connector, initial_ports));

        }

        self.wake_up_connector_with_ping();

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

    }

    fn wake_up_connector_with_ping(&self) {

        let connector = self.runtime.get_component_public(self.connector_id);

        connector.inbox.insert_message(Message::Control(ControlMessage {

            id: 0,

            sending_component_id: self.connector_id,

            content: ControlContent::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.push_work(key);

        }

    }

}

impl Drop for ApplicationInterface {

    fn drop(&mut self) {

        {

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

            lock.push_back(ApplicationJob::Shutdown);

        }

        self.wake_up_connector_with_ping();

        self.runtime.decrement_active_interfaces();

    }

}

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

    }

}

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

pub struct ChannelId {

    pub index: u32,

}

impl ChannelId {

    pub fn new(id: u32) -> Self {

        return Self{ index: id };

    }

}

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

pub enum PortKind {

    Putter,

    Getter,

}

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

pub enum PortState {

    Open,

    Closed,

}

/// Represents a port inside of the runtime. This is generally the local view of

/// a connector on its port, which may not be consistent with the rest of the

/// global system (e.g. its peer was moved to a new connector, or the peer might

/// have died in the meantime, so it is no longer usable).

#[derive(Clone)]

pub struct Port {

    pub self_id: PortIdLocal,

    pub peer_id: PortIdLocal,

    pub channel_id: ChannelId,

    pub kind: PortKind,

    pub state: PortState,

    pub peer_connector: ConnectorId, // might be temporarily inconsistent while peer port is sent around in non-sync phase

}

// TODO: Turn port ID into its own type

pub struct Channel {

    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

}

            }

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

        }

    }

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

        debug_assert_eq!(connector_key.index, connector.ctx_fancy.id.0);

        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)

            }

        }

    }

    #[inline]

    fn get_message_target_port(message: &Message) -> Option<PortIdLocal> {

        match message {

            Message::Data(data) => return Some(data.data_header.target_port),

            Message::Sync(_) => {},

            Message::Control(control) => {

                match &control.content {

                    ControlContent::PortPeerChanged(port_id, _) => return Some(*port_id),

                    ControlContent::CloseChannel(port_id) => return Some(*port_id),

                    ControlContent::Ping | ControlContent::Ack => {},

                }

            },

        }

        return None

    }

    // TODO: Remove, this is debugging stuff

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

    }

    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.

pub(crate) struct ComponentCtx {

    // Mostly managed by the scheduler

    pub(crate) id: ConnectorId,

    ports: Vec<Port>,

    inbox_messages: Vec<Message>, // never control or ping messages

    inbox_len_read: usize,

    // Submitted by the component

    is_in_sync: bool,

    changed_in_sync: bool,

    outbox: VecDeque<Message>,

    state_changes: VecDeque<ComponentStateChange>,

    // Workspaces that may be used by components to (generally) prevent

    // allocations. Be a good scout and leave it empty after you've used it.

    // TODO: Move to scheduler ctx, this is the wrong place

    pub workspace_ports: Vec<PortIdLocal>,

    pub workspace_branches: Vec<BranchId>,

}

impl ComponentCtx {

    pub(crate) fn new_empty() -> Self {

        return Self{

            id: ConnectorId::new_invalid(),

            ports: Vec::new(),

            inbox_messages: Vec::new(),

use super::*;

use crate::{PortId, ProtocolDescription};

use crate::common::Id;

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

fn create_runtime(pdl: &str) -> Runtime {

    let protocol = ProtocolDescription::parse(pdl.as_bytes()).expect("parse pdl");

    let runtime = Runtime::new(NUM_THREADS, protocol);

    return runtime;

}

fn run_test_in_runtime<F: Fn(&mut ApplicationInterface)>(pdl: &str, constructor: F) {

    let protocol = ProtocolDescription::parse(pdl.as_bytes())

        .expect("parse PDL");

    let runtime = Runtime::new(NUM_THREADS, protocol);

    let mut api = runtime.create_interface();

    for _ in 0..NUM_INSTANCES {

        constructor(&mut api);

    }

    // Wait until done :)

}

#[test]

fn test_put_and_get() {

    const CODE: &'static str = "

    primitive putter(out<bool> sender, u32 loops) {

        u32 index = 0;

        while (index < loops) {

            synchronous {

                put(sender, true);

            }

            index += 1;

        }

    }

    primitive getter(in<bool> receiver, u32 loops) {

        u32 index = 0;

        while (index < loops) {

            synchronous {

                auto result = get(receiver);

                assert(result);

            }

            index += 1;

        }

    }

    ";

    run_test_in_runtime(CODE, |api| {

        let channel = api.create_channel();

        api.create_connector("", "putter", ValueGroup::new_stack(vec![

            Value::Output(PortId(Id{ connector_id: 0, u32_suffix: channel.putter_id.index })),

            Value::UInt32(NUM_LOOPS)

        ])).expect("create putter");

        api.create_connector("", "getter", ValueGroup::new_stack(vec![

            Value::Input(PortId(Id{ connector_id: 0, u32_suffix: channel.getter_id.index })),

            Value::UInt32(NUM_LOOPS)

        ])).expect("create getter");

    });

}