vj_sim/sched/

mailbox.rs

//! Classes plan (`docs/superpowers/plans/2026-10-03-classes.md`) Task 18: the
//! scheduler-side state of the built-in `mailbox` / `mailbox #(T)` class (LRM
//! 1800-2023 15.4).
//!
//! A mailbox is an ordinary heap object ([`crate::objects`]); its message queue
//! and the processes blocked on it live here, in one [`Mailbox`] per object
//! handle, because a message is a variable-sized run of words (the object's
//! fixed-size field array cannot hold it) and a blocked process must be found
//! again by the operation that releases it and by the engine when it is killed.
//!
//! # Messages
//!
//! A [`Message`] is the stored words of one value, a type fingerprint and a flag
//! saying the value is a class handle (word 0), in which case the message owns
//! one reference to the object (the object heap's usual rule for a holder: see
//! `vj-codegen`'s `collections`). Whoever takes a message out of the mailbox
//! takes that reference over; a `peek` takes a new one (`objects::retain`).
//! The fingerprint is how an untyped mailbox does the run-time type check of
//! 15.4.5-15.4.8: a `get`/`peek` whose variable is not of an equivalent type to
//! the message at the head of the queue is an error and takes nothing; the
//! `try_` forms report it as a negative result.
//!
//! # Blocking
//!
//! `get`/`peek` on an empty mailbox and `put` on a full one block the caller.
//! This module only decides who waits and who wakes; the scheduler blocks and
//! wakes through the `emit` callback each function takes (`SyncBlock` for the
//! caller that must wait, `SyncWake` per process released), the same protocol as
//! [`super::semaphore`]. Waiters queue in arrival order ("The mailbox waiting
//! queue is FIFO", 15.4.5).
//!
//! A blocked `get`/`peek` cannot hand its message to the caller directly (the
//! call has already returned to compiled code, which must store the message
//! into the destination variable so that the store is an observable value
//! change). So [`get`] is a two-phase call, issued twice by the compiled
//! statement (the CFG walker follows the statement with a suspend point and
//! emits it again): the first call either completes at once or queues the
//! process; the message handed to a woken waiter waits in a per-process slot
//! (`pending`) until the second call, after the process resumed, collects it.
//! A call that completed at once leaves a marker so the second call is a no-op.
//! A woken waiter's message is chosen when it is woken (a `get` takes it out of
//! the queue then), so a waiter killed after being woken loses its message --
//! the same reading as the keys of a semaphore waiter killed after its grant.
//!
//! A waiter retains its mailbox object so it cannot be freed (and its slot
//! reused) while a process is blocked on it. Messages still queued when the
//! mailbox object itself is freed are not released (a documented leak when
//! they are class handles: the object heap has no destructor hook for the
//! side table).

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

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

/// One queued message.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Message {
    /// The value's stored words.
    pub words: Box<[u64]>,
    /// The type fingerprint (equal exactly for equivalent types, 6.22.1).
    pub fingerprint: u64,
    /// The value is a class handle (word 0) and the message owns one reference
    /// to the object.
    pub obj: bool,
    /// The value is a queue, dynamic array or associative array (word 0 is its
    /// collection handle) and the message owns that collection: a `peek` hands
    /// out a copy-on-write clone, a delivering `get` hands the handle itself
    /// over, a dropped message frees it.
    pub coll: bool,
}

impl Message {
    /// A copy of a message that stays in the queue (`peek`): the copy owns its
    /// own object reference.
    fn peeked(&self) -> Message {
        if self.obj {
            objects::retain(self.words[0]);
        }
        let mut m = self.clone();
        if self.coll {
            m.words[0] = crate::collections::clone(self.words[0]);
        }
        m
    }

    /// Drops the reference an undelivered message owns.
    pub fn discard(self) {
        if self.obj {
            objects::release(self.words[0]);
        }
        if self.coll {
            crate::collections::release(self.words[0]);
        }
    }
}

/// What a woken `get`/`peek` finds when it collects its result.
#[derive(Debug, PartialEq, Eq)]
enum Delivery {
    Message(Message),
    /// The message at the head of the queue was not of the waiter's type.
    Mismatch,
}

#[derive(Debug)]
enum Pending {
    /// The first phase completed at once: the second one has nothing to do.
    Completed,
    /// The process is blocked in a `get`/`peek` on this mailbox, or in a `put`.
    Waiting(u64),
    /// The process was woken; its result waits for the second phase.
    Delivered(Delivery),
}

struct Getter {
    proc: ProcessId,
    peek: bool,
    fingerprint: u64,
}

#[derive(Default)]
struct Mailbox {
    /// `0`: unbounded.
    bound: usize,
    messages: VecDeque<Message>,
    getters: VecDeque<Getter>,
    putters: VecDeque<(ProcessId, Message)>,
}

#[derive(Default)]
struct Table {
    mailboxes: HashMap<u64, Mailbox>,
    pending: HashMap<ProcessId, Pending>,
    /// Words a blocking `get`/`peek` statement evaluated in its first phase
    /// (a side-effecting receiver index or destination selector) for the
    /// second phase to read back, keyed by process and slot.
    stash: HashMap<(ProcessId, u32), u64>,
}

// One table per OS thread, like the semaphore table and every other piece of
// per-run scheduler state.
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()))
}

/// Object references to drop once the table is no longer borrowed.
#[derive(Default)]
struct Releases {
    handles: Vec<u64>,
    /// Collection handles owned by dropped messages.
    colls: Vec<u64>,
}

impl Releases {
    /// Drops what the undelivered message `m` owns.
    fn owned(&mut self, m: &Message) {
        if m.obj {
            self.handles.push(m.words[0]);
        }
        if m.coll {
            self.colls.push(m.words[0]);
        }
    }

    /// Every mailbox operation ends here with the table unborrowed, so this is
    /// also where a queue change is reported: `wait (mb.num() > 0)` sleeps on
    /// object state, which no signal change reports.
    fn run(self) {
        for h in self.handles {
            objects::release(h);
        }
        for h in self.colls {
            crate::collections::release(h);
        }
        super::notify_object_state_change();
    }
}

/// `new(bound)` (15.4.1). A negative bound is illegal (indeterminate, 15.4.1);
/// it is read as `0`, unbounded. `false` when `h` is not a live object.
pub fn init(h: u64, bound: i64) -> bool {
    if !objects::is_live(h) {
        return false;
    }
    let bound = usize::try_from(bound).unwrap_or(0);
    with_table(|t| {
        t.mailboxes.insert(
            h,
            Mailbox {
                bound,
                ..Mailbox::default()
            },
        );
    });
    true
}

/// `num()` (15.4.2).
pub fn num(h: u64) -> usize {
    with_table(|t| t.mailboxes.get(&h).map_or(0, |m| m.messages.len()))
}

/// How many processes are blocked on `h`.
pub fn waiters(h: u64) -> usize {
    with_table(|t| {
        t.mailboxes
            .get(&h)
            .map_or(0, |m| m.getters.len() + m.putters.len())
    })
}

fn is_full(m: &Mailbox) -> bool {
    m.bound != 0 && m.messages.len() >= m.bound
}

/// Wakes `proc` (its waiter reference on the mailbox goes back to `rel`). A
/// woken `get`/`peek` has a `delivery` to collect; a woken `put` is finished
/// (its message is queued) and leaves no marker.
fn wake(
    t: &mut Table,
    proc: ProcessId,
    h: u64,
    delivery: Option<Delivery>,
    emit: &mut dyn FnMut(ProcessRequest),
    rel: &mut Releases,
) {
    match delivery {
        Some(d) => {
            t.pending.insert(proc, Pending::Delivered(d));
        }
        None => {
            t.pending.remove(&proc);
        }
    }
    emit(ProcessRequest::SyncWake(proc));
    rel.handles.push(h);
}

/// Moves blocked putters into the queue while there is room.
fn admit_putters(t: &mut Table, h: u64, emit: &mut dyn FnMut(ProcessRequest), rel: &mut Releases) {
    loop {
        let Some(m) = t.mailboxes.get_mut(&h) else {
            return;
        };
        if is_full(m) {
            return;
        }
        let Some((proc, msg)) = m.putters.pop_front() else {
            return;
        };
        m.messages.push_back(msg);
        wake(t, proc, h, None, emit, rel);
    }
}

/// Serves the blocked `get`/`peek`s while a message is queued, oldest first: a
/// `peek` copies the head, a `get` removes it (which may admit a blocked
/// putter), and the next waiter then sees the new head. "As long as a message
/// remains in the mailbox queue, any process blocked in either a peek() or
/// get() operation shall become unblocked" (15.4.7).
fn serve_getters(t: &mut Table, h: u64, emit: &mut dyn FnMut(ProcessRequest), rel: &mut Releases) {
    loop {
        let Some(m) = t.mailboxes.get_mut(&h) else {
            return;
        };
        if m.messages.is_empty() {
            return;
        }
        let Some(g) = m.getters.pop_front() else {
            return;
        };
        let head_fp = m.messages[0].fingerprint;
        let delivery = if head_fp != g.fingerprint {
            Delivery::Mismatch
        } else if g.peek {
            Delivery::Message(m.messages[0].peeked())
        } else {
            let msg = m.messages.pop_front().expect("checked non-empty");
            Delivery::Message(msg)
        };
        let removed = matches!(&delivery, Delivery::Message(_)) && !g.peek;
        wake(t, g.proc, h, Some(delivery), emit, rel);
        if removed {
            admit_putters(t, h, emit, rel);
        }
    }
}

/// The result of [`put`].
#[derive(Debug, PartialEq, Eq)]
pub enum PutResult {
    /// The message is in the queue.
    Stored,
    /// The mailbox is full: the caller was queued (`emit` got `SyncBlock`) and
    /// the message is held for it until there is room.
    Blocked,
}

/// `put(message)` (15.4.3) for process `proc`. `Err(message)` when `h` is not a
/// live mailbox (the message is handed back so the caller can drop it).
pub fn put(
    h: u64,
    proc: ProcessId,
    msg: Message,
    emit: &mut dyn FnMut(ProcessRequest),
) -> Result<PutResult, Message> {
    let mut rel = Releases::default();
    let result = with_table(|t| {
        let Some(m) = t.mailboxes.get_mut(&h) else {
            return Err(msg);
        };
        if !is_full(m) {
            m.messages.push_back(msg);
            serve_getters(t, h, emit, &mut rel);
            return Ok(PutResult::Stored);
        }
        assert!(
            !t.pending.contains_key(&proc),
            "mailbox: process {proc} called put() while already blocked in a mailbox call"
        );
        objects::retain(h);
        t.mailboxes
            .get_mut(&h)
            .expect("looked up above")
            .putters
            .push_back((proc, msg));
        t.pending.insert(proc, Pending::Waiting(h));
        emit(ProcessRequest::SyncBlock);
        Ok(PutResult::Blocked)
    });
    rel.run();
    result
}

/// `try_put(message)` (15.4.4): `Ok(())` when stored, `Err(message)` when the
/// mailbox is full (or `h` is no live mailbox) -- the caller still owns the
/// message and what it holds.
pub fn try_put(h: u64, msg: Message, emit: &mut dyn FnMut(ProcessRequest)) -> Result<(), Message> {
    let mut rel = Releases::default();
    let result = with_table(|t| {
        let Some(m) = t.mailboxes.get_mut(&h) else {
            return Err(msg);
        };
        if is_full(m) {
            return Err(msg);
        }
        m.messages.push_back(msg);
        serve_getters(t, h, emit, &mut rel);
        Ok(())
    });
    rel.run();
    result
}

/// The result of a `try_get` / `try_peek`.
#[derive(Debug, PartialEq, Eq)]
pub enum Taken {
    Empty,
    /// The message is not of the requested type (nothing was taken).
    Mismatch,
    Got(Message),
}

fn take_front(
    t: &mut Table,
    h: u64,
    fingerprint: u64,
    peek: bool,
    emit: &mut dyn FnMut(ProcessRequest),
    rel: &mut Releases,
) -> Taken {
    let Some(m) = t.mailboxes.get_mut(&h) else {
        return Taken::Empty;
    };
    let Some(head) = m.messages.front() else {
        return Taken::Empty;
    };
    if head.fingerprint != fingerprint {
        return Taken::Mismatch;
    }
    if peek {
        return Taken::Got(head.peeked());
    }
    let msg = m.messages.pop_front().expect("checked non-empty");
    admit_putters(t, h, emit, rel);
    Taken::Got(msg)
}

/// `try_get` (`peek = false`, 15.4.6) / `try_peek` (`peek = true`, 15.4.8).
pub fn try_take(
    h: u64,
    fingerprint: u64,
    peek: bool,
    emit: &mut dyn FnMut(ProcessRequest),
) -> Taken {
    let mut rel = Releases::default();
    let taken = with_table(|t| take_front(t, h, fingerprint, peek, emit, &mut rel));
    rel.run();
    taken
}

/// What one phase of a blocking `get`/`peek` produced.
#[derive(Debug, PartialEq, Eq)]
pub enum Got {
    /// Nothing to do: the first phase already finished the call.
    Nothing,
    /// The first phase found no such mailbox (a null handle): the call is
    /// finished, and the second phase will report [`Got::Nothing`].
    NoMailbox,
    Message(Message),
    /// The queued message is not of the caller's type (a run-time error).
    Mismatch,
    /// The caller was queued (`emit` got `SyncBlock`).
    Blocked,
}

/// `get` / `peek` (15.4.5, 15.4.7) for process `proc`, in its two phases (see
/// the module comment): the first call takes the head of the queue or queues
/// the process; the second, issued after the process resumed, collects what a
/// waking operation delivered -- or does nothing when the first completed.
pub fn get(
    h: u64,
    proc: ProcessId,
    fingerprint: u64,
    peek: bool,
    emit: &mut dyn FnMut(ProcessRequest),
) -> Got {
    let mut rel = Releases::default();
    let got = with_table(|t| {
        match t.pending.remove(&proc) {
            Some(Pending::Completed) => return Got::Nothing,
            Some(Pending::Delivered(Delivery::Message(m))) => return Got::Message(m),
            Some(Pending::Delivered(Delivery::Mismatch)) => return Got::Mismatch,
            Some(Pending::Waiting(on)) => {
                panic!("mailbox: process {proc} resumed while still blocked on mailbox {on:#x}")
            }
            None => {}
        }
        if !t.mailboxes.contains_key(&h) {
            t.pending.insert(proc, Pending::Completed);
            return Got::NoMailbox;
        }
        match take_front(t, h, fingerprint, peek, emit, &mut rel) {
            Taken::Got(m) => {
                t.pending.insert(proc, Pending::Completed);
                Got::Message(m)
            }
            Taken::Mismatch => {
                t.pending.insert(proc, Pending::Completed);
                Got::Mismatch
            }
            Taken::Empty => {
                objects::retain(h);
                t.mailboxes
                    .get_mut(&h)
                    .expect("looked up above")
                    .getters
                    .push_back(Getter {
                        proc,
                        peek,
                        fingerprint,
                    });
                t.pending.insert(proc, Pending::Waiting(h));
                emit(ProcessRequest::SyncBlock);
                Got::Blocked
            }
        }
    });
    rel.run();
    got
}

/// Whether the blocking `get`/`peek` statement `proc` is running is in its
/// SECOND phase: the first phase always leaves a `pending` entry (a
/// completed marker, a wait or a delivery) that the second one consumes, so a
/// process with none is starting the statement. Compiled code evaluates a
/// side-effecting receiver or destination in the first phase only.
pub fn second_phase(proc: ProcessId) -> bool {
    with_table(|t| t.pending.contains_key(&proc))
}

/// Saves `value` in `slot` of `proc`'s stash (see [`Table::stash`]).
pub fn stash_set(proc: ProcessId, slot: u32, value: u64) {
    with_table(|t| {
        t.stash.insert((proc, slot), value);
    });
}

/// The word last saved by [`stash_set`] for `slot` (`0` when none).
pub fn stash_get(proc: ProcessId, slot: u32) -> u64 {
    with_table(|t| t.stash.get(&(proc, slot)).copied().unwrap_or(0))
}

/// A blocked process ended (killed) before the call finished: drops it from
/// its queue and releases what it held -- a blocked putter's message, a woken
/// waiter's undelivered message. A no-op for a process in no mailbox call.
pub fn cancel(proc: ProcessId) {
    let mut rel = Releases::default();
    with_table(|t| t.stash.retain(|(p, _), _| *p != proc));
    with_table(|t| match t.pending.remove(&proc) {
        None | Some(Pending::Completed) => {}
        Some(Pending::Delivered(Delivery::Mismatch)) => {}
        Some(Pending::Delivered(Delivery::Message(m))) => {
            rel.owned(&m);
        }
        Some(Pending::Waiting(h)) => {
            if let Some(m) = t.mailboxes.get_mut(&h) {
                m.getters.retain(|g| g.proc != proc);
                let mut kept = VecDeque::new();
                for (p, msg) in std::mem::take(&mut m.putters) {
                    if p == proc {
                        rel.owned(&msg);
                    } else {
                        kept.push_back((p, msg));
                    }
                }
                m.putters = kept;
            }
            rel.handles.push(h);
        }
    });
    rel.run();
}

/// Forgets every mailbox call in flight (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 old = with_table(std::mem::take);
    let mut rel = Releases::default();
    for (_, p) in old.pending {
        if let Pending::Delivered(Delivery::Message(m)) = p {
            rel.owned(&m);
        }
    }
    for (h, m) in old.mailboxes {
        for _ in 0..(m.getters.len() + m.putters.len()) {
            rel.handles.push(h);
        }
        for (_, msg) in m.putters {
            rel.owned(&msg);
        }
        for msg in m.messages {
            msg.discard();
        }
    }
    rel.run();
}

#[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 allocates its own mailbox and a disjoint range of process
    // ids, and asserts only on those.

    const FP: u64 = 0x11;
    const OTHER_FP: u64 = 0x22;

    fn mbx(bound: i64, template: u32) -> u64 {
        let h = objects::alloc(ClassTemplateId(template), 1);
        assert!(init(h, bound));
        h
    }

    fn msg(v: u64) -> Message {
        Message {
            words: vec![v].into_boxed_slice(),
            fingerprint: FP,
            obj: false,
            coll: false,
        }
    }

    fn value(m: &Message) -> u64 {
        m.words[0]
    }

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

    #[test]
    fn messages_come_out_in_fifo_order() {
        let h = mbx(0, 18_000);
        let mut log = Log::default();
        for v in [10, 20, 30] {
            assert_eq!(
                put(h, 2_000, msg(v), &mut log.sink()),
                Ok(PutResult::Stored)
            );
        }
        assert_eq!(num(h), 3);
        for want in [10, 20, 30] {
            let Taken::Got(m) = try_take(h, FP, false, &mut log.sink()) else {
                panic!("a message was queued");
            };
            assert_eq!(value(&m), want);
        }
        assert_eq!(try_take(h, FP, false, &mut log.sink()), Taken::Empty);
        assert!(log.0.is_empty());
    }

    #[test]
    fn try_put_fails_on_a_full_bounded_mailbox_and_hands_the_message_back() {
        let h = mbx(2, 18_001);
        let mut log = Log::default();
        assert!(try_put(h, msg(1), &mut log.sink()).is_ok());
        assert!(try_put(h, msg(2), &mut log.sink()).is_ok());
        let back = try_put(h, msg(3), &mut log.sink()).unwrap_err();
        assert_eq!(value(&back), 3);
        assert_eq!(num(h), 2);
    }

    #[test]
    fn an_unbounded_mailbox_is_never_full() {
        let h = mbx(0, 18_002);
        let mut log = Log::default();
        for v in 0..1000 {
            assert!(try_put(h, msg(v), &mut log.sink()).is_ok());
        }
        assert_eq!(num(h), 1000);
        // A negative bound is illegal (15.4.1); it is read as unbounded.
        let n = mbx(-3, 18_003);
        assert!(try_put(n, msg(1), &mut log.sink()).is_ok());
        assert!(try_put(n, msg(2), &mut log.sink()).is_ok());
    }

    #[test]
    fn a_get_on_an_empty_mailbox_blocks_and_a_put_delivers_to_it() {
        let h = mbx(0, 18_004);
        let mut log = Log::default();
        assert_eq!(get(h, 2_010, FP, false, &mut log.sink()), Got::Blocked);
        assert_eq!(log.0, [ProcessRequest::SyncBlock]);
        assert_eq!(waiters(h), 1);
        log.0.clear();
        assert_eq!(
            put(h, 2_011, msg(7), &mut log.sink()),
            Ok(PutResult::Stored)
        );
        assert_eq!(log.0, [ProcessRequest::SyncWake(2_010)]);
        assert_eq!(num(h), 0, "the get took the message");
        match get(h, 2_010, FP, false, &mut log.sink()) {
            Got::Message(m) => assert_eq!(value(&m), 7),
            other => panic!("second phase must deliver: {other:?}"),
        }
        // The collected call is over: the next get starts afresh.
        assert_eq!(get(h, 2_010, FP, false, &mut log.sink()), Got::Blocked);
    }

    #[test]
    fn a_get_that_finds_a_message_completes_and_its_second_phase_is_a_no_op() {
        let h = mbx(0, 18_005);
        let mut log = Log::default();
        put(h, 2_020, msg(5), &mut log.sink()).unwrap();
        match get(h, 2_021, FP, false, &mut log.sink()) {
            Got::Message(m) => assert_eq!(value(&m), 5),
            other => panic!("{other:?}"),
        }
        assert_eq!(get(h, 2_021, FP, false, &mut log.sink()), Got::Nothing);
        assert!(log.0.is_empty());
        assert_eq!(num(h), 0);
    }

    #[test]
    fn waiters_are_served_in_arrival_order() {
        let h = mbx(0, 18_006);
        let mut log = Log::default();
        for p in [2_030, 2_031, 2_032] {
            assert_eq!(get(h, p, FP, false, &mut log.sink()), Got::Blocked);
        }
        log.0.clear();
        put(h, 2_039, msg(1), &mut log.sink()).unwrap();
        assert_eq!(log.0, [ProcessRequest::SyncWake(2_030)]);
        put(h, 2_039, msg(2), &mut log.sink()).unwrap();
        assert_eq!(
            log.0,
            [
                ProcessRequest::SyncWake(2_030),
                ProcessRequest::SyncWake(2_031)
            ]
        );
        assert_eq!(waiters(h), 1);
    }

    #[test]
    fn one_message_wakes_every_peeker_and_the_first_getter() {
        let h = mbx(0, 18_007);
        let mut log = Log::default();
        assert_eq!(get(h, 2_040, FP, true, &mut log.sink()), Got::Blocked);
        assert_eq!(get(h, 2_041, FP, true, &mut log.sink()), Got::Blocked);
        assert_eq!(get(h, 2_042, FP, false, &mut log.sink()), Got::Blocked);
        assert_eq!(get(h, 2_043, FP, true, &mut log.sink()), Got::Blocked);
        log.0.clear();
        put(h, 2_049, msg(9), &mut log.sink()).unwrap();
        assert_eq!(
            log.0,
            [
                ProcessRequest::SyncWake(2_040),
                ProcessRequest::SyncWake(2_041),
                ProcessRequest::SyncWake(2_042)
            ],
            "two peeks copy, the get removes it, the later peek stays blocked"
        );
        assert_eq!(num(h), 0);
        assert_eq!(waiters(h), 1);
        for p in [2_040, 2_041, 2_042] {
            match get(h, p, FP, p == 2_040 || p == 2_041, &mut log.sink()) {
                Got::Message(m) => assert_eq!(value(&m), 9),
                other => panic!("process {p}: {other:?}"),
            }
        }
    }

    #[test]
    fn peek_leaves_the_message_in_the_queue() {
        let h = mbx(0, 18_008);
        let mut log = Log::default();
        put(h, 2_050, msg(4), &mut log.sink()).unwrap();
        let Taken::Got(m) = try_take(h, FP, true, &mut log.sink()) else {
            panic!("a message was queued");
        };
        assert_eq!(value(&m), 4);
        assert_eq!(num(h), 1);
    }

    #[test]
    fn a_put_on_a_full_mailbox_blocks_until_a_get_makes_room() {
        let h = mbx(1, 18_009);
        let mut log = Log::default();
        put(h, 2_060, msg(1), &mut log.sink()).unwrap();
        assert_eq!(
            put(h, 2_061, msg(2), &mut log.sink()),
            Ok(PutResult::Blocked)
        );
        assert_eq!(log.0, [ProcessRequest::SyncBlock]);
        assert_eq!(num(h), 1, "the blocked message is not queued yet");
        log.0.clear();
        let Taken::Got(m) = try_take(h, FP, false, &mut log.sink()) else {
            panic!("a message was queued");
        };
        assert_eq!(value(&m), 1);
        assert_eq!(log.0, [ProcessRequest::SyncWake(2_061)]);
        assert_eq!(num(h), 1, "the blocked putter's message moved in");
        let Taken::Got(m) = try_take(h, FP, false, &mut log.sink()) else {
            panic!("a message was queued");
        };
        assert_eq!(value(&m), 2);
    }

    #[test]
    fn a_blocked_putter_is_not_overtaken_by_try_put() {
        let h = mbx(1, 18_010);
        let mut log = Log::default();
        put(h, 2_070, msg(1), &mut log.sink()).unwrap();
        put(h, 2_071, msg(2), &mut log.sink()).unwrap();
        try_take(h, FP, false, &mut log.sink());
        assert!(
            try_put(h, msg(3), &mut log.sink()).is_err(),
            "the freed slot went to the blocked putter"
        );
    }

    #[test]
    fn a_type_mismatch_takes_nothing() {
        let h = mbx(0, 18_011);
        let mut log = Log::default();
        put(h, 2_080, msg(1), &mut log.sink()).unwrap();
        assert_eq!(
            try_take(h, OTHER_FP, false, &mut log.sink()),
            Taken::Mismatch
        );
        assert_eq!(
            try_take(h, OTHER_FP, true, &mut log.sink()),
            Taken::Mismatch
        );
        assert_eq!(
            get(h, 2_081, OTHER_FP, false, &mut log.sink()),
            Got::Mismatch
        );
        assert_eq!(num(h), 1, "the message is still there");
        // A waiter whose type is wrong is woken with the error, the message
        // stays for the next one.
        let e = mbx(0, 18_012);
        assert_eq!(
            get(e, 2_082, OTHER_FP, false, &mut log.sink()),
            Got::Blocked
        );
        assert_eq!(get(e, 2_083, FP, false, &mut log.sink()), Got::Blocked);
        log.0.clear();
        put(e, 2_084, msg(8), &mut log.sink()).unwrap();
        assert_eq!(
            log.0,
            [
                ProcessRequest::SyncWake(2_082),
                ProcessRequest::SyncWake(2_083)
            ]
        );
        assert_eq!(
            get(e, 2_082, OTHER_FP, false, &mut log.sink()),
            Got::Mismatch
        );
        assert!(matches!(
            get(e, 2_083, FP, false, &mut log.sink()),
            Got::Message(_)
        ));
    }

    fn obj_msg(h: u64) -> Message {
        Message {
            words: vec![h].into_boxed_slice(),
            fingerprint: FP,
            obj: true,
            coll: false,
        }
    }

    #[test]
    fn a_class_message_owns_one_reference_until_it_is_taken() {
        let h = mbx(0, 18_013);
        let payload = objects::alloc(ClassTemplateId(18_014), 1);
        let mut log = Log::default();
        put(h, 2_090, obj_msg(payload), &mut log.sink()).unwrap();
        assert_eq!(objects::refcount(payload), Some(1), "the message holds it");
        let Taken::Got(copy) = try_take(h, FP, true, &mut log.sink()) else {
            panic!("a message was queued");
        };
        assert_eq!(
            objects::refcount(payload),
            Some(2),
            "a peeked copy owns its own"
        );
        copy.discard();
        let Taken::Got(taken) = try_take(h, FP, false, &mut log.sink()) else {
            panic!("a message was queued");
        };
        assert_eq!(
            objects::refcount(payload),
            Some(1),
            "taking transfers the queue's reference"
        );
        taken.discard();
        assert_eq!(objects::refcount(payload), None, "the last holder freed it");
    }

    #[test]
    fn cancelling_a_blocked_putter_releases_its_message_and_the_mailbox() {
        let h = mbx(1, 18_015);
        let payload = objects::alloc(ClassTemplateId(18_016), 1);
        let mut log = Log::default();
        put(h, 2_100, msg(1), &mut log.sink()).unwrap();
        assert_eq!(
            put(h, 2_101, obj_msg(payload), &mut log.sink()),
            Ok(PutResult::Blocked)
        );
        assert_eq!(objects::refcount(h), Some(2));
        cancel(2_101);
        assert_eq!(objects::refcount(payload), None, "its message was dropped");
        assert_eq!(objects::refcount(h), Some(1));
        assert_eq!(waiters(h), 0);
        // Cancelling a process in no mailbox call is a no-op.
        cancel(2_101);
    }

    #[test]
    fn cancelling_a_blocked_getter_removes_it_and_a_woken_one_loses_its_message() {
        let h = mbx(0, 18_017);
        let mut log = Log::default();
        assert_eq!(get(h, 2_110, FP, false, &mut log.sink()), Got::Blocked);
        cancel(2_110);
        assert_eq!(waiters(h), 0);
        assert_eq!(objects::refcount(h), Some(1));
        put(h, 2_111, msg(1), &mut log.sink()).unwrap();
        assert_eq!(num(h), 1, "nobody is waiting any more");

        let payload = objects::alloc(ClassTemplateId(18_018), 1);
        let e = mbx(0, 18_019);
        assert_eq!(get(e, 2_112, FP, false, &mut log.sink()), Got::Blocked);
        put(e, 2_113, obj_msg(payload), &mut log.sink()).unwrap();
        cancel(2_112);
        assert_eq!(
            objects::refcount(payload),
            None,
            "the lost message was freed"
        );
    }

    #[test]
    fn a_waiter_keeps_its_mailbox_alive_until_it_leaves() {
        let h = mbx(0, 18_020);
        let mut log = Log::default();
        assert_eq!(get(h, 2_120, FP, false, &mut log.sink()), Got::Blocked);
        assert_eq!(objects::refcount(h), Some(2));
        put(h, 2_121, msg(1), &mut log.sink()).unwrap();
        assert_eq!(objects::refcount(h), Some(1), "the woken waiter let go");
    }

    #[test]
    fn a_non_mailbox_handle_stores_and_delivers_nothing() {
        let mut log = Log::default();
        assert!(!init(0, 0));
        assert_eq!(num(0), 0);
        assert_eq!(put(0, 2_130, msg(1), &mut log.sink()), Err(msg(1)));
        assert!(try_put(0, msg(1), &mut log.sink()).is_err());
        assert_eq!(try_take(0, FP, false, &mut log.sink()), Taken::Empty);
        // Phase one names the missing mailbox (the caller reports it once);
        // phase two of the finished call does nothing.
        assert_eq!(get(0, 2_131, FP, false, &mut log.sink()), Got::NoMailbox);
        assert_eq!(get(0, 2_131, FP, false, &mut log.sink()), Got::Nothing);
        assert!(log.0.is_empty());
    }
}