format!("Cannot send on this port, as the peer (id:{}) has shut down", port_info.peer_comp_id.0)

        ))

    } else if port_info.state.is_blocked() {

        // Port is blocked, so we cannot send

        exec_state.set_as_blocked_put(transmitting_port_id, value);

        return Ok(CompScheduling::Sleep);

    } else {

        // Port is not blocked, so send to the peer

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

        let peer_info = comp_ctx.get_peer(peer_handle);

        let annotated_message = consensus.annotate_data_message(comp_ctx, port_info, value);

        return Ok(CompScheduling::Immediate);

    }

}

pub(crate) enum IncomingData {

    PlacedInSlot,

    SlotFull(DataMessage),

}

/// Default handling of receiving a data message. In case there is no room for

/// the message it is returned from this function. Note that this function is

        return IncomingData::PlacedInSlot

    } else {

        // Slot is already full, so if the port was previously opened, it will

        // now become closed

        let port_info = comp_ctx.get_port_mut(port_handle);

        debug_assert!(port_info.state == PortState::Open || port_info.state.is_blocked()); // i.e. not closed, but will go off if more states are added in the future

        if port_info.state == PortState::Open {

            comp_ctx.set_port_state(port_handle, PortState::BlockedDueToFullBuffers);

            let (peer_handle, message) =

                control.initiate_port_blocking(comp_ctx, port_handle);

            let peer = comp_ctx.get_peer(peer_handle);

        }

        return IncomingData::SlotFull(incoming_message)

    }

}

pub(crate) enum GetResult {

    Received(DataMessage),

    NoMessage,

    Error((PortInstruction, String)),

}

            *slot = Some(message);

            return Ok(());

        }

    }

    // Did not have any more messages, so if we were blocked, then we need to

    // unblock the port now (and inform the peer of this unblocking)

    if port_info.state == PortState::BlockedDueToFullBuffers {

        comp_ctx.set_port_state(port_handle, PortState::Open);

        let (peer_handle, message) = control.cancel_port_blocking(comp_ctx, port_handle);

        let peer_info = comp_ctx.get_peer(peer_handle);

    }

    return Ok(());

}

/// Handles control messages in the default way. Note that this function may

/// take a lot of actions in the name of the caller: pending messages may be

/// sent, ports may become blocked/unblocked, etc. So the execution

/// (`CompExecState`), control (`ControlLayer`) and consensus (`Consensus`)

/// state may all change.

pub(crate) fn default_handle_control_message(

    exec_state: &mut CompExecState, control: &mut ControlLayer, consensus: &mut Consensus,

        }

    }

    return Ok(());

}

/// Handles a component entering the synchronous block. Will ensure that the

/// `Consensus` and the `ComponentCtx` are initialized properly.

pub(crate) fn default_handle_sync_start(

    exec_state: &mut CompExecState, inbox_main: &mut InboxMainRef,

    sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, consensus: &mut Consensus

) {

    // If any messages are present for this sync round, set the appropriate flag

    // and notify the consensus handler of the present messages

    consensus.notify_sync_start(comp_ctx);

    for (port_index, message) in inbox_main.iter().enumerate() {

        if let Some(message) = message {

            consensus.handle_incoming_data_message(comp_ctx, message);

            let port_info = comp_ctx.get_port_by_index_mut(port_index);

            port_info.received_message_for_sync = true;

        }

    }

    debug_assert_eq!(exec_state.mode, CompMode::NonSync);

    exec_state.mode = CompMode::Sync;

}

/// Handles a component that has reached the end of the sync block. This does

/// not necessarily mean that the component will go into the `NonSync` mode, as

/// it might have to wait for the leader to finish the round for everyone (see

/// `default_handle_sync_decision`)

pub(crate) fn default_handle_sync_end(

    exec_state: &mut CompExecState, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx,

    consensus: &mut Consensus

) {

    debug_assert_eq!(exec_state.mode, CompMode::Sync);

    let decision = consensus.notify_sync_end_success(sched_ctx, comp_ctx);

    exec_state.mode = CompMode::SyncEnd;

    default_handle_sync_decision(sched_ctx, exec_state, comp_ctx, decision, consensus);

}

/// Handles a component initiating the exiting procedure, and closing all of its

/// ports. Should only be called once per component (which is ensured by

/// checking and modifying the mode in the execution state).

#[must_use]

pub(crate) fn default_handle_start_exit(

    exec_state: &mut CompExecState, control: &mut ControlLayer,

    sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, consensus: &mut Consensus

) -> CompScheduling {

    debug_assert_eq!(exec_state.mode, CompMode::StartExit);

    exec_state.mode = CompMode::BusyExit;

    let exit_inside_sync = exec_state.exit_reason.is_in_sync();

    // If exiting while inside sync mode, report to the leader of the current

    // round that we've failed.

    if exit_inside_sync {

        let decision = consensus.notify_sync_end_failure(sched_ctx, comp_ctx);

        default_handle_sync_decision(sched_ctx, exec_state, comp_ctx, decision, consensus);

    }

    // Iterating over ports by index to work around borrowing rules

    for port_index in 0..comp_ctx.num_ports() {

            // Already closed, or in the process of being closed

            continue;

        }

        // Mark as closed

        let port_id = port.self_id;

        port.state = PortState::Closed;

        // Notify peer of closing

        let port_handle = comp_ctx.get_port_handle(port_id);

        let (peer, message) = control.initiate_port_closing(port_handle, exit_inside_sync, comp_ctx);

        let peer_info = comp_ctx.get_peer(peer);

    }

    return CompScheduling::Immediate; // to check if we can shut down immediately

}

/// Handles a component waiting until all peers are notified that it is quitting

/// (i.e. after calling `default_handle_start_exit`).

#[must_use]

pub(crate) fn default_handle_busy_exit(

    exec_state: &mut CompExecState, control: &ControlLayer,

    sched_ctx: &SchedulerCtx

) -> CompScheduling {

    debug_assert_eq!(exec_state.mode, CompMode::BusyExit);

    if control.has_acks_remaining() {

        return CompScheduling::Sleep;

    } else {

        exec_state.mode = CompMode::Exit;

        return CompScheduling::Exit;

    }

}

/// Handles a potential synchronous round decision. If there was a decision then

/// the `Some(success)` value indicates whether the round succeeded or not.

/// Might also end up changing the `ExecState`.

///

/// Might be called in two cases:

/// 1. The component is in regular execution mode, at the end of a sync round,

///     and is waiting for a solution to the round.

        SyncRoundDecision::Solution => true,

        SyncRoundDecision::Failure => false,

    };

    debug_assert!(

        exec_state.mode == CompMode::SyncEnd || (

            exec_state.mode.is_busy_exiting() && exec_state.exit_reason.is_error()

        ) || (

            exec_state.mode.is_in_sync_block() && decision == SyncRoundDecision::Failure

        )

    );

    consensus.notify_sync_decision(decision);

    if success {

        // We cannot get a success message if the component has encountered an

        // error.

        for port_index in 0..comp_ctx.num_ports() {

            let port_info = comp_ctx.get_port_by_index_mut(port_index);

            if port_info.close_at_sync_end {

                port_info.state = PortState::Closed;

            }

        }

        debug_assert_eq!(exec_state.mode, CompMode::SyncEnd);

        exec_state.mode = CompMode::NonSync;

fn default_handle_ack(

    control: &mut ControlLayer, control_id: ControlId,

    sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx

) {

    // Since an `Ack` may cause another one, handle them in a loop

    let mut to_ack = control_id;

    loop {

        let (action, new_to_ack) = control.handle_ack(to_ack, sched_ctx, comp_ctx);

        match action {

            AckAction::SendMessage(target_comp, message) => {

                // FIX @NoDirectHandle

                let mut handle = sched_ctx.runtime.get_component_public(target_comp);

                let _should_remove = handle.decrement_users();

                debug_assert!(_should_remove.is_none());

            },

            AckAction::ScheduleComponent(to_schedule) => {

                // FIX @NoDirectHandle

                let mut handle = sched_ctx.runtime.get_component_public(to_schedule);

                // Note that the component is intentionally not

                // sleeping, so we just wake it up

                debug_assert!(!handle.sleeping.load(std::sync::atomic::Ordering::Acquire));

                let key = unsafe { to_schedule.upgrade() };

                sched_ctx.runtime.enqueue_work(key);

            Some(new_to_ack) => to_ack = new_to_ack,

            None => break,

        }

    }

}

/// Little helper for sending the most common kind of `Ack`

fn default_send_ack(

    causer_of_ack_id: ControlId, peer_handle: LocalPeerHandle,

    sched_ctx: &SchedulerCtx, comp_ctx: &CompCtx

) {

    let peer_info = comp_ctx.get_peer(peer_handle);

        id: causer_of_ack_id,

        sender_comp_id: comp_ctx.id,

        target_port_id: None,

        content: ControlMessageContent::Ack

    }), true);

}

/// Handles the unblocking of a putter port. In case there is a pending message

/// on that port then it will be sent.

fn default_handle_unblock_put(

    exec_state: &mut CompExecState, consensus: &mut Consensus,

    port_handle: LocalPortHandle, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx,

    port_info.state = PortState::Open;

    if exec_state.is_blocked_on_put(port_id) {

        // Annotate the message that we're going to send

        let port_info = comp_ctx.get_port(port_handle); // for immutable access

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

        let to_send = exec_state.mode_value.take();

        let to_send = consensus.annotate_data_message(comp_ctx, port_info, to_send);

        // Retrieve peer to send the message

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

        let peer_info = comp_ctx.get_peer(peer_handle);

        exec_state.mode = CompMode::Sync; // because we're blocked on a `put`, we must've started in the sync state.

        exec_state.mode_port = PortId::new_invalid();

    }

}

#[inline]

pub(crate) fn port_id_from_eval(port_id: EvalPortId) -> PortId {

    return PortId(port_id.id);

}

#[inline]

                let decision = self.consensus.receive_sync_message(sched_ctx, comp_ctx, message);

                component::default_handle_sync_decision(sched_ctx, &mut self.exec_state, comp_ctx, decision, &mut self.consensus);

            },

            Message::Control(message) => {

                if let Err(location_and_message) = component::default_handle_control_message(

                    &mut self.exec_state, &mut self.control, &mut self.consensus,

                    message, sched_ctx, comp_ctx

                ) {

                    component::default_handle_error_for_builtin(&mut self.exec_state, sched_ctx, location_and_message);

                }

            },

            Message::Poll => {

            },

        }

    }

    fn run(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx) -> CompScheduling {

        match self.exec_state.mode {

            CompMode::BlockedSelect => {

                // Not possible: we never enter this state

                unreachable!();

            },

            CompMode::NonSync => {

                // When in non-sync mode

                match &mut self.socket_state {

                    SocketState::Connected(_socket) => {

                        if self.sync_state == SyncState::FinishSyncThenQuit {

                            // Previous request was to let the component shut down

        let port_index = comp_ctx.get_port_index(port_handle);

        if self.inbox_main[port_index].is_none() {

            self.inbox_main[port_index] = Some(message);

        } else {

            self.inbox_backup.push(message);

        }

    }

    fn handle_message(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx, mut message: Message) {

        // sched_ctx.log(&format!("handling message: {:?}", message));

        if let Some(new_target) = self.control.should_reroute(&mut message) {

            let mut target = sched_ctx.runtime.get_component_public(new_target); // TODO: @NoDirectHandle

            let _should_remove = target.decrement_users();

            debug_assert!(_should_remove.is_none());

            return;

        }

        match message {

            Message::Data(message) => {

                self.handle_incoming_data_message(sched_ctx, comp_ctx, message);

            },

            Message::Control(message) => {

                if let Err(location_and_message) = component::default_handle_control_message(

                    &mut self.exec_state, &mut self.control, &mut self.consensus,

            Message::Sync(message) => {

                self.handle_incoming_sync_message(sched_ctx, comp_ctx, message);

            },

            Message::Poll => {

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

            }

        }

    }

    fn run(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx) -> CompScheduling {

        use EvalContinuation as EC;

        // Depending on the mode don't do anything at all, take some special

        // actions, or fall through and run the PDL code.

        match self.exec_state.mode {

            CompMode::NonSync | CompMode::Sync => {

                // continue and run PDL code

            },

            CompMode::SyncEnd | CompMode::BlockedGet | CompMode::BlockedPut | CompMode::BlockedSelect => {

                return CompScheduling::Sleep;

            }

            CompMode::StartExit => return component::default_handle_start_exit(

                &mut self.exec_state, &mut self.control, sched_ctx, comp_ctx, &mut self.consensus

                    },

                    GetResult::NoMessage => {

                        return CompScheduling::Sleep;

                    },

                    GetResult::Error(location_and_message) => {

                        self.handle_generic_component_error(sched_ctx, location_and_message);

                        return CompScheduling::Immediate;

                    }

                }

            },

            EC::Put(expr_id, port_id, value) => {

                debug_assert_eq!(self.exec_state.mode, CompMode::Sync);

                // Send the message

                let target_port_id = port_id_from_eval(port_id);

                let send_result = component::default_send_data_message(

                    &mut self.exec_state, target_port_id,

                    PortInstruction::SourceLocation(expr_id), value,

                    sched_ctx, &mut self.consensus, comp_ctx

                );

                if let Err(location_and_message) = send_result {

                    self.handle_generic_component_error(sched_ctx, location_and_message);

                    return CompScheduling::Immediate;

                } else {

            let created_ctx = &component.ctx;

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

            for pair in opened_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);

                }

            }

        } else {

            // Peer can be scheduled immediately

            sched_ctx.runtime.enqueue_work(created_key);

        }

    }

}

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

                if let Err(location_and_message) = component::default_handle_control_message(

                    &mut self.exec_state, &mut self.control, &mut self.consensus,

                    message, sched_ctx, comp_ctx

                ) {

                    component::default_handle_error_for_builtin(&mut self.exec_state, sched_ctx, location_and_message);

                }

            },

            Message::Poll => unreachable!(),

        }

    }

    fn run(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx) -> CompScheduling {

        match self.exec_state.mode {

            CompMode::BlockedGet | CompMode::BlockedSelect => {

                // impossible for this component, no input ports and no select

                // blocks

                unreachable!();

            }

            CompMode::NonSync => {

                // If in non-sync mode then we check if the arguments make sense

                // (at some point in the future, this is just a testing

                // component).

                if self.random_minimum >= self.random_maximum {

                    // Could throw an evaluation error, but lets just panic

                    panic!("going to crash 'n burn your system now, please provide valid arguments");

                }

                if self.num_sends >= self.max_num_sends {

                    self.exec_state.set_as_start_exit(ExitReason::Termination);

                } else {

                    self.did_perform_send = false;

                    component::default_handle_sync_start(

                        &mut self.exec_state, &mut [], sched_ctx, comp_ctx, &mut self.consensus

                    );

                }

                return CompScheduling::Immediate;

            },

            CompMode::Sync => {

                // This component just sends a single message, then waits until

                // consensus has been reached

                if !self.did_perform_send {

                    let mut random = self.generator.next_u32() - self.random_minimum;

                    let random_delta = self.random_maximum - self.random_minimum;

                    random %= random_delta;

                    random += self.random_minimum;

                    let value_group = ValueGroup::new_stack(vec![Value::UInt32(random)]);

                    let send_result = component::default_send_data_message(

                        &mut self.exec_state, self.output_port_id,

                        PortInstruction::NoSource, value_group,

                        sched_ctx, &mut self.consensus, comp_ctx

                    );

                        break;

                    }

                }

                if !performed_communication {

                    continue;

                }

                let message = SyncMessage{

                    sync_header: self.create_sync_header(comp_ctx),

                    content: SyncMessageContent::NotificationOfLeader,

                };

            }

            self.forward_partial_solution(sched_ctx, comp_ctx);

        } else if header.highest_id.0 < self.highest_id.0 {

            // Sender has a lower ID, so notify it of our higher one

            let message = SyncMessage{

                sync_header: self.create_sync_header(comp_ctx),

                content: SyncMessageContent::NotificationOfLeader,

            };

            let peer_handle = comp_ctx.get_peer_handle(header.sending_id);

            let peer_info = comp_ctx.get_peer(peer_handle);

        } // else: exactly equal

    }

    fn set_annotation(&mut self, source_comp_id: CompId, data_header: &MessageDataHeader) {

        for annotation in self.ports.iter_mut() {

            if annotation.self_port_id == data_header.target_port {

                // Message should have already passed the `handle_new_data_message` function, so we

                // should have already annotated the peer of the port.

                debug_assert!(

                    annotation.peer_discovered &&

                    annotation.peer_comp_id == source_comp_id &&

                    annotation.peer_port_id == data_header.source_port

            }

            if getter.peer_comp_id != comp_ctx.id && !peers.contains(&getter.peer_comp_id) {

                peers.push(getter.peer_comp_id);

            }

        }

        for peer in peers {

            let mut handle = sched_ctx.runtime.get_component_public(peer);

            let message = Message::Sync(SyncMessage{

                sync_header: self.create_sync_header(comp_ctx),

                content: if is_success { SyncMessageContent::GlobalSolution } else { SyncMessageContent::GlobalFailure },

            });

            let _should_remove = handle.decrement_users();

            debug_assert!(_should_remove.is_none());

        }

    }

    fn send_to_leader(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &CompCtx, message: Message) {

        debug_assert_ne!(self.highest_id, comp_ctx.id); // we're not the leader, // TODO: @NoDirectHandle

        let mut leader_info = sched_ctx.runtime.get_component_public(self.highest_id);

        let should_remove = leader_info.decrement_users();

        if let Some(key) = should_remove {

            sched_ctx.runtime.destroy_component(key);

        }

    }

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

    // Small utilities

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

    fn generate_local_solution(&self, comp_ctx: &CompCtx, failed: bool) -> SyncLocalSolution {

        let mut local_solution = Vec::with_capacity(self.ports.len());

use libc::{self, c_int};

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::store::queue_mpsc::*;

pub(crate) type FileDescriptor = c_int;

pub(crate) trait AsFileDescriptor {

    fn as_file_descriptor(&self) -> FileDescriptor;

}

#[derive(Copy, Clone)]

pub(crate) struct UserData(u64);

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

enum PollCmd {

    Register(CompHandle, UserData),

    Unregister(FileDescriptor, UserData),

    Shutdown,

}

pub struct PollingThread {

    poller: Arc<Poller>,

    runtime: Arc<RuntimeInner>,

    queue: QueueDynMpsc<PollCmd>,

}

impl PollingThread {

        let poller = Poller::new()

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

        let poller = Arc::new(poller);

        let queue = QueueDynMpsc::new(64);

        let queue_producers = queue.producer_factory();

        let mut thread_data = PollingThread{

            poller: poller.clone(),

            runtime: runtime.clone(),

            queue,

        };

        let thread_handle = PollingThreadHandle{

            queue: Some(queue_producers.producer()),

            handle: Some(thread_handle),

        };

        let client_factory = PollingClientFactory{

            poller,

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

            queue_factory: queue_producers,

        };

        return Ok((thread_handle, client_factory));

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

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

        }

    }

}

// bit convoluted, but it works

pub(crate) struct PollingThreadHandle {

    // requires Option, because:

    queue: Option<QueueDynProducer<PollCmd>>, // destructor needs to be called

    handle: Option<thread::JoinHandle<()>>, // we need to call `join`

}

use std::thread;

use std::collections::VecDeque;

use crate::protocol::*;

use crate::runtime2::poll::{PollingThread, PollingThreadHandle};

use crate::runtime2::RtError;

use super::communication::Message;

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

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

use super::scheduler::*;

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

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

        };

        handle.increment_users();

        return handle;

    }

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

        self.inbox.push(message);

        if try_wake_up {

            wake_up_if_sleeping(runtime, self.id, self);

        }

    }

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

        return self.id;

    }

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

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

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

// Runtime

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

pub struct Runtime {

    pub(crate) inner: Arc<RuntimeInner>,

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

    polling_handle: PollingThreadHandle,

}

impl Runtime {

    // TODO: debug_logging should be removed at some point

        if num_threads == 0 {

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

        }

        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 (polling_handle, polling_clients) = rt_error_try!(

            "failed to build polling thread"

        );

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

        for thread_index in 0..num_threads {

            let mut scheduler = Scheduler::new(

                runtime_inner.clone(), polling_clients.client(),

            );

            scheduler_threads.push(thread_handle);

        }

        return Ok(Runtime{

            inner: runtime_inner,

            scheduler_threads,

            polling_handle,

        });

    }

    pub fn create_component(&self, module_name: &[u8], routine_name: &[u8]) -> Result<(), ComponentCreationError> {

use std::sync::Arc;

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

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

use super::component::*;

use super::runtime::*;

/// Data associated with a scheduler thread

pub(crate) struct Scheduler {

    runtime: Arc<RuntimeInner>,

    polling: PollingClient,

    scheduler_id: u32,

}

pub(crate) struct SchedulerCtx<'a> {

    pub runtime: &'a RuntimeInner,

    pub polling: &'a PollingClient,

    pub id: u32,

    pub comp: u32,

}

impl<'a> SchedulerCtx<'a> {