diff --git a/src/channel.rs b/src/channel.rs index b3a1f7e..c461a13 100644 --- a/src/channel.rs +++ b/src/channel.rs @@ -39,6 +39,9 @@ pub fn channel() -> (Sender, Receiver) { parked_receiver: None, senders: 1, receiver_alive: true, + cur_wait: None, + next_wait_seq: 0, + timed_out: false, })); (Sender { inner: inner.clone() }, Receiver { inner }) } @@ -48,6 +51,15 @@ struct Inner { parked_receiver: Option, senders: usize, receiver_alive: bool, + /// The seq of the receiver's current *bounded* wait (`recv_timeout`), or + /// `None` outside one. Lets `on_timeout` tell "this wait" apart from "a + /// later wait on the same channel" — same convention as `Mutex`. + cur_wait: Option, + /// Monotonic source for `cur_wait` seqs. + next_wait_seq: u64, + /// Set by `on_timeout` when it cancels the current bounded wait; consumed + /// by the woken receiver to distinguish a timeout from a message wake. + timed_out: bool, } pub struct Sender { @@ -72,6 +84,27 @@ impl std::fmt::Display for RecvError { impl std::error::Error for RecvError {} +/// Returned by [`Receiver::recv_timeout`]. +#[derive(Debug, PartialEq, Eq, Clone, Copy)] +pub enum RecvTimeoutError { + /// The deadline passed with no message available. + Timeout, + /// All senders dropped with no message available — the bounded analogue + /// of [`RecvError`]. + Disconnected, +} + +impl std::fmt::Display for RecvTimeoutError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + RecvTimeoutError::Timeout => write!(f, "recv timed out"), + RecvTimeoutError::Disconnected => write!(f, "channel closed"), + } + } +} + +impl std::error::Error for RecvTimeoutError {} + impl Clone for Sender { fn clone(&self) -> Self { self.inner.lock().senders += 1; @@ -155,6 +188,87 @@ impl Receiver { } } + /// Bounded receive: like [`recv`](Self::recv), but gives up once + /// `timeout` has elapsed, returning [`RecvTimeoutError::Timeout`]. + /// + /// Built on the same timer machinery as `Mutex::lock_timeout`: the wait + /// registers a `WaitTimeout` entry tagged with a per-wait seq; on expiry + /// the channel (as the [`TimerTarget`](crate::timer::TimerTarget)) checks + /// whether *this* wait is still parked and, only then, cancels it. A wake + /// that races the deadline resolves message-first: if a message is + /// available when the receiver runs, it is delivered even if the timer + /// had already fired. A satisfied or abandoned wait leaves its timer + /// entry to expire as a no-op (seq mismatch), per the no-cancellation + /// convention in `timer.rs`. + /// + /// `Duration::ZERO` is a valid timeout: it parks until the immediately- + /// due timer is drained, then reports `Timeout` unless a message was + /// already queued. + pub fn recv_timeout(&self, timeout: std::time::Duration) -> Result + where + T: Send + 'static, + { + let me = crate::actor::current_pid() + .expect("recv_timeout() called outside an actor"); + + // Fast path + wait registration, one critical section. + let seq; + { + let mut g = self.inner.lock(); + if let Some(v) = g.queue.pop_front() { + return Ok(v); + } + if g.senders == 0 { + return Err(RecvTimeoutError::Disconnected); + } + debug_assert!( + g.parked_receiver.is_none(), + "channel has more than one receiver" + ); + seq = g.next_wait_seq; + g.next_wait_seq = g.next_wait_seq.wrapping_add(1); + g.cur_wait = Some(seq); + g.timed_out = false; + g.parked_receiver = Some(me); + crate::te!(crate::trace::Event::RecvPark(me)); + } + + // Arm the timer after releasing the channel lock (insert takes the + // timers lock; never nest under a Channel lock). A send or even the + // timer itself may unpark us before we park — the RunningNotified + // protocol makes the park below return immediately in that case. + let deadline = crate::timer::deadline_from_now(timeout); + let target: std::sync::Arc = self.inner.clone(); + crate::scheduler::insert_wait_timer(deadline, me, target, seq); + + loop { + crate::scheduler::park_current(); + crate::te!(crate::trace::Event::RecvWake(crate::actor::current_pid().unwrap())); + let mut g = self.inner.lock(); + if let Some(v) = g.queue.pop_front() { + g.cur_wait = None; + g.timed_out = false; + return Ok(v); + } + if g.senders == 0 { + g.cur_wait = None; + g.timed_out = false; + return Err(RecvTimeoutError::Disconnected); + } + if g.timed_out { + g.cur_wait = None; + g.timed_out = false; + return Err(RecvTimeoutError::Timeout); + } + // Defensive: no current wake source reaches here (a send leaves a + // message, closure zeroes senders, the timer sets timed_out, and + // a stop unwinds out of park_current) — but if one ever does, + // re-register and keep waiting; cur_wait == Some(seq) keeps the + // already-armed timer valid for this wait. + g.parked_receiver = Some(me); + } + } + /// Selective receive: remove and return the first queued message for which /// `pred` holds, leaving the rest in arrival order. If no queued message /// matches, parks and re-scans on every send (a selective receiver may park @@ -226,3 +340,31 @@ impl Receiver { Ok(None) } } + +// --------------------------------------------------------------------------- +// TimerTarget — the expiry half of recv_timeout +// --------------------------------------------------------------------------- + +impl crate::timer::TimerTarget for RawMutex> { + fn on_timeout(&self, pid: Pid, wait_seq: u64) { + // Cancel the wait only if THIS wait (seq match) is still parked. If a + // sender already took `parked_receiver`, the receiver is waking with + // a message — message wins, the timer no-ops. If `cur_wait` moved on, + // this entry is stale (the wait was satisfied or abandoned) — no-op. + let unpark = { + let mut g = self.lock(); + if g.cur_wait == Some(wait_seq) && g.parked_receiver == Some(pid) { + g.parked_receiver = None; + g.timed_out = true; + true + } else { + false + } + }; + // Unpark outside the channel lock — it may take the run-queue lock; + // legal under a Channel lock, but pointless to nest. + if unpark { + crate::scheduler::unpark(pid); + } + } +} diff --git a/src/lib.rs b/src/lib.rs index e90c8ab..806986b 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -45,7 +45,7 @@ static ALLOCATOR: preempt::PreemptingAllocator = preempt::PreemptingAllocator; // Public API re-exports // --------------------------------------------------------------------------- -pub use channel::{channel, Receiver, RecvError, Sender}; +pub use channel::{channel, Receiver, RecvError, RecvTimeoutError, Sender}; pub use gen_server::{CallError, CastError, GenServer, ServerRef}; pub use link::{link, trap_exit, unlink, ExitSignal}; pub use monitor::{demonitor, monitor, Down, DownReason, Monitor, MonitorId}; diff --git a/tests/channel.rs b/tests/channel.rs index 41d9575..7fd2e64 100644 --- a/tests/channel.rs +++ b/tests/channel.rs @@ -156,3 +156,141 @@ fn channel_ops_interleaved_with_monitor_churn_multi_thread() { }); assert_eq!(total.load(std::sync::atomic::Ordering::Relaxed), (0..32).sum::()); } + +// --------------------------------------------------------------------------- +// recv_timeout +// --------------------------------------------------------------------------- + +use smarm::RecvTimeoutError; +use std::time::{Duration, Instant}; + +#[test] +fn recv_timeout_returns_queued_message_immediately() { + run(|| { + let (tx, rx) = channel::(); + 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::(); + 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::(); + 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::(); + 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::(); + 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::(); + 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::(); + 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); +}