diff --git a/CHANGELOG.md b/CHANGELOG.md index 66b5f57..4b7f69c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -32,3 +32,4 @@ All notable changes to this project will be documented in this file. ### Improvements * Remove the `slab` dependency in favor of a focused internal waiter arena. +* Replace the unbounded MPSC standard-library backend with a segmented single-consumer queue. diff --git a/Cargo.lock b/Cargo.lock index d293387..21f59e1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -97,6 +97,7 @@ dependencies = [ "async-broadcast", "async-channel", "asyncband", + "crossbeam-channel", "divan", "flume", "pollster", @@ -254,6 +255,15 @@ dependencies = [ "url", ] +[[package]] +name = "crossbeam-channel" +version = "0.5.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d85363c37faeca707aef026efa9f3b34d077bce547e48f770770625c6013679e" +dependencies = [ + "crossbeam-utils", +] + [[package]] name = "crossbeam-utils" version = "0.8.22" diff --git a/Cargo.toml b/Cargo.toml index 11086fb..07401bb 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -39,6 +39,7 @@ async-broadcast = { version = "0.7.2" } async-channel = { version = "2.5.0" } cargo_metadata = { version = "0.23.1" } clap = { version = "4.6.5" } +crossbeam-channel = { version = "0.5.16" } divan = { version = "0.1.21" } flume = { version = "0.12.0", default-features = false } pollster = { version = "1.0.1" } diff --git a/asyncband/src/channel/mpsc/unbounded.rs b/asyncband/src/channel/mpsc/unbounded.rs index 3ee892a..3e4ca2b 100644 --- a/asyncband/src/channel/mpsc/unbounded.rs +++ b/asyncband/src/channel/mpsc/unbounded.rs @@ -18,14 +18,21 @@ //! An unbounded multi-producer, single-consumer queue for sending values between asynchronous //! tasks. +mod queue; + use std::fmt; use std::future::poll_fn; 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::Poll; +use queue::Consumer; +use queue::Pop; +use queue::Queue; + use super::RecvError; use super::SendError; use super::TryRecvError; @@ -41,33 +48,47 @@ use crate::internal::atomic_waker::AtomicWaker; /// the channel. Using an `unbounded` channel has the ability of causing the /// process to run out of memory. In this case, the process will be aborted. pub fn unbounded() -> (UnboundedSender, UnboundedReceiver) { + let (queue, consumer) = Queue::new(); let state = Arc::new(UnboundedState { senders: AtomicUsize::new(1), + rx_waiting: AtomicBool::new(false), rx_waker: AtomicWaker::new(), + queue, }); - let (sender, receiver) = std::sync::mpsc::channel(); let sender = UnboundedSender { state: state.clone(), - sender: Some(sender), - }; - let receiver = UnboundedReceiver { - state: state.clone(), - receiver, }; + let receiver = UnboundedReceiver { state, consumer }; (sender, receiver) } -struct UnboundedState { +struct UnboundedState { senders: AtomicUsize, + rx_waiting: AtomicBool, rx_waker: AtomicWaker, + queue: Queue, +} + +impl UnboundedState { + fn wake_receiver(&self) { + // Keep the common active-receiver path read-only. If the receiver is parked, exactly one + // producer claims this registration and performs the wake. + if self.rx_waiting.load(Ordering::SeqCst) + && self + .rx_waiting + .compare_exchange(true, false, Ordering::SeqCst, Ordering::SeqCst) + .is_ok() + { + self.rx_waker.wake(); + } + } } /// Send values to the associated [`UnboundedReceiver`]. /// /// Instances are created by the [`unbounded`] function. pub struct UnboundedSender { - state: Arc, - sender: Option>, + state: Arc>, } impl Clone for UnboundedSender { @@ -75,7 +96,6 @@ impl Clone for UnboundedSender { self.state.senders.fetch_add(1, Ordering::Release); UnboundedSender { state: self.state.clone(), - sender: self.sender.clone(), } } } @@ -88,14 +108,11 @@ impl fmt::Debug for UnboundedSender { impl Drop for UnboundedSender { fn drop(&mut self) { - // drop the sender; this closes the channel if it is the last sender - drop(self.sender.take()); - match self.state.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. - self.state.rx_waker.wake(); + self.state.wake_receiver(); } _ => { // there are still other senders left, do nothing @@ -114,11 +131,9 @@ impl UnboundedSender { /// If the receiver has been dropped, this function returns an error. The error includes /// the value passed to `send`. pub fn send(&self, value: T) -> Result<(), SendError> { - // SAFETY: The sender is guaranteed to be non-null before dropped. - let sender = self.sender.as_ref().unwrap(); - sender.send(value).map_err(|err| SendError::new(err.0))?; + self.state.queue.push(value).map_err(SendError::new)?; - self.state.rx_waker.wake(); + self.state.wake_receiver(); Ok(()) } @@ -128,13 +143,16 @@ impl UnboundedSender { /// /// Instances are created by the [`unbounded`] function. pub struct UnboundedReceiver { - state: Arc, - receiver: std::sync::mpsc::Receiver, + state: Arc>, + consumer: Consumer, } -/// The only `!Sync` field `receiver` is protected by `&mut self` in `recv` and `try_recv`. -/// That is, `UnboundedReceiver` can only be accessed by one thread at a time. -unsafe impl Sync for UnboundedReceiver {} +impl Drop for UnboundedReceiver { + fn drop(&mut self) { + self.state.rx_waiting.store(false, Ordering::SeqCst); + self.state.queue.close(&mut self.consumer); + } +} impl fmt::Debug for UnboundedReceiver { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { @@ -177,10 +195,21 @@ impl UnboundedReceiver { /// # } /// ``` pub fn try_recv(&mut self) -> Result { - match self.receiver.try_recv() { - Ok(v) => Ok(v), - Err(std::sync::mpsc::TryRecvError::Disconnected) => Err(TryRecvError::Disconnected), - Err(std::sync::mpsc::TryRecvError::Empty) => Err(TryRecvError::Empty), + match self.consumer.pop(&self.state.queue) { + Pop::Value(value) => return Ok(value), + Pop::Empty | Pop::Pending => {} + } + + if self.state.senders.load(Ordering::Acquire) == 0 { + // The last sender publishes its queue write before decrementing this count. Rechecking + // after acquiring zero drains that final value before reporting disconnection. + match self.consumer.pop(&self.state.queue) { + Pop::Value(value) => Ok(value), + Pop::Empty => Err(TryRecvError::Disconnected), + Pop::Pending => unreachable!("the final sender cannot leave a pending queue slot"), + } + } else { + Err(TryRecvError::Empty) } } @@ -243,9 +272,20 @@ impl UnboundedReceiver { Err(TryRecvError::Empty) => { self.state.rx_waker.register(cx.waker()); + // The queue reservation, slot publication, and notification gate are sequentially + // consistent: either this recheck observes a completed send, or that sender sees + // the armed gate and wakes this task after publishing its slot. + self.state.rx_waiting.store(true, Ordering::SeqCst); + match self.try_recv() { - Ok(v) => Poll::Ready(Ok(v)), - Err(TryRecvError::Disconnected) => Poll::Ready(Err(RecvError::Disconnected)), + Ok(v) => { + self.state.rx_waiting.store(false, Ordering::SeqCst); + Poll::Ready(Ok(v)) + } + Err(TryRecvError::Disconnected) => { + self.state.rx_waiting.store(false, Ordering::SeqCst); + Poll::Ready(Err(RecvError::Disconnected)) + } Err(TryRecvError::Empty) => Poll::Pending, } } diff --git a/asyncband/src/channel/mpsc/unbounded/queue.rs b/asyncband/src/channel/mpsc/unbounded/queue.rs new file mode 100644 index 0000000..8410fb6 --- /dev/null +++ b/asyncband/src/channel/mpsc/unbounded/queue.rs @@ -0,0 +1,449 @@ +// 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. + +// Portions are adapted from crossbeam-channel, copyright (c) 2019 The Crossbeam Project +// Developers, and used under the Apache License, Version 2.0. + +//! Segmented storage for the unbounded MPSC channel. +//! +//! The producer-side layout is adapted from the list flavor in `crossbeam-channel`, licensed +//! under Apache-2.0 OR MIT. Unlike that MPMC implementation, this queue has one consumer, so the +//! consumer owns its position and a block can be reclaimed as soon as its final slot is read. + +use std::cell::UnsafeCell; +use std::hint; +use std::marker::PhantomData; +use std::mem::MaybeUninit; +use std::ptr; +use std::sync::atomic::AtomicBool; +use std::sync::atomic::AtomicPtr; +use std::sync::atomic::AtomicUsize; +use std::sync::atomic::Ordering; +use std::thread; + +const LAP: usize = 32; +const BLOCK_CAPACITY: usize = LAP - 1; +const SHIFT: usize = 1; +const STEP: usize = 1 << SHIFT; +const CLOSED: usize = 1; + +#[repr(align(128))] +struct CachePadded(T); + +struct Slot { + value: UnsafeCell>, + ready: AtomicBool, +} + +impl Slot { + fn new() -> Self { + Self { + value: UnsafeCell::new(MaybeUninit::uninit()), + ready: AtomicBool::new(false), + } + } +} + +struct Block { + next: AtomicPtr>, + slots: [Slot; BLOCK_CAPACITY], +} + +impl Block { + fn new() -> Box { + Box::new(Self { + next: AtomicPtr::new(ptr::null_mut()), + slots: std::array::from_fn(|_| Slot::new()), + }) + } +} + +struct Position { + index: AtomicUsize, + block: AtomicPtr>, +} + +/// Producer-owned half of an unbounded MPSC queue. +pub struct Queue { + tail: CachePadded>, + _marker: PhantomData, +} + +// SAFETY: each producer reserves a distinct slot through `tail.index`. A value placed in a slot is +// only accessed by the single consumer after the producer publishes `ready` with release semantics. +unsafe impl Send for Queue {} +// SAFETY: the producer algorithm coordinates all shared mutation through atomics. The consumer is +// separate and may only read a slot after acquiring its `ready` publication. +unsafe impl Sync for Queue {} + +/// Consumer-owned position in an unbounded MPSC queue. +pub struct Consumer { + index: usize, + block: *mut Block, + _marker: PhantomData, +} + +// SAFETY: moving the unique consumer moves exclusive ownership of its position. Queue values cross +// the thread boundary only when `T: Send`. +unsafe impl Send for Consumer {} +// SAFETY: shared references cannot pop or close the consumer because both operations require +// exclusive access. Sharing an idle consumer therefore exposes neither its position nor `T`. +unsafe impl Sync for Consumer {} + +/// Result of attempting to pop one queue slot. +pub enum Pop { + /// A published value was removed. + Value(T), + /// No producer has reserved the next slot. + Empty, + /// A producer reserved the next slot but has not published its value yet. + Pending, +} + +impl Queue { + /// Creates the producer queue and its unique consumer position. + pub fn new() -> (Self, Consumer) { + let block = Box::into_raw(Block::new()); + let queue = Self { + tail: CachePadded(Position { + index: AtomicUsize::new(0), + block: AtomicPtr::new(block), + }), + _marker: PhantomData, + }; + let consumer = Consumer { + index: 0, + block, + _marker: PhantomData, + }; + (queue, consumer) + } + + /// Appends a value, or returns it if the consumer has closed the queue. + pub fn push(&self, value: T) -> Result<(), T> { + let mut backoff = Backoff::new(); + let mut tail = self.tail.0.index.load(Ordering::Acquire); + let mut block = self.tail.0.block.load(Ordering::Acquire); + let mut next_block = None; + + loop { + if tail & CLOSED != 0 { + return Err(value); + } + + let offset = (tail >> SHIFT) % LAP; + if offset == BLOCK_CAPACITY { + backoff.snooze(); + tail = self.tail.0.index.load(Ordering::Acquire); + block = self.tail.0.block.load(Ordering::Acquire); + continue; + } + + // The producer that reserves a block's final slot also installs its successor. Doing + // the allocation before the reservation keeps the boundary transition short. + if offset + 1 == BLOCK_CAPACITY && next_block.is_none() { + next_block = Some(Block::new()); + } + + let new_tail = tail.wrapping_add(STEP); + match self.tail.0.index.compare_exchange_weak( + tail, + new_tail, + Ordering::SeqCst, + Ordering::Acquire, + ) { + Ok(_) => { + // SAFETY: a successful reservation gives this producer exclusive ownership of + // `block.slots[offset]`. The receiver cannot reclaim the block until this slot + // publishes `ready` and is read in FIFO order. + unsafe { + if offset + 1 == BLOCK_CAPACITY { + let next = Box::into_raw(next_block.take().unwrap()); + (*block).next.store(next, Ordering::Release); + self.tail.0.block.store(next, Ordering::Release); + + // The reserved sentinel index keeps other producers and close cleanup + // from entering the next block before both pointers are installed. + self.tail.0.index.fetch_add(STEP, Ordering::Release); + } + + let slot = (*block).slots.get_unchecked(offset); + (*slot.value.get()).write(value); + // This publication joins the notification gate's sequentially consistent + // order. If a producer checks the gate before the receiver arms it, the + // receiver's subsequent recheck must observe this completed publication. + slot.ready.store(true, Ordering::SeqCst); + } + return Ok(()); + } + Err(observed) => { + tail = observed; + block = self.tail.0.block.load(Ordering::Acquire); + backoff.spin(); + } + } + } + } + + /// Closes the producer side, waits for already-reserved slots, and discards buffered values. + pub fn close(&self, consumer: &mut Consumer) { + let mut backoff = Backoff::new(); + let mut tail = self.tail.0.index.fetch_or(CLOSED, Ordering::SeqCst) | CLOSED; + + // A producer at the sentinel owns the block transition. It reserved before close and must + // finish installing the next block before cleanup can traverse it. + while (tail >> SHIFT) % LAP == BLOCK_CAPACITY { + backoff.snooze(); + tail = self.tail.0.index.load(Ordering::Acquire); + } + + let mut cleanup = Cleanup { + queue: self, + consumer, + complete: false, + }; + cleanup.drain(); + cleanup.complete = true; + } +} + +impl Consumer { + /// Attempts to remove the next value without waiting for a producer to finish publishing it. + pub fn pop(&mut self, queue: &Queue) -> Pop { + let tail = queue.tail.0.index.load(Ordering::SeqCst); + if self.index >> SHIFT == tail >> SHIFT { + return Pop::Empty; + } + + let offset = (self.index >> SHIFT) % LAP; + debug_assert!(offset < BLOCK_CAPACITY); + + // SAFETY: `block` is exclusively owned by this consumer position. A non-empty queue means + // the corresponding producer reserved this slot; acquiring `ready` publishes its value. + unsafe { + let slot = (*self.block).slots.get_unchecked(offset); + if !slot.ready.load(Ordering::SeqCst) { + return Pop::Pending; + } + + let value = (*slot.value.get()).assume_init_read(); + let new_index = self.index.wrapping_add(STEP); + + if offset + 1 == BLOCK_CAPACITY { + let old = self.block; + let next = (*old).next.load(Ordering::Acquire); + debug_assert!(!next.is_null()); + self.block = next; + self.index = new_index.wrapping_add(STEP); + + // Observing the last slot's publication also observes the producer's earlier next + // pointer publication. FIFO consumption means every producer using `old` has + // finished publishing before the consumer reaches this point. + drop(Box::from_raw(old)); + } else { + self.index = new_index; + } + + Pop::Value(value) + } + } + + fn finish(&mut self, queue: &Queue) { + debug_assert!(!self.block.is_null()); + + // SAFETY: close prevents new reservations, and cleanup reaches this point only after all + // reserved values have been read. The current empty block is therefore exclusively owned. + unsafe { drop(Box::from_raw(self.block)) }; + self.block = ptr::null_mut(); + queue.tail.0.block.store(ptr::null_mut(), Ordering::Release); + } +} + +struct Cleanup<'a, T> { + queue: &'a Queue, + consumer: &'a mut Consumer, + complete: bool, +} + +impl Cleanup<'_, T> { + fn drain(&mut self) { + let mut backoff = Backoff::new(); + loop { + match self.consumer.pop(self.queue) { + Pop::Value(value) => { + backoff.reset(); + drop(value); + } + Pop::Pending => backoff.snooze(), + Pop::Empty => { + self.consumer.finish(self.queue); + return; + } + } + } + } +} + +impl Drop for Cleanup<'_, T> { + fn drop(&mut self) { + if !self.complete { + // Continue reclaiming if dropping a buffered value unwinds. A second destructor panic + // follows Rust's usual double-panic behavior and aborts the process. + self.drain(); + } + } +} + +struct Backoff { + step: u32, +} + +impl Backoff { + fn new() -> Self { + Self { step: 0 } + } + + fn spin(&mut self) { + let iterations = 1 << self.step.min(6); + for _ in 0..iterations { + hint::spin_loop(); + } + self.step = self.step.saturating_add(1); + } + + fn snooze(&mut self) { + if self.step <= 6 { + self.spin(); + } else { + thread::yield_now(); + self.step = self.step.saturating_add(1); + } + } + + fn reset(&mut self) { + self.step = 0; + } +} + +#[cfg(test)] +mod tests { + use std::sync::Arc; + use std::sync::Barrier; + use std::sync::atomic::AtomicUsize; + use std::sync::atomic::Ordering; + use std::thread; + + use super::Pop; + use super::Queue; + + #[test] + fn crosses_block_boundaries_in_fifo_order() { + let (queue, mut consumer) = Queue::new(); + + for value in 0..1_000 { + queue.push(value).unwrap(); + } + for expected in 0..1_000 { + match consumer.pop(&queue) { + Pop::Value(value) => assert_eq!(value, expected), + Pop::Empty | Pop::Pending => panic!("reserved value should be ready"), + } + } + assert!(matches!(consumer.pop(&queue), Pop::Empty)); + + queue.close(&mut consumer); + } + + #[test] + fn concurrent_producers_preserve_per_producer_order() { + const PRODUCERS: usize = 4; + const VALUES: usize = 128; + + let (queue, mut consumer) = Queue::new(); + let queue = Arc::new(queue); + let start = Arc::new(Barrier::new(PRODUCERS + 1)); + let workers = (0..PRODUCERS) + .map(|producer| { + let queue = queue.clone(); + let start = start.clone(); + thread::spawn(move || { + start.wait(); + for sequence in 0..VALUES { + queue.push((producer, sequence)).unwrap(); + } + }) + }) + .collect::>(); + + start.wait(); + let mut next = [0; PRODUCERS]; + while next.iter().sum::() < PRODUCERS * VALUES { + match consumer.pop(&queue) { + Pop::Value((producer, sequence)) => { + assert_eq!(sequence, next[producer]); + next[producer] += 1; + } + Pop::Empty | Pop::Pending => thread::yield_now(), + } + } + + for worker in workers { + worker.join().unwrap(); + } + queue.close(&mut consumer); + } + + #[test] + fn close_drops_or_returns_every_racing_value_once() { + const PRODUCERS: usize = 4; + const VALUES: usize = 128; + + struct DropProbe(Arc); + + impl Drop for DropProbe { + fn drop(&mut self) { + self.0.fetch_add(1, Ordering::Relaxed); + } + } + + let (queue, mut consumer) = Queue::new(); + let queue = Arc::new(queue); + let dropped = Arc::new(AtomicUsize::new(0)); + let start = Arc::new(Barrier::new(PRODUCERS + 1)); + let workers = (0..PRODUCERS) + .map(|_| { + let queue = queue.clone(); + let dropped = dropped.clone(); + let start = start.clone(); + thread::spawn(move || { + start.wait(); + for _ in 0..VALUES { + drop(queue.push(DropProbe(dropped.clone()))); + } + }) + }) + .collect::>(); + + start.wait(); + queue.close(&mut consumer); + for worker in workers { + worker.join().unwrap(); + } + + assert_eq!(dropped.load(Ordering::Relaxed), PRODUCERS * VALUES); + } +} diff --git a/benchmarks/Cargo.toml b/benchmarks/Cargo.toml index 15a724f..196897a 100644 --- a/benchmarks/Cargo.toml +++ b/benchmarks/Cargo.toml @@ -44,10 +44,11 @@ asyncband = { workspace = true, features = [ "singleflight", "waitgroup", ] } +crossbeam-channel = { workspace = true } divan = { workspace = true } flume = { workspace = true, features = ["async"] } pollster = { workspace = true } -tokio = { workspace = true, features = ["sync"] } +tokio = { workspace = true, features = ["rt-multi-thread", "sync"] } [[bench]] harness = false diff --git a/benchmarks/ecosystem/mpsc/adapters.rs b/benchmarks/ecosystem/mpsc/adapters.rs index 7e78c21..b065b58 100644 --- a/benchmarks/ecosystem/mpsc/adapters.rs +++ b/benchmarks/ecosystem/mpsc/adapters.rs @@ -15,6 +15,7 @@ // specific language governing permissions and limitations // under the License. +use std::future::Future; use std::task::Context; use crate::support::poll_ready; @@ -22,6 +23,7 @@ use crate::support::poll_ready; pub struct Asyncband; pub struct Tokio; pub struct AsyncChannel; +pub struct Crossbeam; pub struct Flume; pub trait BoundedMpsc: Send + Sync + 'static { @@ -48,6 +50,10 @@ pub trait UnboundedMpsc: Send + Sync + 'static { fn recv_blocking(receiver: &mut Self::Receiver) -> usize; } +pub trait AsyncUnboundedMpsc: UnboundedMpsc { + fn recv_async(receiver: &mut Self::Receiver) -> impl Future + Send + '_; +} + impl BoundedMpsc for Asyncband { type Receiver = asyncband::mpsc::BoundedReceiver; type Sender = asyncband::mpsc::BoundedSender; @@ -205,6 +211,12 @@ impl UnboundedMpsc for Asyncband { } } +impl AsyncUnboundedMpsc for Asyncband { + async fn recv_async(receiver: &mut Self::Receiver) -> usize { + receiver.recv().await.unwrap() + } +} + impl UnboundedMpsc for Tokio { type Receiver = tokio::sync::mpsc::UnboundedReceiver; type Sender = tokio::sync::mpsc::UnboundedSender; @@ -230,6 +242,12 @@ impl UnboundedMpsc for Tokio { } } +impl AsyncUnboundedMpsc for Tokio { + async fn recv_async(receiver: &mut Self::Receiver) -> usize { + receiver.recv().await.unwrap() + } +} + impl UnboundedMpsc for AsyncChannel { type Receiver = async_channel::Receiver; type Sender = async_channel::Sender; @@ -255,6 +273,37 @@ impl UnboundedMpsc for AsyncChannel { } } +impl AsyncUnboundedMpsc for AsyncChannel { + async fn recv_async(receiver: &mut Self::Receiver) -> usize { + receiver.recv().await.unwrap() + } +} + +impl UnboundedMpsc for Crossbeam { + type Receiver = crossbeam_channel::Receiver; + type Sender = crossbeam_channel::Sender; + + fn channel() -> (Self::Sender, Self::Receiver) { + crossbeam_channel::unbounded() + } + + fn send(sender: &Self::Sender, value: usize) { + sender.send(value).unwrap(); + } + + fn try_recv(receiver: &mut Self::Receiver) -> usize { + receiver.try_recv().unwrap() + } + + fn recv_ready(receiver: &mut Self::Receiver, _context: &mut Context<'_>) -> usize { + receiver.try_recv().unwrap() + } + + fn recv_blocking(receiver: &mut Self::Receiver) -> usize { + receiver.recv().unwrap() + } +} + impl UnboundedMpsc for Flume { type Receiver = flume::Receiver; type Sender = flume::Sender; @@ -279,3 +328,9 @@ impl UnboundedMpsc for Flume { pollster::block_on(receiver.recv_async()).unwrap() } } + +impl AsyncUnboundedMpsc for Flume { + async fn recv_async(receiver: &mut Self::Receiver) -> usize { + receiver.recv_async().await.unwrap() + } +} diff --git a/benchmarks/ecosystem/mpsc/support.rs b/benchmarks/ecosystem/mpsc/support.rs index aea2885..8a08c28 100644 --- a/benchmarks/ecosystem/mpsc/support.rs +++ b/benchmarks/ecosystem/mpsc/support.rs @@ -28,7 +28,7 @@ use super::adapters::UnboundedMpsc; pub const BOUNDED_CAPACITY: usize = 64; pub const BATCH_MESSAGES: usize = 16_384; -pub const PRODUCER_COUNTS: &[usize] = &[1, 2, 4, 8]; +pub const PRODUCER_COUNTS: &[usize] = &[1, 2, 4, 8, 16, 32]; pub trait ConcurrentMpsc: Send + Sync + 'static { type Sender: Clone + Send + 'static; diff --git a/benchmarks/ecosystem/mpsc/unbounded.rs b/benchmarks/ecosystem/mpsc/unbounded.rs index 6c8ac54..99b9c0a 100644 --- a/benchmarks/ecosystem/mpsc/unbounded.rs +++ b/benchmarks/ecosystem/mpsc/unbounded.rs @@ -20,7 +20,9 @@ use divan::black_box; use divan::counter::ItemsCount; use super::adapters::AsyncChannel; +use super::adapters::AsyncUnboundedMpsc; use super::adapters::Asyncband; +use super::adapters::Crossbeam; use super::adapters::Flume; use super::adapters::Tokio; use super::adapters::UnboundedMpsc; @@ -30,7 +32,25 @@ use super::support::PRODUCER_COUNTS; use super::support::Unbounded; use crate::support::bench_context; -#[divan::bench(types = [Asyncband, Tokio, AsyncChannel, Flume])] +const SEQUENTIAL_MESSAGES: usize = 5_000; +const ASYNC_PRODUCERS: usize = 5; +const ASYNC_MESSAGES_PER_PRODUCER: usize = 1_000; + +#[divan::bench(types = [Asyncband, Tokio, AsyncChannel, Crossbeam, Flume])] +fn create(bencher: Bencher) { + bencher.bench_local(|| black_box(C::channel())); +} + +#[divan::bench(types = [Asyncband, Tokio, AsyncChannel, Crossbeam, Flume])] +fn oneshot(bencher: Bencher) { + bencher.bench_local(|| { + let (sender, mut receiver) = C::channel(); + C::send(&sender, black_box(usize::MAX)); + black_box(C::try_recv(&mut receiver)) + }); +} + +#[divan::bench(types = [Asyncband, Tokio, AsyncChannel, Crossbeam, Flume])] fn ready_round_trip(bencher: Bencher) { let mut context = bench_context(); let (sender, mut receiver) = C::channel(); @@ -41,7 +61,7 @@ fn ready_round_trip(bencher: Bencher) { }); } -#[divan::bench(types = [Asyncband, Tokio, AsyncChannel, Flume])] +#[divan::bench(types = [Asyncband, Tokio, AsyncChannel, Crossbeam, Flume])] fn try_round_trip(bencher: Bencher) { let (sender, mut receiver) = C::channel(); @@ -52,7 +72,26 @@ fn try_round_trip(bencher: Bencher) { } #[divan::bench( - types = [Asyncband, Tokio, AsyncChannel, Flume], + types = [Asyncband, Tokio, AsyncChannel, Crossbeam, Flume], + sample_count = 100, + sample_size = 1, + counter = ItemsCount::new(SEQUENTIAL_MESSAGES), +)] +fn sequential(bencher: Bencher) { + bencher.bench_local(|| { + let (sender, mut receiver) = C::channel(); + + for value in 0..SEQUENTIAL_MESSAGES { + C::send(&sender, black_box(value)); + } + for _ in 0..SEQUENTIAL_MESSAGES { + black_box(C::try_recv(&mut receiver)); + } + }); +} + +#[divan::bench( + types = [Asyncband, Tokio, AsyncChannel, Crossbeam, Flume], args = PRODUCER_COUNTS, sample_count = 20, sample_size = 1, @@ -63,3 +102,43 @@ fn concurrent(bencher: Bencher, producer_count: usize) { .with_inputs(|| ConcurrentBatch::>::new(producer_count)) .bench_local_refs(|batch| batch.run()); } + +#[divan::bench( + types = [Asyncband, Tokio, AsyncChannel, Flume], + sample_count = 20, + sample_size = 1, + counter = ItemsCount::new(ASYNC_PRODUCERS * ASYNC_MESSAGES_PER_PRODUCER), +)] +fn async_contention(bencher: Bencher) { + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(6) + .build() + .unwrap(); + + bencher.bench_local(|| { + runtime.block_on(async { + let (sender, mut receiver) = C::channel(); + let workers = (0..ASYNC_PRODUCERS) + .map(|producer| { + let sender = sender.clone(); + tokio::spawn(async move { + let first = producer * ASYNC_MESSAGES_PER_PRODUCER; + for offset in 0..ASYNC_MESSAGES_PER_PRODUCER { + C::send(&sender, black_box(first + offset)); + } + }) + }) + .collect::>(); + drop(sender); + + let mut checksum = 0usize; + for _ in 0..ASYNC_PRODUCERS * ASYNC_MESSAGES_PER_PRODUCER { + checksum = checksum.wrapping_add(C::recv_async(&mut receiver).await); + } + for worker in workers { + worker.await.unwrap(); + } + black_box(checksum) + }) + }); +}