Learn Rust Series (#63) - Channels: mpsc, Ownership Transfer, and Backpressure

@scipio 72

4 days ago

StemSocial

Learn Rust Series (#63) - Channels: mpsc, Ownership Transfer, and Backpressure

What will I learn

You will learn the "share by communicating" model and how std::sync::mpsc implements it;
how send moves ownership of a value into the channel, so there is nothing left to race on;
how multiple producers work by cloning the sender, and how the receiver ends when all senders drop;
how to iterate a receiver as a stream of values with for or rx.iter();
what backpressure is, and how a bounded sync_channel provides it.

Requirements

A working modern computer running macOS, Windows or Ubuntu;
An installed Rust toolchain (via rustup, from rustup.rs);
The previous sixty-two episodes, especially ownership, threads, and Send;
The ambition to learn systems programming from the ground up.

Difficulty

Intermediate

Curriculum (of the `Learn Rust Series`):

Learn Rust Series (#63) - Channels: mpsc, Ownership Transfer, and Backpressure

There is a famous slogan that came out of the Go community: "do not communicate by sharing memory; share memory by communicating". It is a lovely piece of advice, and it captures a genuine shift in how you think about concurrent programs. In stead of parking a lump of shared state behind a lock and having every thread reach in to poke at it, you give each thread its own data and let them pass messages back and forth down a pipe. The moving parts stop being "who holds the lock right now" and become "who is waiting for the next message". Last episode we shared data by borrowing it across scoped threads; today we start moving data between threads while they run, which is a completely different, and often much cleaner, style of concurrent design.

Rust takes that slogan and does the one thing Go cannot: it makes the safety structural rather than advisory. A channel is a one-way pipe with two ends. One end sends values, the other receives them, and here is the part that matters more than anything else in this episode -- send moves ownership of each value into the channel. Once you have sent something, you no longer have it. There is nothing left in your hand to race on, because the value now belongs to whoever pulls it out the other end. The standard library gives you this in std::sync::mpsc, which stands for multiple producer, single consumer, and it pairs beautifully with the threads we have been building up over the last two episodes ;-)

We first touched channels all the way back in episode 13, in the concurrency primer, but only in passing. Now that we understand ownership, Send, and scoped threads properly, we can look at them the way they deserve. First though, as always, last episode's homework.

Solutions to Episode 62 Exercises

Episode 62 was scoped threads, and all three exercises were about borrowing local data across thread::scope.

Exercise 1 asked you to sum two halves of a stack array [i64; 8] on two separate scoped threads, then add the two partial sums back in the main thread and print the total:

use std::thread;

fn main() {
    let data: [i64; 8] = [1, 2, 3, 4, 5, 6, 7, 8];
    let total = thread::scope(|s| {
        let left = s.spawn(|| data[..4].iter().sum::<i64>());  // borrows the first half
        let right = s.spawn(|| data[4..].iter().sum::<i64>()); // borrows the second half
        left.join().unwrap() + right.join().unwrap()           // add the partial sums
    });
    println!("total: {total}"); // total: 36
}

Both threads borrow data immutably at the same time, which is always sound -- shared reads never conflict. The whole thread::scope expression evaluates to the combined total, so nothing escapes the scope and data is never moved.

Exercise 2 wanted split_at_mut plus two scoped threads to negate every element in the first half of a slice and square every element in the second half, then print the mutated slice:

use std::thread;

fn main() {
    let mut v = [1, 2, 3, 4, 5, 6];
    let (left, right) = v.split_at_mut(3); // two &mut halves that cannot overlap
    thread::scope(|s| {
        s.spawn(|| { for x in left.iter_mut()  { *x = -*x;  } }); // negate the left half
        s.spawn(|| { for x in right.iter_mut() { *x *= *x; } });  // square the right half
    });
    println!("{v:?}"); // [-1, -2, -3, 16, 25, 36]
}

This is disjoint parallel mutation with no lock at all. split_at_mut proves the two halves cannot alias, so the borrow checker happily lets two threads write to them simultaneously -- the clobbering bug is not "guarded against", it is impossible to even express.

Exercise 3 was the harder one: spawn one scoped thread per line of a borrowed multi-line String (via .lines()), have each thread return its line's length, collect the lengths into a Vec, and print it:

use std::thread;

fn main() {
    let text = String::from("one\ntwo\nthree\nfour");
    let lengths: Vec<usize> = thread::scope(|s| {
        let handles: Vec<_> = text
            .lines()
            .map(|line| s.spawn(move || line.len())) // each thread borrows one &str line
            .collect();
        handles.into_iter().map(|h| h.join().unwrap()).collect()
    });
    println!("{lengths:?}"); // [3, 3, 5, 4]
}

Each &str line points straight into the one String buffer -- no copying -- and the scope guarantees every thread is joined before text could be dropped. Right, homework cleared. Now, channels.

The basic channel

mpsc::channel() returns a (Sender, Receiver) pair, conventionally named (tx, rx) for transmitter and receiver. The sender's send hands a value into the channel; the receiver's recv blocks the current thread until a value arrives. Move the sender into a spawned thread and you have thread-to-thread communication in five lines:

use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel();
    thread::spawn(move || {
        tx.send("hello from the worker").unwrap(); // the message moves into the channel
    });
    let msg = rx.recv().unwrap(); // blocks until a message arrives
    println!("{msg}"); // hello from the worker
}

Notice the shape. The worker thread owns tx (we moved it in), the main thread keeps rx, and the two communicate purely by passing a value down the pipe. No shared variable, no Arc, no Mutex -- the channel is the shared thing, and it is designed from the ground up to be safe to share.

recv returns a Result, and that is not bureaucratic noise. It errors -- returns Err(RecvError) -- when every sender has been dropped and no message will ever come again. That is the channel's clean, race-free way of saying "the other side hung up". You will lean on that signal constantly, because it is how a receiver knows a producer has finished for good rather than merely gone quiet for a moment.

recv, try_recv, and recv_timeout

recv blocks, which is usually exactly what you want. But sometimes a thread has other work to do and cannot afford to sit and wait. The Receiver gives you two non-committal alternatives. try_recv returns immediately: Ok(value) if one was waiting, or an Err telling you the channel is currently empty (but still open) versus disconnected (all senders gone). recv_timeout waits, but only up to a Duration:

use std::sync::mpsc;
use std::thread;
use std::time::Duration;

fn main() {
    let (tx, rx) = mpsc::channel();
    thread::spawn(move || {
        thread::sleep(Duration::from_millis(20));
        tx.send("late").unwrap();
    });

    match rx.try_recv() {
        Ok(v) => println!("got {v} straight away"),
        Err(_) => println!("nothing yet, doing other work"), // this branch runs first
    }

    let v = rx.recv_timeout(Duration::from_millis(50)).unwrap();
    println!("eventually got: {v}"); // eventually got: late
}

The distinction matters in real systems. A UI thread or a game loop cannot block on recv -- it must keep ticking -- so it polls with try_recv each frame and processes whatever has arrived. A worker that should give up if nothing comes for a while uses recv_timeout. And the plain blocking recv is for the common case where the thread's entire job is to wait for and process messages.

Ownership transfer is the safety story

Let us slow down on the single most important idea here, because it is why channels are safe rather than merely convenient. send moves the value. After you send something, the compiler considers it gone from your scope, exactly as if you had passed it to a function that took it by value. The producer no longer owns it, so the producer and consumer can never touch it at the same instant. Watch a String leave the sending thread entirely:

use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel();
    thread::spawn(move || {
        let owned = String::from("this String is moved, not shared");
        tx.send(owned).unwrap();
        // println!("{owned}"); // ERROR: borrow of moved value: `owned`
        // `owned` is gone here: the channel took ownership, so no aliasing is possible
    });
    let got = rx.recv().unwrap();
    println!("received {} bytes", got.len()); // received 32 bytes
}

Uncomment that middle line and the compiler stops you cold, the same way it did when we moved a value into a thread back in episode 61. The String's heap buffer was never copied -- only its three-word handle (pointer, length, capacity) moved down the channel -- and now it lives in the receiving thread and nowhere else. There is no lock here, no Arc, no shared mutable state, because there is no sharing at all: the data has exactly one owner at every point in time, and the type system enforces that as rigidly as it enforces any other move. That is the whole trick. A data race needs two threads reaching for the same memory; ownership transfer guarantees there is only ever one thread that can.

Multiple producers

The "mp" in mpsc is the interesting half: multiple producers. The Sender is Clone, and each clone feeds into the same single receiver. So you can hand a clone to each of several threads and let them all send concurrently, and the receiver merges the streams. The one rule to internalise is how the channel decides it is finished: the receiver stops yielding values once every sender has been dropped. Which means you must remember to drop the original tx after cloning, or the receiver will wait forever for a sender that exists but never sends:

use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel();
    for id in 0..3 {
        let tx = tx.clone(); // each producer gets its own sender
        thread::spawn(move || tx.send(id * 10).unwrap());
    }
    drop(tx); // drop the original so the receiver knows when everyone is finished

    let mut received: Vec<i32> = rx.iter().collect();
    received.sort();
    println!("{received:?}"); // [0, 10, 20]
}

That drop(tx) is genuinely load-bearing, and forgetting it is one of the most common channel bugs there is. The symptom is a program that produces all its output and then simply hangs, never exiting, because the receiver's iterator is still politely waiting for a sender that is never going to speak. When a channel program hangs at the end, "did I drop every extra sender?" is the very first question to ask. We sort the results here because the three threads race to send, so their arrival order is not deterministic -- another thing worth getting used to early.

The receiver as a stream

A Receiver implements Iterator, and this is where the ergonomics get lovely. Writing for value in rx (or rx.iter()) yields each message in turn and ends the loop automatically when the channel closes -- that is, when the last sender drops. Consuming a whole stream of values becomes a one-liner, and you never write an explicit "is it done yet" check:

use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel();
    thread::spawn(move || {
        for i in 1..=5 { tx.send(i).unwrap(); }
        // tx is dropped at the end of the closure, which ends the iteration below
    });
    let sum: i32 = rx.iter().sum(); // iterates until every sender is dropped, then stops
    println!("sum of the stream: {sum}"); // 15
}

The elegance here is that the channel's "the other side hung up" signal and the iterator's "I am exhausted" signal are the same event. When the worker's closure ends, tx drops, the channel closes, rx.iter() returns None, and sum finishes. All of that plumbing is invisible; you just wrote rx.iter().sum() as if it were any other iterator, and it happens to be fed live from another thread.

Backpressure with a bounded channel

There is a hidden danger in everything above: mpsc::channel() is unbounded. Its internal buffer grows without limit, so if a producer is faster than its consumer, messages pile up in memory indefinitely. Send a million items a second into a channel whose consumer handles a thousand a second, and you have written a memory leak that looks like a happy, busy program right up until the machine falls over. In a real pipeline that is a serious bug.

The fix is mpsc::sync_channel(n), a bounded channel that holds at most n unconsumed items. The difference is entirely in send: on a full buffer, send blocks the producer until the consumer removes an item and frees a slot. That blocking is the mechanism, and the effect it produces has a name -- backpressure. The consumer's pace pushes back up the pipe and throttles the producer to match:

use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::sync_channel(2); // at most 2 unconsumed items in flight
    let producer = thread::spawn(move || {
        for i in 0..5 {
            tx.send(i).unwrap(); // blocks once 2 items are queued, until the consumer catches up
            println!("sent {i}");
        }
    });

    let mut total = 0;
    for value in rx {
        total += value; // consumer sets the pace; the producer cannot outrun it
    }
    producer.join().unwrap();
    println!("total: {total}"); // total: 10
}

Backpressure is a feature, not a limitation, and this is a mindset worth adopting. An unbounded queue does not actually make a slow consumer faster -- it just hides the imbalance by spending memory, until it cannot. A bounded channel surfaces the imbalance honestly and keeps the pipeline's memory footprint fixed, coupling producer and consumer speeds automatically. As a special case, sync_channel(0) gives you a rendezvous channel: the buffer holds nothing at all, so each send blocks until a recv is ready to receive it hand-to-hand, forcing the two threads into lockstep.

A job/result pipeline

Put two channels together and you have the seed of every thread pool and work queue you will ever build. One channel carries jobs in to a worker; a second carries results out. The worker loops over incoming jobs with the iterator trick, and the "no more jobs" signal is, once again, simply dropping the job sender:

use std::sync::mpsc;
use std::thread;

fn main() {
    let (job_tx, job_rx) = mpsc::channel::<u32>();
    let (res_tx, res_rx) = mpsc::channel::<u32>();

    let worker = thread::spawn(move || {
        for job in job_rx {                 // process until the job sender is dropped
            res_tx.send(job * job).unwrap(); // send each result back out
        }
    });                                     // res_tx drops here, closing the result channel

    for n in 1..=4 { job_tx.send(n).unwrap(); }
    drop(job_tx);                           // signal "no more jobs"

    let mut squares: Vec<u32> = res_rx.iter().collect();
    worker.join().unwrap();
    squares.sort();
    println!("{squares:?}"); // [1, 4, 9, 16]
}

Trace the two shutdown signals, because they are the skeleton of the whole pattern. Dropping job_tx in main ends the worker's for job in job_rx loop; the loop ending drops res_tx inside the worker; and that ends the res_rx.iter() collection in main. Two channels, two clean "hung up" signals, and no explicit flags or sentinel values anywhere. Scale this up -- clone job_tx to many producers, or spawn several workers that all pull from a shared job queue -- and you have a real thread pool, which is precisely the kind of thing later episodes build.

How Python and Go would frame this

A glance sideways sharpens the picture, as it usually does. Python has queue.Queue, which is genuinely close in spirit: a thread-safe queue with an optional maxsize that gives you backpressure much like sync_channel. But the safety is entirely a matter of runtime discipline. Nothing stops you from keeping a reference to an object after you have put it on the queue and mutating it from two threads at once:

import queue, threading

q = queue.Queue(maxsize=2)   # bounded: put() blocks when full, like sync_channel(2)

def worker():
    while True:
        item = q.get()       # blocks until an item is available
        if item is None:     # a sentinel value you must invent yourself
            break
        print(item * item)

t = threading.Thread(target=worker)
t.start()
for n in range(1, 5):
    q.put(n)
q.put(None)                  # no "all senders dropped" signal -- you fake one
t.join()

Look at what Python cannot give you: there is no ownership transfer, so put(n) does not move anything, and there is no automatic "the channel closed" event, so you invent a None sentinel and hope nobody sends a real None. It works, but correctness lives in your head.

Go made channels a first-class language feature, and they are genuinely excellent -- a bounded make(chan int, 2), close() to signal completion, for v := range ch to consume. It is the closest cousin to Rust's design:

ch := make(chan int, 2)      // bounded channel, capacity 2
go func() {
    for n := 1; n <= 4; n++ {
        ch <- n * n          // blocks when the buffer is full
    }
    close(ch)                // signal "no more values"
}()
for v := range ch {          // ends automatically when ch is closed
    fmt.Println(v)
}

The ergonomics are lovely and the range-until-closed loop mirrors rx.iter() almost exactly. What Go does not give you is Rust's ownership guarantee: a value you send on a Go channel is still fully accessible to the sender afterward, so you can send a pointer and then race on what it points at, and only the -race detector might catch it at runtime. In Rust, send moved the value out of your hands, so that race is not a bug you avoid by being careful -- it is a program that does not compile.

Wrapping up

So there we have it: message passing, standard-library style. A channel is a one-way pipe where send moves ownership of each value in and recv (or the iterator) pulls it out the other end, and it is safe for the simplest possible reason -- once a value is sent, exactly one thread owns it, so there is nothing to race on. mpsc gives you multiple producers by cloning the Sender, and the receiver knows the show is over when the last sender drops, which is the signal behind both recv's Err and the iterator's None. When you need to bound memory, sync_channel(n) makes a full send block, and that blocking is backpressure -- a feature that keeps a pipeline honest rather than a limitation to work around.

The two things to carry forward are the ownership story and the drop-to-close story. Ownership transfer is why channels are safe; dropping the last sender is how a channel signals completion, and forgetting to drop a spare sender is the classic hang. Get those two reflexes wired in and channels become one of the most pleasant tools in the whole language.

Having said that, std::sync::mpsc has real limits. It is single-consumer, so only one receiver. There is no built-in way to select over several channels at once, and while it is perfectly good, it is not the fastest implementation you can get. The moment you need multiple consumers, a select across channels, or simply more speed, there is a well-loved crate that takes over from here -- and that is exactly where we are headed next ;-)

Exercises

Send the numbers 1 through 10 from a worker thread, one at a time, and sum them on the main thread using rx.iter().
Use three cloned senders, one moved into each of three threads, and collect all the results into a Vec that you then sort and print (remember to drop the original sender).
Build a job/result channel pair where the main thread sends several String jobs, the worker sends back each string uppercased, and the main thread collects the uppercased results.

Tot de volgende keer, en veel plezier met je eerste pipelines! ;-)

@scipio

stem stemsocial steemstem rust programming

0.000

0 comments

Learn Rust Series (#63) - Channels: mpsc, Ownership Transfer, and Backpressure

Learn Rust Series (#63) - Channels: mpsc, Ownership Transfer, and Backpressure

What will I learn

Requirements

Difficulty

Curriculum (of the Learn Rust Series):

Learn Rust Series (#63) - Channels: mpsc, Ownership Transfer, and Backpressure

Solutions to Episode 62 Exercises

The basic channel

recv, try_recv, and recv_timeout

Ownership transfer is the safety story

Multiple producers

The receiver as a stream

Backpressure with a bounded channel

A job/result pipeline

How Python and Go would frame this

Wrapping up

Exercises

Curriculum (of the `Learn Rust Series`):