vj-sim 0.3.0

//! Classes plan (`docs/superpowers/plans/2026-10-03-classes.md`) Task 17: the
//! scheduler-side state of the built-in `semaphore` class (LRM 1800-2023
//! 15.3).
//!
//! A semaphore is an ordinary heap object ([`crate::objects`]) whose one field
//! word holds the key count as an `i64` (`new(keyCount)` takes an `int`, which
//! may be negative: 15.3.1 -- the keys must climb above zero before any `get`
//! succeeds). The processes blocked in `get(n)` live here, in one FIFO queue per
//! semaphore ("The semaphore waiting queue is first-in first-out (FIFO)",
//! 15.3.3), so a blocked process can be found again by `put` (to hand it keys)
//! and by the engine (to drop it from the queue when it is killed).
//!
//! # Grant policy
//!
//! The queue is strictly FIFO: `get(n)` takes keys at once only when no process
//! is already queued; `put` serves the head of the queue first and stops at the
//! first waiter that needs more keys than the bucket holds, so a large request
//! is never overtaken by later small ones. `try_get` is the one deliberate
//! exception: it never blocks, so it takes keys whenever the bucket has enough
//! (it cannot starve a queued process of a key it would otherwise have been
//! granted, but it can slip ahead of one that needs more than is available).
//!
//! Keys are handed over at the moment a waiter is chosen: the count drops when
//! `put` wakes the process, not when the process runs. A waiter killed between
//! being granted and running therefore loses its keys (it was granted them; the
//! LRM does not say what a killed holder does with its keys either).
//!
//! # Blocking
//!
//! This module only decides who waits and who wakes; the scheduler does the
//! blocking and waking. Both go through the `emit` callback each function takes
//! (`SyncBlock` for the caller of a `get` that must wait, `SyncWake` per process a
//! `put` released), so the requests reach the engine's queue in the order the
//! decisions were made -- always the order `SyncBlock(p)` then `SyncWake(p)`,
//! because the callback runs inside the table update.
//!
//! A queued waiter holds a reference to its semaphore, so the object cannot be
//! freed (and its slot reused) while a process is blocked on it.

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

use crate::objects::{self, FieldSlot};
use crate::sched::{ProcessId, ProcessRequest};

/// The one field word of a semaphore's object: the key count.
const COUNT_SLOT: FieldSlot = FieldSlot {
    offset: 0,
    words: 1,
};

/// How many field words a semaphore object is allocated with.
pub const SEMAPHORE_FIELD_WORDS: usize = 1;

#[derive(Default)]
struct Table {
    /// Waiters per semaphore handle, oldest first, with the keys each wants.
    queues: HashMap<u64, VecDeque<(ProcessId, i64)>>,
    /// The semaphore each blocked process waits on (a process blocks in at most
    /// one `get`).
    waiting_on: HashMap<ProcessId, u64>,
}

// One table per OS thread, like every other piece of per-run scheduler state
// (`kernel_adapter`'s cells, the current-process bridge): a run is one thread,
// process ids restart from zero in each run, and the test binaries run several
// designs concurrently on different threads. (The object heap, whose handles
// are globally unique, is shared.)
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()))
}

fn count_of(h: u64) -> Option<i64> {
    let mut w = [0u64];
    objects::field_get(h, COUNT_SLOT, &mut w).then_some(w[0] as i64)
}

fn set_count(h: u64, count: i64) {
    objects::field_set(h, COUNT_SLOT, &[count as u64]);
}

/// `new(keyCount)`: stores the initial count. `false` when `h` is not a live
/// semaphore object.
pub fn init(h: u64, count: i64) -> bool {
    objects::field_set(h, COUNT_SLOT, &[count as u64])
}

/// Hands keys to the queue head(s) while the bucket can serve them; returns the
/// woken processes (each held a reference on `h` that the caller must release).
fn grant(t: &mut Table, h: u64, emit: &mut dyn FnMut(ProcessRequest)) -> usize {
    let Some(mut count) = count_of(h) else {
        return 0;
    };
    let mut woken = 0;
    if let Some(queue) = t.queues.get_mut(&h) {
        while let Some(&(proc, need)) = queue.front() {
            if count < need {
                break;
            }
            count -= need;
            queue.pop_front();
            t.waiting_on.remove(&proc);
            emit(ProcessRequest::SyncWake(proc));
            woken += 1;
        }
        if queue.is_empty() {
            t.queues.remove(&h);
        }
    }
    set_count(h, count);
    woken
}

/// `put(n)` (15.3.3): adds `n` keys, then serves the queue. Returns `false`
/// when `h` is not a live semaphore. A negative `n` "shall result in an error"
/// (15.3.2): the compiled caller has reported it, and the call changes nothing.
pub fn put(h: u64, n: i64, emit: &mut dyn FnMut(ProcessRequest)) -> bool {
    if n < 0 {
        return count_of(h).is_some();
    }
    let woken = with_table(|t| {
        let count = count_of(h)?;
        set_count(h, count.saturating_add(n));
        Some(grant(t, h, emit))
    });
    let Some(woken) = woken else {
        return false;
    };
    for _ in 0..woken {
        objects::release(h);
    }
    true
}

/// `get(n)` (15.3.4) for process `proc`. Returns `true` when the process must
/// block (it is queued and `emit` was given `SyncBlock`), `false` when it took
/// its keys at once. A handle that is not a live semaphore blocks nothing; nor
/// does a negative `n`, which "shall result in an error" (15.3.3): the compiled
/// caller has reported it, and the call takes nothing and queues nothing.
pub fn get(h: u64, proc: ProcessId, n: i64, emit: &mut dyn FnMut(ProcessRequest)) -> bool {
    with_table(|t| {
        let Some(count) = count_of(h) else {
            return false;
        };
        if n < 0 {
            return false;
        }
        let queued = t.queues.get(&h).is_some_and(|q| !q.is_empty());
        if !queued && count >= n {
            set_count(h, count - n);
            return false;
        }
        assert!(
            !t.waiting_on.contains_key(&proc),
            "semaphore: process {proc} called get() while already blocked in one"
        );
        objects::retain(h);
        t.queues.entry(h).or_default().push_back((proc, n));
        t.waiting_on.insert(proc, h);
        emit(ProcessRequest::SyncBlock);
        true
    })
}

/// `try_get(n)` (15.3.4): takes the keys iff the bucket has them. Never blocks.
/// A negative `n` "returns 0 and shall result in an error": the compiled caller
/// has reported it; the count is left unchanged.
pub fn try_get(h: u64, n: i64) -> bool {
    let Some(count) = count_of(h) else {
        return false;
    };
    if n < 0 || count < n {
        return false;
    }
    set_count(h, count - n);
    true
}

/// The key count, for tests and diagnostics; `None` for a non-semaphore.
pub fn keys(h: u64) -> Option<i64> {
    count_of(h)
}

/// How many processes are queued on `h`.
pub fn waiters(h: u64) -> usize {
    with_table(|t| t.queues.get(&h).map_or(0, VecDeque::len))
}

/// A blocked process ended (killed) before it was woken: drops it from its
/// queue, then serves the queue again (it may have been the large request
/// that was holding the others back). A no-op for a process not blocked on a
/// semaphore.
pub fn cancel(proc: ProcessId, emit: &mut dyn FnMut(ProcessRequest)) {
    let cancelled = with_table(|t| {
        let h = t.waiting_on.remove(&proc)?;
        if let Some(queue) = t.queues.get_mut(&h) {
            queue.retain(|&(p, _)| p != proc);
            if queue.is_empty() {
                t.queues.remove(&h);
            }
        }
        Some((h, grant(t, h, emit)))
    });
    let Some((h, woken)) = cancelled else {
        return;
    };
    // The cancelled waiter's own reference, then one per process woken.
    for _ in 0..=woken {
        objects::release(h);
    }
}

/// Forgets every waiter (a new run starts with no process blocked; ids are
/// reused from zero) and drops the references they held.
pub fn reset_for_new_run() {
    let queues = with_table(|t| {
        t.waiting_on.clear();
        std::mem::take(&mut t.queues)
    });
    for (h, queue) in queues {
        for _ in queue {
            objects::release(h);
        }
    }
}

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

    // The object heap and this table are process-wide and every test shares
    // them: each test below allocates its own semaphore and a disjoint range of
    // process ids, and asserts only on those.

    fn sem(count: i64, template: u32) -> u64 {
        let h = objects::alloc(ClassTemplateId(template), SEMAPHORE_FIELD_WORDS);
        assert!(init(h, count));
        h
    }

    #[derive(Default)]
    struct Log(Vec<ProcessRequest>);
    impl Log {
        fn sink(&mut self) -> impl FnMut(ProcessRequest) + '_ {
            |r| self.0.push(r)
        }
    }

    #[test]
    fn get_takes_keys_without_blocking_when_enough_are_free() {
        let h = sem(3, 17_000);
        let mut log = Log::default();
        assert!(!get(h, 1_000, 2, &mut log.sink()));
        assert_eq!(keys(h), Some(1));
        assert!(log.0.is_empty(), "an immediate get queues no request");
        assert!(get(h, 1_000, 2, &mut log.sink()), "one key is not two");
        assert_eq!(log.0, [ProcessRequest::SyncBlock]);
        assert_eq!(keys(h), Some(1), "a blocked get takes nothing yet");
    }

    #[test]
    fn put_serves_waiters_in_arrival_order_and_hands_over_the_keys() {
        let h = sem(0, 17_001);
        let mut log = Log::default();
        for p in [1_010, 1_011, 1_012] {
            assert!(get(h, p, 1, &mut log.sink()));
        }
        log.0.clear();
        assert!(put(h, 2, &mut log.sink()));
        assert_eq!(
            log.0,
            [
                ProcessRequest::SyncWake(1_010),
                ProcessRequest::SyncWake(1_011)
            ]
        );
        assert_eq!(
            keys(h),
            Some(0),
            "the woken processes already hold the keys"
        );
        assert_eq!(waiters(h), 1);
    }

    #[test]
    fn a_large_request_is_not_overtaken_by_a_later_small_one() {
        let h = sem(0, 17_002);
        let mut log = Log::default();
        assert!(get(h, 1_020, 3, &mut log.sink()));
        assert!(
            get(h, 1_021, 1, &mut log.sink()),
            "queues behind the get(3)"
        );
        log.0.clear();
        put(h, 2, &mut log.sink());
        assert!(
            log.0.is_empty(),
            "two keys serve neither the head (3) nor, by FIFO, the next"
        );
        assert_eq!(keys(h), Some(2));
        put(h, 2, &mut log.sink());
        assert_eq!(
            log.0,
            [
                ProcessRequest::SyncWake(1_020),
                ProcessRequest::SyncWake(1_021)
            ]
        );
        assert_eq!(keys(h), Some(0));
    }

    #[test]
    fn a_get_arriving_behind_a_queue_waits_even_when_keys_are_free() {
        let h = sem(1, 17_003);
        let mut log = Log::default();
        assert!(get(h, 1_030, 2, &mut log.sink()));
        assert!(
            get(h, 1_031, 1, &mut log.sink()),
            "FIFO: the one free key belongs to the queue's order, not to the newcomer"
        );
        assert_eq!(waiters(h), 2);
    }

    #[test]
    fn try_get_never_blocks_and_takes_keys_only_when_available() {
        let h = sem(2, 17_004);
        assert!(try_get(h, 1));
        assert!(!try_get(h, 2));
        assert_eq!(keys(h), Some(1));
        assert!(try_get(h, 1));
        assert!(!try_get(h, 1));
        assert!(try_get(h, 0), "taking zero keys succeeds");
        assert_eq!(waiters(h), 0);
    }

    #[test]
    fn a_negative_count_must_climb_above_zero_first() {
        let h = sem(-2, 17_005);
        let mut log = Log::default();
        assert!(!try_get(h, 1));
        put(h, 2, &mut log.sink());
        assert!(!try_get(h, 1), "two puts from -2 reach zero, not one");
        put(h, 1, &mut log.sink());
        assert!(try_get(h, 1));
    }

    #[test]
    fn cancelling_a_waiter_removes_it_and_lets_the_next_one_through() {
        let h = sem(1, 17_006);
        let mut log = Log::default();
        assert!(get(h, 1_040, 2, &mut log.sink()), "blocks the queue head");
        assert!(get(h, 1_041, 1, &mut log.sink()));
        log.0.clear();
        cancel(1_040, &mut log.sink());
        assert_eq!(log.0, [ProcessRequest::SyncWake(1_041)]);
        assert_eq!(keys(h), Some(0));
        assert_eq!(waiters(h), 0);
        // Cancelling a process that is not blocked is a no-op.
        cancel(1_040, &mut log.sink());
        assert_eq!(log.0.len(), 1);
    }

    #[test]
    fn a_waiter_keeps_its_semaphore_alive_until_it_leaves_the_queue() {
        let h = sem(0, 17_007);
        let mut log = Log::default();
        assert!(get(h, 1_050, 1, &mut log.sink()));
        assert_eq!(
            objects::refcount(h),
            Some(2),
            "the allocation's and the waiter's"
        );
        put(h, 1, &mut log.sink());
        assert_eq!(objects::refcount(h), Some(1), "the woken waiter let go");

        assert!(get(h, 1_051, 1, &mut log.sink()));
        cancel(1_051, &mut log.sink());
        assert_eq!(
            objects::refcount(h),
            Some(1),
            "a cancelled waiter let go too"
        );
    }

    /// A negative key count is an error (15.3.2-15.3.4) and changes nothing:
    /// `try_get(-1)` returns 0 (it used to return 1 and ADD a key), `put(-n)`
    /// does not take keys away, and `get(-n)` neither blocks nor gives keys.
    #[test]
    fn a_negative_key_count_changes_nothing() {
        let h = sem(2, 17_008);
        let mut log = Log::default();
        assert!(!try_get(h, -1));
        assert_eq!(keys(h), Some(2));
        assert!(put(h, -5, &mut log.sink()));
        assert_eq!(keys(h), Some(2));
        assert!(!get(h, 1_070, -3, &mut log.sink()));
        assert_eq!(keys(h), Some(2));
        assert!(log.0.is_empty());
        assert_eq!(waiters(h), 0);
    }

    #[test]
    fn a_non_semaphore_handle_blocks_and_grants_nothing() {
        let mut log = Log::default();
        assert!(!get(0, 1_060, 1, &mut log.sink()));
        assert!(!put(0, 1, &mut log.sink()));
        assert!(!try_get(0, 1));
        assert!(log.0.is_empty());
    }
}