Skip to content

feat(broadcast): add an SPMC specialization - #317

Open
onenewcode wants to merge 10 commits into
apache:mainfrom
onenewcode:refactor/broadcast-spmc-internals
Open

onenewcode wants to merge 10 commits into
apache:mainfrom
onenewcode:refactor/broadcast-spmc-internals

Conversation

@onenewcode

@onenewcode onenewcode commented Sep 15, 2026

Copy link
Copy Markdown
Contributor

Summary

Closes #214.

  • Add broadcast::spmc, a lossless single-producer broadcast family with a non-cloneable sender whose publish methods take &mut self.
  • Keep the same public retention contract as broadcast::mpmc: bounded waits at capacity for the slowest active subscription, unbounded never waits, and a new subscription starts at the committed tail.
  • Match the broadcast::mpmc public surface method for method, except the sender is not Clone. A receive drains accepted values before reporting disconnection. Unbounded value storage tracks the live window: discarded sends with no subscribers do not grow the log, and retained storage is released as the backlog drains.
  • Keep subscriptions independent. Slot-log machinery stays private to asyncband/src/broadcast/spmc.

Design Notes

Public retention matches broadcast::mpmc: lossless fan-out, bounded wait-at-capacity, unbounded growth, join at the committed tail, and discard with no subscribers. try_recv drains accepted values before Disconnected. Unbounded discarded sends do not grow the log.

The sender is not Clone; publish takes &mut self. A blocked bounded send cannot subscribe; use resubscribe. The slot log stays family-private. Publication shares the waiter mutex with subscribe and parking; receivers drain published slots without that lock.

Benchmark comparison

One producer. broadcast::spmc senders are not cloned. mpmc 1P is broadcast::mpmc with one sender.

Bounded is the lossless wait-at-capacity path (async-broadcast overflow off). Tokio is omitted: it overwrites at capacity. Unbounded is the non-blocking batch path; Tokio and async-broadcast get capacity 4096 so the batch stays off Lagged. Native-thread concurrent waits use FutureExt::block_on. Figures are divan medians from one run on Apple M4 (cargo bench -p benchmarks --bench ecosystem -- broadcast::spmc).

Bounded (lossless wait-at-capacity)

Bench async-broadcast spmc mpmc 1P
try_round_trip 24.6 ns 12.8 ns 32.4 ns
ready_round_trip 39.3 ns 16.2 ns 42.5 ns

Native threads (concurrent). Lower time is better.

Shape async-broadcast spmc mpmc 1P
cap 1 / 1 recv 16.2 ms 9.25 ms 15.5 ms
cap 1 / 8 recv 123 ms 72.5 ms 76.1 ms
cap 1 / 32 recv 526 ms 216 ms 235 ms
cap 64 / 1 recv 491 µs 241 µs 537 µs
cap 64 / 8 recv 10.1 ms 21.8 ms 14.5 ms
cap 64 / 32 recv 37.8 ms 151 ms 171 ms

Tokio 4-worker (scheduled):

Shape async-broadcast spmc mpmc 1P
cap 1 / 1 recv 1.28 ms 585 µs 908 µs
cap 1 / 8 recv 14.5 ms 3.59 ms 5.21 ms
cap 1 / 32 recv 67.1 ms 14.9 ms 29.6 ms
cap 64 / 1 recv 233 µs 113 µs 255 µs
cap 64 / 8 recv 2.14 ms 392 µs 2.13 ms
cap 64 / 32 recv 5.92 ms 1.02 ms 7.48 ms

Unbounded (non-blocking batch)

Bench async-broadcast tokio spmc mpmc 1P
try_round_trip 24.5 ns 17.4 ns 12.8 ns 32.5 ns
ready_round_trip 26.3 ns 21.5 ns 14.8 ns 36.1 ns

Fan-out (publish 4096, then every subscription drains):

Receivers async-broadcast tokio spmc mpmc 1P
1 127 µs 136 µs 87.4 µs 162 µs
2 235 µs 149 µs 120 µs 249 µs
4 476 µs 165 µs 142 µs 701 µs
8 907 µs 188 µs 179 µs 1.50 ms
32 3.32 ms 588 µs 436 µs 5.52 ms

Summary
Add `broadcast::spmc` with a non-cloneable sender whose publish methods take `&mut self`. Bounded waits for the slowest active subscription; unbounded never waits. Existing `broadcast::mpmc` is unchanged.
Receivers clone T from a single-writer log without taking the waiter mutex.
The last remaining reader takes the payload.
Native-thread send_blocking/recv_blocking wait on one epoch condvar instead
of an async waker per task.
Unbounded send writes the slot and publishes tail without the waiter mutex
when receivers exist. Fan-out at 32 receivers is now under Tokio's lossy
ring.
disconnect() notifies the shared condvar, and recv_blocking rechecks
disconnection after taking that lock so a parked native-thread receive
cannot hang.
Drop the extra wait helper. Clear send_in_progress once, park without
cloning the channel Arc, and recover condvar poison the same way as Mutex.
@onenewcode
onenewcode force-pushed the refactor/broadcast-spmc-internals branch from b072b5e to 43a8da8 Compare September 15, 2026 05:44
@onenewcode onenewcode changed the title refactor(broadcast): hide SPMC internals and lock the sender contract feat(broadcast): add an SPMC specialization Sep 15, 2026
Keep slot-log types family-private. Compile-fail examples show the sender
is not Clone and publish takes &mut self. Cover unbounded fan-out, late
subscribe, and slow-subscription growth.
@onenewcode
onenewcode force-pushed the refactor/broadcast-spmc-internals branch from 43a8da8 to a2081b9 Compare September 15, 2026 05:49
Hold the waiter mutex across publication so a slot's remaining-reader
count matches live subscriptions and a parking receiver cannot miss a
send. Release unbounded chunk storage as head advances, and wake the
producer if a panicking clone is the last reader. Drop native-thread
send_blocking/recv_blocking; FutureExt::block_on covers that path.
Reload tail after observing a dropped sender so try_recv cannot report
Disconnected while a published value is still unread. Discard unbounded
sends without advancing the version log so empty chunks are not allocated
for messages that were never retained.
@onenewcode
onenewcode marked this pull request as ready for review September 16, 2026 03:59
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(broadcast): add an SPMC specialization

1 participant