Module mpmc 

An async multi-producer multi-consumer channel in which each message goes to one receiver.

bounded and unbounded create a channel and return its first Sender and Receiver. Both can be cloned. Both can be sent to other threads, and shared with them, if the message type T is Send.

A sender moves each message into the channel, and a receiver moves it out. Messages are never cloned, so T need not be Clone. Messages are received in the order they were sent. Each message goes to exactly one receiver.

A bounded channel holds at most as many messages as its capacity. send waits while the channel is full, and try_send fails. An unbounded channel holds any number of messages, and sending to it never waits.

A receiver receives through &self, so several tasks can share one receiver by reference and each wait for the next message. A receiver is also a Stream of messages. The stream ends when the channel is closed and empty.

A channel is closed by Sender::close, by Receiver::close, when its last sender is dropped, and when its last receiver is dropped. Sending to a closed channel fails and returns the message. Receiving still returns the messages left in the channel, and fails only when none is left. When the last receiver is dropped, the messages left in the channel are dropped at once, because nothing can receive them.

The channel needs no runtime and works under any executor. It can be used from any thread, inside a task or not. A thread with no task can wait for a message with the block_on of any executor, as the example below does.

Example

Two senders and two receivers on a channel that holds two messages:

use std::num::NonZeroUsize;

use futures::executor::block_on;
use zruntime::mpmc::{RecvError, TrySendError, bounded};

block_on(async {
    let (s1, r1) = bounded(NonZeroUsize::new(2).unwrap());
    let s2 = s1.clone();
    let r2 = r1.clone();

    // Two messages, from two different senders.
    s1.send("hello").await.unwrap();
    s2.send("world").await.unwrap();

    // The channel is full, so `try_send` fails. A `send` would wait for room.
    assert_eq!(s1.try_send("again"), Err(TrySendError::Full("again")));

    // Each message goes to one receiver, oldest first.
    assert_eq!(r1.recv().await, Ok("hello"));
    assert_eq!(r2.recv().await, Ok("world"));
    assert!(r1.try_recv().unwrap_err().is_empty());

    // Dropping the last sender closes the channel. The receivers see that once there is no
    // message left in it.
    drop(s1);
    drop(s2);
    assert_eq!(r1.recv().await, Err(RecvError));
    assert_eq!(r2.recv().await, Err(RecvError));
});

Spreading work over threads

A runtime of this crate runs its tasks on one thread at a time, so all of its tasks share one core. To use more cores, run one runtime per thread and spread the work over them with a channel. Each thread waits for the next job on its own clone of the same receiver. Each job goes to whichever thread takes it first, so a busy thread leaves new jobs to the others. The results come back through a second channel.

use std::{num::NonZeroUsize, thread};

use zruntime::{LocalRuntime, mpmc};

let (jobs, job_receiver) = mpmc::bounded::<u64>(NonZeroUsize::new(16).unwrap());
let (result_sender, results) = mpmc::unbounded();

let workers: Vec<_> = (0..4)
    .map(|_| {
        let jobs = job_receiver.clone();
        let results = result_sender.clone();

        thread::spawn(move || {
            // A runtime that belongs to this thread.
            let runtime = LocalRuntime::new().unwrap();
            runtime.block_on(async {
                // The loop ends when the channel is closed and empty.
                while let Ok(n) = jobs.recv().await {
                    let sum: u64 = (1..=n).sum();
                    results.send(sum).await.unwrap();
                }
            });
        })
    })
    .collect();
// The workers are the only receivers of jobs and senders of results from here on. The loop
// over the results below ends only when every result sender is dropped, including this one.
drop(job_receiver);
drop(result_sender);

let runtime = LocalRuntime::new().unwrap();
let total = runtime.block_on(async {
    for n in 1..=100 {
        jobs.send(n).await.unwrap();
    }
    // No more jobs. The workers' loops end after they take the last one.
    drop(jobs);

    // Results arrive in the order the workers finish their jobs. The loop ends when every
    // worker has dropped its sender.
    let mut total = 0;
    while let Ok(sum) = results.recv().await {
        total += sum;
    }
    total
});

for worker in workers {
    worker.join().unwrap();
}
assert_eq!(total, (1..=100u64).map(|n| n * (n + 1) / 2).sum());

If jobs mostly wait for I/O, a worker can spawn a task for each job on its runtime. The jobs then wait at the same time instead of one after the other.

Giving up a wait

It is safe to drop the future returned by Sender::send or Receiver::recv before it completes. A timeout or a select can do this. A dropped Recv has taken no message. A dropped Send has sent none, and its message is dropped with it. If a future was woken for a message or for room and is then dropped, the wake-up passes to the next task waiting, so no waiting task is left stranded.

Difference with broadcast

The broadcast module has the same two sides, a sender and a receiver. It delivers every message to every receiver, as a clone, so its messages must be Clone. This channel delivers each message to one receiver only, and moves it there. Use broadcast when every receiver must see every message, and this channel when receivers share the work.

The broadcast module requires the broadcast feature.

Difference with async-channel

This channel is modelled on async-channel. It differs in these ways:

• The capacity of bounded is a NonZeroUsize, so the type rules out a zero capacity. async-channel panics on bounded(0).
• Dropping the last receiver drops the messages left in the channel. async-channel only closes the channel, and the messages stay in it until the channel itself is freed.
• There is no blocking API. Instead of send_blocking and recv_blocking, a thread with no task can wait with the block_on of an executor.

Performance

Benchmarks compared this channel with async-channel 2.5.0 and with the mpsc channel of tokio 1.53.1. Tokio’s channel has only one receiver, so it is left out of the rows with several receivers. The messages are u64s. The machine is a shared x86-64 one with four cores. Each thread drives its whole part with one block_on call of futures-lite. Tasks run on a tokio runtime with four workers.

In the rows with threads or tasks, each sender sends 10,000 messages, and the row times all of them. The row marked “a message” times one message. The times are medians. Times on one thread stayed within 3% from run to run. The others varied by up to 30%. Sending to tokio’s unbounded channel is not a future, so it needs no poll.

What is timed	mpmc	async-channel	tokio
A send and a receive, one thread, capacity 1	55 ns	112 ns	100 ns
A send and a receive, one thread, unbounded	54 ns	123 ns	56 ns
1024 try_sends then 1024 try_recvs, capacity 1024	36 µs	99 µs	67 µs
1024 try_sends then 1024 try_recvs, unbounded	36 µs	112 µs	47 µs
4 sender and 4 receiver threads, capacity 16	72 ms	240 ms
4 sender and 4 receiver threads, unbounded	9.7 ms	16 ms
4 sender tasks and 1 receiver task, capacity 16	12 ms	33 ms	13 ms
4 sender and 4 receiver tasks, capacity 16	18 ms	24 ms
1 sender task and 1 receiver task, capacity 16, a message	280 ns	550 ns	540 ns
1 sender thread and 1 receiver thread, capacity 16	23 ms	20 ms	22 ms
1 sender thread and 1 receiver thread, capacity 1024	2.3 ms	1.7 ms	2.8 ms
4 sender threads and 1 receiver thread, capacity 16	470 ms	500 ms	460 ms

In the last three rows, the channel keeps filling up or running empty, so a thread has to wait every few messages. Waiting parks the thread, and waking it takes the OS tens of microseconds on that machine. That wake-up makes up nearly all of the time. So in these rows the channels differ in how often a thread had to wait, not in the cost of an operation. Between tasks, which wake cheaply, this channel takes half as long as the other two, as the “a message” row shows.

Structs

Receiver

The receiving side of a channel.

Recv

The future returned by Receiver::recv.

RecvError

The error returned by a Receiver::recv that failed.

Send

The future returned by Sender::send.

SendError

The error returned by a Sender::send that failed because the channel is closed.

Sender

The sending side of a channel.

Enums

TryRecvError

The error returned by a Receiver::try_recv that failed.

TrySendError

The error returned by a Sender::try_send that failed.

Functions

bounded

Creates a channel that holds at most cap messages.

unbounded

Creates a channel that holds any number of messages.