Skip to content
Learn Rust
Theme

Channels

Concurrency · Lesson 43

Channels send values from one thread to another. Sending a value moves it into the channel.

Coming from Gompsc::channel() is a queue that grows, so send does not wait for the receiver the way a send on make(chan T) does. sync_channel(n) is make(chan T, n), and sync_channel(0) is the unbuffered kind. There is no close: the channel ends when every sender is dropped, and there is no select in the standard library.

Coming from KotlinA channel is like a LinkedBlockingQueue, but send moves the value in, so the sender can't touch it afterwards, and only one thread receives (a Kotlin coroutine Channel may have several receivers; a Rust Receiver cannot be shared). When every sender is dropped the channel closes and for msg in rx ends, so no "DONE" marker is needed.

Coming from PythonA channel is like queue.Queue, and for msg in rx stops by itself when every sender is gone, so you don't need a None marker (or shutdown(), new in Python 3.13). send moves the value, so the sender can't change it afterwards.

Coming from TypeScriptA channel is like postMessage between workers, except that postMessage copies the data (unless you transfer it) and send moves it. for msg in rx blocks a thread, with no callback, until every sender has been dropped; a MessagePort has no such count.

Coming from Javampsc::channel() is a LinkedBlockingQueue, but send moves the value, so the sender can't touch it afterwards. The channel closes by itself when every sender is dropped, so no DONE marker is needed.

Coming from C#mpsc::channel() is Channel.CreateUnbounded<T>(), and for msg in rx is await foreach over ReadAllAsync(), except it blocks a thread instead of awaiting. Dropping the last tx closes it like Writer.Complete(), and send moves the value so the sender can't change it later.

Coming from CC has no channels, so you'd build a queue with a mutex and a condition variable (or use a pipe for bytes). mpsc::channel() is that queue, ready made; it closes when the last sender is dropped, and the compiler stops the sender from using a value after sending it.

Coming from C++The standard library has no channel, so mpsc::channel() replaces your mutex, condition_variable and queue class. send moves like std::move, but Rust then rejects any later use of the moved-from variable, which C++ accepts.

The quick comparison

Rustsrc/main.rsRun ↗
use std::sync::mpsc;
use std::thread;


fn main() {
    let (tx, rx) = mpsc::channel(); // tx sends, rx receives


    thread::spawn(move || {
        for i in 1..=3 {
            let msg = format!("job {i}");
            tx.send(msg).unwrap(); // msg moves into the channel
        }
    }); // tx goes away here: channel closes


    // ends when the channel closes
    for msg in rx {
        println!("got {msg}");
    }
}
$ cargo run
got job 1
got job 2
got job 3
package main


import "fmt"


func main() {
  ch := make(chan string)


  go func() {
    for i := 1; i <= 3; i++ {
      ch <- fmt.Sprintf("job %d", i)
    }
    close(ch)
  }()


  for msg := range ch {
    fmt.Println("got", msg)
  }
}
$ go run main.go
got job 1
got job 2
got job 3
import java.util.concurrent.LinkedBlockingQueue
import kotlin.concurrent.thread


fun main() {
    val queue = LinkedBlockingQueue<String>()


    val worker = thread {
        for (i in 1..3) {
            queue.put("job $i")
        }
        queue.put("DONE")      // no built-in close, so send a marker
    }


    while (true) {
        val msg = queue.take()
        if (msg == "DONE") break
        println("got $msg")
    }
    worker.join()
}
$ kotlinc Main.kt -include-runtime -d main.jar && java -jar main.jar
got job 1
got job 2
got job 3
import queue
import threading


q: queue.Queue[str | None] = queue.Queue()




def produce() -> None:
    for i in range(1, 4):
        q.put(f"job {i}")
    q.put(None)  # no "close" on a Queue, so send a marker




threading.Thread(target=produce).start()


while (msg := q.get()) is not None:
    print(f"got {msg}")
$ python3 main.py
got job 1
got job 2
got job 3
import { Worker, isMainThread, parentPort } from "node:worker_threads";


if (isMainThread) {
  const worker = new Worker(__filename);
  // messages arrive in the order they were sent
  worker.on("message", (msg: string) => console.log(`got ${msg}`));
} else {
  for (let i = 1; i <= 3; i++) {
    parentPort!.postMessage(`job ${i}`); // the message is copied across
  }
} // the worker ends here and the program exits once its messages are handled
$ tsc --strict --module nodenext --types node main.ts && node main.js
got job 1
got job 2
got job 3
import java.util.concurrent.LinkedBlockingQueue;


public class Main {
    public static void main(String[] args) throws InterruptedException {
        var queue = new LinkedBlockingQueue<String>();


        var worker = Thread.startVirtualThread(() -> {
            for (int i = 1; i <= 3; i++) {
                queue.add("job " + i);
            }
            queue.add("DONE");   // queues can't be closed, so send a marker
        });


        while (true) {
            var msg = queue.take();
            if (msg.equals("DONE")) break;
            System.out.println("got " + msg);
        }
        worker.join();
    }
}
$ javac Main.java && java Main
got job 1
got job 2
got job 3
using System.Threading.Channels;


var ch = Channel.CreateUnbounded<string>();


_ = Task.Run(async () =>
{
    for (var i = 1; i <= 3; i++)
    {
        await ch.Writer.WriteAsync($"job {i}");
    }
    ch.Writer.Complete();            // close the channel
});


await foreach (var msg in ch.Reader.ReadAllAsync())   // ends when it closes
{
    Console.WriteLine($"got {msg}");
}
$ dotnet run
got job 1
got job 2
got job 3
#include <assert.h>
#include <pthread.h>
#include <stdbool.h>
#include <stdio.h>
#include <stdlib.h>


// C has no channels: a small queue guarded by a mutex and a condition variable
struct Channel {
    pthread_mutex_t lock;
    pthread_cond_t ready;
    char *items[8];      // a sketch: room for 8 messages in all, no wrap-around
    int head, tail;
    bool closed;
};


void send(struct Channel *ch, char *msg) {
    pthread_mutex_lock(&ch->lock);
    assert(ch->tail < 8);          // a real queue would wait here until there is room
    ch->items[ch->tail++] = msg;   // the receiver now owns msg
    pthread_cond_signal(&ch->ready);
    pthread_mutex_unlock(&ch->lock);
}


void close_channel(struct Channel *ch) {
    pthread_mutex_lock(&ch->lock);
    ch->closed = true;
    pthread_cond_broadcast(&ch->ready);   // wake every receiver
    pthread_mutex_unlock(&ch->lock);
}


// returns NULL once the channel is closed and empty
char *recv(struct Channel *ch) {
    pthread_mutex_lock(&ch->lock);
    while (ch->head == ch->tail && !ch->closed) {
        pthread_cond_wait(&ch->ready, &ch->lock);
    }
    char *msg = ch->head < ch->tail ? ch->items[ch->head++] : NULL;
    pthread_mutex_unlock(&ch->lock);
    return msg;
}


void *producer(void *arg) {
    struct Channel *ch = arg;
    for (int i = 1; i <= 3; i++) {
        char *msg = malloc(16);
        snprintf(msg, 16, "job %d", i);
        send(ch, msg);
    }
    close_channel(ch);
    return NULL;
}


int main(void) {
    struct Channel ch = {
        .lock = PTHREAD_MUTEX_INITIALIZER,
        .ready = PTHREAD_COND_INITIALIZER,
    };
    pthread_t t;
    pthread_create(&t, NULL, producer, &ch);


    char *msg;
    while ((msg = recv(&ch)) != NULL) {   // ends when the channel closes
        printf("got %s\n", msg);
        free(msg);
    }
    pthread_join(t, NULL);
    return 0;
}
$ gcc -Wall -Wextra main.c -o main && ./main
got job 1
got job 2
got job 3
#include <condition_variable>
#include <format>
#include <iostream>
#include <mutex>
#include <optional>
#include <queue>
#include <string>
#include <thread>
#include <utility>


// The standard library has no channel, so here is a small one.
template <class T>
class Channel {
public:
    void send(T value) {
        { std::lock_guard lock(m_); q_.push(std::move(value)); }
        cv_.notify_one();
    }
    void close() {
        { std::lock_guard lock(m_); closed_ = true; }
        cv_.notify_all();
    }
    std::optional<T> recv() {            // empty optional once closed and drained
        std::unique_lock lock(m_);
        cv_.wait(lock, [&] { return !q_.empty() || closed_; });
        if (q_.empty()) return std::nullopt;
        T value = std::move(q_.front());
        q_.pop();
        return value;
    }


private:
    std::mutex m_;
    std::condition_variable cv_;
    std::queue<T> q_;
    bool closed_ = false;
};


int main() {
    Channel<std::string> ch;


    std::jthread producer([&ch] {
        for (int i = 1; i <= 3; ++i) {
            ch.send(std::format("job {}", i));   // moved into the channel
        }
        ch.close();                              // no more messages
    });


    while (auto msg = ch.recv()) {               // ends when the channel closes
        std::cout << "got " << *msg << "\n";
    }
}
$ g++ -std=c++20 -Wall -Wextra main.cpp -o main && ./main
got job 1
got job 2
got job 3

What's different

  • mpsc means multiple producers, single consumer. Clone tx to have several senders; the receiver cannot be cloned.
  • You never call close. The channel closes when every sender has gone away, here when the thread ends, and for msg in rx ends with it.
  • send returns a Result, because the receiving side might already be gone. .unwrap() here means "crash if nobody is listening".
Gotcha

Changing a value after sending it

Looks fine: in TypeScript it runs fine; in C++ it runs but gives a wrong answer; in Go, Kotlin, Python, Java, C# and C it runs but has a data race. Rust won't compile it.

Broken in Go. Rust catches it.Broken in Kotlin. Rust catches it.Broken in Python. Rust catches it.Works in TypeScript. In Rust it's just a rule.Broken in Java. Rust catches it.Broken in C#. Rust catches it.Broken in C. Rust catches it.Broken in C++. Rust catches it.
Rustsrc/main.rs✗ won't compile
use std::sync::mpsc;
use std::thread;


fn main() {
    let (tx, rx) = mpsc::channel();
    thread::spawn(move || {
        let mut msg = vec![1, 2];
        tx.send(msg).unwrap();
        msg.push(3);
    });
    println!("{:?}", rx.recv().unwrap());
}
package main


import (
  "fmt"
  "time"
)


func main() {
  ch := make(chan []int)
  done := make(chan bool)
  go func() {
    got := <-ch
    time.Sleep(10 * time.Millisecond)
    fmt.Println("receiver sees", got)
    done <- true
  }()
  msg := []int{1, 2}
  ch <- msg
  msg[0] = 99 // sender keeps changing what it already sent
  <-done
}
$ go run -race main.go! runs · data race
==================
WARNING: DATA RACE
Read at 0x············ by goroutine 8:
  runtime.typedmemmove()
      /opt/go/src/runtime/mbarrier.go:152 +0x7c
  reflect.packEfaceData()
      /opt/go/src/reflect/value.go:143 +0xe8
  reflect.packEface()
      /opt/go/src/reflect/value.go:126 +0x15c
  reflect.valueInterface()
      /opt/go/src/reflect/value.go:1512 +0x1b0
  reflect.Value.Interface()
      /opt/go/src/reflect/value.go:1490 +0x78
  fmt.(*pp).printValue()
      /opt/go/src/fmt/print.go:759 +0x64
  fmt.(*pp).printValue()
      /opt/go/src/fmt/print.go:898 +0x1a68
  fmt.(*pp).printArg()
      /opt/go/src/fmt/print.go:749 +0x8ec
  fmt.(*pp).doPrintln()
      /opt/go/src/fmt/print.go:1205 +0x40
  fmt.Fprintln()
      /opt/go/src/fmt/print.go:297 +0x48
  fmt.Println()
      /opt/go/src/fmt/print.go:307 +0xe0
  main.main.func1()
      main.go:14 +0x64

Previous write at 0x············ by main goroutine:
  main.main()
      main.go:19 +0x12c

Goroutine 8 (running) created at:
  main.main()
      main.go:11 +0xdc
==================
receiver sees [99 2]
Found 1 data race(s)
import java.util.concurrent.LinkedBlockingQueue
import kotlin.concurrent.thread


fun main() {
    val queue = LinkedBlockingQueue<MutableList<Int>>()
    val receiver = thread {
        val got = queue.take()
        Thread.sleep(10)
        println("receiver sees $got")
    }
    val msg = mutableListOf(1, 2)
    queue.put(msg)
    msg[0] = 99     // sender keeps changing what it already sent
    receiver.join()
}
$ kotlinc Main.kt -include-runtime -d main.jar && java -jar main.jar! runs · data race
receiver sees [99, 2]
import queue
import threading
import time


q = queue.Queue()




def receive():
    got = q.get()
    time.sleep(0.01)
    print("receiver sees", got)




r = threading.Thread(target=receive)
r.start()
msg = [1, 2]
q.put(msg)
msg[0] = 99  # sender keeps changing what it already sent
r.join()
$ python3 main.py! runs · data race
receiver sees [99, 2]
import { Worker, isMainThread, parentPort } from "node:worker_threads";


if (isMainThread) {
  const worker = new Worker(__filename);
  worker.on("message", (msg: number[]) => console.log(msg));
} else {
  const msg = [1, 2];
  parentPort!.postMessage(msg);
  msg.push(3); // change it after sending
}
$ tsc --strict --module nodenext --types node main.ts && node main.js✓ runs
[ 1, 2 ]
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;


public class Main {
    public static void main(String[] args) throws InterruptedException {
        var queue = new LinkedBlockingQueue<List<Integer>>();
        Thread.startVirtualThread(() -> {
            var msg = new ArrayList<>(List.of(1, 2));
            queue.add(msg);
            msg.add(3);   // sender keeps changing what it already sent
        });
        var got = queue.take();
        Thread.sleep(100);
        System.out.println(got);
    }
}
$ javac Main.java && java Main! runs · data race
[1, 2, 3]
using System.Threading.Channels;


var ch = Channel.CreateUnbounded<List<int>>();


var receiver = Task.Run(async () =>
{
    var got = await ch.Reader.ReadAsync();
    await Task.Delay(50);   // a little work before looking at it
    Console.WriteLine($"receiver sees [{string.Join(", ", got)}]");
});


List<int> msg = [1, 2];
await ch.Writer.WriteAsync(msg);
msg[0] = 99;   // sender keeps changing what it already sent
await receiver;
$ dotnet run! runs · data race
receiver sees [99, 2]
#define _POSIX_C_SOURCE 200809L  // for nanosleep
#include <pthread.h>
#include <stdio.h>
#include <time.h>


// a one-slot "channel": a pointer handed over under a lock
pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t ready = PTHREAD_COND_INITIALIZER;
int *slot = NULL;


void sleep_ms(long ms) {
    struct timespec ts = {0, ms * 1000000L};
    nanosleep(&ts, NULL);
}


void *receiver(void *arg) {
    (void)arg;
    pthread_mutex_lock(&lock);
    while (slot == NULL) pthread_cond_wait(&ready, &lock);
    int *got = slot;
    pthread_mutex_unlock(&lock);


    sleep_ms(10);
    printf("receiver sees [%d, %d]\n", got[0], got[1]);
    return NULL;
}


int main(void) {
    pthread_t t;
    pthread_create(&t, NULL, receiver, NULL);


    int msg[2] = {1, 2};
    pthread_mutex_lock(&lock);
    slot = msg;                  // send
    pthread_cond_signal(&ready);
    pthread_mutex_unlock(&lock);


    msg[0] = 99;                 // sender keeps changing what it already sent
    pthread_join(t, NULL);
    return 0;
}
$ gcc -fsanitize=thread -g main.c -o main && ./main! runs · data race
==================
WARNING: ThreadSanitizer: data race (pid=1)
  Read of size 4 at 0x············ by thread T1:
    #0 receiver main.c:24 (main+0xd74)

  Previous write of size 4 at 0x············ by main thread:
    #0 main main.c:38 (main+0xe64)

  As if synchronized via sleep:
    #0 nanosleep ../../../../src/libsanitizer/tsan/tsan_interceptors_posix.cpp:397 (libtsan.so.2+0x59f60)
    #1 sleep_ms main.c:13 (main+0xcc0)
    #2 receiver main.c:23 (main+0xd6c)

  Location is stack of main thread.

  Location is global '<null>' at 0x············ ([stack]+0x209b0)

  Thread T1 (tid=1, running) created by main thread at:
    #0 pthread_create ../../../../src/libsanitizer/tsan/tsan_interceptors_posix.cpp:1022 (libtsan.so.2+0x5bbbc)
    #1 main main.c:30 (main+0xdf8)

SUMMARY: ThreadSanitizer: data race main.c:24 in receiver
==================
receiver sees [99, 2]
ThreadSanitizer: reported 1 warnings
#include <condition_variable>
#include <iostream>
#include <mutex>
#include <queue>
#include <thread>
#include <utility>
#include <vector>


template <class T>
class Channel {   // the same small channel as the lesson, cut down
public:
    void send(T v) {
        { std::lock_guard lock(m_); q_.push(std::move(v)); }
        cv_.notify_one();
    }
    T recv() {
        std::unique_lock lock(m_);
        cv_.wait(lock, [&] { return !q_.empty(); });
        T v = std::move(q_.front());
        q_.pop();
        return v;
    }
private:
    std::mutex m_;
    std::condition_variable cv_;
    std::queue<T> q_;
};


void print(const char* label, const std::vector<int>& xs) {
    std::cout << label << " [";
    for (std::size_t i = 0; i < xs.size(); ++i) std::cout << (i ? ", " : "") << xs[i];
    std::cout << "]\n";
}


int main() {
    Channel<std::vector<int>> ch;
    std::vector<int> msg{1, 2};
    std::thread sender([&] {
        ch.send(std::move(msg));
        msg.push_back(3);
    });
    print("received", ch.recv());
    sender.join();
    print("sender's msg is now", msg);
}
$ g++ -std=c++20 -Wall -Wextra main.cpp -o main && ./main! runs · wrong answer
received [1, 2]
sender's msg is now [3]
What does the compiler say? Guess first, then open.
$ cargo buildWhat is E0382?
error[E0382]: borrow of moved value: `msg`
 --> src/main.rs:9:9
  |
7 |         let mut msg = vec![1, 2];
  |             ------- move occurs because `msg` has type `Vec<i32>`, which does not implement the `Copy` trait
8 |         tx.send(msg).unwrap();
  |                 --- value moved here
9 |         msg.push(3);
  |         ^^^ value borrowed here after move
What happens in Go

Go printed receiver sees [99 2]. The channel passed a slice that shares its array, so the sender's later change showed up on the other side. Built with the race detector, as here, it also reports a data race.

What happens in Kotlin

Kotlin printed receiver sees [99, 2]. The queue passed the same list object, not a copy, so the sender's change after put showed up on the other side.

What happens in Python

Python prints receiver sees [99, 2]. The queue passes the same list object, not a copy, so the sender's change after put shows up on the other side.

What happens in TypeScript

Prints [ 1, 2 ]. postMessage copies the array when you send it, so the later push only changes the sender's copy.

What happens in Java

The queue passes a reference, not a copy, so the receiver sees the 3 the sender added after sending. Nothing stops both threads from touching the same list.

What happens in C#

The channel passes a reference to the same List, so the sender's later change shows up and the receiver sees [99, 2]. Send a copy, or stop touching it after sending.

What happens in C

C hands over a pointer, so the sender still has the same array. The receiver sees 99, and ThreadSanitizer reports the data race.

What happens in C++

This compiles cleanly. std::move hands the vector to the channel and leaves msg in a valid but unspecified state (empty with libstdc++), so push_back(3) lands in a leftover vector the receiver never sees.

What Rust sees

send(msg) moves msg into the channel. After that line, the sending thread doesn't own it anymore, so it can't touch it.

Why Rust says no

Once you send something, it belongs to the receiver. The sender can't touch it anymore, so two threads never fight over it, and the compiler checks this for you.

The fix

Compiles and runs.

Rust, fixedsrc/main.rsRun ↗
use std::sync::mpsc;
use std::thread;


fn main() {
    let (tx, rx) = mpsc::channel();
    thread::spawn(move || {
        let mut msg = vec![1, 2];
        msg.push(3); // finish building it first
        tx.send(msg).unwrap(); // then hand it over
    });
    println!("{:?}", rx.recv().unwrap());
}
$ cargo run
[1, 2, 3]

The toolkit

Outputs on this page come from Linux on 64-bit ARM (the VM had 8 cores) with Rust 1.99.1, built in debug mode. Every program here prints the same thing on every run: it joins its threads, then sorts or counts before it prints. The Rust tables are asserted by programs the site runs. Threads here are OS threads, not goroutines or virtual threads, and recv blocks one. Threads introduced spawn and join; Shared state covers Arc and Mutex.

A channel is a queue with two ends

mpsc::channel() returns a pair: a Sender (tx) that puts values in, and a Receiver (rx) that takes them out, first in, first out. mpsc stands for “multiple producer, single consumer”: you can clone the sender, never the receiver, so one thread receives. Go channels, Java queues and C# channels accept several consumers; mpsc does not, and the pool shows what that costs.

Rustsrc/main.rsRun ↗
use std::sync::mpsc;
use std::thread;


fn main() {
    // tx is the sending end, rx the receiving end. T is inferred from the first send.
    let (tx, rx) = mpsc::channel();


    let worker = thread::spawn(move || {
        for i in 1..=3 {
            tx.send(format!("job {i}")).unwrap();
        }
        // tx is dropped here, when the closure ends
    });


    for _ in 0..3 {
        println!("recv: {:?}", rx.recv());
    }
    worker.join().unwrap();


    // Nothing is queued and the only sender is gone, so waiting would be pointless.
    println!("recv after the worker ended: {:?}", rx.recv());
}
$ cargo run
recv: Ok("job 1")
recv: Ok("job 2")
recv: Ok("job 3")
recv after the worker ended: Err(RecvError)

send and recv both return a Result. recv waits for a value, and returns Err only when the queue is empty and every sender is gone.

Sending moves the value

send(value) takes value by value, so the sender no longer owns it (the gotcha is that error, with a Vec). A Copy integer is copied, so the variable stays usable. A Vec moves, and what moves is its three words; the numbers stay where they were:

Rustsrc/main.rsRun ↗
use std::mem::size_of;
use std::sync::mpsc;
use std::thread;


fn main() {
    // A Copy value: send copies it, so n is still usable.
    let (tx, rx) = mpsc::channel();
    let n = 7;
    tx.send(n).unwrap();
    println!("n after send: {n}, received: {}", rx.recv().unwrap());


    // A Vec: send moves it. The Vec is 3 words; its numbers stay where they were.
    let (tx, rx) = mpsc::channel::<Vec<i32>>();
    let v = vec![10, 20, 30];
    let before = v.as_ptr();
    println!("size_of::<Vec<i32>>(): {} bytes", size_of::<Vec<i32>>());


    tx.send(v).unwrap(); // v is moved: using it here would not compile
    let got = thread::spawn(move || rx.recv().unwrap()).join().unwrap();


    println!("received: {got:?}");
    println!(
        "same heap buffer after the trip: {}",
        got.as_ptr() == before
    );
}
$ cargo run
n after send: 7, received: 7
size_of::<Vec<i32>>(): 24 bytes
received: [10, 20, 30]
same heap buffer after the trip: true
tx.send(v) v is gone, the channel holds the Vec, the numbers never moved
STACKHEAPv: Vec<i32> (moved into the channel)ptrlen 3cap 3in the channel: a queued messageptrlen 3cap 33 × i32102030
Between the send and the recv, the channel holds the three words of the Vec. The program printed same heap buffer after the trip: true. Where the channel keeps its queue is an implementation detail; this picture claims only what travel.rs printed.

While a value is in flight one thread owns it, so it needs no lock. It must be Send, “allowed to move to another thread” (Send and Sync). An Rc is not, and the error comes from thread::spawn, because the closure holds the Receiver:

Rustsrc/main.rs✗ won't compile
use std::rc::Rc;
use std::sync::mpsc;
use std::thread;


fn main() {
    let (tx, rx) = mpsc::channel();
    tx.send(Rc::new(5)).unwrap();


    let reader = thread::spawn(move || {
        println!("got {}", rx.recv().unwrap());
    });
    reader.join().unwrap();
}
$ cargo buildWhat is E0277?
error[E0277]: `Rc<i32>` cannot be sent between threads safely
  --> src/main.rs:9:32
   |
 9 |       let reader = thread::spawn(move || {
   |  __________________-------------_^
   | |                  |
   | |                  required by a bound introduced by this call
10 | |         println!("got {}", rx.recv().unwrap());
11 | |     });
   | |_____^ `Rc<i32>` cannot be sent between threads safely
   |
   = help: the trait `Send` is not implemented for `Rc<i32>`
   = note: required for `std::sync::mpsc::Receiver<Rc<i32>>` to implement `Send`
note: required because it's used within this closure
  --> src/main.rs:9:32
   |
 9 |     let reader = thread::spawn(move || {
   |                                ^^^^^^^
  ... (notes pointing into the standard library trimmed)

Read the notes in order: Rc<i32> is not Send, so Receiver<Rc<i32>> is not, so the closure that owns it is not, and thread::spawn demands a Send closure (Reading signatures).

Several producers

tx.clone() makes another sender for the same channel. Give each thread its own (Go calls this fan-in):

Rustsrc/main.rsRun ↗
use std::sync::mpsc;
use std::thread;


fn main() {
    let (tx, rx) = mpsc::channel();


    let mut workers = Vec::new();
    for id in 1..=3 {
        let tx = tx.clone(); // one more sender, same channel
        workers.push(thread::spawn(move || {
            for step in 1..=3 {
                tx.send((id, step)).unwrap();
            }
        }));
    }
    drop(tx); // the original sender; without this the loop below would never end


    let mut all = Vec::new();
    let mut last_step = [0; 4];
    for (id, step) in rx {
        // messages from ONE sender arrive in the order it sent them
        assert_eq!(step, last_step[id] + 1);
        last_step[id] = step;
        all.push((id, step));
    }
    for w in workers {
        w.join().unwrap();
    }


    // the interleaving of different senders changes from run to run: sort before printing
    all.sort();
    println!("received {} messages", all.len());
    println!("sorted: {all:?}");
}
$ cargo run
received 9 messages
sorted: [(1, 1), (1, 2), (1, 3), (2, 1), (2, 2), (2, 3), (3, 1), (3, 2), (3, 3)]

One sender’s messages arrive in the order it sent them (the assertion checks it). Different senders interleave differently on every run, so the program sorts before it prints. If order matters, put it in the message, as the first exercise does with an index.

When does for msg in rx end?

When the queue is empty and every sender has been dropped. The classic mistake is a main thread that keeps its own tx while it loops on rx: the workers finish, one sender is still alive, and the loop waits for a message that cannot come. No error is reported; the loop prints what arrived and then stops making progress. The mpsc docs say the program “will not complete” without the drop. A hang cannot be shown here, so this is the state just before it, measured with calls that give up:

Rustsrc/main.rsRun ↗
use std::sync::mpsc::{self, TryRecvError};
use std::thread;
use std::time::Duration;


fn main() {
    let (tx, rx) = mpsc::channel();


    // Two workers, each with its own clone. We join them, so both are finished.
    let workers: Vec<_> = (1..=2)
        .map(|id| {
            let tx = tx.clone();
            thread::spawn(move || tx.send(format!("from worker {id}")).unwrap())
        })
        .collect();
    for w in workers {
        w.join().unwrap();
    }


    // The workers are gone, but main still holds `tx`.
    let mut msgs = vec![rx.try_recv().unwrap(), rx.try_recv().unwrap()];
    msgs.sort();
    println!("queued messages: {msgs:?}");
    println!("try_recv, tx alive: {:?}", rx.try_recv());
    println!(
        "recv_timeout, tx alive: {:?}",
        rx.recv_timeout(Duration::from_millis(50))
    );
    // `for msg in rx` would wait here for ever: a sender exists that nobody will use.


    drop(tx);
    println!("try_recv, tx dropped: {:?}", rx.try_recv());
    println!("recv, tx dropped: {:?}", rx.recv());
    assert_eq!(rx.try_recv(), Err(TryRecvError::Disconnected));
}
$ cargo run
queued messages: ["from worker 1", "from worker 2"]
try_recv, tx alive: Err(Empty)
recv_timeout, tx alive: Err(Timeout)
try_recv, tx dropped: Err(Disconnected)
recv, tx dropped: Err(RecvError)

Empty means “nothing now, but a sender exists”, Disconnected “nothing, ever”, and Timeout is recv_timeout giving up. All the ways to receive:

Call Queue has a message Empty, a sender exists Empty, every sender dropped
recv() the message waits Err(RecvError)
try_recv() the message Err(Empty) Err(Disconnected)
recv_timeout(d) the message waits up to d, then Err(Timeout) Err(Disconnected)
for m in rx, rx.iter() each message waits the loop ends
rx.try_iter() each queued message ends ends

“Waits” is the docs’ word (“will always block the current thread”); every other cell is asserted. Whoever makes the channel holds the first tx: move it into the last worker or drop(tx) before you loop.

The other side can hang up too

Dropping the receiver makes send fail, and the value comes back in the error:

Rustsrc/main.rsRun ↗
use std::sync::mpsc;


fn main() {
    let (tx, rx) = mpsc::channel();
    tx.send(String::from("first")).unwrap(); // the receiver still exists
    drop(rx);


    let err = tx.send(String::from("second")).unwrap_err();
    println!("error: {err}");
    println!("the value comes back: {:?}", err.0);


    // A producer can use this to notice that nobody is listening and stop.
    let mut sent = 0;
    for i in 0..5 {
        if tx.send(i.to_string()).is_err() {
            break;
        }
        sent += 1;
    }
    println!("sent after the receiver left: {sent}");
}
$ cargo run
error: sending on a closed channel
the value comes back: "second"
sent after the receiver left: 0

Examples write .unwrap() after a send to mean “crash if nobody is listening”; a producer usually does better to treat the error as “stop” (the third exercise). A thread that panics drops what it owned, its senders included, and so does a function that returns early:

Rustsrc/main.rsRun ↗
use std::sync::mpsc;
use std::thread;


fn main() {
    std::panic::set_hook(Box::new(|_| {})); // keep the panic message out of this demo


    let (tx, rx) = mpsc::channel();
    let worker = thread::spawn(move || {
        tx.send(1).unwrap();
        tx.send(2).unwrap();
        panic!("worker failed"); // unwinding drops tx
    });


    // The loop ends, because the panicked thread's sender was dropped like any other.
    let received: Vec<i32> = rx.iter().collect();
    println!("received before the failure: {received:?}");
    println!("worker.join() returned Err: {}", worker.join().is_err());
}
$ cargo run
received before the failure: [1, 2]
worker.join() returned Err: true

(A panic hook silences the message. The loop ended, and join returned Err.)

What was dropped The other side sees
one of several Senders nothing
the last Sender queued messages first, then Err / Disconnected / the loop ends
the Receiver send returns Err(SendError(value)) on every clone of the sender

Bounded channels and back-pressure

The docs call channel() “infinitely buffered”: a producer faster than its consumer makes the queue grow without a limit. sync_channel(n) holds at most n messages: send waits while it is full, and try_send returns Full(value) instead of waiting. With 0 it is a rendezvous: each send completes only when a receiver takes the value.

Rustsrc/main.rsRun ↗
use std::sync::mpsc::{self, TrySendError};


fn main() {
    // channel(): the queue grows. Sending never waits for the receiver.
    let (tx, rx) = mpsc::channel();
    for i in 0..100_000 {
        tx.send(i).unwrap();
    }
    println!(
        "channel(): 100000 sends done, {} waiting",
        rx.try_iter().count()
    );


    // sync_channel(2): room for two messages. try_send says "full" instead of waiting.
    let (tx, rx) = mpsc::sync_channel(2);
    println!("try_send 1: {:?}", tx.try_send(1));
    println!("try_send 2: {:?}", tx.try_send(2));
    println!("try_send 3: {:?}", tx.try_send(3));
    println!("recv: {:?}", rx.recv());
    println!("try_send 3 again: {:?}", tx.try_send(3));
    if let Err(TrySendError::Full(value)) = tx.try_send(4) {
        println!("full: the value {value} is handed back");
    }


    // sync_channel(0): a rendezvous. A send completes only when a receiver takes it.
    let (tx, _rx) = mpsc::sync_channel(0);
    println!("try_send 1 on sync_channel(0): {:?}", tx.try_send(1));
}
$ cargo run
channel(): 100000 sends done, 100000 waiting
try_send 1: Ok(())
try_send 2: Ok(())
try_send 3: Err(TrySendError::Full(..))
recv: Ok(1)
try_send 3 again: Ok(())
full: the value 4 is handed back
try_send 1 on sync_channel(0): Err(TrySendError::Full(..))

That limit is back-pressure: the producer is held to the consumer’s pace. This program asserts it at every receive, for three bounds, then shows channel() letting the producer run ahead:

Rustsrc/main.rsRun ↗
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, mpsc};
use std::thread;


// How many sends have returned by the time the receiver has taken `taken` messages?
// With a bound of `cap` the producer cannot be more than `cap` ahead.
fn run(cap: usize) {
    let (tx, rx) = mpsc::sync_channel(cap);
    let done = Arc::new(AtomicUsize::new(0));
    let producer = {
        let done = Arc::clone(&done);
        thread::spawn(move || {
            for i in 0..1000 {
                tx.send(i).unwrap();
                done.fetch_add(1, Ordering::SeqCst); // this send has returned
            }
        })
    };
    let mut taken = 0;
    for _ in rx {
        taken += 1;
        assert!(done.load(Ordering::SeqCst) <= taken + cap);
    }
    producer.join().unwrap();
    println!("sync_channel({cap}): never more than {cap} ahead of the receiver");
}


fn main() {
    for cap in [0, 1, 4] {
        run(cap);
    }


    // channel() has no limit: the producer finishes before anyone receives.
    let (tx, rx) = mpsc::channel();
    thread::spawn(move || (0..1000).for_each(|i| tx.send(i).unwrap()))
        .join()
        .unwrap();
    println!("channel(): producer finished 1000 sends, receiver has taken 0");
    println!("then the receiver finds {}", rx.try_iter().count());
}
$ cargo run
sync_channel(0): never more than 0 ahead of the receiver
sync_channel(1): never more than 1 ahead of the receiver
sync_channel(4): never more than 4 ahead of the receiver
channel(): producer finished 1000 sends, receiver has taken 0
then the receiver finds 1000

try_send exists only on SyncSender, a different type from Sender: a function that takes one does not accept the other.

More than numbers

A message is any Send value. A channel can even carry an Arc<Mutex<..>>, and then both threads share memory again, with the lock visible in the type; Shared state says when a lock is the simpler tool. Channels are a design tool:

  • an enum for several kinds of message, which also merges several sources into one receiver: one tx clone per source, the variant says who sent it;
  • a Sender inside a request, so the other thread knows where to reply;
  • a Result, so an error crosses to another thread as data;
  • a boxed closure, Box<dyn FnOnce() + Send>, for “run this” (Closures, Trait objects).

Here one thread owns a HashMap, so no lock exists; other code asks it by message:

Rustsrc/main.rsRun ↗
use std::collections::HashMap;
use std::sync::mpsc::{self, Sender};
use std::thread;


// Messages are an enum: the usual way to send "several kinds of thing".
enum Cmd {
    Set(String, i32),
    // A request carries a Sender for the reply.
    Get(String, Sender<Option<i32>>),
    Divide(i32, Sender<Result<i32, String>>), // errors cross threads as values
}


fn main() {
    let (tx, rx) = mpsc::channel();


    // This thread owns the map. No other thread can reach it, so there is no lock.
    let owner = thread::spawn(move || {
        let mut map = HashMap::new();
        for cmd in rx {
            match cmd {
                Cmd::Set(k, v) => {
                    map.insert(k, v);
                }
                Cmd::Get(k, reply) => reply.send(map.get(&k).copied()).unwrap(),
                Cmd::Divide(d, reply) => {
                    let total: i32 = map.values().sum();
                    let r = if d == 0 {
                        Err("divide by zero".to_string())
                    } else {
                        Ok(total / d)
                    };
                    reply.send(r).unwrap();
                }
            }
        }
        map.len() // the loop ends when every Sender is dropped
    });


    tx.send(Cmd::Set("a".into(), 10)).unwrap();
    tx.send(Cmd::Set("b".into(), 20)).unwrap();


    let (reply_tx, reply_rx) = mpsc::channel();
    tx.send(Cmd::Get("a".into(), reply_tx.clone())).unwrap();
    println!("get a: {:?}", reply_rx.recv().unwrap());
    tx.send(Cmd::Get("z".into(), reply_tx)).unwrap();
    println!("get z: {:?}", reply_rx.recv().unwrap());


    let (rtx, rrx) = mpsc::channel();
    tx.send(Cmd::Divide(5, rtx.clone())).unwrap();
    println!("sum / 5: {:?}", rrx.recv().unwrap());
    tx.send(Cmd::Divide(0, rtx)).unwrap();
    println!("sum / 0: {:?}", rrx.recv().unwrap());


    drop(tx);
    println!("entries when the owner ended: {}", owner.join().unwrap());
}
$ cargo run
get a: Some(10)
get z: None
sum / 5: Ok(6)
sum / 0: Err("divide by zero")
entries when the owner ended: 2

A pool of workers

This is fan-out, with a cost. A Receiver cannot be cloned, so workers share one behind Arc<Mutex<..>> and take the lock only to receive the next job. Idle workers wait on the lock, not on the queue; a Java ThreadPoolExecutor or N Go goroutines on one channel need nothing like it. (crossbeam-channel has cloneable receivers; not run here.) Dropping job_tx ends the workers:

Rustsrc/main.rsRun ↗
use std::sync::{Arc, Mutex, mpsc};
use std::thread;


fn main() {
    let (job_tx, job_rx) = mpsc::channel::<u64>();
    let (result_tx, result_rx) = mpsc::channel();


    // A Receiver cannot be cloned (single consumer), so the workers share one behind a lock.
    let job_rx = Arc::new(Mutex::new(job_rx));


    let workers: Vec<_> = (0..4)
        .map(|_| {
            let job_rx = Arc::clone(&job_rx);
            let result_tx = result_tx.clone();
            thread::spawn(move || {
                let mut handled = 0;
                loop {
                    // The guard is a temporary: it is released at the end of this statement,
                    // so other workers can take the lock while this one handles its job.
                    let job = job_rx.lock().unwrap().recv();
                    let Ok(n) = job else { break }; // closed and empty
                    result_tx.send((n, n * n)).unwrap();
                    handled += 1;
                }
                handled
            })
        })
        .collect();
    drop(result_tx); // workers hold the other clones


    for n in 1..=20 {
        job_tx.send(n).unwrap();
    }
    drop(job_tx); // no more jobs: each worker's recv will return Err


    let mut results: Vec<(u64, u64)> = result_rx.iter().collect();
    results.sort(); // arrival order depends on the scheduler
    let handled: usize = workers.into_iter().map(|w| w.join().unwrap()).sum();


    println!("jobs handled by all workers: {handled}");
    println!("first three results: {:?}", &results[..3]);
    println!(
        "sum of squares: {}",
        results.iter().map(|r| r.1).sum::<u64>()
    );
}
$ cargo run
jobs handled by all workers: 20
first three results: [(1, 1), (2, 4), (3, 9)]
sum of squares: 2870

The lock must be released before the job runs. In let job = rx.lock().unwrap().recv(); the guard is a temporary and ends at the semicolon. In while let Ok(job) = rx.lock().unwrap().recv() { .. } it lives through the loop body, so no other worker can take a job while this one is busy (try_lock returns Ok only if the lock is free):

Rustsrc/main.rsRun ↗
use std::sync::{Mutex, mpsc};


fn main() {
    let (tx, rx) = mpsc::channel();
    for job in 1..=3 {
        tx.send(job).unwrap();
    }
    drop(tx);
    let rx = Mutex::new(rx);


    // The guard is dropped at the semicolon: the lock is free while we work.
    let job = rx.lock().unwrap().recv().unwrap();
    println!(
        "let: job {job}, lock free during the job: {}",
        rx.try_lock().is_ok()
    );


    // In `while let` the guard lives as long as the loop body: nobody else can take a job.
    while let Ok(job) = rx.lock().unwrap().recv() {
        println!(
            "while let: job {job}, lock free during the job: {}",
            rx.try_lock().is_ok()
        );
    }
}
$ cargo run
let: job 1, lock free during the job: true
while let: job 2, lock free during the job: false
while let: job 3, lock free during the job: false

That compiles and gives correct answers, but the pool runs one job at a time (the Rust Book shows the same trap). The second exercise catches it.

Translating from what you know

Going deeper

What each end can do

Receiver is neither Clone nor Sync: rx.clone() is E0599 and sharing &rx between threads is E0277, which is why the pool needs a Mutex around rx. Sender is Clone, and has been Sync since Rust 1.72 (the docs say so), so a &Sender may be shared between threads. Send and Sync explains the traits.

Pipelines and closing cascades

Nobody calls close anywhere in a pipeline. Each stage loops until the stage before it drops its sender, then ends and drops its own, and the end of the input travels all the way to main:

Rustsrc/main.rsRun ↗
use std::sync::mpsc;
use std::thread;


fn main() {
    // stage 1 -> stage 2 -> main, each link a bounded channel
    let (num_tx, num_rx) = mpsc::sync_channel(2);
    let (square_tx, square_rx) = mpsc::sync_channel(2);


    let generate = thread::spawn(move || {
        for n in 1..=10u32 {
            num_tx.send(n).unwrap();
        }
        // num_tx is dropped here, so stage 2's loop will end
    });
    let square = thread::spawn(move || {
        for n in num_rx {
            square_tx.send(n * n).unwrap();
        }
        // square_tx is dropped here, so main's loop will end
    });


    let squares: Vec<u32> = square_rx.iter().collect();
    generate.join().unwrap();
    square.join().unwrap();
    println!("squares: {squares:?}");
    println!("sum: {}", squares.iter().sum::<u32>());
}
$ cargo run
squares: [1, 4, 9, 16, 25, 36, 49, 64, 81, 100]
sum: 385

Hang-up travels backwards the same way. Here main takes three results and drops its receiver: the middle stage’s send fails, it stops, its receiver drops, and the generator’s send fails too:

Rustsrc/main.rsRun ↗
use std::sync::mpsc;
use std::thread;


fn main() {
    let (num_tx, num_rx) = mpsc::sync_channel(2);
    let (square_tx, square_rx) = mpsc::sync_channel(2);


    let generate = thread::spawn(move || {
        let mut sent = 0;
        for n in 1..=1_000_000u32 {
            if num_tx.send(n).is_err() {
                break; // stage 2 has quit
            }
            sent += 1;
        }
        sent
    });
    let square = thread::spawn(move || {
        for n in num_rx {
            if square_tx.send(n * n).is_err() {
                break; // main has quit
            }
        }
    });


    // main takes three results and hangs up
    let first: Vec<u32> = square_rx.iter().take(3).collect();
    drop(square_rx);


    square.join().unwrap();
    let sent = generate.join().unwrap();
    println!("main took {first:?} and hung up");
    println!("the generator stopped long before 1000000: {}", sent < 1000);
}
$ cargo run
main took [1, 4, 9] and hung up
the generator stopped long before 1000000: true

What std does not have

  • select over several receivers. Use one receiver fed by many senders, tagged by an enum, or poll try_recv on each. crossbeam-channel provides select, and async runtimes such as Tokio bring their own channels (Async basics); neither is run here.
  • Several consumers on one queue. std::sync::mpmc exists but is unstable on 1.99.1 (as is oneshot):
Rustsrc/main.rs✗ won't compile
use std::sync::mpmc;


fn main() {
    let (_tx, _rx) = mpmc::channel::<i32>(); // many consumers: unstable on 1.99.1
}
$ cargo buildWhat is E0658?
error[E0658]: use of unstable library feature `mpmc_channel`
 --> src/main.rs:1:5
  |
1 | use std::sync::mpmc;
  |     ^^^^^^^^^^^^^^^
  |
  = note: see issue #126840 <https://github.com/rust-lang/rust/issues/126840> for more information

error[E0658]: use of unstable library feature `mpmc_channel`
 --> src/main.rs:4:22
  |
4 |     let (_tx, _rx) = mpmc::channel::<i32>(); // many consumers: unstable on 1.99.1
  |                      ^^^^^^^^^^^^^^^^^^^^
  |
  = note: see issue #126840 <https://github.com/rust-lang/rust/issues/126840> for more information
  • A close call, a sender count or a queue length: rx.len(), tx.close() and tx.sender_count() do not compile. A channel closes when you drop its ends. is_disconnected exists but is unstable on 1.99.1 (E0658).

Practice

Try it

Replace the todo!() calls until main prints all checks passed. The starters compile and stop at the first todo!(). Checks that could hang run inside a helper that gives up after 20 seconds (it uses recv_timeout).

Results in input order

Write parallel_map: run f on every item, each on its own thread, and return the results in input order. Send (index, result) on a channel, so that arrival order does not matter. Clone the sender for each thread, and drop the original Sender: while it is alive the receiving loop never ends, for every input. Item 0 is the slowest and results fall and rise, so arrival order fails and so does sorting the values. (Joining JoinHandles in order, Java’s Futures or Python’s ThreadPoolExecutor.map also pass; this exercise is for the channel and the loop that must end.)

Starterexercises/ordered-results.rs
use std::collections::HashSet;
use std::hint::black_box;
use std::sync::mpsc::{self, RecvTimeoutError};
use std::sync::{Arc, Mutex};
use std::thread::{self, ThreadId};
use std::time::Duration;


/// Runs `f` on every item, each item on its own thread, and returns the results in the
/// same order as `items`.
fn parallel_map<F>(items: Vec<u64>, f: F) -> Vec<u64>
where
    F: Fn(u64) -> u64 + Send + Sync + 'static,
{
    todo!(
        "give each thread a clone of the Sender and send (index, result); drop the original Sender; put every result at its index"
    )
}


// Gives up when `f` takes too long: a receiving loop that never ends is the likeliest bug.
fn within<T: Send + 'static>(what: &str, f: impl FnOnce() -> T + Send + 'static) -> T {
    let (tx, rx) = mpsc::channel();
    thread::spawn(move || {
        let _ = tx.send(f());
    });
    match rx.recv_timeout(Duration::from_secs(20)) {
        Ok(value) => value,
        Err(RecvTimeoutError::Timeout) => {
            panic!("{what}: still waiting after 20 s. Is a Sender still alive somewhere?")
        }
        Err(RecvTimeoutError::Disconnected) => panic!("{what}: the function panicked"),
    }
}


fn main() {
    let squares = within("squares", || parallel_map(vec![1, 2, 3, 4], |x| x * x));
    assert_eq!(squares, [1, 4, 9, 16]);


    // results that get smaller as the input grows: sorting the values is not enough
    let falling = within("falling", || parallel_map(vec![1, 2, 3, 4], |x| 10 - x));
    assert_eq!(
        falling,
        [9, 8, 7, 6],
        "results must follow the input order, not their own size"
    );


    // no items: the receiving loop must still end
    let none = within("no items", || parallel_map(vec![], |x| x + 1));
    assert!(none.is_empty());


    // Item 0 does far more work than the rest, so it finishes last. Results must still
    // come back in input order, run after run.
    for round in 0..20 {
        let items: Vec<u64> = (0..200).collect();
        let out = within("200 items", move || {
            parallel_map(items, |x| {
                if x == 0 {
                    let mut acc = 0u64;
                    // black_box stops the optimizer from deleting this busy loop
                    for i in 0..3_000_000u64 {
                        acc = black_box(acc.wrapping_add(i));
                    }
                    black_box(acc);
                }
                (x * 7919) % 1000 // neither increasing nor decreasing
            })
        });
        let expected: Vec<u64> = (0..200).map(|x| (x * 7919) % 1000).collect();
        assert_eq!(out, expected, "round {round}: results are out of order");
    }


    // the work must not run on the calling thread
    let main_id = thread::current().id();
    let seen: Arc<Mutex<HashSet<ThreadId>>> = Arc::new(Mutex::new(HashSet::new()));
    let seen2 = Arc::clone(&seen);
    within("threads", move || {
        parallel_map(vec![1, 2, 3, 4, 5, 6], move |x| {
            seen2.lock().unwrap().insert(thread::current().id());
            x
        })
    });
    let seen = seen.lock().unwrap();
    assert!(!seen.contains(&main_id), "f ran on the main thread");
    assert_eq!(seen.len(), 6, "each item should get its own thread");


    println!("all checks passed");
}

Open in the Rust Playground ↗Run it first: it stops at todo!() with not yet implemented and a hint, and the compiler warns that the parameter is unused. Both are expected. Replace the todo!() with your code and run it again until it prints all checks passed.

Show a solution
One solutionexercises/ordered-results.solution.rsRun ↗
use std::collections::HashSet;
use std::hint::black_box;
use std::sync::mpsc::{self, RecvTimeoutError};
use std::sync::{Arc, Mutex};
use std::thread::{self, ThreadId};
use std::time::Duration;


/// Runs `f` on every item, each item on its own thread, and returns the results in the
/// same order as `items`.
fn parallel_map<F>(items: Vec<u64>, f: F) -> Vec<u64>
where
    F: Fn(u64) -> u64 + Send + Sync + 'static,
{
    let count = items.len();
    let f = Arc::new(f);
    let (tx, rx) = mpsc::channel();
    for (index, item) in items.into_iter().enumerate() {
        let (tx, f) = (tx.clone(), Arc::clone(&f));
        thread::spawn(move || tx.send((index, f(item))).unwrap());
    }
    drop(tx); // every thread holds a clone; this is the last one that is not working


    let mut results = vec![0; count];
    for (index, result) in rx {
        results[index] = result;
    }
    results
}


// Gives up when `f` takes too long: a receiving loop that never ends is the likeliest bug.
fn within<T: Send + 'static>(what: &str, f: impl FnOnce() -> T + Send + 'static) -> T {
    let (tx, rx) = mpsc::channel();
    thread::spawn(move || {
        let _ = tx.send(f());
    });
    match rx.recv_timeout(Duration::from_secs(20)) {
        Ok(value) => value,
        Err(RecvTimeoutError::Timeout) => {
            panic!("{what}: still waiting after 20 s. Is a Sender still alive somewhere?")
        }
        Err(RecvTimeoutError::Disconnected) => panic!("{what}: the function panicked"),
    }
}


fn main() {
    let squares = within("squares", || parallel_map(vec![1, 2, 3, 4], |x| x * x));
    assert_eq!(squares, [1, 4, 9, 16]);


    // results that get smaller as the input grows: sorting the values is not enough
    let falling = within("falling", || parallel_map(vec![1, 2, 3, 4], |x| 10 - x));
    assert_eq!(
        falling,
        [9, 8, 7, 6],
        "results must follow the input order, not their own size"
    );


    // no items: the receiving loop must still end
    let none = within("no items", || parallel_map(vec![], |x| x + 1));
    assert!(none.is_empty());


    // Item 0 does far more work than the rest, so it finishes last. Results must still
    // come back in input order, run after run.
    for round in 0..20 {
        let items: Vec<u64> = (0..200).collect();
        let out = within("200 items", move || {
            parallel_map(items, |x| {
                if x == 0 {
                    let mut acc = 0u64;
                    // black_box stops the optimizer from deleting this busy loop
                    for i in 0..3_000_000u64 {
                        acc = black_box(acc.wrapping_add(i));
                    }
                    black_box(acc);
                }
                (x * 7919) % 1000 // neither increasing nor decreasing
            })
        });
        let expected: Vec<u64> = (0..200).map(|x| (x * 7919) % 1000).collect();
        assert_eq!(out, expected, "round {round}: results are out of order");
    }


    // the work must not run on the calling thread
    let main_id = thread::current().id();
    let seen: Arc<Mutex<HashSet<ThreadId>>> = Arc::new(Mutex::new(HashSet::new()));
    let seen2 = Arc::clone(&seen);
    within("threads", move || {
        parallel_map(vec![1, 2, 3, 4, 5, 6], move |x| {
            seen2.lock().unwrap().insert(thread::current().id());
            x
        })
    });
    let seen = seen.lock().unwrap();
    assert!(!seen.contains(&main_id), "f ran on the main thread");
    assert_eq!(seen.len(), 6, "each item should get its own thread");


    println!("all checks passed");
}
$ cargo run
all checks passed

A pool of boxed jobs

Finish Pool: new starts workers that share one Receiver, run sends a boxed closure, and finish lets the queued jobs run and joins the workers. jobs is an Option so that finish can take the Sender out and drop it while the workers still run. Two jobs wait for each other at a Barrier (Barrier::new(2): the first wait() sleeps until a second arrives), so they finish only if two workers are inside a job at once; a worker that keeps the lock while it runs a job is caught, as are a thread per job and a finish that does not join. finish drains the queue and cannot abort it: Go would cancel with a ctx; here you would add a shared AtomicBool that workers check (not part of this exercise).

Starterexercises/worker-pool.rs
use std::collections::HashSet;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::mpsc::{self, RecvTimeoutError};
use std::sync::{Arc, Barrier, Mutex};
use std::thread::{self, JoinHandle};
use std::time::Duration;


// A job is any closure that can run once on another thread.
type Job = Box<dyn FnOnce() + Send + 'static>;


struct Pool {
    // an Option so that finish() can take the Sender out and drop it while the workers still run
    jobs: Option<mpsc::Sender<Job>>,
    workers: Vec<JoinHandle<()>>,
}


impl Pool {
    /// Starts `size` worker threads. They all take jobs from one channel.
    fn new(size: usize) -> Pool {
        todo!(
            "make a channel; wrap its Receiver in Arc<Mutex<..>> so every worker can share it; each worker loops: take the job (release the lock before running it!), run it, stop when recv() fails"
        )
    }


    /// Queues a job for any free worker.
    fn run<F: FnOnce() + Send + 'static>(&self, job: F) {
        todo!("box the closure and send it")
    }


    /// Accepts no more jobs, lets the queued ones finish, and waits for the workers.
    fn finish(self) {
        todo!("drop the Sender so the workers' recv() fails, then join every worker")
    }
}


// Gives up when `f` takes too long: a hung pool is the likeliest bug.
fn within<T: Send + 'static>(what: &str, f: impl FnOnce() -> T + Send + 'static) -> T {
    let (tx, rx) = mpsc::channel();
    thread::spawn(move || {
        let _ = tx.send(f());
    });
    match rx.recv_timeout(Duration::from_secs(20)) {
        Ok(value) => value,
        Err(RecvTimeoutError::Timeout) => {
            panic!(
                "{what}: still waiting after 20 s. Is a lock held while waiting, or a Sender alive?"
            )
        }
        Err(RecvTimeoutError::Disconnected) => panic!("{what}: the function panicked"),
    }
}


fn main() {
    // every job runs exactly once, and finish() returns only after they have all run
    let count = Arc::new(AtomicUsize::new(0));
    let counted = Arc::clone(&count);
    within("100 jobs", move || {
        let pool = Pool::new(4);
        for _ in 0..100 {
            let count = Arc::clone(&counted);
            pool.run(move || {
                count.fetch_add(1, Ordering::SeqCst);
            });
        }
        pool.finish();
    });
    assert_eq!(count.load(Ordering::SeqCst), 100);


    // no jobs at all
    within("no jobs", || Pool::new(3).finish());


    // Two jobs that each wait for the other: they only finish if two workers are inside
    // a job AT THE SAME TIME. A worker that keeps the lock while it runs a job fails here.
    let done = Arc::new(AtomicUsize::new(0));
    let done2 = Arc::clone(&done);
    within("two jobs at once", move || {
        let pool = Pool::new(2);
        let meet = Arc::new(Barrier::new(2));
        for _ in 0..2 {
            let (meet, done) = (Arc::clone(&meet), Arc::clone(&done2));
            pool.run(move || {
                meet.wait();
                done.fetch_add(1, Ordering::SeqCst);
            });
        }
        pool.finish();
    });
    assert_eq!(done.load(Ordering::SeqCst), 2);


    // a pool of 3 uses 3 threads, whatever the number of jobs, and never the caller's
    let main_id = thread::current().id();
    let ids = Arc::new(Mutex::new(HashSet::new()));
    let ids2 = Arc::clone(&ids);
    within("thread count", move || {
        let pool = Pool::new(3);
        for _ in 0..60 {
            let ids = Arc::clone(&ids2);
            pool.run(move || {
                ids.lock().unwrap().insert(thread::current().id());
            });
        }
        pool.finish();
    });
    let ids = ids.lock().unwrap();
    assert!(!ids.contains(&main_id), "a job ran on the calling thread");
    assert!(
        ids.len() <= 3,
        "{} threads ran jobs; the pool has 3",
        ids.len()
    );


    println!("all checks passed");
}

Open in the Rust Playground ↗Run it first: it stops at todo!() with not yet implemented and a hint, and the compiler warns that the parameter is unused. Both are expected. Replace the todo!() with your code and run it again until it prints all checks passed.

Show a solution
One solutionexercises/worker-pool.solution.rsRun ↗
use std::collections::HashSet;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::mpsc::{self, RecvTimeoutError};
use std::sync::{Arc, Barrier, Mutex};
use std::thread::{self, JoinHandle};
use std::time::Duration;


// A job is any closure that can run once on another thread.
type Job = Box<dyn FnOnce() + Send + 'static>;


struct Pool {
    // an Option so that finish() can take the Sender out and drop it while the workers still run
    jobs: Option<mpsc::Sender<Job>>,
    workers: Vec<JoinHandle<()>>,
}


impl Pool {
    /// Starts `size` worker threads. They all take jobs from one channel.
    fn new(size: usize) -> Pool {
        let (tx, rx) = mpsc::channel::<Job>();
        let rx = Arc::new(Mutex::new(rx));
        let workers = (0..size)
            .map(|_| {
                let rx = Arc::clone(&rx);
                thread::spawn(move || {
                    loop {
                        // the guard is a temporary: it is released before the job runs
                        let job = rx.lock().unwrap().recv();
                        match job {
                            Ok(job) => job(),
                            Err(_) => break, // every Sender is gone and the queue is empty
                        }
                    }
                })
            })
            .collect();
        Pool {
            jobs: Some(tx),
            workers,
        }
    }


    /// Queues a job for any free worker.
    fn run<F: FnOnce() + Send + 'static>(&self, job: F) {
        self.jobs.as_ref().unwrap().send(Box::new(job)).unwrap();
    }


    /// Accepts no more jobs, lets the queued ones finish, and waits for the workers.
    fn finish(self) {
        drop(self.jobs); // the last Sender: recv() now fails once the queue is empty
        for worker in self.workers {
            worker.join().unwrap();
        }
    }
}


// Gives up when `f` takes too long: a hung pool is the likeliest bug.
fn within<T: Send + 'static>(what: &str, f: impl FnOnce() -> T + Send + 'static) -> T {
    let (tx, rx) = mpsc::channel();
    thread::spawn(move || {
        let _ = tx.send(f());
    });
    match rx.recv_timeout(Duration::from_secs(20)) {
        Ok(value) => value,
        Err(RecvTimeoutError::Timeout) => {
            panic!(
                "{what}: still waiting after 20 s. Is a lock held while waiting, or a Sender alive?"
            )
        }
        Err(RecvTimeoutError::Disconnected) => panic!("{what}: the function panicked"),
    }
}


fn main() {
    // every job runs exactly once, and finish() returns only after they have all run
    let count = Arc::new(AtomicUsize::new(0));
    let counted = Arc::clone(&count);
    within("100 jobs", move || {
        let pool = Pool::new(4);
        for _ in 0..100 {
            let count = Arc::clone(&counted);
            pool.run(move || {
                count.fetch_add(1, Ordering::SeqCst);
            });
        }
        pool.finish();
    });
    assert_eq!(count.load(Ordering::SeqCst), 100);


    // no jobs at all
    within("no jobs", || Pool::new(3).finish());


    // Two jobs that each wait for the other: they only finish if two workers are inside
    // a job AT THE SAME TIME. A worker that keeps the lock while it runs a job fails here.
    let done = Arc::new(AtomicUsize::new(0));
    let done2 = Arc::clone(&done);
    within("two jobs at once", move || {
        let pool = Pool::new(2);
        let meet = Arc::new(Barrier::new(2));
        for _ in 0..2 {
            let (meet, done) = (Arc::clone(&meet), Arc::clone(&done2));
            pool.run(move || {
                meet.wait();
                done.fetch_add(1, Ordering::SeqCst);
            });
        }
        pool.finish();
    });
    assert_eq!(done.load(Ordering::SeqCst), 2);


    // a pool of 3 uses 3 threads, whatever the number of jobs, and never the caller's
    let main_id = thread::current().id();
    let ids = Arc::new(Mutex::new(HashSet::new()));
    let ids2 = Arc::clone(&ids);
    within("thread count", move || {
        let pool = Pool::new(3);
        for _ in 0..60 {
            let ids = Arc::clone(&ids2);
            pool.run(move || {
                ids.lock().unwrap().insert(thread::current().id());
            });
        }
        pool.finish();
    });
    let ids = ids.lock().unwrap();
    assert!(!ids.contains(&main_id), "a job ran on the calling thread");
    assert!(
        ids.len() <= 3,
        "{} threads ran jobs; the pool has 3",
        ids.len()
    );


    println!("all checks passed");
}
$ cargo run
all checks passed

Stop when nobody is listening

Write count_up: send 0, 2, 4, … until send fails, and return how many sends succeeded and the value that bounced. A rendezvous channel makes the counts exact. A version that unwraps panics, and one that ignores the error never stops; the checks name both mistakes. This is what ctx.Done() does for a Go producer, done by the type.

Starterexercises/hang-up.rs
use std::sync::mpsc::{self, RecvTimeoutError, SyncSender};
use std::thread;
use std::time::Duration;


/// Sends 0, 2, 4, ... on `tx`, one value at a time, until the receiver hangs up.
/// Returns how many sends succeeded, and the value that could not be delivered.
fn count_up(tx: SyncSender<u64>) -> (usize, Option<u64>) {
    todo!(
        "loop over 0.. ; send(i * 2) returns Result: on Err(e) the undelivered value is e.0, so stop and return it. Do not unwrap"
    )
}


// Gives up when `f` takes too long: a producer that ignores the error never stops.
fn within<T: Send + 'static>(what: &str, f: impl FnOnce() -> T + Send + 'static) -> T {
    let (tx, rx) = mpsc::channel();
    thread::spawn(move || {
        let _ = tx.send(f());
    });
    match rx.recv_timeout(Duration::from_secs(20)) {
        Ok(value) => value,
        Err(RecvTimeoutError::Timeout) => {
            panic!("{what}: still running after 20 s. Does the loop stop when send fails?")
        }
        Err(RecvTimeoutError::Disconnected) => {
            panic!("{what}: the function panicked (did it unwrap a failed send?)")
        }
    }
}


fn main() {
    // Nobody is listening from the start: nothing is sent, and the first value bounces.
    let (tx, rx) = mpsc::sync_channel(0);
    drop(rx);
    assert_eq!(within("no receiver", move || count_up(tx)), (0, Some(0)));


    // A rendezvous channel (bound 0) hands each value straight to the receiver. After
    // the receiver has taken five and gone, the sixth send is the one that fails.
    let (tx, rx) = mpsc::sync_channel(0);
    let producer = thread::spawn(move || count_up(tx));
    let got: Vec<u64> = rx.iter().take(5).collect();
    drop(rx);
    assert_eq!(got, [0, 2, 4, 6, 8]);
    let result = within("producer", move || producer.join().unwrap());
    assert_eq!(result, (5, Some(10)));


    // the same with a bound of 2: the producer may be up to two values ahead
    let (tx, rx) = mpsc::sync_channel(2);
    let producer = thread::spawn(move || count_up(tx));
    let got: Vec<u64> = rx.iter().take(3).collect();
    drop(rx);
    assert_eq!(got, [0, 2, 4]);
    let (sent, bounced) = within("producer", move || producer.join().unwrap());
    assert!((3..=5).contains(&sent), "sent {sent}");
    assert_eq!(bounced, Some(2 * sent as u64));


    println!("all checks passed");
}

Open in the Rust Playground ↗Run it first: it stops at todo!() with not yet implemented and a hint, and the compiler warns that the parameter is unused. Both are expected. Replace the todo!() with your code and run it again until it prints all checks passed.

Show a solution
One solutionexercises/hang-up.solution.rsRun ↗
use std::sync::mpsc::{self, RecvTimeoutError, SyncSender};
use std::thread;
use std::time::Duration;


/// Sends 0, 2, 4, ... on `tx`, one value at a time, until the receiver hangs up.
/// Returns how many sends succeeded, and the value that could not be delivered.
fn count_up(tx: SyncSender<u64>) -> (usize, Option<u64>) {
    let mut sent = 0;
    for i in 0.. {
        if let Err(e) = tx.send(i * 2) {
            return (sent, Some(e.0)); // the receiver is gone: stop, and say what bounced
        }
        sent += 1;
    }
    (sent, None)
}


// Gives up when `f` takes too long: a producer that ignores the error never stops.
fn within<T: Send + 'static>(what: &str, f: impl FnOnce() -> T + Send + 'static) -> T {
    let (tx, rx) = mpsc::channel();
    thread::spawn(move || {
        let _ = tx.send(f());
    });
    match rx.recv_timeout(Duration::from_secs(20)) {
        Ok(value) => value,
        Err(RecvTimeoutError::Timeout) => {
            panic!("{what}: still running after 20 s. Does the loop stop when send fails?")
        }
        Err(RecvTimeoutError::Disconnected) => {
            panic!("{what}: the function panicked (did it unwrap a failed send?)")
        }
    }
}


fn main() {
    // Nobody is listening from the start: nothing is sent, and the first value bounces.
    let (tx, rx) = mpsc::sync_channel(0);
    drop(rx);
    assert_eq!(within("no receiver", move || count_up(tx)), (0, Some(0)));


    // A rendezvous channel (bound 0) hands each value straight to the receiver. After
    // the receiver has taken five and gone, the sixth send is the one that fails.
    let (tx, rx) = mpsc::sync_channel(0);
    let producer = thread::spawn(move || count_up(tx));
    let got: Vec<u64> = rx.iter().take(5).collect();
    drop(rx);
    assert_eq!(got, [0, 2, 4, 6, 8]);
    let result = within("producer", move || producer.join().unwrap());
    assert_eq!(result, (5, Some(10)));


    // the same with a bound of 2: the producer may be up to two values ahead
    let (tx, rx) = mpsc::sync_channel(2);
    let producer = thread::spawn(move || count_up(tx));
    let got: Vec<u64> = rx.iter().take(3).collect();
    drop(rx);
    assert_eq!(got, [0, 2, 4]);
    let (sent, bounced) = within("producer", move || producer.join().unwrap());
    assert!((3..=5).contains(&sent), "sent {sent}");
    assert_eq!(bounced, Some(2 * sent as u64));


    println!("all checks passed");
}
$ cargo run
all checks passed

Read more

Learn Rust is an independent project, not affiliated with the Rust Project or the Rust Foundation. Code and text are under the MIT licence.Source on GitHub.