perf(mpmc): back bounded queues with a ring buffer

Replace the bounded queue's VecDeque behind one mutex with a preallocated ring of stamped slots adapted from crossbeam-channel's array flavor. Sends, receives, and reservations take no lock unless a sender or receiver has to wait, while wait-queue order, exact capacity, permits, cancellation, and disconnection behave as before. Storage for the full capacity is allocated when the queue is created.

The capacity that permits and grants hold, and a bit marking queued senders, live in the tail cursor beside the position, so every exchange on it compares the reservation state it was computed from. Positions use 64 bits wherever 64-bit atomics exist, as in crossbeam-channel. While senders are queued, a send with a permit grants the capacity that receivers free until it finishes, so a single receiver does not have to wake every queued sender itself.

Copy CachePadded from crossbeam-utils and record the incorporated crossbeam portions in LICENSE, licenserc.toml, and .asf.yaml.

Author: ariesdevil

.asf.yaml

@@ -41,6 +41,7 @@ project:
     source_excludes_lightweight:
       - "/apache-asyncband-*-incubating-src.tar.gz/*/asyncband/src/blocking/executor.rs"
       - "/apache-asyncband-*-incubating-src.tar.gz/*/asyncband/src/blocking/parker.rs"
+      - "/apache-asyncband-*-incubating-src.tar.gz/*/asyncband/src/internal/cache_padded.rs"
       - "/apache-asyncband-*-incubating-src.tar.gz/*/asyncband/src/pool/bounded.rs"
       - "/apache-asyncband-*-incubating-src.tar.gz/*/asyncband/src/pool/common.rs"
       - "/apache-asyncband-*-incubating-src.tar.gz/*/asyncband/src/pool/mod.rs"

CHANGELOG.md

@@ -40,6 +40,7 @@ All notable changes to this project will be documented in this file.
 
 ### Improvements
 
+* Bounded MPMC sends, receives, and reservations no longer take a lock unless a sender or receiver is waiting, which speeds up contended workloads on threads and async runtimes. Creating a bounded queue now allocates storage for its full capacity, and handoffs where every send waits for a receiver, such as through a capacity-1 queue, are slightly slower.
 * Improve `FutureExt::block_on` throughput and reduce context switches when many threads repeatedly block on the same primitive, such as a contended MPMC channel; an isolated cross-thread wake-up costs slightly more.
 * Reduce `ManualResetEvent` state-query and already-set wait overhead, especially with concurrent readers.
 * Add a lock-free fast path for releasing `Semaphore` permits while permits remain available, which speeds up dropping `RwLock` read guards, releasing multi-permit `Semaphore` permits, and returning objects to a bounded pool that is not at capacity. `Mutex` releases are unchanged.

LICENSE

@@ -416,6 +416,22 @@ the following copyright notice:
 
   Copyright 2014-2020 The Rust Project Developers
 
+Portions of asyncband/src/internal/cache_padded.rs are adapted from
+crossbeam-utils 0.8.23. Portions of the slot-stamp algorithm and backoff helper
+in asyncband/src/mpmc/ring.rs are adapted from crossbeam-channel 0.5.17 and
+crossbeam-utils 0.8.23. Both are taken from the following exact revision and
+source paths:
+
+  https://github.com/crossbeam-rs/crossbeam/blob/2920c984290229ab4e0ca0452ef09e48a82063f3/crossbeam-utils/src/cache_padded.rs
+  https://github.com/crossbeam-rs/crossbeam/blob/2920c984290229ab4e0ca0452ef09e48a82063f3/crossbeam-utils/src/backoff.rs
+  https://github.com/crossbeam-rs/crossbeam/blob/2920c984290229ab4e0ca0452ef09e48a82063f3/crossbeam-channel/src/flavors/array.rs
+
+Crossbeam is licensed under Apache-2.0 or MIT. Apache Asyncband uses the
+Apache-2.0 option for the incorporated portions. The upstream project carries
+the following copyright notice:
+
+  Copyright (c) 2019 The Crossbeam Project Developers
+
 Portions of the following files are ported from Fastpool 1.1.1. Each local path
 is followed by its upstream source path or paths:
 

asyncband/src/internal/cache_padded.rs

@@ -0,0 +1,181 @@
+// 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.
+
+// This file contains code adapted from crossbeam-utils 0.8.23.
+// Copyright (c) 2019 The Crossbeam Project Developers
+// Asyncband uses the upstream project's Apache-2.0 license option for the incorporated code.
+// The incorporated code has been modified for use in Apache Asyncband: examples, conversions, and
+// explicit Send and Sync implementations are omitted, and imports follow Asyncband's style.
+// Upstream source:
+// https://github.com/crossbeam-rs/crossbeam/blob/2920c984290229ab4e0ca0452ef09e48a82063f3/crossbeam-utils/src/cache_padded.rs
+
+use std::fmt;
+use std::ops::Deref;
+use std::ops::DerefMut;
+
+/// Pads and aligns a value to the length of a cache line.
+///
+/// In concurrent programming, sometimes it is desirable to make sure commonly accessed pieces of
+/// data are not placed into the same cache line. Updating an atomic value invalidates the whole
+/// cache line it belongs to, which makes the next access to the same cache line slower for other
+/// CPU cores. Use `CachePadded` to ensure updating one piece of data doesn't invalidate other
+/// cached data.
+///
+/// # Size and alignment
+///
+/// Cache lines are assumed to be N bytes long, depending on the architecture:
+///
+/// * On x86-64, aarch64, and powerpc64, N = 128.
+/// * On arm, mips, mips64, sparc, and hexagon, N = 32.
+/// * On m68k, N = 16.
+/// * On s390x, N = 256.
+/// * On all others, N = 64.
+///
+/// Note that N is just a reasonable guess and is not guaranteed to match the actual cache line
+/// length of the machine the program is running on. On modern Intel architectures, spatial
+/// prefetcher is pulling pairs of 64-byte cache lines at a time, so we pessimistically assume that
+/// cache lines are 128 bytes long.
+///
+/// The size of `CachePadded<T>` is the smallest multiple of N bytes large enough to accommodate
+/// a value of type `T`.
+///
+/// The alignment of `CachePadded<T>` is the maximum of N bytes and the alignment of `T`.
+///
+/// # Layout
+///
+/// This type is `#[repr(C)]`, so a pointer to `CachePadded<T>` has the same address as the pointer
+/// to the underlying `T`.
+#[derive(Clone, Copy, Default, Hash, PartialEq, Eq)]
+// Starting from Intel's Sandy Bridge, spatial prefetcher is now pulling pairs of 64-byte cache
+// lines at a time, so we have to align to 128 bytes rather than 64.
+//
+// Sources:
+// - https://www.intel.com/content/dam/www/public/us/en/documents/manuals/64-ia-32-architectures-optimization-manual.pdf
+// - https://github.com/facebook/folly/blob/1b5288e6eea6df074758f877c849b6e73bbb9fbb/folly/lang/Align.h#L107
+//
+// aarch64/arm64ec's big.LITTLE architecture has asymmetric cores and "big" cores have 128-byte
+// cache line size.
+//
+// Sources:
+// - https://www.mono-project.com/news/2016/09/12/arm64-icache/
+//
+// powerpc64 has 128-byte cache line size.
+//
+// Sources:
+// - https://github.com/golang/go/blob/3dd58676054223962cd915bb0934d1f9f489d4d2/src/internal/cpu/cpu_ppc64x.go#L9
+// - https://github.com/torvalds/linux/blob/3516bd729358a2a9b090c1905bd2a3fa926e24c6/arch/powerpc/include/asm/cache.h#L26
+#[cfg_attr(
+    any(
+        target_arch = "x86_64",
+        target_arch = "aarch64",
+        target_arch = "arm64ec",
+        target_arch = "powerpc64",
+    ),
+    repr(align(128))
+)]
+// arm, mips, mips64, sparc, and hexagon have 32-byte cache line size.
+//
+// Sources:
+// - https://github.com/golang/go/blob/3dd58676054223962cd915bb0934d1f9f489d4d2/src/internal/cpu/cpu_arm.go#L7
+// - https://github.com/golang/go/blob/3dd58676054223962cd915bb0934d1f9f489d4d2/src/internal/cpu/cpu_mips.go#L7
+// - https://github.com/golang/go/blob/3dd58676054223962cd915bb0934d1f9f489d4d2/src/internal/cpu/cpu_mipsle.go#L7
+// - https://github.com/golang/go/blob/3dd58676054223962cd915bb0934d1f9f489d4d2/src/internal/cpu/cpu_mips64x.go#L9
+// - https://github.com/torvalds/linux/blob/3516bd729358a2a9b090c1905bd2a3fa926e24c6/arch/sparc/include/asm/cache.h#L17
+// - https://github.com/torvalds/linux/blob/3516bd729358a2a9b090c1905bd2a3fa926e24c6/arch/hexagon/include/asm/cache.h#L12
+#[cfg_attr(
+    any(
+        target_arch = "arm",
+        target_arch = "mips",
+        target_arch = "mips32r6",
+        target_arch = "mips64",
+        target_arch = "mips64r6",
+        target_arch = "sparc",
+        target_arch = "hexagon",
+    ),
+    repr(align(32))
+)]
+// m68k has 16-byte cache line size.
+//
+// Sources:
+// - https://github.com/torvalds/linux/blob/3516bd729358a2a9b090c1905bd2a3fa926e24c6/arch/m68k/include/asm/cache.h#L9
+#[cfg_attr(target_arch = "m68k", repr(align(16)))]
+// s390x has 256-byte cache line size.
+//
+// Sources:
+// - https://github.com/golang/go/blob/3dd58676054223962cd915bb0934d1f9f489d4d2/src/internal/cpu/cpu_s390x.go#L7
+// - https://github.com/torvalds/linux/blob/3516bd729358a2a9b090c1905bd2a3fa926e24c6/arch/s390/include/asm/cache.h#L13
+#[cfg_attr(target_arch = "s390x", repr(align(256)))]
+// x86, wasm, riscv, and sparc64 have 64-byte cache line size.
+//
+// Sources:
+// - https://github.com/golang/go/blob/dda2991c2ea0c5914714469c4defc2562a907230/src/internal/cpu/cpu_x86.go#L9
+// - https://github.com/golang/go/blob/3dd58676054223962cd915bb0934d1f9f489d4d2/src/internal/cpu/cpu_wasm.go#L7
+// - https://github.com/torvalds/linux/blob/3516bd729358a2a9b090c1905bd2a3fa926e24c6/arch/riscv/include/asm/cache.h#L10
+// - https://github.com/torvalds/linux/blob/3516bd729358a2a9b090c1905bd2a3fa926e24c6/arch/sparc/include/asm/cache.h#L19
+//
+// All others are assumed to have 64-byte cache line size.
+#[cfg_attr(
+    not(any(
+        target_arch = "x86_64",
+        target_arch = "aarch64",
+        target_arch = "arm64ec",
+        target_arch = "powerpc64",
+        target_arch = "arm",
+        target_arch = "mips",
+        target_arch = "mips32r6",
+        target_arch = "mips64",
+        target_arch = "mips64r6",
+        target_arch = "sparc",
+        target_arch = "hexagon",
+        target_arch = "m68k",
+        target_arch = "s390x",
+    )),
+    repr(align(64))
+)]
+#[repr(C)]
+pub struct CachePadded<T> {
+    value: T,
+}
+
+impl<T> CachePadded<T> {
+    /// Pads and aligns a value to the length of a cache line.
+    pub const fn new(t: T) -> CachePadded<T> {
+        CachePadded::<T> { value: t }
+    }
+}
+
+impl<T> Deref for CachePadded<T> {
+    type Target = T;
+
+    fn deref(&self) -> &T {
+        &self.value
+    }
+}
+
+impl<T> DerefMut for CachePadded<T> {
+    fn deref_mut(&mut self) -> &mut T {
+        &mut self.value
+    }
+}
+
+impl<T: fmt::Debug> fmt::Debug for CachePadded<T> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("CachePadded")
+            .field("value", &self.value)
+            .finish()
+    }
+}

asyncband/src/internal/mod.rs

@@ -54,6 +54,9 @@ pub fn register_waker(slot: &mut Option<Waker>, waker: &Waker) -> Option<Waker>
 #[allow(dead_code)]
 pub(crate) mod arena;
 
+#[cfg(feature = "mpmc")]
+pub(crate) mod cache_padded;
+
 #[cfg(any(feature = "latch", feature = "once"))]
 pub(crate) mod countdown;
 
@@ -116,6 +119,7 @@ pub(crate) mod waitlist;
     feature = "completion",
     feature = "event",
     feature = "latch",
+    feature = "mpmc",
     feature = "mutex",
     feature = "once",
     feature = "phaser",

asyncband/src/internal/waker_batch.rs

@@ -75,7 +75,9 @@ impl WakerBatch {
             // SAFETY: This prefix was initialized by `push` and is no longer owned by the batch.
             unsafe { slot.assume_init_read() }.wake();
         }
-        self.spilled.drain(..).for_each(Waker::wake);
+        if !self.spilled.is_empty() {
+            self.spilled.drain(..).for_each(Waker::wake);
+        }
     }
 }
 

asyncband/src/mpmc/bounded.rs

@@ -22,25 +22,26 @@ use crate::mpmc::RecvError;
 use crate::mpmc::SendError;
 use crate::mpmc::TryRecvError;
 use crate::mpmc::TrySendError;
-use crate::mpmc::queue::Permit;
-use crate::mpmc::queue::Shared;
+use crate::mpmc::ring::Permit;
+use crate::mpmc::ring::Shared;
 
 /// Creates a bounded multi-producer, multi-consumer queue.
 ///
 /// Queued values, held permits, and capacity granted to waiting senders occupy at most `capacity`
 /// slots. Pending `send` and `reserve` operations receive capacity in wait-queue order. Sending
-/// waits for a receiver to free capacity when none is available.
+/// waits for a receiver to free capacity when none is available. Storage for `capacity` values is
+/// allocated when the queue is created.
 ///
 /// The `try_*` methods do not wait for capacity or messages, but may briefly block on an internal
-/// mutex.
+/// mutex or spin while another sender or receiver finishes with the same slot.
 ///
 /// # Panics
 ///
-/// Panics if `capacity` is zero.
+/// Panics if `capacity` is zero or too large.
 #[track_caller]
 pub fn bounded<T>(capacity: usize) -> (BoundedSender<T>, BoundedReceiver<T>) {
     assert!(capacity > 0, "mpmc bounded queue requires capacity > 0");
-    let shared = Arc::new(Shared::bounded(capacity));
+    let shared = Arc::new(Shared::new(capacity));
     (
         BoundedSender {
             shared: shared.clone(),

asyncband/src/mpmc/mod.rs

@@ -25,7 +25,9 @@
 mod bounded;
 mod error;
 mod queue;
+mod ring;
 mod unbounded;
+mod waiters;
 
 pub use self::bounded::BoundedReceiver;
 pub use self::bounded::BoundedSender;
@@ -34,7 +36,7 @@ pub use self::error::RecvError;
 pub use self::error::SendError;
 pub use self::error::TryRecvError;
 pub use self::error::TrySendError;
-pub use self::queue::Permit;
+pub use self::ring::Permit;
 pub use self::unbounded::UnboundedReceiver;
 pub use self::unbounded::UnboundedSender;
 pub use self::unbounded::unbounded;