ports: Vec::new(),

            peers: Vec::new(),

            port_id_counter: 0,

        }

    }

    /// Creates a new channel that is fully owned by the component associated

    /// with this context.

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

        let putter_id = PortId(self.take_port_id());

        let getter_id = PortId(self.take_port_id());

        self.ports.push(Port{

            self_id: putter_id,

            peer_port_id: getter_id,

            kind: PortKind::Putter,

            state: PortState::Open,

            peer_comp_id: self.id,

            associated_with_peer: false,

        });

        self.ports.push(Port{

            self_id: getter_id,

            peer_port_id: putter_id,

            kind: PortKind::Getter,

            state: PortState::Open,

            peer_comp_id: self.id,

            associated_with_peer: false,

        });

        return Channel{ putter_id, getter_id };

    }

    /// Adds a new port. Make sure to call `add_peer` afterwards.

    pub(crate) fn add_port(&mut self, peer_comp_id: CompId, peer_port_id: PortId, kind: PortKind, state: PortState) -> LocalPortHandle {

        let self_id = PortId(self.take_port_id());

        self.ports.push(Port{

            self_id, peer_comp_id, peer_port_id, kind, state,

            #[cfg(debug_assertions)] associated_with_peer: false,

        });

        return LocalPortHandle(self_id);

    }

    /// Removes a port. Make sure you called `remove_peer` first.

    pub(crate) fn remove_port(&mut self, port_handle: LocalPortHandle) -> Port {

        let port_index = self.must_get_port_index(port_handle);

        let port = self.ports.remove(port_index);

        debug_assert!(!port.associated_with_peer);

        return port;

    }

    /// Adds a new peer. This must be called for every port, no matter the

    /// component the channel is connected to. If a `CompHandle` is supplied,

    /// then it will be used to add the peer. Otherwise it will be retrieved

    /// from the runtime using its ID.

    pub(crate) fn add_peer(&mut self, port_handle: LocalPortHandle, sched_ctx: &SchedulerCtx, peer_comp_id: CompId, handle: Option<&CompHandle>) {

        let self_id = self.id;

        let port = self.get_port_mut(port_handle);

        debug_assert_eq!(port.peer_comp_id, peer_comp_id);

        debug_assert!(!port.associated_with_peer);

        if !Self::requires_peer_reference(port, self_id, false) {

            return;

        }

        dbg_code!(port.associated_with_peer = true);

        match self.get_peer_index_by_id(peer_comp_id) {

            Some(peer_index) => {

                let peer = &mut self.peers[peer_index];

                peer.num_associated_ports += 1;

            },

            None => {

                let handle = match handle {

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

                    None => sched_ctx.runtime.get_component_public(peer_comp_id)

                };

                self.peers.push(Peer{

                    id: peer_comp_id,

                    num_associated_ports: 1,

                    handle,

                });

            }

        }

    }

    /// Removes a peer associated with a port.

    pub(crate) fn remove_peer(&mut self, sched_ctx: &SchedulerCtx, port_handle: LocalPortHandle, peer_id: CompId, also_remove_if_closed: bool) {

        let self_id = self.id;

        let port = self.get_port_mut(port_handle);

        debug_assert_eq!(port.peer_comp_id, peer_id);

        if !Self::requires_peer_reference(port, self_id, also_remove_if_closed) {

            return;

        }

        debug_assert!(port.associated_with_peer);

        dbg_code!(port.associated_with_peer = false);

        let peer_index = self.get_peer_index_by_id(peer_id).unwrap();

        let peer = &mut self.peers[peer_index];

        peer.num_associated_ports -= 1;

        if peer.num_associated_ports == 0 {

            let mut peer = self.peers.remove(peer_index);

            if let Some(key) = peer.handle.decrement_users() {

                debug_assert_ne!(key.downgrade(), self.id); // should be upheld by the code that shuts down a component

                sched_ctx.runtime.destroy_component(key);

            }

        }

    }

    pub(crate) fn set_port_state(&mut self, port_handle: LocalPortHandle, new_state: PortState) {

        let port_info = self.get_port_mut(port_handle);

        debug_assert_ne!(port_info.state, PortState::Closed); // because then we do not expect to change the state

        port_info.state = new_state;

    }

    pub(crate) fn get_port_handle(&self, port_id: PortId) -> LocalPortHandle {

        return LocalPortHandle(port_id);

    }

    // should perhaps be revised, used in main inbox

    pub(crate) fn get_port_index(&self, port_handle: LocalPortHandle) -> usize {

        return self.must_get_port_index(port_handle);

    }

    pub(crate) fn get_peer_handle(&self, peer_id: CompId) -> LocalPeerHandle {

        return LocalPeerHandle(peer_id);

    }

    pub(crate) fn get_port(&self, port_handle: LocalPortHandle) -> &Port {

        let index = self.must_get_port_index(port_handle);

        return &self.ports[index];

    }

    pub(crate) fn get_port_mut(&mut self, port_handle: LocalPortHandle) -> &mut Port {

        let index = self.must_get_port_index(port_handle);

        return &mut self.ports[index];

    }

    pub(crate) fn get_port_by_index_mut(&mut self, index: usize) -> &mut Port {

        return &mut self.ports[index];

    }

    pub(crate) fn get_peer(&self, peer_handle: LocalPeerHandle) -> &Peer {

        let index = self.must_get_peer_index(peer_handle);

        return &self.peers[index];

    }

    pub(crate) fn get_peer_mut(&mut self, peer_handle: LocalPeerHandle) -> &mut Peer {

        let index = self.must_get_peer_index(peer_handle);

        return &mut self.peers[index];

    }

    #[inline]

    pub(crate) fn iter_ports(&self) -> impl Iterator<Item=&Port> {

        return self.ports.iter();

    }

    #[inline]

    pub(crate) fn iter_ports_mut(&mut self) -> impl Iterator<Item=&mut Port> {

        return self.ports.iter_mut();

    }

    #[inline]

    pub(crate) fn iter_peers(&self) -> impl Iterator<Item=&Peer> {

        return self.peers.iter();

    }

    #[inline]

    pub(crate) fn num_ports(&self) -> usize {

        return self.ports.len();

    }

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

    // Local utilities

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

    #[inline]

    fn requires_peer_reference(port: &Port, self_id: CompId, required_if_closed: bool) -> bool {

        return (port.state != PortState::Closed || required_if_closed) && port.peer_comp_id != self_id;

    }

    fn must_get_port_index(&self, handle: LocalPortHandle) -> usize {

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

            if port.self_id == handle.0 {

                return index;

            }

        }

        unreachable!()

    }

    fn must_get_peer_index(&self, handle: LocalPeerHandle) -> usize {

        for (index, peer) in self.peers.iter().enumerate() {

            if peer.id == handle.0 {

                return index;

            }

        if self.mode == Mode::BlockedPut && port_id == self.mode_port {

            // We were blocked on the port that just became unblocked, so

            // send the message.

            debug_assert_eq!(port_info.kind, PortKind::Putter);

            let mut replacement = ValueGroup::default();

            std::mem::swap(&mut replacement, &mut self.mode_value);

            self.send_data_message_and_wake_up(sched_ctx, comp_ctx, port_handle, replacement);

            self.mode = Mode::Sync;

            self.mode_port = PortId::new_invalid();

        }

    }

    fn create_component_and_transfer_ports(

        &mut self,

        sched_ctx: &SchedulerCtx, creator_ctx: &mut CompCtx,

        definition_id: DefinitionId, monomorph_index: i32, mut arguments: ValueGroup

    ) {

        struct PortPair{

            creator_handle: LocalPortHandle,

            creator_id: PortId,

            created_handle: LocalPortHandle,

            created_id: PortId,

        }

        let mut port_id_pairs = Vec::new();

        let reservation = sched_ctx.runtime.start_create_pdl_component();

        let mut created_ctx = CompCtx::new(&reservation);

        // Take all the ports ID that are in the `args` (and currently belong to

        // the creator component) and translate them into new IDs that are

        // associated with the component we're about to create

        let mut arg_iter = ValueGroupIter::new(&mut arguments);

        while let Some(port_reference) = arg_iter.next() {

            // Create port entry for new component

            let creator_port_id = port_reference.id;

            let creator_port_handle = creator_ctx.get_port_handle(creator_port_id);

            let creator_port = creator_ctx.get_port(creator_port_handle);

            let created_port_handle = created_ctx.add_port(

                creator_port.peer_comp_id, creator_port.peer_port_id,

                creator_port.kind, creator_port.state

            );

            let created_port = created_ctx.get_port(created_port_handle);

            let created_port_id = created_port.self_id;

            port_id_pairs.push(PortPair{

                creator_handle: creator_port_handle,

                creator_id: creator_port_id,

                created_handle: created_port_handle,

                created_id: created_port_id,

            });

            // Modify value in arguments (bit dirty, but double vec in ValueGroup causes lifetime issues)

            let arg_value = if let Some(heap_pos) = port_reference.heap_pos {

                &mut arg_iter.group.regions[heap_pos][port_reference.index]

            } else {

                &mut arg_iter.group.values[port_reference.index]

            };

            match arg_value {

                Value::Input(id) => *id = port_id_to_eval(created_port_id),

                Value::Output(id) => *id = port_id_to_eval(created_port_id),

                _ => unreachable!(),

            }

        }

        // For each transferred port pair set their peer components to the

        // correct values. This will only change the values for the ports of

        // the new component.

        let mut created_component_has_remote_peers = false;

        for pair in port_id_pairs.iter() {

            let creator_port_info = creator_ctx.get_port(pair.creator_handle);

            let created_port_info = created_ctx.get_port_mut(pair.created_handle);

            if created_port_info.peer_comp_id == creator_ctx.id {

                // Port peer is owned by the creator as well

                let created_peer_port_index = port_id_pairs

                    .iter()

                    .position(|v| v.creator_id == creator_port_info.peer_port_id);

                match created_peer_port_index {

                    Some(created_peer_port_index) => {

                        // Peer port moved to the new component as well. So

                        // adjust IDs appropriately.

                        let peer_pair = &port_id_pairs[created_peer_port_index];

                        created_port_info.peer_port_id = peer_pair.created_id;

                        created_port_info.peer_comp_id = reservation.id();

                        todo!("either add 'self peer', or remove that idea from Ctx altogether")

                    },

                    None => {

                        // Peer port remains with creator component.

                        println!("DEBUG: Setting peer for port {:?} of component {:?} to {:?}", created_port_info.self_id, reservation.id(), creator_ctx.id);

                        created_port_info.peer_comp_id = creator_ctx.id;

                        created_ctx.add_peer(pair.created_handle, sched_ctx, creator_ctx.id, None);

                    }

                }

            } else {

                let peer_handle = creator_ctx.get_peer_handle(created_port_info.peer_comp_id);

                let peer_info = creator_ctx.get_peer(peer_handle);

                created_ctx.add_peer(pair.created_handle, sched_ctx, peer_info.id, Some(&peer_info.handle));

                created_component_has_remote_peers = true;

            }

        }

        // We'll now actually turn our reservation for a new component into an

        // actual component. Note that we initialize it as "not sleeping" as

        // its initial scheduling might be performed based on `Ack`s in response

        // to message exchanges between remote peers.

        let prompt = Prompt::new(

            &sched_ctx.runtime.protocol.types, &sched_ctx.runtime.protocol.heap,

            definition_id, monomorph_index, arguments,

        );

        let component = CompPDL::new(prompt, port_id_pairs.len());

        let (created_key, component) = sched_ctx.runtime.finish_create_pdl_component(

            reservation, component, created_ctx, false,

        );

        let created_ctx = &component.ctx;

        // Now modify the creator's ports: remove every transferred port and

        // potentially remove the peer component. Here is also where we will

        // transfer messages in the main inbox.

        for pair in port_id_pairs.iter() {

            // Remove peer if appropriate

            let creator_port_info = creator_ctx.get_port(pair.creator_handle);

            let creator_port_index = creator_ctx.get_port_index(pair.creator_handle);

            let creator_peer_comp_id = creator_port_info.peer_comp_id;

            creator_ctx.remove_peer(sched_ctx, pair.creator_handle, creator_peer_comp_id, false);

            creator_ctx.remove_port(pair.creator_handle);

            // Transfer any messages

            let created_port_index = created_ctx.get_port_index(pair.created_handle);

            let created_port_info = created_ctx.get_port(pair.created_handle);

            debug_assert!(component.code.inbox_main[created_port_index].is_none());

            if let Some(mut message) = self.inbox_main.remove(creator_port_index) {

                message.data_header.target_port = pair.created_id;

                component.code.inbox_main[created_port_index] = Some(message);

            }

            let mut message_index = 0;

            while message_index < self.inbox_backup.len() {

                let message = &self.inbox_backup[message_index];

                if message.data_header.target_port == pair.creator_id {

                    // transfer message

                    let mut message = self.inbox_backup.remove(message_index);

                    message.data_header.target_port = pair.created_id;

                    component.code.inbox_backup.push(message);

                } else {

                    message_index += 1;

                }

            }

            // Handle potential channel between creator and created component

            if created_port_info.peer_comp_id == creator_ctx.id {

                let peer_port_handle = creator_ctx.get_port_handle(created_port_info.peer_port_id);

                let peer_port_info = creator_ctx.get_port_mut(peer_port_handle);

                peer_port_info.peer_comp_id = created_ctx.id;

                creator_ctx.add_peer(peer_port_handle, sched_ctx, created_ctx.id, None);

            }

        }

        // By now all ports have been transferred. We'll now do any of the setup

        // for rerouting/messaging

        if created_component_has_remote_peers {

            let schedule_entry_id = self.control.add_schedule_entry(created_ctx.id);

            for pair in port_id_pairs.iter() {

                let port_info = created_ctx.get_port(pair.created_handle);

                if port_info.peer_comp_id != creator_ctx.id && port_info.peer_comp_id != created_ctx.id {

                    let message = self.control.add_reroute_entry(

                        creator_ctx.id, port_info.peer_port_id, port_info.peer_comp_id,

                        pair.creator_id, pair.created_id, created_ctx.id,

                        schedule_entry_id

                    );

                    let peer_handle = created_ctx.get_peer_handle(port_info.peer_comp_id);

                    let peer_info = created_ctx.get_peer(peer_handle);

                    peer_info.handle.send_message(sched_ctx, message, true);

                }

            }

        } else {

            // Peer can be scheduled immediately

            sched_ctx.runtime.enqueue_work(created_key);

        }

    }

}

#[inline]

fn port_id_from_eval(port_id: EvalPortId) -> PortId {

    return PortId(port_id.id);

}

#[inline]

fn port_id_to_eval(port_id: PortId) -> EvalPortId {

    return EvalPortId{ id: port_id.0 };

}

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

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

    fn find_port_in_value(group: &ValueGroup, value: &Value, ports: &mut Vec<PortId>) {

        match value {

            Value::Input(port_id) | Value::Output(port_id) => {

                // This is an actual port

                let cur_port = PortId(port_id.id);

                for prev_port in ports.iter() {

                    if *prev_port == cur_port {

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

                // Reference to some dynamic thing which might contain ports,

                // so recurse

                let heap_region = &group.regions[*heap_pos as usize];

                for embedded_value in heap_region {

                    find_port_in_value(group, embedded_value, ports);

                }

            },

            _ => {}, // values we don't care about

        }

    }

    // Clear the ports, then scan all the available values

    ports.clear();

    for value in &value_group.values {

        find_port_in_value(value_group, value, ports);

    }

}

struct ValueGroupIter<'a> {

    group: &'a mut ValueGroup,

    heap_stack: Vec<(usize, usize)>,

    index: usize,

}

impl<'a> ValueGroupIter<'a> {

    fn new(group: &'a mut ValueGroup) -> Self {

        return Self{ group, heap_stack: Vec::new(), index: 0 }

    }

}

struct ValueGroupPortRef {

    id: PortId,

    heap_pos: Option<usize>, // otherwise: on stack

    index: usize,

}

pub(crate) struct CompKey(pub u32);

impl CompKey {

    pub(crate) fn downgrade(&self) -> CompId {

        return CompId(self.0);

    }

}

/// Generational ID of a component

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

pub struct CompId(pub u32);

impl CompId {

    pub(crate) fn new_invalid() -> CompId {

        return CompId(u32::MAX);

    }

    /// Upgrade component ID to component key. Unsafe because the caller needs

    /// to make sure that only one component key can exist at a time (to ensure

    /// a component can only be scheduled/executed by one thread).

    pub(crate) unsafe fn upgrade(&self) -> CompKey {

        return CompKey(self.0);

    }

}

/// Handle to a component that is being created.

pub(crate) struct CompReserved {

    reservation: ComponentReservation,

}

impl CompReserved {

    pub(crate) fn id(&self) -> CompId {

        return CompId(self.reservation.index)

    }

}

/// Private fields of a component, may only be modified by a single thread at

/// a time.

pub(crate) struct RuntimeComp {

    pub public: CompPublic,

    pub code: CompPDL,

    pub ctx: CompCtx,

    pub inbox: QueueDynMpsc<Message>,

    pub exiting: bool,

}

/// Should contain everything that is accessible in a thread-safe manner

// TODO: Do something about the `num_handles` thing. This needs to be a bit more

//  "foolproof" to lighten the mental burden of using the `num_handles`

//  variable.

pub(crate) struct CompPublic {

    pub sleeping: AtomicBool,

    pub num_handles: AtomicU32, // manually modified (!)

    inbox: QueueDynProducer<Message>,

}

/// Handle to public part of a component. Would be nice if we could

/// automagically manage the `num_handles` counter. But when it reaches zero we

/// need to manually remove the handle from the runtime. So we just have debug

/// code to make sure this actually happens.

pub(crate) struct CompHandle {

    target: *const CompPublic,

    id: CompId, // TODO: @Remove after debugging

    #[cfg(debug_assertions)] decremented: bool,

}

impl CompHandle {

    fn new(id: CompId, public: &CompPublic) -> CompHandle {

        let handle = CompHandle{

            target: public,

            id,

            #[cfg(debug_assertions)] decremented: false,

        };

        handle.increment_users();

        return handle;

    }

    pub(crate) fn send_message(&self, sched_ctx: &SchedulerCtx, message: Message, try_wake_up: bool) {

        sched_ctx.log(&format!("Sending message to [c:{:03}, wakeup:{}]: {:?}", self.id.0, try_wake_up, message));

        self.inbox.push(message);

        if try_wake_up {

            wake_up_if_sleeping(sched_ctx, self.id, self);

        }

    }

    fn increment_users(&self) {

        let old_count = self.num_handles.fetch_add(1, Ordering::AcqRel);

        debug_assert!(old_count > 0); // because we should never be able to retrieve a handle when the component is (being) destroyed

    }

    /// Returns the `CompKey` to the component if it should be destroyed

    pub(crate) fn decrement_users(&mut self) -> Option<CompKey> {

        debug_assert!(!self.decremented, "illegal to 'decrement_users' twice");

        let old_count = self.num_handles.fetch_sub(1, Ordering::AcqRel);

        let new_count = old_count - 1;

        dbg_code!(self.decremented = true);

        if new_count == 0 {

            return Some(unsafe{ self.id.upgrade() });

        }

        return None;

    }

}

impl Clone for CompHandle {

    fn clone(&self) -> Self {

        debug_assert!(!self.decremented, "illegal to clone after 'decrement_users'");

        self.increment_users();

        return CompHandle{

            target: self.target,

            id: self.id,

            #[cfg(debug_assertions)] decremented: false,

        };

    }

}

impl std::ops::Deref for CompHandle {

    type Target = CompPublic;

    fn deref(&self) -> &Self::Target {

        debug_assert!(!self.decremented); // cannot access if control is relinquished

        return unsafe{ &*self.target };

    }

}

impl Drop for CompHandle {

    fn drop(&mut self) {

        debug_assert!(self.decremented, "need call to 'decrement_users' before dropping");

    }

}

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

// Runtime

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

pub struct Runtime {

    pub(crate) inner: Arc<RuntimeInner>,

    threads: Vec<std::thread::JoinHandle<()>>,

}

impl Runtime {

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

        assert!(num_threads > 0, "need a thread to perform work");

        let runtime_inner = Arc::new(RuntimeInner {

            protocol: protocol_description,

            components: ComponentStore::new(128),

            work_queue: Mutex::new(VecDeque::with_capacity(128)),

            work_condvar: Condvar::new(),

            active_elements: AtomicU32::new(1),

        });

        let mut runtime = Runtime {

            inner: runtime_inner,

            threads: Vec::with_capacity(num_threads as usize),

        };

        for thread_index in 0..num_threads {

            let mut scheduler = Scheduler::new(runtime.inner.clone(), thread_index);

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

                scheduler.run();

            });

            runtime.threads.push(thread_handle);

        }

        return runtime;

    }

}

impl Drop for Runtime {

    fn drop(&mut self) {

        self.inner.decrement_active_components();

        for handle in self.threads.drain(..) {

            handle.join().expect("join scheduler thread");

        }

    }

}

/// Memory that is maintained by "the runtime". In practice it is maintained by

/// multiple schedulers, and this serves as the common interface to that memory.

pub(crate) struct RuntimeInner {

    pub protocol: ProtocolDescription,

    components: ComponentStore<RuntimeComp>,

    work_queue: Mutex<VecDeque<CompKey>>,

    work_condvar: Condvar,

    active_elements: AtomicU32, // active components and APIs (i.e. component creators)

}

impl RuntimeInner {

    // Scheduling and retrieving work

    pub(crate) fn take_work(&self) -> Option<CompKey> {

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

use std::sync::Arc;

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

use super::component::*;

use super::runtime::*;

/// Data associated with a scheduler thread

pub(crate) struct Scheduler {

    runtime: Arc<RuntimeInner>,

    scheduler_id: u32,

}

pub(crate) struct SchedulerCtx<'a> {

    pub runtime: &'a RuntimeInner,

    pub id: u32,

    pub comp: u32,

}

impl<'a> SchedulerCtx<'a> {

    pub fn new(runtime: &'a RuntimeInner, id: u32) -> Self {

        return Self {

            runtime,

            id,

            comp: 0,

        }

    }

    pub(crate) fn log(&self, text: &str) {

        println!("[s:{:02}, c:{:03}] {}", self.id, self.comp, text);

    }

}

impl Scheduler {

    // public interface to thread

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

        return Scheduler{ runtime, scheduler_id }

    }

    pub fn run(&mut self) {

        let mut scheduler_ctx = SchedulerCtx::new(&*self.runtime, self.scheduler_id);

        'run_loop: loop {

            // Wait until we have something to do (or need to quit)

            let comp_key = self.runtime.take_work();

            if comp_key.is_none() {

                break 'run_loop;

            }

            let comp_key = comp_key.unwrap();

            let component = self.runtime.get_component(comp_key);

            scheduler_ctx.comp = comp_key.0;

            // Run the component until it no longer indicates that it needs to

            // be re-executed immediately.

            let mut new_scheduling = CompScheduling::Immediate;

            while let CompScheduling::Immediate = new_scheduling {

                while let Some(message) = component.inbox.pop() {

                    component.code.handle_message(&mut scheduler_ctx, &mut component.ctx, message);

                }

                new_scheduling = component.code.run(&mut scheduler_ctx, &mut component.ctx).expect("TODO: Handle error");

            }

            // Handle the new scheduling

            match new_scheduling {

                CompScheduling::Immediate => unreachable!(),

                CompScheduling::Requeue => { self.runtime.enqueue_work(comp_key); },

                CompScheduling::Sleep => { self.mark_component_as_sleeping(comp_key, component); },

                CompScheduling::Exit => { self.mark_component_as_exiting(&scheduler_ctx, component); }

            }

        }

    }

    // local utilities

    /// Marks component as sleeping, if after marking itself as sleeping the

    /// inbox contains messages then the component will be immediately

    /// rescheduled. After calling this function the component should not be

    /// executed anymore.

    fn mark_component_as_sleeping(&self, key: CompKey, component: &mut RuntimeComp) {

        debug_assert_eq!(key.downgrade(), component.ctx.id); // make sure component matches key

        debug_assert_eq!(component.public.sleeping.load(Ordering::Acquire), false); // we're executing it, so it cannot be sleeping

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

        if component.inbox.can_pop() {

            let should_reschedule = component.public.sleeping

                .compare_exchange(true, false, Ordering::AcqRel, Ordering::Relaxed)

                .is_ok();

            if should_reschedule {

                self.runtime.enqueue_work(key);

            }

        }

    }

    /// Marks the component as exiting by removing the reference it holds to

    /// itself. Afterward the component will enter "normal" sleeping mode (if it

    /// has not yet been destroyed)

    fn mark_component_as_exiting(&self, sched_ctx: &SchedulerCtx, component: &mut RuntimeComp) {

        // If we didn't yet decrement our reference count, do so now

        let comp_key = unsafe{ component.ctx.id.upgrade() };

        if !component.exiting {

            component.exiting = true;

            let old_count = component.public.num_handles.fetch_sub(1, Ordering::AcqRel);

            let new_count = old_count - 1;

            if new_count == 0 {

                sched_ctx.runtime.destroy_component(comp_key);

                return;

            }

        }

        // Enter "regular" sleeping mode

        self.mark_component_as_sleeping(comp_key, component);

    }

}

use crate::protocol::*;

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

use crate::runtime2::runtime::*;

use crate::runtime2::component::{CompCtx, CompPDL};

fn create_component(rt: &Runtime, module_name: &str, routine_name: &str, args: ValueGroup) {

    let prompt = rt.inner.protocol.new_component(

        module_name.as_bytes(), routine_name.as_bytes(), args

    ).expect("create prompt");

    let reserved = rt.inner.start_create_pdl_component();

    let ctx = CompCtx::new(&reserved);

    let (key, _) = rt.inner.finish_create_pdl_component(reserved, CompPDL::new(prompt, 0), ctx, false);

    rt.inner.enqueue_work(key);

}

fn no_args() -> ValueGroup { ValueGroup::new_stack(Vec::new()) }

#[test]

fn test_component_creation() {

    let pd = ProtocolDescription::parse(b"

    primitive nothing_at_all() {

        s32 a = 5;

        auto b = 5 + a;

    }

    ").expect("compilation");

    let rt = Runtime::new(1, pd);

    for i in 0..20 {

        create_component(&rt, "", "nothing_at_all", no_args());

    }

}

#[test]

    let pd = ProtocolDescription::parse(b"

    primitive sender(out<u32> o, u32 outside_loops, u32 inside_loops) {

        u32 outside_index = 0;

        while (outside_index < outside_loops) {

            u32 inside_index = 0;

            sync while (inside_index < inside_loops) {

                put(o, inside_index);

                inside_index += 1;

            }

            outside_index += 1;

        }

    }

    primitive receiver(in<u32> i, u32 outside_loops, u32 inside_loops) {

        u32 outside_index = 0;

        while (outside_index < outside_loops) {

            u32 inside_index = 0;

            sync while (inside_index < inside_loops) {

                auto val = get(i);

                while (val != inside_index) {} // infinite loop if incorrect value is received

                inside_index += 1;

            }

            outside_index += 1;

        }

    }

    composite constructor() {

        channel o_orom -> i_orom;

        channel o_mrom -> i_mrom;

        channel o_ormm -> i_ormm;

        channel o_mrmm -> i_mrmm;

        // one round, one message per round

        new sender(o_orom, 1, 1);

        new receiver(i_orom, 1, 1);

        // multiple rounds, one message per round

        new sender(o_mrom, 5, 1);

        new receiver(i_mrom, 5, 1);

        // one round, multiple messages per round

        new sender(o_ormm, 1, 5);

        new receiver(i_ormm, 1, 5);

        // multiple rounds, multiple messages per round

        new sender(o_mrmm, 5, 5);

        new receiver(i_mrmm, 5, 5);

    }").expect("compilation");

    let rt = Runtime::new(1, pd);

    create_component(&rt, "", "constructor", no_args());

}