section.hash(state);

                }

                in_use_index += 1;

            }

        }

    }

}

impl PartialEq for MonomorphKey {

    fn eq(&self, other: &Self) -> bool {

        if self.in_use.is_empty() {

            let temp_result = self.parts == other.parts;

            return temp_result;

        } else {

            // Outer type does not match

            if self.parts[0] != other.parts[0] {

                return false;

            }

            debug_assert_eq!(self.parts[0].num_embedded() as usize, self.in_use.len());

            let mut iter_self = ConcreteTypeIter::new(self.parts.as_slice(), 0);

            let mut iter_other = ConcreteTypeIter::new(other.parts.as_slice(), 0);

            let mut index = 0;

            while let Some(section_self) = iter_self.next() {

                let section_other = iter_other.next().unwrap();

                let in_use = self.in_use[index];

                index += 1;

                if !in_use {

                    continue;

                }

                if section_self != section_other {

                    return false;

                }

            }

            return true;

        }

    }

}

impl Eq for MonomorphKey {}

use std::cell::UnsafeCell;

/// Lookup table for monomorphs. Wrapped in a special struct because we don't

/// want to allocate for each lookup (what we really want is a HashMap that

pub(crate) struct MonomorphTable {

    lookup: HashMap<MonomorphKey, i32>, // indexes into `monomorphs`

    pub(crate) monomorphs: Vec<TypeMonomorph>,

    // We use an UnsafeCell because this is only used internally per call to

    // `get_monomorph_index` calls. This is safe because `&TypeMonomorph`s

    // retrieved for this class remain valid when the key is mutated and the

    // type table is not multithreaded.

    //

    // I added this because we don't want to allocate for each lookup, hence we

    // need a reusable `key` internal to this class. This in turn makes

    // `get_monomorph_index` a mutable call. Now the code that calls this

    // function (even though we're not mutating the table!) needs a lot of extra

    // boilerplate. I opted for the `UnsafeCell` instead of the boilerplate.

    key: UnsafeCell<MonomorphKey>,

}

impl MonomorphTable {

    fn new() -> Self {

        return Self {

            lookup: HashMap::with_capacity(256),

            monomorphs: Vec::with_capacity(256),

            key: UnsafeCell::new(MonomorphKey{

                parts: Vec::with_capacity(32),

                in_use: Vec::with_capacity(32),

            }),

        }

    }

    fn insert_with_zero_size_and_alignment(&mut self, concrete_type: ConcreteType, in_use: &[PolymorphicVariable], variant: MonomorphVariant) -> i32 {

        let key = MonomorphKey{

            parts: Vec::from(concrete_type.parts.as_slice()),

            in_use: in_use.iter().map(|v| v.is_in_use).collect(),

        };

        let index = self.monomorphs.len();

        let _result = self.lookup.insert(key, index as i32);

        debug_assert!(_result.is_none()); // did not exist yet

        self.monomorphs.push(TypeMonomorph{

            concrete_type,

            size: 0,

            alignment: 0,

            variant,

        });

        return index as i32;

    }

    fn get_monomorph_index(&self, parts: &[ConcreteTypePart], in_use: &[PolymorphicVariable]) -> Option<i32> {

        let key = unsafe {

            // Clear-and-extend to, at some point, prevent future allocations

            let key = &mut *self.key.get();

            key.parts.clear();

            key.parts.extend_from_slice(parts);

            key.in_use.clear();

            key.in_use.extend(in_use.iter().map(|v| v.is_in_use));

            &*key

        };

        match self.lookup.get(key) {

            Some(index) => return Some(*index),

            None => return None,

        }

    }

mod store;

mod runtime;

mod component;

mod communication;

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

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

use std::collections::VecDeque;

use crate::protocol::*;

use super::communication::Message;

use super::component::{CompCtx, CompPDL};

use super::store::{ComponentStore, QueueDynMpsc, QueueDynProducer};

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

// Component

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

/// Key to a component. Type system somewhat ensures that there can only be one

/// of these. Only with a key one may retrieve privately-accessible memory for

/// a component. Practically just a generational index, like `CompId` is.

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

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

    }

}

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

}

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

        };

        handle.increment_users();

        return handle;

    }

    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 true if the component should be destroyed

    pub(crate) fn decrement_users(&mut self) -> bool {

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

        dbg_code!(self.decremented = true);

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

        return old_count == 1;

    }

}

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,

            #[cfg(debug_assertions)] decremented: false,

        };

    }

}

impl std::ops::Deref for CompHandle {

    type Target = CompPublic;

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

        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 {

}

impl Runtime {

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

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

            protocol: protocol_description,

            components: ComponentStore::new(128),

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

            work_condvar: Condvar::new(),

        };

    }

    // Scheduling and retrieving work

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

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

        while lock.is_empty() && self.active_elements.load(Ordering::Acquire) != 0 {

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

        }

        return lock.pop_front();

    }

    pub(crate) fn enqueue_work(&self, key: CompKey) {

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

        lock.push_back(key);

        self.work_condvar.notify_one();

    }

    // Creating/destroying components

    /// Creates a new component. Note that the public part will be properly

    /// initialized, but the private fields (e.g. owned ports, peers, etc.)

    /// are not.

    pub(crate) fn create_pdl_component(&self, comp: CompPDL, initially_sleeping: bool) -> (CompKey, &mut RuntimeComp) {

        let inbox_queue = QueueDynMpsc::new(16);

        let inbox_producer = inbox_queue.producer();

        let comp = RuntimeComp{

            public: CompPublic{

                sleeping: AtomicBool::new(initially_sleeping),

                num_handles: AtomicU32::new(1), // the component itself acts like a handle

                inbox: inbox_producer,

            },

            code: comp,

            ctx: CompCtx::default(),

            inbox: inbox_queue,

        };

        let index = self.components.create(comp);

        // TODO: just do a reserve_index followed by a consume_index or something

        let component = self.components.get_mut(index);

        component.ctx.id = CompId(index);

        return (CompKey(index), component);

    }

    pub(crate) fn get_component(&self, key: CompKey) -> &mut RuntimeComp {

        let component = self.components.get_mut(key.0);

        return component;

    }

    pub(crate) fn get_component_public(&self, id: CompId) -> CompHandle {

        let component = self.components.get(id.0);

        return CompHandle::new(&component.public);

    }

    pub(crate) fn destroy_component(&self, key: CompKey) {

        debug_assert_eq!(self.get_component(key).public.num_handles.load(Ordering::Acquire), 0);

        self.components.destroy(key.0);

    }

}

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

use super::component::*;

use super::runtime::*;

use super::communication::*;

/// Data associated with a scheduler thread

pub(crate) struct Scheduler {

    scheduler_id: u32,

}

pub(crate) struct SchedulerCtx<'a> {

}

impl<'a> SchedulerCtx<'a> {

        return Self {

            runtime,

        }

    }

}

impl Scheduler {

    // public interface to thread

        return Scheduler{ runtime, scheduler_id }

    }

    pub fn run(&mut self) {

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

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

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

                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

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

            }