Poll,

            Message::Poll =>

                return None,

            Message::Poll => unreachable!(),

        peer_info.handle.send_message(&sched_ctx.runtime, Message::Control(message), true);

                handle.send_message(&sched_ctx.runtime, Message::Control(message), true);

    peer_info.handle.send_message(&sched_ctx.runtime, Message::Control(ControlMessage{

        peer_info.handle.send_message(&sched_ctx.runtime, Message::Data(to_send), true);

            },

            Message::Poll => unreachable!(),

                        peer_info.handle.send_message(&sched_ctx.runtime, Message::Data(message), true);

            target.send_message(&sched_ctx.runtime, message, false); // not waking up: we schedule once we've received all PortPeerChanged Acks

            },

            Message::Poll => {

                unreachable!(); // because we never register at the polling thread

            peer_info.handle.send_message(&sched_ctx.runtime, Message::Control(message), true);

        peer_info.handle.send_message(&sched_ctx.runtime, Message::Data(annotated_message), true);

            peer.handle.send_message(&sched_ctx.runtime, Message::Control(message), true);

            peer_info.handle.send_message(&sched_ctx.runtime, Message::Control(message), true);

        let other_proc = &sched_ctx.runtime.protocol.heap[definition_id];

        let self_proc = &sched_ctx.runtime.protocol.heap[self.prompt.frames[0].definition];

        dbg_code!({

            sched_ctx.log(&format!(

            ));

        });

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

                peer.handle.send_message(&sched_ctx.runtime, Message::Sync(message), true);

            peer_info.handle.send_message(&sched_ctx.runtime, Message::Sync(message), true);

            handle.send_message(&sched_ctx.runtime, message, true);

        leader_info.send_message(&sched_ctx.runtime, message, true);

pub(crate) fn wake_up_if_sleeping(runtime: &RuntimeInner, comp_id: CompId, handle: &CompHandle) {

        runtime.enqueue_work(comp_key);

use std::fmt::{Write, Debug, Display, Formatter as FmtFormatter, Result as FmtResult};

/// Represents an unrecoverable runtime error that is reported to the user (for

/// debugging purposes). Basically a human-readable message with its source

/// location. The error is chainable.

pub struct RtError {

    file: &'static str,

    line: u32,

    message: String,

    cause: Option<Box<RtError>>,

}

impl RtError {

    pub(crate) fn new(file: &'static str, line: u32, message: String) -> RtError {

        return RtError {

            file, line, message, cause: None,

        }

    }

    pub(crate) fn wrap(self, file: &'static str, line: u32, message: String) -> RtError {

        return RtError {

            file, line, message, cause: Some(Box::new(self))

        }

    }

}

impl Display for RtError {

    fn fmt(&self, f: &mut FmtFormatter<'_>) -> FmtResult {

        let mut error = self;

        loop {

            write!(f, "[{}:{}] {}", self.file, self.line, self.message).unwrap();

            match &error.cause {

                Some(cause) => {

                    writeln!(f, " ...");

                    error = cause.as_ref()

                },

                None => {

                    writeln!(f).unwrap();

                },

            }

        }

    }

}

impl Debug for RtError {

    fn fmt(&self, f: &mut FmtFormatter<'_>) -> FmtResult {

        return (self as &dyn Display).fmt(f);

    }

}

macro_rules! rt_error {

    ($fmt:expr) => {

        $crate::runtime2::error::RtError::new(file!(), line!(), $fmt.to_string())

    };

    ($fmt:expr, $($args:expr),*) => {

        $crate::runtime2::error::RtError::new(file!(), line!(), format!($fmt, $($args),*))

    };

}

macro_rules! rt_error_try {

    ($prev:expr, $($fmt_and_args:expr),*) => {

        {

            let result = $prev;

            match result {

                Ok(result) => result,

                Err(result) => return Err(result.wrap(file!(), line!(), format!($($fmt_and_args),*))),

            }

        }

    }

}

#[macro_use] mod error;

pub(crate) use error::RtError;

use std::{io, ptr, time, thread};

use std::sync::Arc;

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

use std::collections::HashMap;

use crate::runtime2::RtError;

use crate::runtime2::runtime::{CompHandle, RuntimeInner};

use crate::runtime2::store::queue_mpsc::*;

#[derive(Copy, Clone)]

pub(crate) struct UserData(u64);

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

// Poller

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

    pub fn new() -> io::Result<Self> {

    /// Performs `epoll_wait`, waiting for the provided timeout or until events

    /// are reported. They are stored in the `events` variable (up to

    /// `events.cap()` are reported, so ensure it is preallocated).

    pub fn wait(&self, events: &mut Vec<libc::epoll_event>, timeout: time::Duration) -> io::Result<()> {

        let timeout_millis = timeout.as_millis();

        let timeout_millis = if timeout_millis > MAX_TIMEOUT {

            -1 // effectively infinite

        } else {

            timeout_millis as c_int

        };

        debug_assert!(events.is_empty());

        debug_assert!(events.capacity() > 0 && events.capacity() < i32::MAX as usize);

            libc::epoll_wait(self.handle, events.as_mut_ptr(), events.capacity() as i32, timeout_millis)

            events.set_len(num_events as usize);

#[cfg(unix)]

impl Drop for Poller {

    fn drop(&mut self) {

        unsafe{ libc::close(self.handle); }

    }

}

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

// Polling Thread

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

enum PollCmd {

    Register(CompHandle, UserData),

    Unregister(FileDescriptor, UserData),

    Shutdown,

}

/// Represents the data needed to build interfaces to the polling thread (which

/// should happen first) and to create the polling thread itself.

pub(crate) struct PollingThreadBuilder {

    poller: Arc<Poller>,

    generation_counter: Arc<AtomicU32>,

    queue: QueueDynMpsc<PollCmd>,

    runtime: Arc<RuntimeInner>,

    logging_enabled: bool,

}

impl PollingThreadBuilder {

    pub(crate) fn new(runtime: Arc<RuntimeInner>, logging_enabled: bool) -> Result<PollingThreadBuilder, RtError> {

        let poller = Poller::new()

            .map_err(|e| rt_error!("failed to create poller, because: {}", e))?;

        return Ok(PollingThreadBuilder {

            poller: Arc::new(poller),

            generation_counter: Arc::new(AtomicU32::new(0)),

            queue: QueueDynMpsc::new(64),

            runtime,

            logging_enabled,

        })

    }

    pub(crate) fn client(&self) -> PollingClient {

        return PollingClient{

            poller: self.poller.clone(),

            generation_counter: self.generation_counter.clone(),

            queue: self.queue.producer(),

        }

    }

    pub(crate) fn into_thread(self) -> (PollingThread, PollingThreadDestroyer) {

        let destroyer = self.queue.producer();

        return (

            PollingThread{

                poller: self.poller,

                runtime: self.runtime,

                queue: self.queue,

                logging_enabled: self.logging_enabled,

            },

            PollingThreadDestroyer::new(destroyer)

        );

    }

}

pub(crate) struct PollingThread {

    poller: Arc<Poller>,

    runtime: Arc<RuntimeInner>,

    queue: QueueDynMpsc<PollCmd>,

    logging_enabled: bool,

}

impl PollingThread {

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

        use crate::runtime2::scheduler::SchedulerCtx;

        use crate::runtime2::communication::Message;

        const NUM_EVENTS: usize = 256;

        const EPOLL_DURATION: time::Duration = time::Duration::from_millis(250);

        // @performance: Lot of improvements possible here, a HashMap is likely

        // a horrible way to do this.

        let mut events = Vec::with_capacity(NUM_EVENTS);

        let mut lookup = HashMap::with_capacity(64);

        self.log("Starting polling thread");

        loop {

            // Retrieve events first (because the PollingClient will first

            // register at epoll, and then push a command into the queue).

            self.poller.wait(&mut events, EPOLL_DURATION).unwrap();

            // Then handle everything in the command queue.

            while let Some(command) = self.queue.pop() {

                match command {

                    PollCmd::Register(handle, user_data) => {

                        self.log(&format!("Registering component {:?} as {}", handle.id(), user_data.0));

                        let key = Self::user_data_as_key(user_data);

                        debug_assert!(!lookup.contains_key(&key));

                        lookup.insert(key, handle);

                    },

                    PollCmd::Unregister(_file_descriptor, user_data) => {

                        let key = Self::user_data_as_key(user_data);

                        debug_assert!(lookup.contains_key(&key));

                        let mut handle = lookup.remove(&key).unwrap();

                        self.log(&format!("Unregistering component {:?} as {}", handle.id(), user_data.0));

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

                            self.runtime.destroy_component(key);

                        }

                    },

                    PollCmd::Shutdown => {

                        // The contract is that all scheduler threads shutdown

                        // before the polling thread. This happens when all

                        // components are removed.

                        self.log("Received shutdown signal");

                        debug_assert!(lookup.is_empty());

                        return;

                    }

                }

            }

            // Now process all of the events. Because we might have had a

            // `Register` command followed by an `Unregister` command (e.g. a

            // component has died), we might get events that are not associated

            // with an entry in the lookup.

            for event in events.drain(..) {

                let key = event.u64;

                if let Some(handle) = lookup.get(&key) {

                    self.log(&format!("Sending poll to {:?} (event: {:x})", handle.id(), event.events));

                    handle.send_message(&self.runtime, Message::Poll, true);

                }

            }

        }

    }

    #[inline]

    fn user_data_as_key(data: UserData) -> u64 {

        return data.0;

    }

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

        if self.logging_enabled {

            println!("[polling] {}", message);

        }

    }

}

// bit convoluted, but it works

pub(crate) struct PollingThreadDestroyer {

    queue: Option<QueueDynProducer<PollCmd>>,

}

impl PollingThreadDestroyer {

    fn new(queue: QueueDynProducer<PollCmd>) -> Self {

        return Self{ queue: Some(queue) };

    }

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

        self.queue.take().unwrap().push(PollCmd::Shutdown);

    }

}

impl Drop for PollingThreadDestroyer {

    fn drop(&mut self) {

        debug_assert!(self.queue.is_none());

    }

}

pub(crate) struct PollTicket(FileDescriptor, u64);

/// A structure that allows the owner to register components at the polling

/// thread. Because of assumptions in the communication queue all of these

/// clients should be dropped before stopping the polling thread.

pub(crate) struct PollingClient {

    poller: Arc<Poller>,

    generation_counter: Arc<AtomicU32>,

    queue: QueueDynProducer<PollCmd>,

}

impl PollingClient {

    fn register<F: AsFileDescriptor>(&self, entity: F, handle: CompHandle, read: bool, write: bool) -> Result<PollTicket, RtError> {

        let generation = self.generation_counter.fetch_add(1, Ordering::Relaxed);

        let user_data = user_data_for_component(handle.id().0, generation);

        self.queue.push(PollCmd::Register(handle, user_data));

        let file_descriptor = entity.as_file_descriptor();

        self.poller.register(file_descriptor, user_data, read, write)

            .map_err(|e| rt_error!("failed to register for polling, because: {}", e))?;

        return Ok(PollTicket(file_descriptor, user_data.0));

    }

    fn unregister(&self, ticket: PollTicket) -> Result<(), RtError> {

        let file_descriptor = ticket.0;

        let user_data = UserData(ticket.1);

        self.queue.push(PollCmd::Unregister(file_descriptor, user_data));

        self.poller.unregister(file_descriptor)

            .map_err(|e| rt_error!("failed to unregister polling, because: {}", e))?;

        return Ok(());

    }

}

#[inline]

fn user_data_for_component(component_id: u32, generation: u32) -> UserData {

    return UserData((generation as u64) << 32 | (component_id as u64));

}

use std::thread;

use crate::runtime2::poll::{PollingThreadBuilder, PollingThreadDestroyer};

use crate::runtime2::RtError;

/// Generational ID of a component.

    id: CompId,

    pub(crate) fn send_message(&self, runtime: &RuntimeInner, message: Message, try_wake_up: bool) {

            wake_up_if_sleeping(runtime, self.id, self);

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

        return self.id;

    }

    scheduler_threads: Vec<thread::JoinHandle<()>>,

    polling_destroyer: PollingThreadDestroyer,

    polling_thread: Option<thread::JoinHandle<()>>,

    pub fn new(num_threads: u32, debug_logging: bool, protocol_description: ProtocolDescription) -> Result<Runtime, RtError> {

        if num_threads == 0 {

            return Err(rt_error!("need at least one thread to create the runtime"));

        }

        let polling_builder = rt_error_try!(

            PollingThreadBuilder::new(runtime_inner.clone(), debug_logging),

            "failed to build polling thread"

        );

        let mut scheduler_threads = Vec::with_capacity(num_threads as usize);

            let mut scheduler = Scheduler::new(

                runtime_inner.clone(), polling_builder.client(),

                thread_index, debug_logging

            );

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

            scheduler_threads.push(thread_handle);

        let (mut poller, polling_destroyer) = polling_builder.into_thread();

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

            poller.run();

        });

        return Ok(Runtime{

            inner: runtime_inner,

            scheduler_threads,

            polling_destroyer,

            polling_thread: Some(polling_thread),

        });

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

        self.polling_destroyer.initiate_destruction();

        self.polling_thread.take().unwrap().join().expect("join polling thread");

use crate::runtime2::poll::PollingClient;

    polling: PollingClient,

    pub polling: &'a PollingClient,

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

            polling,

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

        return Scheduler{ runtime, polling, scheduler_id, debug_logging }

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