smarm
134ff52c8a
The per-recv deadline the roadmap deferred, built exactly the way Mutex::lock_timeout already works: register the wait with a per-wait seq, arm a timer::Reason::WaitTimeout with the channel inner (now a TimerTarget) as the target, park. On expiry on_timeout cancels the wait only if that same seq is still parked; satisfied or abandoned waits leave a stale heap entry that no-ops on seq mismatch, per the no-cancellation convention in timer.rs. Race resolution is message-first: a send that lands by the time the woken receiver runs is delivered even if the deadline had also passed. Closure is reported as RecvTimeoutError::Disconnected, keeping timeout and server/sender death distinguishable for callers (gen_server call timeout builds on exactly this distinction next). The timer is armed outside the channel critical section (the timers lock must never nest under a Channel-class lock); the unpark race this opens is absorbed by the RunningNotified protocol. Tests: immediate delivery, actual timeout (with elapsed check), prompt wake on send, Disconnected on close, zero-duration poll, post-timeout seq isolation (stale entry must not cancel later waits), and a 4-scheduler mixed-outcome run (12 fed / 12 timed out).
297 lines
9.2 KiB
Rust
297 lines
9.2 KiB
Rust
//! Channel tests. These run under the scheduler because `recv()` needs to
|
|
//! be able to park, which requires a live runtime.
|
|
|
|
use smarm::{channel, run, spawn};
|
|
use std::cell::Cell;
|
|
|
|
thread_local! {
|
|
static OUT: Cell<i64> = const { Cell::new(0) };
|
|
}
|
|
|
|
#[test]
|
|
fn send_then_recv_same_actor() {
|
|
OUT.with(|c| c.set(0));
|
|
run(|| {
|
|
let (tx, rx) = channel::<i64>();
|
|
tx.send(42).unwrap();
|
|
let v = rx.recv().unwrap();
|
|
OUT.with(|c| c.set(v));
|
|
});
|
|
assert_eq!(OUT.with(|c| c.get()), 42);
|
|
}
|
|
|
|
#[test]
|
|
fn recv_parks_until_send_from_other_actor() {
|
|
OUT.with(|c| c.set(0));
|
|
run(|| {
|
|
let (tx, rx) = channel::<i64>();
|
|
let h = spawn(move || {
|
|
// This actor blocks on an empty channel.
|
|
let v = rx.recv().unwrap();
|
|
OUT.with(|c| c.set(v));
|
|
});
|
|
// Parent runs, then yields to let the child block,
|
|
// then sends, then joins.
|
|
smarm::yield_now();
|
|
tx.send(7).unwrap();
|
|
h.join().unwrap();
|
|
});
|
|
assert_eq!(OUT.with(|c| c.get()), 7);
|
|
}
|
|
|
|
#[test]
|
|
fn multiple_messages_arrive_in_order() {
|
|
let captured: std::sync::Arc<std::sync::Mutex<Vec<i64>>> =
|
|
std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
|
|
let cap2 = captured.clone();
|
|
|
|
run(move || {
|
|
let (tx, rx) = channel::<i64>();
|
|
let h = spawn(move || {
|
|
for _ in 0..3 {
|
|
let v = rx.recv().unwrap();
|
|
cap2.lock().unwrap().push(v);
|
|
}
|
|
});
|
|
for v in 1..=3i64 {
|
|
tx.send(v).unwrap();
|
|
}
|
|
h.join().unwrap();
|
|
});
|
|
|
|
assert_eq!(*captured.lock().unwrap(), vec![1, 2, 3]);
|
|
}
|
|
|
|
#[test]
|
|
fn cloned_senders_both_deliver() {
|
|
let captured: std::sync::Arc<std::sync::Mutex<Vec<i64>>> =
|
|
std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
|
|
let cap2 = captured.clone();
|
|
|
|
run(move || {
|
|
let (tx, rx) = channel::<i64>();
|
|
let tx2 = tx.clone();
|
|
let h = spawn(move || {
|
|
for _ in 0..2 {
|
|
let v = rx.recv().unwrap();
|
|
cap2.lock().unwrap().push(v);
|
|
}
|
|
});
|
|
tx.send(10).unwrap();
|
|
tx2.send(20).unwrap();
|
|
h.join().unwrap();
|
|
});
|
|
|
|
let mut got = captured.lock().unwrap().clone();
|
|
got.sort();
|
|
assert_eq!(got, vec![10, 20]);
|
|
}
|
|
|
|
#[test]
|
|
fn recv_returns_err_when_all_senders_dropped() {
|
|
let saw_err: std::sync::Arc<std::sync::atomic::AtomicBool> =
|
|
std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
|
|
let saw_err2 = saw_err.clone();
|
|
|
|
run(move || {
|
|
let (tx, rx) = channel::<i64>();
|
|
let h = spawn(move || {
|
|
// Receiver waits; no message will ever come.
|
|
if rx.recv().is_err() {
|
|
saw_err2.store(true, std::sync::atomic::Ordering::SeqCst);
|
|
}
|
|
});
|
|
smarm::yield_now();
|
|
drop(tx); // last sender gone; rx.recv must return Err.
|
|
h.join().unwrap();
|
|
});
|
|
|
|
assert!(saw_err.load(std::sync::atomic::Ordering::SeqCst));
|
|
}
|
|
|
|
#[test]
|
|
fn channel_ops_interleaved_with_monitor_churn_multi_thread() {
|
|
// Regression for the RawMutex migration: monitor registration clones the
|
|
// Down sender under the target's cold (Leaf) lock, which now nests a
|
|
// Channel-class lock under it. Debug builds enforce the Leaf -> Channel
|
|
// ordering on every acquisition, so driving channels, monitors, and actor
|
|
// death concurrently across schedulers makes any ordering regression
|
|
// panic here rather than deadlock in the field.
|
|
use std::sync::atomic::{AtomicI64, Ordering};
|
|
use std::sync::Arc;
|
|
|
|
let total = Arc::new(AtomicI64::new(0));
|
|
let total2 = total.clone();
|
|
smarm::init(smarm::Config::exact(4)).run(move || {
|
|
let (tx, rx) = channel::<i64>();
|
|
let consumer = spawn(move || {
|
|
let mut sum = 0;
|
|
while let Ok(v) = rx.recv() {
|
|
sum += v;
|
|
}
|
|
OUT.with(|c| c.set(sum)); // not asserted cross-thread; see total
|
|
total2.fetch_add(sum, Ordering::Relaxed);
|
|
});
|
|
|
|
let mut handles = Vec::new();
|
|
for i in 0..32i64 {
|
|
let tx = tx.clone();
|
|
handles.push(spawn(move || {
|
|
// Short-lived target whose death fires the monitor below.
|
|
let t = spawn(move || {
|
|
tx.send(i).unwrap();
|
|
});
|
|
let m = smarm::monitor(t.pid());
|
|
t.join().unwrap();
|
|
// Down delivery exercises send-from-finalize.
|
|
let d = m.rx.recv().unwrap();
|
|
assert_eq!(d.reason, smarm::DownReason::Exit);
|
|
}));
|
|
}
|
|
drop(tx);
|
|
for h in handles {
|
|
h.join().unwrap();
|
|
}
|
|
consumer.join().unwrap();
|
|
});
|
|
assert_eq!(total.load(std::sync::atomic::Ordering::Relaxed), (0..32).sum::<i64>());
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// recv_timeout
|
|
// ---------------------------------------------------------------------------
|
|
|
|
use smarm::RecvTimeoutError;
|
|
use std::time::{Duration, Instant};
|
|
|
|
#[test]
|
|
fn recv_timeout_returns_queued_message_immediately() {
|
|
run(|| {
|
|
let (tx, rx) = channel::<i64>();
|
|
tx.send(5).unwrap();
|
|
assert_eq!(rx.recv_timeout(Duration::from_secs(10)), Ok(5));
|
|
});
|
|
}
|
|
|
|
#[test]
|
|
fn recv_timeout_times_out_on_silent_channel() {
|
|
run(|| {
|
|
let (_tx, rx) = channel::<i64>();
|
|
let start = Instant::now();
|
|
let r = rx.recv_timeout(Duration::from_millis(50));
|
|
assert_eq!(r, Err(RecvTimeoutError::Timeout));
|
|
assert!(start.elapsed() >= Duration::from_millis(50));
|
|
});
|
|
}
|
|
|
|
#[test]
|
|
fn recv_timeout_wakes_promptly_on_send() {
|
|
run(|| {
|
|
let (tx, rx) = channel::<i64>();
|
|
let h = spawn(move || {
|
|
let start = Instant::now();
|
|
assert_eq!(rx.recv_timeout(Duration::from_secs(10)), Ok(9));
|
|
// Far below the timeout: the send woke us, not the deadline.
|
|
assert!(start.elapsed() < Duration::from_secs(1));
|
|
});
|
|
smarm::yield_now();
|
|
tx.send(9).unwrap();
|
|
h.join().unwrap();
|
|
});
|
|
}
|
|
|
|
#[test]
|
|
fn recv_timeout_reports_disconnected_on_close() {
|
|
run(|| {
|
|
let (tx, rx) = channel::<i64>();
|
|
let h = spawn(move || {
|
|
assert_eq!(
|
|
rx.recv_timeout(Duration::from_secs(10)),
|
|
Err(RecvTimeoutError::Disconnected)
|
|
);
|
|
});
|
|
smarm::yield_now();
|
|
drop(tx);
|
|
h.join().unwrap();
|
|
});
|
|
}
|
|
|
|
#[test]
|
|
fn recv_timeout_zero_duration_is_a_bounded_poll() {
|
|
run(|| {
|
|
let (_tx, rx) = channel::<i64>();
|
|
assert_eq!(rx.recv_timeout(Duration::ZERO), Err(RecvTimeoutError::Timeout));
|
|
});
|
|
}
|
|
|
|
#[test]
|
|
fn channel_remains_usable_after_a_timeout() {
|
|
// The stale timer entry from the first (timed-out) wait must not cancel
|
|
// or corrupt later waits — seq isolation.
|
|
run(|| {
|
|
let (tx, rx) = channel::<i64>();
|
|
assert_eq!(
|
|
rx.recv_timeout(Duration::from_millis(10)),
|
|
Err(RecvTimeoutError::Timeout)
|
|
);
|
|
// Plain recv still works...
|
|
tx.send(1).unwrap();
|
|
assert_eq!(rx.recv(), Ok(1));
|
|
// ...and so does a second bounded wait, woken by a send.
|
|
let h = spawn(move || {
|
|
tx.send(2).unwrap();
|
|
});
|
|
assert_eq!(rx.recv_timeout(Duration::from_secs(10)), Ok(2));
|
|
h.join().unwrap();
|
|
});
|
|
}
|
|
|
|
#[test]
|
|
fn recv_timeout_many_waiters_multi_thread() {
|
|
// Mixed outcomes under real parallelism: half the channels get fed,
|
|
// half time out; every actor must resolve correctly.
|
|
use std::sync::atomic::{AtomicU32, Ordering};
|
|
use std::sync::Arc;
|
|
|
|
let got = Arc::new(AtomicU32::new(0));
|
|
let timed_out = Arc::new(AtomicU32::new(0));
|
|
let (got2, timed_out2) = (got.clone(), timed_out.clone());
|
|
smarm::init(smarm::Config::exact(4)).run(move || {
|
|
let mut handles = Vec::new();
|
|
for i in 0..24i64 {
|
|
let (tx, rx) = channel::<i64>();
|
|
let got = got2.clone();
|
|
let timed_out = timed_out2.clone();
|
|
handles.push(spawn(move || match rx.recv_timeout(Duration::from_millis(100)) {
|
|
Ok(v) => {
|
|
assert_eq!(v, i);
|
|
got.fetch_add(1, Ordering::Relaxed);
|
|
}
|
|
Err(RecvTimeoutError::Timeout) => {
|
|
timed_out.fetch_add(1, Ordering::Relaxed);
|
|
}
|
|
Err(e) => panic!("unexpected: {e}"),
|
|
}));
|
|
if i % 2 == 0 {
|
|
handles.push(spawn(move || {
|
|
tx.send(i).unwrap();
|
|
}));
|
|
}
|
|
// odd i: tx drops here -> Disconnected, not Timeout! Keep it alive
|
|
// instead by leaking the sender into a holder actor that outlives
|
|
// the deadline.
|
|
else {
|
|
handles.push(spawn(move || {
|
|
smarm::sleep(Duration::from_millis(200));
|
|
drop(tx);
|
|
}));
|
|
}
|
|
}
|
|
for h in handles {
|
|
h.join().unwrap();
|
|
}
|
|
});
|
|
assert_eq!(got.load(std::sync::atomic::Ordering::Relaxed), 12);
|
|
assert_eq!(timed_out.load(std::sync::atomic::Ordering::Relaxed), 12);
|
|
}
|