This workshop combines the tools from the Shared State and Channels section into five programs. Each project starts with a problem and a requirement list, grows through stages that each compile and run, and ends with a full program, a table of its parts, and extension tasks.

Project	Builds	Main tools
5. Rate-limited job runner	a runner that limits concurrency and start rate	`Semaphore`, `spawn`, `add_permits`
6. Three-stage pipeline	source, transform, and sink joined by bounded channels	bounded `mpsc`, backpressure
7. Worker pool	workers that pull jobs from one queue	`mpsc`, `Arc<Mutex<Receiver>>`, results channel
8. Bank account actor	one task that owns the state, with a handle to talk to it	`mpsc`, `oneshot`
9. Live configuration reload	workers that pick up new settings without a restart	`watch`, `broadcast`, `select!`

Conventions used in every program:

Time-based programs call tokio::time::pause() on a current_thread runtime (Time: sleep, timeout, and interval). Sleeps then finish in virtual time, and printed times are exact.
Output lines that could come from several tasks at once are collected first and printed in a fixed order.
Gauge is a small helper that counts how many things are active and remembers the highest count. It wraps a std::sync::Mutex, which is never held across an .await.

Project 5: Rate-limited job runner

Run a list of jobs while respecting two limits. At most a fixed number of jobs run at the same time, and new jobs may start only at a fixed rate.

Requirements:

Accept any number of jobs and run each one as its own task.
Never run more than max_concurrent jobs at once.
Allow at most rate job starts per second after an initial burst.
A failing job must not stop the others.
Report each job's start time, end time, and result in job order.

Stage 1: limit concurrency with a semaphore

A Semaphore holds a number of permits. acquire_owned waits until a permit is free and returns an owned permit that returns to the semaphore when dropped. Each job takes a permit before it works, so no more than three jobs run together. Eight tasks start at once and wait their turn.

Rust

use std::sync::{Arc, Mutex};
use std::time::Duration;
use tokio::sync::Semaphore;
use tokio::time::{self, sleep};

#[derive(Default)]
struct Gauge {
    state: Mutex<(usize, usize)>, // (current, peak)
}

impl Gauge {
    fn enter(&self) {
        let mut state = self.state.lock().unwrap();
        state.0 += 1;
        state.1 = state.1.max(state.0);
    }

    fn leave(&self) {
        self.state.lock().unwrap().0 -= 1;
    }

    fn peak(&self) -> usize {
        self.state.lock().unwrap().1
    }
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    time::pause();

    let slots = Arc::new(Semaphore::new(3));
    let gauge = Arc::new(Gauge::default());

    let mut handles = Vec::new();
    for id in 0..8u32 {
        let slots = Arc::clone(&slots);
        let gauge = Arc::clone(&gauge);
        handles.push(tokio::spawn(async move {
            let _permit = slots.acquire_owned().await.unwrap();
            gauge.enter();
            sleep(Duration::from_millis(100)).await;
            gauge.leave();
            id * 10
        }));
    }

    let mut results = Vec::new();
    for handle in handles {
        results.push(handle.await.unwrap());
    }
    println!("results: {results:?}");
    println!("peak concurrency: {}", gauge.peak());
}

Output

results: [0, 10, 20, 30, 40, 50, 60, 70]
peak concurrency: 3

The permit lives until the end of the task body, so it is released even if the job returns early. Awaiting the handles in spawn order returns the results in job order.

Stage 2: run any list of jobs and keep failures

The runner becomes a function. A job returns Result<u32, String>, and the runner turns each result into an Outcome. Jobs with an id that leaves remainder 3 when divided by 4 fail on purpose.

Rust

use std::sync::{Arc, Mutex};
use std::time::Duration;
use tokio::sync::Semaphore;
use tokio::time::{self, sleep};

#[derive(Default)]
struct Gauge {
    state: Mutex<(usize, usize)>,
}

impl Gauge {
    fn enter(&self) {
        let mut state = self.state.lock().unwrap();
        state.0 += 1;
        state.1 = state.1.max(state.0);
    }

    fn leave(&self) {
        self.state.lock().unwrap().0 -= 1;
    }

    fn peak(&self) -> usize {
        self.state.lock().unwrap().1
    }
}

#[derive(Debug)]
enum Outcome {
    Done { id: u32, value: u32 },
    Failed { id: u32, reason: String },
}

async fn run_job(id: u32) -> Result<u32, String> {
    sleep(Duration::from_millis(100)).await;
    if id % 4 == 3 {
        Err(format!("job {id} rejected its input"))
    } else {
        Ok(id * 10)
    }
}

async fn run_all(ids: Vec<u32>, max_concurrent: usize) -> (Vec<Outcome>, usize) {
    let slots = Arc::new(Semaphore::new(max_concurrent));
    let gauge = Arc::new(Gauge::default());

    let mut handles = Vec::new();
    for id in ids {
        let slots = Arc::clone(&slots);
        let gauge = Arc::clone(&gauge);
        handles.push(tokio::spawn(async move {
            let _permit = slots.acquire_owned().await.unwrap();
            gauge.enter();
            let result = run_job(id).await;
            gauge.leave();
            match result {
                Ok(value) => Outcome::Done { id, value },
                Err(reason) => Outcome::Failed { id, reason },
            }
        }));
    }

    let mut outcomes = Vec::new();
    for handle in handles {
        outcomes.push(handle.await.expect("job task panicked"));
    }
    (outcomes, gauge.peak())
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    time::pause();

    let (outcomes, peak) = run_all((0..8).collect(), 3).await;
    for outcome in &outcomes {
        println!("{outcome:?}");
    }
    let failed = outcomes
        .iter()
        .filter(|o| matches!(o, Outcome::Failed { .. }))
        .count();
    println!("failed: {failed} of {}, peak concurrency: {peak}", outcomes.len());
}

Output

Done { id: 0, value: 0 }
Done { id: 1, value: 10 }
Done { id: 2, value: 20 }
Failed { id: 3, reason: "job 3 rejected its input" }
Done { id: 4, value: 40 }
Done { id: 5, value: 50 }
Done { id: 6, value: 60 }
Failed { id: 7, reason: "job 7 rejected its input" }
failed: 2 of 8, peak concurrency: 3

A failed job returns an Outcome::Failed value instead of panicking, so the other tasks are unaffected. A job that panics would make handle.await return an error; the runner treats that as a bug and stops with a message.

Stage 3: add a start-rate limit (full program)

A second semaphore acts as a token bucket. It starts with burst permits. Each job removes one permit permanently with forget (which keeps the permit from returning to the semaphore when it is dropped). A refill task adds per_second permits every second with add_permits, without going above burst. available_permits reports how many permits are free. A job takes a concurrency slot first, then a token.

Rust

use std::sync::{Arc, Mutex};
use std::time::Duration;
use tokio::sync::Semaphore;
use tokio::time::{self, Instant, interval, sleep};

#[derive(Default)]
struct Gauge {
    state: Mutex<(usize, usize)>,
}

impl Gauge {
    fn enter(&self) {
        let mut state = self.state.lock().unwrap();
        state.0 += 1;
        state.1 = state.1.max(state.0);
    }

    fn leave(&self) {
        self.state.lock().unwrap().0 -= 1;
    }

    fn peak(&self) -> usize {
        self.state.lock().unwrap().1
    }
}

struct Report {
    id: u32,
    start_ms: u128,
    end_ms: u128,
    result: Result<u32, String>,
}

struct Limits {
    max_concurrent: usize,
    burst: usize,
    per_second: usize,
}

/// Formats milliseconds as seconds with one decimal, rounding down.
/// Timers round up to the next millisecond, so 1500 ms may read as 1501.
fn seconds(ms: u128) -> String {
    format!("{}.{}s", ms / 1000, (ms % 1000) / 100)
}

async fn run_job(id: u32) -> Result<u32, String> {
    sleep(Duration::from_millis(1500)).await;
    if id == 3 {
        Err(String::from("bad input"))
    } else {
        Ok(id * 10)
    }
}

async fn refill(tokens: Arc<Semaphore>, per_second: usize, burst: usize) {
    let mut ticker = interval(Duration::from_secs(1));
    ticker.tick().await; // the first tick completes immediately
    loop {
        ticker.tick().await;
        let missing = burst.saturating_sub(tokens.available_permits());
        tokens.add_permits(per_second.min(missing));
    }
}

async fn run_all(ids: Vec<u32>, limits: Limits) -> (Vec<Report>, usize) {
    let slots = Arc::new(Semaphore::new(limits.max_concurrent));
    let tokens = Arc::new(Semaphore::new(limits.burst));
    let gauge = Arc::new(Gauge::default());
    let clock = Instant::now();

    let refiller = tokio::spawn(refill(Arc::clone(&tokens), limits.per_second, limits.burst));

    let mut handles = Vec::new();
    for id in ids {
        let slots = Arc::clone(&slots);
        let tokens = Arc::clone(&tokens);
        let gauge = Arc::clone(&gauge);
        handles.push(tokio::spawn(async move {
            let _slot = slots.acquire_owned().await.unwrap();
            tokens.acquire().await.unwrap().forget();
            gauge.enter();
            let start_ms = clock.elapsed().as_millis();
            let result = run_job(id).await;
            let end_ms = clock.elapsed().as_millis();
            gauge.leave();
            Report { id, start_ms, end_ms, result }
        }));
    }

    let mut reports = Vec::new();
    for handle in handles {
        reports.push(handle.await.expect("job task panicked"));
    }
    refiller.abort();
    (reports, gauge.peak())
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    time::pause();

    let limits = Limits { max_concurrent: 3, burst: 2, per_second: 2 };
    let (reports, peak) = run_all((0..6).collect(), limits).await;

    for r in &reports {
        let status = match &r.result {
            Ok(value) => format!("ok({value})"),
            Err(reason) => format!("failed({reason})"),
        };
        println!("job {}: start {}, end {}, {status}", r.id, seconds(r.start_ms), seconds(r.end_ms));
    }
    println!("peak concurrency: {peak}");
}

Output

job 0: start 0.0s, end 1.5s, ok(0)
job 1: start 0.0s, end 1.5s, ok(10)
job 2: start 1.0s, end 2.5s, ok(20)
job 3: start 1.5s, end 3.0s, failed(bad input)
job 4: start 2.0s, end 3.5s, ok(40)
job 5: start 2.5s, end 4.0s, ok(50)
peak concurrency: 3

Reading the output (jobs take 1.5 s): the bucket starts with two tokens, so jobs 0 and 1 start at 0.0 s. Job 2 takes the third slot and waits for a token. The refill at 1.0 s adds two tokens, and job 2 starts. Job 3 has no slot until jobs 0 and 1 end at 1.5 s, and then it takes the token left over from the 1.0 s refill. Job 4 gets a slot at 1.5 s and starts at 2.0 s, when the next refill arrives. Job 5 gets the slot freed by job 2 at 2.5 s. Both limits act: no start is faster than the token supply, and the peak of 3 running jobs equals the slot limit.

What each part does:

Part	Purpose
`slots` semaphore	Caps the number of running jobs. The owned permit is released when the task ends.
`tokens` semaphore	Token bucket. `forget` consumes a token for good.
`refill` task	Adds `per_second` tokens each second, capped at `burst`. Started with `spawn` and stopped with `abort`.
`Gauge`	Records the highest number of simultaneous jobs.
`Report`	Carries start time, end time, and result back through the `JoinHandle`.

Extension tasks:

Take the token first and the slot second. Compare the reported start times and explain why holding a slot while waiting for a token can leave slots idle.
Add a try_acquire on the token semaphore and count how many jobs would be dropped instead of delayed if the runner shed load.
Give each job a priority and run high-priority jobs first when several wait for a slot.

Project 6: Three-stage processing pipeline

Process a stream of numbers through three connected tasks: a source that generates them, a transform that validates and changes them, and a sink that collects the result. Bounded channels connect the stages so a slow stage slows the stages before it.

Requirements:

Three tasks joined by two bounded mpsc channels.
The transform drops numbers divisible by 5 and squares the rest.
When the source finishes, the shutdown passes down the pipeline without extra signals.
The number of items waiting between stages is bounded by the channel capacity (plus the one item a blocked sender holds).
Each stage reports how many items it handled.

Stage 1: backpressure between two tasks

A bounded channel refuses new messages when it is full. try_send reports this as an error at once. send().await waits for space, which is what slows the sender.

Rust

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel::<u32>(2);

    for n in 1..=4 {
        match tx.try_send(n) {
            Ok(()) => println!("queued {n}"),
            Err(error) => println!("rejected {n}: {error}"),
        }
    }

    println!("received {:?}", rx.recv().await);
    match tx.try_send(5) {
        Ok(()) => println!("queued 5"),
        Err(error) => println!("rejected 5: {error}"),
    }
}

Output

queued 1
queued 2
rejected 3: no available capacity
rejected 4: no available capacity
received Some(1)
queued 5

Capacity 2 holds two messages. The third and fourth are refused. After the receiver takes one message, one slot is free again.

The next program uses send().await with a slow consumer. Time is paused, so the consumer's 100 ms per item is virtual. A Gauge counts items that were sent (or are waiting to be sent) and not yet received.

Rust

use std::sync::{Arc, Mutex};
use std::time::Duration;
use tokio::sync::mpsc;
use tokio::time::{self, sleep};

#[derive(Default)]
struct Gauge {
    state: Mutex<(usize, usize)>,
}

impl Gauge {
    fn enter(&self) {
        let mut state = self.state.lock().unwrap();
        state.0 += 1;
        state.1 = state.1.max(state.0);
    }

    fn leave(&self) {
        self.state.lock().unwrap().0 -= 1;
    }

    fn peak(&self) -> usize {
        self.state.lock().unwrap().1
    }
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    time::pause();

    let (tx, mut rx) = mpsc::channel::<u32>(2);
    let gauge = Arc::new(Gauge::default());

    let producer_gauge = Arc::clone(&gauge);
    let producer = tokio::spawn(async move {
        for n in 1..=10 {
            producer_gauge.enter();
            tx.send(n).await.unwrap();
        }
    });

    let mut received = Vec::new();
    while let Some(n) = rx.recv().await {
        gauge.leave();
        sleep(Duration::from_millis(100)).await;
        received.push(n);
    }
    producer.await.unwrap();

    println!("received: {received:?}");
    println!("peak items sent but not received: {}", gauge.peak());
}

Output

received: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
peak items sent but not received: 3

The producer wants to send ten items at once, but the peak stays near the channel capacity. The count is capacity plus the one item the producer is holding while it waits for space.

Stage 2: source, transform, and sink

Three tasks connected by two channels. Each stage owns the sender of the next channel. When the source ends, its sender drops, recv in the transform returns None, the transform ends and drops its sender, and the sink ends the same way. No shutdown message is needed.

Rust

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    let (raw_tx, mut raw_rx) = mpsc::channel::<u64>(4);
    let (clean_tx, mut clean_rx) = mpsc::channel::<u64>(4);

    let source = tokio::spawn(async move {
        for n in 1..=12 {
            raw_tx.send(n).await.unwrap();
        }
    });

    let transform = tokio::spawn(async move {
        while let Some(n) = raw_rx.recv().await {
            if n % 5 != 0 {
                clean_tx.send(n * n).await.unwrap();
            }
        }
    });

    let sink = tokio::spawn(async move {
        let mut collected = Vec::new();
        while let Some(n) = clean_rx.recv().await {
            collected.push(n);
        }
        collected
    });

    source.await.unwrap();
    transform.await.unwrap();
    let collected = sink.await.unwrap();
    println!("collected: {collected:?}");
    println!("sum: {}", collected.iter().sum::<u64>());
}

Output

collected: [1, 4, 9, 16, 36, 49, 64, 81, 121, 144]
sum: 525

Numbers 5 and 10 were dropped by the transform. Order is preserved because each channel is first-in, first-out and each stage is one task.

Stage 3: stage statistics and capacity comparison (full program)

Each stage returns a Stats value through its JoinHandle. The pipeline is a function, so the program can run it with three different channel capacities and compare the peak queue length. The sink is slower than the other stages (50 ms per item, virtual).

Rust

use std::sync::{Arc, Mutex};
use std::time::Duration;
use tokio::sync::mpsc;
use tokio::time::{self, sleep};

#[derive(Default)]
struct Gauge {
    state: Mutex<(usize, usize)>,
}

impl Gauge {
    fn enter(&self) {
        let mut state = self.state.lock().unwrap();
        state.0 += 1;
        state.1 = state.1.max(state.0);
    }

    fn leave(&self) {
        self.state.lock().unwrap().0 -= 1;
    }

    fn peak(&self) -> usize {
        self.state.lock().unwrap().1
    }
}

#[derive(Debug, Default)]
struct Stats {
    received: usize,
    sent: usize,
}

struct PipelineReport {
    output: Vec<u64>,
    source: Stats,
    transform: Stats,
    sink: Stats,
    peak_first_queue: usize,
    peak_second_queue: usize,
}

async fn run_pipeline(count: u64, capacity: usize) -> PipelineReport {
    let (raw_tx, mut raw_rx) = mpsc::channel::<u64>(capacity);
    let (clean_tx, mut clean_rx) = mpsc::channel::<u64>(capacity);
    let first = Arc::new(Gauge::default());
    let second = Arc::new(Gauge::default());

    let source = {
        let first = Arc::clone(&first);
        tokio::spawn(async move {
            let mut stats = Stats::default();
            for n in 1..=count {
                first.enter();
                raw_tx.send(n).await.unwrap();
                stats.sent += 1;
            }
            stats
        })
    };

    let transform = {
        let first = Arc::clone(&first);
        let second = Arc::clone(&second);
        tokio::spawn(async move {
            let mut stats = Stats::default();
            while let Some(n) = raw_rx.recv().await {
                first.leave();
                stats.received += 1;
                if n % 5 != 0 {
                    second.enter();
                    clean_tx.send(n * n).await.unwrap();
                    stats.sent += 1;
                }
            }
            stats
        })
    };

    let sink = {
        let second = Arc::clone(&second);
        tokio::spawn(async move {
            let mut stats = Stats::default();
            let mut output = Vec::new();
            while let Some(n) = clean_rx.recv().await {
                second.leave();
                stats.received += 1;
                sleep(Duration::from_millis(50)).await;
                output.push(n);
            }
            (stats, output)
        })
    };

    let source = source.await.unwrap();
    let transform = transform.await.unwrap();
    let (sink, output) = sink.await.unwrap();

    PipelineReport {
        output,
        source,
        transform,
        sink,
        peak_first_queue: first.peak(),
        peak_second_queue: second.peak(),
    }
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    time::pause();

    for capacity in [1, 4, 64] {
        let report = run_pipeline(20, capacity).await;
        println!("capacity {capacity}");
        println!("  output: {:?}", report.output);
        println!(
            "  source sent {}, transform received {} and sent {}, sink received {}",
            report.source.sent, report.transform.received, report.transform.sent, report.sink.received
        );
        println!(
            "  peak waiting: first channel {}, second channel {}",
            report.peak_first_queue, report.peak_second_queue
        );
    }
}

Output

capacity 1
  output: [1, 4, 9, 16, 36, 49, 64, 81, 121, 144, 169, 196, 256, 289, 324, 361]
  source sent 20, transform received 20 and sent 16, sink received 16
  peak waiting: first channel 2, second channel 2
capacity 4
  output: [1, 4, 9, 16, 36, 49, 64, 81, 121, 144, 169, 196, 256, 289, 324, 361]
  source sent 20, transform received 20 and sent 16, sink received 16
  peak waiting: first channel 5, second channel 5
capacity 64
  output: [1, 4, 9, 16, 36, 49, 64, 81, 121, 144, 169, 196, 256, 289, 324, 361]
  source sent 20, transform received 20 and sent 16, sink received 16
  peak waiting: first channel 20, second channel 16

Reading the output: the items and counts are the same for every capacity, because capacity changes when work waits and not what the stages compute. The slow sink backs up the second channel first. Then the transform blocks on its send, and the first channel backs up as well. With capacity 1 and 4 the peaks are capacity plus one (2 and 5). With capacity 64 nothing ever blocks, so the peaks equal the number of items that pass through each channel: 20 on the first, 16 on the second (the transform removed 4 numbers).

What each part does:

Part	Purpose
`raw_tx` / `raw_rx`	Bounded channel from source to transform.
`clean_tx` / `clean_rx`	Bounded channel from transform to sink.
`Stats`	Counts per stage, returned by the task.
`Gauge` per channel	Measures how many items are waiting or blocked at the sender.
Dropping senders	Passes the end of input down the pipeline.

Extension tasks:

Add a fourth stage that groups squared numbers into batches of three and passes each batch as a Vec<u64>.
Replace the transform's send().await with try_send and count the numbers dropped when the second channel is full.
Run two transform tasks that share the first channel (see Project 7) and check that the output still contains every squared number once. The output order may change.

Project 7: Worker pool with a shared queue

Run a fixed number of workers that all take jobs from the same queue. A worker that finishes early takes the next job, so slow jobs do not hold back the rest.

Requirements:

A pool of n workers started once, with a job queue in front of them.
Every job is processed exactly once.
Results come back on a separate channel.
Closing the queue lets workers finish the remaining jobs and stop.
Compare the shared queue with a design that gives each worker its own queue.

An mpsc::Receiver cannot be cloned, because a channel has one receiver. To let several workers take from one queue, the receiver goes inside Arc<tokio::sync::Mutex<...>>. A worker locks the mutex, waits for a job, and releases the lock when the job arrives. The lock is held while the worker waits, so the other workers wait for the lock and take the following jobs in turn.

Rust

use std::sync::Arc;
use std::time::Duration;
use tokio::sync::{Mutex, mpsc};
use tokio::time::{self, sleep};

#[tokio::main(flavor = "current_thread")]
async fn main() {
    time::pause();

    let (job_tx, job_rx) = mpsc::channel::<u32>(16);
    let (result_tx, mut result_rx) = mpsc::channel::<(u32, u32)>(16);
    let job_rx = Arc::new(Mutex::new(job_rx));

    for _worker in 0..3 {
        let job_rx = Arc::clone(&job_rx);
        let result_tx = result_tx.clone();
        tokio::spawn(async move {
            loop {
                // The guard is a temporary: it is dropped when this statement ends.
                let job = job_rx.lock().await.recv().await;
                let Some(n) = job else { break };
                sleep(Duration::from_millis(100)).await;
                result_tx.send((n, n * n)).await.unwrap();
            }
        });
    }
    drop(result_tx);

    for n in 1..=9 {
        job_tx.send(n).await.unwrap();
    }
    drop(job_tx);

    let mut results = Vec::new();
    while let Some(result) = result_rx.recv().await {
        results.push(result);
    }
    results.sort();
    println!("{results:?}");
}

Output

[(1, 1), (2, 4), (3, 9), (4, 16), (5, 25), (6, 36), (7, 49), (8, 64), (9, 81)]

The main task drops its own result_tx. The channel closes when the last worker drops its clone, so the collecting loop ends after the last worker stops. Results arrive in completion order, so the program sorts them before printing.

Stage 2: join the workers and check exactly-once

Keeping the workers' JoinHandles lets the program wait for them and read a count from each. The total of the counts must equal the number of jobs.

Rust

use std::sync::Arc;
use std::time::Duration;
use tokio::sync::{Mutex, mpsc};
use tokio::time::{self, sleep};

#[tokio::main(flavor = "current_thread")]
async fn main() {
    time::pause();

    let (job_tx, job_rx) = mpsc::channel::<u32>(16);
    let (result_tx, mut result_rx) = mpsc::channel::<(u32, u32)>(16);
    let job_rx = Arc::new(Mutex::new(job_rx));

    let mut workers = Vec::new();
    for _worker in 0..3 {
        let job_rx = Arc::clone(&job_rx);
        let result_tx = result_tx.clone();
        workers.push(tokio::spawn(async move {
            let mut handled = 0usize;
            loop {
                let job = job_rx.lock().await.recv().await;
                let Some(n) = job else { break };
                sleep(Duration::from_millis(100)).await;
                result_tx.send((n, n * n)).await.unwrap();
                handled += 1;
            }
            handled
        }));
    }
    drop(result_tx);

    let collector = tokio::spawn(async move {
        let mut results = Vec::new();
        while let Some(result) = result_rx.recv().await {
            results.push(result);
        }
        results
    });

    for n in 1..=9 {
        job_tx.send(n).await.unwrap();
    }
    drop(job_tx);

    let mut total_handled = 0;
    for worker in workers {
        total_handled += worker.await.unwrap();
    }
    let mut results = collector.await.unwrap();
    results.sort();

    let ids: Vec<u32> = results.iter().map(|(id, _)| *id).collect();
    println!("job ids: {ids:?}");
    println!("handled by workers: {total_handled}");
    println!("results collected: {}", results.len());
}

Output

job ids: [1, 2, 3, 4, 5, 6, 7, 8, 9]
handled by workers: 9
results collected: 9

Nine jobs, nine results, and a worker total of nine: no job was lost and none ran twice. Which worker ran which job is not part of the output, because that depends on scheduling.

Stage 3: a pool type and a per-worker queue for comparison (full program)

The pool becomes a type with submit and finish. A second design, RoundRobinPool, gives every worker its own queue and sends job i to worker i % n. Both pools run the same uneven job list: one job costs 900 ms and five cost 100 ms, on two workers. The program prints the virtual time each pool needs.

Rust

use std::sync::Arc;
use std::time::Duration;
use tokio::sync::{Mutex, mpsc};
use tokio::task::JoinHandle;
use tokio::time::{self, Instant, sleep};

#[derive(Debug, Clone, Copy)]
struct Job {
    id: u32,
    cost_ms: u64,
}

/// Formats milliseconds as seconds with one decimal, rounding down.
/// Timers round up to the next millisecond, so 900 ms may read as 901.
fn seconds(ms: u128) -> String {
    format!("{}.{}s", ms / 1000, (ms % 1000) / 100)
}

async fn process(job: Job) -> (u32, u64) {
    sleep(Duration::from_millis(job.cost_ms)).await;
    (job.id, job.cost_ms * 2)
}

/// Workers share one queue through Arc<Mutex<Receiver>>.
struct SharedQueuePool {
    jobs: mpsc::Sender<Job>,
    workers: Vec<JoinHandle<usize>>,
    results: JoinHandle<Vec<(u32, u64)>>,
}

impl SharedQueuePool {
    fn new(worker_count: usize) -> Self {
        let (jobs, job_rx) = mpsc::channel::<Job>(16);
        let (result_tx, mut result_rx) = mpsc::channel::<(u32, u64)>(16);
        let job_rx = Arc::new(Mutex::new(job_rx));

        let mut workers = Vec::new();
        for _ in 0..worker_count {
            let job_rx = Arc::clone(&job_rx);
            let result_tx = result_tx.clone();
            workers.push(tokio::spawn(async move {
                let mut handled = 0;
                loop {
                    let job = job_rx.lock().await.recv().await;
                    let Some(job) = job else { break };
                    result_tx.send(process(job).await).await.unwrap();
                    handled += 1;
                }
                handled
            }));
        }

        let results = tokio::spawn(async move {
            let mut all = Vec::new();
            while let Some(result) = result_rx.recv().await {
                all.push(result);
            }
            all
        });

        SharedQueuePool { jobs, workers, results }
    }

    async fn submit(&self, job: Job) {
        self.jobs.send(job).await.expect("pool is running");
    }

    async fn finish(self) -> (Vec<(u32, u64)>, usize) {
        drop(self.jobs);
        let mut handled = 0;
        for worker in self.workers {
            handled += worker.await.unwrap();
        }
        let mut results = self.results.await.unwrap();
        results.sort();
        (results, handled)
    }
}

/// Each worker has its own queue. Job number i goes to worker i % n.
struct RoundRobinPool {
    queues: Vec<mpsc::Sender<Job>>,
    workers: Vec<JoinHandle<Vec<(u32, u64)>>>,
    next: usize,
}

impl RoundRobinPool {
    fn new(worker_count: usize) -> Self {
        let mut queues = Vec::new();
        let mut workers = Vec::new();
        for _ in 0..worker_count {
            let (tx, mut rx) = mpsc::channel::<Job>(16);
            queues.push(tx);
            workers.push(tokio::spawn(async move {
                let mut done = Vec::new();
                while let Some(job) = rx.recv().await {
                    done.push(process(job).await);
                }
                done
            }));
        }
        RoundRobinPool { queues, workers, next: 0 }
    }

    async fn submit(&mut self, job: Job) {
        let queue = &self.queues[self.next % self.queues.len()];
        self.next += 1;
        queue.send(job).await.expect("pool is running");
    }

    async fn finish(self) -> (Vec<(u32, u64)>, Vec<usize>) {
        drop(self.queues);
        let mut results = Vec::new();
        let mut per_worker = Vec::new();
        for worker in self.workers {
            let done = worker.await.unwrap();
            per_worker.push(done.len());
            results.extend(done);
        }
        results.sort();
        (results, per_worker)
    }
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    time::pause();

    let costs = [900, 100, 100, 100, 100, 100];
    let jobs: Vec<Job> = costs
        .iter()
        .enumerate()
        .map(|(i, &cost_ms)| Job { id: i as u32, cost_ms })
        .collect();

    let started = Instant::now();
    let pool = SharedQueuePool::new(2);
    for job in &jobs {
        pool.submit(*job).await;
    }
    let (results, handled) = pool.finish().await;
    println!("shared queue: {} results, {handled} handled, {}", results.len(), seconds(started.elapsed().as_millis()));
    println!("  results: {results:?}");

    let started = Instant::now();
    let mut pool = RoundRobinPool::new(2);
    for job in &jobs {
        pool.submit(*job).await;
    }
    let (results, per_worker) = pool.finish().await;
    println!("round robin: {} results, jobs per worker {per_worker:?}, {}", results.len(), seconds(started.elapsed().as_millis()));
    println!("  results: {results:?}");
}

Output

shared queue: 6 results, 6 handled, 0.9s
  results: [(0, 1800), (1, 200), (2, 200), (3, 200), (4, 200), (5, 200)]
round robin: 6 results, jobs per worker [3, 3], 1.1s
  results: [(0, 1800), (1, 200), (2, 200), (3, 200), (4, 200), (5, 200)]

Reading the output: with a shared queue, one worker takes the 900 ms job and the other worker takes all five short jobs, so the batch ends at 0.9 s. With round robin, worker 0 gets jobs 0, 2, and 4 (900 + 100 + 100 ms) and worker 1 gets the other three, so the batch ends at 1.1 s. The results are the same in both designs.

What each part does:

Part	Purpose
`Arc<Mutex<Receiver>>`	Lets many workers take from one queue. Only one worker waits on the receiver at a time.
`result_tx` clones	One per worker. The results channel closes after the last worker exits.
`finish`	Drops the queue sender, waits for every worker, then collects the sorted results.
`RoundRobinPool`	Alternative with one queue per worker and a fixed assignment rule.

Extension tasks:

Make submit use try_send and return an error when the queue is full, so callers can decide to wait or shed load.
Add a Stop job variant that makes one worker exit, and check that the other workers keep running.
Replace the shared mutex with a design where each idle worker sends its id on an "idle" channel and a dispatcher task sends the next job to that worker.

Project 8: Bank account actor

Keep an account balance inside one task. Other tasks never touch the balance directly. They send commands over a channel and wait for a reply on a oneshot channel. Because one task handles the commands one at a time, no lock is needed.

Requirements:

One task owns the balance and a transaction history.
Callers use a cloneable Account handle with deposit, withdraw, balance, and history methods.
A withdrawal larger than the balance fails and leaves the balance unchanged.
Many tasks can use the account at once without losing updates.
A transfer moves money between two accounts and restores the money if the second step fails.
When every handle is dropped, the actor ends and returns its final balance.

Stage 1: deposit and balance

A command carries its data and a oneshot::Sender for the reply. The handle hides the channel: deposit creates the reply channel, sends the command, and awaits the answer.

Rust

use tokio::sync::{mpsc, oneshot};

enum Command {
    Deposit { amount: u64, reply: oneshot::Sender<u64> },
    Balance { reply: oneshot::Sender<u64> },
}

#[derive(Clone)]
struct Account {
    commands: mpsc::Sender<Command>,
}

impl Account {
    fn open(initial: u64) -> (Account, tokio::task::JoinHandle<u64>) {
        let (commands, mut inbox) = mpsc::channel(32);
        let actor = tokio::spawn(async move {
            let mut balance = initial;
            while let Some(command) = inbox.recv().await {
                match command {
                    Command::Deposit { amount, reply } => {
                        balance += amount;
                        let _ = reply.send(balance);
                    }
                    Command::Balance { reply } => {
                        let _ = reply.send(balance);
                    }
                }
            }
            balance
        });
        (Account { commands }, actor)
    }

    async fn deposit(&self, amount: u64) -> u64 {
        let (reply, answer) = oneshot::channel();
        self.commands.send(Command::Deposit { amount, reply }).await.unwrap();
        answer.await.unwrap()
    }

    async fn balance(&self) -> u64 {
        let (reply, answer) = oneshot::channel();
        self.commands.send(Command::Balance { reply }).await.unwrap();
        answer.await.unwrap()
    }
}

#[tokio::main]
async fn main() {
    let (account, actor) = Account::open(100);

    println!("deposit 50 -> balance {}", account.deposit(50).await);
    println!("balance query -> {}", account.balance().await);

    drop(account);
    println!("actor ended with {}", actor.await.unwrap());
}

Output

deposit 50 -> balance 150
balance query -> 150
actor ended with 150

Dropping the last Account handle closes the command channel. recv returns None, the loop ends, and the task returns the final balance through its JoinHandle.

Stage 2: withdrawals, errors, and history

A withdrawal can fail, so its reply carries a Result. The actor records each successful change in a history list. Because the balance check and the change happen inside one command, no other command can run between them.

Rust

use tokio::sync::{mpsc, oneshot};

#[derive(Debug, PartialEq)]
enum BankError {
    InsufficientFunds { balance: u64, requested: u64 },
    AccountClosed,
}

#[derive(Debug, Clone, PartialEq)]
enum Entry {
    Deposit(u64),
    Withdraw(u64),
}

enum Command {
    Deposit { amount: u64, reply: oneshot::Sender<u64> },
    Withdraw { amount: u64, reply: oneshot::Sender<Result<u64, BankError>> },
    Balance { reply: oneshot::Sender<u64> },
    History { reply: oneshot::Sender<Vec<Entry>> },
}

#[derive(Clone)]
struct Account {
    commands: mpsc::Sender<Command>,
}

impl Account {
    fn open(initial: u64) -> (Account, tokio::task::JoinHandle<u64>) {
        let (commands, mut inbox) = mpsc::channel(32);
        let actor = tokio::spawn(async move {
            let mut balance = initial;
            let mut history = Vec::new();
            while let Some(command) = inbox.recv().await {
                match command {
                    Command::Deposit { amount, reply } => {
                        balance += amount;
                        history.push(Entry::Deposit(amount));
                        let _ = reply.send(balance);
                    }
                    Command::Withdraw { amount, reply } => {
                        let outcome = if amount > balance {
                            Err(BankError::InsufficientFunds { balance, requested: amount })
                        } else {
                            balance -= amount;
                            history.push(Entry::Withdraw(amount));
                            Ok(balance)
                        };
                        let _ = reply.send(outcome);
                    }
                    Command::Balance { reply } => {
                        let _ = reply.send(balance);
                    }
                    Command::History { reply } => {
                        let _ = reply.send(history.clone());
                    }
                }
            }
            balance
        });
        (Account { commands }, actor)
    }

    async fn deposit(&self, amount: u64) -> Result<u64, BankError> {
        let (reply, answer) = oneshot::channel();
        self.commands
            .send(Command::Deposit { amount, reply })
            .await
            .map_err(|_| BankError::AccountClosed)?;
        answer.await.map_err(|_| BankError::AccountClosed)
    }

    async fn withdraw(&self, amount: u64) -> Result<u64, BankError> {
        let (reply, answer) = oneshot::channel();
        self.commands
            .send(Command::Withdraw { amount, reply })
            .await
            .map_err(|_| BankError::AccountClosed)?;
        answer.await.map_err(|_| BankError::AccountClosed)?
    }

    async fn balance(&self) -> Result<u64, BankError> {
        let (reply, answer) = oneshot::channel();
        self.commands
            .send(Command::Balance { reply })
            .await
            .map_err(|_| BankError::AccountClosed)?;
        answer.await.map_err(|_| BankError::AccountClosed)
    }

    async fn history(&self) -> Result<Vec<Entry>, BankError> {
        let (reply, answer) = oneshot::channel();
        self.commands
            .send(Command::History { reply })
            .await
            .map_err(|_| BankError::AccountClosed)?;
        answer.await.map_err(|_| BankError::AccountClosed)
    }
}

#[tokio::main]
async fn main() {
    let (account, actor) = Account::open(100);

    println!("withdraw 30: {:?}", account.withdraw(30).await);
    println!("withdraw 500: {:?}", account.withdraw(500).await);
    println!("deposit 5: {:?}", account.deposit(5).await);
    println!("balance: {:?}", account.balance().await);
    println!("history: {:?}", account.history().await);

    drop(account);
    actor.await.unwrap();

    let (account, actor) = Account::open(1);
    actor.abort();
    let _ = actor.await;
    println!("after the actor stopped: {:?}", account.deposit(1).await);
}

Output

withdraw 30: Ok(70)
withdraw 500: Err(InsufficientFunds { balance: 70, requested: 500 })
deposit 5: Ok(75)
balance: Ok(75)
history: Ok([Withdraw(30), Deposit(5)])
after the actor stopped: Err(AccountClosed)

The rejected withdrawal left the balance at 75 and added nothing to the history. The last lines show what a caller sees when the actor is gone: sending fails, and the handle reports AccountClosed instead of panicking.

Stage 3: concurrent callers and transfers (full program)

Two accounts are used by many tasks at once. Ten tasks each move money from account A to account B in small steps, and ten tasks do the opposite. A transfer function withdraws from the source, deposits into the target, and refunds the source if the deposit fails. The program also shows a transfer that fails because of insufficient funds.

Rust

use tokio::sync::{mpsc, oneshot};

#[derive(Debug, PartialEq)]
enum BankError {
    InsufficientFunds { balance: u64, requested: u64 },
    AccountClosed,
}

enum Command {
    Deposit { amount: u64, reply: oneshot::Sender<u64> },
    Withdraw { amount: u64, reply: oneshot::Sender<Result<u64, BankError>> },
    Balance { reply: oneshot::Sender<u64> },
}

struct Summary {
    balance: u64,
    operations: usize,
}

#[derive(Clone)]
struct Account {
    commands: mpsc::Sender<Command>,
}

impl Account {
    fn open(initial: u64) -> (Account, tokio::task::JoinHandle<Summary>) {
        let (commands, mut inbox) = mpsc::channel(64);
        let actor = tokio::spawn(async move {
            let mut balance = initial;
            let mut operations = 0;
            while let Some(command) = inbox.recv().await {
                operations += 1;
                match command {
                    Command::Deposit { amount, reply } => {
                        balance += amount;
                        let _ = reply.send(balance);
                    }
                    Command::Withdraw { amount, reply } => {
                        let outcome = if amount > balance {
                            Err(BankError::InsufficientFunds { balance, requested: amount })
                        } else {
                            balance -= amount;
                            Ok(balance)
                        };
                        let _ = reply.send(outcome);
                    }
                    Command::Balance { reply } => {
                        let _ = reply.send(balance);
                    }
                }
            }
            Summary { balance, operations }
        });
        (Account { commands }, actor)
    }

    async fn deposit(&self, amount: u64) -> Result<u64, BankError> {
        let (reply, answer) = oneshot::channel();
        self.commands
            .send(Command::Deposit { amount, reply })
            .await
            .map_err(|_| BankError::AccountClosed)?;
        answer.await.map_err(|_| BankError::AccountClosed)
    }

    async fn withdraw(&self, amount: u64) -> Result<u64, BankError> {
        let (reply, answer) = oneshot::channel();
        self.commands
            .send(Command::Withdraw { amount, reply })
            .await
            .map_err(|_| BankError::AccountClosed)?;
        answer.await.map_err(|_| BankError::AccountClosed)?
    }

    async fn balance(&self) -> Result<u64, BankError> {
        let (reply, answer) = oneshot::channel();
        self.commands
            .send(Command::Balance { reply })
            .await
            .map_err(|_| BankError::AccountClosed)?;
        answer.await.map_err(|_| BankError::AccountClosed)
    }
}

async fn transfer(from: &Account, to: &Account, amount: u64) -> Result<(), BankError> {
    from.withdraw(amount).await?;
    if let Err(error) = to.deposit(amount).await {
        // The money left `from` but never reached `to`: put it back.
        from.deposit(amount).await?;
        return Err(error);
    }
    Ok(())
}

#[tokio::main]
async fn main() {
    let (alice, alice_actor) = Account::open(1_000);
    let (bob, bob_actor) = Account::open(1_000);

    let mut tasks = Vec::new();
    for _ in 0..10 {
        let (alice, bob) = (alice.clone(), bob.clone());
        tasks.push(tokio::spawn(async move {
            for _ in 0..20 {
                transfer(&alice, &bob, 7).await.unwrap();
            }
        }));
    }
    for _ in 0..10 {
        let (alice, bob) = (alice.clone(), bob.clone());
        tasks.push(tokio::spawn(async move {
            for _ in 0..20 {
                transfer(&bob, &alice, 5).await.unwrap();
            }
        }));
    }
    for task in tasks {
        task.await.unwrap();
    }

    let alice_balance = alice.balance().await.unwrap();
    let bob_balance = bob.balance().await.unwrap();
    println!("alice: {alice_balance}");
    println!("bob: {bob_balance}");
    println!("total: {}", alice_balance + bob_balance);

    let too_much = transfer(&alice, &bob, 1_000_000).await;
    println!("oversized transfer: {too_much:?}");
    println!("alice unchanged: {}", alice.balance().await.unwrap() == alice_balance);

    drop(alice);
    drop(bob);
    let alice_summary = alice_actor.await.unwrap();
    let bob_summary = bob_actor.await.unwrap();
    println!("alice final {} after {} commands", alice_summary.balance, alice_summary.operations);
    println!("bob final {} after {} commands", bob_summary.balance, bob_summary.operations);
}

Output

alice: 600
bob: 1400
total: 2000
oversized transfer: Err(InsufficientFunds { balance: 600, requested: 1000000 })
alice unchanged: true
alice final 600 after 403 commands
bob final 1400 after 401 commands

Reading the output: Alice sends 10 tasks x 20 x 7 = 1400 and receives 10 x 20 x 5 = 1000, so she ends 400 lower and Bob 400 higher. The total stays 2000 regardless of how the tasks interleave. Each transfer sends one command to each account. Alice handled 403 commands: 200 withdrawals and 200 deposits from the 400 transfers, one balance query before the oversized transfer, the rejected withdrawal, and one balance query after it. Bob handled 401: 400 transfer commands and one balance query.

What each part does:

Part	Purpose
Actor task	Owns the balance. Handles one command at a time.
`Account` handle	Cloneable sender. Hides the reply channel.
`oneshot` reply	Carries the answer back to the specific caller.
`BankError::AccountClosed`	Reports a stopped actor to callers.
`transfer`	Withdraw, deposit, and refund on failure.
`Summary`	Returned when the last handle drops.

The transfer is not atomic: between the withdrawal and the deposit, the total across both accounts is briefly lower. The actor design guarantees that each account is consistent, not that a pair of accounts is.

Extension tasks:

Add a Transfer command handled by a third "bank" actor that owns both balances, and check that the total is never observed below 2000.
Add an overdraft limit that is set when the account opens and checked in Withdraw.
Make history return only the last n entries and cap the stored history length.

Project 9: Live configuration reload

Workers read settings that can change while the program runs. A reloader parses new settings text, rejects invalid text, and publishes valid settings to all workers at once. Workers apply the change without restarting.

Requirements:

Settings have a prefix and a limit, written as key=value lines.
Invalid text is rejected and the previous settings stay active.
Text that produces identical settings does not wake the workers.
Workers handle requests and pick up new settings between requests.
Reload results are published as events to any number of listeners, such as an audit log.
A shutdown event stops all workers.

A watch channel holds the current settings. The worker's loop uses select! with biased, which checks its branches in written order. The settings branch comes first, so a worker always applies a pending change before it handles the next request. Each request has a oneshot reply.

Rust

use tokio::sync::{mpsc, oneshot, watch};

#[derive(Debug, Clone, PartialEq)]
struct Config {
    prefix: String,
    limit: u32,
}

type Request = (u32, oneshot::Sender<String>);

async fn worker(mut config: watch::Receiver<Config>, mut requests: mpsc::Receiver<Request>) {
    let mut current = config.borrow_and_update().clone();
    loop {
        tokio::select! {
            biased;
            changed = config.changed() => {
                if changed.is_err() {
                    break;
                }
                current = config.borrow_and_update().clone();
                println!("worker applied {current:?}");
            }
            request = requests.recv() => {
                let Some((n, reply)) = request else { break };
                let answer = if n > current.limit {
                    format!("{}: {n} is above the limit {}", current.prefix, current.limit)
                } else {
                    format!("{}: {n} accepted", current.prefix)
                };
                let _ = reply.send(answer);
            }
        }
    }
}

async fn ask(requests: &mpsc::Sender<Request>, n: u32) -> String {
    let (reply, answer) = oneshot::channel();
    requests.send((n, reply)).await.unwrap();
    answer.await.unwrap()
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let (config_tx, config_rx) = watch::channel(Config { prefix: "v1".into(), limit: 10 });
    let (request_tx, request_rx) = mpsc::channel(8);
    let handle = tokio::spawn(worker(config_rx, request_rx));

    println!("{}", ask(&request_tx, 7).await);
    println!("{}", ask(&request_tx, 20).await);

    config_tx.send(Config { prefix: "v2".into(), limit: 50 }).unwrap();
    println!("{}", ask(&request_tx, 20).await);

    drop(request_tx);
    handle.await.unwrap();
}

Output

v1: 7 accepted
v1: 20 is above the limit 10
worker applied Config { prefix: "v2", limit: 50 }
v2: 20 accepted

The worker prints "applied" before the answer to the first request after the change, because the changed() branch is checked first. changed() is cancel safe, so racing it against recv() loses no data.

Stage 2: parse, validate, and publish events

The Reloader turns text into settings. A parse error produces a Rejected event and leaves the settings alone. send_if_modified replaces the settings only when they differ from the current ones and returns whether it did. Each result is sent on a broadcast channel, and an audit task collects every event.

Rust

use tokio::sync::{broadcast, watch};

#[derive(Debug, Clone, PartialEq)]
struct Config {
    prefix: String,
    limit: u32,
}

impl Config {
    fn parse(text: &str) -> Result<Config, String> {
        let mut prefix = None;
        let mut limit = None;
        for line in text.lines().filter(|line| !line.trim().is_empty()) {
            let (key, value) = line
                .split_once('=')
                .ok_or_else(|| format!("line without '=': {line}"))?;
            match key.trim() {
                "prefix" => prefix = Some(value.trim().to_string()),
                "limit" => {
                    let parsed = value.trim().parse::<u32>().map_err(|e| format!("limit: {e}"))?;
                    limit = Some(parsed);
                }
                other => return Err(format!("unknown key: {other}")),
            }
        }
        Ok(Config {
            prefix: prefix.ok_or("missing key: prefix")?,
            limit: limit.ok_or("missing key: limit")?,
        })
    }
}

#[derive(Debug, Clone)]
enum Event {
    Applied(u64),
    Unchanged,
    Rejected(String),
}

struct Reloader {
    config: watch::Sender<Config>,
    events: broadcast::Sender<Event>,
    version: u64,
}

impl Reloader {
    fn reload(&mut self, text: &str) {
        let event = match Config::parse(text) {
            Err(reason) => Event::Rejected(reason),
            Ok(new) => {
                let changed = self.config.send_if_modified(|current| {
                    if *current == new {
                        false
                    } else {
                        *current = new;
                        true
                    }
                });
                if changed {
                    self.version += 1;
                    Event::Applied(self.version)
                } else {
                    Event::Unchanged
                }
            }
        };
        let _ = self.events.send(event);
    }
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let (config_tx, config_rx) = watch::channel(Config { prefix: "v1".into(), limit: 10 });
    let (events_tx, mut events_rx) = broadcast::channel(16);
    let mut reloader = Reloader { config: config_tx, events: events_tx, version: 1 };

    let audit = tokio::spawn(async move {
        let mut log = Vec::new();
        while let Ok(event) = events_rx.recv().await {
            log.push(event);
        }
        log
    });

    reloader.reload("prefix = v2\nlimit = 50");
    println!("current: {:?}", *config_rx.borrow());

    reloader.reload("prefix = v2\nlimit = 50");
    reloader.reload("prefix = v3\nlimit = many");
    reloader.reload("prefix = v3");
    reloader.reload("prefix = v3\nlimit = 5\nmode = fast");
    println!("current: {:?}", *config_rx.borrow());

    reloader.reload("prefix = v3\nlimit = 5");
    println!("current: {:?}", *config_rx.borrow());

    drop(reloader);
    for event in audit.await.unwrap() {
        println!("{event:?}");
    }
}

Output

current: Config { prefix: "v2", limit: 50 }
current: Config { prefix: "v2", limit: 50 }
current: Config { prefix: "v3", limit: 5 }
Applied(2)
Unchanged
Rejected("limit: invalid digit found in string")
Rejected("missing key: limit")
Rejected("unknown key: mode")
Applied(3)

Four of the six reload attempts change nothing: one identical text and three invalid texts. Only valid, different settings reach the watch channel. Dropping the reloader drops both senders, and the audit task's recv returns Closed, which ends its loop.

Stage 3: several workers, events, and shutdown (full program)

Three workers each have a request queue. Every worker runs a select! loop over three sources: settings changes, its own requests, and the event channel, where a Shutdown event ends the loop. Requests go to the workers in turn. Replies include the settings version the worker used, so the program can show that every worker applied the new settings. Each worker returns how many requests it served.

Rust

use tokio::sync::{broadcast, mpsc, oneshot, watch};
use tokio::sync::broadcast::error::RecvError;

#[derive(Debug, Clone, PartialEq)]
struct Config {
    prefix: String,
    limit: u32,
}

impl Config {
    fn parse(text: &str) -> Result<Config, String> {
        let mut prefix = None;
        let mut limit = None;
        for line in text.lines().filter(|line| !line.trim().is_empty()) {
            let (key, value) = line
                .split_once('=')
                .ok_or_else(|| format!("line without '=': {line}"))?;
            match key.trim() {
                "prefix" => prefix = Some(value.trim().to_string()),
                "limit" => {
                    let parsed = value.trim().parse::<u32>().map_err(|e| format!("limit: {e}"))?;
                    limit = Some(parsed);
                }
                other => return Err(format!("unknown key: {other}")),
            }
        }
        Ok(Config {
            prefix: prefix.ok_or("missing key: prefix")?,
            limit: limit.ok_or("missing key: limit")?,
        })
    }
}

#[derive(Debug, Clone, PartialEq)]
struct Live {
    version: u64,
    config: Config,
}

#[derive(Debug, Clone)]
enum Event {
    Applied(u64),
    Unchanged,
    Rejected(String),
    Shutdown,
}

type Request = (u32, oneshot::Sender<String>);

struct Reloader {
    live: watch::Sender<Live>,
    events: broadcast::Sender<Event>,
}

impl Reloader {
    fn reload(&self, text: &str) {
        let event = match Config::parse(text) {
            Err(reason) => Event::Rejected(reason),
            Ok(new) => {
                let mut applied = 0;
                let changed = self.live.send_if_modified(|live| {
                    if live.config == new {
                        false
                    } else {
                        live.version += 1;
                        live.config = new;
                        applied = live.version;
                        true
                    }
                });
                if changed { Event::Applied(applied) } else { Event::Unchanged }
            }
        };
        let _ = self.events.send(event);
    }
}

async fn worker(
    id: usize,
    mut live: watch::Receiver<Live>,
    mut events: broadcast::Receiver<Event>,
    mut requests: mpsc::Receiver<Request>,
) -> (usize, usize) {
    let mut current = live.borrow_and_update().clone();
    let mut served = 0;
    loop {
        tokio::select! {
            biased;
            changed = live.changed() => {
                if changed.is_err() {
                    break;
                }
                current = live.borrow_and_update().clone();
            }
            event = events.recv() => match event {
                Ok(Event::Shutdown) | Err(RecvError::Closed) => break,
                Ok(_) | Err(RecvError::Lagged(_)) => {}
            },
            request = requests.recv() => {
                let Some((n, reply)) = request else { break };
                served += 1;
                let verdict = if n > current.config.limit { "rejected" } else { "accepted" };
                let _ = reply.send(format!(
                    "worker {id} [v{} {}] {n} {verdict}",
                    current.version, current.config.prefix
                ));
            }
        }
    }
    (id, served)
}

async fn ask(queues: &[mpsc::Sender<Request>], turn: usize, n: u32) -> String {
    let (reply, answer) = oneshot::channel();
    queues[turn % queues.len()].send((n, reply)).await.unwrap();
    answer.await.unwrap()
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let initial = Live { version: 1, config: Config { prefix: "alpha".into(), limit: 10 } };
    let (live_tx, live_rx) = watch::channel(initial);
    let (events_tx, mut audit_rx) = broadcast::channel(32);
    let reloader = Reloader { live: live_tx, events: events_tx.clone() };

    let mut queues = Vec::new();
    let mut workers = Vec::new();
    for id in 0..3 {
        let (request_tx, request_rx) = mpsc::channel(8);
        queues.push(request_tx);
        workers.push(tokio::spawn(worker(id, live_rx.clone(), events_tx.subscribe(), request_rx)));
    }

    let audit = tokio::spawn(async move {
        let mut log = Vec::new();
        while let Ok(event) = audit_rx.recv().await {
            let stop = matches!(event, Event::Shutdown);
            log.push(event);
            if stop {
                break;
            }
        }
        log
    });

    let mut turn = 0;
    for n in [4, 12, 8] {
        println!("{}", ask(&queues, turn, n).await);
        turn += 1;
    }

    println!("-- reload: valid");
    reloader.reload("prefix = beta\nlimit = 20");
    for n in [12, 12, 12] {
        println!("{}", ask(&queues, turn, n).await);
        turn += 1;
    }

    println!("-- reload: invalid, then identical");
    reloader.reload("prefix = gamma\nlimit = -1");
    reloader.reload("prefix = beta\nlimit = 20");
    for n in [15, 25, 15] {
        println!("{}", ask(&queues, turn, n).await);
        turn += 1;
    }

    println!("-- shutdown");
    let _ = events_tx.send(Event::Shutdown);
    drop(queues);
    for worker in workers {
        let (id, served) = worker.await.unwrap();
        println!("worker {id} served {served}");
    }
    for event in audit.await.unwrap() {
        println!("audit: {event:?}");
    }
}

Output

worker 0 [v1 alpha] 4 accepted
worker 1 [v1 alpha] 12 rejected
worker 2 [v1 alpha] 8 accepted
-- reload: valid
worker 0 [v2 beta] 12 accepted
worker 1 [v2 beta] 12 accepted
worker 2 [v2 beta] 12 accepted
-- reload: invalid, then identical
worker 0 [v2 beta] 15 accepted
worker 1 [v2 beta] 25 rejected
worker 2 [v2 beta] 15 accepted
-- shutdown
worker 0 served 3
worker 1 served 3
worker 2 served 3
audit: Applied(2)
audit: Rejected("limit: invalid digit found in string")
audit: Unchanged
audit: Shutdown

Reading the output: the first three requests use version 1 (alpha, limit 10), so 12 is rejected. After the valid reload every worker reports version 2, and 12 is accepted under limit 20. The invalid reload and the identical reload leave the version at 2. The audit log lists the three reload results and the shutdown event in order.

What each part does:

Part	Purpose
`watch::channel(Live)`	Holds the current settings and version. Workers see only the latest value.
`send_if_modified`	Publishes only real changes, and lets the closure assign the new version.
`Config::parse`	Validates text before any worker sees it.
`broadcast` events	Every listener (workers, audit task) receives every event.
`select!` with `biased`	Applies settings first, then checks events, then handles a request.
Request queue per worker	Lets the program pick the worker, so the output order is fixed.

Extension tasks:

Add a tokio::time::interval branch to each worker that resets a per-worker counter every tick, and test it with paused time.
Add a reload_from_source task that receives new text over an mpsc channel and calls reload, so several producers can request reloads.
Make Config::parse reject limit = 0, and add a matching case to the reload sequence.

Interview points

A Semaphore limits concurrency. A second semaphore with forget and a refill task acts as a token bucket. acquire_owned gives a permit that can move into a spawned task.
Bounded channels give backpressure: send().await waits when the queue is full, and try_send fails at once. Dropping the sender is the shutdown signal for the next stage.
mpsc::Receiver is single-consumer. A shared queue needs Arc<tokio::sync::Mutex<Receiver>> or a dispatcher; a shared queue balances uneven job costs better than a fixed assignment.
The actor pattern keeps mutable state in one task and uses mpsc for commands and oneshot for replies. Each account is consistent because commands run one at a time. A transfer across two actors needs compensation logic, because it is not atomic.
watch suits settings that change over time. Combine it with broadcast when listeners need every event and not only the latest state.
select! with biased sets a fixed branch priority. changed(), recv() on mpsc and broadcast are cancel safe branches.

Summary

Project 5 combines two semaphores (slots and tokens), spawn, add_permits, forget, and abort.
Project 6 connects stages with bounded mpsc channels. Queue length is bounded by capacity and a stalled sender.
Project 7 shares one queue among workers with Arc<Mutex<Receiver>> and compares it with per-worker queues.
Project 8 uses mpsc plus oneshot per request. The handle hides the channels.
Project 9 uses watch for current settings, broadcast for events, and select! to combine them in a worker loop.
All programs print in a fixed order: time is paused, results are sorted, and worker choice is set by the program.

Project 5: Rate-limited job runner

Stage 1: limit concurrency with a semaphore

Stage 2: run any list of jobs and keep failures

Stage 3: add a start-rate limit (full program)

Project 6: Three-stage processing pipeline

Stage 1: backpressure between two tasks

Stage 2: source, transform, and sink

Stage 3: stage statistics and capacity comparison (full program)

Project 7: Worker pool with a shared queue

Stage 1: share one receiver between workers

Stage 2: join the workers and check exactly-once

Stage 3: a pool type and a per-worker queue for comparison (full program)

Project 8: Bank account actor

Stage 1: deposit and balance

Stage 2: withdrawals, errors, and history

Stage 3: concurrent callers and transfers (full program)

Project 9: Live configuration reload

Stage 1: share settings with watch

Stage 2: parse, validate, and publish events

Stage 3: several workers, events, and shutdown (full program)

Interview points

Summary