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 {

        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)

    }

}

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

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

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

}

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

        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

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

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

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

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

}

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

    }

}

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

            },

    }

    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 {

        }

    }

    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

                        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,

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

        }

            },

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

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

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

        }

    }

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 {

}

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{

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

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

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

    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,

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

        return Self {

            runtime,

            polling,

            id,

            comp: 0,

        }

    }

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

        }

    }

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

        // TODO: Probably not always use colour

    }

}

impl Scheduler {

    // public interface to thread

    }

    pub fn run(&mut self) {

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