vj_sim/sched/

repeat.rs

//! SCHED::REPEAT — `EventTerm::Repeat` engine semantics: tracks how many
//! times a `repeat(N)@(...)`-shaped dynamic wait's *inner* term must fire
//! before the waiting process actually resumes (scheduler.md §4; ARCHITECTURE
//! §6.5; ledger SC-004, SC-005, SC-052, SC-053).
//!
//! `SensitivityTable::install_term`'s `Repeat` arm (wave A) installs the
//! inner term at the *outer* term's index, so every firing of the inner edge
//! reaches `EngineNotifySink::notify_change` exactly as an ordinary dynamic
//! match — indistinguishable, at that layer, from a plain `Edge` term firing
//! once. [`RepeatCounters`] is the missing piece: keyed by `(ProcessId,
//! term_index)` (never just `ProcessId` —
//! `two_different_term_indices_for_the_same_process_count_independently`
//! below proves a process can have more than one live `Repeat` term at once),
//! it decrements a live per-firing count and reports whether the process
//! should actually resume yet.
//!
//! **SC-053's reduction (documented here, the one place `EventTerm::Repeat`'s
//! own doc comment in `sched/mod.rs` points back to):** `##N` against a
//! clocking block's bound clock is semantically `repeat(N)@(posedge
//! <bound clock's cell>)` — this module counts live *firings* of the inner
//! term, never ticks or a precomputed absolute time, so "the same compiled
//! `##N` produces different absolute tick timing when its clock's period
//! changes dynamically at runtime" falls out for free: there is no cached
//! tick offset for a period change to invalidate, because none is ever
//! computed. See `sched::control`'s
//! `sc053_repeat_based_wait_re_reads_the_clocks_live_period_not_a_cached_tick_offset`
//! for the scheduler-level (`Engine`-driven) proof of this against a
//! deliberately irregular (non-periodic) clock.

use rustc_hash::FxHashMap as HashMap;
use std::sync::Mutex;

use crate::sched::ProcessId;

/// [`RepeatCounters::on_term_fired`]'s report.
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub enum RepeatOutcome {
    /// The inner term fired, but fewer than `declared_total` times so far —
    /// the caller must NOT resume the process: leave its sensitivity entry
    /// installed and swallow this firing (no enqueue).
    StillWaiting,
    /// This firing was the `declared_total`th — the caller may resume the
    /// process. The counter for this `(proc, term_index)` is removed.
    Satisfied,
}

/// Live decrementing counters for every `(ProcessId, term_index)` pair
/// currently mid-`repeat(N)@(...)` wait. `Mutex`-guarded so `Engine` (which
/// reaches this only through a shared `&self` inside `EngineNotifySink`) can
/// call [`Self::on_term_fired`]/[`Self::clear`] from a `&self`-only context —
/// the same pattern every other `sched` shared-mutable-state field already
/// uses (`NbaLog`, `RegionQueues`, ...).
pub struct RepeatCounters {
    counters: Mutex<HashMap<(ProcessId, usize), u64>>,
}

impl RepeatCounters {
    pub fn new() -> Self {
        Self {
            counters: Mutex::new(HashMap::default()),
        }
    }

    /// Looks up `(proc, term_index)`'s live remaining count, initializing it
    /// to `declared_total` on first sight (the first firing after the
    /// process suspended), decrements by 1, and reports the outcome. At `0`
    /// post-decrement the entry is removed — so a *later*, unrelated
    /// `repeat(N)@(...)` wait re-installed at the same `term_index` starts
    /// fresh rather than inheriting a stale zero (see
    /// `clear_removes_only_the_named_processs_entries` for the companion
    /// "no stale residue after an explicit clear" proof).
    pub fn on_term_fired(
        &self,
        proc: ProcessId,
        term_index: usize,
        declared_total: u64,
    ) -> RepeatOutcome {
        let mut counters = self.counters.lock().expect("RepeatCounters mutex poisoned");
        let remaining = counters.entry((proc, term_index)).or_insert(declared_total);
        *remaining = remaining.saturating_sub(1);
        if *remaining == 0 {
            counters.remove(&(proc, term_index));
            RepeatOutcome::Satisfied
        } else {
            RepeatOutcome::StillWaiting
        }
    }

    /// Removes every counter entry belonging to `proc`, regardless of
    /// `term_index` — called from the same place `SensitivityTable::
    /// remove_dynamic` already is (`sched::control`'s `dispatch_ready_set`),
    /// so a process that resumes via any path (an ordinary event match,
    /// `Disabled`, ...) never leaves a stale partial count behind for a
    /// future, unrelated dynamic wait reusing the same `term_index`.
    pub fn clear(&self, proc: ProcessId) {
        let mut counters = self.counters.lock().expect("RepeatCounters mutex poisoned");
        counters.retain(|&(p, _), _| p != proc);
    }
}

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

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

    #[test]
    fn first_and_second_firing_of_a_three_count_repeat_report_still_waiting_third_reports_satisfied()
     {
        let counters = RepeatCounters::new();
        let proc: ProcessId = 1;
        assert_eq!(
            counters.on_term_fired(proc, 0, 3),
            RepeatOutcome::StillWaiting
        );
        assert_eq!(
            counters.on_term_fired(proc, 0, 3),
            RepeatOutcome::StillWaiting
        );
        assert_eq!(counters.on_term_fired(proc, 0, 3), RepeatOutcome::Satisfied);
    }

    #[test]
    fn two_different_term_indices_for_the_same_process_count_independently() {
        let counters = RepeatCounters::new();
        let proc: ProcessId = 7;
        // term 0 needs 2, term 1 needs 3, interleaved.
        assert_eq!(
            counters.on_term_fired(proc, 0, 2),
            RepeatOutcome::StillWaiting
        );
        assert_eq!(
            counters.on_term_fired(proc, 1, 3),
            RepeatOutcome::StillWaiting
        );
        assert_eq!(
            counters.on_term_fired(proc, 1, 3),
            RepeatOutcome::StillWaiting
        );
        assert_eq!(
            counters.on_term_fired(proc, 0, 2),
            RepeatOutcome::Satisfied,
            "term 0 must reach Satisfied at its own declared count (2), unaffected by \
             term 1's firings"
        );
        assert_eq!(
            counters.on_term_fired(proc, 1, 3),
            RepeatOutcome::Satisfied,
            "term 1 must reach Satisfied at its own declared count (3)"
        );
    }

    #[test]
    fn clear_removes_only_the_named_processs_entries() {
        let counters = RepeatCounters::new();
        let proc_a: ProcessId = 1;
        let proc_b: ProcessId = 2;
        assert_eq!(
            counters.on_term_fired(proc_a, 0, 3),
            RepeatOutcome::StillWaiting
        );
        assert_eq!(
            counters.on_term_fired(proc_b, 0, 5),
            RepeatOutcome::StillWaiting
        );

        counters.clear(proc_a);

        // proc_a restarts at declared_total - 1 (not continuing a stale
        // partial count of "1 remaining out of 3").
        assert_eq!(
            counters.on_term_fired(proc_a, 0, 3),
            RepeatOutcome::StillWaiting
        );
        assert_eq!(
            counters.on_term_fired(proc_a, 0, 3),
            RepeatOutcome::StillWaiting
        );
        assert_eq!(
            counters.on_term_fired(proc_a, 0, 3),
            RepeatOutcome::Satisfied
        );

        // proc_b's own entry (1 fired of 5) must be untouched by proc_a's
        // clear.
        for _ in 0..3 {
            assert_eq!(
                counters.on_term_fired(proc_b, 0, 5),
                RepeatOutcome::StillWaiting,
                "proc_b's own in-progress count must survive proc_a's clear untouched"
            );
        }
        assert_eq!(
            counters.on_term_fired(proc_b, 0, 5),
            RepeatOutcome::Satisfied
        );
    }
}