diff --git a/CHANGELOG.md b/CHANGELOG.md index d6d6f5e6..9543c027 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,7 @@ All notable changes to this project will be documented in this file. * Remove the `asyncband::atomicbox` module and its `AtomicBox` and `AtomicOptionBox` types from the public API. * Remove the lossy `broadcast::overflow` channel; use the new lossless `broadcast::mpmc::unbounded` channel instead. * Remove `OnceMap::with_capacity` and `OnceMap::with_capacity_and_hasher`; use `OnceMap::new` or `OnceMap::with_hasher`, which allocate the backing table lazily. +* Remove the unconstructible `LatchWait` and `OwnedLatchWait` types from the public API; `Latch::wait` and `Latch::wait_owned` continue to return anonymous futures through their `async fn` signatures. * Rename `oneshot::Sender::is_closed` and `oneshot::Receiver::is_closed` to `is_disconnected`. * Remove `Semaphore::try_acquire_and_forget`, `Semaphore::acquire_and_forget`, `Semaphore::try_acquire_owned_and_forget`, and `Semaphore::acquire_owned_and_forget`; acquire a permit and call its `forget` method instead. * Replace `Semaphore::forget` with `Semaphore::drain_permits` and `Semaphore::forget_exact` with `Semaphore::reduce_permits`; permit-level `forget` methods are unchanged. diff --git a/asyncband/src/barrier/mod.rs b/asyncband/src/barrier/mod.rs index 69cef812..9f43625a 100644 --- a/asyncband/src/barrier/mod.rs +++ b/asyncband/src/barrier/mod.rs @@ -15,17 +15,11 @@ // specific language governing permissions and limitations // under the License. -//! A synchronization primitive that enables multiple tasks to wait for each other. +//! Synchronize a fixed number of tasks at a reusable rendezvous point. //! -//! The barrier ensures that no task proceeds past a certain point until all tasks have reached it. -//! This is useful for scenarios where multiple tasks need to proceed together after reaching a -//! certain point in their execution. -//! -//! A barrier enables multiple tasks to synchronize the beginning of some computation. -//! When a barrier is created, it is initialized with a count of the number of tasks -//! that will synchronize on the barrier. Each task can then call [`wait()`] on the -//! barrier to indicate it is ready to proceed. The barrier ensures that no task -//! proceeds past the barrier point until all tasks have made the call. +//! A [`Barrier`] releases one generation after the configured number of participants have awaited +//! [`Barrier::wait`]. It can then be reused for the next generation. Exactly one participant in +//! each generation receives a [`BarrierWaitResult`] marked as the leader. //! //! # Examples //! @@ -37,38 +31,21 @@ //! use asyncband::barrier::Barrier; //! //! let barrier = Arc::new(Barrier::new(3)); -//! let mut handles = Vec::new(); +//! let mut tasks = Vec::new(); //! -//! for i in 0..3 { +//! for _ in 0..3 { //! let barrier = barrier.clone(); -//! handles.push(tokio::spawn(async move { -//! println!("Task {} before barrier", i); -//! let result = barrier.wait().await; -//! println!("Task {} after barrier (leader: {})", i, result.is_leader()); -//! })); -//! } -//! -//! for handle in std::mem::take(&mut handles) { -//! handle.await.unwrap(); +//! let task = tokio::spawn(async move { barrier.wait().await.is_leader() }); +//! tasks.push(task); //! } //! -//! // The barrier can be reused (generation is increased). -//! for i in 0..3 { -//! let barrier = barrier.clone(); -//! handles.push(tokio::spawn(async move { -//! println!("Task {} before barrier", i); -//! let result = barrier.wait().await; -//! println!("Task {} after barrier (leader: {})", i, result.is_leader()); -//! })); -//! } -//! -//! for handle in handles { -//! handle.await.unwrap(); +//! let mut leaders = 0; +//! for task in tasks { +//! leaders += usize::from(task.await.unwrap()); //! } +//! assert_eq!(leaders, 1); //! # } //! ``` -//! -//! [`wait()`]: Barrier::wait use std::fmt; use std::future::Future; @@ -105,20 +82,9 @@ impl fmt::Debug for BarrierState { } } -/// A `BarrierWaitResult` is returned by [`Barrier::wait()`] when all threads -/// in the [`Barrier`] have rendezvoused. +/// The result of participating in one [`Barrier`] generation. /// -/// # Examples -/// -/// ``` -/// # #[tokio::main] -/// # async fn main() { -/// use asyncband::barrier::Barrier; -/// -/// let barrier = Barrier::new(1); -/// let barrier_wait_result = barrier.wait().await; -/// # } -/// ``` +/// Exactly one participant in each completed generation is designated as the leader. pub struct BarrierWaitResult(bool); impl fmt::Debug for BarrierWaitResult { @@ -130,10 +96,9 @@ impl fmt::Debug for BarrierWaitResult { } impl BarrierWaitResult { - /// Returns `true` if this worker is the "leader" for the call to [`Barrier::wait()`]. + /// Returns `true` if this participant is the leader for its barrier generation. /// - /// Only one worker will have `true` returned from their result, all other - /// workers will have `false` returned. + /// Exactly one participant per generation returns `true`. /// /// # Examples /// @@ -143,8 +108,7 @@ impl BarrierWaitResult { /// use asyncband::barrier::Barrier; /// /// let barrier = Barrier::new(1); - /// let barrier_wait_result = barrier.wait().await; - /// println!("{:?}", barrier_wait_result.is_leader()); + /// assert!(barrier.wait().await.is_leader()); /// # } /// ``` #[must_use] @@ -188,9 +152,8 @@ impl Barrier { /// Waits for all tasks to reach this point. /// - /// The barrier will block the current task until all `n` tasks have called `wait()`. - /// The last task to call `wait()` will be designated as the leader and receive `true` - /// as the return value. All other tasks will receive `false`. + /// The barrier holds the current task until all `n` participants have arrived. The final + /// participant is designated as the leader for this generation. /// /// # Cancel safety /// @@ -206,32 +169,17 @@ impl Barrier { /// wait futures. Callers that need to keep waiting after another operation completes first /// should retain and continue polling the same `wait` future instead of creating a new one. /// - /// # Returns - /// - /// Returns a `Future` that resolves to: - /// * `true` if this task is the last (leader) task to arrive at the barrier - /// * `false` for all other tasks + /// Returns a [`BarrierWaitResult`] that identifies whether this participant is the leader. /// /// # Examples /// /// ``` /// # #[tokio::main] /// # async fn main() { - /// use std::sync::Arc; - /// /// use asyncband::barrier::Barrier; /// - /// let barrier = Arc::new(Barrier::new(2)); - /// let barrier2 = barrier.clone(); - /// - /// let handle = tokio::spawn(async move { - /// let result = barrier2.wait().await; - /// println!("Task 1: leader = {}", result.is_leader()); - /// }); - /// - /// let result = barrier.wait().await; - /// println!("Task 2: leader = {}", result.is_leader()); - /// handle.await.unwrap(); + /// let barrier = Barrier::new(1); + /// assert!(barrier.wait().await.is_leader()); /// # } /// ``` pub async fn wait(&self) -> BarrierWaitResult { diff --git a/asyncband/src/condvar/mod.rs b/asyncband/src/condvar/mod.rs index f1a72053..ef3fb2ef 100644 --- a/asyncband/src/condvar/mod.rs +++ b/asyncband/src/condvar/mod.rs @@ -40,23 +40,20 @@ //! use asyncband::mutex::Mutex; //! //! let pair = Arc::new((Mutex::new(false), Condvar::new())); -//! let pair_clone = pair.clone(); +//! let notifier_pair = pair.clone(); //! -//! // Inside our lock, spawn a new thread, and then wait for it to start. -//! tokio::spawn(async move { -//! let (lock, cvar) = &*pair_clone; -//! let mut started = lock.lock().await; -//! *started = true; -//! // We notify the condvar that the value has changed. +//! let notifier = tokio::spawn(async move { +//! let (lock, cvar) = &*notifier_pair; +//! let mut ready = lock.lock().await; +//! *ready = true; //! cvar.notify_one(); //! }); //! -//! // Wait for the thread to start up. //! let (lock, cvar) = &*pair; -//! let mut started = lock.lock().await; -//! while !*started { -//! started = cvar.wait(started).await; -//! } +//! let ready = cvar.wait_while(lock.lock().await, |ready| !*ready).await; +//! assert!(*ready); +//! drop(ready); +//! notifier.await.unwrap(); //! # } //! ``` @@ -247,21 +244,20 @@ impl Condvar { /// let pair = Arc::new((Mutex::new(false), Condvar::new())); /// let pair_clone = pair.clone(); /// - /// tokio::spawn(async move { + /// let notifier = tokio::spawn(async move { /// let (lock, cvar) = &*pair_clone; /// let mut started = lock.lock().await; /// *started = true; - /// // We notify the condvar that the value has changed. /// cvar.notify_one(); /// }); /// - /// // Wait for the thread to start up. /// let (lock, cvar) = &*pair; - /// // As long as the value inside the `Mutex` is `false`, we wait. /// let guard = cvar /// .wait_while(lock.lock().await, |started| !*started) /// .await; /// assert!(*guard); + /// drop(guard); + /// notifier.await.unwrap(); /// # } /// ``` pub async fn wait_while<'a, T, F>( @@ -293,21 +289,20 @@ impl Condvar { /// let pair = (Arc::new(Mutex::new(false)), Arc::new(Condvar::new())); /// let pair_clone = pair.clone(); /// - /// tokio::spawn(async move { + /// let notifier = tokio::spawn(async move { /// let (lock, cvar) = pair_clone; /// let mut started = lock.lock_owned().await; /// *started = true; - /// // We notify the condvar that the value has changed. /// cvar.notify_one(); /// }); /// - /// // Wait for the thread to start up. /// let (lock, cvar) = pair; - /// // As long as the value inside the `Mutex` is `false`, we wait. /// let guard = cvar /// .wait_while_owned(lock.lock_owned().await, |started| !*started) /// .await; /// assert!(*guard); + /// drop(guard); + /// notifier.await.unwrap(); /// # } /// ``` pub async fn wait_while_owned( diff --git a/asyncband/src/latch/mod.rs b/asyncband/src/latch/mod.rs index cfb4e1fb..048c2617 100644 --- a/asyncband/src/latch/mod.rs +++ b/asyncband/src/latch/mod.rs @@ -15,15 +15,12 @@ // specific language governing permissions and limitations // under the License. -//! A countdown latch that allows one or more tasks to wait until a set of operations completes. +//! Wait for a one-way countdown to reach zero. //! -//! Unlike a barrier, a latch's count can only decrease and cannot be reused once it reaches zero. -//! This makes it ideal for scenarios where you need to wait for a specific number of events or -//! operations to complete. -//! -//! A latch starts with an initial count and tasks can wait for this count to reach zero. -//! The count can be decremented by calling [`count_down()`] or [`arrive()`]. Once the count -//! reaches zero, all waiting tasks are unblocked. +//! A [`Latch`] starts with a fixed count. [`Latch::count_down`] decrements it by one and +//! [`Latch::arrive`] decrements it by an arbitrary amount. Both operations saturate at zero. Once +//! zero is reached, all current and future waits complete immediately; a latch cannot be reset or +//! reused for another countdown. //! //! # Examples //! @@ -35,27 +32,22 @@ //! use asyncband::latch::Latch; //! //! let latch = Arc::new(Latch::new(3)); -//! let mut handles = Vec::new(); +//! let mut tasks = Vec::new(); //! -//! for i in 0..3 { +//! for _ in 0..3 { //! let latch = latch.clone(); -//! handles.push(tokio::spawn(async move { -//! println!("Task {} starting", i); -//! // Simulate some work -//! latch.count_down(); // Signal completion -//! })); +//! let task = tokio::spawn(async move { latch.count_down() }); +//! tasks.push(task); //! } //! -//! // Wait for all tasks to complete //! latch.wait().await; -//! println!("All tasks completed"); +//! for task in tasks { +//! task.await.unwrap(); +//! } +//! assert_eq!(latch.count(), 0); //! # } //! ``` -//! -//! [`count_down()`]: Latch::count_down -//! [`arrive()`]: Latch::arrive -use std::fmt; use std::future::Future; use std::pin::Pin; use std::sync::Arc; @@ -65,7 +57,7 @@ use std::task::Poll; use crate::internal::countdown::CountdownState; use crate::internal::waitset::WakerToken; -/// A synchronization primitive that can be used to coordinate multiple tasks. +/// A one-shot countdown that can wake any number of waiting tasks. /// /// See the [module level documentation](self) for more. #[derive(Debug)] @@ -76,16 +68,12 @@ pub struct Latch { impl Latch { /// Creates a new latch initialized with the given count. /// - /// # Arguments - /// - /// * `count` - The initial count value. Tasks will wait until this count reaches zero. - /// /// # Examples /// /// ``` /// use asyncband::latch::Latch; /// - /// let latch = Latch::new(3); // Creates a latch with count of 3 + /// let latch = Latch::new(3); /// ``` pub fn new(count: u32) -> Self { Self { @@ -95,8 +83,6 @@ impl Latch { /// Returns the current count. /// - /// This method is typically used for debugging and testing purposes. - /// /// # Examples /// /// ``` @@ -119,8 +105,8 @@ impl Latch { /// use asyncband::latch::Latch; /// /// let latch = Latch::new(2); - /// latch.count_down(); // Count is now 1 - /// latch.count_down(); // Count is now 0, all waiting tasks are woken + /// latch.count_down(); + /// assert_eq!(latch.count(), 1); /// ``` pub fn count_down(&self) { if self.state.decrement(1) { @@ -130,19 +116,8 @@ impl Latch { /// Decrements the latch count by `n`, waking up all waiting tasks if the counter reaches zero. /// - /// This method provides a way to decrement the counter by more than one at a time. - /// It will not cause an overflow when decrementing the counter. - /// - /// # Arguments - /// - /// * `n` - The amount to decrement the counter by - /// - /// # Behavior - /// - /// * If `n` is zero or the counter has already reached zero, nothing happens - /// * If the current count is greater than `n`, it is decremented by `n` - /// * If the current count is greater than 0 but less than or equal to `n`, the count becomes - /// zero and all waiting tasks are woken + /// The count saturates at zero. Passing zero or calling this after the latch has completed has + /// no effect. /// /// # Examples /// @@ -150,8 +125,10 @@ impl Latch { /// use asyncband::latch::Latch; /// /// let latch = Latch::new(5); - /// latch.arrive(3); // Count is now 2 - /// latch.arrive(2); // Count is now 0, all waiting tasks are woken + /// latch.arrive(3); + /// assert_eq!(latch.count(), 2); + /// latch.arrive(10); + /// assert_eq!(latch.count(), 0); /// ``` pub fn arrive(&self, n: u32) { if n != 0 && self.state.decrement(n) { @@ -161,10 +138,7 @@ impl Latch { /// Attempts to wait for the latch count to reach zero without blocking. /// - /// # Returns - /// - /// * `Ok(())` if the count is zero - /// * `Err(count)` if the count is not zero, where `count` is the current count + /// Returns `Ok(())` if the latch is complete, or `Err(count)` with the current nonzero count. /// /// # Examples /// @@ -184,27 +158,19 @@ impl Latch { /// Returns a future that will complete when the latch count reaches zero. /// + /// This method is cancel safe. Dropping a pending wait does not change the countdown or affect + /// other waiters. + /// /// # Examples /// /// ``` /// # #[tokio::main] /// # async fn main() { - /// use std::sync::Arc; - /// /// use asyncband::latch::Latch; /// - /// let latch = Arc::new(Latch::new(1)); - /// let latch2 = latch.clone(); - /// - /// // Spawn a task that will wait for the latch - /// let handle = tokio::spawn(async move { - /// latch2.wait().await; - /// println!("Latch reached zero!"); - /// }); - /// - /// // Count down the latch + /// let latch = Latch::new(1); /// latch.count_down(); - /// handle.await.unwrap(); + /// latch.wait().await; /// # } /// ``` pub async fn wait(&self) { @@ -217,8 +183,8 @@ impl Latch { /// Returns a future that will complete when the latch count reaches zero. /// - /// The latch must be wrapped in an [`Arc`] to call this method. Thus, the returned future has - /// no lifetime constraints. + /// The latch must be wrapped in an [`Arc`] to call this method. The future owns that `Arc`, so + /// it can be moved into a spawned task. Like [`wait`](Self::wait), this method is cancel safe. /// /// # Examples /// @@ -230,17 +196,9 @@ impl Latch { /// use asyncband::latch::Latch; /// /// let latch = Arc::new(Latch::new(1)); - /// let latch2 = latch.clone(); - /// - /// // Spawn a task that will wait for the latch - /// let handle = tokio::spawn(async move { - /// latch2.wait_owned().await; - /// println!("Latch reached zero!"); - /// }); - /// - /// // Count down the latch + /// let waiter = tokio::spawn(latch.clone().wait_owned()); /// latch.count_down(); - /// handle.await.unwrap(); + /// waiter.await.unwrap(); /// # } /// ``` pub async fn wait_owned(self: Arc) { @@ -258,21 +216,12 @@ impl Latch { } } -/// A wait future returned by [`Latch::wait()`]. -/// -/// This future will complete when the latch count reaches zero. #[must_use = "futures do nothing unless you `.await` or poll them"] -pub struct LatchWait<'a> { +struct LatchWait<'a> { token: Option, latch: &'a Latch, } -impl fmt::Debug for LatchWait<'_> { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - f.debug_struct("LatchWait").finish_non_exhaustive() - } -} - impl Future for LatchWait<'_> { type Output = (); @@ -288,21 +237,12 @@ impl Drop for LatchWait<'_> { } } -/// An owned wait future returned by [`Latch::wait()`]. -/// -/// This future will complete when the latch count reaches zero. #[must_use = "futures do nothing unless you `.await` or poll them"] -pub struct OwnedLatchWait { +struct OwnedLatchWait { token: Option, latch: Arc, } -impl fmt::Debug for OwnedLatchWait { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - f.debug_struct("OwnedLatchWait").finish_non_exhaustive() - } -} - impl Future for OwnedLatchWait { type Output = (); diff --git a/asyncband/src/lib.rs b/asyncband/src/lib.rs index 349f3cc7..6beb257e 100644 --- a/asyncband/src/lib.rs +++ b/asyncband/src/lib.rs @@ -36,6 +36,7 @@ //! Then use the selected APIs directly: //! //! ``` +//! # #[cfg(feature = "mutex")] //! # #[tokio::main] //! # async fn main() { //! use asyncband::mutex::Mutex; @@ -47,6 +48,8 @@ //! } //! assert_eq!(*counter.lock().await, 1); //! # } +//! # #[cfg(not(feature = "mutex"))] +//! # fn main() {} //! ``` //! //! # API map diff --git a/asyncband/src/mpsc/bounded.rs b/asyncband/src/mpsc/bounded.rs index c4d4bcf2..4f888f39 100644 --- a/asyncband/src/mpsc/bounded.rs +++ b/asyncband/src/mpsc/bounded.rs @@ -42,6 +42,10 @@ use crate::internal::semaphore::Semaphore; /// A `send` on this channel will wait if the buffer of the channel is full until a /// `recv` is called on the receiver, which will consume the message and /// free up space in the buffer. +/// +/// # Panics +/// +/// Panics if `buffer` is zero. #[track_caller] pub fn bounded(buffer: usize) -> (BoundedSender, BoundedReceiver) { assert!(buffer > 0, "mpsc bounded channel requires buffer > 0"); @@ -110,7 +114,7 @@ impl Drop for BoundedSender { } impl BoundedSender { - /// Attempts to send a message to the associated receiver. + /// Sends a message to the associated receiver. /// /// This method will wait if the buffer of the channel is full until a `recv` is called on the /// receiver, which will consume the message and free up space in the buffer. @@ -193,7 +197,7 @@ impl BoundedSender { /// # } /// ``` pub fn try_send(&self, value: T) -> Result<(), TrySendError> { - // SAFETY: The sender is guaranteed to be non-null before dropped. + // INVARIANT: A shared borrow of the endpoint cannot overlap its destructor. let sender = self.sender.as_ref().unwrap(); match sender.try_send(value) { Ok(()) => { @@ -270,7 +274,7 @@ impl BoundedReceiver { /// # } /// ``` pub fn try_recv(&mut self) -> Result { - // SAFETY: The receiver is guaranteed to be non-null before dropped. + // INVARIANT: A mutable borrow of the endpoint cannot overlap its destructor. let receiver = self.receiver.as_ref().unwrap(); match receiver.try_recv() { Ok(v) => { @@ -288,7 +292,7 @@ impl BoundedReceiver { /// no buffered messages remain. At that point, this `Receiver` can never receive another /// value. /// - /// If the buffer is empty while a sender remains, this method sleeps until a message is sent or + /// If the buffer is empty while a sender remains, this method waits until a message is sent or /// the final sender is dropped. /// /// # Cancel safety diff --git a/asyncband/src/mpsc/mod.rs b/asyncband/src/mpsc/mod.rs index 45bc7406..040c98b2 100644 --- a/asyncband/src/mpsc/mod.rs +++ b/asyncband/src/mpsc/mod.rs @@ -15,7 +15,11 @@ // specific language governing permissions and limitations // under the License. -//! A multi-producer, single-consumer queue for sending values between asynchronous tasks. +//! Multi-producer, single-consumer channels for asynchronous tasks. +//! +//! [`bounded`] applies backpressure once its fixed-capacity buffer fills. [`unbounded`] never waits +//! to send while the receiver is alive, but a slow receiver can cause memory use to grow without a +//! configured limit. mod bounded; mod error; diff --git a/asyncband/src/mpsc/unbounded.rs b/asyncband/src/mpsc/unbounded.rs index 57552d2a..ed6eb565 100644 --- a/asyncband/src/mpsc/unbounded.rs +++ b/asyncband/src/mpsc/unbounded.rs @@ -104,7 +104,7 @@ impl Drop for UnboundedSender { } impl UnboundedSender { - /// Attempts to send a message without blocking. + /// Sends a message without blocking. /// /// This method is not marked async because sending a message to an unbounded channel /// never requires any form of waiting. Because of this, the `send` method can be @@ -113,7 +113,7 @@ impl UnboundedSender { /// If the receiver has been dropped, this function returns an error. The error includes /// the value passed to `send`. pub fn send(&self, value: T) -> Result<(), SendError> { - // SAFETY: The sender is guaranteed to be non-null before dropped. + // INVARIANT: A shared borrow of the endpoint cannot overlap its destructor. let sender = self.sender.as_ref().unwrap(); sender.send(value).map_err(|err| SendError::new(err.0))?; @@ -189,7 +189,7 @@ impl UnboundedReceiver { /// no buffered messages remain. At that point, this `Receiver` can never receive another /// value. /// - /// If the buffer is empty while a sender remains, this method sleeps until a message is sent or + /// If the buffer is empty while a sender remains, this method waits until a message is sent or /// the final sender is dropped. /// /// # Cancel safety diff --git a/asyncband/src/once/once_cell/mod.rs b/asyncband/src/once/once_cell/mod.rs index ddb23232..7c4e47ff 100644 --- a/asyncband/src/once/once_cell/mod.rs +++ b/asyncband/src/once/once_cell/mod.rs @@ -22,7 +22,7 @@ use crate::internal::value_cell::ValueCell; use crate::semaphore::Semaphore; use crate::semaphore::SemaphorePermit; -/// A thread-safe cell which can nominally be written to only once. +/// A thread-safe cell whose value is asynchronously initialized at most once. /// /// Callers provide an initializer when accessing an empty cell. An initializer that returns an /// error, panics, or is cancelled leaves the cell empty so a later caller can retry. Use @@ -44,12 +44,10 @@ use crate::semaphore::SemaphorePermit; /// let handle2 = tokio::spawn(async { CELL.get_or_init(move || async { 2 }).await }); /// let result1 = handle1.await.unwrap(); /// let result2 = handle2.await.unwrap(); -/// println!("Results: {}, {}", result1, result2); +/// assert_eq!(result1, result2); +/// assert!(*result1 == 1 || *result1 == 2); /// # } /// ``` -/// -/// The outputs must be either `Results: 1, 1` or `Results: 2, 2`, i.e. once the value is set via -/// an asynchronous function, the value inside the `OnceCell` will be immutable. pub struct OnceCell { value: ValueCell, semaphore: Semaphore, @@ -253,7 +251,7 @@ impl OnceCell { /// Initializes the contents of the cell to `value` if the cell was uninitialized, /// then returns a reference to it. /// - /// May wait if another thread is currently attempting to initialize the cell. The cell is + /// May wait if another task is currently attempting to initialize the cell. The cell is /// guaranteed to contain a value when `try_insert` returns, though not necessarily the /// one provided. /// diff --git a/asyncband/src/pool/bounded.rs b/asyncband/src/pool/bounded.rs index d3584e81..0d4dbece 100644 --- a/asyncband/src/pool/bounded.rs +++ b/asyncband/src/pool/bounded.rs @@ -91,7 +91,7 @@ use crate::semaphore::Semaphore; #[derive(Clone, Copy, Debug)] #[non_exhaustive] pub struct PoolConfig { - /// Maximum size of the [`Pool`]. + /// Maximum size of the [`Pool`]. Must be greater than zero. pub max_size: usize, /// Queue strategy of the [`Pool`]. @@ -104,7 +104,9 @@ pub struct PoolConfig { } impl PoolConfig { - /// Creates a new [`PoolConfig`]. + /// Creates a new [`PoolConfig`] for a pool with the given maximum size. + /// + /// [`Pool::new`] panics if `max_size` is zero. pub fn new(max_size: usize) -> Self { Self { max_size, @@ -174,7 +176,16 @@ where impl Pool { /// Creates a new [`Pool`]. + /// + /// # Panics + /// + /// Panics if `config.max_size` is zero. pub fn new(config: PoolConfig, manager: M) -> Arc { + assert!( + config.max_size > 0, + "bounded pool max_size must be greater than zero" + ); + let permits = Arc::new(Semaphore::new(config.max_size)); let slots = Mutex::new(PoolState::new()); @@ -471,14 +482,14 @@ impl Drop for Object { impl Deref for Object { type Target = M::Object; fn deref(&self) -> &M::Object { - // SAFETY: `state` is always `Some` when `Object` is owned. + // INVARIANT: `state` is `Some` until this object is detached or dropped. &self.state.as_ref().unwrap().o } } impl DerefMut for Object { fn deref_mut(&mut self) -> &mut Self::Target { - // SAFETY: `state` is always `Some` when `Object` is owned. + // INVARIANT: `state` is `Some` until this object is detached or dropped. &mut self.state.as_mut().unwrap().o } } @@ -500,7 +511,7 @@ impl Object { /// /// This reduces the size of the pool by one. pub fn detach(mut self) -> M::Object { - // SAFETY: `state` is always `Some` when `Object` is owned. + // INVARIANT: `state` is `Some` until this object is detached or dropped. let mut o = self.state.take().unwrap().o; if let Some(pool) = self.pool.upgrade() { pool.detach_object(&mut o); @@ -510,7 +521,7 @@ impl Object { /// Returns the status of the object. pub fn status(&self) -> ObjectStatus { - // SAFETY: `state` is always `Some` when `Object` is owned. + // INVARIANT: `state` is `Some` until this object is detached or dropped. self.state.as_ref().unwrap().status } } @@ -546,7 +557,7 @@ impl Drop for UnreadyObject { impl UnreadyObject { fn ready(mut self, permit: OwnedSemaphorePermit) -> Object { - // SAFETY: `state` is always `Some` when `UnreadyObject` is owned. + // INVARIANT: `state` is `Some` until this object becomes ready, detaches, or is dropped. let state = Some(self.state.take().unwrap()); let pool = self.pool.clone(); Object { @@ -565,7 +576,7 @@ impl UnreadyObject { } fn state(&mut self) -> &mut ObjectState { - // SAFETY: `state` is always `Some` when `UnreadyObject` is owned. + // INVARIANT: `state` is `Some` until this object becomes ready, detaches, or is dropped. self.state.as_mut().unwrap() } } diff --git a/asyncband/src/pool/unbounded.rs b/asyncband/src/pool/unbounded.rs index 84832244..24358b7b 100644 --- a/asyncband/src/pool/unbounded.rs +++ b/asyncband/src/pool/unbounded.rs @@ -499,14 +499,14 @@ impl> Drop for Object { impl> Deref for Object { type Target = T; fn deref(&self) -> &T { - // SAFETY: `state` is always `Some` when `Object` is owned. + // INVARIANT: `state` is `Some` until this object is detached or dropped. &self.state.as_ref().unwrap().o } } impl> DerefMut for Object { fn deref_mut(&mut self) -> &mut Self::Target { - // SAFETY: `state` is always `Some` when `Object` is owned. + // INVARIANT: `state` is `Some` until this object is detached or dropped. &mut self.state.as_mut().unwrap().o } } @@ -528,7 +528,7 @@ impl> Object { /// /// This reduces the size of the pool by one. pub fn detach(mut self) -> M::Object { - // SAFETY: `state` is always `Some` when `Object` is owned. + // INVARIANT: `state` is `Some` until this object is detached or dropped. let mut o = self.state.take().unwrap().o; if let Some(pool) = self.pool.upgrade() { pool.detach_object(&mut o); @@ -538,7 +538,7 @@ impl> Object { /// Returns the status of the object. pub fn status(&self) -> ObjectStatus { - // SAFETY: `state` is always `Some` when `Object` is owned. + // INVARIANT: `state` is `Some` until this object is detached or dropped. self.state.as_ref().unwrap().status } } @@ -574,7 +574,7 @@ impl> Drop for UnreadyObject { impl> UnreadyObject { fn ready(mut self) -> Object { - // SAFETY: `state` is always `Some` when `UnreadyObject` is owned. + // INVARIANT: `state` is `Some` until this object becomes ready, detaches, or is dropped. let state = Some(self.state.take().unwrap()); let pool = self.pool.clone(); Object { state, pool } @@ -589,7 +589,7 @@ impl> UnreadyObject { } fn state(&mut self) -> &mut ObjectState { - // SAFETY: `state` is always `Some` when `UnreadyObject` is owned. + // INVARIANT: `state` is `Some` until this object becomes ready, detaches, or is dropped. self.state.as_mut().unwrap() } } diff --git a/asyncband/src/semaphore/mod.rs b/asyncband/src/semaphore/mod.rs index c812f516..764561b4 100644 --- a/asyncband/src/semaphore/mod.rs +++ b/asyncband/src/semaphore/mod.rs @@ -15,21 +15,14 @@ // specific language governing permissions and limitations // under the License. -//! An async counting semaphore for controlling access to a set of resources. +//! Limit concurrent access with a set of permits. //! -//! A semaphore maintains a set of permits. Permits are used to synchronize access -//! to a pool of resources. Each [`acquire`] call blocks until a permit is available, -//! and then takes one permit. Each [`release`] call adds a new permit, potentially -//! releasing a blocked acquirer. -//! -//! Semaphores are often used to restrict the number of tasks that can access some -//! (physical or logical) resource. For example, here is a class that uses a -//! semaphore to control access to a pool of connections: +//! [`Semaphore::acquire`] waits for the requested number of permits and returns a guard that puts +//! them back when dropped. [`Semaphore::try_acquire`] performs the same operation without waiting, +//! while [`Semaphore::release`] adds permits that were not represented by a guard. //! //! # Examples //! -//! ## Basic usage -//! //! ``` //! # #[tokio::main] //! # async fn main() { @@ -43,43 +36,11 @@ //! //! let permit_attempt = semaphore.try_acquire(1); //! assert!(permit_attempt.is_none()); -//! # } -//! ``` -//! -//! ## Limit the number of simultaneously opened files in your program -//! -//! Most operating systems have limits on the number of open file -//! handles. Even in systems without explicit limits, resource constraints -//! implicitly set an upper bound on the number of open files. If your -//! program attempts to open a large number of files and exceeds this -//! limit, it will result in an error. -//! -//! This example uses a Semaphore with 100 permits. By acquiring a permit from -//! the Semaphore before accessing a file, you ensure that your program opens -//! no more than 100 files at a time. When trying to open the 101st -//! file, the program will wait until a permit becomes available before -//! proceeding to open another file. -//! -//! ``` -//! use std::fs::File; -//! use std::io::Result; -//! use std::io::Write; -//! use std::sync::LazyLock; -//! -//! use asyncband::semaphore::Semaphore; //! -//! static PERMITS: LazyLock = LazyLock::new(|| Semaphore::new(100)); -//! -//! async fn write_to_file(message: &[u8]) -> Result<()> { -//! let _permit = PERMITS.acquire(1).await; -//! let mut buffer = File::create("example.txt")?; -//! buffer.write_all(message)?; -//! Ok(()) // Permit goes out of scope here, and is available again for acquisition -//! } +//! drop(a_permit); +//! assert_eq!(semaphore.available_permits(), 1); +//! # } //! ``` -//! -//! [`acquire`]: Semaphore::acquire -//! [`release`]: Semaphore::release use std::sync::Arc; @@ -536,7 +497,7 @@ impl OwnedSemaphorePermit { self.permits = 0; } - /// Merge two [`SemaphorePermit`] instances together, consuming `other` + /// Merge two [`OwnedSemaphorePermit`] instances together, consuming `other` /// without releasing the permits it holds. /// /// Permits held by both `self` and `other` are released when `self` drops. @@ -554,10 +515,10 @@ impl OwnedSemaphorePermit { /// use asyncband::semaphore::Semaphore; /// /// let sem = Arc::new(Semaphore::new(10)); - /// let mut permit = sem.try_acquire(1).unwrap(); + /// let mut permit = sem.clone().try_acquire_owned(1).unwrap(); /// /// for _ in 0..9 { - /// let new_permit = sem.try_acquire(1).unwrap(); + /// let new_permit = sem.clone().try_acquire_owned(1).unwrap(); /// // Merge individual permits into a single one. /// permit.merge(new_permit) /// } @@ -621,10 +582,12 @@ impl OwnedSemaphorePermit { /// # Examples /// /// ``` + /// use std::sync::Arc; + /// /// use asyncband::semaphore::Semaphore; /// - /// let sem = Semaphore::new(5); - /// let permit = sem.try_acquire(3).unwrap(); + /// let sem = Arc::new(Semaphore::new(5)); + /// let permit = sem.try_acquire_owned(3).unwrap(); /// assert_eq!(permit.permits(), 3); /// ``` pub fn permits(&self) -> usize { diff --git a/asyncband/src/shutdown/mod.rs b/asyncband/src/shutdown/mod.rs index 424bd523..faae3f51 100644 --- a/asyncband/src/shutdown/mod.rs +++ b/asyncband/src/shutdown/mod.rs @@ -24,10 +24,9 @@ //! shutdown request. //! * [`ShutdownWatch`] can observe the shutdown request without delaying completion. //! -//! Internally, the shutdown signal is implemented using a countdown latch, and the task completion -//! is tracked using a wait group. [`Shutdown`] is cloneable, allowing multiple control handles to -//! request shutdown or wait for completion. [`ShutdownGuard`] is also cloneable; each clone keeps -//! completion pending independently until it is dropped. +//! [`Shutdown`] is cloneable, allowing multiple control handles to request shutdown or wait for +//! completion. [`ShutdownGuard`] is also cloneable; each clone keeps completion pending +//! independently until it is dropped. //! //! Awaiting [`Shutdown`] requests shutdown and then waits until all [`ShutdownGuard`] handles have //! been dropped. The request is made when the future is first polled, not when the value is created @@ -40,18 +39,24 @@ //! # #[tokio::main] //! # async fn main() { //! let (shutdown, guard) = asyncband::shutdown::new(); +//! let mut tasks = Vec::new(); //! -//! for i in 0..3 { +//! for _ in 0..3 { //! let guard = guard.clone(); -//! tokio::spawn(async move { -//! println!("Task {} starting", i); +//! let task = tokio::spawn(async move { //! guard.shutdown_requested().await; -//! println!("Task {} done", i); +//! 1 //! }); +//! tasks.push(task); //! } //! drop(guard); //! //! shutdown.await; +//! let mut completed = 0; +//! for task in tasks { +//! completed += task.await.unwrap(); +//! } +//! assert_eq!(completed, 3); //! # } //! ``` @@ -180,10 +185,9 @@ impl ShutdownGuard { self.latch.wait().await; } - /// Returns an owned future that resolves when shutdown is requested. + /// Returns a future that resolves when shutdown is requested. /// - /// The returned future has no lifetime constraints and does not keep shutdown completion - /// pending. + /// The returned future can be moved into a spawned task. pub fn shutdown_requested_owned(&self) -> impl Future + 'static { self.latch.clone().wait_owned() } diff --git a/asyncband/src/waitgroup/mod.rs b/asyncband/src/waitgroup/mod.rs index 93567f3f..1e082558 100644 --- a/asyncband/src/waitgroup/mod.rs +++ b/asyncband/src/waitgroup/mod.rs @@ -15,43 +15,38 @@ // specific language governing permissions and limitations // under the License. -//! A synchronization primitive for waiting on multiple tasks to complete. +//! Wait for a set of worker handles to be dropped. //! -//! Similar to Go's WaitGroup, this type allows a task to wait for multiple other -//! tasks to finish. Each task holds a handle to the WaitGroup, and the main task -//! can wait for all handles to be dropped before proceeding. -//! -//! A WaitGroup waits for a collection of tasks to finish. The main task calls -//! [`clone()`] to create a new worker handle for each task, and can then wait -//! for all tasks to complete by calling `.await` on the WaitGroup. +//! A [`WaitGroup`] starts with one coordinator handle. Clone that handle once for each unit of work +//! and move the clones into their workers. Dropping a worker handle marks that worker as complete. +//! Awaiting the coordinator consumes it and waits until every remaining handle has been dropped. //! //! # Examples //! //! ``` //! # #[tokio::main] //! # async fn main() { -//! use std::time::Duration; -//! //! use asyncband::waitgroup::WaitGroup; -//! let wg = WaitGroup::new(); //! -//! for i in 0..3 { -//! let wg = wg.clone(); -//! tokio::spawn(async move { -//! println!("Task {} starting", i); -//! tokio::time::sleep(Duration::from_millis(100)).await; -//! // wg is automatically decremented when dropped -//! drop(wg); -//! }); +//! async fn do_work() {} +//! +//! let group = WaitGroup::new(); +//! let mut tasks = Vec::new(); +//! +//! for _ in 0..3 { +//! let worker = group.clone(); +//! tasks.push(tokio::spawn(async move { +//! do_work().await; +//! drop(worker); // Signals completion. This would also happen at the end of the task. +//! })); //! } //! -//! // Wait for all tasks to complete -//! wg.await; -//! println!("All tasks completed"); +//! group.await; +//! for task in tasks { +//! task.await.unwrap(); +//! } //! # } //! ``` -//! -//! [`clone()`]: WaitGroup::clone use std::fmt; use std::future::Future; @@ -67,7 +62,7 @@ use crate::internal::waitset::WakerToken; #[cfg(test)] mod tests; -/// A synchronization primitive for waiting on multiple tasks to complete. +/// A group of handles whose collective completion can be awaited. /// /// See the [module level documentation](self) for more. pub struct WaitGroup { @@ -104,7 +99,7 @@ impl WaitGroup { } impl Clone for WaitGroup { - /// Creates a new worker handle for the WaitGroup. + /// Creates a new worker handle for the wait group. /// /// This increments the WaitGroup counter. The counter will be decremented /// when the new handle is dropped. @@ -133,8 +128,7 @@ impl IntoFuture for WaitGroup { type Output = (); type IntoFuture = Wait; - /// Converts the WaitGroup into a future that completes when all tasks finish. This decreases - /// the WaitGroup counter. + /// Consumes this handle and waits for all other handles to be dropped. fn into_future(self) -> Self::IntoFuture { let state = self.state.clone(); drop(self); @@ -142,11 +136,10 @@ impl IntoFuture for WaitGroup { } } -/// A future that completes when all tasks in a WaitGroup have finished. +/// A future that completes when every [`WaitGroup`] handle has been dropped. /// -/// This type is created by either: (1) calling `.await` on a `WaitGroup`, or (2) cloning -/// itself, which does not increase the WaitGroup counter, but creates a new future that -/// will complete when the WaitGroup counter reaches zero. +/// Awaiting a [`WaitGroup`] creates this future. Cloning a `Wait` creates another observer without +/// adding a worker to the group. #[must_use = "futures do nothing unless you `.await` or poll them"] pub struct Wait { token: Option, diff --git a/tests-integration/tests/broadcast_mpmc_unbounded_test.rs b/tests-integration/tests/broadcast_mpmc_unbounded_test.rs index 5347b34c..8469ab9b 100644 --- a/tests-integration/tests/broadcast_mpmc_unbounded_test.rs +++ b/tests-integration/tests/broadcast_mpmc_unbounded_test.rs @@ -24,7 +24,6 @@ use std::task::Wake; use std::task::Waker; use std::thread; use std::time::Duration; -use std::time::Instant; use asyncband::broadcast::mpmc::*; @@ -330,8 +329,7 @@ fn panicking_clone_leaves_the_channel_consistent() { #[test] fn message_destructors_run_outside_the_channel_lock() { - let finished = Arc::new(AtomicUsize::new(0)); - let flag = finished.clone(); + let (finished_tx, finished_rx) = std::sync::mpsc::channel(); let worker = thread::spawn(move || { let (tx, mut rx1) = unbounded(); @@ -357,33 +355,15 @@ fn message_destructors_run_outside_the_channel_lock() { }); drop(tx); - flag.store(1, Ordering::SeqCst); + finished_tx.send(()).unwrap(); }); - let deadline = Instant::now() + Duration::from_secs(10); - while Instant::now() < deadline && finished.load(Ordering::SeqCst) == 0 { - thread::sleep(Duration::from_millis(10)); - } - assert_eq!( - finished.load(Ordering::SeqCst), - 1, - "a message destructor deadlocked against the channel lock" - ); + finished_rx + .recv_timeout(Duration::from_secs(10)) + .expect("a message destructor deadlocked against the channel lock"); worker.join().unwrap(); } -#[tokio::test] -async fn test_wait_mechanism() { - let (tx, mut rx) = unbounded(); - - let handle = tokio::spawn(async move { rx.recv().await }); - - tokio::time::sleep(Duration::from_millis(100)).await; - tx.send(42); - - assert_eq!(handle.await.unwrap(), Ok(42)); -} - #[test] fn send_wakes_a_parked_receiver_exactly_once() { let (tx, mut rx) = unbounded(); diff --git a/tests-integration/tests/latch_test.rs b/tests-integration/tests/latch_test.rs index 6e5d7a7f..b527c0bc 100644 --- a/tests-integration/tests/latch_test.rs +++ b/tests-integration/tests/latch_test.rs @@ -16,28 +16,14 @@ // under the License. use std::future::Future; -use std::prelude::rust_2015::Vec; use std::sync::Arc; -use std::sync::atomic::AtomicU32; use std::sync::atomic::AtomicUsize; use std::sync::atomic::Ordering; use std::task::Context; use std::task::Wake; use std::task::Waker; -use std::time::Duration; -use std::time::Instant; - -use asyncband::latch::*; - -macro_rules! assert_time { - ($time:expr, $mills:literal $(,)?) => { - assert!( - (Duration::from_millis($mills - 1) - ..Duration::from_millis($mills + std::cmp::max($mills >> 1, 50))) - .contains(&$time) - ) - }; -} + +use asyncband::latch::Latch; struct TrackWake(AtomicUsize); @@ -48,24 +34,22 @@ impl Wake for TrackWake { } #[test] -fn test_count_down() { - let latch = Latch::new(3); - latch.count_down(); - latch.count_down(); - latch.count_down(); - assert_eq!(latch.count(), 0); -} +fn countdown_operations_saturate_at_zero() { + let latch = Latch::new(5); -#[test] -fn test_try_wait() { - let latch = Latch::new(0); + latch.arrive(0); + assert_eq!(latch.try_wait(), Err(5)); + + latch.arrive(3); + assert_eq!(latch.try_wait(), Err(2)); + + latch.count_down(); + latch.arrive(2); assert_eq!(latch.try_wait(), Ok(())); -} -#[test] -fn test_try_wait_err() { - let latch = Latch::new(3); - assert_eq!(latch.try_wait(), Err(3)); + latch.count_down(); + latch.arrive(u32::MAX); + assert_eq!(latch.count(), 0); } #[test] @@ -85,142 +69,62 @@ fn cancelled_wait_releases_its_waker() { } #[test] -fn test_arrive_zero() { +fn final_arrival_wakes_every_waiter() { let latch = Latch::new(2); - latch.arrive(0); - assert_eq!(latch.count(), 2); -} - -#[test] -fn test_more_arrive() { - let latch = Latch::new(10); - for _ in 0..4 { - latch.arrive(3); - } - assert_eq!(latch.count(), 0); -} - -#[tokio::test] -async fn test_arrive() { - let latch = Latch::new(3); - latch.arrive(3); - latch.wait().await; - assert_eq!(latch.count(), 0); -} - -#[tokio::test] -async fn test_last_one_signal() { - let latch = Arc::new(Latch::new(3)); - let l1 = latch.clone(); + let first_tracker = Arc::new(TrackWake(AtomicUsize::new(0))); + let second_tracker = Arc::new(TrackWake(AtomicUsize::new(0))); + let first_waker = Waker::from(first_tracker.clone()); + let second_waker = Waker::from(second_tracker.clone()); + let mut first_context = Context::from_waker(&first_waker); + let mut second_context = Context::from_waker(&second_waker); + let mut first = Box::pin(latch.wait()); + let mut second = Box::pin(latch.wait()); + + assert!(first.as_mut().poll(&mut first_context).is_pending()); + assert!(second.as_mut().poll(&mut second_context).is_pending()); latch.count_down(); - latch.count_down(); + assert_eq!(first_tracker.0.load(Ordering::Relaxed), 0); + assert_eq!(second_tracker.0.load(Ordering::Relaxed), 0); - let start = Instant::now(); - - tokio::spawn(async move { - tokio::time::sleep(Duration::from_millis(32)).await; - l1.count_down() - }); - - latch.wait().await; - assert_time!(start.elapsed(), 32); - assert_eq!(latch.count(), 0); + latch.count_down(); + assert_eq!(first_tracker.0.load(Ordering::Relaxed), 1); + assert_eq!(second_tracker.0.load(Ordering::Relaxed), 1); + assert!(first.as_mut().poll(&mut first_context).is_ready()); + assert!(second.as_mut().poll(&mut second_context).is_ready()); } #[tokio::test] -async fn test_gate_wait() { +async fn owned_wait_can_move_to_another_task() { let latch = Arc::new(Latch::new(1)); - let tasks: Vec<_> = (0..4) - .map(|_| { - let latch = latch.clone(); - let start = Instant::now(); - - tokio::spawn(async move { - latch.wait().await; - start.elapsed() - }) - }) - .collect(); - - tokio::time::sleep(Duration::from_millis(20)).await; - latch.count_down(); + let waiter = tokio::spawn(latch.clone().wait_owned()); - for t in tasks { - assert_time!(t.await.unwrap(), 20); - } + latch.count_down(); + waiter.await.unwrap(); } -#[tokio::test] -async fn test_multi_tasks() { - const SIZE: u32 = 16; - let latch = Arc::new(Latch::new(SIZE)); - let counter = Arc::new(AtomicU32::new(0)); +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn concurrent_arrivals_complete_the_latch() { + const TASKS: usize = 32; - for _ in 0..SIZE { - let latch = latch.clone(); - let counter = counter.clone(); + let latch = Arc::new(Latch::new(TASKS as u32)); + let start = Arc::new(tokio::sync::Barrier::new(TASKS + 1)); + let mut tasks = Vec::with_capacity(TASKS); - tokio::spawn(async move { - tokio::time::sleep(Duration::from_millis(20)).await; - counter.fetch_add(1, Ordering::Relaxed); - latch.count_down() - }); + for _ in 0..TASKS { + let latch = latch.clone(); + let start = start.clone(); + tasks.push(tokio::spawn(async move { + start.wait().await; + latch.count_down(); + })); } - let start = Instant::now(); - + start.wait().await; latch.wait().await; - assert_time!(start.elapsed(), 20); - assert_eq!(counter.load(Ordering::Relaxed), SIZE); - assert_eq!(latch.count(), 0); -} - -#[tokio::test] -async fn test_more_count_down() { - const SIZE: u32 = 16; - let latch = Arc::new(Latch::new(SIZE)); - let counter = Arc::new(AtomicU32::new(0)); - - for _ in 0..(SIZE + (SIZE >> 1)) { - let latch = latch.clone(); - let counter = counter.clone(); - tokio::spawn(async move { - tokio::time::sleep(Duration::from_millis(10)).await; - counter.fetch_add(1, Ordering::Relaxed); - latch.count_down() - }); + for task in tasks { + task.await.unwrap(); } - - latch.wait().await; - assert!(counter.load(Ordering::Relaxed) >= SIZE); - assert_eq!(latch.count(), 0); - latch.count_down(); assert_eq!(latch.count(), 0); } - -#[tokio::test] -async fn test_select_two_wait() { - let latch1 = Arc::new(Latch::new(1)); - let latch2 = Arc::new(Latch::new(1)); - let l1 = latch1.clone(); - let l2 = latch2.clone(); - - tokio::spawn(async move { - tokio::time::sleep(Duration::from_millis(50)).await; - l1.count_down(); - }); - - tokio::spawn(async move { - tokio::time::sleep(Duration::from_millis(10)).await; - l2.count_down(); - }); - - assert!(tokio::select! { - _ = latch1.wait() => false, - _ = latch2.wait() => true, - }); - - latch1.wait().await; -} diff --git a/tests-integration/tests/mpsc_test.rs b/tests-integration/tests/mpsc_test.rs index ed132d36..50d5a108 100644 --- a/tests-integration/tests/mpsc_test.rs +++ b/tests-integration/tests/mpsc_test.rs @@ -16,7 +16,6 @@ // under the License. use std::task::Poll; -use std::time::Instant; use asyncband::mpsc; use asyncband::mpsc::RecvError; @@ -34,31 +33,11 @@ fn expect_ready(poll: Poll) -> T { } #[test] -fn test_unbounded_pressure() { - let n = 1024 * 1024; +fn unbounded_collects_from_multiple_producers() { let (tx, mut rx) = mpsc::unbounded(); test_runtime().block_on(async move { - let start = Instant::now(); - tokio::spawn(async move { - for i in 0..n { - tx.send(i).unwrap(); - } - }); - - for i in 0..n { - assert_eq!(rx.recv().await, Ok(i)); - } - println!("Elapsed: {:?}", start.elapsed()); - }); -} - -#[test] -fn test_unbounded_sum() { - let (tx, mut rx) = mpsc::unbounded(); - - test_runtime().block_on(async move { - for i in 0..100 { + for i in 0..8 { let tx = tx.clone(); tokio::spawn(async move { tx.send(i).unwrap(); @@ -70,7 +49,7 @@ fn test_unbounded_sum() { while let Ok(i) = rx.recv().await { sum += i; } - assert_eq!(sum, 4950); + assert_eq!(sum, 28); }); } @@ -182,31 +161,19 @@ async fn async_send_recv_unbounded() { } #[test] -fn try_recv_unbounded() { - for num in 0..100 { - let (tx, mut rx) = mpsc::unbounded(); - - for i in 0..num { - tx.send(i).unwrap(); - } - - for i in 0..num { - assert_eq!(rx.try_recv(), Ok(i)); - } +fn unbounded_try_recv_preserves_order_and_reports_state() { + let (tx, mut rx) = mpsc::unbounded(); - assert_eq!(rx.try_recv(), Err(TryRecvError::Empty)); - drop(tx); - assert_eq!(rx.try_recv(), Err(TryRecvError::Disconnected)); + for i in 0..4 { + tx.send(i).unwrap(); } -} -#[test] -fn try_recv_reports_disconnection_while_empty_unbounded() { - let (tx, mut rx) = mpsc::unbounded::<()>(); - - assert_eq!(Err(TryRecvError::Empty), rx.try_recv()); + for i in 0..4 { + assert_eq!(rx.try_recv(), Ok(i)); + } + assert_eq!(rx.try_recv(), Err(TryRecvError::Empty)); drop(tx); - assert_eq!(Err(TryRecvError::Disconnected), rx.try_recv()); + assert_eq!(rx.try_recv(), Err(TryRecvError::Disconnected)); } #[tokio::test] @@ -236,17 +203,17 @@ async fn async_send_recv_bounded() { } #[test] -fn try_send_recv_bounded() { - for num in 1..101 { - let (tx, mut rx) = mpsc::bounded(num); +fn bounded_try_send_respects_capacity_and_order() { + for capacity in [1, 4, 16] { + let (tx, mut rx) = mpsc::bounded(capacity); - for i in 0..num { + for i in 0..capacity { tx.try_send(i).unwrap(); } - assert_eq!(tx.try_send(num), Err(TrySendError::Full(num))); + assert_eq!(tx.try_send(capacity), Err(TrySendError::Full(capacity))); - for i in 0..num { + for i in 0..capacity { assert_eq!(rx.try_recv(), Ok(i)); } @@ -340,23 +307,3 @@ fn bounded_receiver_drop_returns_values_to_all_blocked_senders() { assert_eq!(first_error.into_inner(), 1); assert_eq!(second_error.into_inner(), 2); } - -#[test] -fn test_bounded_pressure() { - let n = 1024 * 1024; - let (tx, mut rx) = mpsc::bounded(1024); - - test_runtime().block_on(async move { - let start = Instant::now(); - tokio::spawn(async move { - for i in 0..n { - tx.send(i).await.unwrap(); - } - }); - - for i in 0..n { - assert_eq!(rx.recv().await, Ok(i)); - } - println!("Elapsed: {:?}", start.elapsed()); - }); -} diff --git a/tests-integration/tests/once_cell_test.rs b/tests-integration/tests/once_cell_test.rs index 5370f02d..f9303397 100644 --- a/tests-integration/tests/once_cell_test.rs +++ b/tests-integration/tests/once_cell_test.rs @@ -17,187 +17,122 @@ use std::sync::Arc; use std::sync::atomic::AtomicBool; +use std::sync::atomic::AtomicUsize; use std::sync::atomic::Ordering; -use std::time::Duration; use asyncband::once::OnceCell; -use tokio::sync::Mutex; +use tests_integration::poll_once; -struct Foo { - value: Arc, -} - -impl Foo { - async fn new(value: Arc) -> Self { - // simulate some async initialization work - tokio::time::sleep(Duration::from_millis(100)).await; - Foo { value } - } +struct DropFlag(Arc); - fn value(&self) -> bool { - self.value.load(Ordering::Acquire) - } -} - -impl Drop for Foo { +impl Drop for DropFlag { fn drop(&mut self) { - self.value.store(true, Ordering::Release); + self.0.store(true, Ordering::Release); } } #[tokio::test] -async fn drop_cell() { +async fn dropping_the_cell_drops_its_value() { let dropped = Arc::new(AtomicBool::new(false)); { let cell = OnceCell::new(); - assert!(cell.get().is_none()); - - let state = dropped.clone(); - cell.get_or_init(|| async { Foo::new(state).await }).await; - - let foo = cell.get().unwrap(); - assert!(!foo.value()); + cell.get_or_init(async || DropFlag(dropped.clone())).await; assert!(!dropped.load(Ordering::Acquire)); } assert!(dropped.load(Ordering::Acquire)); } -#[test] -fn multi_init() { - let rt = tokio::runtime::Builder::new_multi_thread() - .worker_threads(4) - .build() - .unwrap(); - - static CELL: OnceCell = OnceCell::new(); - - rt.block_on(async { - const N: usize = 100; - - let values = Arc::new(Mutex::new(vec![0; N])); - let mut handles = Vec::with_capacity(N); - - for i in 0..N { - let values = values.clone(); - handles.push(rt.spawn(async move { - let result = CELL.get_or_init(move || async move { i + 1000 }).await; - let mut values = values.lock().await; - values[i] = *result; - })); - } - - for handle in handles { - handle.await.unwrap(); - } - let cell_value = CELL.get().unwrap(); - for (index, value) in values.lock().await.iter().enumerate() { - assert_eq!(*value, *cell_value, "mismatch at index {index}"); - } - }); -} +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn concurrent_callers_publish_one_value() { + const TASKS: usize = 100; + + let cell = Arc::new(OnceCell::new()); + let attempts = Arc::new(AtomicUsize::new(0)); + let start = Arc::new(tokio::sync::Barrier::new(TASKS + 1)); + let mut tasks = Vec::with_capacity(TASKS); + + for value in 0..TASKS { + let cell = cell.clone(); + let attempts = attempts.clone(); + let start = start.clone(); + tasks.push(tokio::spawn(async move { + start.wait().await; + *cell + .get_or_init(async || { + attempts.fetch_add(1, Ordering::SeqCst); + value + }) + .await + })); + } -#[tokio::test] -async fn init_cancelled() { - static CELL: OnceCell = OnceCell::new(); - - let handle1 = tokio::spawn(async { - let fut = CELL.get_or_init(|| async { - tokio::time::sleep(Duration::from_millis(1000)).await; - 1 - }); - let timeout = tokio::time::timeout(Duration::from_millis(1), fut).await; - assert!(timeout.is_err()); - }); - - let handle2 = tokio::spawn(async { - tokio::time::sleep(Duration::from_millis(100)).await; - let value = CELL.get_or_init(|| async { 2 }).await; - assert_eq!(*value, 2); - }); - - handle1.await.unwrap(); - handle2.await.unwrap(); + start.wait().await; + let expected = tasks.remove(0).await.unwrap(); + for task in tasks { + assert_eq!(task.await.unwrap(), expected); + } + assert_eq!(attempts.load(Ordering::SeqCst), 1); } #[tokio::test] -async fn init_error() { - { - static CELL: OnceCell = OnceCell::new(); - - let handle1 = tokio::spawn(async { - let result = CELL.get_or_try_init(|| async { Err(()) }).await; - assert!(result.is_err()); - }); - - let handle2 = tokio::spawn(async { - tokio::time::sleep(Duration::from_millis(100)).await; - let value = CELL.get_or_try_init(|| async { Ok::<_, ()>(2) }).await; - assert_eq!(*value.unwrap(), 2); - }); - - handle1.await.unwrap(); - handle2.await.unwrap(); - } - - { - static CELL: OnceCell = OnceCell::new(); +async fn cancelled_initialization_can_be_retried() { + let cell = OnceCell::new(); + let mut first = Box::pin(cell.get_or_init(async || std::future::pending::().await)); - let handle1 = tokio::spawn(async { - let value = CELL.get_or_try_init(|| async { Ok::<_, ()>(2) }).await; - assert_eq!(*value.unwrap(), 2); - }); + assert!(poll_once(first.as_mut()).is_pending()); + drop(first); - let handle2 = tokio::spawn(async { - tokio::time::sleep(Duration::from_millis(100)).await; - let value = CELL.get_or_try_init(|| async { Err(()) }).await; - assert_eq!(*value.unwrap(), 2); - }); - - handle1.await.unwrap(); - handle2.await.unwrap(); - } + assert_eq!(*cell.get_or_init(async || 2).await, 2); } #[tokio::test] -async fn get_mut_or_init() { - let mut cell: OnceCell = OnceCell::new(); - let v = cell - .get_mut_or_init(async || { - tokio::time::sleep(Duration::from_millis(1)).await; - 41 +async fn failed_initialization_can_be_retried_and_success_is_cached() { + let cell = OnceCell::new(); + let attempts = AtomicUsize::new(0); + + let error = cell + .get_or_try_init(async || { + attempts.fetch_add(1, Ordering::SeqCst); + Err::("not ready") }) .await; - *v += 1; + assert_eq!(error, Err("not ready")); - let v = tokio::spawn(async move { *cell.get_or_init(async || 0).await }) - .await - .unwrap(); - assert_eq!(v, 42); + let value = cell + .get_or_try_init(async || { + attempts.fetch_add(1, Ordering::SeqCst); + Ok::<_, &str>(2) + }) + .await; + assert_eq!(value, Ok(&2)); + + let cached = cell + .get_or_try_init(async || { + attempts.fetch_add(1, Ordering::SeqCst); + Ok::<_, &str>(3) + }) + .await; + assert_eq!(cached, Ok(&2)); + assert_eq!(attempts.load(Ordering::SeqCst), 2); } #[tokio::test] -async fn get_mut_or_try_init() { +async fn exclusive_initialization_returns_mutable_access() { let mut cell: OnceCell = OnceCell::new(); - let r = cell - .get_mut_or_try_init(async || { - tokio::time::sleep(Duration::from_millis(1)).await; - Err(()) - }) + + let failed = cell + .get_mut_or_try_init(async || Err::("not ready")) .await; - assert!(r.is_err()); + assert_eq!(failed, Err("not ready")); assert_eq!(cell.get_mut(), None); - let v = tokio::spawn(async move { - let v = cell - .get_mut_or_try_init(async || Ok::<_, ()>(10)) - .await - .unwrap(); - *v += 5; - *v - }) - .await - .unwrap(); - assert_eq!(v, 15); + let value = cell.get_mut_or_init(async || 41).await; + *value += 1; + + let value = tokio::spawn(async move { *cell.get_or_init(async || 0).await }) + .await + .unwrap(); + assert_eq!(value, 42); } diff --git a/tests-integration/tests/once_map_test.rs b/tests-integration/tests/once_map_test.rs index aa150d35..ef0cf3d2 100644 --- a/tests-integration/tests/once_map_test.rs +++ b/tests-integration/tests/once_map_test.rs @@ -31,9 +31,6 @@ fn constructors_and_default() { let _: OnceMap = OnceMap::default(); let _: OnceMap = OnceMap::new(); let _: OnceMap = OnceMap::with_hasher(RandomState::new()); - - let map: OnceMap = OnceMap::new(); - assert!(format!("{map:?}").contains("OnceMap")); } #[tokio::test] diff --git a/tests-integration/tests/once_test.rs b/tests-integration/tests/once_test.rs index 71faaaeb..622f746e 100644 --- a/tests-integration/tests/once_test.rs +++ b/tests-integration/tests/once_test.rs @@ -15,160 +15,117 @@ // specific language governing permissions and limitations // under the License. +use std::sync::Arc; use std::sync::atomic::AtomicUsize; use std::sync::atomic::Ordering; -use std::time::Duration; use asyncband::once::Once; -use tests_integration::test_runtime; -use tokio_test::assert_ready; +use tests_integration::poll_once; #[tokio::test] -async fn test_call_once_runs_only_once() { - static ONCE: Once = Once::new(); - static COUNTER: AtomicUsize = AtomicUsize::new(0); +async fn call_once_runs_only_one_initializer() { + let once = Once::new(); + let counter = AtomicUsize::new(0); - assert!(!ONCE.is_completed()); + assert!(!once.is_completed()); - ONCE.call_once(async || { - COUNTER.fetch_add(1, Ordering::SeqCst); + once.call_once(async || { + counter.fetch_add(1, Ordering::SeqCst); }) .await; - assert!(ONCE.is_completed()); - assert_eq!(COUNTER.load(Ordering::SeqCst), 1); + assert!(once.is_completed()); + assert_eq!(counter.load(Ordering::SeqCst), 1); - // Second call should not run the closure - ONCE.call_once(async || { - COUNTER.fetch_add(1, Ordering::SeqCst); + once.call_once(async || { + counter.fetch_add(1, Ordering::SeqCst); }) .await; - assert_eq!(COUNTER.load(Ordering::SeqCst), 1); + assert_eq!(counter.load(Ordering::SeqCst), 1); } -#[test] -fn test_once_multi_task() { - static ONCE: Once = Once::new(); - static COUNTER: AtomicUsize = AtomicUsize::new(0); - - test_runtime().block_on(async { - const N: usize = 100; - - let mut handles = Vec::with_capacity(N); - - for _ in 0..N { - handles.push(tokio::spawn(async move { - ONCE.call_once(async || { - COUNTER.fetch_add(1, Ordering::SeqCst); - }) - .await; - })); - } - - for handle in handles { - handle.await.unwrap(); - } - - // Only one task should have incremented the counter - assert_eq!(COUNTER.load(Ordering::SeqCst), 1); - assert!(ONCE.is_completed()); - }); -} - -#[tokio::test] -async fn test_once_cancelled() { - static ONCE: Once = Once::new(); - static COUNTER: AtomicUsize = AtomicUsize::new(0); - - let handle1 = tokio::spawn(async { - let fut = ONCE.call_once(async || { - tokio::time::sleep(Duration::from_millis(1000)).await; - COUNTER.fetch_add(1, Ordering::SeqCst); - }); - let timeout = tokio::time::timeout(Duration::from_millis(1), fut).await; - assert!(timeout.is_err()); - }); - - let handle2 = tokio::spawn(async { - tokio::time::sleep(Duration::from_millis(100)).await; - ONCE.call_once(async || { - COUNTER.fetch_add(10, Ordering::SeqCst); - }) - .await; - }); +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn concurrent_callers_run_one_initializer() { + const TASKS: usize = 100; + + let once = Arc::new(Once::new()); + let counter = Arc::new(AtomicUsize::new(0)); + let start = Arc::new(tokio::sync::Barrier::new(TASKS + 1)); + let mut tasks = Vec::with_capacity(TASKS); + + for _ in 0..TASKS { + let once = once.clone(); + let counter = counter.clone(); + let start = start.clone(); + tasks.push(tokio::spawn(async move { + start.wait().await; + once.call_once(async || { + counter.fetch_add(1, Ordering::SeqCst); + }) + .await; + })); + } - handle1.await.unwrap(); - handle2.await.unwrap(); + start.wait().await; + for task in tasks { + task.await.unwrap(); + } - // The second task should have run since the first was cancelled - assert_eq!(COUNTER.load(Ordering::SeqCst), 10); - assert!(ONCE.is_completed()); + assert_eq!(counter.load(Ordering::SeqCst), 1); + assert!(once.is_completed()); } #[tokio::test] -async fn test_once_debug() { +async fn cancelled_initializer_can_be_retried() { let once = Once::new(); - let debug_str = format!("{:?}", once); - assert!(debug_str.contains("Once")); - assert!(debug_str.contains("done")); - assert!(debug_str.contains("false")); + let mut first = Box::pin(once.call_once(async || std::future::pending::<()>().await)); - once.call_once(async || {}).await; + assert!(poll_once(first.as_mut()).is_pending()); + drop(first); - let debug_str = format!("{:?}", once); - assert!(debug_str.contains("true")); -} - -#[tokio::test] -async fn test_once_default() { - let once = Once::default(); - assert!(!once.is_completed()); + once.call_once(async || {}).await; + assert!(once.is_completed()); } #[tokio::test] -async fn test_once_retry_after_panic() { - static ONCE: Once = Once::new(); - static COUNTER: AtomicUsize = AtomicUsize::new(0); - - let handle = tokio::spawn(async { - ONCE.call_once(async || { - COUNTER.fetch_add(1, Ordering::SeqCst); - panic!("boom"); - }) - .await; +async fn panicked_initializer_can_be_retried() { + let once = Arc::new(Once::new()); + let counter = Arc::new(AtomicUsize::new(0)); + + let handle = tokio::spawn({ + let once = once.clone(); + let counter = counter.clone(); + async move { + once.call_once(async || { + counter.fetch_add(1, Ordering::SeqCst); + panic!("boom"); + }) + .await; + } }); - let err = handle.await.expect_err("once init should panic"); - assert!(err.is_panic()); + let error = handle.await.expect_err("initializer should panic"); + assert!(error.is_panic()); - ONCE.call_once(async || { - COUNTER.fetch_add(1, Ordering::SeqCst); + once.call_once(async || { + counter.fetch_add(1, Ordering::SeqCst); }) .await; - assert_eq!(COUNTER.load(Ordering::SeqCst), 2); - assert!(ONCE.is_completed()); + assert_eq!(counter.load(Ordering::SeqCst), 2); + assert!(once.is_completed()); } #[tokio::test] -async fn test_once_wait() { - // wait after call_once completed - { - let once = Once::new(); - once.call_once(async || {}).await; - assert_ready!(tokio_test::task::spawn(once.wait()).poll()); - } +async fn wait_observes_completion() { + let once = Once::new(); + let mut waiting = Box::pin(once.wait()); - // wait before call_once completed - { - static ONCE: Once = Once::new(); - let handle = tokio::spawn(async { - ONCE.wait().await; - }); + assert!(poll_once(waiting.as_mut()).is_pending()); + once.call_once(async || {}).await; + assert!(poll_once(waiting.as_mut()).is_ready()); - tokio::time::sleep(Duration::from_millis(100)).await; - ONCE.call_once(async || {}).await; - handle.await.unwrap(); - } + let mut completed = Box::pin(once.wait()); + assert!(poll_once(completed.as_mut()).is_ready()); } diff --git a/tests-integration/tests/pool_behavior_test.rs b/tests-integration/tests/pool_behavior_test.rs index 864ce28e..3ea95a8a 100644 --- a/tests-integration/tests/pool_behavior_test.rs +++ b/tests-integration/tests/pool_behavior_test.rs @@ -56,6 +56,18 @@ impl ManageObject for CountingManager { } } +#[test] +#[should_panic(expected = "bounded pool max_size must be greater than zero")] +fn bounded_pool_rejects_zero_capacity() { + bounded::Pool::new( + bounded::PoolConfig::new(0), + CountingManager { + next: Arc::new(AtomicUsize::new(0)), + detached: Arc::new(AtomicUsize::new(0)), + }, + ); +} + #[test] fn bounded_construction_allocates_idle_storage_lazily() { let pool = bounded::Pool::new( diff --git a/tests-integration/tests/rwlock_test.rs b/tests-integration/tests/rwlock_test.rs index a368f444..1f95bd6a 100644 --- a/tests-integration/tests/rwlock_test.rs +++ b/tests-integration/tests/rwlock_test.rs @@ -17,683 +17,183 @@ use std::num::NonZeroUsize; use std::sync::Arc; -use std::sync::Weak; use asyncband::rwlock::*; +use tests_integration::poll_once; +use tokio_test::assert_pending; +use tokio_test::assert_ready; #[test] -fn test_try_read_write_never_blocks() { - // Test that try_read and try_write never block +fn try_methods_respect_held_guards() { let rwlock = Arc::new(RwLock::new(42)); - let r1 = rwlock.try_read().unwrap(); - let _r2 = rwlock.try_read().unwrap(); - assert_eq!(*r1, 42); - assert_eq!(*_r2, 42); + let first_reader = rwlock.try_read().unwrap(); + let second_reader = rwlock.try_read().unwrap(); + assert_eq!(*first_reader, 42); + assert_eq!(*second_reader, 42); assert!(rwlock.try_write().is_none()); - drop(r1); - drop(_r2); + drop(first_reader); + drop(second_reader); - let w = rwlock.try_write().unwrap(); - assert_eq!(*w, 42); + let writer = rwlock.try_write().unwrap(); + assert_eq!(*writer, 42); assert!(rwlock.try_read().is_none()); assert!(rwlock.clone().try_read_owned().is_none()); } #[test] -fn test_get_mut_provides_exclusive_access() { - // Test that get_mut provides exclusive access to the data +fn get_mut_and_into_inner_use_exclusive_access() { let mut rwlock = RwLock::new(100); - let data = rwlock.get_mut(); - *data = 200; + *rwlock.get_mut() = 200; assert_eq!(*rwlock.get_mut(), 200); - - let inner = rwlock.into_inner(); - assert_eq!(inner, 200); + assert_eq!(rwlock.into_inner(), 200); } #[test] -fn test_with_max_readers() { +fn max_readers_limits_concurrent_readers() { let rwlock = RwLock::with_max_readers(10, NonZeroUsize::new(2).unwrap()); - let r1 = rwlock.try_read().unwrap(); - let r2 = rwlock.try_read().unwrap(); - assert_eq!(*r1, 10); - assert_eq!(*r2, 10); - + let first = rwlock.try_read().unwrap(); + let second = rwlock.try_read().unwrap(); assert!(rwlock.try_read().is_none()); - assert!(rwlock.try_write().is_none()); - drop(r1); - - let r3 = rwlock.try_read().unwrap(); - assert_eq!(*r3, 10); - + drop(first); + let replacement = rwlock.try_read().unwrap(); assert!(rwlock.try_read().is_none()); - drop(r2); - drop(r3); - - let mut w = rwlock.try_write().unwrap(); - *w = 20; - drop(w); - - let r = rwlock.try_read().unwrap(); - assert_eq!(*r, 20); + drop(second); + drop(replacement); + assert!(rwlock.try_write().is_some()); } -#[tokio::test] -async fn test_stress_concurrent_readers_writers() { - // Test concurrent readers and writers with RwLock - let rwlock = Arc::new(RwLock::new(0i32)); - let mut reader_results = Vec::new(); - let mut writer_results = Vec::new(); +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn concurrent_readers_and_writers_preserve_values() { + const READERS: usize = 50; + const WRITERS: usize = 10; - // Spawn reader tasks - for i in 0..50 { - let rwlock_clone = rwlock.clone(); - let handle = tokio::spawn(async move { - let guard = rwlock_clone.read().await; - let value = *guard; + let rwlock = Arc::new(RwLock::new(0)); + let start = Arc::new(tokio::sync::Barrier::new(READERS + WRITERS + 1)); + let mut readers = Vec::with_capacity(READERS); + let mut writers = Vec::with_capacity(WRITERS); + + for _ in 0..READERS { + let rwlock = rwlock.clone(); + let start = start.clone(); + readers.push(tokio::spawn(async move { + start.wait().await; + let value = *rwlock.read().await; tokio::task::yield_now().await; - (i, value) - }); - reader_results.push(handle); + value + })); } - // Spawn writer tasks - for i in 0..10 { - let rwlock_clone = rwlock.clone(); - let handle = tokio::spawn(async move { - let mut guard = rwlock_clone.write().await; - let old_value = *guard; + for _ in 0..WRITERS { + let rwlock = rwlock.clone(); + let start = start.clone(); + writers.push(tokio::spawn(async move { + start.wait().await; + let mut guard = rwlock.write().await; *guard += 1; tokio::task::yield_now().await; - (i + 100, old_value, *guard) - }); - writer_results.push(handle); - } - - for handle in reader_results { - let (reader_id, value) = handle.await.unwrap(); - assert!( - (0..=10).contains(&value), - "Reader {reader_id} saw invalid value: {value}" - ); - } - - let mut writer_values = Vec::new(); - for handle in writer_results { - let (writer_id, old_value, new_value) = handle.await.unwrap(); - assert_eq!( - new_value, - old_value + 1, - "Writer {writer_id} increment failed: {old_value} -> {new_value}" - ); - writer_values.push((old_value, new_value)); - } - - let final_guard = rwlock.read().await; - assert_eq!( - *final_guard, 10, - "Final value should be 10 after 10 increments" - ); - - writer_values.sort_by_key(|(old, _)| *old); - for (i, (old_value, new_value)) in writer_values.iter().enumerate() { - assert_eq!( - *old_value, i as i32, - "Writer operations should be sequential" - ); - assert_eq!( - *new_value, - (i + 1) as i32, - "Each increment should be atomic" - ); + })); } -} - -#[tokio::test] -async fn test_guard_prevents_concurrent_access() { - let rwlock = Arc::new(RwLock::new(0)); - let rwlock_clone = rwlock.clone(); - let writer_queued = Arc::new(std::sync::atomic::AtomicBool::new(false)); - let writer_queued_clone = writer_queued.clone(); - let writer_completed = Arc::new(std::sync::atomic::AtomicBool::new(false)); - let writer_completed_clone = writer_completed.clone(); - - let read_guard = rwlock.read().await; - - assert!(rwlock.try_write().is_none()); - - let handle = tokio::spawn(async move { - writer_queued_clone.store(true, std::sync::atomic::Ordering::SeqCst); - let mut write_guard = rwlock_clone.write().await; - *write_guard = 123; - writer_completed_clone.store(true, std::sync::atomic::Ordering::SeqCst); - *write_guard - }); - while !writer_queued.load(std::sync::atomic::Ordering::SeqCst) { - tokio::task::yield_now().await; + start.wait().await; + for reader in readers { + assert!((0..=WRITERS as i32).contains(&reader.await.unwrap())); } - - for _ in 0..10 { - tokio::task::yield_now().await; + for writer in writers { + writer.await.unwrap(); } - - assert!(!writer_completed.load(std::sync::atomic::Ordering::SeqCst)); - assert!(rwlock.try_write().is_none()); - - drop(read_guard); - - let result = handle.await.unwrap(); - assert_eq!(result, 123); - assert!(writer_completed.load(std::sync::atomic::Ordering::SeqCst)); - - // Verify the write took effect and lock is available - let final_guard = rwlock.try_read().unwrap(); - assert_eq!(*final_guard, 123); -} - -#[test] -fn test_lock_panic_safety() { - // Test panic safety with synchronous locks - use std::panic::AssertUnwindSafe; - - let rwlock = Arc::new(RwLock::new(0)); - let rwlock_clone = rwlock.clone(); - - let result = std::panic::catch_unwind(AssertUnwindSafe(move || { - let _guard = rwlock_clone.try_read().unwrap(); - panic!("test panic"); - })); - - assert!(result.is_err()); - // Lock should be released after panic - assert!(rwlock.try_read().is_some()); -} - -#[tokio::test] -async fn test_async_lock_panic_safety() { - // Test panic safety with async locks - let rwlock = Arc::new(RwLock::new(0)); - let rwlock_clone = rwlock.clone(); - - let handle = tokio::spawn(async move { - let _guard = rwlock_clone.read().await; - panic!("async test panic"); - }); - - // panic - assert!(handle.await.is_err()); - - let guard = rwlock.try_read(); - assert!(guard.is_some()); -} - -#[tokio::test] -async fn test_owned_guard_panic_safety() { - // Test panic safety with owned guards - let rwlock = Arc::new(RwLock::new(0)); - let rwlock_clone = rwlock.clone(); - - let handle = tokio::spawn(async move { - let _guard = rwlock_clone.clone().read_owned().await; - panic!("owned guard panic"); - }); - - assert!(handle.await.is_err()); - - // Lock should be available after the panicked task - let guard = rwlock.try_read(); - assert!(guard.is_some()); -} - -#[tokio::test] -async fn test_mapped_guard_panic_safety() { - // Test panic safety with mapped guards - let rwlock = Arc::new(RwLock::new((66, vec![1, 2, 3]))); - let rwlock_clone = rwlock.clone(); - - let handle = tokio::spawn(async move { - let guard = rwlock_clone.read().await; - let _mapped = RwLockReadGuard::map(guard, |data| &data.0); - panic!("mapped guard panic"); - }); - - assert!(handle.await.is_err()); - - let guard = rwlock.try_read(); - assert!(guard.is_some()); -} - -#[tokio::test] -async fn test_mapped_write_guard_panic_safety() { - // Test panic safety with mapped write guards - let rwlock = Arc::new(RwLock::new((42, String::from("test")))); - let rwlock_clone = rwlock.clone(); - - let handle = tokio::spawn(async move { - let guard = rwlock_clone.write().await; - let _mapped = RwLockWriteGuard::map(guard, |data| &mut data.0); - panic!("mapped write guard panic"); - }); - - assert!(handle.await.is_err()); - - let guard = rwlock.try_write(); - assert!(guard.is_some()); -} - -#[tokio::test] -async fn test_owned_mapped_read_guard_panic_safety() { - // Test panic safety with owned mapped read guards - let rwlock = Arc::new(RwLock::new((100, vec![4, 5, 6]))); - let rwlock_clone = rwlock.clone(); - - let handle = tokio::spawn(async move { - let guard = rwlock_clone.read_owned().await; - let _mapped = OwnedRwLockReadGuard::map(guard, |data| &data.1); - panic!("owned mapped read guard panic"); - }); - - assert!(handle.await.is_err()); - - let guard = rwlock.try_read(); - assert!(guard.is_some()); -} - -#[tokio::test] -async fn test_owned_mapped_write_guard_panic_safety() { - // Test panic safety with owned mapped write guards - let rwlock = Arc::new(RwLock::new((200, String::from("owned")))); - let rwlock_clone = rwlock.clone(); - - let handle = tokio::spawn(async move { - let guard = rwlock_clone.write_owned().await; - let _mapped = OwnedRwLockWriteGuard::map(guard, |data| &mut data.1); - panic!("owned mapped write guard panic"); - }); - - assert!(handle.await.is_err()); - - let guard = rwlock.try_write(); - assert!(guard.is_some()); -} - -#[tokio::test] -async fn test_memory_ordering_correctness() { - // Test that rwlock provides proper memory ordering guarantees - // When one task modifies data under rwlock protection, - // another task should see the modification after acquiring the lock - let rwlock = Arc::new(RwLock::new(vec![1, 2, 3])); - let rwlock_clone = rwlock.clone(); - - let handle = tokio::spawn(async move { - let mut guard = rwlock_clone.write().await; - guard.push(4); - guard[0] = 100; - // Lock is released when guard is dropped - }); - - handle.await.unwrap(); - - let guard = rwlock.read().await; - assert_eq!(*guard, vec![100, 2, 3, 4]); -} - -#[tokio::test] -async fn test_rwlock_debug_when_locked() { - let rwlock = Arc::new(RwLock::new(78)); - - let rwlock_debug_unlocked = format!("{rwlock:?}"); - assert!( - rwlock_debug_unlocked.contains("78"), - "RwLock Debug should show value when unlocked, got: {rwlock_debug_unlocked}" - ); - - let write_guard = rwlock.write().await; - let rwlock_debug_write = format!("{rwlock:?}"); - assert!( - rwlock_debug_write.contains(""), - "RwLock Debug should show when write lock is held, got: {rwlock_debug_write}" - ); - drop(write_guard); - - let read_guard = rwlock.read().await; - let rwlock_debug_read = format!("{rwlock:?}"); - - let shows_value = rwlock_debug_read.contains("78"); - let shows_locked = rwlock_debug_read.contains(""); - assert!( - shows_value || shows_locked, - "RwLock Debug with read lock should show either value or , got: {rwlock_debug_read}" - ); - - drop(read_guard); - - let rwlock_debug_final = format!("{rwlock:?}"); - assert!( - rwlock_debug_final.contains("78"), - "RwLock Debug should show value when all locks are released, got: {rwlock_debug_final}" - ); + assert_eq!(*rwlock.read().await, WRITERS as i32); } #[tokio::test] -async fn test_rwlock_zst() { - // Test that RwLock works correctly with Zero-Sized Types +async fn zero_sized_values_follow_locking_rules() { let rwlock = Arc::new(RwLock::new(())); - let rwlock_clone = rwlock.clone(); - let handle = tokio::spawn(async move { - let guard = rwlock_clone.read().await; - *guard; - }); - - handle.await.unwrap(); - - let guard1 = rwlock.read().await; - let guard2 = rwlock.clone().read_owned().await; - *guard1; - *guard2; - + let borrowed = rwlock.read().await; + let owned = rwlock.clone().read_owned().await; assert!(rwlock.try_write().is_none()); - drop(guard1); - drop(guard2); - - let mut write_guard = rwlock.write().await; - *write_guard = (); - drop(write_guard); - - let try_write_guard = rwlock.try_write().unwrap(); - *try_write_guard; - drop(try_write_guard); - - let guard = rwlock.try_read().unwrap(); - *guard; -} - -#[tokio::test] -async fn test_owned_write_guard_map_memory_leak() { - // Test OwnedRwLockWriteGuard::map memory management - let rwlock = Arc::new(RwLock::new(29u32)); - let weak_ref: Weak> = Arc::downgrade(&rwlock); - - { - let write_guard = rwlock.clone().write_owned().await; - let mut mapped_guard = OwnedRwLockWriteGuard::map(write_guard, |data| data); - *mapped_guard = 100; - assert_eq!(*mapped_guard, 100); - } - - drop(rwlock); - assert!( - weak_ref.upgrade().is_none(), - "expected Arc to be dropped (no strong refs)" - ); -} - -#[tokio::test] -async fn test_owned_write_guard_filter_map_memory_leak() { - // Test OwnedRwLockWriteGuard::filter_map memory management - - // Test success case - { - let rwlock = Arc::new(RwLock::new(Some(29u32))); - let weak_ref: Weak>> = Arc::downgrade(&rwlock); - - { - let write_guard = rwlock.clone().write_owned().await; - let mut mapped_guard = - OwnedRwLockWriteGuard::filter_map(write_guard, |data| data.as_mut()) - .expect("Should succeed"); - *mapped_guard = 100; - assert_eq!(*mapped_guard, 100); - } - - drop(rwlock); - assert!( - weak_ref.upgrade().is_none(), - "expected Arc to be dropped after filter_map success" - ); - } - - // Test failure case - { - let rwlock = Arc::new(RwLock::new(None::)); - let weak_ref: Weak>> = Arc::downgrade(&rwlock); - - { - let write_guard = rwlock.clone().write_owned().await; - let result = OwnedRwLockWriteGuard::filter_map(write_guard, |data| data.as_mut()); - assert!(result.is_err(), "filter_map should have failed for None"); - } - - drop(rwlock); - assert!( - weak_ref.upgrade().is_none(), - "expected Arc to be dropped after filter_map failure" - ); - } -} - -#[tokio::test] -async fn test_owned_mapped_write_guard_map_memory_leak() { - // Test OwnedMappedRwLockWriteGuard::map memory management - let rwlock = Arc::new(RwLock::new("test".to_string())); - let weak_ref: Weak> = Arc::downgrade(&rwlock); - - { - let write_guard = rwlock.clone().write_owned().await; - let mapped_guard1 = OwnedRwLockWriteGuard::map(write_guard, |s| s); - let mut mapped_guard2 = OwnedMappedRwLockWriteGuard::map(mapped_guard1, |s| s.as_mut_str()); - mapped_guard2.make_ascii_uppercase(); - assert_eq!(&*mapped_guard2, "TEST"); - } - - drop(rwlock); - assert!( - weak_ref.upgrade().is_none(), - "expected Arc to be dropped (no strong refs)" - ); + drop(borrowed); + drop(owned); + assert!(rwlock.try_write().is_some()); } #[tokio::test] -async fn test_owned_mapped_write_guard_filter_map_memory_leak() { - // Test OwnedMappedRwLockWriteGuard::filter_map memory management - - // Test success case - { - let rwlock = Arc::new(RwLock::new(vec![1, 2, 3])); - let weak_ref: Weak>> = Arc::downgrade(&rwlock); - - { - let write_guard = rwlock.clone().write_owned().await; - let mapped_guard1 = OwnedRwLockWriteGuard::map(write_guard, |v| v); - let mut mapped_guard2 = OwnedMappedRwLockWriteGuard::filter_map(mapped_guard1, |v| { - if !v.is_empty() { Some(&mut v[0]) } else { None } - }) - .expect("Should succeed"); - *mapped_guard2 = 100; - assert_eq!(*mapped_guard2, 100); - } - - drop(rwlock); - assert!( - weak_ref.upgrade().is_none(), - "Memory leak detected on filter_map success" - ); - } +async fn owned_write_mappings_preserve_lock_ownership() { + let rwlock = Arc::new(RwLock::new(Some(vec![1, 2, 3]))); + let weak = Arc::downgrade(&rwlock); - // Test failure case - { - let rwlock = Arc::new(RwLock::new(Vec::::new())); - let weak_ref: Weak>> = Arc::downgrade(&rwlock); - - { - let write_guard = rwlock.clone().write_owned().await; - let mapped_guard1 = OwnedRwLockWriteGuard::map(write_guard, |v| v); - let result = OwnedMappedRwLockWriteGuard::filter_map(mapped_guard1, |v| { - if !v.is_empty() { Some(&mut v[0]) } else { None } - }); - - assert!( - result.is_err(), - "filter_map should have failed for empty vector" - ); - if let Err(original_guard) = result { - assert!(original_guard.is_empty()); - } - } + let identity = OwnedRwLockWriteGuard::map(rwlock.clone().write_owned().await, |value| value); + drop(identity); - drop(rwlock); - assert!( - weak_ref.upgrade().is_none(), - "Memory leak detected on filter_map failure" - ); - } -} - -#[tokio::test] -async fn test_owned_read_guard_map_memory_leak() { - // Test OwnedRwLockReadGuard::map memory management - let rwlock = Arc::new(RwLock::new(29u32)); - let weak_ref: Weak> = Arc::downgrade(&rwlock); - - { - let read_guard = rwlock.clone().read_owned().await; - let mapped_guard = OwnedRwLockReadGuard::map(read_guard, |data| data); - assert_eq!(*mapped_guard, 29); - } + let failed = OwnedRwLockWriteGuard::filter_map(rwlock.clone().write_owned().await, |_| { + None::<&mut Vec> + }); + let original = match failed { + Ok(_) => panic!("mapping should fail"), + Err(original) => original, + }; + drop(original); + + let value = + OwnedRwLockWriteGuard::filter_map(rwlock.clone().write_owned().await, Option::as_mut) + .unwrap(); + let value = OwnedMappedRwLockWriteGuard::map(value, Vec::as_mut_slice); + let mut value = + OwnedMappedRwLockWriteGuard::filter_map(value, |value| value.first_mut()).unwrap(); + *value = 100; + let value = value.downgrade(); drop(rwlock); - assert!( - weak_ref.upgrade().is_none(), - "expected Arc to be dropped (no strong ref; potential memory leaks)" - ); -} - -#[tokio::test] -async fn test_owned_read_guard_filter_map_memory_leak() { - // Test OwnedRwLockReadGuard::filter_map memory management - - // Test success case - { - let rwlock = Arc::new(RwLock::new(Some(29u32))); - let weak_ref: Weak>> = Arc::downgrade(&rwlock); - - { - let read_guard = rwlock.clone().read_owned().await; - let mapped_guard = OwnedRwLockReadGuard::filter_map(read_guard, |data| data.as_ref()) - .expect("filter_map should succeed for Some(_) value"); - assert_eq!(*mapped_guard, 29); - } - - drop(rwlock); - assert!( - weak_ref.upgrade().is_none(), - "expected Arc to be dropped after filter_map success" - ); - } - - // Test failure case - { - let rwlock = Arc::new(RwLock::new(None::)); - let weak_ref: Weak>> = Arc::downgrade(&rwlock); + assert!(weak.upgrade().is_some()); + assert_eq!(*value, 100); - { - let read_guard = rwlock.clone().read_owned().await; - let result = OwnedRwLockReadGuard::filter_map(read_guard, |data| data.as_ref()); - assert!(result.is_err(), "filter_map should have failed for None"); - } - - drop(rwlock); - assert!( - weak_ref.upgrade().is_none(), - "expected Arc to be dropped after filter_map failure" - ); - } + drop(value); + assert!(weak.upgrade().is_none()); } #[tokio::test] -async fn test_owned_mapped_read_guard_map_memory_leak() { - // Test OwnedMappedRwLockReadGuard::map memory management - let rwlock = Arc::new(RwLock::new("test".to_string())); - let weak_ref: Weak> = Arc::downgrade(&rwlock); - - { - let read_guard = rwlock.clone().read_owned().await; - let mapped_guard1 = OwnedRwLockReadGuard::map(read_guard, |s| s); - let mapped_guard2 = OwnedMappedRwLockReadGuard::map(mapped_guard1, |s| s.as_str()); - assert_eq!(&*mapped_guard2, "test"); - } +async fn owned_read_mappings_preserve_lock_ownership() { + let rwlock = Arc::new(RwLock::new(Some(vec![1, 2, 3]))); + let weak = Arc::downgrade(&rwlock); + + let identity = OwnedRwLockReadGuard::map(rwlock.clone().read_owned().await, |value| value); + drop(identity); + + let failed = + OwnedRwLockReadGuard::filter_map(rwlock.clone().read_owned().await, |_| None::<&Vec>); + let original = match failed { + Ok(_) => panic!("mapping should fail"), + Err(original) => original, + }; + drop(original); + + let value = OwnedRwLockReadGuard::filter_map(rwlock.clone().read_owned().await, Option::as_ref) + .unwrap(); + let value = OwnedMappedRwLockReadGuard::map(value, Vec::as_slice); + let value = OwnedMappedRwLockReadGuard::filter_map(value, |value| value.first()).unwrap(); drop(rwlock); - assert!( - weak_ref.upgrade().is_none(), - "Memory leak detected: Arc was not deallocated" - ); -} - -#[tokio::test] -async fn test_owned_mapped_read_guard_filter_map_memory_leak() { - // Test OwnedMappedRwLockReadGuard::filter_map memory management - - // Test success case - { - let rwlock = Arc::new(RwLock::new(Some(29u32))); - let weak_ref: Weak>> = Arc::downgrade(&rwlock); - - { - let read_guard = rwlock.clone().read_owned().await; - let mapped_guard1 = OwnedRwLockReadGuard::map(read_guard, |data| data); - let mapped_guard2 = - OwnedMappedRwLockReadGuard::filter_map(mapped_guard1, |opt| opt.as_ref()) - .expect("filter_map should succeed for Some(_) value"); - assert_eq!(*mapped_guard2, 29); - } - - drop(rwlock); - assert!( - weak_ref.upgrade().is_none(), - "expected Arc to be dropped after filter_map success" - ); - } - - // Test failure case - { - let rwlock = Arc::new(RwLock::new(None::)); - let weak_ref: Weak>> = Arc::downgrade(&rwlock); + assert!(weak.upgrade().is_some()); + assert_eq!(*value, 1); - { - let read_guard = rwlock.clone().read_owned().await; - let mapped_guard1 = OwnedRwLockReadGuard::map(read_guard, |data| data); - let result = OwnedMappedRwLockReadGuard::filter_map(mapped_guard1, |opt| opt.as_ref()); - assert!(result.is_err(), "filter_map should have failed for None"); - } - - drop(rwlock); - assert!( - weak_ref.upgrade().is_none(), - "Memory leak detected on filter_map failure" - ); - } + drop(value); + assert!(weak.upgrade().is_none()); } #[tokio::test] -async fn test_downgrade_atomicity() { - // Test atomic downgrade behavior for all guard types +async fn every_write_guard_type_can_downgrade() { let rwlock = Arc::new(RwLock::new((42, "test".to_string()))); - // Test basic write guard downgrade { let mut write_guard = rwlock.write().await; write_guard.0 = 100; @@ -701,7 +201,6 @@ async fn test_downgrade_atomicity() { let read_guard = write_guard.downgrade(); assert_eq!(read_guard.0, 100); - // Writers blocked, readers allowed assert!(rwlock.try_write().is_none()); let concurrent_read = rwlock.try_read().unwrap(); assert_eq!(concurrent_read.0, 100); @@ -709,7 +208,6 @@ async fn test_downgrade_atomicity() { drop(read_guard); } - // Test owned write guard downgrade { let mut owned_write = rwlock.clone().write_owned().await; owned_write.1 = "updated".to_string(); @@ -721,7 +219,6 @@ async fn test_downgrade_atomicity() { drop(owned_read); } - // Test mapped write guard downgrade { let write_guard = rwlock.write().await; let mut mapped_write = RwLockWriteGuard::map(write_guard, |data| &mut data.0); @@ -734,7 +231,6 @@ async fn test_downgrade_atomicity() { drop(mapped_read); } - // Test owned mapped write guard downgrade { let owned_write = rwlock.clone().write_owned().await; let mut owned_mapped = OwnedRwLockWriteGuard::map(owned_write, |data| &mut data.1); @@ -751,64 +247,7 @@ async fn test_downgrade_atomicity() { } #[tokio::test] -async fn test_downgrade_allows_concurrent_readers() { - // Test that downgrading a write lock allows other readers to acquire the lock. - let rwlock = Arc::new(RwLock::new(0i32)); - let writer_started = Arc::new(std::sync::atomic::AtomicBool::new(false)); - let downgrade_completed = Arc::new(std::sync::atomic::AtomicBool::new(false)); - - let writer_rwlock = rwlock.clone(); - let writer_started_clone = writer_started.clone(); - let downgrade_completed_clone = downgrade_completed.clone(); - - let writer_handle = tokio::spawn(async move { - let mut write_guard = writer_rwlock.write().await; - *write_guard = 42; - writer_started_clone.store(true, std::sync::atomic::Ordering::SeqCst); - - tokio::time::sleep(tokio::time::Duration::from_millis(10)).await; - - let read_guard = write_guard.downgrade(); - downgrade_completed_clone.store(true, std::sync::atomic::Ordering::SeqCst); - - assert_eq!(*read_guard, 42); - - tokio::time::sleep(tokio::time::Duration::from_millis(20)).await; - *read_guard - }); - - while !writer_started.load(std::sync::atomic::Ordering::SeqCst) { - tokio::task::yield_now().await; - } - - let mut reader_handles = vec![]; - for i in 0..5 { - let reader_rwlock = rwlock.clone(); - let downgrade_completed_clone = downgrade_completed.clone(); - - let handle = tokio::spawn(async move { - while !downgrade_completed_clone.load(std::sync::atomic::Ordering::SeqCst) { - tokio::task::yield_now().await; - } - - let read_guard = reader_rwlock.read().await; - assert_eq!(*read_guard, 42); - (i, *read_guard) - }); - reader_handles.push(handle); - } - - // All tasks should complete successfully - let writer_result = writer_handle.await.unwrap(); - assert_eq!(writer_result, 42); - - for handle in reader_handles { - let (reader_id, value) = handle.await.unwrap(); - assert_eq!(value, 42, "Reader {reader_id} should see the written value"); - } -} -#[tokio::test] -async fn test_downgrade_with_max_readers() { +async fn downgraded_guard_counts_against_max_readers() { let rwlock = Arc::new(RwLock::with_max_readers(0, NonZeroUsize::new(3).unwrap())); let mut write_guard = rwlock.write().await; @@ -839,218 +278,21 @@ async fn test_downgrade_with_max_readers() { } #[tokio::test] -async fn test_downgrade_panic_safety() { - let rwlock = Arc::new(RwLock::new((0, "test".to_string()))); - - { - let rwlock_clone = rwlock.clone(); - let handle = tokio::spawn(async move { - let mut write_guard = rwlock_clone.write().await; - write_guard.0 = 42; - let read_guard = write_guard.downgrade(); - assert_eq!(read_guard.0, 42); - panic!("test panic after downgrade"); - }); - - assert!(handle.await.is_err()); - - let guard = rwlock.try_read().unwrap(); - assert_eq!(guard.0, 42); - drop(guard); - } - - { - let rwlock_clone = rwlock.clone(); - let handle = tokio::spawn(async move { - let owned_write = rwlock_clone.write_owned().await; - let mut mapped_write = OwnedRwLockWriteGuard::map(owned_write, |data| &mut data.1); - *mapped_write = "panic_test".to_string(); - let _mapped_read = mapped_write.downgrade(); - panic!("test panic with owned mapped downgrade"); - }); - - assert!(handle.await.is_err()); - - let guard = rwlock.try_read().unwrap(); - assert_eq!(guard.1, "panic_test"); - } -} - -#[tokio::test] -async fn test_downgrade_prevents_deadlock() { - // Test the classic deadlock prevention scenario - // Demonstrates how downgrade enables safe lock ordering patterns - - let rwlock = Arc::new(RwLock::new(vec![1, 2, 3])); - - let rwlock1 = rwlock.clone(); - let task1 = tokio::spawn(async move { - let mut write_guard = rwlock1.write().await; - write_guard.push(4); - let len_after_write = write_guard.len(); - - let read_guard = write_guard.downgrade(); - - assert_eq!(read_guard.len(), len_after_write); - - tokio::time::sleep(tokio::time::Duration::from_millis(10)).await; - - let final_len = read_guard.len(); - drop(read_guard); - final_len - }); - - let rwlock2 = rwlock.clone(); - let task2 = tokio::spawn(async move { - tokio::time::sleep(tokio::time::Duration::from_millis(5)).await; - - let mut write_guard = rwlock2.write().await; - write_guard.push(5); - - let read_guard = write_guard.downgrade(); - let final_len = read_guard.len(); - - drop(read_guard); - final_len - }); - - let rwlock3 = rwlock.clone(); - let task3 = tokio::spawn(async move { - tokio::time::sleep(tokio::time::Duration::from_millis(15)).await; - - let read_guard = rwlock3.read().await; - read_guard.len() - }); - - let result1 = task1.await.unwrap(); - let result2 = task2.await.unwrap(); - let result3 = task3.await.unwrap(); - - assert_eq!(result1, 4); // After first push - assert_eq!(result2, 5); // After second push - assert_eq!(result3, 5); // Final state - - let final_read = rwlock.read().await; - assert_eq!(final_read.len(), 5); - assert_eq!(*final_read, vec![1, 2, 3, 4, 5]); -} - -#[tokio::test] -async fn test_downgrade_with_waiting_writers() { - let rwlock = Arc::new(RwLock::new(0i32)); - let writer_queued = Arc::new(std::sync::atomic::AtomicBool::new(false)); - let downgrade_done = Arc::new(std::sync::atomic::AtomicBool::new(false)); - - let rwlock1 = rwlock.clone(); - let downgrade_done_clone = downgrade_done.clone(); - let downgrade_task = tokio::spawn(async move { - let mut write_guard = rwlock1.write().await; - *write_guard = 42; - - let read_guard = write_guard.downgrade(); - downgrade_done_clone.store(true, std::sync::atomic::Ordering::SeqCst); - - tokio::time::sleep(tokio::time::Duration::from_millis(50)).await; - assert_eq!(*read_guard, 42); - - drop(read_guard); - 42 - }); - - while !downgrade_done.load(std::sync::atomic::Ordering::SeqCst) { - tokio::task::yield_now().await; - } - - let rwlock2 = rwlock.clone(); - let writer_queued_clone = writer_queued.clone(); - let writer_task = tokio::spawn(async move { - writer_queued_clone.store(true, std::sync::atomic::Ordering::SeqCst); - - let mut write_guard = rwlock2.write().await; - - assert_eq!(*write_guard, 42); - *write_guard = 100; - - drop(write_guard); - 100 - }); - - while !writer_queued.load(std::sync::atomic::Ordering::SeqCst) { - tokio::task::yield_now().await; - } - - tokio::time::sleep(tokio::time::Duration::from_millis(10)).await; - - let mut reader_tasks = vec![]; - for i in 0..3 { - let rwlock_clone = rwlock.clone(); - - let reader_task = tokio::spawn(async move { - let read_guard = rwlock_clone.read().await; - let value = *read_guard; - drop(read_guard); - (i, value) - }); - reader_tasks.push(reader_task); - } - - let downgrade_result = downgrade_task.await.unwrap(); - assert_eq!(downgrade_result, 42); - - let writer_result = writer_task.await.unwrap(); - assert_eq!(writer_result, 100); - - for reader_task in reader_tasks { - let (reader_id, value) = reader_task.await.unwrap(); - // The readers might see either value depending on exact timing, - // so let's not make strict assertions here - println!("Reader {reader_id} saw value: {value}"); - } - - let final_read = rwlock.read().await; - assert_eq!(*final_read, 100); -} - -#[tokio::test] -async fn test_owned_write_guard_downgrade_no_memory_leak() { - let rwlock = Arc::new(RwLock::new(42i32)); - let weak_ref = Arc::downgrade(&rwlock); - - { - let write_guard = rwlock.clone().write_owned().await; - let _read_guard = write_guard.downgrade(); - } - - drop(rwlock); - - assert_eq!( - weak_ref.strong_count(), - 0, - "Memory leak detected in OwnedRwLockWriteGuard downgrade!" - ); -} - -#[tokio::test] -async fn test_owned_mapped_write_guard_downgrade_no_memory_leak() { - #[derive(Debug)] - struct Data { - value: i32, - } - - let rwlock = Arc::new(RwLock::new(Data { value: 42 })); - let weak_ref = Arc::downgrade(&rwlock); - - { - let write_guard = rwlock.clone().write_owned().await; - let mapped_write_guard = OwnedRwLockWriteGuard::map(write_guard, |data| &mut data.value); - let _mapped_read_guard = mapped_write_guard.downgrade(); - } - - drop(rwlock); - - assert_eq!( - weak_ref.strong_count(), - 0, - "Memory leak detected in OwnedMappedRwLockWriteGuard downgrade!" - ); +async fn queued_writer_precedes_a_later_reader() { + let rwlock = RwLock::new(0); + let first_reader = rwlock.read().await; + let mut writer = Box::pin(rwlock.write()); + let mut later_reader = Box::pin(rwlock.read()); + + assert_pending!(poll_once(writer.as_mut())); + assert_pending!(poll_once(later_reader.as_mut())); + + drop(first_reader); + let mut writer_guard = assert_ready!(poll_once(writer.as_mut())); + assert_pending!(poll_once(later_reader.as_mut())); + + *writer_guard = 100; + drop(writer_guard); + let reader_guard = assert_ready!(poll_once(later_reader.as_mut())); + assert_eq!(*reader_guard, 100); } diff --git a/tests-integration/tests/semaphore_test.rs b/tests-integration/tests/semaphore_test.rs index 4adbef70..fead9a3f 100644 --- a/tests-integration/tests/semaphore_test.rs +++ b/tests-integration/tests/semaphore_test.rs @@ -22,7 +22,6 @@ use std::task::Context; use std::task::Poll; use std::task::Wake; use std::task::Waker; -use std::vec::Vec; use asyncband::semaphore::Semaphore; @@ -81,29 +80,6 @@ fn forget() { assert!(sem.try_acquire(1).is_none()); } -#[tokio::test] -async fn stress_test() { - let sem = Arc::new(Semaphore::new(5)); - let mut join_handles = Vec::new(); - for i in 0..100 { - let sem_clone = sem.clone(); - join_handles.push(tokio::spawn(async move { - let _p = sem_clone.acquire(1).await; - tokio::time::sleep(std::time::Duration::from_millis(100 - i)).await; - })); - } - for j in join_handles { - j.await.unwrap(); - } - // there should be exactly 5 semaphores available now - let _p1 = sem.try_acquire(1).unwrap(); - let _p2 = sem.try_acquire(1).unwrap(); - let _p3 = sem.try_acquire(1).unwrap(); - let _p4 = sem.try_acquire(1).unwrap(); - let _p5 = sem.try_acquire(1).unwrap(); - assert!(sem.try_acquire(1).is_none()); -} - #[test] fn add_max_amount_permits() { let s = Semaphore::new(0); diff --git a/tests-integration/tests/waitgroup_test.rs b/tests-integration/tests/waitgroup_test.rs index 281ee498..970db9d8 100644 --- a/tests-integration/tests/waitgroup_test.rs +++ b/tests-integration/tests/waitgroup_test.rs @@ -16,69 +16,50 @@ // under the License. use std::future::IntoFuture; -use std::time::Duration; use asyncband::waitgroup::WaitGroup; -use tests_integration::test_runtime; +use tests_integration::poll_once; -#[test] -fn test_wait_group_drop() { +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn waits_for_all_worker_handles() { let wg = WaitGroup::new(); - for _i in 0..100 { - let w = wg.clone(); - test_runtime().spawn(async move { - drop(w); - }); + let mut tasks = Vec::new(); + for _ in 0..100 { + let worker = wg.clone(); + tasks.push(tokio::spawn(async move { + drop(worker); + })); } - pollster::block_on(wg.into_future()); -} -#[test] -fn test_wait_group_await() { - let wg = WaitGroup::new(); - for _i in 0..100 { - let w = wg.clone(); - test_runtime().spawn(async move { - w.await; - }); + wg.await; + for task in tasks { + task.await.unwrap(); } - pollster::block_on(wg.into_future()); } #[test] -fn test_wait_group_timeout() { +fn wait_is_pending_until_the_last_handle_drops() { let wg = WaitGroup::new(); - let _wg_clone = wg.clone(); - let timeout = test_runtime().block_on(async move { - tokio::select! { - _ = tokio::time::sleep(Duration::from_millis(50)) => true , - _ = wg => false, - } - }); - assert!(timeout); + let worker = wg.clone(); + let mut wait = Box::pin(wg.into_future()); + + assert!(poll_once(wait.as_mut()).is_pending()); + drop(worker); + assert!(poll_once(wait.as_mut()).is_ready()); } #[test] -fn test_wait_group_cancel() { +fn cancelling_one_wait_does_not_cancel_another() { let wg = WaitGroup::new(); - let wg_clone = wg.clone().into_future(); - let wg_clone_2 = wg.clone().into_future(); - test_runtime().block_on(async move { - tokio::select! { - _ = tokio::time::sleep(Duration::ZERO) => {}, - _ = wg_clone => {} - } - }); - let fut = test_runtime().spawn(async move { - wg_clone_2.await; - }); - std::thread::sleep(Duration::from_millis(50)); - drop(wg); - let timeout = test_runtime().block_on(async move { - tokio::select! { - _ = tokio::time::sleep(Duration::from_secs(60)) => true , - _ = fut => false, - } - }); - assert!(!timeout); + let worker = wg.clone(); + let first = wg.into_future(); + let mut second = Box::pin(first.clone()); + let mut first = Box::pin(first); + + assert!(poll_once(first.as_mut()).is_pending()); + assert!(poll_once(second.as_mut()).is_pending()); + drop(first); + + drop(worker); + assert!(poll_once(second.as_mut()).is_ready()); }