smarm
a0a93b61bb
The deadline is one more stamped waker on the select's wait epoch: a unit TimerTarget whose on_timeout is a bare unpark_at. Nothing registers in any arm for it, so there is nothing to cancel or leak — an arm winning leaves the timer entry to die at its epoch CAS; the timer winning leaves the arms' registrations to self-clean like any select loser's. Classification is pure channel state (wakes are precise): some arm ready -> Some(first, priority order); none -> the timer was the only remaining stamped waker -> None. Message-first on a raced deadline, same as recv_timeout. Registration pass factored into register_arms, which owns the retire_wait obligation on the ready-now exit for both entry points.
341 lines
11 KiB
Rust
341 lines
11 KiB
Rust
//! select tests. Beyond the functional cases, the *_stays_precise tests are
|
|
//! soundness probes for the consuming-wake protocol: they fire stale loser-
|
|
//! arm wakes at an actor and then run a one-shot park (`sleep`, whose early
|
|
//! return would be observable as a short elapsed time) to prove the stale
|
|
//! wake died at its epoch CAS instead of corrupting the next wait.
|
|
|
|
use smarm::{channel, run, select, spawn};
|
|
use std::sync::atomic::{AtomicI64, Ordering};
|
|
use std::time::{Duration, Instant};
|
|
|
|
static OUT: AtomicI64 = AtomicI64::new(0);
|
|
|
|
#[test]
|
|
fn ready_arm_returns_immediately_without_parking() {
|
|
OUT.store(0, Ordering::SeqCst);
|
|
run(|| {
|
|
let (txa, rxa) = channel::<i64>();
|
|
let (_txb, rxb) = channel::<i64>();
|
|
txa.send(42).unwrap();
|
|
let i = select(&[&rxb, &rxa]);
|
|
assert_eq!(i, 1);
|
|
OUT.store(rxa.try_recv().unwrap().expect("ready arm must hold a message"), Ordering::SeqCst);
|
|
});
|
|
assert_eq!(OUT.load(Ordering::SeqCst), 42);
|
|
}
|
|
|
|
#[test]
|
|
fn lower_index_wins_when_several_arms_are_ready() {
|
|
run(|| {
|
|
let (txa, rxa) = channel::<i64>();
|
|
let (txb, rxb) = channel::<i64>();
|
|
txa.send(1).unwrap();
|
|
txb.send(2).unwrap();
|
|
// Documented priority order: index 0 first, regardless of send order.
|
|
assert_eq!(select(&[&rxa, &rxb]), 0);
|
|
assert_eq!(select(&[&rxb, &rxa]), 0);
|
|
});
|
|
}
|
|
|
|
#[test]
|
|
fn parks_until_any_arm_fires() {
|
|
OUT.store(0, Ordering::SeqCst);
|
|
run(|| {
|
|
let (txa, _rxa_keepalive) = (channel::<i64>().0, ());
|
|
let _hold = txa; // arm a: sender alive, never sends
|
|
let (txa, rxa) = channel::<i64>();
|
|
let (txb, rxb) = channel::<i64>();
|
|
let _keep_a_open = txa;
|
|
let h = spawn(move || {
|
|
smarm::sleep(Duration::from_millis(10));
|
|
txb.send(7).unwrap();
|
|
});
|
|
let i = select(&[&rxa, &rxb]);
|
|
assert_eq!(i, 1);
|
|
OUT.store(rxb.try_recv().unwrap().unwrap(), Ordering::SeqCst);
|
|
h.join().unwrap();
|
|
});
|
|
assert_eq!(OUT.load(Ordering::SeqCst), 7);
|
|
}
|
|
|
|
#[test]
|
|
fn closed_arm_counts_as_ready() {
|
|
run(|| {
|
|
let (_keep, rxa) = channel::<i64>();
|
|
let (txb, rxb) = channel::<i64>();
|
|
drop(txb);
|
|
let i = select(&[&rxa, &rxb]);
|
|
assert_eq!(i, 1);
|
|
// The caller observes the closure on the arm itself.
|
|
assert!(rxb.try_recv().is_err());
|
|
});
|
|
}
|
|
|
|
#[test]
|
|
fn closure_while_parked_wakes_the_select() {
|
|
run(|| {
|
|
let (_keep, rxa) = channel::<i64>();
|
|
let (txb, rxb) = channel::<i64>();
|
|
let h = spawn(move || {
|
|
smarm::sleep(Duration::from_millis(10));
|
|
drop(txb);
|
|
});
|
|
assert_eq!(select(&[&rxa, &rxb]), 1);
|
|
assert!(rxb.try_recv().is_err());
|
|
h.join().unwrap();
|
|
});
|
|
}
|
|
|
|
#[test]
|
|
fn loser_arm_wake_after_parked_select_stays_precise() {
|
|
// Both arms registered; arm 0 wins the wake; arm 1's sender then fires
|
|
// a wake stamped with the consumed epoch. The subsequent sleep is a
|
|
// one-shot park: an early return would mean the stale wake landed.
|
|
run(|| {
|
|
let (txa, rxa) = channel::<i64>();
|
|
let (txb, rxb) = channel::<i64>();
|
|
let h = spawn(move || {
|
|
smarm::sleep(Duration::from_millis(10));
|
|
txa.send(1).unwrap(); // wins
|
|
txb.send(2).unwrap(); // loser arm: stale-epoch wake
|
|
});
|
|
assert_eq!(select(&[&rxa, &rxb]), 0);
|
|
assert_eq!(rxa.try_recv().unwrap(), Some(1));
|
|
let t0 = Instant::now();
|
|
smarm::sleep(Duration::from_millis(40));
|
|
assert!(
|
|
t0.elapsed() >= Duration::from_millis(40),
|
|
"one-shot park returned early: a stale loser-arm wake landed"
|
|
);
|
|
// The loser's message was never lost.
|
|
assert_eq!(select(&[&rxa, &rxb]), 1);
|
|
assert_eq!(rxb.try_recv().unwrap(), Some(2));
|
|
h.join().unwrap();
|
|
});
|
|
}
|
|
|
|
#[test]
|
|
fn no_park_exit_retires_the_wait() {
|
|
// Arm 1 is ready at registration time, so select returns WITHOUT
|
|
// parking while arm 0 holds a live-epoch registration. Arm 0's sender
|
|
// then fires. retire_wait must have invalidated/eaten that wake;
|
|
// the sleep proves it.
|
|
run(|| {
|
|
let (txa, rxa) = channel::<i64>();
|
|
let (txb, rxb) = channel::<i64>();
|
|
txb.send(2).unwrap();
|
|
assert_eq!(select(&[&rxa, &rxb]), 1);
|
|
assert_eq!(rxb.try_recv().unwrap(), Some(2));
|
|
|
|
let h = spawn(move || {
|
|
txa.send(1).unwrap(); // fires at the (retired) select epoch
|
|
});
|
|
h.join().unwrap();
|
|
|
|
let t0 = Instant::now();
|
|
smarm::sleep(Duration::from_millis(40));
|
|
assert!(
|
|
t0.elapsed() >= Duration::from_millis(40),
|
|
"one-shot park returned early: the no-park exit leaked a wake"
|
|
);
|
|
assert_eq!(rxa.try_recv().unwrap(), Some(1));
|
|
});
|
|
}
|
|
|
|
#[test]
|
|
fn select_then_plain_recv_on_a_loser_arm() {
|
|
// A leftover own-registration from a select must not trip the next
|
|
// direct wait on that channel (it is overwritten, not asserted away).
|
|
run(|| {
|
|
let (txa, rxa) = channel::<i64>();
|
|
let (txb, rxb) = channel::<i64>();
|
|
txa.send(1).unwrap();
|
|
assert_eq!(select(&[&rxa, &rxb]), 0); // rxb may keep a registration
|
|
assert_eq!(rxa.try_recv().unwrap(), Some(1));
|
|
let h = spawn(move || {
|
|
smarm::sleep(Duration::from_millis(5));
|
|
txb.send(9).unwrap();
|
|
});
|
|
assert_eq!(rxb.recv().unwrap(), 9);
|
|
h.join().unwrap();
|
|
});
|
|
}
|
|
|
|
#[test]
|
|
fn select_loop_drains_two_producers_completely() {
|
|
const N: i64 = 200;
|
|
OUT.store(0, Ordering::SeqCst);
|
|
run(|| {
|
|
let (txa, rxa) = channel::<i64>();
|
|
let (txb, rxb) = channel::<i64>();
|
|
let ha = spawn(move || {
|
|
for v in 0..N {
|
|
txa.send(v).unwrap();
|
|
if v % 7 == 0 {
|
|
smarm::yield_now();
|
|
}
|
|
}
|
|
});
|
|
let hb = spawn(move || {
|
|
for v in 0..N {
|
|
txb.send(v).unwrap();
|
|
if v % 5 == 0 {
|
|
smarm::yield_now();
|
|
}
|
|
}
|
|
});
|
|
// A closed arm stays permanently ready, so it must leave the arm
|
|
// set once observed (keeping it would starve the other arm at
|
|
// priority order) — select until the FIRST closure, then drain the
|
|
// survivor with plain recv.
|
|
let mut sum = 0i64;
|
|
let survivor: &smarm::Receiver<i64> = loop {
|
|
let i = select(&[&rxa, &rxb]);
|
|
let arm: &smarm::Receiver<i64> = if i == 0 { &rxa } else { &rxb };
|
|
match arm.try_recv() {
|
|
Ok(Some(v)) => sum += v,
|
|
Ok(None) => {} // defensive: ready arm already drained
|
|
Err(_) => break if i == 0 { &rxb } else { &rxa },
|
|
}
|
|
};
|
|
loop {
|
|
match survivor.recv() {
|
|
Ok(v) => sum += v,
|
|
Err(_) => break,
|
|
}
|
|
}
|
|
OUT.store(sum, Ordering::SeqCst);
|
|
ha.join().unwrap();
|
|
hb.join().unwrap();
|
|
});
|
|
assert_eq!(OUT.load(Ordering::SeqCst), 2 * (0..200i64).sum::<i64>());
|
|
}
|
|
|
|
#[test]
|
|
fn motivating_pattern_inbox_plus_monitor_down() {
|
|
// The gen_server/handle_info shape select unlocks: a server waiting on
|
|
// its request inbox AND a monitor Down channel in one park.
|
|
use smarm::monitor;
|
|
run(|| {
|
|
let (req_tx, req_rx) = channel::<i64>();
|
|
let worker = spawn(|| {
|
|
smarm::sleep(Duration::from_millis(10));
|
|
// worker exits -> Down fires
|
|
});
|
|
let m = monitor(worker.pid());
|
|
let _keep_inbox_open = req_tx;
|
|
|
|
let mut served = 0;
|
|
loop {
|
|
match select(&[&m.rx, &req_rx]) {
|
|
0 => {
|
|
let _down = m.rx.try_recv().unwrap().expect("Down message");
|
|
break;
|
|
}
|
|
_ => {
|
|
if let Ok(Some(_)) = req_rx.try_recv() {
|
|
served += 1;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
assert_eq!(served, 0);
|
|
worker.join().unwrap();
|
|
});
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// select_timeout
|
|
// ---------------------------------------------------------------------------
|
|
|
|
use smarm::select_timeout;
|
|
|
|
#[test]
|
|
fn select_timeout_returns_none_after_the_deadline() {
|
|
run(|| {
|
|
let (_keep_a, rxa) = channel::<i64>();
|
|
let (_keep_b, rxb) = channel::<i64>();
|
|
let t0 = Instant::now();
|
|
let r = select_timeout(&[&rxa, &rxb], Duration::from_millis(30));
|
|
assert_eq!(r, None);
|
|
assert!(t0.elapsed() >= Duration::from_millis(30));
|
|
});
|
|
}
|
|
|
|
#[test]
|
|
fn select_timeout_ready_arm_wins_without_arming_a_timer() {
|
|
run(|| {
|
|
let (txa, rxa) = channel::<i64>();
|
|
let (_keep_b, rxb) = channel::<i64>();
|
|
txa.send(5).unwrap();
|
|
assert_eq!(select_timeout(&[&rxb, &rxa], Duration::from_millis(500)), Some(1));
|
|
assert_eq!(rxa.try_recv().unwrap(), Some(5));
|
|
});
|
|
}
|
|
|
|
#[test]
|
|
fn select_timeout_arm_beats_timer_and_stale_entry_stays_inert() {
|
|
run(|| {
|
|
let (txa, rxa) = channel::<i64>();
|
|
let (_keep_b, rxb) = channel::<i64>();
|
|
let h = spawn(move || {
|
|
smarm::sleep(Duration::from_millis(5));
|
|
txa.send(1).unwrap();
|
|
});
|
|
let t0 = Instant::now();
|
|
let r = select_timeout(&[&rxa, &rxb], Duration::from_millis(200));
|
|
assert_eq!(r, Some(0));
|
|
assert!(t0.elapsed() < Duration::from_millis(200));
|
|
assert_eq!(rxa.try_recv().unwrap(), Some(1));
|
|
h.join().unwrap();
|
|
// The abandoned timer entry expires mid-sleep; a stale-epoch wake
|
|
// landing would return this one-shot park early.
|
|
let t1 = Instant::now();
|
|
smarm::sleep(Duration::from_millis(250));
|
|
assert!(
|
|
t1.elapsed() >= Duration::from_millis(250),
|
|
"one-shot park returned early: the stale select timer landed"
|
|
);
|
|
});
|
|
}
|
|
|
|
#[test]
|
|
fn select_timeout_timer_first_message_still_delivered_later() {
|
|
run(|| {
|
|
let (txa, rxa) = channel::<i64>();
|
|
let h = spawn(move || {
|
|
smarm::sleep(Duration::from_millis(60));
|
|
txa.send(9).unwrap();
|
|
});
|
|
assert_eq!(select_timeout(&[&rxa], Duration::from_millis(10)), None);
|
|
// The wait is over but the channel is intact: the late message
|
|
// arrives and a fresh select sees it.
|
|
assert_eq!(select(&[&rxa]), 0);
|
|
assert_eq!(rxa.try_recv().unwrap(), Some(9));
|
|
h.join().unwrap();
|
|
});
|
|
}
|
|
|
|
#[test]
|
|
fn select_timeout_zero_duration_polls() {
|
|
run(|| {
|
|
let (txa, rxa) = channel::<i64>();
|
|
let (_keep_b, rxb) = channel::<i64>();
|
|
assert_eq!(select_timeout(&[&rxa, &rxb], Duration::ZERO), None);
|
|
txa.send(3).unwrap();
|
|
assert_eq!(select_timeout(&[&rxa, &rxb], Duration::ZERO), Some(0));
|
|
assert_eq!(rxa.try_recv().unwrap(), Some(3));
|
|
});
|
|
}
|
|
|
|
#[test]
|
|
fn select_timeout_closed_arm_is_ready_not_a_timeout() {
|
|
run(|| {
|
|
let (_keep_a, rxa) = channel::<i64>();
|
|
let (txb, rxb) = channel::<i64>();
|
|
drop(txb);
|
|
assert_eq!(select_timeout(&[&rxa, &rxb], Duration::from_millis(200)), Some(1));
|
|
assert!(rxb.try_recv().is_err());
|
|
});
|
|
}
|