vj-sim 0.3.0

//! The scheduler-side state of an `event` that lives in a class (LRM
//! 1800-2023 15.5): an `event` property of an object, an `event` local or
//! formal of a class method.
//!
//! A module's named event is a state cell its waiters are sensitive to
//! (`state::event_pool`), which compiled class code has no access to: a class
//! method has no instance handle. A class event is instead an id into this
//! table -- the 64-bit value the property holds, `0` being `null` -- and its
//! waiters are blocked processes, like the queue of a semaphore
//! ([`super::semaphore`]): `@e` queues the process (`SyncBlock`), `-> e`
//! wakes every queued process (`SyncWake`) and stamps the time step, and
//! `e.triggered` is true when that stamp is the current time step (15.5.3).
//!
//! Ids are never reused: an event object is a few words, and an assignment
//! (`a = b`, which makes both name one object, 15.5.5.1) copies the id with no
//! reference counting. A process killed while queued leaves its queue
//! ([`cancel`]), like a blocked `semaphore.get`.

use std::cell::RefCell;
use std::collections::{HashMap, HashSet};

use crate::sched::{ProcessId, ProcessRequest};

#[derive(Default)]
struct Event {
    /// Blocked `@e` waiters, oldest first, each with the index of the
    /// `@(...)` list term that queued it (a process whose list names this
    /// event in several terms appears once per term).
    waiters: Vec<(ProcessId, u32)>,
    /// The time (ticks) of the latest trigger, if any.
    last_trigger: Option<u64>,
}

#[derive(Default)]
struct Table {
    /// Event `id` is `events[id - 1]`.
    events: Vec<Event>,
    /// The events each blocked process waits on: one for `@e`, several for an
    /// `@(e1 or e2 ...)` list (9.4.2.1), from which the first trigger of any
    /// removes it.
    waiting_on: HashMap<ProcessId, Vec<u64>>,
    /// Processes in a mixed `@(class_event or signal ...)` wait ([`enqueue_dual`]):
    /// they are also waiting on signals, so a trigger wakes them through the
    /// sensitivity table's side too ([`is_dual`]) until they [`leave`].
    dual: HashSet<ProcessId>,
    /// The class-event terms (bit k = the list's kth class term) of a mixed
    /// wait that triggered, kept until the process's next mixed wait
    /// ([`fired_bit`]): the resume judges each fired term's `iff`.
    dual_fired: HashMap<ProcessId, u64>,
}

// One table per OS thread, like every other piece of per-run scheduler state
// (see `super::semaphore`).
thread_local! {
    static TABLE: RefCell<Table> = RefCell::new(Table::default());
}

fn with_table<R>(f: impl FnOnce(&mut Table) -> R) -> R {
    TABLE.with(|t| f(&mut t.borrow_mut()))
}

/// A new event object: its id (never `0`).
pub fn alloc() -> u64 {
    with_table(|t| {
        t.events.push(Event::default());
        t.events.len() as u64
    })
}

thread_local! {
    /// Wake requests raised where no process runs (the NBA commit pass): the
    /// engine applies them right after the pass ([`take_deferred_wakes`]).
    static DEFERRED: RefCell<Vec<ProcessRequest>> = const { RefCell::new(Vec::new()) };
}

/// Queues a request raised by an NBA-region trigger.
pub fn defer_wake(req: ProcessRequest) {
    DEFERRED.with(|d| d.borrow_mut().push(req));
}

/// The requests [`defer_wake`] queued since the last call, in order.
pub fn take_deferred_wakes() -> Vec<ProcessRequest> {
    DEFERRED.with(|d| std::mem::take(&mut *d.borrow_mut()))
}

/// `-> e` at time `now`: stamps the time step and wakes every waiter, in
/// arrival order. A `null` or unknown id triggers nothing (15.5.5.2).
pub fn trigger(id: u64, now: u64, emit: &mut dyn FnMut(ProcessRequest)) {
    let woken = with_table(|t| {
        let ev = usize::try_from(id)
            .ok()
            .and_then(|i| i.checked_sub(1))
            .and_then(|i| t.events.get_mut(i))?;
        ev.last_trigger = Some(now);
        let entries = std::mem::take(&mut ev.waiters);
        // One wake per process; every term of its list that names this event
        // fired (9.4.2.3: each term's `iff` is judged on its own), which the
        // resume reads as the fired-term mask (`sv_rt_fired_mask`).
        let mut woken: Vec<ProcessId> = Vec::new();
        for &(p, term) in &entries {
            if t.dual.contains(&p) {
                if term < 64 {
                    *t.dual_fired.entry(p).or_default() |= 1u64 << term;
                }
            } else {
                crate::sched::kernel_adapter::note_fired_term(p, term as usize);
            }
            if !woken.contains(&p) {
                woken.push(p);
            }
        }
        for p in &woken {
            for other in t.waiting_on.remove(p).unwrap_or_default() {
                if other != id
                    && let Some(oe) = t.events.get_mut((other - 1) as usize)
                {
                    oe.waiters.retain(|w| w.0 != *p);
                }
            }
        }
        Some(woken)
    });
    for p in woken.unwrap_or_default() {
        emit(ProcessRequest::SyncWake(p));
    }
}

/// Queues `proc` on event `id` without blocking it: a term of an
/// `@(e1 or e2 ...)` list (the list's last term is a [`wait`]); `term` is the
/// term's index in the list. A `null` or unknown id queues nothing; a repeated
/// (event, term) pair queues once.
pub fn enqueue(id: u64, proc: ProcessId, term: u32) {
    with_table(|t| {
        let known = usize::try_from(id)
            .ok()
            .and_then(|i| i.checked_sub(1))
            .is_some_and(|i| i < t.events.len());
        if !known {
            return;
        }
        let ws = &mut t.events[(id - 1) as usize].waiters;
        if !ws.contains(&(proc, term)) {
            ws.push((proc, term));
        }
        let on = t.waiting_on.entry(proc).or_default();
        if !on.contains(&id) {
            on.push(id);
        }
    });
}

/// Queues `proc` on event `id` as one term of a mixed `@(class_event or
/// signal ...)` list (LRM 1800-2023 9.4.2.1): like [`enqueue`], and the process
/// is marked as also waiting on signals ([`is_dual`]) until it [`leave`]s.
pub fn enqueue_dual(id: u64, proc: ProcessId, term: u32) {
    with_table(|t| {
        t.dual.insert(proc);
        if term == 0 {
            // The first class term of a new wait: forget the last wait's.
            t.dual_fired.remove(&proc);
        }
    });
    enqueue(id, proc, term);
}

/// Whether class term `term` of the mixed wait `proc` last resumed from
/// triggered (`term` = all ones: whether any class term did). A process woken
/// by a signal term has none.
pub fn fired_bit(proc: ProcessId, term: u64) -> bool {
    with_table(|t| {
        let m = t.dual_fired.get(&proc).copied().unwrap_or(0);
        if term >= 64 {
            m != 0
        } else {
            m >> term & 1 != 0
        }
    })
}

/// Whether `proc` is in a mixed class-event/signal wait ([`enqueue_dual`])
/// that has not been left yet.
pub fn is_dual(proc: ProcessId) -> bool {
    with_table(|t| t.dual.contains(&proc))
}

/// A mixed-wait process resumed (by either side): it leaves every class-event
/// queue it is still on and is no longer dual.
pub fn leave(proc: ProcessId) {
    cancel(proc);
}

/// `@e` for process `proc`: queues it and asks the scheduler to block it.
/// Waiting on `null` blocks forever (15.5.5.2: no event ever wakes it).
pub fn wait(id: u64, proc: ProcessId, term: u32, emit: &mut dyn FnMut(ProcessRequest)) {
    enqueue(id, proc, term);
    emit(ProcessRequest::SyncBlock);
}

/// `e.triggered` (15.5.3): `e` was triggered in the time step `now`.
pub fn triggered(id: u64, now: u64) -> bool {
    with_table(|t| {
        usize::try_from(id)
            .ok()
            .and_then(|i| i.checked_sub(1))
            .and_then(|i| t.events.get(i))
            .is_some_and(|ev| ev.last_trigger == Some(now))
    })
}

/// How many processes are queued on `id`.
pub fn waiters(id: u64) -> usize {
    with_table(|t| {
        usize::try_from(id)
            .ok()
            .and_then(|i| i.checked_sub(1))
            .and_then(|i| t.events.get(i))
            .map_or(0, |ev| {
                let mut ps: Vec<ProcessId> = Vec::new();
                for w in &ev.waiters {
                    if !ps.contains(&w.0) {
                        ps.push(w.0);
                    }
                }
                ps.len()
            })
    })
}

/// A blocked process ended (killed) before it was woken: it leaves its queue.
/// A no-op for a process not blocked on a class event.
pub fn cancel(proc: ProcessId) {
    with_table(|t| {
        t.dual.remove(&proc);
        for id in t.waiting_on.remove(&proc).unwrap_or_default() {
            if let Some(ev) = t.events.get_mut((id - 1) as usize) {
                ev.waiters.retain(|w| w.0 != proc);
            }
        }
    });
}

/// Forgets every event and waiter (a new run starts with none; process ids
/// are reused from zero).
pub fn reset_for_new_run() {
    DEFERRED.with(|d| d.borrow_mut().clear());
    with_table(|t| *t = Table::default());
}

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

    #[derive(Default)]
    struct Log(Vec<ProcessRequest>);

    #[test]
    fn trigger_wakes_every_waiter_in_order_and_stamps_the_step() {
        reset_for_new_run();
        let e = alloc();
        let mut log = Log::default();
        for p in [3, 1, 2] {
            wait(e, p, 0, &mut |r| log.0.push(r));
        }
        assert_eq!(waiters(e), 3);
        log.0.clear();
        assert!(!triggered(e, 5));
        trigger(e, 5, &mut |r| log.0.push(r));
        assert_eq!(
            log.0,
            [
                ProcessRequest::SyncWake(3),
                ProcessRequest::SyncWake(1),
                ProcessRequest::SyncWake(2)
            ]
        );
        assert_eq!(waiters(e), 0);
        assert!(triggered(e, 5));
        assert!(!triggered(e, 6), "triggered lasts one time step");
    }

    #[test]
    fn an_or_list_wakes_once_on_the_first_trigger_and_leaves_the_other_queues() {
        reset_for_new_run();
        let (a, b) = (alloc(), alloc());
        let mut log = Log::default();
        enqueue(a, 4, 0);
        enqueue(a, 4, 0);
        wait(b, 4, 1, &mut |r| log.0.push(r));
        assert_eq!((waiters(a), waiters(b)), (1, 1));
        log.0.clear();
        trigger(b, 1, &mut |r| log.0.push(r));
        assert_eq!(log.0, [ProcessRequest::SyncWake(4)]);
        assert_eq!((waiters(a), waiters(b)), (0, 0));
        trigger(a, 1, &mut |r| log.0.push(r));
        assert_eq!(log.0.len(), 1, "the process was already woken once");
    }

    #[test]
    fn a_dual_waiter_is_marked_until_it_leaves() {
        reset_for_new_run();
        let (a, b) = (alloc(), alloc());
        enqueue_dual(a, 6, 0);
        enqueue_dual(b, 6, 1);
        assert!(is_dual(6) && !is_dual(7));
        assert_eq!((waiters(a), waiters(b)), (1, 1));
        let mut log = Vec::new();
        trigger(a, 1, &mut |r| log.push(r));
        assert_eq!(log, [ProcessRequest::SyncWake(6)]);
        assert!(is_dual(6), "still marked until the process runs again");
        assert!(fired_bit(6, 0) && !fired_bit(6, 1) && fired_bit(6, u64::MAX));
        assert_eq!(waiters(b), 0, "the first trigger drops the other queues");
        leave(6);
        assert!(!is_dual(6));
        enqueue_dual(a, 6, 0);
        assert!(
            !fired_bit(6, u64::MAX),
            "a new wait forgets the last one's terms"
        );
        leave(6);
        assert_eq!(waiters(a), 0, "a signal-woken process leaves its queues");
    }

    #[test]
    fn null_and_unknown_ids_trigger_nothing_and_never_wake() {
        reset_for_new_run();
        let mut log = Log::default();
        trigger(0, 1, &mut |r| log.0.push(r));
        trigger(99, 1, &mut |r| log.0.push(r));
        assert!(log.0.is_empty());
        assert!(!triggered(0, 1));
        wait(0, 7, 0, &mut |r| log.0.push(r));
        assert_eq!(log.0, [ProcessRequest::SyncBlock]);
        trigger(0, 2, &mut |r| log.0.push(r));
        assert_eq!(log.0.len(), 1, "a null wait is never woken");
    }

    #[test]
    fn a_killed_waiter_leaves_the_queue() {
        reset_for_new_run();
        let e = alloc();
        let mut log = Log::default();
        wait(e, 4, 0, &mut |r| log.0.push(r));
        wait(e, 5, 0, &mut |r| log.0.push(r));
        log.0.clear();
        cancel(4);
        trigger(e, 1, &mut |r| log.0.push(r));
        assert_eq!(log.0, [ProcessRequest::SyncWake(5)]);
    }
}