vj_sim/sched/

nba.rs

//! SCHED::NBA — the nonblocking-assignment (`<=`) staging log (scheduler.md §1/§3
//! Key Decisions/Key Data Structures; ARCHITECTURE §6.3; ledger SC-034, SC-035,
//! SC-036, SC-037).
//!
//! `NbaLog` is the *only* place a `<=` write goes before it becomes visible.
//! `stage` never touches signal storage — it has no `&mut dyn SignalAccess`
//! parameter to call through even if it wanted to (SC-036) — so reading a cell
//! immediately after staging an NBA write to it, in the same process, still
//! observes the pre-assignment value (SC-034). The log accumulates every
//! staged write, unapplied, for the whole timestep; only the outer driver's
//! NBA-region commit pass (Task 7's `commit_nba`, once Active and Inactive are
//! both empty) ever calls [`NbaLog::take`] and applies the drained batch, which
//! is what makes multiple processes' NBA writes in one timestep become visible
//! together rather than one at a time (SC-035) — this module's own
//! contribution to that property is that nothing here has any partial-apply
//! path: every entry sits inert in `entries` until a single `take()` drains
//! all of them at once. The full SC-035 atomic-visibility acceptance test
//! (multiple processes' writes becoming visible together, not incrementally)
//! needs a real commit pass to observe "visible", so it is Task 7's
//! `commit_nba` test, not this module's; see this task's own commit message
//! for the exact split.
//!
//! `NbaLog` is cleared on every `now` advance (SC-037) — this module provides
//! the primitive (`clear`); Task 21's outer driver calls it at the right point
//! in the delta cycle.

use crate::sched::ProcessId;
use crate::state::{FourStateValue, StateCellId};

/// One staged nonblocking-assignment write (SC-037's exact field list, `bytes`
/// renamed `value: FourStateValue` to reuse the already-contracted STATE type
/// rather than inventing a `FourStateBits` alias).
#[derive(Clone, Debug)]
pub struct NbaUpdate {
    pub cell: StateCellId,
    pub value: FourStateValue,
    pub source: ProcessId,
    pub order: u32,
}

/// The flat per-timestep NBA staging log (SC-034/SC-035/SC-036/SC-037).
///
/// Structurally cannot apply a write: `stage`'s signature takes no `&mut dyn
/// SignalAccess` (or any other write-capable handle), so there is no call
/// this type could make even if its body were wrong — the only way an
/// `NbaUpdate` ever reaches signal storage is via [`NbaLog::take`] plus code
/// *outside* this module (Task 7's `commit_nba`) that actually writes it
/// (SC-036).
#[derive(Debug, Default)]
pub struct NbaLog {
    entries: Vec<NbaUpdate>,
    masked: Vec<MaskedNbaUpdate>,
    /// Shared staging counter for `entries` and `masked`, so the commit pass
    /// can apply one cell's whole and partial writes in staging order.
    next_order: u32,
}

/// A staged nonblocking write to PART of a cell -- the bits set in `mask`
/// (`q[3:0] <= v`, `{a, b} <= v`) -- and/or to one row of a memory cell
/// (`mem[i] <= v`, `row: Some(i)`). `value` carries the new bits already
/// shifted into position; bits outside `mask` are ignored. `mask` has one
/// word per 64 bits of the cell (or row) -- or, when `window` is `Some(w)`
/// (a write to part of a huge vector), one word per word of the span
/// starting at plane word `w`, `value` and `mask` covering only that span.
#[derive(Clone, Debug)]
pub struct MaskedNbaUpdate {
    pub cell: StateCellId,
    pub row: Option<u64>,
    pub value: FourStateValue,
    pub mask: Vec<u64>,
    pub window: Option<u32>,
    pub source: ProcessId,
    pub order: u32,
    /// A nonblocking event trigger (`->> e`, LRM 15.5.2): the update is not a
    /// value to merge but "trigger the named event" -- the commit pass bumps
    /// the event cell's trigger count and stamps it with the commit time
    /// (`value` and `mask` are unused).
    pub pulse: bool,
    /// A nonblocking write of one element of the queue or dynamic array
    /// whose handle the cell (or row) holds (`a[2][k] <= v`, LRM 1800-2023
    /// 10.4.2, 7.10): the commit pass looks the handle up *then*, so an
    /// out-of-range `index` is ignored at update time. `mask` carries the
    /// element's words; `value` is unused.
    pub coll: Option<CollWrite>,
    /// A nonblocking write of (part of) a class property
    /// (`obj.p <= v`, `obj.v[3:0] <= v`, LRM 1800-2023 10.4.2, 8.x): the
    /// update names no cell, only the object, which the staged write keeps
    /// alive until it commits. `value` and `mask` are unused.
    pub obj: Option<ObjWrite>,
}

/// The property bits a staged class-property write replaces (see
/// [`MaskedNbaUpdate::obj`]). Holds one reference on `h`, released when the
/// write commits ([`ObjWrite::commit`]).
#[derive(Clone, Debug)]
pub struct ObjWrite {
    pub h: u64,
    pub offset: usize,
    /// The property's packed words, the new bits already in position.
    pub val: Vec<u64>,
    /// The bits of `val` that are written; the rest of the property stays.
    pub mask: Vec<u64>,
    /// Not a property write: a nonblocking trigger (`->> e`, LRM 15.5.1) of
    /// class event `h` (`sched::class_event`); `offset`, `val` and `mask` are
    /// unused and `h` is no object (nothing is retained).
    pub class_event: bool,
}

impl ObjWrite {
    /// Retains the object for the write's lifetime.
    pub fn new(h: u64, offset: usize, val: Vec<u64>, mask: Vec<u64>) -> Self {
        crate::objects::retain(h);
        Self {
            h,
            offset,
            val,
            mask,
            class_event: false,
        }
    }

    /// A staged `->> e` of class event `id`.
    pub fn class_event(id: u64) -> Self {
        Self {
            h: id,
            offset: 0,
            val: Vec::new(),
            mask: Vec::new(),
            class_event: true,
        }
    }

    /// Merges the bits into the property (a no-op if the object is gone, which
    /// the retain prevents) and drops the reference.
    pub fn commit(self) {
        debug_assert!(
            !self.class_event,
            "a class-event trigger commits via commit_nba"
        );
        crate::objects::field_merge(self.h, self.offset, &self.val, &self.mask);
        crate::objects::release(self.h);
    }
}

/// The element a staged collection write targets (see
/// [`MaskedNbaUpdate::coll`]).
#[derive(Clone, Debug)]
pub struct CollWrite {
    pub shape: crate::collections::Shape,
    pub index: i64,
    /// An associative array's key (`m[k] <= v`, evaluated when staged):
    /// the element is found by it, not by `index`.
    pub key: Option<crate::collections::Key>,
    /// A queue: writing `index == size` appends.
    pub queue: bool,
}

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

    fn bump_order(&mut self) -> u32 {
        let order = self.next_order;
        self.next_order += 1;
        order
    }

    /// Appends a staged whole-cell write, ordered after every write staged
    /// before it (SC-034 — staging only, this call never applies anything).
    pub fn stage(&mut self, cell: StateCellId, value: FourStateValue, source: ProcessId) {
        let order = self.bump_order();
        self.entries.push(NbaUpdate {
            cell,
            value,
            source,
            order,
        });
    }

    /// Appends a staged partial / memory-row write (see [`MaskedNbaUpdate`]).
    pub fn stage_masked(
        &mut self,
        cell: StateCellId,
        row: Option<u64>,
        value: FourStateValue,
        mask: Vec<u64>,
        window: Option<u32>,
        source: ProcessId,
    ) {
        let order = self.bump_order();
        self.masked.push(MaskedNbaUpdate {
            cell,
            row,
            value,
            mask,
            window,
            source,
            order,
            pulse: false,
            coll: None,
            obj: None,
        });
    }

    /// Appends a staged nonblocking write of element `coll.index` (`elem`:
    /// its words) of the collection whose handle `cell` (row `row`) holds.
    pub fn stage_coll(
        &mut self,
        cell: StateCellId,
        row: Option<u64>,
        elem: Vec<u64>,
        coll: CollWrite,
        source: ProcessId,
    ) {
        let order = self.bump_order();
        self.masked.push(MaskedNbaUpdate {
            cell,
            row,
            value: FourStateValue::Narrow(0),
            mask: elem,
            window: None,
            source,
            order,
            pulse: false,
            coll: Some(coll),
            obj: None,
        });
    }

    /// Appends a staged nonblocking write of class-property bits.
    pub fn stage_obj(&mut self, w: ObjWrite, source: ProcessId) {
        let order = self.bump_order();
        self.masked.push(MaskedNbaUpdate {
            cell: StateCellId(0),
            row: None,
            value: FourStateValue::Narrow(0),
            mask: Vec::new(),
            window: None,
            source,
            order,
            pulse: false,
            coll: None,
            obj: Some(w),
        });
    }

    /// Appends a staged nonblocking event trigger (`->> e`) of event `cell`
    /// (or of row `row` of an event array's memory cell), ordered with every
    /// other staged write.
    pub fn stage_pulse(&mut self, cell: StateCellId, row: Option<u64>, source: ProcessId) {
        let order = self.bump_order();
        self.masked.push(MaskedNbaUpdate {
            cell,
            row,
            value: FourStateValue::Narrow(0),
            mask: Vec::new(),
            window: None,
            source,
            order,
            pulse: true,
            coll: None,
            obj: None,
        });
    }

    pub fn is_empty(&self) -> bool {
        self.entries.is_empty() && self.masked.is_empty()
    }

    pub fn len(&self) -> usize {
        self.entries.len() + self.masked.len()
    }

    /// Empties the log (SC-037: called by the outer driver on `now` advance).
    /// Idempotent — clearing an already-empty log is a no-op, never a panic.
    pub fn clear(&mut self) {
        self.entries.clear();
        self.masked.clear();
        self.next_order = 0;
    }

    /// Drains and returns every staged whole-cell entry (what Task 7's
    /// commit pass consumes). A genuine drain, not a clone: a second `take()`
    /// with no intervening `stage()` returns `[]`, so the same entries can
    /// never be committed twice. Resets the order counter only once the
    /// partial writes are drained too (`take_masked`).
    pub fn take(&mut self) -> Vec<NbaUpdate> {
        if self.masked.is_empty() {
            self.next_order = 0;
        }
        std::mem::take(&mut self.entries)
    }

    /// Drains every staged partial / memory-row write.
    pub fn take_masked(&mut self) -> Vec<MaskedNbaUpdate> {
        if self.entries.is_empty() {
            self.next_order = 0;
        }
        std::mem::take(&mut self.masked)
    }
}

#[cfg(test)]
mod tests {
    // Deliberately only these two `state` types, never `SignalAccess` — see
    // `no_direct_apply_path_exists_on_nba_log` below, which points at this
    // `use` list as the compile-time half of its proof.
    use crate::state::{FourStateValue, StateCellId};

    use super::{NbaLog, NbaUpdate};

    #[test]
    fn stage_appends_without_applying_to_any_state() {
        let mut log = NbaLog::new();
        assert!(log.is_empty());

        log.stage(StateCellId(1), FourStateValue::Narrow(5), 100);
        log.stage(StateCellId(1), FourStateValue::Narrow(9), 200);

        // SC-034's acceptance, at the log level: nothing above could have
        // changed a read path — `stage`'s own signature has no write-capable
        // parameter, so this assertion is backed by a type-level guarantee,
        // not just today's behavior.
        assert_eq!(log.len(), 2);
        assert!(!log.is_empty());

        let entries = log.take();
        assert_eq!(entries.len(), 2);
        assert_eq!(entries[0].source, 100);
        assert_eq!(entries[1].source, 200);
        assert!(matches!(entries[0].value, FourStateValue::Narrow(5)));
        assert!(matches!(entries[1].value, FourStateValue::Narrow(9)));
    }

    #[test]
    fn order_field_is_populated_in_strict_push_order() {
        let mut log = NbaLog::new();
        log.stage(StateCellId(7), FourStateValue::Narrow(1), 1);
        log.stage(StateCellId(3), FourStateValue::Narrow(2), 2);
        log.stage(StateCellId(7), FourStateValue::Narrow(3), 1);

        let entries = log.take();
        assert_eq!(entries[0].order, 0);
        assert_eq!(entries[1].order, 1);
        assert_eq!(entries[2].order, 2);
    }

    #[test]
    fn clear_empties_the_log_and_is_idempotent() {
        let mut log = NbaLog::new();
        log.stage(StateCellId(1), FourStateValue::Narrow(1), 1);
        log.stage(StateCellId(2), FourStateValue::Narrow(2), 1);
        assert_eq!(log.len(), 2);

        log.clear();
        assert!(log.is_empty());
        assert_eq!(log.len(), 0);

        // Idempotent: clearing an already-empty log must not panic.
        log.clear();
        assert!(log.is_empty());
    }

    #[test]
    fn take_drains_exactly_once() {
        let mut log = NbaLog::new();
        log.stage(StateCellId(1), FourStateValue::Narrow(1), 1);
        log.stage(StateCellId(2), FourStateValue::Narrow(2), 2);

        let first = log.take();
        assert_eq!(first.len(), 2);
        assert!(log.is_empty());

        // A second take with no intervening stage must return nothing —
        // fails if `take` cloned instead of draining (double-commit hazard).
        let second = log.take();
        assert!(second.is_empty());
    }

    /// SC-036, negative/compile-level: this test file's own `use` list (see
    /// the top of this module) only names `StateCellId`/`FourStateValue`,
    /// never `SignalAccess` — a reviewer can point at that as evidence
    /// `NbaLog` cannot call a write API even if it wanted to. The runtime-
    /// checkable half of the same claim: `NbaLog` is exactly its two staging
    /// `Vec`s plus the shared order counter -- no embedded arena pointer or
    /// other write handle.
    #[test]
    fn no_direct_apply_path_exists_on_nba_log() {
        #[allow(dead_code)]
        struct ExpectedShape {
            entries: Vec<NbaUpdate>,
            masked: Vec<super::MaskedNbaUpdate>,
            next_order: u32,
        }
        assert_eq!(
            std::mem::size_of::<NbaLog>(),
            std::mem::size_of::<ExpectedShape>()
        );
    }

    /// SC-035's staging-layer contribution: several processes staging in the
    /// same timestep all land inertly in the log — nothing here has a partial-
    /// apply path, so nothing becomes visible until one `take()` drains the
    /// whole batch at once for the (Task 7) commit pass to apply together.
    /// The full "visible atomically, not incrementally" acceptance requires an
    /// actual commit/apply step and is Task 7's `commit_nba` test — this test
    /// is the achievable proxy at this layer: staging never applies, no
    /// matter how many processes or writes are interleaved first.
    #[test]
    fn several_processes_staging_in_one_timestep_stay_unapplied_until_one_take() {
        let mut log = NbaLog::new();
        log.stage(StateCellId(1), FourStateValue::Narrow(10), 1);
        log.stage(StateCellId(2), FourStateValue::Narrow(20), 2);
        log.stage(StateCellId(1), FourStateValue::Narrow(30), 3);
        log.stage(StateCellId(3), FourStateValue::Narrow(40), 4);

        // All four stay staged together — no incremental drain happened as a
        // side effect of any individual `stage` call.
        assert_eq!(log.len(), 4);

        let batch = log.take();
        assert_eq!(
            batch.len(),
            4,
            "the whole timestep's writes drain as one batch"
        );
        assert!(log.is_empty());
    }
}