From a0a93b61bbd67b00a6dc0c7a1f1fbc388850df0f Mon Sep 17 00:00:00 2001 From: smarm Date: Wed, 10 Jun 2026 07:46:52 +0000 Subject: [PATCH] =?UTF-8?q?feat(channel):=20select=5Ftimeout=20=E2=80=94?= =?UTF-8?q?=20bounded=20select=20on=20a=20stateless=20stamped=20timer=20ar?= =?UTF-8?q?m?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- src/channel.rs | 92 +++++++++++++++++++++++++++++++++++++---------- src/lib.rs | 2 +- tests/select.rs | 95 +++++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 170 insertions(+), 19 deletions(-) diff --git a/src/channel.rs b/src/channel.rs index 80f8a20..491a5b7 100644 --- a/src/channel.rs +++ b/src/channel.rs @@ -435,24 +435,7 @@ pub fn select(arms: &[&dyn Selectable]) -> usize { let me = crate::actor::current_pid().expect("select() called outside an actor"); loop { let epoch = crate::scheduler::begin_wait(); - - // Check-or-register each arm, in priority order, each atomically - // under its own lock. Cross-arm atomicity is unnecessary: an arm - // becoming ready right after its registration wakes us through the - // protocol (the prep-to-park window is closed by RunningNotified). - let mut ready = None; - for (i, arm) in arms.iter().enumerate() { - if !arm.sel_register(me, epoch) { - ready = Some(i); - break; - } - } - if let Some(i) = ready { - // Returning WITHOUT parking while earlier arms hold live-epoch - // registrations: retire the wait (bump the epoch, eat a - // notification that already landed, re-observe a pending stop) - // so no stale arm wake can fault a later one-shot park. - crate::scheduler::retire_wait(); + if let Some(i) = register_arms(me, epoch, arms) { return i; } @@ -472,3 +455,76 @@ pub fn select(arms: &[&dyn Selectable]) -> usize { // stale own-registrations are overwritten. } } + +/// The registration pass shared by [`select`] and [`select_timeout`]: +/// check-or-register each arm, in priority order, each atomically under its +/// own lock. Cross-arm atomicity is unnecessary: an arm becoming ready +/// right after its registration wakes the caller through the protocol (the +/// prep-to-park window is closed by RunningNotified). +/// +/// `Some(i)` = arm `i` was ready, the pass stopped, and the wait has been +/// RETIRED (no park may follow): earlier arms hold live-epoch +/// registrations, so the epoch is bumped, a landed notification eaten, and +/// a pending stop re-observed — without which a stale arm wake could fault +/// a later one-shot park. `None` = every arm registered; the caller parks. +fn register_arms(me: Pid, epoch: u32, arms: &[&dyn Selectable]) -> Option { + for (i, arm) in arms.iter().enumerate() { + if !arm.sel_register(me, epoch) { + crate::scheduler::retire_wait(); + return Some(i); + } + } + None +} + +/// The [`select_timeout`] timer target: stateless, because precise wakes +/// make classification a pure function of channel state. The entry is +/// stamped with the select's epoch; if an arm already won, this unpark dies +/// at the word's epoch CAS (the no-cancellation convention in `timer.rs`). +struct SelectTimeout; +impl crate::timer::TimerTarget for SelectTimeout { + fn on_timeout(&self, pid: Pid, epoch: u32) { + crate::scheduler::unpark_at(pid, epoch); + } +} + +/// [`select`] with a deadline: returns `Some(index)` like `select`, or +/// `None` once `timeout` elapses with no arm ready. +/// +/// All of `select`'s semantics carry over (priority order, closed arms +/// permanently ready, no fairness promise). The timeout is one more stamped +/// waker on the same wait epoch — nothing is registered in any arm for it, +/// so there is nothing to cancel or leak: an arm winning leaves the timer +/// entry to expire as a stale-epoch no-op; the timer winning leaves the +/// arms' registrations to self-clean exactly as a `select` loser's would. +/// +/// The wake is classified from state alone (wakes are precise): some arm +/// ready → `Some` of the first, in priority order; none ready → the timer +/// was the only remaining stamped waker → `None`. A message that races the +/// deadline resolves message-first, as `recv_timeout` does. +/// +/// `Duration::ZERO` is a valid timeout: it parks until the immediately-due +/// timer is drained, then reports `None` unless an arm was already ready. +/// +/// Panics if `arms` is empty, or when called outside an actor. +pub fn select_timeout( + arms: &[&dyn Selectable], + timeout: std::time::Duration, +) -> Option { + assert!(!arms.is_empty(), "select_timeout() on an empty arm list"); + let me = crate::actor::current_pid().expect("select_timeout() called outside an actor"); + let epoch = crate::scheduler::begin_wait(); + if let Some(i) = register_arms(me, epoch, arms) { + return Some(i); // ready now: the timer was never armed + } + + // Arm the timer after the registration pass, outside every Channel + // lock (insert takes the timers lock). + let deadline = crate::timer::deadline_from_now(timeout); + let target: std::sync::Arc = std::sync::Arc::new(SelectTimeout); + crate::scheduler::insert_wait_timer(deadline, me, target, epoch); + + crate::scheduler::park_current(); + // Woken precisely: an arm (ready below) or the timer (nothing ready). + arms.iter().position(|arm| arm.sel_ready()) +} diff --git a/src/lib.rs b/src/lib.rs index f42b926..be57621 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, select, Receiver, RecvError, RecvTimeoutError, Selectable, Sender}; +pub use channel::{channel, select, select_timeout, Receiver, RecvError, RecvTimeoutError, Selectable, Sender}; pub use gen_server::{CallError, CallTimeoutError, CastError, GenServer, ServerRef}; pub use link::{link, trap_exit, unlink, ExitSignal}; pub use monitor::{demonitor, monitor, Down, DownReason, Monitor, MonitorId}; diff --git a/tests/select.rs b/tests/select.rs index d51f4ed..5066d38 100644 --- a/tests/select.rs +++ b/tests/select.rs @@ -243,3 +243,98 @@ fn motivating_pattern_inbox_plus_monitor_down() { worker.join().unwrap(); }); } + +// --------------------------------------------------------------------------- +// select_timeout +// --------------------------------------------------------------------------- + +use smarm::select_timeout; + +#[test] +fn select_timeout_returns_none_after_the_deadline() { + run(|| { + let (_keep_a, rxa) = channel::(); + let (_keep_b, rxb) = channel::(); + 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::(); + let (_keep_b, rxb) = channel::(); + 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::(); + let (_keep_b, rxb) = channel::(); + 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::(); + 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::(); + let (_keep_b, rxb) = channel::(); + 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::(); + let (txb, rxb) = channel::(); + drop(txb); + assert_eq!(select_timeout(&[&rxa, &rxb], Duration::from_millis(200)), Some(1)); + assert!(rxb.try_recv().is_err()); + }); +}