Checkpoint: Concurrency

Seven short programs on the Concurrency part: threads, shared state, channels, Send and Sync and async basics. No output here depends on timing, and the programs that do not compile stop at the compiler.

How to use it: read the program, write down your answer (the output, or the error code), then open the answer. Every answer is a real run with Rust 1.99.1, edition 2024.

1. Which step runs first?

Rustsrc/bin/q1.rsCopy

use std::future::Future;
use std::pin::pin;
use std::task::{Context, Poll, Waker};

fn block_on<F: Future>(fut: F) -> F::Output {
    let mut fut = pin!(fut);
    let mut cx = Context::from_waker(Waker::noop());
    loop {
        if let Poll::Ready(value) = fut.as_mut().poll(&mut cx) {
            return value;
        }
    }
}

async fn step(name: &str) {
    println!("step {name}");
}

fn main() {
    let a = step("a");
    let b = step("b");
    println!("built both");
    block_on(b);
    block_on(a);
}

What does it print? (block_on is the small polling loop from the async lesson.)

Answer

terminal

$ cargo run --bin q1
built both
step b
step a

Calling an async fn runs none of its body, so building a and b printed nothing. A future runs when something polls it, and block_on(b) came first. Taught in A future is a value that has not run.

2. Empty, or disconnected?

Rustsrc/bin/q2.rsCopy

use std::sync::mpsc;

fn main() {
    let (tx, rx) = mpsc::channel();
    let tx2 = tx.clone();
    tx.send(1).unwrap();
    drop(tx);
    println!("{:?}", rx.try_recv());
    println!("{:?}", rx.try_recv());
    drop(tx2);
    println!("{:?}", rx.try_recv());
}

What do the three lines print?

Answer

terminal

$ cargo run --bin q2
Ok(1)
Err(Empty)
Err(Disconnected)

drop(tx) left tx2 alive, so an empty queue is Empty: nothing now, but a sender could still send. After tx2 is dropped too, nothing can arrive, and the answer is Disconnected. Taught in When does for msg in rx end?.

3. What does an atomic return?

Rustsrc/bin/q3.rsCopy

use std::sync::atomic::AtomicUsize;
use std::sync::atomic::Ordering::SeqCst;

fn main() {
    let n = AtomicUsize::new(5);
    println!("{}", n.fetch_add(1, SeqCst));
    println!("{}", n.load(SeqCst));
    println!("{:?}", n.compare_exchange(5, 10, SeqCst, SeqCst));
    println!("{:?}", n.compare_exchange(6, 10, SeqCst, SeqCst));
    println!("{}", n.load(SeqCst));
}

What are the five lines?

Answer

terminal

$ cargo run --bin q3
5
6
Err(6)
Ok(6)
10

fetch_add returns the value from before the addition. The first compare_exchange expected 5 but found 6, so it changed nothing and returned Err(6), the value it found. The second expected 6, wrote 10 and returned the old value in Ok. Taught in Atomics.

4. Which error code?

Rustq4.rsCopy

use std::cell::RefCell;
use std::sync::Arc;
use std::thread;

fn main() {
    let total = Arc::new(RefCell::new(0));
    let mut handles = Vec::new();
    for _ in 0..2 {
        let total = Arc::clone(&total);
        handles.push(thread::spawn(move || *total.borrow_mut() += 1));
    }
    for handle in handles {
        handle.join().unwrap();
    }
    println!("{}", total.borrow());
}

Does it compile? If not, which error code, and which trait does RefCell<i32> lack?

Answer

$ cargo run --bin q4

error[E0277]: `RefCell<i32>` cannot be shared between threads safely
  --> src/bin/q4.rs:10:36
   |
10 |         handles.push(thread::spawn(move || *total.borrow_mut() += 1));
   |                      ------------- ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ `RefCell<i32>` cannot be shared between threads safely
   |                      |
   |                      required by a bound introduced by this call
   |
   = help: the trait `Sync` is not implemented for `RefCell<i32>`
   = note: if you want to do aliasing and mutation between multiple threads, use `std::sync::RwLock` instead
   = note: required for `Arc<RefCell<i32>>` to implement `Send`
note: required because it's used within this closure
  --> src/bin/q4.rs:10:36
   |
10 |         handles.push(thread::spawn(move || *total.borrow_mut() += 1));
   |                                    ^^^^^^^
  ... (notes pointing into the standard library trimmed)

E0277, and RefCell<i32> lacks Sync. The compiler also says that Arc<RefCell<i32>> is not Send, and the two notes are one fact: an Arc<T> can move to another thread only when T is Send and Sync. Every thread holding the Arc gets a &T, and a RefCell changes its borrow count through &self with nothing to coordinate the threads. A Mutex is the thread-safe version. Taught in Arc<RefCell<T>> fails, Arc<Mutex<T>> works.

5. Which error code?

Rustq5.rsCopy

use std::thread;

fn main() {
    let name = String::from("worker");
    let handle = thread::spawn(|| println!("hello from {name}"));
    handle.join().unwrap();
}

The thread is joined on the next line. Does it compile? If not, which error code?

Answer

$ cargo run --bin q5

error[E0373]: closure may outlive the current function, but it borrows `name`, which is owned by the current function
 --> src/bin/q5.rs:5:32
  |
5 |     let handle = thread::spawn(|| println!("hello from {name}"));
  |                                ^^                       ---- `name` is borrowed here
  |                                |
  |                                may outlive borrowed value `name`
  |
note: function requires argument type to outlive `'static`
 --> src/bin/q5.rs:5:18
  |
5 |     let handle = thread::spawn(|| println!("hello from {name}"));
  |                  ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
help: to force the closure to take ownership of `name` (and any other referenced variables), use the `move` keyword
  |
5 |     let handle = thread::spawn(move || println!("hello from {name}"));
  |                                ++++

E0373. A thread may outlive the function that started it, and the compiler does not use the join on the next line to decide otherwise, so the closure may not borrow a local. The suggestion is move. thread::scope is the tool that does let a thread borrow. Taught in move: the thread owns what it uses.

6. Fix it: three threads, one list

Rustq6.rsCopy

use std::sync::Arc;
use std::thread;

fn main() {
    let list = Arc::new(Vec::new());
    let mut handles = Vec::new();
    for i in 0..3 {
        let list = Arc::clone(&list);
        handles.push(thread::spawn(move || list.push(i)));
    }
    for handle in handles {
        handle.join().unwrap();
    }
    println!("{}", list.len());
}

This does not compile. Which error code? Then make it print the three numbers in order.

Answer

$ cargo run --bin q6

error[E0596]: cannot borrow data in an `Arc` as mutable
 --> src/bin/q6.rs:9:44
  |
9 |         handles.push(thread::spawn(move || list.push(i)));
  |                                            ^^^^ cannot borrow as mutable
  |
  = help: trait `DerefMut` is required to modify through a dereference, but it is not implemented for `Arc<Vec<i32>>`

E0596. An Arc hands out only shared references, so push has no &mut Vec to work with. Put a Mutex inside the Arc. The threads finish in any order, so one fix is to sort before printing (joining each thread before spawning the next also gives the order, but gives up running them at the same time):

Rustq6.rsCopy

use std::sync::{Arc, Mutex};
use std::thread;

fn main() {
    let list = Arc::new(Mutex::new(Vec::new()));
    let mut handles = Vec::new();
    for i in 0..3 {
        let list = Arc::clone(&list);
        handles.push(thread::spawn(move || list.lock().unwrap().push(i)));
    }
    for handle in handles {
        handle.join().unwrap();
    }
    let mut items = list.lock().unwrap().clone();
    items.sort();
    println!("{items:?}");
}

$ cargo run --bin q6

[0, 1, 2]

Taught in What Rust refuses and Mutex<T>.

7. Which error code?

Rustq7.rsCopy

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

fn main() {
    let (tx, rx) = mpsc::channel();
    let handle = thread::spawn(move || {
        let report = String::from("done");
        tx.send(report).unwrap();
        println!("sent {report}");
    });
    println!("{}", rx.recv().unwrap());
    handle.join().unwrap();
}

Does it compile? If not, which error code?

Answer

$ cargo run --bin q7

error[E0382]: borrow of moved value: `report`
 --> src/bin/q7.rs:9:25
  |
7 |         let report = String::from("done");
  |             ------ move occurs because `report` has type `String`, which does not implement the `Copy` trait
8 |         tx.send(report).unwrap();
  |                 ------ value moved here
9 |         println!("sent {report}");
  |                         ^^^^^^ value borrowed here after move

E0382. send takes the value, so the thread no longer owns report. One thread owns a value at a time, which is why you write no lock around a channel. Taught in Sending moves the value.

If you missed one

• Question 1: Async basics.
• Questions 2 and 7: Channels.
• Questions 3 and 6: Shared state.
• Question 4: Send and Sync.
• Question 5: Threads.
• Next: Checkpoint: Standard library.