Module broadcast

Expand description

A multi-producer, multi-consumer broadcast channel: each value sent reaches every receiver that exists when it is sent.

channel returns a Sender and a Receiver around a capacity. Sender::send puts a value in the channel for every receiver; each Receiver takes the values in order with Receiver::recv, cloning them (T: Clone). The channel holds the last capacity values (rounded up to a power of two): a receiver that falls further behind loses the oldest ones, and its next receive reports how many with RecvError::Lagged before it goes on from the oldest value still held. The channel closes once every sender is gone: receivers take what is left, then get RecvError::Closed. The API is the established async broadcast-channel API; nothing here needs a runtime.

use nuclei::sync::broadcast;

let (tx, mut rx1) = broadcast::channel(16);
let mut rx2 = tx.subscribe();
tx.send(10).unwrap();
tx.send(20).unwrap();
assert_eq!(rx1.try_recv(), Ok(10));
assert_eq!(rx2.try_recv(), Ok(10));
assert_eq!(rx1.try_recv(), Ok(20));

§Values live in nodes published lock-free

The channel is a ring of capacity slots, each an atomic pointer holding the node of the latest value sent to it with that value’s position. A sender takes the next position with one fetch_add, publishes its node into the position’s slot with a compare-and-swap that never replaces a later position, and retires the node it replaced. A receive is a protected load of its slot and a clone of the value: it writes nothing another thread reads, so receivers never contend with one another or with the senders. Two consequences, both differing from the lock-based API:

The channel needs T: Send + Sync + 'static: values are shared by reference between threads and reclaimed later.
A value is dropped when a later send takes its slot (or the channel goes), not as soon as every receiver has taken it: the channel holds up to capacity values whatever the receivers have read. Sender::len still counts what the slowest receiver has yet to take.

Under the recycle feature its nodes come from a per-thread cache instead of the global allocator.

§Progress

Wait-free: Receiver::try_recv, Receiver::len, Receiver::is_empty, Sender::receiver_count, Sender::same_channel, Sender::subscribe, Receiver::resubscribe, Sender::downgrade, the sender counts, and the clones and drops but the last sender’s and the last receiver’s.
Lock-free: Sender::send (its compare-and-swap is retried only if a send a whole ring later published into the same slot meanwhile, and it ends by waking the waiting receivers through Notify::notify_waiters, which is lock-free); Sender::len and Sender::is_empty (they look at every receiver); polling Receiver::recv and Sender::closed (a poll may queue a waiter); WeakSender::upgrade (a compare-and-swap on the sender count, retried when a sender was cloned or dropped meanwhile); and the last sender’s or receiver’s drop, for its wake.

Re-exports§

pub use self::error::RecvError;
pub use self::error::SendError;
pub use self::error::TryRecvError;

Modules§

error: Error types of the broadcast channel.

Structs§

Receiver: Receives the values of the channel, in order.
Sender: Sends values to every Receiver of the channel.
WeakSender: A sender that does not keep the channel open.

Functions§

channel: Creates a broadcast channel holding the last capacity values (rounded up to a power of two), returning a sender and a first receiver.