sv_sim_sim/sched/

dedup.rs

//! SCHED::DEDUP — the `queued_this_pass` per-process dedup gate: the exact
//! `if !process.queued_this_pass.swap(true, Ordering::AcqRel) { ... }` idiom
//! from scheduler.md §4, plus the `notify_change`-time wiring that combines
//! it with Task 9's [`SensitivityTable::matches`] (scheduler.md §4/§Risks;
//! ledger SC-055, SC-056).
//!
//! A process sensitive to several signals that all change in the same
//! Active/Inactive/NBA delta pass must be pushed onto the ready set exactly
//! once (SC-055) — otherwise it would run twice for what LRM semantics treats
//! as a single reactivation. The dedup flag's lifecycle is
//! enqueue -> (stays set across any further matches this pass) -> reset at
//! *dequeue* time, not at enqueue time (SC-056): resetting on enqueue would
//! make a legitimate re-trigger arriving in the gap between a process being
//! popped off the ready set and its compiled kernel re-registering its next
//! sensitivity look identical to "already queued, skip it", silently
//! swallowing a real stimulus. [`QueuedGate::reset_on_dequeue`] is therefore
//! the caller's responsibility to invoke the instant a process is actually
//! popped off the ready set to run, strictly *before* `resume()` is called.

use std::collections::HashMap;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Mutex, RwLock};

use crate::sched::ProcessId;
use crate::sched::sensitivity::SensitivityTable;
use crate::state::{EdgeKind, StateCellId};

/// Per-process "already queued this pass" flags (scheduler.md §4's literal
/// `queued_this_pass` bit, lifted out of any one process's storage into its
/// own gate so `sensitivity_match_to_enqueue` can dedup across every cell a
/// process is sensitive to, not just one).
///
/// The flags are a `Vec<AtomicBool>` indexed by `ProcessId` behind an
/// `RwLock`: both methods take `&self` and must create a flag on first touch
/// (no registration step pre-populates them), which needs the write lock
/// once per process; every later touch takes only the read lock, and the
/// flag itself is flipped and reset by a genuine atomic swap/store (SC-056's
/// acceptance: "the `AtomicBool` must be real, not a `Cell<bool>`"), so
/// `QueuedGate` stays `Send + Sync` with no lock needed around it. (It was a
/// `Mutex<HashMap<ProcessId, Arc<AtomicBool>>>`, whose hash, lock and `Arc`
/// clone on every match cost 3-5% of a fan-out-heavy run.)
pub struct QueuedGate {
    /// Indexed by `ProcessId` (dense from 0), grown on first touch; a read
    /// lock suffices for every flag that already exists, so the per-pass hot
    /// path is one uncontended read lock and one atomic.
    flags: RwLock<Vec<AtomicBool>>,
    /// A process id too large to index densely (never produced by the
    /// engine's own sequential allocation, but accepted).
    sparse: Mutex<HashMap<ProcessId, bool>>,
}

/// Ids at or above this go to [`QueuedGate::sparse`].
const DENSE_LIMIT: ProcessId = 1 << 24;

impl QueuedGate {
    pub fn new() -> Self {
        Self {
            flags: RwLock::new(Vec::new()),
            sparse: Mutex::new(HashMap::new()),
        }
    }

    /// Runs `f` on `proc`'s flag, creating it (`false`) on first touch.
    fn with_flag<R>(&self, proc: ProcessId, f: impl FnOnce(&AtomicBool) -> R) -> R {
        if proc >= DENSE_LIMIT {
            let mut sparse = self.sparse.lock().expect("QueuedGate mutex poisoned");
            let flag = AtomicBool::new(*sparse.get(&proc).unwrap_or(&false));
            let r = f(&flag);
            sparse.insert(proc, flag.into_inner());
            return r;
        }
        let i = proc as usize;
        {
            let flags = self.flags.read().expect("QueuedGate lock poisoned");
            if let Some(flag) = flags.get(i) {
                return f(flag);
            }
        }
        let mut flags = self.flags.write().expect("QueuedGate lock poisoned");
        if flags.len() <= i {
            flags.resize_with(i + 1, || AtomicBool::new(false));
        }
        f(&flags[i])
    }

    /// The exact `if !process.queued_this_pass.swap(true, Ordering::AcqRel)
    /// { ... }` idiom from scheduler.md §4: atomically marks `proc` as
    /// queued and returns `true` iff *this* call is the one that transitioned
    /// the flag from unqueued to queued — i.e. iff this call is the one that
    /// should push `proc` onto `active_q`. A second call for the same `proc`
    /// before the flag is reset returns `false` (SC-055's "queued only
    /// once").
    pub fn mark_queued_if_not_already(&self, proc: ProcessId) -> bool {
        self.with_flag(proc, |flag| !flag.swap(true, Ordering::AcqRel))
    }

    /// Clears `proc`'s queued flag. Callers invoke this the instant a
    /// process is popped off the ready set to run, **before** `resume()` is
    /// called — never at enqueue time (SC-056) — so a legitimate re-trigger
    /// landing in the dequeue-to-re-registration gap is correctly seen as a
    /// fresh, unqueued state and re-queues the process rather than being
    /// silently swallowed.
    pub fn reset_on_dequeue(&self, proc: ProcessId) {
        self.with_flag(proc, |flag| flag.store(false, Ordering::Release));
    }
}

impl Default for QueuedGate {
    fn default() -> Self {
        Self::new()
    }
}

/// The actual `notify_change`-time dedup step: combines Task 9's
/// [`SensitivityTable::matches`] with `gate`, returning only the
/// [`ProcessId`]s that were *not* already queued this pass — i.e. exactly
/// the set that should be pushed onto the ready set for this one committed
/// change. Every `ProcessId` `matches` yields for `cell`/`edge` is passed
/// through `gate.mark_queued_if_not_already`, so a process matched by more
/// than one cell/edge combination in the same pass appears in the returned
/// `Vec` at most once, on whichever call first marks it queued.
pub fn sensitivity_match_to_enqueue(
    table: &SensitivityTable,
    gate: &QueuedGate,
    cell: StateCellId,
    edge: EdgeKind,
) -> Vec<ProcessId> {
    // Task 3 (ledger SC-052) widened `SensitivityTable::matches`'s item type
    // to `(ProcessId, usize, Option<u64>)`, and Task 5 (SC-068) widened it
    // again to `(ProcessId, usize, Option<u64>, Option<PredicateFn>)` — so
    // `EngineNotifySink::notify_change` can see a `Repeat` term's
    // `term_index`/declared count and a `Dynamic` entry's `iff` predicate.
    // This dedup gate has no repeat-counting or `iff`-evaluation role
    // (both happen one layer up, in `sched::control`, before a matched
    // firing ever reaches here as something to enqueue) and only ever
    // needed the `ProcessId`.
    table
        .matches(cell, edge)
        .map(|(proc, _, _, _)| proc)
        .filter(|&proc| gate.mark_queued_if_not_already(proc))
        .collect()
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::sched::StaticSensitivity;

    const PROC_MAIN: ProcessId = 1;

    /// SC-055 — an `always_comb`-shaped process sensitive to two signals
    /// (`a`, `b`) that both change in the same Active pass must be queued
    /// exactly once: the first `sensitivity_match_to_enqueue` call (for `a`)
    /// returns `[proc]`, the second (for `b`, against the *same* gate)
    /// returns `[]` because the process is already queued this pass.
    #[test]
    fn process_sensitive_to_two_signals_changing_same_pass_queues_once() {
        let a = StateCellId(1);
        let b = StateCellId(2);

        let mut table = SensitivityTable::new();
        table.install_static(
            &StaticSensitivity {
                edges: vec![(a, None)],
            },
            PROC_MAIN,
        );
        table.install_static(
            &StaticSensitivity {
                edges: vec![(b, None)],
            },
            PROC_MAIN,
        );

        let gate = QueuedGate::new();

        let first = sensitivity_match_to_enqueue(&table, &gate, a, EdgeKind::None);
        assert_eq!(
            first,
            vec![PROC_MAIN],
            "the first match this pass must be the one that enqueues the process"
        );

        let second = sensitivity_match_to_enqueue(&table, &gate, b, EdgeKind::None);
        assert!(
            second.is_empty(),
            "a second signal changing in the same pass must not re-enqueue an already-queued process"
        );
    }

    /// SC-056 — the flag's lifecycle is enqueue -> (stays set across
    /// multiple matches) -> dequeue-reset, not enqueue -> immediate-reset.
    #[test]
    fn reset_on_dequeue_happens_at_dequeue_not_enqueue() {
        let gate = QueuedGate::new();

        assert!(
            gate.mark_queued_if_not_already(PROC_MAIN),
            "first mark must succeed (transitions unqueued -> queued)"
        );
        assert!(
            !gate.mark_queued_if_not_already(PROC_MAIN),
            "a repeat mark before any dequeue-reset must report already-queued"
        );

        gate.reset_on_dequeue(PROC_MAIN);

        assert!(
            gate.mark_queued_if_not_already(PROC_MAIN),
            "after reset_on_dequeue, the process must be markable as queued again"
        );
    }

    /// SC-056 — the literal "stimulus timing that lands exactly in the gap"
    /// test: mark queued, dequeue-reset (the process is now "running", about
    /// to re-register its next sensitivity, which in this fixture hasn't
    /// happened yet), then — *before* any re-registration — a legitimate new
    /// change arrives and marks the process queued again. This must succeed:
    /// a buggy "reset on enqueue instead of dequeue" implementation would
    /// have already reset the flag before step (2) even runs, making its own
    /// step (3) indistinguishable from correct — ordering the reset strictly
    /// between the two `mark_queued_if_not_already` calls is what catches
    /// that exact class of bug.
    #[test]
    fn retrigger_arriving_in_the_dequeue_to_reregister_gap_is_not_swallowed() {
        let gate = QueuedGate::new();

        // (1) queued.
        assert!(gate.mark_queued_if_not_already(PROC_MAIN));

        // (2) dequeued to run — reset happens here, not at step (1)'s enqueue.
        gate.reset_on_dequeue(PROC_MAIN);

        // (3) before re-registration, a legitimate new change arrives.
        assert!(
            gate.mark_queued_if_not_already(PROC_MAIN),
            "a retrigger landing in the dequeue-to-re-registration gap must \
             still be able to re-queue the process, not be silently eaten"
        );
    }

    /// Guards against an implementation that accidentally gates on
    /// `StateCellId` instead of `ProcessId`: two *different* processes both
    /// sensitive to the same cell must both appear in the one
    /// `sensitivity_match_to_enqueue` call for that cell's change — the gate
    /// is per-`ProcessId`, so neither one's dedup entry should suppress the
    /// other.
    #[test]
    fn dedup_does_not_falsely_eat_a_second_legitimate_process() {
        let cell = StateCellId(5);
        const PROC_B: ProcessId = 2;

        let mut table = SensitivityTable::new();
        table.install_static(
            &StaticSensitivity {
                edges: vec![(cell, None)],
            },
            PROC_MAIN,
        );
        table.install_static(
            &StaticSensitivity {
                edges: vec![(cell, None)],
            },
            PROC_B,
        );

        let gate = QueuedGate::new();

        let mut hits = sensitivity_match_to_enqueue(&table, &gate, cell, EdgeKind::None);
        hits.sort_unstable();
        assert_eq!(
            hits,
            vec![PROC_MAIN, PROC_B],
            "both processes sensitive to the same cell must be enqueued on its change"
        );
    }
}