sync_channel: bounded buffers and backpressure

Build a three-stage pipeline. Each stage is one thread, connected by sync_channel(buffer).

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

Stages, in order: x + 1, then x * 2, then x - 3. So each value comes out as 2x - 1, and the output is in input order.

The tests run the same input at buffer = 0, 1, 3 and 8 and demand identical output. If your answer depends on the buffer size, your pipeline is wrong.

Backpressure is what separates a queue from a pipeline

mpsc::channel() is unbounded. send never blocks; it allocates. That sounds like a feature until the producer is faster than the consumer, and then it is an unbounded memory leak wearing a queue costume: the queue grows until the process is killed, and the symptom appears somewhere else entirely, minutes later.

mpsc::sync_channel(n) is bounded. When the buffer is full, send blocks until the consumer takes something out. That is backpressure: the slowest stage sets the pace of the whole pipeline, automatically, with no coordination and no configuration. Memory use is bounded by construction, and the queue depth becomes a design parameter you chose rather than a number you discover in a postmortem.

The tuning knob is real: 0 is maximum coupling and minimum memory; larger buffers absorb bursts at the cost of latency and footprint. What must not change is the result, which is what the tests check.

buffer == 0 is a rendezvous, and it is where the starter dies

With capacity zero there is no buffer at all. A send completes only when a receiver is simultaneously taking the value — the two threads meet, hand the value over, and both continue. That is a rendezvous, and it is a genuine synchronisation point, not just a small queue.

The consequence dictates your program’s structure: you cannot send into a stage whose consumer is not running yet. The starter feeds the pipeline from the main thread before spawning any stage, and:

• buffer = 8, three items — works. All three fit in the buffer.
• buffer = 1 — blocks on the second send.
• buffer = 0 — blocks on the first send, immediately, forever.

A bug whose appearance depends on a capacity constant is a nasty one to meet in production. Spawn every consumer first, then feed.

Why the order is preserved

Each stage is a single thread reading one channel and writing one channel. A channel is FIFO, and one thread cannot reorder what it processes sequentially, so order in equals order out — through three stages and four channels.

This is exactly what breaks the moment a stage fans out to several workers: two threads pulling from one queue finish in whatever order they finish, and the sequence is gone. There is nothing wrong with fanning out — it is how you use more cores — but you then have to reconstruct the order deliberately, with index tagging. Single-threaded stages are the reason you can skip that here.

Shutdown cascades

Each stage owns its outbound Sender. When its inbound Receiver ends, the loop exits, the stage’s function returns, and the Sender drops — which ends the next stage’s receiver, and so on down the line. Drop the very first sender when the input is exhausted and the whole pipeline drains and stops by itself.

Handle a failed send too: if a downstream stage has gone away, send returns Err and the sensible response is to stop rather than panic.

The API

• sync_channel(n) -> (SyncSender<T>, Receiver<T>);
• SyncSender::send blocks when full, returning Err(SendError(value)) if the receiver is gone — and handing the value back;
• try_send never blocks and distinguishes TrySendError::Full (try again) from Disconnected (never again). Full is the signal a load-shedder acts on.

  Loading visualization…