From 196ec8f3447e15e435a6fb5b51be8c8e50a272ce Mon Sep 17 00:00:00 2001 From: mxsm Date: Tue, 1 Sep 2026 21:22:32 +0800 Subject: [PATCH 1/2] feat(mpmc): add competing queues --- CHANGELOG.md | 4 + README.md | 1 + asyncband/Cargo.toml | 1 + asyncband/src/internal/mod.rs | 15 +- asyncband/src/internal/semaphore.rs | 35 +++ asyncband/src/lib.rs | 3 + asyncband/src/mpmc/bounded.rs | 140 +++++++++ asyncband/src/mpmc/error.rs | 138 +++++++++ asyncband/src/mpmc/mod.rs | 39 +++ asyncband/src/mpmc/queue.rs | 217 +++++++++++++ asyncband/src/mpmc/unbounded.rs | 127 ++++++++ benchmarks/Cargo.toml | 1 + benchmarks/asyncband/main.rs | 1 + benchmarks/asyncband/mpmc/bounded.rs | 49 +++ benchmarks/asyncband/mpmc/mod.rs | 20 ++ benchmarks/asyncband/mpmc/support.rs | 136 ++++++++ benchmarks/asyncband/mpmc/unbounded.rs | 48 +++ benchmarks/ecosystem/main.rs | 1 + benchmarks/ecosystem/mpmc/adapters.rs | 142 +++++++++ benchmarks/ecosystem/mpmc/bounded.rs | 42 +++ benchmarks/ecosystem/mpmc/mod.rs | 21 ++ benchmarks/ecosystem/mpmc/support.rs | 149 +++++++++ benchmarks/ecosystem/mpmc/unbounded.rs | 41 +++ tests-integration/Cargo.toml | 1 + tests-integration/tests/mpmc_test.rs | 414 +++++++++++++++++++++++++ tests-integration/tests/traits_test.rs | 23 ++ 26 files changed, 1806 insertions(+), 3 deletions(-) create mode 100644 asyncband/src/mpmc/bounded.rs create mode 100644 asyncband/src/mpmc/error.rs create mode 100644 asyncband/src/mpmc/mod.rs create mode 100644 asyncband/src/mpmc/queue.rs create mode 100644 asyncband/src/mpmc/unbounded.rs create mode 100644 benchmarks/asyncband/mpmc/bounded.rs create mode 100644 benchmarks/asyncband/mpmc/mod.rs create mode 100644 benchmarks/asyncband/mpmc/support.rs create mode 100644 benchmarks/asyncband/mpmc/unbounded.rs create mode 100644 benchmarks/ecosystem/mpmc/adapters.rs create mode 100644 benchmarks/ecosystem/mpmc/bounded.rs create mode 100644 benchmarks/ecosystem/mpmc/mod.rs create mode 100644 benchmarks/ecosystem/mpmc/support.rs create mode 100644 benchmarks/ecosystem/mpmc/unbounded.rs create mode 100644 tests-integration/tests/mpmc_test.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 43ffd8e0..1b9eea73 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,10 @@ All notable changes to this project will be documented in this file. ## Unreleased +### New features + +* Add opt-in bounded and unbounded `asyncband::mpmc` queues with cloneable producers and competing consumers, delivering each accepted value to exactly one receiver while a receiver remains. + ### Bug fixes * Release MPSC receiver wakers when the receiver is dropped, avoiding retained tasks and ownership cycles when a waker holds a sender. diff --git a/README.md b/README.md index 1203ae01..ccec0d05 100644 --- a/README.md +++ b/README.md @@ -80,6 +80,7 @@ Runnable examples live in the [`examples`](examples) workspace crate. They demon | | [`Group`](https://docs.rs/asyncband/*/asyncband/singleflight/struct.Group.html) | `singleflight` | Coalesce overlapping work per key without retaining completed values. | | Communication | [`Completion`](https://docs.rs/asyncband/*/asyncband/completion/struct.Completion.html) | `completion` | Publish one shared result to any number of current and future observers. | | | [`oneshot`](https://docs.rs/asyncband/*/asyncband/oneshot/) | `oneshot` | Send one value from one sender to one receiver. | +| | [`mpmc`](https://docs.rs/asyncband/*/asyncband/mpmc/) | `mpmc` | Distribute each value to exactly one of multiple competing receivers. | | | [`mpsc`](https://docs.rs/asyncband/*/asyncband/mpsc/) | `mpsc` | Send each value from multiple producers to one receiver with bounded backpressure or an unbounded queue. | | | [`broadcast`](https://docs.rs/asyncband/*/asyncband/broadcast/) | `broadcast` | Deliver every value to receivers active at send time; retain an unbounded backlog until each consumes or drops. | | | [`watch`](https://docs.rs/asyncband/*/asyncband/watch/) | `watch` | Publish cloneable latest state from one or more senders; receivers independently coalesce intermediate updates. | diff --git a/asyncband/Cargo.toml b/asyncband/Cargo.toml index d2d74f45..1b517dd8 100644 --- a/asyncband/Cargo.toml +++ b/asyncband/Cargo.toml @@ -52,6 +52,7 @@ condvar = ["mutex"] event = [] latch = [] lazy-cell = ["mutex"] +mpmc = [] mpsc = [] mutex = [] once = ["semaphore"] diff --git a/asyncband/src/internal/mod.rs b/asyncband/src/internal/mod.rs index 045a1c2b..3c9d5d8f 100644 --- a/asyncband/src/internal/mod.rs +++ b/asyncband/src/internal/mod.rs @@ -60,6 +60,7 @@ pub(crate) mod atomic_waker; feature = "event", feature = "completion", feature = "latch", + feature = "mpmc", feature = "mpsc", feature = "mutex", feature = "rwlock", @@ -90,6 +91,7 @@ pub(crate) mod value_cell; feature = "event", feature = "completion", feature = "latch", + feature = "mpmc", feature = "mpsc", feature = "mutex", feature = "rwlock", @@ -101,14 +103,20 @@ pub(crate) mod value_cell; #[allow(dead_code)] pub(crate) mod mutex; -#[cfg(any(feature = "mutex", feature = "rwlock", feature = "semaphore"))] -// Mutexes and rwlocks use the acquire/release operations; the public semaphore also exposes -// permit accounting. Each single-primitive build leaves part of this shared API unused. +#[cfg(any( + feature = "mpmc", + feature = "mutex", + feature = "rwlock", + feature = "semaphore", +))] +// MPMC uses waiter notifications; mutexes and rwlocks use acquire/release operations; the public +// semaphore also exposes permit accounting. Single-primitive builds leave part of this API unused. #[allow(dead_code)] pub(crate) mod semaphore; #[cfg(any( feature = "event", + feature = "mpmc", feature = "mpsc", feature = "mutex", feature = "rwlock", @@ -125,6 +133,7 @@ pub(crate) mod waitlist; feature = "event", feature = "completion", feature = "latch", + feature = "mpmc", feature = "mpsc", feature = "mutex", feature = "once", diff --git a/asyncband/src/internal/semaphore.rs b/asyncband/src/internal/semaphore.rs index 38283614..1b2525f0 100644 --- a/asyncband/src/internal/semaphore.rs +++ b/asyncband/src/internal/semaphore.rs @@ -205,6 +205,41 @@ impl Semaphore { } } + /// Adds `n` permits to the semaphore if there is any waiter. + #[cfg(feature = "mpmc")] + pub fn release_if_nonempty(&self, n: usize) { + let waiters = self.waiters.lock(); + if !waiters.is_empty() { + self.insert_permits_with_lock(n, waiters); + } + } + + /// Adds as many permits until there is no waiter. + #[cfg(feature = "mpmc")] + pub fn notify_all(&self) { + let mut waiters = self.waiters.lock(); + let mut wakers = vec![]; + loop { + match waiters.unlink_first_waiter(|node| { + node.permits = 0; + true + }) { + None => break, + Some((id, waiter)) => { + let remove_now = waiter.waker.is_none(); + if let Some(waker) = waiter.waker.take() { + wakers.push(waker); + } + if remove_now { + waiters.remove_unlinked_waiter(id); + } + } + } + } + drop(waiters); + crate::internal::wake_all(wakers.into_iter()); + } + fn insert_permits_with_lock( &self, mut rem: usize, diff --git a/asyncband/src/lib.rs b/asyncband/src/lib.rs index 93ce485e..8ffdb1f7 100644 --- a/asyncband/src/lib.rs +++ b/asyncband/src/lib.rs @@ -72,6 +72,7 @@ //! | | [`Group`](singleflight::Group) | `singleflight` | Coalesce overlapping work per key without retaining completed values. | //! | Communication | [`Completion`](completion::Completion) | `completion` | Publish one shared result to any number of current and future observers. | //! | | [`oneshot`] | `oneshot` | Send one value from one sender to one receiver. | +//! | | [`mpmc`] | `mpmc` | Distribute each value to exactly one of multiple competing receivers. | //! | | [`mpsc`] | `mpsc` | Send each value from multiple producers to one receiver with bounded backpressure or an unbounded queue. | //! | | [`broadcast`] | `broadcast` | Deliver every value to receivers active at send time; retain an unbounded backlog until each consumes or drops. | //! | | [`watch`] | `watch` | Publish cloneable latest state from one or more senders; receivers independently coalesce intermediate updates. | @@ -132,6 +133,8 @@ pub mod condvar; pub mod event; #[cfg(feature = "latch")] pub mod latch; +#[cfg(feature = "mpmc")] +pub mod mpmc; #[cfg(feature = "mpsc")] pub mod mpsc; #[cfg(feature = "mutex")] diff --git a/asyncband/src/mpmc/bounded.rs b/asyncband/src/mpmc/bounded.rs new file mode 100644 index 00000000..1a735a23 --- /dev/null +++ b/asyncband/src/mpmc/bounded.rs @@ -0,0 +1,140 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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::fmt; +use std::sync::Arc; + +use super::RecvError; +use super::SendError; +use super::TryRecvError; +use super::TrySendError; +use super::queue::Shared; + +/// Creates a bounded multi-producer, multi-consumer queue. +/// +/// The queue stores at most `capacity` values. Sending waits for a receiver to free capacity when +/// the queue is full. +/// +/// # Panics +/// +/// Panics if `capacity` is zero. +#[track_caller] +pub fn bounded(capacity: usize) -> (BoundedSender, BoundedReceiver) { + assert!(capacity > 0, "mpmc bounded queue requires capacity > 0"); + let shared = Arc::new(Shared::bounded(capacity)); + ( + BoundedSender { + shared: shared.clone(), + }, + BoundedReceiver { shared }, + ) +} + +/// Sends values to the associated [`BoundedReceiver`] handles. +/// +/// Instances are created by [`bounded`] and can be cloned to add producers. +pub struct BoundedSender { + shared: Arc>, +} + +impl Clone for BoundedSender { + fn clone(&self) -> Self { + self.shared.clone_sender(); + Self { + shared: self.shared.clone(), + } + } +} + +impl fmt::Debug for BoundedSender { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("BoundedSender").finish_non_exhaustive() + } +} + +impl Drop for BoundedSender { + fn drop(&mut self) { + self.shared.drop_sender(); + } +} + +impl BoundedSender { + /// Sends a value, waiting until capacity is available if the queue is full. + /// + /// If all receivers have been dropped, the value is returned in [`SendError`]. This method is + /// cancel safe: cancelling a pending send leaves its value with the future and passes any + /// selected capacity notification to the next sender. + pub async fn send(&self, value: T) -> Result<(), SendError> { + self.shared.send(value).await + } + + /// Attempts to send a value without waiting. + /// + /// Returns [`TrySendError::Full`] when the queue has reached its exact capacity and + /// [`TrySendError::Disconnected`] when all receivers have been dropped. + pub fn try_send(&self, value: T) -> Result<(), TrySendError> { + self.shared.try_send(value) + } +} + +/// Receives values from the associated [`BoundedSender`] handles. +/// +/// Cloned receivers compete for values, and every accepted value is returned by exactly one +/// receiver while a receiver remains. Dropping the final receiver releases buffered values. +pub struct BoundedReceiver { + shared: Arc>, +} + +impl Clone for BoundedReceiver { + fn clone(&self) -> Self { + self.shared.clone_receiver(); + Self { + shared: self.shared.clone(), + } + } +} + +impl fmt::Debug for BoundedReceiver { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("BoundedReceiver").finish_non_exhaustive() + } +} + +impl Drop for BoundedReceiver { + fn drop(&mut self) { + self.shared.drop_receiver(); + } +} + +impl BoundedReceiver { + /// Receives the next available value. + /// + /// Buffered values remain available after the final sender is dropped. Once they are drained, + /// this method returns [`RecvError::Disconnected`]. This method is cancel safe and passes a + /// selected value notification to another receiver if the pending future is cancelled. + pub async fn recv(&self) -> Result { + self.shared.recv().await + } + + /// Attempts to receive the next available value without waiting. + /// + /// Returns [`TryRecvError::Empty`] while the queue is empty and a sender remains, or + /// [`TryRecvError::Disconnected`] once the queue is empty and all senders have been dropped. + pub fn try_recv(&self) -> Result { + self.shared.try_recv() + } +} diff --git a/asyncband/src/mpmc/error.rs b/asyncband/src/mpmc/error.rs new file mode 100644 index 00000000..7bb7525e --- /dev/null +++ b/asyncband/src/mpmc/error.rs @@ -0,0 +1,138 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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::any::type_name; +use std::fmt; + +/// An error returned when trying to send on a disconnected queue. +/// +/// The value that could not be sent can be retrieved with [`SendError::into_inner`]. +#[derive(Clone, PartialEq, Eq)] +pub struct SendError(T); + +impl SendError { + /// Gets a reference to the value that failed to be sent. + pub fn as_inner(&self) -> &T { + &self.0 + } + + /// Consumes the error and returns the value that failed to be sent. + pub fn into_inner(self) -> T { + self.0 + } + + pub(super) fn new(value: T) -> Self { + Self(value) + } +} + +impl fmt::Display for SendError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str("sending on a disconnected queue") + } +} + +impl fmt::Debug for SendError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "SendError<{}>(..)", type_name::()) + } +} + +impl std::error::Error for SendError {} + +/// Error returned by [`BoundedSender::try_send`](crate::mpmc::BoundedSender::try_send). +#[derive(Clone, PartialEq, Eq)] +pub enum TrySendError { + /// The queue is full, so the value cannot be sent without waiting for capacity. + Full(T), + /// All receivers have been dropped, so the value can never be received. + Disconnected(T), +} + +impl TrySendError { + /// Gets a reference to the value that failed to be sent. + pub fn as_inner(&self) -> &T { + match self { + TrySendError::Full(value) | TrySendError::Disconnected(value) => value, + } + } + + /// Consumes the error and returns the value that failed to be sent. + pub fn into_inner(self) -> T { + match self { + TrySendError::Full(value) | TrySendError::Disconnected(value) => value, + } + } +} + +impl fmt::Display for TrySendError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(match self { + TrySendError::Full(_) => "sending on a full queue", + TrySendError::Disconnected(_) => "sending on a disconnected queue", + }) + } +} + +impl fmt::Debug for TrySendError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let ty = type_name::(); + match self { + TrySendError::Full(_) => write!(f, "TrySendError<{ty}>::Full(..)"), + TrySendError::Disconnected(_) => { + write!(f, "TrySendError<{ty}>::Disconnected(..)") + } + } + } +} + +impl std::error::Error for TrySendError {} + +/// Error returned by a receive operation. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum RecvError { + /// All senders have been dropped, and no buffered values remain. + Disconnected, +} + +impl fmt::Display for RecvError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str("receiving on a disconnected queue") + } +} + +impl std::error::Error for RecvError {} + +/// Error returned by a non-blocking receive operation. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum TryRecvError { + /// No value is currently available, but at least one sender remains. + Empty, + /// All senders have been dropped, and no buffered values remain. + Disconnected, +} + +impl fmt::Display for TryRecvError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(match self { + TryRecvError::Empty => "receiving on an empty queue", + TryRecvError::Disconnected => "receiving on a disconnected queue", + }) + } +} + +impl std::error::Error for TryRecvError {} diff --git a/asyncband/src/mpmc/mod.rs b/asyncband/src/mpmc/mod.rs new file mode 100644 index 00000000..8a3f0392 --- /dev/null +++ b/asyncband/src/mpmc/mod.rs @@ -0,0 +1,39 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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 queue for sending values between asynchronous tasks. +//! +//! Receivers compete for values: each value accepted by a sender is delivered to exactly one +//! receiver while a receiver remains. Clone a receiver to distribute work across multiple +//! asynchronous tasks. Dropping the final receiver releases any buffered values and makes later +//! sends return their value in an error. + +mod bounded; +mod error; +mod queue; +mod unbounded; + +pub use self::bounded::BoundedReceiver; +pub use self::bounded::BoundedSender; +pub use self::bounded::bounded; +pub use self::error::RecvError; +pub use self::error::SendError; +pub use self::error::TryRecvError; +pub use self::error::TrySendError; +pub use self::unbounded::UnboundedReceiver; +pub use self::unbounded::UnboundedSender; +pub use self::unbounded::unbounded; diff --git a/asyncband/src/mpmc/queue.rs b/asyncband/src/mpmc/queue.rs new file mode 100644 index 00000000..4b20c981 --- /dev/null +++ b/asyncband/src/mpmc/queue.rs @@ -0,0 +1,217 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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::collections::VecDeque; +use std::future::Future; +use std::future::poll_fn; +use std::pin::Pin; +use std::task::Context; +use std::task::Poll; + +use super::RecvError; +use super::SendError; +use super::TryRecvError; +use super::TrySendError; +use crate::internal::mutex::Mutex; +use crate::internal::semaphore::Acquire; +use crate::internal::semaphore::Semaphore; + +pub(super) struct Shared { + state: Mutex>, + recv_waiters: Semaphore, + send_waiters: Semaphore, + capacity: Option, +} + +struct State { + values: VecDeque, + senders: usize, + receivers: usize, +} + +impl Shared { + pub fn bounded(capacity: usize) -> Self { + Self::new(Some(capacity)) + } + + pub fn unbounded() -> Self { + Self::new(None) + } + + fn new(capacity: Option) -> Self { + Self { + state: Mutex::new(State { + values: VecDeque::new(), + senders: 1, + receivers: 1, + }), + recv_waiters: Semaphore::new(0), + send_waiters: Semaphore::new(0), + capacity, + } + } + + pub fn clone_sender(&self) { + let mut state = self.state.lock(); + state.senders = state + .senders + .checked_add(1) + .expect("mpmc sender count overflow"); + } + + pub fn drop_sender(&self) { + let is_last = { + let mut state = self.state.lock(); + state.senders -= 1; + state.senders == 0 + }; + if is_last { + self.recv_waiters.notify_all(); + } + } + + pub fn clone_receiver(&self) { + let mut state = self.state.lock(); + state.receivers = state + .receivers + .checked_add(1) + .expect("mpmc receiver count overflow"); + } + + pub fn drop_receiver(&self) { + let discarded = { + let mut state = self.state.lock(); + state.receivers -= 1; + (state.receivers == 0).then(|| std::mem::take(&mut state.values)) + }; + if discarded.is_some() { + self.send_waiters.notify_all(); + } + drop(discarded); + } + + pub fn try_send(&self, value: T) -> Result<(), TrySendError> { + { + let mut state = self.state.lock(); + if state.receivers == 0 { + return Err(TrySendError::Disconnected(value)); + } + if self + .capacity + .is_some_and(|capacity| state.values.len() >= capacity) + { + return Err(TrySendError::Full(value)); + } + state.values.push_back(value); + } + self.recv_waiters.release_if_nonempty(1); + Ok(()) + } + + pub async fn send(&self, value: T) -> Result<(), SendError> { + let value = match self.try_send(value) { + Ok(()) => return Ok(()), + Err(TrySendError::Disconnected(value)) => return Err(SendError::new(value)), + Err(TrySendError::Full(value)) => value, + }; + let mut send = Send { + shared: self, + value: Some(value), + acquire: self.send_waiters.poll_acquire(1), + }; + poll_fn(|cx| send.poll(cx)).await + } + + pub fn try_recv(&self) -> Result { + let value = { + let mut state = self.state.lock(); + match state.values.pop_front() { + Some(value) => value, + None if state.senders == 0 => return Err(TryRecvError::Disconnected), + None => return Err(TryRecvError::Empty), + } + }; + if self.capacity.is_some() { + self.send_waiters.release_if_nonempty(1); + } + Ok(value) + } + + pub async fn recv(&self) -> Result { + match self.try_recv() { + Ok(value) => return Ok(value), + Err(TryRecvError::Disconnected) => return Err(RecvError::Disconnected), + Err(TryRecvError::Empty) => {} + } + let mut recv = Recv { + shared: self, + acquire: self.recv_waiters.poll_acquire(1), + }; + poll_fn(|cx| recv.poll(cx)).await + } +} + +struct Send<'a, T> { + shared: &'a Shared, + value: Option, + acquire: Acquire<'a>, +} + +impl Send<'_, T> { + fn poll(&mut self, cx: &mut Context<'_>) -> Poll>> { + let mut value = self.value.take().expect("pending send must own its value"); + loop { + let notified = Pin::new(&mut self.acquire).poll(cx); + value = match self.shared.try_send(value) { + Ok(()) => return Poll::Ready(Ok(())), + Err(TrySendError::Disconnected(value)) => { + return Poll::Ready(Err(SendError::new(value))); + } + Err(TrySendError::Full(value)) => value, + }; + if notified.is_ready() { + self.acquire = self.shared.send_waiters.poll_acquire(1); + } else { + self.value = Some(value); + return Poll::Pending; + } + } + } +} + +struct Recv<'a, T> { + shared: &'a Shared, + acquire: Acquire<'a>, +} + +impl Recv<'_, T> { + fn poll(&mut self, cx: &mut Context<'_>) -> Poll> { + loop { + let notified = Pin::new(&mut self.acquire).poll(cx); + match self.shared.try_recv() { + Ok(value) => return Poll::Ready(Ok(value)), + Err(TryRecvError::Disconnected) => { + return Poll::Ready(Err(RecvError::Disconnected)); + } + Err(TryRecvError::Empty) if notified.is_ready() => { + self.acquire = self.shared.recv_waiters.poll_acquire(1); + } + Err(TryRecvError::Empty) => return Poll::Pending, + } + } + } +} diff --git a/asyncband/src/mpmc/unbounded.rs b/asyncband/src/mpmc/unbounded.rs new file mode 100644 index 00000000..479c1609 --- /dev/null +++ b/asyncband/src/mpmc/unbounded.rs @@ -0,0 +1,127 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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::fmt; +use std::sync::Arc; + +use super::RecvError; +use super::SendError; +use super::TryRecvError; +use super::TrySendError; +use super::queue::Shared; + +/// Creates an unbounded multi-producer, multi-consumer queue. +/// +/// Sends are synchronous and values may be buffered until available memory is exhausted. +pub fn unbounded() -> (UnboundedSender, UnboundedReceiver) { + let shared = Arc::new(Shared::unbounded()); + ( + UnboundedSender { + shared: shared.clone(), + }, + UnboundedReceiver { shared }, + ) +} + +/// Sends values to the associated [`UnboundedReceiver`] handles. +/// +/// Instances are created by [`unbounded`] and can be cloned to add producers. +pub struct UnboundedSender { + shared: Arc>, +} + +impl Clone for UnboundedSender { + fn clone(&self) -> Self { + self.shared.clone_sender(); + Self { + shared: self.shared.clone(), + } + } +} + +impl fmt::Debug for UnboundedSender { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("UnboundedSender").finish_non_exhaustive() + } +} + +impl Drop for UnboundedSender { + fn drop(&mut self) { + self.shared.drop_sender(); + } +} + +impl UnboundedSender { + /// Sends a value without waiting. + /// + /// If all receivers have been dropped, the value is returned in [`SendError`]. + pub fn send(&self, value: T) -> Result<(), SendError> { + match self.shared.try_send(value) { + Ok(()) => Ok(()), + Err(TrySendError::Disconnected(value)) => Err(SendError::new(value)), + Err(TrySendError::Full(_)) => unreachable!("unbounded queue cannot be full"), + } + } +} + +/// Receives values from the associated [`UnboundedSender`] handles. +/// +/// Cloned receivers compete for values, and every accepted value is returned by exactly one +/// receiver while a receiver remains. Dropping the final receiver releases buffered values. +pub struct UnboundedReceiver { + shared: Arc>, +} + +impl Clone for UnboundedReceiver { + fn clone(&self) -> Self { + self.shared.clone_receiver(); + Self { + shared: self.shared.clone(), + } + } +} + +impl fmt::Debug for UnboundedReceiver { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("UnboundedReceiver").finish_non_exhaustive() + } +} + +impl Drop for UnboundedReceiver { + fn drop(&mut self) { + self.shared.drop_receiver(); + } +} + +impl UnboundedReceiver { + /// Receives the next available value. + /// + /// Buffered values remain available after the final sender is dropped. Once they are drained, + /// this method returns [`RecvError::Disconnected`]. This method is cancel safe and passes a + /// selected value notification to another receiver if the pending future is cancelled. + pub async fn recv(&self) -> Result { + self.shared.recv().await + } + + /// Attempts to receive the next available value without waiting. + /// + /// Returns [`TryRecvError::Empty`] while the queue is empty and a sender remains, or + /// [`TryRecvError::Disconnected`] once the queue is empty and all senders have been dropped. + pub fn try_recv(&self) -> Result { + self.shared.try_recv() + } +} diff --git a/benchmarks/Cargo.toml b/benchmarks/Cargo.toml index 19e5ac33..6a54dba6 100644 --- a/benchmarks/Cargo.toml +++ b/benchmarks/Cargo.toml @@ -33,6 +33,7 @@ asyncband = { workspace = true, features = [ "condvar", "event", "latch", + "mpmc", "mpsc", "mutex", "once", diff --git a/benchmarks/asyncband/main.rs b/benchmarks/asyncband/main.rs index 53ae706a..1dd1877b 100644 --- a/benchmarks/asyncband/main.rs +++ b/benchmarks/asyncband/main.rs @@ -22,6 +22,7 @@ mod completion; mod condvar; mod event; mod latch; +mod mpmc; mod mpsc; mod mutex; mod once; diff --git a/benchmarks/asyncband/mpmc/bounded.rs b/benchmarks/asyncband/mpmc/bounded.rs new file mode 100644 index 00000000..ba2863be --- /dev/null +++ b/benchmarks/asyncband/mpmc/bounded.rs @@ -0,0 +1,49 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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 asyncband::mpmc; +use divan::Bencher; +use divan::counter::ItemsCount; + +use super::support::BATCH_MESSAGES; +use super::support::BOUNDED_CAPACITY; +use super::support::ConcurrentBatch; +use super::support::TOPOLOGIES; +use super::support::Topology; + +fn send(sender: &mpmc::BoundedSender, value: usize) { + pollster::block_on(sender.send(value)).expect("benchmark sender disconnected"); +} + +fn recv(receiver: &mpmc::BoundedReceiver) -> usize { + pollster::block_on(receiver.recv()).expect("benchmark receiver disconnected") +} + +#[divan::bench( + args = TOPOLOGIES, + sample_count = 20, + sample_size = 1, + counter = ItemsCount::new(BATCH_MESSAGES), +)] +fn concurrent(bencher: Bencher, topology: Topology) { + bencher + .with_inputs(|| { + let (sender, receiver) = mpmc::bounded(BOUNDED_CAPACITY); + ConcurrentBatch::new(sender, receiver, topology, send, recv) + }) + .bench_local_refs(|batch| batch.run()); +} diff --git a/benchmarks/asyncband/mpmc/mod.rs b/benchmarks/asyncband/mpmc/mod.rs new file mode 100644 index 00000000..551a5993 --- /dev/null +++ b/benchmarks/asyncband/mpmc/mod.rs @@ -0,0 +1,20 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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. + +mod bounded; +mod support; +mod unbounded; diff --git a/benchmarks/asyncband/mpmc/support.rs b/benchmarks/asyncband/mpmc/support.rs new file mode 100644 index 00000000..a8e43db4 --- /dev/null +++ b/benchmarks/asyncband/mpmc/support.rs @@ -0,0 +1,136 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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::sync::Arc; +use std::sync::Barrier; +use std::thread; +use std::thread::JoinHandle; + +use divan::black_box; + +pub const BATCH_MESSAGES: usize = 16_384; +pub const BOUNDED_CAPACITY: usize = 64; + +#[derive(Clone, Copy, Debug)] +pub struct Topology { + pub producers: usize, + pub consumers: usize, +} + +pub const TOPOLOGIES: &[Topology] = &[ + Topology { + producers: 1, + consumers: 1, + }, + Topology { + producers: 1, + consumers: 8, + }, + Topology { + producers: 8, + consumers: 1, + }, + Topology { + producers: 8, + consumers: 8, + }, +]; + +pub struct ConcurrentBatch { + start: Arc, + done: Arc, + workers: Vec>, +} + +impl ConcurrentBatch { + pub fn new( + sender: S, + receiver: R, + topology: Topology, + send: fn(&S, usize), + recv: fn(&R) -> usize, + ) -> Self + where + S: Clone + Send + 'static, + R: Clone + Send + 'static, + { + assert_eq!(BATCH_MESSAGES % topology.producers, 0); + assert_eq!(BATCH_MESSAGES % topology.consumers, 0); + + let participants = topology.producers + topology.consumers; + let start = Arc::new(Barrier::new(participants + 1)); + let done = Arc::new(Barrier::new(participants + 1)); + let messages_per_producer = BATCH_MESSAGES / topology.producers; + let messages_per_consumer = BATCH_MESSAGES / topology.consumers; + let mut workers = Vec::with_capacity(participants); + + for producer in 0..topology.producers { + let sender = sender.clone(); + let start = start.clone(); + let done = done.clone(); + workers.push(thread::spawn(move || { + start.wait(); + let first = producer * messages_per_producer; + for offset in 0..messages_per_producer { + send(&sender, black_box(first + offset)); + } + done.wait(); + })); + } + + for _ in 0..topology.consumers { + let receiver = receiver.clone(); + let start = start.clone(); + let done = done.clone(); + workers.push(thread::spawn(move || { + start.wait(); + let mut checksum = 0usize; + for _ in 0..messages_per_consumer { + checksum = checksum.wrapping_add(recv(&receiver)); + } + black_box(checksum); + done.wait(); + })); + } + + drop(sender); + drop(receiver); + + Self { + start, + done, + workers, + } + } + + pub fn run(&self) { + self.start.wait(); + self.done.wait(); + } +} + +impl Drop for ConcurrentBatch { + fn drop(&mut self) { + let panicking = thread::panicking(); + for worker in self.workers.drain(..) { + let result = worker.join(); + if !panicking { + result.expect("benchmark worker panicked"); + } + } + } +} diff --git a/benchmarks/asyncband/mpmc/unbounded.rs b/benchmarks/asyncband/mpmc/unbounded.rs new file mode 100644 index 00000000..940cc803 --- /dev/null +++ b/benchmarks/asyncband/mpmc/unbounded.rs @@ -0,0 +1,48 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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 asyncband::mpmc; +use divan::Bencher; +use divan::counter::ItemsCount; + +use super::support::BATCH_MESSAGES; +use super::support::ConcurrentBatch; +use super::support::TOPOLOGIES; +use super::support::Topology; + +fn send(sender: &mpmc::UnboundedSender, value: usize) { + sender.send(value).expect("benchmark sender disconnected"); +} + +fn recv(receiver: &mpmc::UnboundedReceiver) -> usize { + pollster::block_on(receiver.recv()).expect("benchmark receiver disconnected") +} + +#[divan::bench( + args = TOPOLOGIES, + sample_count = 20, + sample_size = 1, + counter = ItemsCount::new(BATCH_MESSAGES), +)] +fn concurrent(bencher: Bencher, topology: Topology) { + bencher + .with_inputs(|| { + let (sender, receiver) = mpmc::unbounded(); + ConcurrentBatch::new(sender, receiver, topology, send, recv) + }) + .bench_local_refs(|batch| batch.run()); +} diff --git a/benchmarks/ecosystem/main.rs b/benchmarks/ecosystem/main.rs index 0197cee3..f2faee4e 100644 --- a/benchmarks/ecosystem/main.rs +++ b/benchmarks/ecosystem/main.rs @@ -16,6 +16,7 @@ // under the License. mod broadcast; +mod mpmc; mod mpsc; mod waitgroup; mod watch; diff --git a/benchmarks/ecosystem/mpmc/adapters.rs b/benchmarks/ecosystem/mpmc/adapters.rs new file mode 100644 index 00000000..84f6c12a --- /dev/null +++ b/benchmarks/ecosystem/mpmc/adapters.rs @@ -0,0 +1,142 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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. + +pub struct Asyncband; +pub struct AsyncChannel; +pub struct Flume; + +pub trait BoundedMpmc: Send + Sync + 'static { + type Sender: Clone + Send + 'static; + type Receiver: Clone + Send + 'static; + + fn channel(capacity: usize) -> (Self::Sender, Self::Receiver); + fn send(sender: &Self::Sender, value: usize); + fn recv(receiver: &Self::Receiver) -> usize; +} + +pub trait UnboundedMpmc: Send + Sync + 'static { + type Sender: Clone + Send + 'static; + type Receiver: Clone + Send + 'static; + + fn channel() -> (Self::Sender, Self::Receiver); + fn send(sender: &Self::Sender, value: usize); + fn recv(receiver: &Self::Receiver) -> usize; +} + +impl BoundedMpmc for Asyncband { + type Receiver = asyncband::mpmc::BoundedReceiver; + type Sender = asyncband::mpmc::BoundedSender; + + fn channel(capacity: usize) -> (Self::Sender, Self::Receiver) { + asyncband::mpmc::bounded(capacity) + } + + fn send(sender: &Self::Sender, value: usize) { + pollster::block_on(sender.send(value)).expect("benchmark sender disconnected"); + } + + fn recv(receiver: &Self::Receiver) -> usize { + pollster::block_on(receiver.recv()).expect("benchmark receiver disconnected") + } +} + +impl BoundedMpmc for AsyncChannel { + type Receiver = async_channel::Receiver; + type Sender = async_channel::Sender; + + fn channel(capacity: usize) -> (Self::Sender, Self::Receiver) { + async_channel::bounded(capacity) + } + + fn send(sender: &Self::Sender, value: usize) { + pollster::block_on(sender.send(value)).expect("benchmark sender disconnected"); + } + + fn recv(receiver: &Self::Receiver) -> usize { + pollster::block_on(receiver.recv()).expect("benchmark receiver disconnected") + } +} + +impl BoundedMpmc for Flume { + type Receiver = flume::Receiver; + type Sender = flume::Sender; + + fn channel(capacity: usize) -> (Self::Sender, Self::Receiver) { + flume::bounded(capacity) + } + + fn send(sender: &Self::Sender, value: usize) { + pollster::block_on(sender.send_async(value)).expect("benchmark sender disconnected"); + } + + fn recv(receiver: &Self::Receiver) -> usize { + pollster::block_on(receiver.recv_async()).expect("benchmark receiver disconnected") + } +} + +impl UnboundedMpmc for Asyncband { + type Receiver = asyncband::mpmc::UnboundedReceiver; + type Sender = asyncband::mpmc::UnboundedSender; + + fn channel() -> (Self::Sender, Self::Receiver) { + asyncband::mpmc::unbounded() + } + + fn send(sender: &Self::Sender, value: usize) { + sender.send(value).expect("benchmark sender disconnected"); + } + + fn recv(receiver: &Self::Receiver) -> usize { + pollster::block_on(receiver.recv()).expect("benchmark receiver disconnected") + } +} + +impl UnboundedMpmc for AsyncChannel { + type Receiver = async_channel::Receiver; + type Sender = async_channel::Sender; + + fn channel() -> (Self::Sender, Self::Receiver) { + async_channel::unbounded() + } + + fn send(sender: &Self::Sender, value: usize) { + sender + .try_send(value) + .expect("benchmark sender disconnected"); + } + + fn recv(receiver: &Self::Receiver) -> usize { + pollster::block_on(receiver.recv()).expect("benchmark receiver disconnected") + } +} + +impl UnboundedMpmc for Flume { + type Receiver = flume::Receiver; + type Sender = flume::Sender; + + fn channel() -> (Self::Sender, Self::Receiver) { + flume::unbounded() + } + + fn send(sender: &Self::Sender, value: usize) { + sender.send(value).expect("benchmark sender disconnected"); + } + + fn recv(receiver: &Self::Receiver) -> usize { + pollster::block_on(receiver.recv_async()).expect("benchmark receiver disconnected") + } +} diff --git a/benchmarks/ecosystem/mpmc/bounded.rs b/benchmarks/ecosystem/mpmc/bounded.rs new file mode 100644 index 00000000..9daa6067 --- /dev/null +++ b/benchmarks/ecosystem/mpmc/bounded.rs @@ -0,0 +1,42 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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 divan::Bencher; +use divan::counter::ItemsCount; + +use super::adapters::AsyncChannel; +use super::adapters::Asyncband; +use super::adapters::BoundedMpmc; +use super::adapters::Flume; +use super::support::BATCH_MESSAGES; +use super::support::BOUNDED_CAPACITY; +use super::support::ConcurrentBatch; +use super::support::TOPOLOGIES; +use super::support::Topology; + +#[divan::bench( + types = [Asyncband, AsyncChannel, Flume], + args = TOPOLOGIES, + sample_count = 20, + sample_size = 1, + counter = ItemsCount::new(BATCH_MESSAGES), +)] +fn concurrent(bencher: Bencher, topology: Topology) { + bencher + .with_inputs(|| ConcurrentBatch::new_bounded::(BOUNDED_CAPACITY, topology)) + .bench_local_refs(|batch| batch.run()); +} diff --git a/benchmarks/ecosystem/mpmc/mod.rs b/benchmarks/ecosystem/mpmc/mod.rs new file mode 100644 index 00000000..dd09282b --- /dev/null +++ b/benchmarks/ecosystem/mpmc/mod.rs @@ -0,0 +1,21 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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. + +mod adapters; +mod bounded; +mod support; +mod unbounded; diff --git a/benchmarks/ecosystem/mpmc/support.rs b/benchmarks/ecosystem/mpmc/support.rs new file mode 100644 index 00000000..4955f291 --- /dev/null +++ b/benchmarks/ecosystem/mpmc/support.rs @@ -0,0 +1,149 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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::sync::Arc; +use std::sync::Barrier; +use std::thread; +use std::thread::JoinHandle; + +use divan::black_box; + +use super::adapters::BoundedMpmc; +use super::adapters::UnboundedMpmc; + +pub const BATCH_MESSAGES: usize = 16_384; +pub const BOUNDED_CAPACITY: usize = 64; + +#[derive(Clone, Copy, Debug)] +pub struct Topology { + pub producers: usize, + pub consumers: usize, +} + +pub const TOPOLOGIES: &[Topology] = &[ + Topology { + producers: 1, + consumers: 1, + }, + Topology { + producers: 1, + consumers: 8, + }, + Topology { + producers: 8, + consumers: 1, + }, + Topology { + producers: 8, + consumers: 8, + }, +]; + +pub struct ConcurrentBatch { + start: Arc, + done: Arc, + workers: Vec>, +} + +impl ConcurrentBatch { + pub fn new_bounded(capacity: usize, topology: Topology) -> Self { + let (sender, receiver) = C::channel(capacity); + Self::new(sender, receiver, topology, C::send, C::recv) + } + + pub fn new_unbounded(topology: Topology) -> Self { + let (sender, receiver) = C::channel(); + Self::new(sender, receiver, topology, C::send, C::recv) + } + + fn new( + sender: S, + receiver: R, + topology: Topology, + send: fn(&S, usize), + recv: fn(&R) -> usize, + ) -> Self + where + S: Clone + Send + 'static, + R: Clone + Send + 'static, + { + assert_eq!(BATCH_MESSAGES % topology.producers, 0); + assert_eq!(BATCH_MESSAGES % topology.consumers, 0); + + let participants = topology.producers + topology.consumers; + let start = Arc::new(Barrier::new(participants + 1)); + let done = Arc::new(Barrier::new(participants + 1)); + let messages_per_producer = BATCH_MESSAGES / topology.producers; + let messages_per_consumer = BATCH_MESSAGES / topology.consumers; + let mut workers = Vec::with_capacity(participants); + + for producer in 0..topology.producers { + let sender = sender.clone(); + let start = start.clone(); + let done = done.clone(); + workers.push(thread::spawn(move || { + start.wait(); + let first = producer * messages_per_producer; + for offset in 0..messages_per_producer { + send(&sender, black_box(first + offset)); + } + done.wait(); + })); + } + + for _ in 0..topology.consumers { + let receiver = receiver.clone(); + let start = start.clone(); + let done = done.clone(); + workers.push(thread::spawn(move || { + start.wait(); + let mut checksum = 0usize; + for _ in 0..messages_per_consumer { + checksum = checksum.wrapping_add(recv(&receiver)); + } + black_box(checksum); + done.wait(); + })); + } + + drop(sender); + drop(receiver); + + Self { + start, + done, + workers, + } + } + + pub fn run(&self) { + self.start.wait(); + self.done.wait(); + } +} + +impl Drop for ConcurrentBatch { + fn drop(&mut self) { + let panicking = thread::panicking(); + for worker in self.workers.drain(..) { + let result = worker.join(); + if !panicking { + result.expect("benchmark worker panicked"); + } + } + } +} diff --git a/benchmarks/ecosystem/mpmc/unbounded.rs b/benchmarks/ecosystem/mpmc/unbounded.rs new file mode 100644 index 00000000..4dfc7e11 --- /dev/null +++ b/benchmarks/ecosystem/mpmc/unbounded.rs @@ -0,0 +1,41 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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 divan::Bencher; +use divan::counter::ItemsCount; + +use super::adapters::AsyncChannel; +use super::adapters::Asyncband; +use super::adapters::Flume; +use super::adapters::UnboundedMpmc; +use super::support::BATCH_MESSAGES; +use super::support::ConcurrentBatch; +use super::support::TOPOLOGIES; +use super::support::Topology; + +#[divan::bench( + types = [Asyncband, AsyncChannel, Flume], + args = TOPOLOGIES, + sample_count = 20, + sample_size = 1, + counter = ItemsCount::new(BATCH_MESSAGES), +)] +fn concurrent(bencher: Bencher, topology: Topology) { + bencher + .with_inputs(|| ConcurrentBatch::new_unbounded::(topology)) + .bench_local_refs(|batch| batch.run()); +} diff --git a/tests-integration/Cargo.toml b/tests-integration/Cargo.toml index 04df5894..36deaf36 100644 --- a/tests-integration/Cargo.toml +++ b/tests-integration/Cargo.toml @@ -35,6 +35,7 @@ asyncband = { workspace = true, features = [ "event", "latch", "lazy-cell", + "mpmc", "mpsc", "mutex", "once", diff --git a/tests-integration/tests/mpmc_test.rs b/tests-integration/tests/mpmc_test.rs new file mode 100644 index 00000000..40f05969 --- /dev/null +++ b/tests-integration/tests/mpmc_test.rs @@ -0,0 +1,414 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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::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 std::task::Wake; +use std::task::Waker; +use std::time::Duration; + +use asyncband::mpmc; +use asyncband::mpmc::RecvError; +use asyncband::mpmc::TryRecvError; +use asyncband::mpmc::TrySendError; +use tests_integration::poll_once; + +struct WakeCounter(AtomicUsize); + +impl WakeCounter { + fn count(&self) -> usize { + self.0.load(Ordering::SeqCst) + } +} + +impl Wake for WakeCounter { + fn wake(self: Arc) { + self.0.fetch_add(1, Ordering::SeqCst); + } +} + +fn expect_ready(poll: Poll) -> T { + match poll { + Poll::Ready(value) => value, + Poll::Pending => panic!("future should be ready"), + } +} + +fn poll_with_waker(future: Pin<&mut F>, waker: &Waker) -> Poll { + future.poll(&mut Context::from_waker(waker)) +} + +#[test] +fn bounded_enforces_exact_capacity_and_fifo_order() { + let (sender, receiver) = mpmc::bounded(2); + let competing = receiver.clone(); + + sender.try_send(0).unwrap(); + sender.try_send(1).unwrap(); + assert_eq!(sender.try_send(2), Err(TrySendError::Full(2))); + + assert_eq!(receiver.try_recv(), Ok(0)); + assert_eq!(competing.try_recv(), Ok(1)); + assert_eq!(receiver.try_recv(), Err(TryRecvError::Empty)); +} + +#[test] +#[should_panic(expected = "mpmc bounded queue requires capacity > 0")] +fn bounded_rejects_zero_capacity() { + let _ = mpmc::bounded::<()>(0); +} + +#[test] +fn receiver_and_sender_clone_counts_control_disconnection() { + let (sender, receiver) = mpmc::unbounded(); + let sender_clone = sender.clone(); + let receiver_clone = receiver.clone(); + + drop(receiver); + sender.send(1).unwrap(); + assert_eq!(receiver_clone.try_recv(), Ok(1)); + + drop(sender); + assert_eq!(receiver_clone.try_recv(), Err(TryRecvError::Empty)); + drop(sender_clone); + assert_eq!(receiver_clone.try_recv(), Err(TryRecvError::Disconnected)); + + drop(receiver_clone); +} + +#[test] +fn last_receiver_returns_each_unsent_value_once() { + let (bounded_sender, bounded_receiver) = mpmc::bounded(1); + let bounded_receiver_clone = bounded_receiver.clone(); + drop(bounded_receiver); + drop(bounded_receiver_clone); + assert_eq!( + bounded_sender.try_send(1), + Err(TrySendError::Disconnected(1)) + ); + assert_eq!( + bounded_sender.try_send(2), + Err(TrySendError::Disconnected(2)) + ); + + let (unbounded_sender, unbounded_receiver) = mpmc::unbounded(); + drop(unbounded_receiver); + assert_eq!(unbounded_sender.send(3).unwrap_err().into_inner(), 3); +} + +#[tokio::test] +async fn buffered_values_drain_before_disconnection() { + let (sender, receiver) = mpmc::bounded(3); + sender.send(0).await.unwrap(); + sender.send(1).await.unwrap(); + sender.send(2).await.unwrap(); + drop(sender); + + assert_eq!(receiver.recv().await, Ok(0)); + assert_eq!(receiver.recv().await, Ok(1)); + assert_eq!(receiver.recv().await, Ok(2)); + assert_eq!(receiver.recv().await, Err(RecvError::Disconnected)); +} + +#[tokio::test] +async fn unbounded_preserves_fifo_order_and_drains_before_disconnection() { + let (sender, receiver) = mpmc::unbounded(); + sender.send(0).unwrap(); + sender.send(1).unwrap(); + sender.send(2).unwrap(); + drop(sender); + + assert_eq!(receiver.recv().await, Ok(0)); + assert_eq!(receiver.recv().await, Ok(1)); + assert_eq!(receiver.recv().await, Ok(2)); + assert_eq!(receiver.recv().await, Err(RecvError::Disconnected)); +} + +#[test] +fn bounded_send_wakes_only_the_first_receiver() { + let (sender, receiver) = mpmc::bounded(2); + let competing = receiver.clone(); + let mut first = Box::pin(receiver.recv()); + let mut second = Box::pin(competing.recv()); + let first_wakes = Arc::new(WakeCounter(AtomicUsize::new(0))); + let second_wakes = Arc::new(WakeCounter(AtomicUsize::new(0))); + let first_waker = Waker::from(first_wakes.clone()); + let second_waker = Waker::from(second_wakes.clone()); + + assert!(poll_with_waker(first.as_mut(), &first_waker).is_pending()); + assert!(poll_with_waker(second.as_mut(), &second_waker).is_pending()); + sender.try_send(1).unwrap(); + + assert_eq!(first_wakes.count(), 1); + assert_eq!(second_wakes.count(), 0); + assert_eq!( + expect_ready(poll_with_waker(first.as_mut(), &first_waker)), + Ok(1) + ); + assert!(poll_with_waker(second.as_mut(), &second_waker).is_pending()); + assert_eq!(second_wakes.count(), 0); +} + +#[test] +fn unbounded_send_wakes_only_the_first_receiver() { + let (sender, receiver) = mpmc::unbounded(); + let competing = receiver.clone(); + let mut first = Box::pin(receiver.recv()); + let mut second = Box::pin(competing.recv()); + let first_wakes = Arc::new(WakeCounter(AtomicUsize::new(0))); + let second_wakes = Arc::new(WakeCounter(AtomicUsize::new(0))); + let first_waker = Waker::from(first_wakes.clone()); + let second_waker = Waker::from(second_wakes.clone()); + + assert!(poll_with_waker(first.as_mut(), &first_waker).is_pending()); + assert!(poll_with_waker(second.as_mut(), &second_waker).is_pending()); + sender.send(1).unwrap(); + + assert_eq!(first_wakes.count(), 1); + assert_eq!(second_wakes.count(), 0); + assert_eq!( + expect_ready(poll_with_waker(first.as_mut(), &first_waker)), + Ok(1) + ); + assert!(poll_with_waker(second.as_mut(), &second_waker).is_pending()); + assert_eq!(second_wakes.count(), 0); +} + +#[test] +fn cancelled_notified_receiver_passes_value_to_next_receiver() { + let (sender, receiver) = mpmc::unbounded(); + let competing = receiver.clone(); + let mut cancelled = Box::pin(receiver.recv()); + let mut waiting = Box::pin(competing.recv()); + let cancelled_wakes = Arc::new(WakeCounter(AtomicUsize::new(0))); + let waiting_wakes = Arc::new(WakeCounter(AtomicUsize::new(0))); + let cancelled_waker = Waker::from(cancelled_wakes.clone()); + let waiting_waker = Waker::from(waiting_wakes.clone()); + + assert!(poll_with_waker(cancelled.as_mut(), &cancelled_waker).is_pending()); + assert!(poll_with_waker(waiting.as_mut(), &waiting_waker).is_pending()); + sender.send(1).unwrap(); + assert_eq!(cancelled_wakes.count(), 1); + assert_eq!(waiting_wakes.count(), 0); + drop(cancelled); + + assert_eq!(waiting_wakes.count(), 1); + assert_eq!( + expect_ready(poll_with_waker(waiting.as_mut(), &waiting_waker)), + Ok(1) + ); +} + +#[test] +fn bounded_cancelled_notified_receiver_passes_value_to_next_receiver() { + let (sender, receiver) = mpmc::bounded(1); + let competing = receiver.clone(); + let mut cancelled = Box::pin(receiver.recv()); + let mut waiting = Box::pin(competing.recv()); + let cancelled_wakes = Arc::new(WakeCounter(AtomicUsize::new(0))); + let waiting_wakes = Arc::new(WakeCounter(AtomicUsize::new(0))); + let cancelled_waker = Waker::from(cancelled_wakes.clone()); + let waiting_waker = Waker::from(waiting_wakes.clone()); + + assert!(poll_with_waker(cancelled.as_mut(), &cancelled_waker).is_pending()); + assert!(poll_with_waker(waiting.as_mut(), &waiting_waker).is_pending()); + sender.try_send(1).unwrap(); + assert_eq!(cancelled_wakes.count(), 1); + assert_eq!(waiting_wakes.count(), 0); + drop(cancelled); + + assert_eq!(waiting_wakes.count(), 1); + assert_eq!( + expect_ready(poll_with_waker(waiting.as_mut(), &waiting_waker)), + Ok(1) + ); +} + +#[test] +fn bounded_cancelled_notified_sender_passes_capacity_to_next_sender() { + let (sender, receiver) = mpmc::bounded(1); + sender.try_send(0).unwrap(); + let first_sender = sender.clone(); + let second_sender = sender.clone(); + let mut cancelled = Box::pin(first_sender.send(1)); + let mut waiting = Box::pin(second_sender.send(2)); + let cancelled_wakes = Arc::new(WakeCounter(AtomicUsize::new(0))); + let waiting_wakes = Arc::new(WakeCounter(AtomicUsize::new(0))); + let cancelled_waker = Waker::from(cancelled_wakes.clone()); + let waiting_waker = Waker::from(waiting_wakes.clone()); + + assert!(poll_with_waker(cancelled.as_mut(), &cancelled_waker).is_pending()); + assert!(poll_with_waker(waiting.as_mut(), &waiting_waker).is_pending()); + assert_eq!(receiver.try_recv(), Ok(0)); + assert_eq!(cancelled_wakes.count(), 1); + assert_eq!(waiting_wakes.count(), 0); + drop(cancelled); + + assert_eq!(waiting_wakes.count(), 1); + assert_eq!( + expect_ready(poll_with_waker(waiting.as_mut(), &waiting_waker)), + Ok(()) + ); + assert_eq!(receiver.try_recv(), Ok(2)); +} + +#[test] +fn last_endpoint_wakes_all_opposite_waiters() { + let (sender, receiver) = mpmc::bounded(1); + sender.try_send(0).unwrap(); + let sender_clone = sender.clone(); + let mut first_send = Box::pin(sender.send(1)); + let mut second_send = Box::pin(sender_clone.send(2)); + assert!(poll_once(first_send.as_mut()).is_pending()); + assert!(poll_once(second_send.as_mut()).is_pending()); + drop(receiver); + assert_eq!( + expect_ready(poll_once(first_send.as_mut())) + .unwrap_err() + .into_inner(), + 1 + ); + assert_eq!( + expect_ready(poll_once(second_send.as_mut())) + .unwrap_err() + .into_inner(), + 2 + ); + + let (sender, receiver) = mpmc::unbounded::(); + let competing = receiver.clone(); + let mut first_recv = Box::pin(receiver.recv()); + let mut second_recv = Box::pin(competing.recv()); + assert!(poll_once(first_recv.as_mut()).is_pending()); + assert!(poll_once(second_recv.as_mut()).is_pending()); + drop(sender); + assert_eq!( + expect_ready(poll_once(first_recv.as_mut())), + Err(RecvError::Disconnected) + ); + assert_eq!( + expect_ready(poll_once(second_recv.as_mut())), + Err(RecvError::Disconnected) + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn bounded_values_are_delivered_exactly_once_under_contention() { + const PRODUCERS: usize = 8; + const CONSUMERS: usize = 8; + const VALUES_PER_PRODUCER: usize = 512; + const TOTAL: usize = PRODUCERS * VALUES_PER_PRODUCER; + + let (sender, receiver) = mpmc::bounded(32); + let consumers = (0..CONSUMERS) + .map(|_| { + let receiver = receiver.clone(); + tokio::spawn(async move { + let mut values = Vec::new(); + while let Ok(value) = receiver.recv().await { + values.push(value); + } + values + }) + }) + .collect::>(); + drop(receiver); + + let producers = (0..PRODUCERS) + .map(|producer| { + let sender = sender.clone(); + tokio::spawn(async move { + let first = producer * VALUES_PER_PRODUCER; + for value in first..first + VALUES_PER_PRODUCER { + sender.send(value).await.unwrap(); + } + }) + }) + .collect::>(); + drop(sender); + + for producer in producers { + producer.await.unwrap(); + } + let mut received = Vec::with_capacity(TOTAL); + for consumer in consumers { + received.extend( + tokio::time::timeout(Duration::from_secs(10), consumer) + .await + .expect("bounded consumers must make progress") + .unwrap(), + ); + } + received.sort_unstable(); + assert_eq!(received, (0..TOTAL).collect::>()); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn unbounded_values_are_delivered_exactly_once_under_contention() { + const PRODUCERS: usize = 8; + const CONSUMERS: usize = 8; + const VALUES_PER_PRODUCER: usize = 512; + const TOTAL: usize = PRODUCERS * VALUES_PER_PRODUCER; + + let (sender, receiver) = mpmc::unbounded(); + let consumers = (0..CONSUMERS) + .map(|_| { + let receiver = receiver.clone(); + tokio::spawn(async move { + let mut values = Vec::new(); + while let Ok(value) = receiver.recv().await { + values.push(value); + } + values + }) + }) + .collect::>(); + drop(receiver); + + let producers = (0..PRODUCERS) + .map(|producer| { + let sender = sender.clone(); + tokio::spawn(async move { + let first = producer * VALUES_PER_PRODUCER; + for value in first..first + VALUES_PER_PRODUCER { + sender.send(value).unwrap(); + } + }) + }) + .collect::>(); + drop(sender); + + for producer in producers { + producer.await.unwrap(); + } + let mut received = Vec::with_capacity(TOTAL); + for consumer in consumers { + received.extend( + tokio::time::timeout(Duration::from_secs(10), consumer) + .await + .expect("unbounded consumers must make progress") + .unwrap(), + ); + } + received.sort_unstable(); + assert_eq!(received, (0..TOTAL).collect::>()); +} diff --git a/tests-integration/tests/traits_test.rs b/tests-integration/tests/traits_test.rs index dfbefb8c..a4607a2f 100644 --- a/tests-integration/tests/traits_test.rs +++ b/tests-integration/tests/traits_test.rs @@ -26,6 +26,7 @@ use asyncband::completion; use asyncband::condvar::Condvar; use asyncband::event::ManualResetEvent; use asyncband::latch::Latch; +use asyncband::mpmc; use asyncband::mpsc; use asyncband::mutex::Mutex; use asyncband::mutex::MutexGuard; @@ -113,6 +114,15 @@ fn public_types_are_send_and_sync() { assert_send_and_sync::>(); assert_send_and_sync::>(); assert_send_and_sync::>(); + assert_send_and_sync::>(); + assert_send_and_sync::>(); + assert_send_and_sync::>(); + assert_send_and_sync::>(); + assert_send_and_sync::>(); + assert_send_and_sync::>>(); + assert_send_and_sync::>>(); + assert_send_and_sync::>>(); + assert_send_and_sync::>>(); assert_send_and_sync::>(); assert_send_and_sync::>(); assert_send_and_sync::>(); @@ -135,6 +145,14 @@ fn movable_public_types_are_send() { let (_completer, completion) = completion::new::(); assert_send_value(completion.wait()); + + let (unbounded_sender, unbounded_receiver) = mpmc::unbounded::>(); + assert_send_value(unbounded_receiver.recv()); + drop(unbounded_sender); + + let (bounded_sender, bounded_receiver) = mpmc::bounded::>(1); + assert_send_value(bounded_sender.send(Cell::new(0))); + assert_send_value(bounded_receiver.recv()); } #[test] @@ -181,6 +199,11 @@ fn public_types_are_unpin() { assert_unpin::>(); assert_unpin::>(); assert_unpin::>(); + assert_unpin::>(); + assert_unpin::>(); + assert_unpin::>(); + assert_unpin::>(); + assert_unpin::>(); assert_unpin::>(); assert_unpin::>(); assert_unpin::>(); From 4b4c3346ab882714122f8917e575075a31820967 Mon Sep 17 00:00:00 2001 From: mxsm Date: Tue, 8 Sep 2026 07:57:42 -0700 Subject: [PATCH 2/2] fix(benchmarks): close MPMC senders before completion barrier --- benchmarks/asyncband/mpmc/support.rs | 3 + benchmarks/ecosystem/mpmc/support.rs | 3 + benchmarks/tests/mpmc_batch.rs | 111 +++++++++++++++++++++++++++ 3 files changed, 117 insertions(+) create mode 100644 benchmarks/tests/mpmc_batch.rs diff --git a/benchmarks/asyncband/mpmc/support.rs b/benchmarks/asyncband/mpmc/support.rs index a8e43db4..554fa1fe 100644 --- a/benchmarks/asyncband/mpmc/support.rs +++ b/benchmarks/asyncband/mpmc/support.rs @@ -88,6 +88,9 @@ impl ConcurrentBatch { for offset in 0..messages_per_producer { send(&sender, black_box(first + offset)); } + // Close the channel when production ends so pending receivers can finish + // draining it before all workers rendezvous at the completion barrier. + drop(sender); done.wait(); })); } diff --git a/benchmarks/ecosystem/mpmc/support.rs b/benchmarks/ecosystem/mpmc/support.rs index 4955f291..48404a70 100644 --- a/benchmarks/ecosystem/mpmc/support.rs +++ b/benchmarks/ecosystem/mpmc/support.rs @@ -101,6 +101,9 @@ impl ConcurrentBatch { for offset in 0..messages_per_producer { send(&sender, black_box(first + offset)); } + // Close the channel when production ends so pending receivers can finish + // draining it before all workers rendezvous at the completion barrier. + drop(sender); done.wait(); })); } diff --git a/benchmarks/tests/mpmc_batch.rs b/benchmarks/tests/mpmc_batch.rs new file mode 100644 index 00000000..2d855998 --- /dev/null +++ b/benchmarks/tests/mpmc_batch.rs @@ -0,0 +1,111 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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::sync::mpsc; +use std::thread; +use std::time::Duration; + +#[allow(dead_code)] +#[path = "../ecosystem/mpmc/adapters.rs"] +mod adapters; +#[allow(dead_code)] +#[path = "../asyncband/mpmc/support.rs"] +mod asyncband_support; +#[allow(dead_code)] +#[path = "../ecosystem/mpmc/support.rs"] +mod ecosystem_support; + +// Deliberately drain only after the last producer drops its sender. This makes +// retaining senders across the completion barrier deadlock deterministically. +struct DrainAfterClose; + +impl adapters::UnboundedMpmc for DrainAfterClose { + type Receiver = flume::Receiver; + type Sender = flume::Sender; + + fn channel() -> (Self::Sender, Self::Receiver) { + flume::unbounded() + } + + fn send(sender: &Self::Sender, value: usize) { + sender.send(value).unwrap(); + } + + fn recv(receiver: &Self::Receiver) -> usize { + while !receiver.is_disconnected() { + thread::sleep(Duration::from_millis(1)); + } + receiver.try_recv().unwrap() + } +} + +fn assert_completes(run: impl FnOnce() + Send + 'static) { + let (done, completion) = mpsc::channel(); + let worker = thread::spawn(move || { + run(); + done.send(()).unwrap(); + }); + completion + .recv_timeout(Duration::from_secs(30)) + .expect("benchmark batch did not finish after production ended"); + worker.join().unwrap(); +} + +#[test] +fn ecosystem_batch_closes_before_waiting_for_consumers() { + assert_completes(|| { + for &topology in ecosystem_support::TOPOLOGIES { + let batch = + ecosystem_support::ConcurrentBatch::new_unbounded::(topology); + batch.run(); + } + }); +} + +#[test] +fn asyncband_batch_closes_before_waiting_for_consumers() { + use adapters::UnboundedMpmc; + + assert_completes(|| { + for &topology in asyncband_support::TOPOLOGIES { + let (sender, receiver) = DrainAfterClose::channel(); + let batch = asyncband_support::ConcurrentBatch::new( + sender, + receiver, + topology, + DrainAfterClose::send, + DrainAfterClose::recv, + ); + batch.run(); + } + }); +} + +#[test] +fn flume_batches_complete_with_competing_consumers() { + assert_completes(|| { + for _ in 0..100 { + let batch = ecosystem_support::ConcurrentBatch::new_unbounded::( + ecosystem_support::Topology { + producers: 1, + consumers: 8, + }, + ); + batch.run(); + } + }); +}