smarm/tests/cancel.rs

//! Cooperative cancellation tests (roadmap item #1, the keystone).
//!
//! `request_stop(pid)` flags an actor for cancellation. The actor realizes the
//! stop as a *controlled unwind*: at the next observation point (a `check!()`
//! / allocation via `maybe_preempt`, or the wakeup side of any blocking park)
//! a dedicated sentinel panic is raised, the trampoline's `catch_unwind` tears
//! the stack down — running Drop guards — and reports `Outcome::Stopped`,
//! distinct from a user `Panic`. Monitors see `DownReason::Stopped`.
//!
//! These cases are deterministic under the single-thread runtime: the parent
//! runs until it parks, so the relative order of `request_stop`, the child
//! reaching its observation point, and the monitor `Down` is fixed.

use smarm::{channel, monitor, request_stop, run, spawn, yield_now, DownReason};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;

/// Sets its flag when dropped — used to prove the cancellation unwind runs
/// Drop guards rather than leaking the stack.
struct DropFlag(Arc<AtomicBool>);
impl Drop for DropFlag {
    fn drop(&mut self) {
        self.0.store(true, Ordering::SeqCst);
    }
}

#[test]
fn looping_actor_on_check_is_stopped() {
    let dropped = Arc::new(AtomicBool::new(false));
    let saw_stopped = Arc::new(AtomicBool::new(false));
    let (d, s) = (dropped.clone(), saw_stopped.clone());
    run(move || {
        let h = spawn(move || {
            let _g = DropFlag(d);
            // Tight loop whose only observation point is check!().
            loop {
                smarm::check!();
            }
        });
        let pid = h.pid();
        let down = monitor(pid);
        // Flag the stop before the child is ever resumed; it will observe the
        // flag once its check!() loop reaches the amortised preempt check.
        request_stop(pid);
        let dn = down.rx.recv().expect("monitor channel closed before Down");
        assert_eq!(dn.pid, pid);
        if matches!(dn.reason, DownReason::Stopped) {
            s.store(true, Ordering::SeqCst);
        }
        let _ = h.join();
    });
    assert!(saw_stopped.load(Ordering::SeqCst), "expected DownReason::Stopped");
    assert!(dropped.load(Ordering::SeqCst), "Drop guard must run during the cancellation unwind");
}

#[test]
fn parked_on_recv_actor_is_stopped() {
    let dropped = Arc::new(AtomicBool::new(false));
    let saw_stopped = Arc::new(AtomicBool::new(false));
    let (d, s) = (dropped.clone(), saw_stopped.clone());
    run(move || {
        let h = spawn(move || {
            let _g = DropFlag(d);
            let (tx, rx) = channel::<u8>();
            // Keep a sender alive so the channel stays open and recv() parks
            // indefinitely rather than returning Err.
            let _keep = tx;
            let _ = rx.recv(); // parks here until the stop unwinds us out
        });
        let pid = h.pid();
        let down = monitor(pid);
        // Let the child run and park in recv() before we request the stop.
        yield_now();
        request_stop(pid);
        let dn = down.rx.recv().expect("monitor channel closed before Down");
        assert_eq!(dn.pid, pid);
        if matches!(dn.reason, DownReason::Stopped) {
            s.store(true, Ordering::SeqCst);
        }
        let _ = h.join();
    });
    assert!(saw_stopped.load(Ordering::SeqCst), "expected DownReason::Stopped");
    assert!(dropped.load(Ordering::SeqCst), "Drop guard must run on cancellation of a parked actor");
}

#[test]
fn no_check_no_alloc_loop_is_not_stopped() {
    // Documents the inherent gap: an actor that never reaches an observation
    // point (no check!(), no allocation, no blocking op) cannot be stopped —
    // the same limitation as preemption. Here the child runs a bounded,
    // allocation-free arithmetic loop, so the stop flagged before it runs is
    // silently never honored and the actor exits normally.
    let saw_exit = Arc::new(AtomicBool::new(false));
    let s = saw_exit.clone();
    run(move || {
        let h = spawn(|| {
            let mut x: u64 = 0;
            for i in 0..2_000_000u64 {
                x = x.wrapping_add(i ^ (x >> 1));
            }
            std::hint::black_box(x);
        });
        let pid = h.pid();
        let down = monitor(pid);
        request_stop(pid); // no observation point => ignored
        let dn = down.rx.recv().expect("monitor channel closed before Down");
        if matches!(dn.reason, DownReason::Exit) {
            s.store(true, Ordering::SeqCst);
        }
        let _ = h.join();
    });
    assert!(
        saw_exit.load(Ordering::SeqCst),
        "a loop with no observation points must exit normally, never Stopped"
    );
}

#[test]
fn join_on_stopped_actor_returns_ok() {
    // A cooperative stop carries no panic payload to propagate, so join()
    // reports Ok(()); the *fact* of the stop is observable via monitors
    // (DownReason::Stopped), which is the channel that carries termination
    // reason.
    run(|| {
        let h = spawn(|| loop {
            smarm::check!();
        });
        let pid = h.pid();
        request_stop(pid);
        assert!(h.join().is_ok(), "join on a stopped actor returns Ok(())");
    });
}