vj_sim/state/

gvt.rs

//! Task 12 — GVT gating: speculative-immediate for Scheduler, committed-only
//! for Waveform/VPI/`$display` (ledger ST-075, ST-078, ST-081).
//!
//! Two asymmetric visibility rules, both already wired at the primitives
//! this task assembles:
//!
//! - **Scheduler** (ST-075): notified eagerly, unconditionally, the instant
//!   `write::commit_write` sees a real change — never gated on
//!   [`StateEngine::advance_gvt`] at all. See `Dispatch::notify_scheduler`'s
//!   own doc comment (`state::dispatch`) for why this is correct even once a
//!   future wave enables real speculation (ARCHITECTURE §4): in v1's
//!   conservative `LOOKAHEAD_CYCLES = 0` engine there is no outstanding
//!   speculation to hide, and the scheduler's own re-execution-then-possible-
//!   rollback *is* what speculation means once it exists.
//! - **Waveform/VPI/`$display`** (ST-078): observe committed changes
//!   exclusively through [`StateEngine::drain_committed_changes`] below,
//!   which filters every registered worker thread's speculative buffer
//!   (`write::VcState`, Task 8/9) down to records with `at <= up_to` — a
//!   value written but not yet passed by the caller-supplied GVT watermark
//!   is invisible through this path even though a direct
//!   [`super::SignalAccess::read`] would already show it (that distinction
//!   is this row's entire point, and is what
//!   `display_style_observation_never_sees_a_pre_gvt_value_even_if_committed_later`
//!   below proves).
//!
//! [`StateEngine::advance_gvt`] (ST-081) is a pure setter with no dependency
//! on any scheduler-internal type — STATE only buffers/gates on GVT, it
//! never computes GVT itself; that computation is ENGINE's job, pushed in
//! from outside this crate.

use vj_abi::SimTime;

use super::instance::StateEngine;
use super::write::VcRecord;

impl StateEngine {
    /// ST-081: a pure setter (`self.vc.set_gvt_watermark(gvt)`) — no
    /// dependency on any scheduler-internal type, callable with a bare
    /// `SimTime` the caller (ENGINE, later) computed however it likes. See
    /// `write::VcState`'s `gvt_watermark` field doc comment for why this
    /// field exists even though `drain_committed_changes` below takes its
    /// own explicit `up_to` parameter rather than reading this field back.
    ///
    /// **Task 20 (ST-124) — open joint sizing risk with ENGINE, not bounded
    /// by this crate alone.** Every un-drained speculative VC record between
    /// two `advance_gvt` calls stays buffered (`write::VcState`'s per-thread
    /// buffer, Task 8/9): this is only bounded in practice if the caller
    /// (ENGINE) advances GVT "cheaply and frequently enough" relative to the
    /// write rate. STATE has no way to observe or enforce ENGINE's actual
    /// GVT-advancement cadence from inside this crate — a coarse-grained
    /// cadence (e.g. only at major time-step boundaries) could make the
    /// buffering cost far larger than assumed here. Tracked as an open risk
    /// (`docs/design/state.md` Risks/Open Questions), not silently assumed
    /// safe; restated in `docs/CONTRACTS.md`'s "Wave A — wA-state notes"
    /// section so ENGINE's own wave can pick it up without re-deriving it
    /// from this doc comment.
    pub fn advance_gvt(&mut self, gvt: SimTime) {
        self.vc.set_gvt_watermark(gvt);
    }

    /// ST-078, ST-081: drains every registered worker thread's speculative
    /// buffer (review fix — this plan's header "cross-thread visibility"
    /// decision) of records with `at <= up_to`, merged by `(at, registration
    /// index)` — see `write::VcState::drain_committed_changes`'s own doc
    /// comment for the full removal/merge algorithm this forwards to.
    /// Waveform/VPI/`$display`-style consumers are meant to be the only
    /// callers of this method in a real simulation run.
    ///
    /// See [`StateEngine::advance_gvt`]'s doc comment (ST-124) for the
    /// buffer-growth sizing risk this method's drain rate must keep bounded.
    pub fn drain_committed_changes(&mut self, up_to: SimTime) -> Vec<VcRecord> {
        self.vc.drain_committed_changes(up_to)
    }

    /// Whether value changes are logged for [`Self::drain_committed_changes`]
    /// (on by default). A run whose observers read STATE directly and never
    /// drain must turn the log off, or it grows with every change of the run
    /// (ST-124: 70-130 GB on a few minutes of a real design).
    pub fn set_change_log(&mut self, on: bool) {
        self.vc.set_record_changes(on);
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::state::instance::TemplateRegistry;
    use crate::state::write::SlotLocation;
    use crate::state::{
        EdgeKind, FourStateValue, ListenerRole, SignalAccess, SlotClass, StateCellId,
        ValueChangeSink,
    };
    use std::sync::{Arc, Mutex};
    use vj_abi::Region;

    fn fresh_engine() -> StateEngine {
        StateEngine::new(TemplateRegistry::new())
    }

    /// Allocates one inline cell (mirrors `write::tests::inline_cell`,
    /// duplicated locally per this crate's existing per-task-module test
    /// helper convention).
    fn inline_cell(engine: &mut StateEngine) -> StateCellId {
        // SAFETY: local, single-owned engine; no other live borrow of the
        // inline plane, and every test below performs no concurrent
        // allocation on the same engine (the one cross-thread test
        // allocates both cells before spawning any thread).
        let word_offset = unsafe { engine.arena.alloc_inline_word() };
        let loc = SlotLocation {
            class: SlotClass::Inline,
            word_offset,
            bit_width: 8,
            elem_word_count: 1,
            row_stride_words: 0,
            num_rows: 0,
        };
        engine.alloc_cell(loc)
    }

    fn narrow_word(val: u64, unk: u64) -> u128 {
        (val as u128) | ((unk as u128) << 64)
    }

    fn t(units: i64) -> SimTime {
        SimTime {
            sim_units: units,
            delta: 0,
        }
    }

    /// Records every `notify_change` call it receives into a shared log.
    struct SpySink {
        log: Arc<Mutex<Vec<(StateCellId, SimTime)>>>,
    }

    impl ValueChangeSink for SpySink {
        fn notify_change(
            &mut self,
            cell: StateCellId,
            _edge: EdgeKind,
            at: SimTime,
            _region: Region,
        ) {
            self.log.lock().unwrap().push((cell, at));
        }
    }

    /// ST-075: register a scheduler spy; `commit_write` a real change
    /// *without* ever calling `advance_gvt`; the spy has already recorded
    /// the call. Fails if scheduler notification were routed through the
    /// same GVT-gated path as Waveform/VPI (it would never fire here, since
    /// `advance_gvt`/`drain_committed_changes` are never called in this
    /// test at all).
    #[test]
    fn scheduler_listener_fires_immediately_without_waiting_for_advance_gvt() {
        let mut engine = fresh_engine();
        let cell = inline_cell(&mut engine);
        let log = Arc::new(Mutex::new(Vec::new()));
        engine.subscribe(
            cell,
            ListenerRole::Scheduler,
            Box::new(SpySink { log: log.clone() }),
        );

        let result = engine.commit_write(cell, FourStateValue::Narrow(narrow_word(5, 0)), &mut ());
        assert!(result.changed);

        let calls = log.lock().unwrap();
        assert_eq!(
            calls.len(),
            1,
            "the Scheduler listener must fire immediately on commit_write, with no advance_gvt call anywhere in this test"
        );
        assert_eq!(calls[0].0, cell);
    }

    /// ST-124: with the log off a write still lands but leaves no record;
    /// turned back on, records resume.
    #[test]
    fn a_disabled_change_log_records_nothing() {
        let mut engine = fresh_engine();
        let cell = inline_cell(&mut engine);
        engine.set_change_log(false);
        engine.set_current_time(t(1));
        let r = engine.commit_write(cell, FourStateValue::Narrow(narrow_word(5, 0)), &mut ());
        assert!(r.changed, "the write itself still lands");
        assert!(engine.drain_committed_changes(t(100)).is_empty());
        engine.set_change_log(true);
        let r = engine.commit_write(cell, FourStateValue::Narrow(narrow_word(6, 0)), &mut ());
        assert!(r.changed);
        assert_eq!(engine.drain_committed_changes(t(100)).len(), 1);
    }

    /// ST-078, ST-081: two writes at different `SimTime`s (stamped via
    /// `set_current_time` before each `commit_write`, per this plan's own
    /// "how STATE learns now" design); `drain_committed_changes` with an
    /// `up_to` between the two `at`s returns only the earlier record, a
    /// repeat call with the same `up_to` returns nothing further (already
    /// drained), and the later record only appears once a later
    /// `drain_committed_changes` call raises `up_to` past it. Fails if
    /// gating compares the wrong direction (`>=` vs `<=`) or if drain
    /// doesn't actually remove what it returns.
    #[test]
    fn drain_committed_changes_excludes_records_after_the_watermark() {
        let mut engine = fresh_engine();
        let cell_a = inline_cell(&mut engine);
        let cell_b = inline_cell(&mut engine);

        engine.set_current_time(t(5));
        let ra = engine.commit_write(cell_a, FourStateValue::Narrow(narrow_word(1, 0)), &mut ());
        assert!(ra.changed);

        engine.set_current_time(t(10));
        let rb = engine.commit_write(cell_b, FourStateValue::Narrow(narrow_word(1, 0)), &mut ());
        assert!(rb.changed);

        engine.advance_gvt(t(7));
        let drained = engine.drain_committed_changes(t(7));
        assert_eq!(
            drained.len(),
            1,
            "only cell_a's at=5 record is <= the up_to=7 watermark"
        );
        assert_eq!(drained[0].cell, cell_a);
        assert_eq!(drained[0].at, t(5));

        let drained_again = engine.drain_committed_changes(t(7));
        assert!(
            drained_again.is_empty(),
            "a second drain with the same up_to must not re-return an already-drained record"
        );

        engine.advance_gvt(t(10));
        let drained_b = engine.drain_committed_changes(t(10));
        assert_eq!(
            drained_b.len(),
            1,
            "cell_b's at=10 record becomes visible once up_to reaches it"
        );
        assert_eq!(drained_b[0].cell, cell_b);
        assert_eq!(drained_b[0].at, t(10));
    }

    /// ST-081: a type-level proof, deliberately written with no
    /// `crate::sched`/`crate::determinism` import anywhere in this function
    /// (nor in this module) — `advance_gvt`'s signature takes only
    /// `SimTime`. Also asserts the setter actually stores the value (via
    /// `VcState`'s `#[cfg(test)]` getter), so this is a real, failable test,
    /// not merely a compile-only gate.
    #[test]
    fn advance_gvt_has_no_dependency_on_scheduler_internal_state() {
        let mut engine = fresh_engine();
        engine.advance_gvt(t(42));
        assert_eq!(
            engine.vc.gvt_watermark(),
            t(42),
            "advance_gvt must actually store the watermark it was given"
        );
    }

    /// ST-078: a value written via `commit_write` is immediately visible via
    /// `SignalAccess::read` (a direct-state query, not the commit-stream
    /// path this row is about — that's expected and fine), but is *not*
    /// visible via `drain_committed_changes` until `up_to` reaches its `at`.
    /// Fails if the commit-stream path leaked a pre-watermark value, making
    /// the gating-vs-direct-read distinction explicit and regression-tested.
    #[test]
    fn display_style_observation_never_sees_a_pre_gvt_value_even_if_committed_later() {
        let mut engine = fresh_engine();
        let cell = inline_cell(&mut engine);

        engine.set_current_time(t(5));
        let result = engine.commit_write(cell, FourStateValue::Narrow(narrow_word(9, 0)), &mut ());
        assert!(result.changed);

        match engine.read(cell) {
            FourStateValue::Narrow(w) => assert_eq!(
                w,
                narrow_word(9, 0),
                "a direct SignalAccess::read already sees the new value -- that's fine, it's not the commit-stream path"
            ),
            FourStateValue::Wide { .. } => panic!("expected Narrow"),
        }

        let too_early = engine.drain_committed_changes(t(3));
        assert!(
            too_early.is_empty(),
            "drain_committed_changes must reflect nothing past its own watermark, even though read() already shows the new value"
        );
    }

    /// Review fix, closes the "no task can ever exercise this" gap: build a
    /// `StateEngine` on the main thread, allocate two cells; spawn real OS
    /// thread T1, which writes cell A and is `join`ed to completion; spawn
    /// real OS thread T2 (T1's thread -- and its `thread_local!` storage --
    /// no longer exists at this point), which writes cell B and is `join`ed;
    /// back on the *main* thread (a third, distinct OS thread from both T1
    /// and T2), `drain_committed_changes` must see **both** records. Fails
    /// (returns only one, or neither) if `drain_committed_changes` read only
    /// a single `thread_local!` buffer instead of iterating the
    /// `worker_buffers` registry -- the exact bug this plan's header
    /// "cross-thread visibility" decision exists to fix, and one no
    /// single-threaded test above can ever catch.
    #[test]
    fn drain_committed_changes_sees_records_appended_from_a_different_real_os_thread() {
        let mut engine = fresh_engine();
        let cell_a = inline_cell(&mut engine);
        let cell_b = inline_cell(&mut engine);

        /// A bare pointer wrapper made `Send` on purpose: `*mut StateEngine`
        /// itself is not `Send`, but this test's own structure (T1 fully
        /// `join`ed before T2 is even spawned, and the main thread never
        /// touches `engine` again until both have finished) guarantees at
        /// most one thread ever dereferences the pointee at a time, so no
        /// two threads ever hold this pointer concurrently.
        struct SendPtr(*mut StateEngine);
        // SAFETY: see this struct's own doc comment above -- this test's
        // control flow (strict join-before-next-spawn, no concurrent access
        // from the main thread) is what makes sending this pointer across
        // threads sound.
        unsafe impl Send for SendPtr {}

        let ptr1 = SendPtr(&mut engine as *mut StateEngine);
        let t1 = std::thread::spawn(move || {
            let ptr1 = ptr1;
            // SAFETY: `SendPtr`'s doc comment above -- no other thread
            // touches `engine` while T1 runs.
            let engine_ref = unsafe { &mut *ptr1.0 };
            engine_ref.set_current_time(t(1));
            engine_ref.commit_write(cell_a, FourStateValue::Narrow(narrow_word(1, 0)), &mut ());
        });
        t1.join().unwrap();

        let ptr2 = SendPtr(&mut engine as *mut StateEngine);
        let t2 = std::thread::spawn(move || {
            let ptr2 = ptr2;
            // SAFETY: T1 has already been `join`ed (its OS thread, and its
            // `thread_local!` storage, no longer exists), and the main
            // thread has not touched `engine` in between -- still only one
            // thread holding the pointer.
            let engine_ref = unsafe { &mut *ptr2.0 };
            engine_ref.set_current_time(t(2));
            engine_ref.commit_write(cell_b, FourStateValue::Narrow(narrow_word(1, 0)), &mut ());
        });
        t2.join().unwrap();

        // Back on the main thread -- a third, distinct OS thread from both
        // T1 and T2.
        engine.advance_gvt(t(5));
        let drained = engine.drain_committed_changes(t(5));
        let cells: Vec<StateCellId> = drained.iter().map(|r| r.cell).collect();
        assert!(
            cells.contains(&cell_a),
            "T1's committed change must be visible from the draining thread"
        );
        assert!(
            cells.contains(&cell_b),
            "T2's committed change must be visible from the draining thread"
        );
        assert_eq!(
            drained.len(),
            2,
            "both worker threads' buffers must be drained, not just one"
        );
    }
}