Skip to content

← Fearless Concurrency: Threads, Channels, Shared State step 18 of 24

Medium Framework

sync_channel: bounded buffers and backpressure

Build a three-stage pipeline in which each stage runs on its own thread and the stages are connected by bounded channels created with std::sync::mpsc::sync_channel(buffer).

pub fn pipeline_stages(input: Vec<i64>, buffer: usize) -> Vec<i64>

The contract

  • The stages apply, in order, x + 1, then x * 2, then x - 3. Together they turn each value x into 2 * (x + 1) - 3, which equals 2x - 1. For example, 5 becomes 6, then 12, then 9.
  • Return one output value per input value, in input order. Input values may repeat, be negative or be zero. An empty input returns an empty vector.
  • Every value in the tests lies between -(2^62) + 1 and 2^62 - 2 inclusive, so every intermediate result fits in i64. You do not need to handle overflow.
  • buffer is the capacity of every channel. The tests use 0, 1, 3 and 8; the same ten-value input is checked at 0, 1 and 8. The answer must not depend on buffer, and the function must return (not hang) for every capacity, including 0.
  • Only the returned vector is graded. The tests cannot see how many threads you used, but the exercise is about the threaded design below.

Unbounded and bounded channels

A channel is a queue with a sending end and a receiving end. Values come out in the order they went in (first in, first out).

mpsc::channel() creates an unbounded channel: Sender::send never blocks, and the queue grows to hold every value that has been sent but not yet received. If the producer keeps running faster than the consumer, that backlog, and the memory holding it, keeps growing.

mpsc::sync_channel(n) creates a bounded channel that holds at most n values. When it is full, SyncSender::send blocks (waits) until the receiver takes a value out. This is backpressure: a stage that cannot keep up makes the stage before it wait, so in a chain of bounded channels the slowest stage limits the rate of the whole pipeline, and the number of values waiting in each channel never exceeds its capacity.

The capacity is a trade-off. A small buffer keeps fewer values waiting and makes neighbouring stages wait for each other more often. A larger buffer lets a fast stage run further ahead and absorb short bursts, at the cost of more values held in memory and possibly longer waits for a value sitting behind others in a full queue. The capacity changes timing, never the results, and the tests check the results.

A capacity of zero is a rendezvous

With sync_channel(0) there is no storage at all. The standard library calls this a rendezvous channel: each send does not return until a receiver takes that value. The two threads meet, hand the value over, and both continue.

So a sender can only make progress if something is receiving at the same time. The starter sends every input value from the main thread before spawning any stage, so nothing is receiving yet:

  • buffer = 8 with three values: works, because all three fit in the channel.
  • buffer = 8 with the ten-value test input: the ninth send waits forever.
  • buffer = 1: the second send waits forever.
  • buffer = 0: the first send waits forever.

Because whether it hangs depends on the capacity and the input length, this kind of bug can pass small tests and fail later. Start every consumer before you send anything.

There is a second trap. Even with all stages running, the main thread cannot first send every value and only then read the results. While it is still sending, nobody reads the last channel, so once the values in flight fill the channels and the stages, every thread is waiting on another and nothing moves. Send the input from a separate thread so that the main thread can collect results at the same time.

Why the order is preserved

Each stage is a single thread that reads one channel and writes one channel. A channel delivers values in the order they were sent, and a single thread handles its values one at a time in the order it receives them, so the order going in is the order coming out, through all four channels.

That guarantee is lost if a stage is split across several worker threads reading from one channel: the workers can finish in any order. Using several workers is a normal way to use more cores, but then you have to restore the order yourself, for example by tagging each value with its index. With one thread per stage you do not need to.

Shutting down

A for x in rx loop over a Receiver ends when the channel is empty and every sender for it has been dropped. Each stage owns the sender for its outbound channel. When the stage's inbound loop ends, its closure returns and that sender is dropped, which ends the next stage's loop, and so on. So once the feeding thread has sent the last value and dropped the first sender, the whole pipeline drains and stops by itself, and the main thread's loop over the last receiver ends after the final result.

Make sure no extra sender is kept alive, for example a clone or one left in the main thread. A sender that is never dropped keeps the receiving loop waiting forever.

A send fails only when the receiving end has been dropped, which here happens only if a later stage has stopped, for example by panicking. Stopping the loop on that error is simpler and quieter than panicking in every stage.

The API

  • sync_channel(n) returns (SyncSender<T>, Receiver<T>).
  • SyncSender::send(value) blocks while the channel is full. If the receiver has been dropped it returns Err(SendError(value)), which gives the unsent value back.
  • SyncSender::try_send(value) never blocks. It returns Err(TrySendError::Full(value)) when the channel has no room (with capacity 0, when no receiver is waiting) and Err(TrySendError::Disconnected(value)) when the receiver is gone. A program that prefers dropping work to waiting can act on Full. This exercise does not need try_send.