Hand-rolled context switching on mmap'd stacks with guard pages, allocator-driven RDTSC preemption, unbounded MPSC channels, supervision via per-slot Signal mailboxes, root supervisor as sentinel PID. Lib + tests + benches clean check/clippy. All 29 tests pass. Bench: smarm 3.4% over serial baseline, within 160us of tokio current-thread on prime-counting fan-out.
172 lines
5.5 KiB
Rust
172 lines
5.5 KiB
Rust
//! End-to-end scheduler tests: spawning, joining, panic delivery,
|
|
//! yield_now, self_pid.
|
|
|
|
use smarm::{channel, run, self_pid, spawn, spawn_under, yield_now, Signal};
|
|
use std::cell::Cell;
|
|
use std::sync::atomic::{AtomicI64, Ordering};
|
|
use std::sync::Arc;
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Single root actor runs to completion
|
|
// ---------------------------------------------------------------------------
|
|
|
|
#[test]
|
|
fn root_actor_runs() {
|
|
let captured = Arc::new(AtomicI64::new(0));
|
|
let c = captured.clone();
|
|
run(move || { c.store(99, Ordering::SeqCst); });
|
|
assert_eq!(captured.load(Ordering::SeqCst), 99);
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Spawn child, join it
|
|
// ---------------------------------------------------------------------------
|
|
|
|
#[test]
|
|
fn spawn_and_join_returns_exit() {
|
|
let captured = Arc::new(AtomicI64::new(0));
|
|
let c = captured.clone();
|
|
run(move || {
|
|
let h = spawn(move || { c.store(7, Ordering::SeqCst); });
|
|
let res = h.join();
|
|
assert!(res.is_ok(), "join returned {:?}", res);
|
|
});
|
|
assert_eq!(captured.load(Ordering::SeqCst), 7);
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// yield_now lets a sibling run
|
|
// ---------------------------------------------------------------------------
|
|
|
|
#[test]
|
|
fn yield_now_interleaves_actors() {
|
|
let log: Arc<std::sync::Mutex<Vec<u8>>> = Arc::new(std::sync::Mutex::new(Vec::new()));
|
|
let l1 = log.clone();
|
|
let l2 = log.clone();
|
|
run(move || {
|
|
let h1 = spawn(move || {
|
|
l1.lock().unwrap().push(1);
|
|
yield_now();
|
|
l1.lock().unwrap().push(3);
|
|
});
|
|
let h2 = spawn(move || {
|
|
l2.lock().unwrap().push(2);
|
|
yield_now();
|
|
l2.lock().unwrap().push(4);
|
|
});
|
|
h1.join().unwrap();
|
|
h2.join().unwrap();
|
|
});
|
|
// Both actors get their first step before either second step. Exact order
|
|
// is FIFO: 1, 2, then 3, 4.
|
|
assert_eq!(*log.lock().unwrap(), vec![1, 2, 3, 4]);
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// self_pid returns this actor's pid inside the actor
|
|
// ---------------------------------------------------------------------------
|
|
|
|
#[test]
|
|
fn self_pid_is_stable_within_an_actor() {
|
|
let pid_cell: Arc<std::sync::Mutex<Option<smarm::Pid>>> =
|
|
Arc::new(std::sync::Mutex::new(None));
|
|
let p2 = pid_cell.clone();
|
|
run(move || {
|
|
let h = spawn(move || {
|
|
let me = self_pid();
|
|
yield_now();
|
|
assert_eq!(me, self_pid(), "self_pid changed across yield");
|
|
*p2.lock().unwrap() = Some(me);
|
|
});
|
|
h.join().unwrap();
|
|
});
|
|
assert!(pid_cell.lock().unwrap().is_some());
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Panic is captured; join returns Err; supervisor receives Signal::Panic
|
|
// ---------------------------------------------------------------------------
|
|
|
|
#[test]
|
|
fn panicking_child_returns_join_error() {
|
|
let saw_err = Arc::new(std::sync::atomic::AtomicBool::new(false));
|
|
let s = saw_err.clone();
|
|
run(move || {
|
|
let h = spawn(|| panic!("kaboom"));
|
|
if h.join().is_err() {
|
|
s.store(true, Ordering::SeqCst);
|
|
}
|
|
});
|
|
|
|
assert!(saw_err.load(Ordering::SeqCst));
|
|
}
|
|
|
|
#[test]
|
|
fn supervisor_receives_panic_signal() {
|
|
let saw_panic_signal = Arc::new(std::sync::atomic::AtomicBool::new(false));
|
|
let s = saw_panic_signal.clone();
|
|
|
|
run(move || {
|
|
// Build a supervisor actor with its own mailbox.
|
|
let (sig_tx, sig_rx) = channel::<Signal>();
|
|
let sup_handle = spawn(move || {
|
|
// Wait for exactly one signal.
|
|
let sig = sig_rx.recv().unwrap();
|
|
if let Signal::Panic(_, _) = sig {
|
|
s.store(true, Ordering::SeqCst);
|
|
}
|
|
});
|
|
// Tell the runtime: when I spawn the next child, route signals here.
|
|
let sup_pid = sup_handle.pid();
|
|
smarm::scheduler::register_supervisor_channel(sup_pid, sig_tx);
|
|
|
|
let child = spawn_under(sup_pid, || panic!("oops"));
|
|
let _ = child.join();
|
|
sup_handle.join().unwrap();
|
|
});
|
|
|
|
assert!(saw_panic_signal.load(Ordering::SeqCst));
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Multiple children, all complete, parent gets back control
|
|
// ---------------------------------------------------------------------------
|
|
|
|
#[test]
|
|
fn many_children_all_complete() {
|
|
let counter = Arc::new(AtomicI64::new(0));
|
|
let c = counter.clone();
|
|
run(move || {
|
|
let mut handles = Vec::new();
|
|
for _ in 0..10 {
|
|
let cc = c.clone();
|
|
handles.push(spawn(move || {
|
|
cc.fetch_add(1, Ordering::SeqCst);
|
|
}));
|
|
}
|
|
for h in handles {
|
|
h.join().unwrap();
|
|
}
|
|
});
|
|
assert_eq!(counter.load(Ordering::SeqCst), 10);
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Repeated yield_now inside an actor with no other actors completes
|
|
// ---------------------------------------------------------------------------
|
|
|
|
#[test]
|
|
fn yield_alone_terminates() {
|
|
thread_local! {
|
|
static N: Cell<i32> = const { Cell::new(0) };
|
|
}
|
|
N.with(|c| c.set(0));
|
|
run(|| {
|
|
for _ in 0..5 {
|
|
N.with(|c| c.set(c.get() + 1));
|
|
yield_now();
|
|
}
|
|
});
|
|
assert_eq!(N.with(|c| c.get()), 5);
|
|
}
|