Skip to content

feat(mpmc): add competing queues - #265

Open
mxsm wants to merge 2 commits into
apache:mainfrom
mxsm:feat-mxsm-211
Open

feat(mpmc): add competing queues#265
mxsm wants to merge 2 commits into
apache:mainfrom
mxsm:feat-mxsm-211

Conversation

@mxsm

@mxsm mxsm commented Sep 1, 2026

Copy link
Copy Markdown
Member

Summary

  • Add asyncband::mpmc bounded and unbounded queues with cloneable senders and receivers.
  • Deliver every accepted value to exactly one competing receiver while receivers remain, preserve FIFO queue order, enforce strict bounded capacity, and drain buffered values before reporting sender disconnection.
  • Use targeted FIFO waiter notification for ordinary sends and receives, pass selected notifications to the next waiter after cancellation, and reserve wake-all behavior for terminal disconnection.
  • Add behavior, contention, trait, and cancellation tests plus bounded and unbounded ecosystem benchmarks for 1P/1C, 1P/8C, 8P/1C, and 8P/8C topologies.

Closes #211

Design Notes

Queue storage, endpoint counts, capacity checks, and disconnection state are linearized under one private Mutex<VecDeque<T>>. Separate internal semaphores track blocked receivers and capacity-waiting bounded senders, so an ordinary queue transition selects one waiter without coupling public endpoints to storage or waiter types.

Dropping the final sender wakes all receivers, which first drain buffered values and then observe Disconnected. Dropping the final receiver releases buffered values and wakes every blocked bounded sender so each pending send returns its own unsent value.

The new mpmc Cargo feature is an independent compile, audit, and code-size boundary for the new public module. This keeps the implementation opt-in without placing MPMC APIs behind the existing mpsc feature; maintainers should confirm this boundary against the feature policy discussed in #216.

Testing

  • cargo x check — passed the no-feature build, the standalone mpmc and event feature builds, every other standalone feature build, and the all-feature build.
  • cargo x test --no-capture — passed the full workspace suite, including 14 MPMC integration tests, upstream panic-recovery tests, and all documentation tests.
  • cargo x lint — passed Clippy with warnings denied, rustfmt, Taplo, typos, license checks, and documentation generation.
  • The MPMC tests cover clone counts, FIFO ordering, strict capacity, disconnection, buffered draining, unsent-value recovery, targeted wakeups, sender and receiver cancellation handoff, T: Send + !Sync endpoint traits, future Send traits, exact-once delivery, and bounded progress under 8P/8C contention.

Performance

Command: cargo bench -p benchmarks --bench ecosystem -- mpmc

The following values are median time per 16,384-value batch from the optimized Windows run after merging the current main. Lower is better; the ratio compares Asyncband with the fastest result in the same row.

Bounded, capacity 64

Topology Asyncband async-channel flume Asyncband / fastest
1P/1C 6.151 ms 3.390 ms 4.415 ms 1.81x
1P/8C 24.52 ms 34.51 ms 26.57 ms 1.00x
8P/1C 24.12 ms 10.81 ms 21.30 ms 2.23x
8P/8C 7.219 ms 7.681 ms 7.045 ms 1.02x

Unbounded

Topology Asyncband async-channel flume Asyncband / fastest
1P/1C 5.216 ms 2.052 ms 2.251 ms 2.54x
1P/8C 24.04 ms 3.661 ms 10.69 ms 6.57x
8P/1C 3.322 ms 3.674 ms 2.060 ms 1.61x
8P/8C 5.897 ms 4.315 ms 3.513 ms 1.68x

The largest observed median gap was 6.57x for unbounded 1P/8C, below the order-of-magnitude regression threshold in #211. The private mutex-backed queue is expected to serialize more heavily than async-channel in some producer or consumer skews; the benchmark matrix is retained so future changes can track and improve those paths.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

feat(mpmc): add a competing queue

1 participant