sv_sim_sim/sched/

queues.rs

//! SCHED::QUEUES — region queues & the Active/Inactive/NBA delta-cycle
//! convergence primitives (scheduler.md §1; ARCHITECTURE §6.3; ledger SC-017,
//! SC-018, SC-019, SC-021, SC-022).
//!
//! This module owns the four region queues (`Active`, `Inactive`, `Reactive`,
//! `Postponed` — NBA has no queue of its own, it is Task 6's staging log) and
//! the primitives an outer driver (Task 21's `run_until`) composes into the
//! actual Active/Inactive/NBA fixpoint loop and the Reactive/Postponed
//! exactly-once-per-timestep passes. This module deliberately does **not**
//! drain Active/Inactive to a fixpoint internally — that orchestration,
//! including the livelock guard (Task 15), stays external and pluggable
//! (this task's own acceptance note).

use std::collections::VecDeque;

use sv_sim_abi::Region;

use crate::sched::PendingActivation;

/// `push`'s one typed-error variant (SC-036's "loud, typed, never silent"
/// posture, applied here to NBA's non-queue). Never a `debug_assert!` — that
/// compiles to nothing in release builds, which is exactly the silent-drop
/// this type exists to prevent (see [`RegionQueues::push`]'s doc comment).
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub enum RegionQueuePushError {
    /// NBA has no queue of its own — it is [`crate::sched`]'s NBA staging log
    /// (Task 6), not a `RegionQueues` member. Pushing `Region::Nba` here is
    /// always a caller bug, never a legitimate no-op.
    NbaHasNoQueue,
    /// **sva T1 (§1.4, review fix, blocking finding 2).** `Region::Preponed`
    /// has no queue of its own either — it is a pure snapshot push
    /// (`state::snapshot::SnapshotRing`), fired as an unconditional inline
    /// side effect at time-slot entry (`sched::control::Engine::run_until`/
    /// `step_region`), never a region a `CompiledProcess` is "ready" in.
    /// Pushing `Region::Preponed` here is always a caller bug, the same
    /// class `NbaHasNoQueue` already names for `Region::Nba`.
    PreponedHasNoQueue,
    /// **sva T2 (§1.4, review fix, blocking finding 2), Observed's half.**
    /// `Region::Observed` has no queue of its own either -- a live SVA
    /// thread is owned by the automaton engine's own thread table (a later
    /// task), never dispatched through `RegionQueues`/`ProcessTable` as a
    /// `PendingActivation`: the Observed-region engine step is reached only
    /// through the region-callback mechanism (`Engine::push_region_
    /// callback`/`drain_region_callbacks`), never a queued activation.
    /// Pushing `Region::Observed` here is always a caller bug, the same
    /// class `NbaHasNoQueue`/`PreponedHasNoQueue` already name.
    ObservedHasNoQueue,
}

impl std::fmt::Display for RegionQueuePushError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            RegionQueuePushError::NbaHasNoQueue => {
                write!(
                    f,
                    "Region::Nba has no RegionQueues queue; route NBA writes through the NBA staging log (Task 6) instead"
                )
            }
            RegionQueuePushError::PreponedHasNoQueue => {
                write!(
                    f,
                    "Region::Preponed has no RegionQueues queue; it is a pure snapshot push \
                     (state::snapshot::SnapshotRing), never a region with ready processes"
                )
            }
            RegionQueuePushError::ObservedHasNoQueue => {
                write!(
                    f,
                    "Region::Observed has no RegionQueues queue; a live SVA thread is owned by \
                     the automaton engine's own thread table, never a region with ready \
                     processes dispatched as a PendingActivation"
                )
            }
        }
    }
}

impl std::error::Error for RegionQueuePushError {}

/// The four region queues (SC-017..SC-022). `active`/`inactive` are
/// `VecDeque` (FIFO-drained, and re-populated *during* draining by whatever
/// processes the drained batch — that re-population is what makes the outer
/// fixpoint loop possible); `reactive`/`postponed` are plain `Vec`s since
/// they are drained exactly once per timestep and never need FIFO semantics
/// mid-drain (SC-018, SC-019).
#[derive(Default)]
pub struct RegionQueues {
    active: VecDeque<PendingActivation>,
    inactive: VecDeque<PendingActivation>,
    reactive: Vec<PendingActivation>,
    /// LRM 4.4.2.7 Re-Inactive: a `#0` taken by a Reactive-region process.
    /// Not a [`Region`] variant (the ABI enum stays at seven); the scheduler
    /// drains it directly once Reactive is empty.
    re_inactive: VecDeque<PendingActivation>,
    postponed: Vec<PendingActivation>,
}

impl RegionQueues {
    pub fn new() -> Self {
        RegionQueues::default()
    }

    /// Routes `act` to its region's queue. Exhaustive `match region` over all
    /// 7 [`Region`] variants, no wildcard arm: `Active`/`Inactive`/
    /// `Reactive`/`Postponed` route to their own queue and return `Ok(())`;
    /// `Nba`/`Preponed`/`Observed` return their own named
    /// `Err(RegionQueuePushError)` variant rather than a `debug_assert!`
    /// (which compiles to nothing — and so silently drops the activation
    /// with no queue, no panic, no error — in a release build). `push` must
    /// return `Result` so misuse is loud in every build profile and every
    /// caller (Tasks 11, 13, 15, 21) must handle/propagate it.
    pub fn push(
        &mut self,
        region: Region,
        act: PendingActivation,
    ) -> Result<(), RegionQueuePushError> {
        match region {
            Region::Active => {
                self.active.push_back(act);
                Ok(())
            }
            Region::Inactive => {
                self.inactive.push_back(act);
                Ok(())
            }
            Region::Nba => Err(RegionQueuePushError::NbaHasNoQueue),
            Region::Reactive => {
                self.reactive.push(act);
                Ok(())
            }
            Region::Postponed => {
                self.postponed.push(act);
                Ok(())
            }
            Region::Preponed => Err(RegionQueuePushError::PreponedHasNoQueue),
            // sva T2 (§1.4, review fix, blocking finding 2), Observed's
            // half: queue-less for the identical reason `Preponed` is.
            Region::Observed => Err(RegionQueuePushError::ObservedHasNoQueue),
        }
    }

    /// Pops everything currently queued in Active, preserving FIFO order.
    /// Pushes made *during* draining (by whatever calls this) land in the
    /// queue for the *next* `drain_active` call, not this one — this is what
    /// makes the outer fixpoint loop possible.
    pub fn drain_active(&mut self) -> Vec<PendingActivation> {
        self.active.drain(..).collect()
    }

    /// Pops everything currently queued in Inactive, preserving FIFO order.
    /// Same "not this call's own later pushes" semantics as
    /// [`Self::drain_active`].
    pub fn drain_inactive(&mut self) -> Vec<PendingActivation> {
        self.inactive.drain(..).collect()
    }

    /// `true` only once both Active and Inactive are empty *at the moment of
    /// the call* — callers must re-check after each drain round, since a
    /// drain's own processing may re-populate either queue (SC-017, SC-022's
    /// joint fixpoint).
    pub fn active_and_inactive_both_empty(&self) -> bool {
        self.active.is_empty() && self.inactive.is_empty()
    }

    /// `true` when nothing is queued in Reactive at the moment of the call
    /// (LRM 4.5: the Reactive region runs until it is empty, and a process it
    /// wakes in Reactive -- a checker's -- joins it).
    pub fn reactive_is_empty(&self) -> bool {
        self.reactive.is_empty()
    }

    /// Pops everything currently queued in Reactive. Intended to be called
    /// exactly once per timestep by the outer driver (Task 21 wires the
    /// "exactly once" part; this module only provides the primitive and
    /// proves it drains everything present, including zero items, without
    /// erroring). Returns `[]`, never panics, when Reactive is empty (SC-018,
    /// SC-019 — "usually-empty... drained exactly once per timestep even when
    /// empty" must not need special-casing at call sites).
    pub fn drain_reactive_once(&mut self) -> Vec<PendingActivation> {
        std::mem::take(&mut self.reactive)
    }

    /// Queues `act` in the Re-Inactive region (LRM 4.4.2.7).
    pub fn push_re_inactive(&mut self, act: PendingActivation) {
        self.re_inactive.push_back(act);
    }

    /// Pops everything currently queued in Re-Inactive, FIFO.
    pub fn drain_re_inactive(&mut self) -> Vec<PendingActivation> {
        self.re_inactive.drain(..).collect()
    }

    /// `true` when nothing is queued in Re-Inactive.
    pub fn re_inactive_is_empty(&self) -> bool {
        self.re_inactive.is_empty()
    }

    /// Pops everything currently queued in Postponed. Same "exactly once,
    /// even when empty, never errors" semantics as
    /// [`Self::drain_reactive_once`].
    pub fn drain_postponed_once(&mut self) -> Vec<PendingActivation> {
        std::mem::take(&mut self.postponed)
    }
}

/// Pure helper: normalizes `Suspend::Delay(0)` to land in `Region::Inactive`
/// rather than re-entering the time wheel (SC-021 — "Delay(0) normalized here
/// rather than re-entering the wheel"; this helper IS that normalization
/// point, called by Task 5's wheel-arming code before a wheel insert is even
/// attempted). Only `Region::Active` redirects: a caller already targeting
/// `Inactive` (or any other region) with a zero delay is left alone — Inactive
/// itself never re-redirects, so there is no infinite Active->Inactive bounce.
pub fn redirect_delay_zero_to_inactive(region: Region, delay_ticks: u64) -> Region {
    if region == Region::Active && delay_ticks == 0 {
        Region::Inactive
    } else {
        region
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::sched::{ResumeArg, SchedTime};

    fn act(id: u64, proc: u32, ticks: u64, region: Region, reg_order: u64) -> PendingActivation {
        PendingActivation {
            id,
            proc,
            arg: ResumeArg::Start,
            reconv_iter: 0,
            logical_order: (SchedTime { ticks }, region, reg_order),
        }
    }

    #[test]
    fn active_drains_fifo_and_is_empty_after() {
        let mut q = RegionQueues::new();
        q.push(Region::Active, act(1, 1, 0, Region::Active, 0))
            .unwrap();
        q.push(Region::Active, act(2, 2, 0, Region::Active, 1))
            .unwrap();
        q.push(Region::Active, act(3, 3, 0, Region::Active, 2))
            .unwrap();

        let drained = q.drain_active();
        let ids: Vec<u64> = drained.iter().map(|a| a.id).collect();
        assert_eq!(ids, vec![1, 2, 3], "drain must preserve FIFO push order");

        assert!(
            q.drain_active().is_empty(),
            "a second drain must return [] — the queue must be cleared, not just iterated"
        );
    }

    #[test]
    fn push_to_nba_returns_a_typed_error_never_silently_drops() {
        let mut q = RegionQueues::new();
        let a = act(1, 1, 0, Region::Nba, 0);
        let result = q.push(Region::Nba, a);
        assert_eq!(result, Err(RegionQueuePushError::NbaHasNoQueue));

        // No queue's contents may have changed — the rejection must be a
        // loud, typed, side-effect-free error, not a debug_assert!-in-debug/
        // silent-drop-in-release gate.
        assert!(q.drain_active().is_empty());
        assert!(q.drain_inactive().is_empty());
        assert!(q.drain_reactive_once().is_empty());
        assert!(q.drain_postponed_once().is_empty());
    }

    /// sva T1 (§1.4, review fix, blocking finding 2): `Region::Preponed` is
    /// the second queue-less region, mirroring `Region::Nba`'s own existing
    /// `push_to_nba_returns_a_typed_error_never_silently_drops` test exactly
    /// -- a named, typed `Result::Err`, never a silent no-op or a panic.
    #[test]
    fn region_queues_push_preponed_returns_named_error() {
        let mut q = RegionQueues::new();
        let a = act(1, 1, 0, Region::Preponed, 0);
        let result = q.push(Region::Preponed, a);
        assert_eq!(result, Err(RegionQueuePushError::PreponedHasNoQueue));

        // No queue's contents may have changed.
        assert!(q.drain_active().is_empty());
        assert!(q.drain_inactive().is_empty());
        assert!(q.drain_reactive_once().is_empty());
        assert!(q.drain_postponed_once().is_empty());
    }

    /// sva T2 (§1.4, review fix, blocking finding 2), Observed's half:
    /// mirrors `region_queues_push_preponed_returns_named_error` exactly --
    /// a named, typed `Result::Err`, never a silent no-op or a panic.
    #[test]
    fn region_queues_push_observed_returns_named_error() {
        let mut q = RegionQueues::new();
        let a = act(1, 1, 0, Region::Observed, 0);
        let result = q.push(Region::Observed, a);
        assert_eq!(result, Err(RegionQueuePushError::ObservedHasNoQueue));

        // No queue's contents may have changed.
        assert!(q.drain_active().is_empty());
        assert!(q.drain_inactive().is_empty());
        assert!(q.drain_reactive_once().is_empty());
        assert!(q.drain_postponed_once().is_empty());
    }

    #[test]
    fn inactive_is_purely_a_landing_zone_for_delay_zero() {
        assert_eq!(
            redirect_delay_zero_to_inactive(Region::Active, 0),
            Region::Inactive
        );
        assert_eq!(
            redirect_delay_zero_to_inactive(Region::Active, 1),
            Region::Active
        );
        assert_eq!(
            redirect_delay_zero_to_inactive(Region::Nba, 0),
            Region::Nba,
            "only Active redirects — Inactive itself never re-redirects, no infinite bounce"
        );
    }

    #[test]
    fn active_inactive_reach_a_joint_fixpoint() {
        // Round 1: one item queued in Active.
        let mut q = RegionQueues::new();
        q.push(Region::Active, act(1, 1, 0, Region::Active, 0))
            .unwrap();
        assert!(
            !q.active_and_inactive_both_empty(),
            "work is queued — the pair must not report converged yet"
        );

        // Processing the Active batch pushes a new item to Inactive.
        let batch = q.drain_active();
        assert_eq!(batch.len(), 1);
        q.push(Region::Inactive, act(2, 2, 0, Region::Inactive, 0))
            .unwrap();
        assert!(
            !q.active_and_inactive_both_empty(),
            "Inactive was just re-populated — must not be computed from a stale snapshot"
        );

        // Processing the Inactive batch pushes a new item back to Active.
        let batch = q.drain_inactive();
        assert_eq!(batch.len(), 1);
        q.push(Region::Active, act(3, 3, 0, Region::Active, 0))
            .unwrap();
        assert!(
            !q.active_and_inactive_both_empty(),
            "Active was just re-populated by Inactive's own processing — the loop must keep re-checking both queues"
        );

        // Round 3: drain Active with no further pushes; the pair converges.
        let batch = q.drain_active();
        assert_eq!(batch.len(), 1);
        assert!(
            q.active_and_inactive_both_empty(),
            "both queues are now empty and nothing re-populated them — this is the fixpoint"
        );
    }

    #[test]
    fn reactive_and_postponed_drain_exactly_once_even_when_empty() {
        let mut q = RegionQueues::new();
        assert!(q.drain_reactive_once().is_empty());
        assert!(q.drain_postponed_once().is_empty());
    }

    #[test]
    fn activity_scheduled_during_reactive_does_not_rerun_reactive_this_timestep() {
        let mut q = RegionQueues::new();
        q.push(Region::Reactive, act(1, 1, 0, Region::Reactive, 0))
            .unwrap();

        let drained = q.drain_reactive_once();
        assert_eq!(drained.len(), 1);

        // While "processing" the drained item, new activity is scheduled
        // into Reactive again.
        q.push(Region::Reactive, act(2, 2, 0, Region::Reactive, 0))
            .unwrap();

        // This module exposes no "drain reactive again" method for the same
        // timestep — only the outer driver (Task 21) may call
        // drain_reactive_once once per timestep. The newly-pushed item must
        // still be sitting in the queue (genuinely deferred to next
        // timestep), never lost.
        let drained_again = q.drain_reactive_once();
        assert_eq!(
            drained_again.iter().map(|a| a.id).collect::<Vec<_>>(),
            vec![2],
            "the item pushed during processing must survive to the next drain, not be lost"
        );
    }
}