From 171c83a491cd9194580d48f7cf9dcf945d2ae13c Mon Sep 17 00:00:00 2001 From: daylily Date: Fri, 5 Dec 2025 22:47:53 -0400 Subject: [PATCH] Implement `Unparker::unpark_with_result()` This method is like `Unpark::unpark()`, but also returns the previous state of the token. This change also introduces one new public enum `UnparkResult`. --- crossbeam-utils/src/sync/mod.rs | 2 +- crossbeam-utils/src/sync/parker.rs | 62 ++++++++++++++++++++++++++++-- crossbeam-utils/tests/parker.rs | 30 ++++++++++++++- 3 files changed, 87 insertions(+), 7 deletions(-) diff --git a/crossbeam-utils/src/sync/mod.rs b/crossbeam-utils/src/sync/mod.rs index ed5542117..108479b32 100644 --- a/crossbeam-utils/src/sync/mod.rs +++ b/crossbeam-utils/src/sync/mod.rs @@ -11,7 +11,7 @@ mod parker; mod sharded_lock; mod wait_group; -pub use self::parker::{Parker, UnparkReason, Unparker}; +pub use self::parker::{Parker, UnparkReason, UnparkResult, Unparker}; #[cfg(not(crossbeam_loom))] pub use self::sharded_lock::{ShardedLock, ShardedLockReadGuard, ShardedLockWriteGuard}; pub use self::wait_group::WaitGroup; diff --git a/crossbeam-utils/src/sync/parker.rs b/crossbeam-utils/src/sync/parker.rs index 82d48e151..91899bc1c 100644 --- a/crossbeam-utils/src/sync/parker.rs +++ b/crossbeam-utils/src/sync/parker.rs @@ -255,6 +255,44 @@ impl Unparker { /// [`park`]: Parker::park /// [`park_timeout`]: Parker::park_timeout pub fn unpark(&self) { + self.inner.unpark(); + } + + /// Atomically makes the token available if it is not already, returning an [`UnparkResult`] + /// indicating the result of the unpark operation. + /// + /// This method will wake up the thread blocked on [`park`] or [`park_timeout`], if there is + /// any. + /// + /// # Examples + /// + /// ``` + /// use std::thread; + /// use std::time::Duration; + /// use crossbeam_utils::sync::{Parker, UnparkResult}; + /// + /// let p = Parker::new(); + /// let u = p.unparker().clone(); + /// + /// let result = u.unpark_with_result(); + /// assert_eq!(result, UnparkResult::NotParked); + /// p.park(); // consume the token and immediately return + /// + /// # let t = + /// thread::spawn(move || { + /// thread::sleep(Duration::from_millis(500)); + /// let result = u.unpark_with_result(); + /// assert_eq!(result, UnparkResult::Notified); + /// }); + /// + /// // Wakes up when `u.unpark()` provides the token. + /// p.park(); + /// # t.join().unwrap(); // join thread to avoid https://github.com/rust-lang/miri/issues/1371 + /// ``` + /// + /// [`park`]: Parker::park + /// [`park_timeout`]: Parker::park_timeout + pub fn unpark_with_result(&self) -> UnparkResult { self.inner.unpark() } @@ -324,6 +362,21 @@ pub enum UnparkReason { Timeout, } +/// An enum that reports the result of an `Unparker::unpark` call. This includes whether the parker +/// was parked when `unpark` was called, and if so, whether it was already notified by a previous +/// `unpark` call. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum UnparkResult { + /// The parker was not parked when `unpark` was called. + NotParked, + + /// The parker was parked, but was already notified by an earlier `unpark` call. + AlreadyNotified, + + /// The parker was parked and has been successfully notified by this `unpark` call. + Notified, +} + const EMPTY: usize = 0; const PARKED: usize = 1; const NOTIFIED: usize = 2; @@ -407,15 +460,15 @@ impl Inner { } } - pub(crate) fn unpark(&self) { + pub(crate) fn unpark(&self) -> UnparkResult { // To ensure the unparked thread will observe any writes we made before this call, we must // perform a release operation that `park` can synchronize with. To do that we must write // `NOTIFIED` even if `state` is already `NOTIFIED`. That is why this must be a swap rather // than a compare-and-swap that returns if it reads `NOTIFIED` on failure. match self.state.swap(NOTIFIED, SeqCst) { - EMPTY => return, // no one was waiting - NOTIFIED => return, // already unparked - PARKED => {} // gotta go wake someone up + EMPTY => return UnparkResult::NotParked, // no one was waiting + NOTIFIED => return UnparkResult::AlreadyNotified, // already unparked + PARKED => {} // gotta go wake someone up _ => panic!("inconsistent state in unpark"), } @@ -429,5 +482,6 @@ impl Inner { // it doesn't get woken only to have to wait for us to release `lock`. drop(self.lock.lock().unwrap()); self.cvar.notify_one(); + UnparkResult::Notified } } diff --git a/crossbeam-utils/tests/parker.rs b/crossbeam-utils/tests/parker.rs index da8b133c3..00fad9717 100644 --- a/crossbeam-utils/tests/parker.rs +++ b/crossbeam-utils/tests/parker.rs @@ -1,7 +1,7 @@ -use std::thread::sleep; +use std::thread::{sleep, spawn}; use std::time::Duration; -use crossbeam_utils::sync::{Parker, UnparkReason}; +use crossbeam_utils::sync::{Parker, UnparkReason, UnparkResult}; use crossbeam_utils::thread; #[test] @@ -47,3 +47,29 @@ fn park_timeout_unpark_called_other_thread() { .unwrap(); } } + +#[test] +fn unpark_with_result_called_before_park() { + let p = Parker::new(); + let u = p.unparker().clone(); + + let result = u.unpark_with_result(); + assert_eq!(result, UnparkResult::NotParked); + + p.park(); // consume the token and immediately return +} + +#[test] +fn unpark_with_result_called_after_park() { + let p = Parker::new(); + let u = p.unparker().clone(); + + let t = spawn(move || { + sleep(Duration::from_millis(500)); + let result = u.unpark_with_result(); + assert_eq!(result, UnparkResult::Notified); + }); + + p.park(); // consume the token and immediately return + t.join().unwrap(); +}