diff --git a/CHANGELOG.md b/CHANGELOG.md index 6b4bcfc..1e89274 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,15 @@ All notable changes to this project will be documented in this file. ## Unreleased +### New features + +* Implement `broadcast::unbounded`, an unbounded broadcast channel that retains messages until all + active receivers consume them or are dropped. + +### Bug fixes + +* Clean up cancelled broadcast receive futures registered through the internal `WaitSet`. + ## [0.6.3] - 2026-01-21 ### Improvements diff --git a/mea/src/barrier/mod.rs b/mea/src/barrier/mod.rs index 76d12d3..99dc303 100644 --- a/mea/src/barrier/mod.rs +++ b/mea/src/barrier/mod.rs @@ -75,6 +75,7 @@ use std::task::Poll; use crate::internal::Mutex; use crate::internal::WaitSet; +use crate::internal::WaiterId; #[cfg(test)] mod tests; @@ -228,7 +229,11 @@ impl Barrier { if state.arrived == self.n { state.arrived = 0; state.generation += 1; - state.waiters.wake_all(); + let wakers = state.waiters.take_wakers(); + drop(state); + for waker in wakers { + waker.wake(); + } return BarrierWaitResult(true); } @@ -250,7 +255,7 @@ impl Barrier { /// This future will complete when all tasks have reached the barrier point. #[must_use = "futures do nothing unless you `.await` or poll them"] struct BarrierWait<'a> { - idx: Option, + idx: Option, generation: usize, barrier: &'a Barrier, } @@ -277,7 +282,9 @@ impl Future for BarrierWait<'_> { if *generation < state.generation { Poll::Ready(()) } else { - state.waiters.register_waker(idx, cx); + let waker = state.waiters.register_waker(idx, cx); + drop(state); + drop(waker); Poll::Pending } } diff --git a/mea/src/broadcast/mod.rs b/mea/src/broadcast/mod.rs index 2460601..5763a00 100644 --- a/mea/src/broadcast/mod.rs +++ b/mea/src/broadcast/mod.rs @@ -16,6 +16,8 @@ //! //! This module provides broadcast channels in one of the following policies: //! +//! * [`unbounded`]: messages are buffered until all active receivers consume them. //! * [`overflow`]: when the channel is full, the oldest messages are overwritten. pub mod overflow; +pub mod unbounded; diff --git a/mea/src/broadcast/overflow/mod.rs b/mea/src/broadcast/overflow/mod.rs index 34b7036..68281ff 100644 --- a/mea/src/broadcast/overflow/mod.rs +++ b/mea/src/broadcast/overflow/mod.rs @@ -73,6 +73,7 @@ use std::task::Poll; use crate::internal::Mutex; use crate::internal::RwLock; use crate::internal::WaitSet; +use crate::internal::WaiterId; #[cfg(test)] mod tests; @@ -218,7 +219,13 @@ impl Drop for Sender { 1 => { // If this is the last sender, we need to wake up the receiver so it can // observe the disconnected state. - self.shared.waiters.lock().wake_all(); + let wakers = { + let mut waiters = self.shared.waiters.lock(); + waiters.take_wakers() + }; + for waker in wakers { + waker.wake(); + } } _ => { // there are still other senders left, do nothing @@ -254,7 +261,13 @@ impl Sender { } // Notify all waiting receivers. - self.shared.waiters.lock().wake_all(); + let wakers = { + let mut waiters = self.shared.waiters.lock(); + waiters.take_wakers() + }; + for waker in wakers { + waker.wake(); + } } /// Creates a new receiver that starts receiving messages from the current tail of the channel. @@ -435,7 +448,22 @@ impl Receiver { struct Recv<'a, T> { receiver: &'a mut Receiver, - index: Option, + index: Option, +} + +impl Drop for Recv<'_, T> { + fn drop(&mut self) { + // Ready paths clear the waiter ID, so only a cancelled pending receive takes this lock. + if self.index.is_none() { + return; + } + + let waker = { + let mut waiters = self.receiver.shared.waiters.lock(); + waiters.remove_waker(&mut self.index) + }; + drop(waker); + } } impl Future for Recv<'_, T> { @@ -446,9 +474,16 @@ impl Future for Recv<'_, T> { loop { match receiver.try_recv() { - Ok(val) => return Poll::Ready(Ok(val)), - Err(TryRecvError::Lagged(n)) => return Poll::Ready(Err(RecvError::Lagged(n))), + Ok(val) => { + *index = None; + return Poll::Ready(Ok(val)); + } + Err(TryRecvError::Lagged(n)) => { + *index = None; + return Poll::Ready(Err(RecvError::Lagged(n))); + } Err(TryRecvError::Disconnected) => { + *index = None; return Poll::Ready(Err(RecvError::Disconnected)); } Err(TryRecvError::Empty) => {} @@ -468,11 +503,14 @@ impl Future for Recv<'_, T> { // Check for Closed // Use Acquire to ensure we see all writes before the sender dropped. if shared.senders.load(Ordering::Acquire) == 0 { + *index = None; return Poll::Ready(Err(RecvError::Disconnected)); } // Register Waker - waiters.register_waker(index, cx); + let waker = waiters.register_waker(index, cx); + drop(waiters); + drop(waker); return Poll::Pending; } } diff --git a/mea/src/broadcast/overflow/tests.rs b/mea/src/broadcast/overflow/tests.rs index 0a1bb9b..78b279e 100644 --- a/mea/src/broadcast/overflow/tests.rs +++ b/mea/src/broadcast/overflow/tests.rs @@ -12,7 +12,12 @@ // See the License for the specific language governing permissions and // limitations under the License. +use std::future::Future; +use std::sync::atomic::Ordering; +use std::task::Context; + use super::*; +use crate::count_waker; #[tokio::test] async fn test_broadcast_basic() { @@ -77,6 +82,22 @@ async fn test_wait_mechanism() { assert_eq!(handle.await.unwrap(), Ok(42)); } +#[tokio::test] +async fn test_recv_cancellation_removes_waiter() { + let (tx, mut rx) = channel::(10); + let (waker, wake_count) = count_waker(); + let mut cx = Context::from_waker(&waker); + let mut recv = Box::pin(rx.recv()); + + assert!(recv.as_mut().poll(&mut cx).is_pending()); + + drop(recv); + tx.send(1); + + assert_eq!(wake_count.load(Ordering::Relaxed), 0); + assert_eq!(rx.try_recv(), Ok(1)); +} + #[tokio::test] async fn test_subscribe() { let (tx, _rx) = channel(10); diff --git a/mea/src/broadcast/unbounded/mod.rs b/mea/src/broadcast/unbounded/mod.rs new file mode 100644 index 0000000..35e1913 --- /dev/null +++ b/mea/src/broadcast/unbounded/mod.rs @@ -0,0 +1,665 @@ +// Copyright 2024 tison +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! A multi-producer multi-consumer broadcast channel with an unbounded buffer. +//! +//! This channel supports multiple senders and multiple receivers. Each message sent by any +//! sender is received by all active receivers. If a receiver falls behind, messages are buffered +//! until the receiver consumes them or is dropped. +//! +//! # Memory usage +//! +//! This channel does not impose a capacity limit. A slow or stalled receiver can cause the +//! buffer to grow without bound, because messages are retained until every active receiver has +//! consumed them or the receiver is dropped. Use [`Sender::buffer_len`] to monitor the number of +//! messages currently retained by the shared buffer. +//! +//! # Receivers +//! +//! Each receiver has an independent cursor. Use [`Sender::subscribe`] to create a receiver that +//! starts at the current tail of the channel, or [`Receiver::resubscribe`] to skip this receiver's +//! backlog and start a new receiver at the current tail. +//! +//! # Examples +//! +//! Basic usage: +//! +//! ``` +//! use mea::broadcast::unbounded; +//! +//! # #[tokio::main] +//! # async fn main() { +//! let (tx, mut rx1) = unbounded::channel(); +//! let mut rx2 = tx.subscribe(); +//! +//! tx.send(10); +//! tx.send(20); +//! +//! assert_eq!(rx1.recv().await, Ok(10)); +//! assert_eq!(rx1.recv().await, Ok(20)); +//! assert_eq!(rx2.recv().await, Ok(10)); +//! assert_eq!(rx2.recv().await, Ok(20)); +//! # } +//! ``` +//! +//! Slow receivers do not miss messages: +//! +//! ``` +//! use mea::broadcast::unbounded; +//! +//! # #[tokio::main] +//! # async fn main() { +//! let (tx, mut rx) = unbounded::channel(); +//! +//! tx.send(1); +//! tx.send(2); +//! tx.send(3); +//! +//! assert_eq!(rx.recv().await, Ok(1)); +//! assert_eq!(rx.recv().await, Ok(2)); +//! assert_eq!(rx.recv().await, Ok(3)); +//! # } +//! ``` + +use std::collections::VecDeque; +use std::fmt; +use std::future::Future; +use std::pin::Pin; +use std::sync::Arc; +use std::sync::atomic::AtomicUsize; +use std::sync::atomic::Ordering; +use std::task::Context; +use std::task::Poll; + +use slab::Slab; + +use crate::internal::Mutex; +use crate::internal::WaitSet; +use crate::internal::WaiterId; + +#[cfg(test)] +mod tests; + +/// Creates a new broadcast channel with an unbounded buffer. +/// +/// See [module-level documentation](self) for broadcast channel semantics. +/// +/// # Examples +/// +/// ``` +/// use mea::broadcast::unbounded; +/// +/// let (tx, mut rx) = unbounded::channel(); +/// tx.send(10); +/// assert_eq!(rx.try_recv(), Ok(10)); +/// ``` +pub fn channel() -> (Sender, Receiver) { + let mut receivers = Slab::new(); + let index = receivers.insert(0); + let shared = Arc::new(Shared { + inner: Mutex::new(Inner { + buffer: VecDeque::new(), + head: 0, + head_receivers: 1, + tail: 0, + receivers, + }), + senders: AtomicUsize::new(1), + waiters: Mutex::new(WaitSet::new()), + }); + let sender = Sender { + shared: shared.clone(), + }; + let receiver = Receiver { shared, index }; + (sender, receiver) +} + +/// Error returned by [`Receiver::recv`]. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum RecvError { + /// The sender has become disconnected, and there will never be any more data received on it. + Disconnected, +} + +impl fmt::Display for RecvError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + RecvError::Disconnected => write!(f, "receiving on a closed channel"), + } + } +} + +impl std::error::Error for RecvError {} + +/// Error returned by [`Receiver::try_recv`]. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum TryRecvError { + /// This channel is currently empty, but the sender(s) have not yet disconnected, so data may + /// yet become available. + Empty, + /// The sender has become disconnected, and there will never be any more data received on it. + Disconnected, +} + +impl fmt::Display for TryRecvError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + TryRecvError::Empty => write!(f, "receiving on an empty channel"), + TryRecvError::Disconnected => write!(f, "receiving on a closed channel"), + } + } +} + +impl std::error::Error for TryRecvError {} + +struct Inner { + /// Messages whose versions are in the range `[head, tail)`. + buffer: VecDeque>, + /// The version of the first message in `buffer`. + head: u64, + /// The number of active receivers whose cursor equals `head`. + head_receivers: usize, + /// The next message version to assign. + tail: u64, + /// Cursor for each active receiver. + receivers: Slab, +} + +impl Inner { + fn insert_receiver(&mut self, head: u64) -> usize { + if head == self.head { + self.head_receivers += 1; + } + + self.receivers.insert(head) + } + + fn remove_receiver(&mut self, index: usize) -> Vec> { + let head = self.receivers.remove(index); + + if head == self.head { + self.release_head_receiver() + } else { + Vec::new() + } + } + + fn advance_receiver(&mut self, index: usize, next_head: u64) -> Vec> { + let head = self.receivers[index]; + self.receivers[index] = next_head; + + if head == self.head { + self.release_head_receiver() + } else { + Vec::new() + } + } + + fn release_head_receiver(&mut self) -> Vec> { + self.head_receivers -= 1; + + if self.head_receivers == 0 { + self.reclaim_consumed() + } else { + Vec::new() + } + } + + fn receive(&mut self, index: usize) -> Option<(Arc, Vec>)> { + let head = self.receivers[index]; + + if head < self.tail { + debug_assert!(head >= self.head); + let offset = (head - self.head) as usize; + let msg = self.buffer[offset].clone(); + let reclaimed = self.advance_receiver(index, head + 1); + Some((msg, reclaimed)) + } else { + None + } + } + + fn reclaim_consumed(&mut self) -> Vec> { + let mut next_head = self.tail; + let mut head_receivers = 0; + + for (_, head) in self.receivers.iter() { + if *head < next_head { + next_head = *head; + head_receivers = 1; + } else if *head == next_head { + head_receivers += 1; + } + } + + debug_assert!(next_head >= self.head); + let consumed = usize::try_from(next_head - self.head) + .expect("retained broadcast message count exceeds usize"); + // Move reclaimed messages out so their Drop impls run after `inner` is unlocked. + let reclaimed = self.buffer.drain(..consumed).collect(); + + self.head = next_head; + self.head_receivers = head_receivers; + reclaimed + } +} + +struct Shared { + inner: Mutex>, + /// Number of active senders. + senders: AtomicUsize, + /// Waiters (receivers) waiting for new messages. + waiters: Mutex, +} + +/// A sender handle to the broadcast channel. +/// +/// The sender can be cloned to create multiple producers. When all senders are dropped, +/// the channel is closed. +pub struct Sender { + shared: Arc>, +} + +impl Clone for Sender { + fn clone(&self) -> Self { + self.shared.senders.fetch_add(1, Ordering::Relaxed); + Self { + shared: self.shared.clone(), + } + } +} + +impl Drop for Sender { + fn drop(&mut self) { + match self.shared.senders.fetch_sub(1, Ordering::AcqRel) { + 1 => { + // If this is the last sender, we need to wake up the receiver so it can + // observe the disconnected state. + let wakers = { + let mut waiters = self.shared.waiters.lock(); + waiters.take_wakers() + }; + for waker in wakers { + waker.wake(); + } + } + _ => { + // there are still other senders left, do nothing + } + } + } +} + +impl Sender { + /// Broadcasts a value to all active receivers. + /// + /// This operation does not wait for receiver capacity. If receivers fall behind, messages + /// remain buffered until all active receivers have consumed them or the lagging receivers + /// are dropped. + /// + /// If no receivers are active, the message is dropped immediately. + /// + /// # Panics + /// + /// Panics if the internal message version counter overflows. After `u64::MAX` successful sends + /// on one channel instance, the next send panics. + /// + /// # Examples + /// + /// ``` + /// use mea::broadcast::unbounded; + /// + /// let (tx, mut rx) = unbounded::channel(); + /// tx.send(10); + /// assert_eq!(rx.try_recv(), Ok(10)); + /// ``` + pub fn send(&self, msg: T) { + let msg = Arc::new(msg); + + { + let mut inner = self.shared.inner.lock(); + inner.tail = inner + .tail + .checked_add(1) + .expect("broadcast channel version counter overflowed"); + + if inner.receivers.is_empty() { + // No receivers means no one will read this message; advance `head` so the + // invariant that `buffer` covers versions `[head, tail)` still holds without + // buffering anything. The buffer is already drained when the last receiver was + // dropped, so there is nothing to clear here. + debug_assert!(inner.buffer.is_empty()); + debug_assert_eq!(inner.head_receivers, 0); + inner.head = inner.tail; + } else { + inner.buffer.push_back(msg); + } + } + + // Notify all waiting receivers. + let wakers = { + let mut waiters = self.shared.waiters.lock(); + waiters.take_wakers() + }; + for waker in wakers { + waker.wake(); + } + } + + /// Returns the number of messages currently retained by the shared buffer. + /// + /// This is not the number of messages any single receiver can still read. It is the shared + /// backlog kept alive by the slowest active receiver. + /// + /// # Examples + /// + /// ``` + /// use mea::broadcast::unbounded; + /// + /// let (tx, mut rx) = unbounded::channel(); + /// tx.send(10); + /// assert_eq!(tx.buffer_len(), 1); + /// + /// assert_eq!(rx.try_recv(), Ok(10)); + /// assert_eq!(tx.buffer_len(), 0); + /// ``` + pub fn buffer_len(&self) -> usize { + self.shared.inner.lock().buffer.len() + } + + /// Returns the number of active receivers. + /// + /// # Examples + /// + /// ``` + /// use mea::broadcast::unbounded; + /// + /// let (tx, rx) = unbounded::channel::(); + /// assert_eq!(tx.receiver_count(), 1); + /// + /// let rx2 = tx.subscribe(); + /// assert_eq!(tx.receiver_count(), 2); + /// + /// drop(rx); + /// drop(rx2); + /// assert_eq!(tx.receiver_count(), 0); + /// ``` + pub fn receiver_count(&self) -> usize { + self.shared.inner.lock().receivers.len() + } + + /// Creates a new receiver that starts receiving messages from the current tail of the channel. + /// + /// # Examples + /// + /// ``` + /// use mea::broadcast::unbounded; + /// use mea::broadcast::unbounded::TryRecvError; + /// + /// # #[tokio::main] + /// # async fn main() { + /// let (tx, _) = unbounded::channel(); + /// tx.send(10); + /// + /// let mut rx = tx.subscribe(); + /// assert_eq!(rx.try_recv(), Err(TryRecvError::Empty)); + /// tx.send(20); + /// assert_eq!(rx.recv().await, Ok(20)); + /// # } + /// ``` + pub fn subscribe(&self) -> Receiver { + let mut inner = self.shared.inner.lock(); + let head = inner.tail; + let index = inner.insert_receiver(head); + let shared = self.shared.clone(); + Receiver { shared, index } + } +} + +/// A receiver handle to the broadcast channel. +/// +/// Each receiver sees every message sent to the channel while the receiver is active. +pub struct Receiver { + shared: Arc>, + index: usize, +} + +impl Drop for Receiver { + fn drop(&mut self) { + let reclaimed = { + let mut inner = self.shared.inner.lock(); + inner.remove_receiver(self.index) + }; + drop(reclaimed); + } +} + +impl Receiver { + /// Receives the next value for this receiver. + /// + /// # Returns + /// + /// * `Ok(T)`: The next message. + /// * `Err(RecvError::Disconnected)`: All senders have been dropped and no more messages are + /// available. + /// + /// # Cancel safety + /// + /// This method is cancel safe. If `recv` is used as the event in a `select` statement and some + /// other branch completes first, it is guaranteed that no messages were received on this + /// channel. + /// + /// # Examples + /// + /// ``` + /// use mea::broadcast::unbounded; + /// + /// # #[tokio::main] + /// # async fn main() { + /// let (tx, mut rx) = unbounded::channel(); + /// tx.send(10); + /// assert_eq!(rx.recv().await, Ok(10)); + /// # } + /// ``` + pub async fn recv(&mut self) -> Result { + Recv { + receiver: self, + index: None, + } + .await + } + + /// Attempts to receive the next value for this receiver without blocking. + /// + /// # Returns + /// + /// * `Ok(T)`: The next message. + /// * `Err(TryRecvError::Empty)`: No message is currently available. + /// * `Err(TryRecvError::Disconnected)`: All senders have been dropped and no more messages are + /// available. + /// + /// # Examples + /// + /// ``` + /// use mea::broadcast::unbounded; + /// + /// let (tx, mut rx) = unbounded::channel(); + /// tx.send(10); + /// assert_eq!(rx.try_recv(), Ok(10)); + /// ``` + pub fn try_recv(&mut self) -> Result { + let (msg, reclaimed) = self.try_recv_shared()?; + drop(reclaimed); + Ok((*msg).clone()) + } +} + +impl Receiver { + fn try_recv_shared(&mut self) -> Result<(Arc, Vec>), TryRecvError> { + // Check this receiver's cursor while holding `inner` before observing `senders`. Senders + // append messages under the same lock before they can be dropped, so an empty result here + // means this receiver has no unread buffered message. + let mut inner = self.shared.inner.lock(); + if let Some(received) = inner.receive(self.index) { + return Ok(received); + } + + if self.shared.senders.load(Ordering::Acquire) == 0 { + Err(TryRecvError::Disconnected) + } else { + Err(TryRecvError::Empty) + } + } + + /// Re-subscribes to the channel, returning a new receiver that starts receiving messages from + /// the *current* tail of the channel. + /// + /// This is useful if the receiver wants to jump to the latest message, skipping everything in + /// between. The original receiver is unchanged and continues to retain its own backlog until + /// it consumes those messages or is dropped. + /// + /// # Examples + /// + /// ``` + /// use mea::broadcast::unbounded; + /// + /// let (tx, mut rx) = unbounded::channel(); + /// tx.send(1); + /// tx.send(2); + /// + /// let mut rx2 = rx.resubscribe(); + /// tx.send(3); + /// + /// assert_eq!(rx2.try_recv(), Ok(3)); + /// ``` + pub fn resubscribe(&self) -> Self { + let mut inner = self.shared.inner.lock(); + let head = inner.tail; + let index = inner.insert_receiver(head); + let shared = self.shared.clone(); + Self { shared, index } + } + + /// Returns the number of messages this receiver can still read. + /// + /// This count is specific to this receiver, unlike [`Sender::buffer_len`], which reports the + /// shared backlog retained by the slowest active receiver. + /// + /// # Examples + /// + /// ``` + /// use mea::broadcast::unbounded; + /// + /// let (tx, mut rx) = unbounded::channel(); + /// assert_eq!(rx.len(), 0); + /// + /// tx.send(10); + /// tx.send(20); + /// assert_eq!(rx.len(), 2); + /// + /// assert_eq!(rx.try_recv(), Ok(10)); + /// assert_eq!(rx.len(), 1); + /// ``` + pub fn len(&self) -> usize { + let inner = self.shared.inner.lock(); + let head = inner.receivers[self.index]; + usize::try_from(inner.tail - head).expect("unread broadcast message count exceeds usize") + } + + /// Returns `true` if this receiver has no currently available messages. + /// + /// # Examples + /// + /// ``` + /// use mea::broadcast::unbounded; + /// + /// let (tx, rx) = unbounded::channel(); + /// assert!(rx.is_empty()); + /// + /// tx.send(10); + /// assert!(!rx.is_empty()); + /// ``` + pub fn is_empty(&self) -> bool { + self.len() == 0 + } +} + +struct Recv<'a, T> { + receiver: &'a mut Receiver, + index: Option, +} + +impl Drop for Recv<'_, T> { + fn drop(&mut self) { + // Ready paths clear the waiter ID, so only a cancelled pending receive takes this lock. + if self.index.is_none() { + return; + } + + let waker = { + let mut waiters = self.receiver.shared.waiters.lock(); + waiters.remove_waker(&mut self.index) + }; + drop(waker); + } +} + +impl Future for Recv<'_, T> { + type Output = Result; + + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + let Self { receiver, index } = self.get_mut(); + + match receiver.try_recv_shared() { + Ok((msg, reclaimed)) => { + *index = None; + drop(reclaimed); + return Poll::Ready(Ok((*msg).clone())); + } + Err(TryRecvError::Disconnected) => { + *index = None; + return Poll::Ready(Err(RecvError::Disconnected)); + } + Err(TryRecvError::Empty) => {} + } + + let received = { + let mut waiters = receiver.shared.waiters.lock(); + let mut inner = receiver.shared.inner.lock(); + + if let Some(received) = inner.receive(receiver.index) { + received + } else { + // A sender may have disconnected after the first `try_recv` returned `Empty`. + // Check again before registering the waker so `recv` does not miss the final wake. + if receiver.shared.senders.load(Ordering::Acquire) == 0 { + *index = None; + return Poll::Ready(Err(RecvError::Disconnected)); + } + + // Register Waker + let waker = waiters.register_waker(index, cx); + drop(waiters); + drop(inner); + drop(waker); + return Poll::Pending; + } + }; + + let (msg, reclaimed) = received; + *index = None; + drop(reclaimed); + Poll::Ready(Ok((*msg).clone())) + } +} diff --git a/mea/src/broadcast/unbounded/tests.rs b/mea/src/broadcast/unbounded/tests.rs new file mode 100644 index 0000000..62e31e4 --- /dev/null +++ b/mea/src/broadcast/unbounded/tests.rs @@ -0,0 +1,391 @@ +// Copyright 2024 tison +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::future::Future; +use std::sync::atomic::Ordering; +use std::task::Context; +use std::task::Poll; + +use super::*; +use crate::count_waker; + +#[tokio::test] +async fn test_broadcast_basic() { + let (tx, mut rx1) = channel(); + let mut rx2 = tx.subscribe(); + + tx.send(10); + tx.send(20); + + assert_eq!(rx1.recv().await, Ok(10)); + assert_eq!(rx1.recv().await, Ok(20)); + assert_eq!(rx2.recv().await, Ok(10)); + assert_eq!(rx2.recv().await, Ok(20)); +} + +#[tokio::test] +async fn test_broadcast_slow_receiver() { + let (tx, mut rx) = channel(); + + tx.send(1); + tx.send(2); + tx.send(3); + tx.send(4); + + assert_eq!(rx.recv().await, Ok(1)); + assert_eq!(rx.recv().await, Ok(2)); + assert_eq!(rx.recv().await, Ok(3)); + assert_eq!(rx.recv().await, Ok(4)); +} + +#[tokio::test] +async fn test_broadcast_closed() { + let (tx, mut rx) = channel::<()>(); + drop(tx); + assert_eq!(rx.recv().await, Err(RecvError::Disconnected)); +} + +#[tokio::test] +async fn test_broadcast_closed_after_buffered_messages() { + let (tx, mut rx) = channel(); + + tx.send(1); + tx.send(2); + drop(tx); + + assert_eq!(rx.recv().await, Ok(1)); + assert_eq!(rx.recv().await, Ok(2)); + assert_eq!(rx.recv().await, Err(RecvError::Disconnected)); +} + +#[test] +fn test_wait_mechanism() { + let (tx, mut rx) = channel(); + let (waker, wake_count) = count_waker(); + let mut cx = Context::from_waker(&waker); + let mut recv = Box::pin(rx.recv()); + + assert!(recv.as_mut().poll(&mut cx).is_pending()); + + tx.send(42); + + assert_eq!(wake_count.load(Ordering::Relaxed), 1); + assert_eq!(recv.as_mut().poll(&mut cx), Poll::Ready(Ok(42))); +} + +#[tokio::test] +async fn test_recv_cancellation_removes_waiter() { + let (tx, mut rx) = channel::(); + let (waker, wake_count) = count_waker(); + let mut cx = Context::from_waker(&waker); + let mut recv = Box::pin(rx.recv()); + + assert!(recv.as_mut().poll(&mut cx).is_pending()); + + drop(recv); + tx.send(1); + + assert_eq!(wake_count.load(Ordering::Relaxed), 0); + assert_eq!(rx.try_recv(), Ok(1)); +} + +#[tokio::test] +async fn test_dropped_woken_recv_does_not_remove_reused_waiter() { + let (tx, mut rx1) = channel::(); + let mut rx2 = tx.subscribe(); + let (waker1, wake_count1) = count_waker(); + let mut cx1 = Context::from_waker(&waker1); + let mut recv1 = Box::pin(rx1.recv()); + + assert!(recv1.as_mut().poll(&mut cx1).is_pending()); + + tx.send(1); + assert_eq!(wake_count1.load(Ordering::Relaxed), 1); + assert_eq!(rx2.try_recv(), Ok(1)); + + let (waker2, wake_count2) = count_waker(); + let mut cx2 = Context::from_waker(&waker2); + let mut recv2 = Box::pin(rx2.recv()); + assert!(recv2.as_mut().poll(&mut cx2).is_pending()); + + drop(recv1); + tx.send(2); + + assert_eq!(wake_count2.load(Ordering::Relaxed), 1); + drop(recv2); +} + +#[tokio::test] +async fn test_subscribe() { + let (tx, _rx) = channel(); + let mut rx = tx.subscribe(); + + tx.send(100); + assert_eq!(rx.recv().await, Ok(100)); +} + +#[tokio::test] +async fn test_receiver_count_and_len() { + let (tx, mut rx1) = channel(); + assert_eq!(tx.receiver_count(), 1); + assert_eq!(rx1.len(), 0); + assert!(rx1.is_empty()); + + tx.send(1); + tx.send(2); + assert_eq!(rx1.len(), 2); + assert!(!rx1.is_empty()); + + let mut rx2 = tx.subscribe(); + assert_eq!(tx.receiver_count(), 2); + assert_eq!(rx2.len(), 0); + assert!(rx2.is_empty()); + + tx.send(3); + assert_eq!(rx1.len(), 3); + assert_eq!(rx2.len(), 1); + + assert_eq!(rx2.try_recv(), Ok(3)); + assert_eq!(rx2.len(), 0); + drop(rx2); + assert_eq!(tx.receiver_count(), 1); + + assert_eq!(rx1.try_recv(), Ok(1)); + assert_eq!(rx1.len(), 2); +} + +#[tokio::test] +async fn test_resubscribe() { + let (tx, mut rx) = channel(); + + tx.send(1); + tx.send(2); + + let mut rx2 = rx.resubscribe(); + + // rx sees 1, 2 + // rx2 sees nothing yet (starts at tail=2) + + tx.send(3); + + assert_eq!(rx.recv().await, Ok(1)); + assert_eq!(rx.recv().await, Ok(2)); + assert_eq!(rx2.recv().await, Ok(3)); +} + +#[tokio::test] +async fn test_try_recv() { + let (tx, mut rx) = channel(); + + // Empty + assert_eq!(rx.try_recv(), Err(TryRecvError::Empty)); + + // Success + tx.send(10); + assert_eq!(rx.try_recv(), Ok(10)); + assert_eq!(rx.try_recv(), Err(TryRecvError::Empty)); + + // Closed + drop(tx); + assert_eq!(rx.try_recv(), Err(TryRecvError::Disconnected)); +} + +#[tokio::test] +async fn test_consumed_messages_are_reclaimed() { + let (tx, mut rx1) = channel(); + let mut rx2 = tx.subscribe(); + + tx.send(1); + tx.send(2); + assert_eq!(tx.buffer_len(), 2); + + assert_eq!(rx1.recv().await, Ok(1)); + assert_eq!(tx.buffer_len(), 2); + + assert_eq!(rx2.recv().await, Ok(1)); + assert_eq!(tx.buffer_len(), 1); + + assert_eq!(rx1.recv().await, Ok(2)); + assert_eq!(tx.buffer_len(), 1); + + assert_eq!(rx2.recv().await, Ok(2)); + assert_eq!(tx.buffer_len(), 0); +} + +#[tokio::test] +async fn test_drop_receiver_reclaims_messages() { + let (tx, mut rx1) = channel(); + let rx2 = tx.subscribe(); + + tx.send(1); + tx.send(2); + tx.send(3); + + assert_eq!(rx1.recv().await, Ok(1)); + assert_eq!(rx1.recv().await, Ok(2)); + assert_eq!(rx1.recv().await, Ok(3)); + assert_eq!(tx.buffer_len(), 3); + + drop(rx2); + assert_eq!(tx.buffer_len(), 0); +} + +#[tokio::test] +async fn test_buffer_len_tracks_shared_backlog() { + let (tx, mut rx1) = channel(); + let mut rx2 = tx.subscribe(); + + tx.send(1); + tx.send(2); + assert_eq!(tx.buffer_len(), 2); + + assert_eq!(rx1.recv().await, Ok(1)); + assert_eq!(rx1.recv().await, Ok(2)); + assert_eq!(tx.buffer_len(), 2); + + assert_eq!(rx2.recv().await, Ok(1)); + assert_eq!(tx.buffer_len(), 1); + assert_eq!(rx2.recv().await, Ok(2)); + assert_eq!(tx.buffer_len(), 0); +} + +#[tokio::test] +async fn test_resubscribe_keeps_original_receiver_backlog() { + let (tx, mut rx) = channel(); + + tx.send(1); + tx.send(2); + + let mut rx2 = rx.resubscribe(); + assert_eq!(tx.buffer_len(), 2); + + tx.send(3); + + assert_eq!(rx2.recv().await, Ok(3)); + assert_eq!(tx.buffer_len(), 3); + + assert_eq!(rx.recv().await, Ok(1)); + assert_eq!(rx.recv().await, Ok(2)); + assert_eq!(rx.recv().await, Ok(3)); + assert_eq!(tx.buffer_len(), 0); +} + +#[tokio::test] +#[should_panic(expected = "broadcast channel version counter overflowed")] +async fn test_send_panics_on_version_overflow() { + let (tx, _) = channel(); + tx.shared.inner.lock().tail = u64::MAX; + tx.send(()); +} + +#[tokio::test] +async fn test_send_without_receivers_does_not_buffer() { + let (tx, rx) = channel(); + drop(rx); + + tx.send(1); + tx.send(2); + assert_eq!(tx.buffer_len(), 0); + + let mut rx = tx.subscribe(); + assert_eq!(rx.try_recv(), Err(TryRecvError::Empty)); + + tx.send(3); + assert_eq!(rx.recv().await, Ok(3)); +} + +#[tokio::test] +async fn test_multi_senders_concurrent() { + let (tx, mut rx) = channel(); + let tx1 = tx.clone(); + let tx2 = tx.clone(); + + let handle1 = tokio::spawn(async move { + for i in 0..10 { + tx1.send(i); + } + }); + + let handle2 = tokio::spawn(async move { + for i in 10..20 { + tx2.send(i); + } + }); + + // Main tx can also send + for i in 20..30 { + tx.send(i); + } + + handle1.await.unwrap(); + handle2.await.unwrap(); + drop(tx); + + let mut received = Vec::new(); + while let Ok(n) = rx.recv().await { + received.push(n); + } + received.sort(); + + let expected = (0..30).collect::>(); + assert_eq!(received, expected); +} + +#[tokio::test] +async fn test_multi_senders_multiple_receivers_receive_all() { + let (tx, mut rx1) = channel(); + let mut rx2 = tx.subscribe(); + let mut rx3 = tx.subscribe(); + let tx1 = tx.clone(); + let tx2 = tx.clone(); + + let handle1 = tokio::spawn(async move { + for i in 0..10 { + tx1.send(i); + } + }); + let handle2 = tokio::spawn(async move { + for i in 10..20 { + tx2.send(i); + } + }); + + for i in 20..30 { + tx.send(i); + } + + handle1.await.unwrap(); + handle2.await.unwrap(); + drop(tx); + + let received1 = drain(&mut rx1).await; + let received2 = drain(&mut rx2).await; + let received3 = drain(&mut rx3).await; + + assert_eq!(received1, received2); + assert_eq!(received1, received3); + + let mut sorted = received1; + sorted.sort(); + let expected = (0..30).collect::>(); + assert_eq!(sorted, expected); +} + +async fn drain(rx: &mut Receiver) -> Vec { + let mut received = Vec::new(); + while let Ok(n) = rx.recv().await { + received.push(n); + } + received +} diff --git a/mea/src/internal/countdown.rs b/mea/src/internal/countdown.rs index 66ecd81..6c4c855 100644 --- a/mea/src/internal/countdown.rs +++ b/mea/src/internal/countdown.rs @@ -18,6 +18,7 @@ use std::task::Context; use crate::internal::Mutex; use crate::internal::WaitSet; +use crate::internal::WaiterId; #[derive(Debug)] pub(crate) struct CountdownState { @@ -56,17 +57,26 @@ impl CountdownState { /// Drain and wake up all waiters. pub(crate) fn wake_all(&self) { - let mut waiters = self.waiters.lock(); - waiters.wake_all(); + let wakers = { + let mut waiters = self.waiters.lock(); + waiters.take_wakers() + }; + + for waker in wakers { + waker.wake(); + } } /// Registers a waker to be woken up when the countdown reaches zero. /// - /// `idx` must be `None` when the waker is not registered, or `Some(key)` where `key` is - /// a value previously returned by this method. - pub(crate) fn register_waker(&self, idx: &mut Option, cx: &mut Context<'_>) { - let mut waiters = self.waiters.lock(); - waiters.register_waker(idx, cx); + /// `id` must be `None` when the waker is not registered, or `Some` with a waiter ID previously + /// stored by this method. + pub(crate) fn register_waker(&self, id: &mut Option, cx: &mut Context<'_>) { + let waker = { + let mut waiters = self.waiters.lock(); + waiters.register_waker(id, cx) + }; + drop(waker); } /// Returns `Ok(())` if the counter is zero, otherwise returns `Err(s)` where `s` is the current diff --git a/mea/src/internal/waitset.rs b/mea/src/internal/waitset.rs index d57096f..99fa330 100644 --- a/mea/src/internal/waitset.rs +++ b/mea/src/internal/waitset.rs @@ -17,9 +17,26 @@ use std::task::Waker; use slab::Slab; +/// Identifies a registered waker in a [`WaitSet`]. +/// +/// The generation distinguishes reused slab slots so stale waiter IDs cannot remove or update a +/// newer waiter that happens to occupy the same index. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) struct WaiterId { + index: usize, + generation: u64, +} + +#[derive(Debug)] +struct Waiter { + generation: u64, + waker: Waker, +} + #[derive(Debug)] pub(crate) struct WaitSet { - waiters: Slab, + waiters: Slab, + next_generation: u64, } impl WaitSet { @@ -27,6 +44,7 @@ impl WaitSet { pub const fn new() -> Self { Self { waiters: Slab::new(), + next_generation: 0, } } @@ -34,47 +52,176 @@ impl WaitSet { pub fn with_capacity(capacity: usize) -> Self { Self { waiters: Slab::with_capacity(capacity), + next_generation: 0, } } - /// Drain and wake up all waiters. - pub(crate) fn wake_all(&mut self) { - for w in self.waiters.drain() { - w.wake(); + /// Drains all wakers from the wait set. + /// + /// Callers should release any lock that protects the wait set before calling [`Waker::wake`] on + /// the returned wakers, to avoid running user-provided waker code with that lock held. + pub(crate) fn take_wakers(&mut self) -> Vec { + self.waiters.drain().map(|waiter| waiter.waker).collect() + } + + /// Removes a previously registered waker from the wait set, returning it if it was still + /// registered. + /// + /// The returned waker should be dropped after releasing any lock that protects the wait set, + /// so that user-provided `Waker::Drop` does not run with that lock held. + pub(crate) fn remove_waker(&mut self, id: &mut Option) -> Option { + let key = id.take()?; + let waiter = self.waiters.get(key.index)?; + + if waiter.generation == key.generation { + Some(self.waiters.remove(key.index).waker) + } else { + None } } /// Registers a waker to the wait set. /// - /// `idx` must be `None` when the waker is not registered, or `Some(key)` where `key` is - /// a value previously returned by this method. - pub(crate) fn register_waker(&mut self, idx: &mut Option, cx: &mut Context<'_>) { - match *idx { - None => { - let key = self.waiters.insert(cx.waker().clone()); - *idx = Some(key); - } - Some(key) => { - if self.waiters.contains(key) { - if !self.waiters[key].will_wake(cx.waker()) { - self.waiters[key] = cx.waker().clone(); + /// `id` must be `None` when the waker is not registered, or `Some` with a waiter ID previously + /// stored by this method. + /// + /// If a stored waker is replaced by a different one, the old waker is returned. The caller + /// should drop the returned waker after releasing any lock that protects the wait set, so + /// that user-provided `Waker::Drop` does not run with that lock held. + pub(crate) fn register_waker( + &mut self, + id: &mut Option, + cx: &mut Context<'_>, + ) -> Option { + if let Some(key) = *id { + if let Some(waiter) = self.waiters.get_mut(key.index) { + if waiter.generation == key.generation { + if !waiter.waker.will_wake(cx.waker()) { + return Some(std::mem::replace(&mut waiter.waker, cx.waker().clone())); } - } else { - // DEFENSIVE NOTE: - // - // This is possible if latch/waitgroup is fired between the first and second - // state check. - // - // In this case, it does not harm to re-register the waker. Because - // the second state check will finish the future and the WaitSet gets - // dropped. - // - // Barrier holds the lock during check and register, so the race condition - // above won't happen. - let key = self.waiters.insert(cx.waker().clone()); - *idx = Some(key); + return None; } } } + + // The stored WaiterId may be stale if the waiter was removed, drained by `take_wakers`, + // or if the slab slot has since been reused by another waiter. Register the current waker + // again and replace the stale ID. + *id = Some(self.insert_waker(cx.waker())); + None + } + + /// Allocates a fresh waiter ID and stores the waker in the wait set. + fn insert_waker(&mut self, waker: &Waker) -> WaiterId { + let generation = self.next_generation; + + // Do not wrap the generation counter: wrapping could make a stale WaiterId valid again + // after enough insertions. + self.next_generation = self + .next_generation + .checked_add(1) + .expect("wait set generation counter overflowed"); + + let index = self.waiters.insert(Waiter { + generation, + waker: waker.clone(), + }); + WaiterId { index, generation } + } +} + +#[cfg(test)] +mod tests { + use std::sync::Arc; + use std::sync::atomic::AtomicBool; + use std::sync::atomic::AtomicUsize; + use std::sync::atomic::Ordering; + use std::task::Context; + use std::task::Wake; + use std::task::Waker; + + use super::WaitSet; + + struct DropWake { + dropped: Arc, + wake_count: AtomicUsize, + } + + impl Wake for DropWake { + fn wake(self: Arc) { + self.wake_count.fetch_add(1, Ordering::Relaxed); + } + } + + impl Drop for DropWake { + fn drop(&mut self) { + self.dropped.store(true, Ordering::Relaxed); + } + } + + fn drop_waker() -> (Waker, Arc) { + let dropped = Arc::new(AtomicBool::new(false)); + let waker = Waker::from(Arc::new(DropWake { + dropped: dropped.clone(), + wake_count: AtomicUsize::new(0), + })); + (waker, dropped) + } + + #[test] + fn test_remove_waker_delays_drop() { + let mut waiters = WaitSet::new(); + let mut id = None; + let (waker, dropped) = drop_waker(); + let mut cx = Context::from_waker(&waker); + + assert!(waiters.register_waker(&mut id, &mut cx).is_none()); + drop(waker); + + let removed = waiters.remove_waker(&mut id).unwrap(); + assert!(!dropped.load(Ordering::Relaxed)); + + drop(removed); + assert!(dropped.load(Ordering::Relaxed)); + } + + #[test] + fn test_replace_waker_delays_drop() { + let mut waiters = WaitSet::new(); + let mut id = None; + let (waker, dropped) = drop_waker(); + let mut cx = Context::from_waker(&waker); + + assert!(waiters.register_waker(&mut id, &mut cx).is_none()); + drop(waker); + + let (replacement, _) = drop_waker(); + let mut cx = Context::from_waker(&replacement); + let replaced = waiters.register_waker(&mut id, &mut cx).unwrap(); + assert!(!dropped.load(Ordering::Relaxed)); + + drop(replaced); + assert!(dropped.load(Ordering::Relaxed)); + } + + #[test] + fn test_stale_waiter_id_does_not_remove_reused_slot() { + let mut waiters = WaitSet::new(); + let mut stale_id = None; + let (waker, _) = drop_waker(); + let mut cx = Context::from_waker(&waker); + + assert!(waiters.register_waker(&mut stale_id, &mut cx).is_none()); + let stale_id_value = stale_id.unwrap(); + drop(waiters.take_wakers()); + + let mut current_id = None; + assert!(waiters.register_waker(&mut current_id, &mut cx).is_none()); + let current_id_value = current_id.unwrap(); + assert_eq!(stale_id_value.index, current_id_value.index); + assert_ne!(stale_id_value.generation, current_id_value.generation); + + assert!(waiters.remove_waker(&mut stale_id).is_none()); + assert!(waiters.remove_waker(&mut current_id).is_some()); } } diff --git a/mea/src/latch/mod.rs b/mea/src/latch/mod.rs index d82b5ce..e644ed7 100644 --- a/mea/src/latch/mod.rs +++ b/mea/src/latch/mod.rs @@ -60,6 +60,7 @@ use std::task::Context; use std::task::Poll; use crate::internal::CountdownState; +use crate::internal::WaiterId; #[cfg(test)] mod tests; @@ -252,7 +253,7 @@ impl Latch { } impl Latch { - fn intern_poll(&self, idx: &mut Option, cx: &mut Context<'_>) -> Poll<()> { + fn intern_poll(&self, idx: &mut Option, cx: &mut Context<'_>) -> Poll<()> { // register waker if the counter is not zero if self.state.spin_wait(16).is_err() { self.state.register_waker(idx, cx); @@ -271,7 +272,7 @@ impl Latch { /// 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> { - idx: Option, + idx: Option, latch: &'a Latch, } @@ -295,7 +296,7 @@ impl Future for LatchWait<'_> { /// This future will complete when the latch count reaches zero. #[must_use = "futures do nothing unless you `.await` or poll them"] pub struct OwnedLatchWait { - idx: Option, + idx: Option, latch: Arc, } diff --git a/mea/src/lib.rs b/mea/src/lib.rs index 969c5d4..73f4b93 100644 --- a/mea/src/lib.rs +++ b/mea/src/lib.rs @@ -95,6 +95,38 @@ fn test_runtime() -> &'static tokio::runtime::Runtime { RT.get_or_init(|| Runtime::new().unwrap()) } +#[cfg(test)] +use std::sync::Arc; +#[cfg(test)] +use std::sync::atomic::AtomicUsize; +#[cfg(test)] +use std::sync::atomic::Ordering; +#[cfg(test)] +use std::task::Wake; +#[cfg(test)] +use std::task::Waker; + +#[cfg(test)] +struct CountWake(Arc); + +#[cfg(test)] +impl Wake for CountWake { + fn wake(self: Arc) { + self.0.fetch_add(1, Ordering::Relaxed); + } + + fn wake_by_ref(self: &Arc) { + self.0.fetch_add(1, Ordering::Relaxed); + } +} + +#[cfg(test)] +fn count_waker() -> (Waker, Arc) { + let wake_count = Arc::new(AtomicUsize::new(0)); + let waker = Waker::from(Arc::new(CountWake(wake_count.clone()))); + (waker, wake_count) +} + #[cfg(test)] mod tests { use crate::barrier::Barrier; @@ -141,6 +173,10 @@ mod tests { do_assert_send_and_sync::>(); do_assert_send_and_sync::(); do_assert_send_and_sync::(); + do_assert_send_and_sync::>(); + do_assert_send_and_sync::>(); + do_assert_send_and_sync::(); + do_assert_send_and_sync::(); do_assert_send_and_sync::>(); do_assert_send_and_sync::>(); do_assert_send_and_sync::>(); @@ -181,6 +217,10 @@ mod tests { do_assert_unpin::>(); do_assert_unpin::(); do_assert_unpin::(); + do_assert_unpin::>(); + do_assert_unpin::>(); + do_assert_unpin::(); + do_assert_unpin::(); do_assert_unpin::>(); do_assert_unpin::>(); do_assert_unpin::>(); diff --git a/mea/src/once/once/mod.rs b/mea/src/once/once/mod.rs index cb5efc3..76712a2 100644 --- a/mea/src/once/once/mod.rs +++ b/mea/src/once/once/mod.rs @@ -18,6 +18,7 @@ use std::task::Context; use std::task::Poll; use crate::internal::CountdownState; +use crate::internal::WaiterId; use crate::semaphore::Semaphore; #[cfg(test)] @@ -215,7 +216,7 @@ impl Once { } struct OnceWait<'a> { - idx: Option, + idx: Option, once: &'a Once, } diff --git a/mea/src/waitgroup/mod.rs b/mea/src/waitgroup/mod.rs index 66fc42a..9b273ef 100644 --- a/mea/src/waitgroup/mod.rs +++ b/mea/src/waitgroup/mod.rs @@ -59,6 +59,7 @@ use std::task::Context; use std::task::Poll; use crate::internal::CountdownState; +use crate::internal::WaiterId; #[cfg(test)] mod tests; @@ -145,7 +146,7 @@ impl IntoFuture for WaitGroup { /// will complete when the WaitGroup counter reaches zero. #[must_use = "futures do nothing unless you `.await` or poll them"] pub struct Wait { - idx: Option, + idx: Option, state: Arc, }