nexosim 1.0.0

//! Scheduling functions and types.
use std::error::Error;
use std::fmt;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;

use crate::ports::ReplyReader;
use crate::simulation::queue_items::{Event, EventId, EventKey, Query, QueryId, QueueItem};
use crate::time::{AtomicTimeReader, ClockReader, Deadline, MonotonicTime};
use crate::util::priority_queue::PriorityQueue;

#[cfg(all(test, not(nexosim_loom)))]
use crate::{time::AtomicTime, time::TearableAtomicTime, util::sync_cell::SyncCell};

use super::GLOBAL_ORIGIN_ID;

/// A scheduler for events and queries meant to be processed at specified
/// deadlines.
///
/// The `Scheduler` handle is `Clone`-able and can be shared or sent to other
/// threads.
///
/// When scheduling an event or query, it is important to consider that its
/// deadline must be in the future of the current simulation time and that
/// stepping method such as
/// [`Simulation::step`](crate::simulation::Simulation::step) or
/// [`Simulation::run`](crate::simulation::Simulation::run) eagerly advance the
/// simulation time to the deadline of the next scheduled event or simulation
/// tick. If a stepping method is executed concurrently, therefore, events or
/// queries can only be scheduled after the deadline associated with the next
/// scheduler event or simulation tick.
#[derive(Clone, Debug)]
pub struct Scheduler(GlobalScheduler);

impl Scheduler {
    /// Creates a new scheduler.
    pub(crate) fn new(
        scheduler_queue: Arc<Mutex<SchedulerQueue>>,
        time: AtomicTimeReader,
        is_halted: Arc<AtomicBool>,
    ) -> Self {
        Self(GlobalScheduler::new(scheduler_queue, time, is_halted))
    }

    /// Creates a dummy scheduler (for testing purposes only).
    #[cfg(all(test, not(nexosim_loom)))]
    #[allow(dead_code)]
    pub(crate) fn dummy() -> Self {
        let time = AtomicTime::new(TearableAtomicTime::new(MonotonicTime::EPOCH)).reader();
        let scheduler_queue = Arc::new(Mutex::new(SchedulerQueue::new()));
        let is_halted = Arc::new(AtomicBool::default());

        Self(GlobalScheduler::new(scheduler_queue, time, is_halted))
    }

    /// Returns the current simulation time.
    ///
    /// # Examples
    ///
    /// ```
    /// use nexosim::simulation::Scheduler;
    /// use nexosim::time::MonotonicTime;
    ///
    /// fn is_third_millennium(scheduler: &Scheduler) -> bool {
    ///     let time = scheduler.time();
    ///     time >= MonotonicTime::new(978307200, 0).unwrap()
    ///         && time < MonotonicTime::new(32535216000, 0).unwrap()
    /// }
    /// ```
    pub fn time(&self) -> MonotonicTime {
        self.0.time()
    }

    /// Schedules a readily-built event at a future time.
    #[cfg(feature = "server")]
    pub(crate) fn schedule(
        &self,
        deadline: impl Deadline,
        event: Event,
    ) -> Result<(), SchedulingError> {
        self.0.schedule_from(deadline, event, GLOBAL_ORIGIN_ID)
    }

    /// Schedules an event at a future time.
    ///
    /// An error is returned if the specified time is not in the future of the
    /// current simulation time.
    ///
    /// Events scheduled for the same time and targeting the same model are
    /// guaranteed to be processed according to the scheduling order.
    pub fn schedule_event<T>(
        &self,
        deadline: impl Deadline,
        event_id: &EventId<T>,
        arg: T,
    ) -> Result<(), SchedulingError>
    where
        T: Send + Clone + 'static,
    {
        self.0
            .schedule_event_from(deadline, event_id, arg, GLOBAL_ORIGIN_ID)
    }

    /// Schedules a cancellable event at a future time and returns an event key.
    ///
    /// An error is returned if the specified time is not in the future of the
    /// current simulation time.
    ///
    /// Events scheduled for the same time and targeting the same model are
    /// guaranteed to be processed according to the scheduling order.
    pub fn schedule_keyed_event<T>(
        &self,
        deadline: impl Deadline,
        event_id: &EventId<T>,
        arg: T,
    ) -> Result<EventKey, SchedulingError>
    where
        T: Send + Clone + 'static,
    {
        self.0
            .schedule_keyed_event_from(deadline, event_id, arg, GLOBAL_ORIGIN_ID)
    }

    /// Schedules a periodically recurring event at a future time.
    ///
    /// An error is returned if the specified time is not in the future of the
    /// current simulation time, or if the specified period is null.
    ///
    /// Events scheduled for the same time and targeting the same model are
    /// guaranteed to be processed according to the scheduling order.
    pub fn schedule_periodic_event<T>(
        &self,
        deadline: impl Deadline,
        period: Duration,
        event_id: &EventId<T>,
        arg: T,
    ) -> Result<(), SchedulingError>
    where
        T: Send + Clone + 'static,
    {
        self.0
            .schedule_periodic_event_from(deadline, period, event_id, arg, GLOBAL_ORIGIN_ID)
    }

    /// Schedules a cancellable, periodically recurring event at a future time
    /// and returns an event key.
    ///
    /// An error is returned if the specified time is not in the future of the
    /// current simulation time, or if the specified period is null.
    ///
    /// Events scheduled for the same time and targeting the same model are
    /// guaranteed to be processed according to the scheduling order.
    pub fn schedule_keyed_periodic_event<T>(
        &self,
        deadline: impl Deadline,
        period: Duration,
        event_id: &EventId<T>,
        arg: T,
    ) -> Result<EventKey, SchedulingError>
    where
        T: Send + Clone + 'static,
    {
        self.0
            .schedule_keyed_periodic_event_from(deadline, period, event_id, arg, GLOBAL_ORIGIN_ID)
    }

    /// Schedules a query at a future time.
    ///
    /// An error is returned if the specified time is not in the future of the
    /// current simulation.
    ///
    /// Queries scheduled for the same time and targeting the same model are
    /// guaranteed to be processed according to the scheduling order.
    pub fn schedule_query<T, R>(
        &self,
        deadline: impl Deadline,
        query_id: &QueryId<T, R>,
        arg: T,
    ) -> Result<ReplyReader<R>, SchedulingError>
    where
        T: Send + Clone + 'static,
        R: Send + 'static,
    {
        self.0
            .schedule_query_from(deadline, query_id, arg, GLOBAL_ORIGIN_ID)
    }

    /// Requests the simulation to be interrupted at the earliest opportunity.
    ///
    /// If a stepping method such as
    /// [`Simulation::step`](crate::simulation::Simulation::step) or
    /// [`Simulation::run`](crate::simulation::Simulation::run) is concurrently
    /// being executed, this will cause such method to return before it steps to
    /// next scheduler deadline or simulation tick (if any) with
    /// [`ExecutionError::Halted`](crate::simulation::ExecutionError::Halted).
    ///
    /// Otherwise, this will cause the next call to a `Simulation::step*` or
    /// `Simulation::process*` method to return immediately with
    /// [`ExecutionError::Halted`](crate::simulation::ExecutionError::Halted).
    ///
    /// In all cases, once
    /// [`ExecutionError::Halted`](crate::simulation::ExecutionError::Halted) is
    /// returned, the simulation can be resumed at any moment with another call
    /// to a stepping method or a
    /// [`Simulation::process_*`](crate::simulation::Simulation::process_event)
    /// methods.
    pub fn halt(&self) {
        self.0.halt()
    }
}

/// An error returned when the scheduled time or the repetition period are
/// invalid.
#[derive(Debug, PartialEq, Eq, Clone, Copy)]
#[non_exhaustive]
pub enum SchedulingError {
    /// The scheduled time does not lie in the future of the current simulation
    /// time.
    InvalidScheduledTime,
    /// The repetition period is zero.
    NullRepetitionPeriod,
}

impl fmt::Display for SchedulingError {
    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::InvalidScheduledTime => write!(
                fmt,
                "the scheduled time should be in the future of the current simulation time"
            ),
            Self::NullRepetitionPeriod => write!(fmt, "the repetition period cannot be zero"),
        }
    }
}

impl Error for SchedulingError {}

/// Alias for the scheduler queue type.
///
/// Why use both time and origin as the key? The short answer is that this
/// allows the preservation of the relative ordering of events which have the
/// same origin (where the origin is either a model instance or the global
/// scheduler). The preservation of this ordering is implemented by the event
/// loop, which aggregate events with the same origin into single sequential
/// futures, thus ensuring that they are not executed concurrently.
pub(crate) type SchedulerQueue = PriorityQueue<SchedulerKey, QueueItem>;

pub(crate) type SchedulerKey = (MonotonicTime, usize);

/// Internal implementation of the global scheduler.
#[derive(Clone)]
pub(crate) struct GlobalScheduler {
    scheduler_queue: Arc<Mutex<SchedulerQueue>>,
    time: AtomicTimeReader,
    is_halted: Arc<AtomicBool>,
}

impl GlobalScheduler {
    pub(crate) fn new(
        scheduler_queue: Arc<Mutex<SchedulerQueue>>,
        time: AtomicTimeReader,
        is_halted: Arc<AtomicBool>,
    ) -> Self {
        Self {
            scheduler_queue,
            time,
            is_halted,
        }
    }

    /// Returns the current simulation time.
    pub(crate) fn time(&self) -> MonotonicTime {
        // We use `read` rather than `try_read` because the scheduler can be
        // sent to another thread than the simulator's and could thus
        // potentially see a torn read if the simulator increments time
        // concurrently. The chances of this happening are very small since
        // simulation time is not changed frequently.
        self.time.read()
    }

    /// Returns a clock reader.
    pub(crate) fn clock_reader(&self) -> ClockReader {
        ClockReader::from_atomic_time_reader(&self.time)
    }

    /// Schedules a readily-built event identified by its origin at a future
    /// time.
    #[cfg(feature = "server")]
    pub(crate) fn schedule_from(
        &self,
        deadline: impl Deadline,
        event: Event,
        origin_id: usize,
    ) -> Result<(), SchedulingError> {
        // The scheduler queue must always be locked when reading the time,
        // otherwise the following race could occur:
        // 1) this method reads the time and concludes that it is not too late to
        //    schedule the action,
        // 2) the `Simulation` object takes the lock, increments simulation time and
        //    runs the simulation step,
        // 3) this method takes the lock and schedules the now-outdated action.
        let mut scheduler_queue = self.scheduler_queue.lock().unwrap();

        let now = self.time();
        let time = deadline.into_time(now);
        if now >= time {
            return Err(SchedulingError::InvalidScheduledTime);
        }

        scheduler_queue.insert((time, origin_id), QueueItem::Event(event));

        Ok(())
    }

    /// Schedules an event identified by its idntifier and origin at a future
    /// time.
    pub(crate) fn schedule_event_from<T>(
        &self,
        deadline: impl Deadline,
        event_id: &EventId<T>,
        arg: T,
        origin_id: usize,
    ) -> Result<(), SchedulingError>
    where
        T: Send + Clone + 'static,
    {
        // The scheduler queue must always be locked when reading the time (see
        // `schedule_from`).
        let mut scheduler_queue = self.scheduler_queue.lock().unwrap();
        let now = self.time();
        let time = deadline.into_time(now);
        if now >= time {
            return Err(SchedulingError::InvalidScheduledTime);
        }

        let event = Event::new(event_id, arg);
        scheduler_queue.insert((time, origin_id), QueueItem::Event(event));

        Ok(())
    }

    /// Schedules a cancellable event identified by its identifier and origin at
    /// a future time and returns an event key.
    pub(crate) fn schedule_keyed_event_from<T>(
        &self,
        deadline: impl Deadline,
        event_id: &EventId<T>,
        arg: T,
        origin_id: usize,
    ) -> Result<EventKey, SchedulingError>
    where
        T: Send + Clone + 'static,
    {
        let event_key = EventKey::new();

        // The scheduler queue must always be locked when reading the time (see
        // `schedule_from`).
        let mut scheduler_queue = self.scheduler_queue.lock().unwrap();
        let now = self.time();
        let time = deadline.into_time(now);
        if now >= time {
            return Err(SchedulingError::InvalidScheduledTime);
        }

        let event = Event::new(event_id, arg).with_key(event_key.clone());
        scheduler_queue.insert((time, origin_id), QueueItem::Event(event));

        Ok(event_key)
    }

    /// Schedules a periodically recurring event identified by its id and origin
    /// at a future time.
    pub(crate) fn schedule_periodic_event_from<T>(
        &self,
        deadline: impl Deadline,
        period: Duration,
        event_id: &EventId<T>,
        arg: T,
        origin_id: usize,
    ) -> Result<(), SchedulingError>
    where
        T: Send + Clone + 'static,
    {
        if period.is_zero() {
            return Err(SchedulingError::NullRepetitionPeriod);
        }

        // The scheduler queue must always be locked when reading the time (see
        // `schedule_from`).
        let mut scheduler_queue = self.scheduler_queue.lock().unwrap();
        let now = self.time();
        let time = deadline.into_time(now);
        if now >= time {
            return Err(SchedulingError::InvalidScheduledTime);
        }

        let event = Event::new(event_id, arg).with_period(period);
        scheduler_queue.insert((time, origin_id), QueueItem::Event(event));

        Ok(())
    }

    /// Schedules a cancellable, periodically recurring event identified by its
    /// id and origin at a future time and returns an event key.
    pub(crate) fn schedule_keyed_periodic_event_from<T>(
        &self,
        deadline: impl Deadline,
        period: Duration,
        event_id: &EventId<T>,
        arg: T,
        origin_id: usize,
    ) -> Result<EventKey, SchedulingError>
    where
        T: Send + Clone + 'static,
    {
        if period.is_zero() {
            return Err(SchedulingError::NullRepetitionPeriod);
        }
        let event_key = EventKey::new();

        // The scheduler queue must always be locked when reading the time (see
        // `schedule_from`).
        let mut scheduler_queue = self.scheduler_queue.lock().unwrap();
        let now = self.time();
        let time = deadline.into_time(now);
        if now >= time {
            return Err(SchedulingError::InvalidScheduledTime);
        }

        let event = Event::new(event_id, arg)
            .with_period(period)
            .with_key(event_key.clone());
        scheduler_queue.insert((time, origin_id), QueueItem::Event(event));

        Ok(event_key)
    }

    /// Schedules a query identified by its id and origin at a future time.
    pub(crate) fn schedule_query_from<T, R>(
        &self,
        deadline: impl Deadline,
        query_id: &QueryId<T, R>,
        arg: T,
        origin_id: usize,
    ) -> Result<ReplyReader<R>, SchedulingError>
    where
        T: Send + Clone + 'static,
        R: Send + 'static,
    {
        // The scheduler queue must always be locked when reading the time (see
        // `schedule_from`).
        let mut scheduler_queue = self.scheduler_queue.lock().unwrap();
        let now = self.time();
        let time = deadline.into_time(now);
        if now >= time {
            return Err(SchedulingError::InvalidScheduledTime);
        }

        let (query, rx) = Query::new(query_id, arg);
        scheduler_queue.insert((time, origin_id), QueueItem::Query(query));

        Ok(rx)
    }

    /// Requests the simulation to return as early as possible upon the
    /// completion of the current time step.
    pub(crate) fn halt(&self) {
        self.is_halted.store(true, Ordering::Relaxed);
    }
}

impl fmt::Debug for GlobalScheduler {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("GlobalScheduler")
            .field("time", &self.time())
            .field("is_halted", &self.is_halted.load(Ordering::Relaxed))
            .finish_non_exhaustive()
    }
}

#[cfg(all(test, not(nexosim_loom)))]
impl GlobalScheduler {
    /// Creates a dummy scheduler for testing purposes.
    pub(crate) fn new_dummy() -> Self {
        let dummy_priority_queue = Arc::new(Mutex::new(SchedulerQueue::new()));
        let dummy_time = SyncCell::new(TearableAtomicTime::new(MonotonicTime::EPOCH)).reader();
        let dummy_running = Arc::new(AtomicBool::new(false));
        GlobalScheduler::new(dummy_priority_queue, dummy_time, dummy_running)
    }
}