perf(mpsc): use a segmented unbounded queue

CHANGELOG.md

@@ -17,7 +17,7 @@ All notable changes to this project will be documented in this file.
 ### Improvements
 
 * Finish releasing buffered bounded MPSC messages even if one message destructor panics.
-* Improve unbounded MPSC throughput with batched receiving and incremental storage reclamation; empty-buffer retention is bounded independently of previous peak occupancy.
+* Improve unbounded MPSC throughput for ready-message and contended asynchronous workloads; reclaim consumed storage incrementally and bound empty-buffer retention independently of previous peak occupancy.
 
 ## v0.7.2
 

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" }

LICENSE

@@ -436,3 +436,15 @@ composition. Fastpool is licensed under Apache-2.0, and its source carries the
 following copyright notice:
 
   Copyright 2025 FastLabs Developers
+
+Portions of asyncband/src/mpsc/unbounded/queue.rs are adapted from the
+segmented list in crossbeam-channel 0.5.16 at the following exact revision:
+
+  https://github.com/crossbeam-rs/crossbeam/blob/9b56303b8aa9ff8ec5bbebb9d2da05e034977889/crossbeam-channel/src/flavors/list.rs
+
+The single-consumer implementation replaces multi-reader reclamation with FIFO
+block ownership, bounds slots by payload size, and separates receiver notification
+from storage. Crossbeam is licensed under Apache-2.0 or MIT; Asyncband uses the
+Apache-2.0 option. The upstream distribution carries this copyright notice:
+
+  Copyright (c) 2019 The Crossbeam Project Developers

asyncband/src/mpsc/unbounded/mod.rs

@@ -19,12 +19,15 @@
 //! tasks.
 
 use std::sync::Arc;
+use std::sync::atomic::AtomicBool;
+use std::sync::atomic::AtomicUsize;
+use std::sync::atomic::Ordering;
 use std::task::Waker;
 
-use self::buffer::Buffer;
+use self::queue::Queue;
 use crate::internal::mutex::Mutex;
 
-mod buffer;
+mod queue;
 mod receiver;
 mod sender;
 
@@ -43,28 +46,92 @@ pub use self::sender::UnboundedSender;
 /// Storage is reclaimed incrementally as messages are received. A bounded amount of empty
 /// storage may be retained for reuse, independently of the channel's previous peak occupancy.
 ///
-/// Operations briefly acquire an internal mutex; no lock is held across an await point or while
-/// invoking waker callbacks or message destructors. Sending and trying to receive may wait to
-/// acquire this mutex, but never wait for capacity or new messages.
+/// Sending and receiving may briefly wait for an in-progress producer or an internal mutex, but
+/// never wait for capacity or new messages in `send` or `try_recv`. No lock is held across an
+/// await point or while invoking waker callbacks or message destructors.
 pub fn unbounded<T>() -> (UnboundedSender<T>, UnboundedReceiver<T>) {
-    let shared = Arc::new(Mutex::new(State {
-        buffer: Buffer::new(),
-        senders: 1,
-        receiver: true,
-        recv_waker: None,
-    }));
+    let (queue, consumer) = Queue::new();
+    let shared = Arc::new(State {
+        queue,
+        senders: CachePadded(AtomicUsize::new(1)),
+        recv: ReceiverWake {
+            waiting: CachePadded(AtomicBool::new(false)),
+            waker: Mutex::new(None),
+        },
+    });
     (
         UnboundedSender::new(shared.clone()),
-        UnboundedReceiver::new(shared),
+        UnboundedReceiver::new(shared, consumer),
     )
 }
 
-// Queue contents, endpoint liveness, and wake registration share one lock. Only the receiver
-// accesses its current batch; refilling that batch preserves the order of concurrent sends.
+// Separate producer reservations, sender counts, and the receiver's mostly-read waiting flag.
+#[cfg_attr(
+    any(
+        target_arch = "aarch64",
+        target_arch = "arm64ec",
+        target_arch = "x86_64",
+        target_arch = "powerpc64"
+    ),
+    repr(align(128))
+)]
+#[cfg_attr(target_arch = "s390x", repr(align(256)))]
+#[cfg_attr(
+    not(any(
+        target_arch = "aarch64",
+        target_arch = "arm64ec",
+        target_arch = "x86_64",
+        target_arch = "powerpc64",
+        target_arch = "s390x"
+    )),
+    repr(align(64))
+)]
+struct CachePadded<T>(T);
+
 struct State<T> {
-    buffer: Buffer<T>,
-    senders: usize,
-    // True while the receiving endpoint is alive.
-    receiver: bool,
-    recv_waker: Option<Waker>,
+    queue: Queue<T>,
+    senders: CachePadded<AtomicUsize>,
+    recv: ReceiverWake,
+}
+
+struct ReceiverWake {
+    waiting: CachePadded<AtomicBool>,
+    waker: Mutex<Option<Waker>>,
+}
+
+impl ReceiverWake {
+    fn register(&self, waker: &Waker) {
+        // Clone, replacement destruction, and wake may reenter this channel.
+        let waker = waker.clone();
+        let mut slot = self.waker.lock();
+        let old = slot.replace(waker);
+        // Publication and this flag share an SC order: after register -> recheck, either the
+        // receiver sees the value or its producer observes this registration and wakes it.
+        self.waiting.0.store(true, Ordering::SeqCst);
+        drop(slot);
+        drop(old);
+    }
+
+    fn take(&self) -> Option<Waker> {
+        let mut slot = self.waker.lock();
+        let waker = slot.take();
+        // Clear under the registration lock so a delayed notifier cannot erase a newer wait.
+        self.waiting.0.store(false, Ordering::SeqCst);
+        waker
+    }
+
+    fn wake(&self) {
+        if !self.waiting.0.load(Ordering::SeqCst)
+            || self
+                .waiting
+                .0
+                .compare_exchange(true, false, Ordering::SeqCst, Ordering::SeqCst)
+                .is_err()
+        {
+            return;
+        }
+        if let Some(waker) = self.take() {
+            waker.wake();
+        }
+    }
 }