}

fn is_reserved_expression_keyword(text: &[u8]) -> bool {

    match text {

        KW_LET | KW_CAST |

        KW_LIT_TRUE | KW_LIT_FALSE | KW_LIT_NULL |

        _ => false,

    }

}

fn is_reserved_type_keyword(text: &[u8]) -> bool {

    match text {

#[derive(Debug)]

pub enum Message {

    Data(DataMessage),

    Sync(SyncMessage),

    Control(ControlMessage),

}

impl Message {

    pub(crate) fn target_port(&self) -> Option<PortId> {

        match self {

            Message::Data(v) =>

                return Some(v.data_header.target_port),

            Message::Control(v) =>

                return v.target_port_id,

            Message::Sync(_) =>

                return None,

        }

    }

    pub(crate) fn modify_target_port(&mut self, port_id: PortId) {

        match self {

            Message::Data(v) =>

                v.data_header.target_port = port_id,

            Message::Control(v) =>

                v.target_port_id = Some(port_id),

            Message::Sync(_) => unreachable!(), // should never be called for this message type

        }

    }

}

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

    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::Control(message) => {

                component::default_handle_control_message(

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

                    message, sched_ctx, comp_ctx

                );

        }

    }

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

        sched_ctx.log(&format!("Running component ComponentRandomU32 (mode: {:?})", self.exec_state.mode));

                        CompScheduling::Sleep

                    } else {

                        let message = self.consensus.annotate_data_message(comp_ctx, port_info, value_group);

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

                        let peer_info = comp_ctx.get_peer(peer_handle);

                        // Remain in sync mode, but after `did_perform_send` was

                        // set to true.

                        CompScheduling::Immediate

                    };

    }

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

            let _should_remove = target.decrement_users();

            debug_assert!(_should_remove.is_none());

            return;

        }

        match message {

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

                    message, sched_ctx, comp_ctx

                );

            },

            Message::Sync(message) => {

                self.handle_incoming_sync_message(sched_ctx, comp_ctx, message);

            }

        }

    }

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

        use EvalContinuation as EC;

            port.state = PortState::Closed;

            // Notify peer of closing

            let port_handle = comp_ctx.get_port_handle(port_id);

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

            let peer_info = comp_ctx.get_peer(peer);

        }

    }

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

    // Handling messages

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

    fn send_data_message_and_wake_up(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &CompCtx, source_port_handle: LocalPortHandle, value: ValueGroup) {

        let port_info = comp_ctx.get_port(source_port_handle);

        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 = self.consensus.annotate_data_message(comp_ctx, port_info, value);

    }

    /// Handles a message that came in through the public inbox. This function

    /// will handle putting it in the correct place, and potentially blocking

    /// the port in case too many messages are being received.

    fn handle_incoming_data_message(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, message: DataMessage) {

        if port_info.state == PortState::Open {

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

            let (peer_handle, message) =

                self.control.initiate_port_blocking(comp_ctx, port_handle);

            let peer = comp_ctx.get_peer(peer_handle);

        }

        // But we still need to remember the message, so:

        self.inbox_backup.push(message);

    }

        // Did not have any more messages. So if we were blocked, then we need

        // to send the "unblock" message.

        if port_info.state == PortState::BlockedDueToFullBuffers {

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

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

            let peer_info = comp_ctx.get_peer(peer_handle);

        }

    }

    fn handle_incoming_sync_message(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, message: SyncMessage) {

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

        self.handle_sync_decision(sched_ctx, comp_ctx, decision);

            created_handle: LocalPortHandle,

            created_id: PortId,

        }

        let mut opened_port_id_pairs = Vec::new();

        let mut closed_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 = ValueGroupPortIter::new(&mut arguments);

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

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

        }

                }

                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

        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 super::scheduler::*;

use super::runtime::*;

/// If the component is sleeping, then that flag will be atomically set to

/// false. If we're the ones that made that happen then we add it to the work

/// queue.

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

    let should_wake_up = handle.sleeping

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

        .is_ok();

    if should_wake_up {

        let comp_key = unsafe{ comp_id.upgrade() };

    }

}

mod store;

mod runtime;

mod component;

mod communication;

mod scheduler;

mod poll;

mod stdlib;

#[cfg(test)] mod tests;

pub use runtime::Runtime;

pub(crate) use scheduler::SchedulerCtx;

pub(crate) use communication::{

    PortId, PortKind, PortState,

    Message, ControlMessage, SyncMessage, DataMessage,

    SyncRoundDecision

};

use libc::{self, c_int};

pub(crate) type FileDescriptor = c_int;

pub(crate) trait AsFileDescriptor {

    fn as_file_descriptor(&self) -> FileDescriptor;

}

#[cfg(unix)]

pub(crate) struct Poller {

    handle: c_int,

}

// All of this is gleaned from the `mio` crate.

#[cfg(unix)]

impl Poller {

        let handle = syscall_result(unsafe{ libc::epoll_create1(libc::EPOLL_CLOEXEC) })?;

        return Ok(Self{

            handle,

        })

    }

    fn register(&self, fd: FileDescriptor, user: UserData, read: bool, write: bool) -> io::Result<()> {

        let mut event = libc::epoll_event{

            events: Self::events_from_rw_flags(read, write),

            libc::epoll_ctl(self.handle, libc::EPOLL_CTL_DEL, fd, ptr::null_mut())

        })?;

        return Ok(());

    }

        // See `mio` for the reason. Works around a linux bug

        #[cfg(target_pointer_width = "32")]

        const MAX_TIMEOUT: u128 = 1789569;

        #[cfg(not(target_pointer_width = "32"))]

        const MAX_TIMEOUT: u128 = c_int::MAX as u128;

        let num_events = syscall_result(unsafe{

        })?;

        unsafe{

            debug_assert!(num_events >= 0);

        }

        return Ok(());

    }

    fn events_from_rw_flags(read: bool, write: bool) -> u32 {

        }

        return events as u32;

    }

}

#[inline]

fn syscall_result(result: c_int) -> io::Result<c_int> {

    if result < 0 {

        return Err(io::Error::last_os_error());

    } else {

        return Ok(result);

    }

}

#[cfg(not(unix))]

struct Poller {

}

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::{Component, wake_up_if_sleeping, CompPDL, CompCtx};

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

use super::scheduler::*;

impl CompKey {

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

        return CompId(self.0);

    }

}

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

pub struct CompId(pub u32);

impl CompId {

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

        return CompId(u32::MAX);

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

    #[cfg(debug_assertions)] decremented: bool,

}

impl CompHandle {

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

        let handle = CompHandle{

            #[cfg(debug_assertions)] decremented: false,

        };

        handle.increment_users();

        return handle;

    }