forge_orchestration/scheduler/

deadline.rs

//! Tick-deadline scheduling discipline (EEVDF-as-frame-deadline).
//!
//! A simulation that must render/advance a frame every `tick` is a *soft
//! real-time* workload: the value of running it is highest right before its next
//! tick is due and drops (or is lost) once the tick is missed. Static priority
//! cannot express this — two cells with equal priority but different tick phases
//! must still be ordered by *which tick is due first*. This module orders
//! schedulables by an **earliest-virtual-deadline-first** rule, where the
//! virtual deadline is the simulation's next tick time.
//!
//! # EEVDF mapping
//!
//! EEVDF (Earliest Eligible Virtual Deadline First) schedules by two per-entity
//! quantities computed in a *virtual time* `v` that advances inversely to total
//! weight:
//!
//! - **lag** `l_i = w_i * (v - v_i)` — how far entity `i` is from its fair share.
//!   `l_i > 0` means it is owed service (under-served); `l_i < 0` means it has
//!   over-run. An entity is **eligible** only when `l_i >= 0` (`v >= v_i`,
//!   i.e. its `eligible_time` has arrived).
//! - **virtual deadline** `d_i = v_i + r_i / w_i` — eligible time plus the
//!   virtual cost of one request `r_i` scaled by weight `w_i`.
//!
//! EEVDF runs the **eligible** entity with the **earliest virtual deadline**.
//!
//! ## How we map a sim tick onto this
//!
//! | EEVDF concept            | This module                                              |
//! |--------------------------|----------------------------------------------------------|
//! | virtual deadline `d_i`   | `virtual_deadline` = the cell's **next tick wall time**  |
//! | eligible time `v_i`      | `eligible_time` = tick start = `next_deadline - tick`    |
//! | request size `r_i`       | one tick's wall budget = `tick`                          |
//! | weight `w_i`             | derived from static `priority` (`weight = priority+1>=1`) |
//! | lag `l_i`                | `weight * (now - eligible_time)` in milliseconds         |
//!
//! So instead of EEVDF's abstract virtual time we use **wall-clock time as the
//! virtual timeline** (every entity shares one global clock, weight 1 in time),
//! and let `weight` break ties and bias lag. This is the documented divergence
//! from textbook EEVDF (see *Differences* below) and is exact for the common
//! case where every cell advances on real wall-clock ticks.
//!
//! ## Ordering rule
//!
//! Eligible entities are ordered by `(virtual_deadline asc, weight desc,
//! sequence asc)`. Ineligible entities (tick not yet started) sort after all
//! eligible ones. This is **earliest-deadline-first among the runnable set**,
//! the EEVDF discipline restricted to our wall-clock virtual time.
//!
//! ## Missing a tick (overrun handling)
//!
//! When `now > virtual_deadline` the tick is **late**. [`TickOutcome`] selects
//! the policy:
//!
//! - [`MissPolicy::Late`] — run anyway, then re-arm the deadline forward
//!   (accumulating lag); good for replay/offline sims where every frame matters.
//! - [`MissPolicy::Drop`] — skip the missed frame, re-arm to the next future
//!   tick boundary; good for live/interactive sims (catch up, don't pile up).
//! - [`MissPolicy::Backpressure`] — refuse to enqueue further ticks until the
//!   backlog (`lag`) falls below a bound; surfaced as [`Eligibility::Backpressured`].
//!
//! # Differences from textbook EEVDF
//!
//! 1. **Virtual time is wall time.** Textbook EEVDF advances `v` as
//!    `dv = dt / W` (W = sum of weights) so the timeline stretches under load.
//!    We pin `v == now` (wall clock). Deadlines are therefore real instants the
//!    sim genuinely must hit, not load-relative virtual instants. Weight then
//!    only biases *lag* and *ties*, never the deadline scale.
//! 2. **Bounded, integer lag.** Lag is computed in whole milliseconds
//!    (`i64`), saturating, so ordering is total and panic-free — no float NaN
//!    can corrupt the heap (textbook EEVDF carries real-valued lag).
//! 3. **Explicit miss policy.** Classic EEVDF has no notion of dropping a
//!    request; a missed sim frame is a first-class outcome here.

use std::cmp::Ordering;
use std::collections::BinaryHeap;
use std::time::Duration;

use chrono::{DateTime, Utc};
use parking_lot::Mutex;

use super::algorithms::SchedulingAlgorithm;
use super::{NodeResources, Workload};

/// Policy for handling a tick whose deadline has already passed.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum MissPolicy {
    /// Run the late tick anyway; re-arm the deadline forward by one tick.
    Late,
    /// Skip the missed frame(s); re-arm to the next future tick boundary.
    Drop,
    /// Stop admitting new ticks while backlog exceeds `max_lag`.
    Backpressure {
        /// Maximum tolerated lag before backpressure engages.
        max_lag: Duration,
    },
}

impl Default for MissPolicy {
    fn default() -> Self {
        MissPolicy::Late
    }
}

/// Result of evaluating a tick's eligibility against `now`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Eligibility {
    /// Tick has not started yet (`now < eligible_time`); not runnable.
    Pending,
    /// Tick is runnable and on time (`eligible_time <= now <= virtual_deadline`).
    Eligible,
    /// Deadline already passed (`now > virtual_deadline`); the entity is late.
    Late,
    /// Backlog exceeds the configured bound; admission is paused.
    Backpressured,
}

/// What to do with a late tick, per [`MissPolicy`].
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TickOutcome {
    /// Run the tick now.
    Run,
    /// Drop this frame and advance to the next boundary.
    DropFrame,
    /// Apply backpressure (do not admit).
    Hold,
}

/// A schedulable carrying EEVDF-style timing. Cheap to copy except for the
/// owned [`Workload`].
#[derive(Debug, Clone)]
pub struct DeadlineEntry {
    /// The underlying workload (typically a sim cell member or its world anchor).
    pub workload: Workload,
    /// Eligible (tick-start) time `v_i`: when this entity *may* begin running.
    pub eligible_time: DateTime<Utc>,
    /// Virtual deadline `d_i`: the next tick's wall-clock due time.
    pub virtual_deadline: DateTime<Utc>,
    /// Tick cadence; one request's virtual cost `r_i`.
    pub tick: Duration,
    /// EEVDF weight `w_i` (>= 1), derived from static priority.
    pub weight: u32,
    /// FIFO tiebreaker assigned on enqueue.
    pub sequence: u64,
}

impl DeadlineEntry {
    /// Build an entry. `weight` is clamped to `>= 1`.
    pub fn new(
        workload: Workload,
        eligible_time: DateTime<Utc>,
        virtual_deadline: DateTime<Utc>,
        tick: Duration,
    ) -> Self {
        let weight = (workload.priority.max(0) as u32).saturating_add(1);
        Self {
            workload,
            eligible_time,
            virtual_deadline,
            tick,
            weight,
            sequence: 0,
        }
    }

    /// Construct directly from a sim cell's next tick.
    ///
    /// `next_deadline` is the cell's `next_deadline`; `eligible_time` is
    /// `next_deadline - tick` (the tick's start). This is the bridge from
    /// [`crate::scheduler::sim::SimCell`] timing into the deadline queue.
    pub fn from_tick(workload: Workload, next_deadline: DateTime<Utc>, tick: Duration) -> Self {
        let tick_chrono = chrono::Duration::from_std(tick)
            .unwrap_or_else(|_| chrono::Duration::milliseconds(0));
        let eligible_time = next_deadline - tick_chrono;
        Self::new(workload, eligible_time, next_deadline, tick)
    }

    /// Lag `l_i` in milliseconds: `weight * (now - eligible_time)`, saturating.
    ///
    /// Positive => owed service (eligible/under-served). Negative => its tick has
    /// not started yet. Bounded integer math keeps ordering total.
    pub fn lag_ms(&self, now: DateTime<Utc>) -> i64 {
        let elapsed = (now - self.eligible_time).num_milliseconds();
        elapsed.saturating_mul(self.weight as i64)
    }

    /// Classify the entity against `now` and a [`MissPolicy`].
    pub fn eligibility(&self, now: DateTime<Utc>, policy: MissPolicy) -> Eligibility {
        if now < self.eligible_time {
            return Eligibility::Pending;
        }
        if now <= self.virtual_deadline {
            return Eligibility::Eligible;
        }
        // now > virtual_deadline: late.
        if let MissPolicy::Backpressure { max_lag } = policy {
            let max_lag_ms = chrono::Duration::from_std(max_lag)
                .map(|d| d.num_milliseconds())
                .unwrap_or(i64::MAX);
            // Use unit-weight overrun for the backpressure bound (real time owed).
            let overrun_ms = (now - self.virtual_deadline).num_milliseconds();
            if overrun_ms > max_lag_ms {
                return Eligibility::Backpressured;
            }
        }
        Eligibility::Late
    }

    /// Decide what to do with this entity right now under `policy`.
    pub fn outcome(&self, now: DateTime<Utc>, policy: MissPolicy) -> TickOutcome {
        match self.eligibility(now, policy) {
            Eligibility::Eligible | Eligibility::Late => match policy {
                MissPolicy::Drop if now > self.virtual_deadline => TickOutcome::DropFrame,
                _ => TickOutcome::Run,
            },
            Eligibility::Backpressured => TickOutcome::Hold,
            Eligibility::Pending => TickOutcome::Hold,
        }
    }

    /// Re-arm timing after a tick is serviced.
    ///
    /// - `Late`: advance both eligible and deadline by exactly one tick
    ///   (preserve accumulated lag).
    /// - `Drop`: jump the deadline to the first future boundary at or after
    ///   `now`, discarding skipped frames; eligible time is `deadline - tick`.
    pub fn rearm(&mut self, now: DateTime<Utc>, policy: MissPolicy) {
        let tick_chrono = chrono::Duration::from_std(self.tick)
            .unwrap_or_else(|_| chrono::Duration::milliseconds(0));
        match policy {
            MissPolicy::Drop if self.tick.as_millis() > 0 && now > self.virtual_deadline => {
                // Smallest k >= 1 such that deadline + k*tick >= now.
                let behind_ms = (now - self.virtual_deadline).num_milliseconds().max(0);
                let tick_ms = tick_chrono.num_milliseconds().max(1);
                let k = behind_ms / tick_ms + 1;
                self.virtual_deadline += tick_chrono * (k as i32);
                self.eligible_time = self.virtual_deadline - tick_chrono;
            }
            _ => {
                self.eligible_time = self.virtual_deadline;
                self.virtual_deadline += tick_chrono;
            }
        }
    }
}

/// A min-heap ordering wrapper: earliest virtual deadline pops first.
///
/// [`BinaryHeap`] is a max-heap, so [`Ord`] here is *reversed* — "greater" means
/// "should run later". The full order is:
/// `(virtual_deadline asc, weight desc, sequence asc)`.
#[derive(Debug, Clone)]
struct Earliest(DeadlineEntry);

impl PartialEq for Earliest {
    fn eq(&self, other: &Self) -> bool {
        self.0.virtual_deadline == other.0.virtual_deadline
            && self.0.weight == other.0.weight
            && self.0.sequence == other.0.sequence
    }
}
impl Eq for Earliest {}

impl PartialOrd for Earliest {
    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
        Some(self.cmp(other))
    }
}

impl Ord for Earliest {
    fn cmp(&self, other: &Self) -> Ordering {
        // Reverse so the heap's max is our *least* urgent.
        // Earliest deadline = most urgent = should be heap-max.
        other
            .0
            .virtual_deadline
            .cmp(&self.0.virtual_deadline)
            // Higher weight is more urgent => heap-greater.
            .then_with(|| self.0.weight.cmp(&other.0.weight))
            // Lower sequence (older) is more urgent => heap-greater.
            .then_with(|| other.0.sequence.cmp(&self.0.sequence))
    }
}

/// EEVDF-style deadline-ordered queue. Pops the earliest-virtual-deadline entry.
///
/// Thread-safe via an internal [`parking_lot::Mutex`], matching the existing
/// [`super::queue::SchedulingQueue`] design.
pub struct DeadlineQueue {
    heap: Mutex<BinaryHeap<Earliest>>,
    sequence: Mutex<u64>,
    policy: MissPolicy,
}

impl DeadlineQueue {
    /// Create a queue with the given miss policy.
    pub fn new(policy: MissPolicy) -> Self {
        Self {
            heap: Mutex::new(BinaryHeap::new()),
            sequence: Mutex::new(0),
            policy,
        }
    }

    /// The queue's miss policy.
    pub fn policy(&self) -> MissPolicy {
        self.policy
    }

    /// Enqueue a tick. Assigns a FIFO sequence for tie-breaking.
    pub fn enqueue(&self, mut entry: DeadlineEntry) {
        let mut seq = self.sequence.lock();
        entry.sequence = *seq;
        *seq += 1;
        drop(seq);
        self.heap.lock().push(Earliest(entry));
    }

    /// Number of queued ticks.
    pub fn len(&self) -> usize {
        self.heap.lock().len()
    }

    /// Whether the queue is empty.
    pub fn is_empty(&self) -> bool {
        self.heap.lock().is_empty()
    }

    /// Peek the workload id of the most-urgent (earliest-deadline) entry.
    pub fn peek_id(&self) -> Option<String> {
        self.heap.lock().peek().map(|e| e.0.workload.id.clone())
    }

    /// Pop the most-urgent (earliest virtual deadline) entry unconditionally.
    pub fn pop_earliest(&self) -> Option<DeadlineEntry> {
        self.heap.lock().pop().map(|e| e.0)
    }

    /// Pop the most-urgent entry that is runnable *now* under the miss policy.
    ///
    /// Pending entries (tick not started) and backpressured entries are kept in
    /// the queue. Returns the entry plus the [`TickOutcome`] decided for it.
    /// `Drop`ped frames are returned too (the caller re-arms and re-enqueues);
    /// the outcome tells the caller whether to actually execute.
    pub fn pop_runnable(&self, now: DateTime<Utc>) -> Option<(DeadlineEntry, TickOutcome)> {
        let mut heap = self.heap.lock();
        let mut deferred: Vec<Earliest> = Vec::new();
        let result = loop {
            match heap.pop() {
                Some(top) => {
                    let outcome = top.0.outcome(now, self.policy);
                    match outcome {
                        TickOutcome::Hold => {
                            // Pending or backpressured: keep it, look deeper.
                            deferred.push(top);
                        }
                        TickOutcome::Run | TickOutcome::DropFrame => {
                            break Some((top.0, outcome));
                        }
                    }
                }
                None => break None,
            }
        };
        for d in deferred {
            heap.push(d);
        }
        result
    }
}

/// A thin [`SchedulingAlgorithm`] that biases node scoring toward deadline-tagged
/// workloads, so urgent ticks land on the emptiest (lowest-latency) node.
///
/// Ordering of *which* tick to run is the [`DeadlineQueue`]'s job; this is the
/// *placement* half: given a chosen tick, prefer the node that will start it
/// soonest (least contended). It wraps `inner` for the base score and adds an
/// urgency bonus based on the workload's static priority (the deadline itself is
/// queue-side state, not visible on a bare [`Workload`]).
pub struct TickDeadlineScheduler<A: SchedulingAlgorithm> {
    inner: A,
    /// Weight of the urgency bonus added to the inner score.
    urgency_weight: f64,
}

impl<A: SchedulingAlgorithm> TickDeadlineScheduler<A> {
    /// Wrap an inner algorithm with a default urgency weight of `1.0`.
    ///
    /// At `1.0`, a maximally-urgent workload's headroom bonus can fully override
    /// the inner algorithm's packing preference, so urgent ticks land on the
    /// emptiest (lowest-latency-to-start) node. Lower it toward `0` to make
    /// urgency only a tiebreaker on top of the inner score.
    pub fn new(inner: A) -> Self {
        Self {
            inner,
            urgency_weight: 1.0,
        }
    }

    /// Set the urgency weight.
    pub fn with_urgency_weight(mut self, w: f64) -> Self {
        self.urgency_weight = w;
        self
    }
}

impl<A: SchedulingAlgorithm> SchedulingAlgorithm for TickDeadlineScheduler<A> {
    fn score(&self, workload: &Workload, node: &NodeResources) -> f64 {
        let base = self.inner.score(workload, node);
        // Urgent (high-priority/short-tick) workloads prefer the *emptiest* node
        // so they start immediately: reward available CPU headroom.
        let headroom = if node.cpu_capacity > 0 {
            node.cpu_available() as f64 / node.cpu_capacity as f64
        } else {
            0.0
        };
        // Bounded weight derived from static priority (0..1).
        let urgency = (workload.priority.max(0) as f64 / 1000.0).min(1.0);
        base + self.urgency_weight * urgency * headroom
    }

    fn name(&self) -> &str {
        "tick-deadline"
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::scheduler::algorithms::BinPackScheduler;
    use crate::types::NodeId;

    fn at(ms: i64) -> DateTime<Utc> {
        // A fixed epoch + ms; deterministic for ordering tests.
        DateTime::<Utc>::from_timestamp(1_700_000_000, 0).unwrap()
            + chrono::Duration::milliseconds(ms)
    }

    fn entry(id: &str, deadline_ms: i64, priority: i32) -> DeadlineEntry {
        let wl = Workload::new(id, id).with_priority(priority);
        DeadlineEntry::from_tick(wl, at(deadline_ms), Duration::from_millis(50))
    }

    #[test]
    fn pops_earliest_deadline_first() {
        let q = DeadlineQueue::new(MissPolicy::Late);
        // Enqueue out of deadline order.
        q.enqueue(entry("c", 300, 0));
        q.enqueue(entry("a", 100, 0));
        q.enqueue(entry("b", 200, 0));

        assert_eq!(q.pop_earliest().unwrap().workload.id, "a");
        assert_eq!(q.pop_earliest().unwrap().workload.id, "b");
        assert_eq!(q.pop_earliest().unwrap().workload.id, "c");
    }

    #[test]
    fn equal_deadline_breaks_ties_by_weight_then_sequence() {
        let q = DeadlineQueue::new(MissPolicy::Late);
        // Same deadline; higher priority (weight) should pop first.
        q.enqueue(entry("low", 100, 1));
        q.enqueue(entry("high", 100, 500));
        q.enqueue(entry("low2", 100, 1)); // ties low on weight; later sequence

        assert_eq!(q.pop_earliest().unwrap().workload.id, "high");
        // Between the two weight-1 entries, the earlier-enqueued ("low") wins.
        assert_eq!(q.pop_earliest().unwrap().workload.id, "low");
        assert_eq!(q.pop_earliest().unwrap().workload.id, "low2");
    }

    #[test]
    fn eligibility_classifies_pending_eligible_late() {
        // Deadline at t=200, tick=50 => eligible_time at t=150.
        let e = entry("x", 200, 0);
        assert_eq!(e.eligibility(at(100), MissPolicy::Late), Eligibility::Pending);
        assert_eq!(e.eligibility(at(175), MissPolicy::Late), Eligibility::Eligible);
        assert_eq!(e.eligibility(at(200), MissPolicy::Late), Eligibility::Eligible);
        assert_eq!(e.eligibility(at(250), MissPolicy::Late), Eligibility::Late);
    }

    #[test]
    fn lag_is_weighted_and_bounded() {
        // priority 0 => weight 1; priority 3 => weight 4.
        let e0 = entry("a", 200, 0); // eligible at 150
        let e3 = entry("b", 200, 3);
        // 50ms past eligible.
        assert_eq!(e0.lag_ms(at(200)), 50);
        assert_eq!(e3.lag_ms(at(200)), 200); // 50ms * weight 4
        // Before eligible => negative.
        assert!(e0.lag_ms(at(100)) < 0);
    }

    #[test]
    fn pop_runnable_skips_pending_and_returns_eligible() {
        let q = DeadlineQueue::new(MissPolicy::Late);
        // "future" deadline at 1000 (eligible 950) is most-urgent by deadline?
        // No: smaller deadline = more urgent. Make the pending one the smallest
        // deadline so we prove pop_runnable skips it despite being heap-top.
        q.enqueue(entry("pending", 100, 0)); // eligible 50
        q.enqueue(entry("ready", 500, 0)); // eligible 450

        // now = 460: "pending" deadline 100 < now (late, runnable);
        // To make pending truly pending, push its eligibility into the future.
        // Re-do with explicit times:
        let q2 = DeadlineQueue::new(MissPolicy::Late);
        let future = DeadlineEntry::from_tick(
            Workload::new("future", "future"),
            at(1000), // eligible 950
            Duration::from_millis(50),
        );
        let ready = DeadlineEntry::from_tick(
            Workload::new("ready", "ready"),
            at(500), // eligible 450
            Duration::from_millis(50),
        );
        q2.enqueue(future);
        q2.enqueue(ready);

        // now = 470: future is Pending (now < 950), ready is Eligible.
        let (got, outcome) = q2.pop_runnable(at(470)).unwrap();
        assert_eq!(got.workload.id, "ready");
        assert_eq!(outcome, TickOutcome::Run);
        // The pending "future" remains queued.
        assert_eq!(q2.len(), 1);
        assert_eq!(q2.peek_id().as_deref(), Some("future"));

        // Silence unused warnings from the first scratch queue.
        let _ = q.len();
    }

    #[test]
    fn drop_policy_advances_past_missed_frames() {
        let policy = MissPolicy::Drop;
        // Deadline 200, tick 50, eligible 150. now = 380 => 180ms late => skipped
        // frames; next boundary should be the first deadline >= 380.
        let mut e = entry("d", 200, 0);
        assert_eq!(e.outcome(at(380), policy), TickOutcome::DropFrame);
        e.rearm(at(380), policy);
        // behind = 180ms, tick = 50 => k = 180/50 + 1 = 4; 200 + 4*50 = 400.
        assert!(e.virtual_deadline >= at(380));
        assert_eq!((e.virtual_deadline - at(0)).num_milliseconds(), 400);
        assert_eq!((e.eligible_time - at(0)).num_milliseconds(), 350);
    }

    #[test]
    fn late_policy_advances_exactly_one_tick() {
        let policy = MissPolicy::Late;
        let mut e = entry("l", 200, 0);
        assert_eq!(e.outcome(at(280), policy), TickOutcome::Run);
        e.rearm(at(280), policy);
        // Late: eligible <- old deadline (200); deadline <- 250.
        assert_eq!((e.eligible_time - at(0)).num_milliseconds(), 200);
        assert_eq!((e.virtual_deadline - at(0)).num_milliseconds(), 250);
    }

    #[test]
    fn backpressure_holds_when_lag_exceeds_bound() {
        let policy = MissPolicy::Backpressure {
            max_lag: Duration::from_millis(100),
        };
        // Deadline 200; now = 500 => 300ms overrun > 100ms bound => backpressure.
        let e = entry("bp", 200, 0);
        assert_eq!(e.eligibility(at(500), policy), Eligibility::Backpressured);
        assert_eq!(e.outcome(at(500), policy), TickOutcome::Hold);

        // Within bound (250 => 50ms overrun) => just Late, runs.
        assert_eq!(e.eligibility(at(250), policy), Eligibility::Late);
        assert_eq!(e.outcome(at(250), policy), TickOutcome::Run);
    }

    #[test]
    fn tick_deadline_scheduler_prefers_emptier_node_for_urgent() {
        let sched = TickDeadlineScheduler::new(BinPackScheduler::new());
        let urgent = Workload::new("u", "u").with_priority(900);

        let mut empty = NodeResources::new(NodeId::new(), 4000, 8192);
        let mut busy = NodeResources::new(NodeId::new(), 4000, 8192);
        busy.cpu_allocated = 3000;
        empty.cpu_allocated = 0;

        let s_empty = sched.score(&urgent, &empty);
        let s_busy = sched.score(&urgent, &busy);
        // Urgency bonus rewards headroom => empty node scores higher overall for
        // an urgent workload, overriding bin-pack's lean toward the busy node.
        assert!(
            s_empty > s_busy,
            "urgent tick should prefer emptier node: empty={s_empty} busy={s_busy}"
        );
        assert_eq!(sched.name(), "tick-deadline");
    }
}