Module mpsc 

Multi-producer, single-consumer channels for sending values between tasks.

The module has two flavors. The bounded one (channel) holds at most a fixed number of messages, and a send waits for room, which is backpressure. The unbounded one (unbounded_channel) never makes a send wait, so UnboundedSender::send is an ordinary function that synchronous and asynchronous code can both call. The names, signatures, error types and documented behavior are those of the established async channel API, so code written against that API compiles against these; nothing here needs a runtime, and every wait is a plain Waker handshake.

use nuclei::sync::mpsc;

let (tx, mut rx) = mpsc::channel(8);
let senders: Vec<_> = (0..4)
    .map(|id| {
        let tx = tx.clone();
        std::thread::spawn(move || {
            for n in 0..100 {
                tx.blocking_send((id, n)).expect("the receiver is still there");
            }
        })
    })
    .collect();
drop(tx);

let mut seen = [0; 4];
while let Some((id, _n)) = rx.blocking_recv() {
    seen[id] += 1;
}
assert_eq!(seen, [100; 4]);
for sender in senders {
    sender.join().unwrap();
}

Ordering

What one sender sends comes out in the order it was sent, as the established API promises. The message queue is a linearizable FIFO queue of cells: a send claims the next cell, and the receiver takes cells in the order they were claimed, so a send that returned before another began comes out first. Values that different senders send at the same time come out in the order their sends claimed cells.

Disconnection

When every sender is dropped, no value can be sent. That is the end of the stream: once the buffered values are received, recv returns None.

If the receiver is dropped, its unread values are dropped with it, and every later send fails with an error that holds the value.

Clean shutdown

Dropping the receiver loses the values still buffered. For a clean shutdown the receiver calls close first, which stops further sends but keeps the buffered values (and any slot a sender reserved before) usable, then receives until recv returns None.

Weak senders

A WeakSender does not keep the channel open: when only weak senders remain the channel is closed. WeakSender::upgrade gives a Sender while one exists. An actor can hold a weak handle to its own mailbox without keeping it alive.

Communicating between sync and async code

Every method that is not async can be called from any thread. For synchronous code on the receiving or sending end, the blocking_* methods park the calling thread until their wait is over. On a nuclei shard they run a nested block_on instead, which keeps the shard running its other tasks; do not call them from a task of another runtime, whose thread they block.

Differences from the established API

• Sender::send_timeout waits on nuclei’s own timers: on a thread that runs no nuclei shard, the default runtime fires it.
• The blocking_* methods called from inside a task do not panic: on a nuclei shard they run a nested block_on, and inside another runtime’s task they block its thread (on wasm32, where nothing else can run, a call that would wait forever panics instead).
• When the receiver closes, a reservation still waiting gives back the slots set aside for it at the close, where the established API keeps them with the failed wait until its future is dropped. So a closed receiver’s drain never waits for a failed reservation to be dropped.
• Receiver::try_recv does not park the thread for a send that is in progress on another thread. Such a value is not in the channel yet, and its sender wakes the receiver when it is; the established API parks the thread until the send finishes.

Cooperative scheduling

On a nuclei shard, every receive that completes (recv, recv_many and the two poll_ forms) spends one operation of its task’s cooperative budget, as IO does, and so does a send or a reservation that completes at its first attempt; one whose slots were granted after it waited in line spends nothing, as giving them back would cost it its place. When the budget is spent the call takes nothing, wakes its task and returns Pending, so a loop on a channel that is always ready yields to the shard’s other tasks within BUDGET operations. Receiver::try_recv, Sender::try_send and the other calls that never wait spend nothing, and off a nuclei shard there is no budget: nothing yields for it.

Progress

The message queue is built for its one receiver: a bounded channel’s is a ring of cells used lap after lap (a radix directory past 65,536 cells), an unbounded channel’s a radix directory of leaves. A send claims the next cell with one fetch-add, finds its segment (in the ring by its position, in the directory by at most one level per ten of a 64-bit position, making a missing node with one compare-and-swap that, lost, leaves another send’s), writes its value and marks the cell ready: the push is wait-free. A receive reads the next cell’s mark and takes the value, with no read-modify-write and nothing another thread contends for, and never waits for a sender: when the next cell is claimed by a send that has not written it yet, the receive finds nothing for now, registers its waker and returns Pending, and that send wakes it once its value is in. A sender the scheduler stops between its claim and its write holds back no one else’s value in the radix directory, unless the position right after it is stalled too: the receive takes the ready values past it (within 64 positions), each sender’s still in its order, and its own once written. In a bounded channel’s ring the values behind it wait until it runs again, as its cell holds their place in the lap; no call waits for it either way.

• Wait-free (a bounded number of the caller’s own atomic steps, whatever other threads do): the accessors (is_closed, len, is_empty, capacity, max_capacity, the sender counts, same_channel), clone, downgrade, and dropping a sender or weak sender (the last sender’s drop also wakes the receiver, a two-step handshake on the waker cell). Registering the receiver’s waker is wait-free too, and so is UnboundedSender::send: its admission is one fetch-add (taken back with one fetch-sub on a closed channel), its push the queue’s.
• Lock-free: every bounded send, try_send, reservation and permit send, whose admission is the semaphore’s compare-and-swap (exact: it never refuses a free slot); closed; WeakSender::upgrade (a compare-and-swap loop on the sender count); and Receiver::close. A bounded wait that has to queue is lock-free for the reason every wait queue is (see the parent module): its poll pushes a node, and may serve the queue for one pass plus one for each operation of another thread that lands meanwhile.
• When a waiting bounded send is served is not a bound on its own calls. Its slot is set aside and granted by whichever thread serves the queue once the slot is free (the receiver taking a value, a permit dropped), so a thread the scheduler stops while it serves the queue delays that grant until it runs again, and a try_send fails meanwhile even if the last waiting send has just left. No call waits for it.
• Lock-free: every receive (recv, recv_many, try_recv, poll_recv, poll_recv_many), none of which waits for a sender inside the call, and the receiver’s drop, which drops every value left and so takes time proportional to them.
• A Waker the channel calls is the executor’s, and so is what it does; the destructor of a value runs in the call that drops it; and the blocking_* methods park the thread until their wait is over, which is what they are for.

Cancel safety

recv, recv_many and closed are cancel safe: dropping the future loses nothing. send is cancel safe in the sense that a dropped future sent nothing, but it drops the value it held; reserve and its relatives hold no slot when dropped, and a dropped reserve_owned also drops its sender.

Fairness

The bounded channel serves waiting sends and reservations strictly in the order they began to wait, as the established API does. Slots that come free are set aside for the oldest waiting request until it has all it asked for, so a reserve_many(n) is not starved by smaller requests and nothing behind it is served first; a try_send or try_reserve never takes a slot while a request waits. Dropping a waiting future gives up its place and passes on whatever was set aside or granted to it, so no slot and no wakeup is lost. The unbounded channel never waits, so it has no order to keep.

Modules

error

Error types of the multi-producer, single-consumer channels.

Structs

OwnedPermit

An owned permit to send one value into the channel.

Permit

Permits to send one value into the channel.

PermitIterator

An Iterator of Permits that holds n slots of the channel.

Receiver

Receives values from the associated Senders.

Sender

Sends values to the associated Receiver.

UnboundedReceiver

Receives values from the associated UnboundedSenders.

UnboundedSender

Sends values to the associated UnboundedReceiver.

WeakSender

A sender that does not keep the channel open.

WeakUnboundedSender

A sender that does not keep the channel open.

Functions

channel

Creates a bounded channel for communicating between tasks with backpressure.

unbounded_channel

Creates an unbounded channel for communicating between tasks without backpressure.