Workshop 4: Concurrency Patterns

Each project starts with a problem and requirements, then grows through stages. Every stage is a complete program that compiles and runs. The last stage is the full program, followed by a table of its parts and extension tasks.

The projects use only earlier lessons: JoinSet and Structured Concurrency, Graceful Shutdown, Backpressure and Load Shedding, Actors, Worker Pools, and Pipelines, Semaphore, Notify, Barrier, and OnceCell, oneshot, broadcast, and watch, TCP Clients and Servers, and Time: sleep, timeout, and interval.

Projects 15, 17, 18, and 20 run under the paused clock, so their times are exact. Projects 16 and 19 use real sockets and real (short) timers. Their output is fixed by waiting for events (a reply, a watch change) and never by sleeping for a guessed time.

Project 15: Bounded-concurrency crawler

A crawler fetches a list of pages. Fetching every page at once overloads the remote site, so at most three fetches may run at the same time. The fetch is simulated with a sleep, and one page fails.

Requirements:

Every page is fetched in its own task, collected in a JoinSet.
At most limit fetches run at once, enforced by a Semaphore.
Results print in the order they finish, with the time since the start.
Two error policies: collect every error, or stop at the first one and cancel the rest.
The program reports the peak number of fetches that ran at once.

Stage 1: All pages at once

Each page becomes a task. join_next returns results in completion order. The page delays are all different, so the order is fixed.

Rust

use std::time::Duration;
use tokio::task::JoinSet;
use tokio::time::{Instant, sleep};

const PAGES: [(&str, u64); 7] = [
    ("/home", 120), ("/about", 80), ("/blog", 200), ("/blog/1", 60),
    ("/blog/2", 90), ("/contact", 150), ("/faq", 70),
];

async fn fetch(path: &str, ms: u64) -> Result<usize, String> {
    sleep(Duration::from_millis(ms)).await;
    if path == "/about" { Err(format!("{path}: timeout")) } else { Ok(path.len() * 100) }
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let start = Instant::now();
    let mut set = JoinSet::new();
    for (path, ms) in PAGES {
        set.spawn(async move { (path, fetch(path, ms).await) });
    }
    while let Some(joined) = set.join_next().await {
        let (path, result) = joined.unwrap();
        println!("[{:>3} ms] {path} -> {result:?}", start.elapsed().as_millis());
    }
}

Output

[ 60 ms] /blog/1 -> Ok(700)
[ 70 ms] /faq -> Ok(400)
[ 80 ms] /about -> Err("/about: timeout")
[ 90 ms] /blog/2 -> Ok(700)
[120 ms] /home -> Ok(500)
[150 ms] /contact -> Ok(800)
[200 ms] /blog -> Ok(500)

All seven fetches start at time 0 and finish at their own delay. Nothing limits them.

Stage 2: Limit with a semaphore

Each task first acquires a permit and holds it while it fetches. With three permits, a fourth task waits until one finishes. Two counters record how many fetches are active and the highest value seen.

Rust

use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering::SeqCst};
use std::time::Duration;
use tokio::sync::Semaphore;
use tokio::task::JoinSet;
use tokio::time::{Instant, sleep};

const PAGES: [(&str, u64); 7] = [
    ("/home", 120), ("/about", 80), ("/blog", 200), ("/blog/1", 60),
    ("/blog/2", 90), ("/contact", 150), ("/faq", 70),
];

async fn fetch(path: &str, ms: u64) -> Result<usize, String> {
    sleep(Duration::from_millis(ms)).await;
    if path == "/about" { Err(format!("{path}: timeout")) } else { Ok(path.len() * 100) }
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let start = Instant::now();
    let sem = Arc::new(Semaphore::new(3));
    let (active, peak) = (Arc::new(AtomicUsize::new(0)), Arc::new(AtomicUsize::new(0)));
    let mut set = JoinSet::new();
    for (path, ms) in PAGES {
        let (sem, active, peak) = (sem.clone(), active.clone(), peak.clone());
        set.spawn(async move {
            let _permit = sem.acquire().await.unwrap();
            peak.fetch_max(active.fetch_add(1, SeqCst) + 1, SeqCst);
            let result = fetch(path, ms).await;
            active.fetch_sub(1, SeqCst);
            (path, result)
        });
    }
    while let Some(joined) = set.join_next().await {
        let (path, result) = joined.unwrap();
        println!("[{:>3} ms] {path} -> {result:?}", start.elapsed().as_millis());
    }
    println!("peak in flight: {}", peak.load(SeqCst));
}

Output

[ 80 ms] /about -> Err("/about: timeout")
[120 ms] /home -> Ok(500)
[140 ms] /blog/1 -> Ok(700)
[200 ms] /blog -> Ok(500)
[210 ms] /blog/2 -> Ok(700)
[270 ms] /faq -> Ok(400)
[290 ms] /contact -> Ok(800)
peak in flight: 3

The semaphore hands out permits in the order tasks asked for them, so /blog/1 starts when /about finishes at 80 ms. The peak is 3, never 7.

Stage 3: Error policy

crawl takes the policy as an argument. Under FailFast, the first error calls abort_all. Aborted tasks come back from join_next as cancelled JoinErrors, and the loop counts them. This is the full program.

Rust

use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering::SeqCst};
use std::time::Duration;
use tokio::sync::Semaphore;
use tokio::task::JoinSet;
use tokio::time::{Instant, sleep};

const PAGES: [(&str, u64); 7] = [
    ("/home", 120), ("/about", 80), ("/blog", 200), ("/blog/1", 60),
    ("/blog/2", 90), ("/contact", 150), ("/faq", 70),
];

#[derive(Debug, Clone, Copy)]
enum Policy { CollectAll, FailFast }

async fn fetch(path: &str, ms: u64) -> Result<usize, String> {
    sleep(Duration::from_millis(ms)).await;
    if path == "/about" { Err(format!("{path}: timeout")) } else { Ok(path.len() * 100) }
}

async fn crawl(policy: Policy, limit: usize) {
    let start = Instant::now();
    let sem = Arc::new(Semaphore::new(limit));
    let (active, peak) = (Arc::new(AtomicUsize::new(0)), Arc::new(AtomicUsize::new(0)));
    let mut set = JoinSet::new();
    for (path, ms) in PAGES {
        let (sem, active, peak) = (sem.clone(), active.clone(), peak.clone());
        set.spawn(async move {
            let _permit = sem.acquire().await.unwrap();
            peak.fetch_max(active.fetch_add(1, SeqCst) + 1, SeqCst);
            let result = fetch(path, ms).await;
            active.fetch_sub(1, SeqCst);
            (path, result)
        });
    }
    let (mut ok, mut errors, mut cancelled) = (0, 0, 0);
    while let Some(joined) = set.join_next().await {
        let at = start.elapsed().as_millis();
        match joined {
            Ok((path, Ok(size))) => { ok += 1; println!("[{at:>3} ms] {path} -> {size} bytes"); }
            Ok((_, Err(e))) => {
                errors += 1;
                println!("[{at:>3} ms] error: {e}");
                if matches!(policy, Policy::FailFast) { set.abort_all(); }
            }
            Err(e) if e.is_cancelled() => cancelled += 1,
            Err(e) => panic!("task failed: {e}"),
        }
    }
    println!("{policy:?}: ok={ok} errors={errors} cancelled={cancelled} peak={}", peak.load(SeqCst));
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    crawl(Policy::CollectAll, 3).await;
    println!("---");
    crawl(Policy::FailFast, 3).await;
}

Output

[ 80 ms] error: /about: timeout
[120 ms] /home -> 500 bytes
[140 ms] /blog/1 -> 700 bytes
[200 ms] /blog -> 500 bytes
[210 ms] /blog/2 -> 700 bytes
[270 ms] /faq -> 400 bytes
[290 ms] /contact -> 800 bytes
CollectAll: ok=6 errors=1 cancelled=0 peak=3
---
[ 80 ms] error: /about: timeout
FailFast: ok=0 errors=1 cancelled=6 peak=3

CollectAll reports six pages and one error. FailFast stops at 80 ms: the two fetches in progress and the four waiting ones are cancelled, so the run ends after one error.

Part	Role
`Semaphore`	limits concurrent fetches; permits are released when `_permit` drops
`JoinSet`	owns the tasks; `join_next` yields results as they finish
`Policy`	chooses between collecting errors and cancelling on the first
`abort_all`	cancels every task in the set, including tasks waiting for a permit
`peak` counter	proves the limit held

Extension tasks:

Retry each failed page once before counting it as an error.
Give each fetch a timeout of 100 ms and count timeouts separately.
Replace the fixed list with a queue: each fetched page adds new paths, and the crawler stops when the queue and the set are both empty.

Project 16: Graceful-shutdown server

A line-based TCP server must stop cleanly: stop accepting, let work in progress finish, and give up on work that takes too long.

Requirements:

hello style lines are answered with echo <line>. WORK <ms> replies STARTED, works for that many milliseconds, then replies DONE <ms>.
One task per connection, counted by a TaskTracker.
Shutdown stops the accept loop, closes idle connections, and lets running WORK finish.
After a drain deadline, a second token cancels the work that is still running.

Stage 1: Stop accepting and close idle connections

The accept loop and each connection use select! on the token. next_line is cancel safe, so a cancelled read loses no data. The client type wraps a connection so later stages can reuse it.

Rust

use std::net::SocketAddr;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, Lines};
use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf};
use tokio::net::{TcpListener, TcpStream};
use tokio_util::sync::CancellationToken;
use tokio_util::task::TaskTracker;

async fn handle(stream: TcpStream, stop: CancellationToken) {
    let (reader, mut writer) = stream.into_split();
    let mut lines = BufReader::new(reader).lines();
    loop {
        tokio::select! {
            _ = stop.cancelled() => break,
            line = lines.next_line() => {
                let Ok(Some(line)) = line else { break };
                writer.write_all(format!("echo {line}\n").as_bytes()).await.unwrap();
            }
        }
    }
}

async fn serve(listener: TcpListener, stop: CancellationToken, tracker: TaskTracker) {
    loop {
        tokio::select! {
            _ = stop.cancelled() => break,
            accepted = listener.accept() => {
                let (stream, _) = accepted.unwrap();
                tracker.spawn(handle(stream, stop.clone()));
            }
        }
    }
}

struct Client { lines: Lines<BufReader<OwnedReadHalf>>, writer: OwnedWriteHalf }

impl Client {
    async fn connect(addr: SocketAddr) -> Client {
        let (reader, writer) = TcpStream::connect(addr).await.unwrap().into_split();
        Client { lines: BufReader::new(reader).lines(), writer }
    }
    async fn send(&mut self, line: &str) {
        self.writer.write_all(format!("{line}\n").as_bytes()).await.unwrap();
    }
    async fn recv(&mut self) -> Option<String> {
        self.lines.next_line().await.ok().flatten()
    }
}

#[tokio::main]
async fn main() {
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let (stop, tracker) = (CancellationToken::new(), TaskTracker::new());
    let server = tokio::spawn(serve(listener, stop.clone(), tracker.clone()));

    let mut client = Client::connect(addr).await;
    client.send("hello").await;
    println!("client got: {:?}", client.recv().await);

    println!("shutdown: stop accepting");
    stop.cancel();
    server.await.unwrap();
    tracker.close();
    tracker.wait().await;
    println!("drained, open connections: {}", tracker.len());
    println!("client sees: {:?}", client.recv().await);
}

Output

client got: Some("echo hello")
shutdown: stop accepting
drained, open connections: 0
client sees: None

The server task is awaited first, so no new connection can join the tracker. Then close and wait return once every handler has exited. The idle client sees the end of its stream.

Stage 2: Let running work finish

WORK runs outside the select!, so cancelling the token does not interrupt it. The handler checks the token again on the next loop turn and exits then. The client waits for STARTED before shutdown begins, which fixes the order.

Rust

use std::net::SocketAddr;
use std::time::Duration;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, Lines};
use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf};
use tokio::net::{TcpListener, TcpStream};
use tokio::time::sleep;
use tokio_util::sync::CancellationToken;
use tokio_util::task::TaskTracker;

async fn handle(stream: TcpStream, stop: CancellationToken) {
    let (reader, mut writer) = stream.into_split();
    let mut lines = BufReader::new(reader).lines();
    loop {
        tokio::select! {
            _ = stop.cancelled() => break,
            line = lines.next_line() => {
                let Ok(Some(line)) = line else { break };
                let reply = if let Some(ms) = line.strip_prefix("WORK ") {
                    writer.write_all(b"STARTED\n").await.unwrap();
                    let ms: u64 = ms.parse().unwrap();
                    sleep(Duration::from_millis(ms)).await;
                    format!("DONE {ms}\n")
                } else {
                    format!("echo {line}\n")
                };
                writer.write_all(reply.as_bytes()).await.unwrap();
            }
        }
    }
}

async fn serve(listener: TcpListener, stop: CancellationToken, tracker: TaskTracker) {
    loop {
        tokio::select! {
            _ = stop.cancelled() => break,
            accepted = listener.accept() => {
                let (stream, _) = accepted.unwrap();
                tracker.spawn(handle(stream, stop.clone()));
            }
        }
    }
}

struct Client { lines: Lines<BufReader<OwnedReadHalf>>, writer: OwnedWriteHalf }

impl Client {
    async fn connect(addr: SocketAddr) -> Client {
        let (reader, writer) = TcpStream::connect(addr).await.unwrap().into_split();
        Client { lines: BufReader::new(reader).lines(), writer }
    }
    async fn send(&mut self, line: &str) {
        self.writer.write_all(format!("{line}\n").as_bytes()).await.unwrap();
    }
    async fn recv(&mut self) -> Option<String> {
        self.lines.next_line().await.ok().flatten()
    }
}

#[tokio::main]
async fn main() {
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let (stop, tracker) = (CancellationToken::new(), TaskTracker::new());
    let server = tokio::spawn(serve(listener, stop.clone(), tracker.clone()));

    let mut worker = Client::connect(addr).await;
    worker.send("WORK 100").await;
    println!("worker: {:?}", worker.recv().await);

    println!("shutdown: stop accepting");
    stop.cancel();
    server.await.unwrap();
    tracker.close();
    tracker.wait().await;
    println!("drained, open connections: {}", tracker.len());
    println!("worker: {:?}", worker.recv().await);
    println!("worker: {:?}", worker.recv().await);
}

Output

worker: Some("STARTED")
shutdown: stop accepting
drained, open connections: 0
worker: Some("DONE 100")
worker: None

wait returned only after DONE 100 was written. The reply is still in the socket buffer, so the client reads it after the server has finished. The next read sees the end of the stream.

Stage 3: Drain deadline and hard stop

A second token, hard, cancels work that outlives the deadline. run_until_cancelled returns None when the token fires first, and the handler then drops the connection. timeout around tracker.wait() is the deadline. This is the full program.

Rust

use std::net::SocketAddr;
use std::time::Duration;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, Lines};
use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf};
use tokio::net::{TcpListener, TcpStream};
use tokio::time::{sleep, timeout};
use tokio_util::sync::CancellationToken;
use tokio_util::task::TaskTracker;

async fn handle(stream: TcpStream, stop: CancellationToken, hard: CancellationToken) {
    let (reader, mut writer) = stream.into_split();
    let mut lines = BufReader::new(reader).lines();
    loop {
        tokio::select! {
            _ = stop.cancelled() => break,
            line = lines.next_line() => {
                let Ok(Some(line)) = line else { break };
                let reply = if let Some(ms) = line.strip_prefix("WORK ") {
                    writer.write_all(b"STARTED\n").await.unwrap();
                    let ms: u64 = ms.parse().unwrap();
                    if hard.run_until_cancelled(sleep(Duration::from_millis(ms))).await.is_none() {
                        break;
                    }
                    format!("DONE {ms}\n")
                } else {
                    format!("echo {line}\n")
                };
                writer.write_all(reply.as_bytes()).await.unwrap();
            }
        }
    }
}

async fn serve(
    listener: TcpListener,
    stop: CancellationToken,
    hard: CancellationToken,
    tracker: TaskTracker,
) {
    loop {
        tokio::select! {
            _ = stop.cancelled() => break,
            accepted = listener.accept() => {
                let (stream, _) = accepted.unwrap();
                tracker.spawn(handle(stream, stop.clone(), hard.clone()));
            }
        }
    }
}

struct Client { lines: Lines<BufReader<OwnedReadHalf>>, writer: OwnedWriteHalf }

impl Client {
    async fn connect(addr: SocketAddr) -> Client {
        let (reader, writer) = TcpStream::connect(addr).await.unwrap().into_split();
        Client { lines: BufReader::new(reader).lines(), writer }
    }
    async fn send(&mut self, line: &str) {
        self.writer.write_all(format!("{line}\n").as_bytes()).await.unwrap();
    }
    async fn recv(&mut self) -> Option<String> {
        self.lines.next_line().await.ok().flatten()
    }
}

#[tokio::main]
async fn main() {
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let (stop, hard, tracker) = (CancellationToken::new(), CancellationToken::new(), TaskTracker::new());
    let server = tokio::spawn(serve(listener, stop.clone(), hard.clone(), tracker.clone()));

    let (mut quick, mut stuck, mut idle) =
        (Client::connect(addr).await, Client::connect(addr).await, Client::connect(addr).await);
    quick.send("WORK 100").await;
    stuck.send("WORK 60000").await;
    idle.send("hi").await;
    println!("quick: {:?}", quick.recv().await);
    println!("stuck: {:?}", stuck.recv().await);
    println!("idle: {:?}", idle.recv().await);

    println!("shutdown: stop accepting");
    stop.cancel();
    server.await.unwrap();
    tracker.close();
    if timeout(Duration::from_millis(300), tracker.wait()).await.is_err() {
        println!("deadline passed, still busy: {}", tracker.len());
        hard.cancel();
        tracker.wait().await;
    }
    println!("drained, open connections: {}", tracker.len());
    println!("quick: {:?}", quick.recv().await);
    println!("quick: {:?}", quick.recv().await);
    println!("stuck: {:?}", stuck.recv().await);
    println!("idle: {:?}", idle.recv().await);
}

Output

quick: Some("STARTED")
stuck: Some("STARTED")
idle: Some("echo hi")
shutdown: stop accepting
deadline passed, still busy: 1
drained, open connections: 0
quick: Some("DONE 100")
quick: None
stuck: None
idle: None

The idle connection closes at once and quick finishes within the deadline. The 60-second job does not, so the deadline fires with one connection busy. Cancelling hard ends it and the client sees the stream close without DONE.

Part	Role
`stop` token	tells the accept loop and idle connections to exit
`hard` token	cancels work still running after the deadline
`TaskTracker`	counts connection tasks; `close` plus `wait` is the drain
`timeout` around `wait`	the drain deadline
`run_until_cancelled`	makes one await point cancellable by `hard`

Extension tasks:

Send a BYE line from the server to each idle client before it closes.
Make the deadline a command-line style argument and test a zero deadline.
Add a SHUTDOWN command that any client can send to start the sequence.

Project 17: Cron-style scheduler

A scheduler runs named jobs at fixed periods. Jobs run at the same time, a long job must not pile up runs, and the scheduler stops without cutting a run in half.

Requirements:

Each job has a name, a period, and a run time.
Every job loops on its own interval, first firing one period after the start.
A job that is still running when its next tick arrives skips that tick.
Stopping lets a run in progress finish, then reports how many runs each job made.

Stage 1: One job on an interval

interval_at sets the first tick to one period after the start. The paused clock makes each line's time exact.

Rust

use std::time::Duration;
use tokio::time::{Instant, interval_at, sleep};

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let start = Instant::now();
    let period = Duration::from_millis(300);
    let mut ticker = interval_at(start + period, period);
    for _ in 0..3 {
        ticker.tick().await;
        println!("[{:>4} ms] report starts", start.elapsed().as_millis());
        sleep(Duration::from_millis(50)).await;
    }
}

Output

[ 300 ms] report starts
[ 600 ms] report starts
[ 900 ms] report starts

Stage 2: Several jobs in a JoinSet

Each job is a function, and a JoinSet runs all of them. The tasks return (name, runs). Jobs finish at different times, so the results are sorted by name before printing.

Rust

use std::time::Duration;
use tokio::task::JoinSet;
use tokio::time::{Instant, interval_at, sleep};

async fn job(name: &'static str, period_ms: u64, work_ms: u64, rounds: u32, start: Instant) -> (&'static str, u32) {
    let period = Duration::from_millis(period_ms);
    let mut ticker = interval_at(start + period, period);
    for _ in 0..rounds {
        ticker.tick().await;
        println!("[{:>4} ms] {name} starts", start.elapsed().as_millis());
        sleep(Duration::from_millis(work_ms)).await;
    }
    (name, rounds)
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let start = Instant::now();
    let mut set = JoinSet::new();
    set.spawn(job("report", 300, 50, 3, start));
    set.spawn(job("cleanup", 470, 100, 2, start));
    let mut done = Vec::new();
    while let Some(joined) = set.join_next().await {
        done.push(joined.unwrap());
    }
    done.sort();
    println!("{done:?}");
}

Output

[ 300 ms] report starts
[ 470 ms] cleanup starts
[ 600 ms] report starts
[ 900 ms] report starts
[ 940 ms] cleanup starts
[("cleanup", 2), ("report", 3)]

The two jobs interleave by time. Neither waits for the other.

Stage 3: Skip missed ticks and stop on a token

MissedTickBehavior::Skip drops ticks that came due while the job was still running. A CancellationToken replaces the fixed round count. The job checks it only between runs, so a run in progress always finishes. This is the full program.

Rust

use std::time::Duration;
use tokio::task::JoinSet;
use tokio::time::{Instant, MissedTickBehavior, interval_at, sleep};
use tokio_util::sync::CancellationToken;

async fn job(
    name: &'static str,
    period_ms: u64,
    work_ms: u64,
    start: Instant,
    stop: CancellationToken,
) -> (&'static str, u32, u128) {
    let period = Duration::from_millis(period_ms);
    let mut ticker = interval_at(start + period, period);
    ticker.set_missed_tick_behavior(MissedTickBehavior::Skip);
    let mut runs = 0;
    loop {
        tokio::select! {
            _ = stop.cancelled() => break,
            _ = ticker.tick() => {}
        }
        println!("[{:>4} ms] {name} starts", start.elapsed().as_millis());
        sleep(Duration::from_millis(work_ms)).await;
        runs += 1;
    }
    (name, runs, start.elapsed().as_millis())
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let start = Instant::now();
    let stop = CancellationToken::new();
    let mut set = JoinSet::new();
    for (name, period, work) in [("report", 300, 50), ("cleanup", 470, 100), ("backup", 400, 650)] {
        set.spawn(job(name, period, work, start, stop.clone()));
    }

    sleep(Duration::from_millis(1000)).await;
    println!("[{:>4} ms] stop requested", start.elapsed().as_millis());
    stop.cancel();

    let mut done = Vec::new();
    while let Some(joined) = set.join_next().await {
        done.push(joined.unwrap());
    }
    done.sort();
    for (name, runs, ended) in done {
        println!("{name}: {runs} runs, exited at {ended} ms");
    }
}

Output

[ 300 ms] report starts
[ 400 ms] backup starts
[ 470 ms] cleanup starts
[ 600 ms] report starts
[ 900 ms] report starts
[ 940 ms] cleanup starts
[1000 ms] stop requested
backup: 1 runs, exited at 1050 ms
cleanup: 2 runs, exited at 1040 ms
report: 3 runs, exited at 1000 ms

backup starts at 400 ms and runs for 650 ms. Its tick at 800 ms comes due during the run and is skipped, so its next tick would be at 1200 ms. The stop request at 1000 ms arrives during the run, which ends at 1050 ms. The other two jobs are idle at 1000 ms and exit at once.

Part	Role
`interval_at`	first tick one period after the start, so no job fires at time 0
`MissedTickBehavior::Skip`	avoids catch-up runs after a slow run
`select!` on token and tick	stops between runs, never during one
`JoinSet`	runs the jobs and returns their run counts
paused clock	makes every printed time exact

Extension tasks:

Add a jitter of a few milliseconds to each period, computed from the job name so the output stays fixed.
Compare Skip with Burst and Delay for backup and print the start times.
Add a job that is removed while running by aborting its AbortHandle, and report what the abort lost.

Project 18: Async connection pool

A pool keeps a fixed number of expensive connections and lends them out. A caller waits when all are in use, and a borrowed connection returns to the pool by itself when the caller is done with it.

Requirements:

The pool has a maximum size. Connections are created on demand, up to that size.
get waits for a free connection. get_timeout gives up after a deadline.
The borrowed value is an RAII guard: dropping it returns the connection.
A guard can mark its connection broken. The pool discards it and creates a replacement later.

Stage 1: Guard that returns the connection

The idle list is a std::sync::Mutex<Vec<Conn>>. Its critical sections are short and never cross an .await. The semaphore counts free slots. Drop pushes the connection back before the _permit field drops, because fields drop after the drop method runs.

Rust

use std::ops::Deref;
use std::sync::atomic::{AtomicUsize, Ordering::SeqCst};
use std::sync::{Arc, Mutex};
use tokio::sync::{OwnedSemaphorePermit, Semaphore};

struct Conn { id: usize }

struct Pool {
    idle: Mutex<Vec<Conn>>,
    slots: Arc<Semaphore>,
    created: AtomicUsize,
}

struct PooledConn { conn: Option<Conn>, pool: Arc<Pool>, _permit: OwnedSemaphorePermit }

impl Pool {
    fn new(size: usize) -> Arc<Pool> {
        Arc::new(Pool { idle: Mutex::new(Vec::new()), slots: Arc::new(Semaphore::new(size)), created: AtomicUsize::new(0) })
    }

    async fn get(self: &Arc<Self>) -> PooledConn {
        let permit = self.slots.clone().acquire_owned().await.unwrap();
        let reused = self.idle.lock().unwrap().pop();
        let conn = reused.unwrap_or_else(|| Conn { id: self.created.fetch_add(1, SeqCst) });
        PooledConn { conn: Some(conn), pool: self.clone(), _permit: permit }
    }
}

impl Deref for PooledConn {
    type Target = Conn;
    fn deref(&self) -> &Conn { self.conn.as_ref().unwrap() }
}

impl Drop for PooledConn {
    fn drop(&mut self) {
        if let Some(conn) = self.conn.take() {
            println!("  returning conn {}", conn.id);
            self.pool.idle.lock().unwrap().push(conn);
        }
    }
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let pool = Pool::new(2);
    let a = pool.get().await;
    println!("got conn {}", a.id);
    drop(a);
    let b = pool.get().await;
    let c = pool.get().await;
    println!("got conn {} and conn {}", b.id, c.id);
    println!("created: {}", pool.created.load(SeqCst));
}

Output

got conn 0
  returning conn 0
got conn 0 and conn 1
created: 2
  returning conn 1
  returning conn 0

The first connection is reused after it is returned. Only when two are held at once does the pool create a second. At the end of main, c then b drop and return their connections.

Stage 2: Callers wait for a free connection

Five tasks share a pool of two. Each holds its connection for a different time, so the schedule is fixed under the paused clock. Tasks 2 to 4 wait until a slot frees.

Rust

use std::ops::Deref;
use std::sync::atomic::{AtomicUsize, Ordering::SeqCst};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use tokio::sync::{OwnedSemaphorePermit, Semaphore};
use tokio::time::{Instant, sleep};

struct Conn { id: usize }

struct Pool {
    idle: Mutex<Vec<Conn>>,
    slots: Arc<Semaphore>,
    created: AtomicUsize,
}

struct PooledConn { conn: Option<Conn>, pool: Arc<Pool>, _permit: OwnedSemaphorePermit }

impl Pool {
    fn new(size: usize) -> Arc<Pool> {
        Arc::new(Pool { idle: Mutex::new(Vec::new()), slots: Arc::new(Semaphore::new(size)), created: AtomicUsize::new(0) })
    }

    async fn get(self: &Arc<Self>) -> PooledConn {
        let permit = self.slots.clone().acquire_owned().await.unwrap();
        let reused = self.idle.lock().unwrap().pop();
        let conn = reused.unwrap_or_else(|| Conn { id: self.created.fetch_add(1, SeqCst) });
        PooledConn { conn: Some(conn), pool: self.clone(), _permit: permit }
    }
}

impl Deref for PooledConn {
    type Target = Conn;
    fn deref(&self) -> &Conn { self.conn.as_ref().unwrap() }
}

impl Drop for PooledConn {
    fn drop(&mut self) {
        if let Some(conn) = self.conn.take() {
            self.pool.idle.lock().unwrap().push(conn);
        }
    }
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let start = Instant::now();
    let pool = Pool::new(2);
    let mut tasks = Vec::new();
    for i in 0..5u64 {
        let pool = pool.clone();
        tasks.push(tokio::spawn(async move {
            let conn = pool.get().await;
            println!("[{:>3} ms] task {i} uses conn {}", start.elapsed().as_millis(), conn.id);
            sleep(Duration::from_millis(100 + 10 * i)).await;
        }));
    }
    for task in tasks {
        task.await.unwrap();
    }
    println!("created: {}", pool.created.load(SeqCst));
}

Output

[  0 ms] task 0 uses conn 0
[  0 ms] task 1 uses conn 1
[100 ms] task 2 uses conn 0
[110 ms] task 3 uses conn 1
[220 ms] task 4 uses conn 0
created: 2

Five tasks used two connections. created stays at 2 because a returned connection is reused before a new one is made.

Stage 3: Timeout and broken connections

get_timeout wraps get in timeout. The dropped get future releases its place in the semaphore queue, so a timed-out caller leaks nothing. discard marks a guard so that Drop does not return the connection. Its permit still frees the slot, and the next get creates a new connection. This is the full program.

Rust

use std::ops::Deref;
use std::sync::atomic::{AtomicUsize, Ordering::SeqCst};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use tokio::sync::{OwnedSemaphorePermit, Semaphore};
use tokio::time::{Instant, sleep, timeout};

struct Conn { id: usize }

struct Pool {
    idle: Mutex<Vec<Conn>>,
    slots: Arc<Semaphore>,
    created: AtomicUsize,
}

struct PooledConn { conn: Option<Conn>, pool: Arc<Pool>, _permit: OwnedSemaphorePermit }

impl Pool {
    fn new(size: usize) -> Arc<Pool> {
        Arc::new(Pool { idle: Mutex::new(Vec::new()), slots: Arc::new(Semaphore::new(size)), created: AtomicUsize::new(0) })
    }

    async fn get(self: &Arc<Self>) -> PooledConn {
        let permit = self.slots.clone().acquire_owned().await.unwrap();
        let reused = self.idle.lock().unwrap().pop();
        let conn = reused.unwrap_or_else(|| Conn { id: self.created.fetch_add(1, SeqCst) });
        PooledConn { conn: Some(conn), pool: self.clone(), _permit: permit }
    }

    async fn get_timeout(self: &Arc<Self>, wait: Duration) -> Option<PooledConn> {
        timeout(wait, self.get()).await.ok()
    }
}

impl PooledConn {
    fn discard(mut self) {
        let conn = self.conn.take().unwrap();
        println!("  discarding broken conn {}", conn.id);
    }
}

impl Deref for PooledConn {
    type Target = Conn;
    fn deref(&self) -> &Conn { self.conn.as_ref().unwrap() }
}

impl Drop for PooledConn {
    fn drop(&mut self) {
        if let Some(conn) = self.conn.take() {
            self.pool.idle.lock().unwrap().push(conn);
        }
    }
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let start = Instant::now();
    let pool = Pool::new(2);
    let mut holders = Vec::new();
    for _ in 0..2 {
        let pool = pool.clone();
        holders.push(tokio::spawn(async move {
            let _conn = pool.get().await;
            sleep(Duration::from_millis(200)).await;
        }));
    }
    sleep(Duration::from_millis(1)).await;

    let waited = pool.get_timeout(Duration::from_millis(50)).await;
    println!("[{:>3} ms] get_timeout(50 ms): {}", start.elapsed().as_millis(), if waited.is_some() { "got one" } else { "gave up" });

    let waited = pool.get_timeout(Duration::from_millis(500)).await.unwrap();
    println!("[{:>3} ms] get_timeout(500 ms): conn {}", start.elapsed().as_millis(), waited.id);
    waited.discard();

    let fresh = pool.get().await;
    println!("after discard: conn {}, created {}", fresh.id, pool.created.load(SeqCst));
    println!("free slots while held: {}", pool.slots.available_permits());
    for holder in holders {
        holder.await.unwrap();
    }
}

Output

[ 51 ms] get_timeout(50 ms): gave up
[200 ms] get_timeout(500 ms): conn 1
  discarding broken conn 1
after discard: conn 0, created 2
free slots while held: 1

Both connections are busy for 200 ms. The 50 ms wait gives up. The 500 ms wait succeeds at 201 ms and gets one of the released connections. After discard, the pool has one idle connection and one free slot. The next get takes the idle one, so no new connection is created yet.

Part	Role
`Semaphore`	one permit per allowed connection; bounds waiting callers
`Mutex<Vec<Conn>>`	idle list, locked briefly and never across an `.await`
`PooledConn`	RAII guard: `Deref` to use it, `Drop` to return it
field order	`_permit` is last so the connection is back in the list before the slot frees
`get_timeout`	bounds the wait with `timeout`
`discard`	drops a broken connection instead of returning it

Extension tasks:

Add a health check on get: if the idle connection reports itself closed, discard it and create a new one.
Record the maximum number of callers waiting at once, using an atomic counter around acquire_owned.
Add close, which calls Semaphore::close so waiting callers get an error.

Project 19: TCP proxy with round-robin and health checks

A proxy accepts client connections and forwards each to one of several backends in turn. A health checker probes the backends and the proxy skips any that do not answer.

Requirements:

Three backends answer PING with PONG and any other line with b<N>: <line>. A backend can be switched into a hung state where it accepts connections but never replies.
The proxy picks backends round-robin among the healthy ones and forwards bytes with copy_bidirectional.
A health task probes every backend on an interval. Each probe has a timeout. The result is published on a watch channel.
The proxy drops a client immediately when no backend is healthy.

Stage 1: Round-robin proxy

Each client connection is forwarded to addrs[n % 3]. The client helper sends one line and reads one reply.

Rust

use std::net::SocketAddr;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, copy_bidirectional};
use tokio::net::{TcpListener, TcpStream};

async fn backend(name: usize, listener: TcpListener) {
    loop {
        let (stream, _) = listener.accept().await.unwrap();
        tokio::spawn(async move {
            let (reader, mut writer) = stream.into_split();
            let mut lines = BufReader::new(reader).lines();
            while let Ok(Some(line)) = lines.next_line().await {
                if writer.write_all(format!("b{name}: {line}\n").as_bytes()).await.is_err() {
                    break;
                }
            }
        });
    }
}

async fn proxy(listener: TcpListener, addrs: Vec<SocketAddr>) {
    let mut n = 0;
    loop {
        let (mut client, _) = listener.accept().await.unwrap();
        let target = addrs[n % addrs.len()];
        n += 1;
        tokio::spawn(async move {
            let mut upstream = TcpStream::connect(target).await.unwrap();
            let _ = copy_bidirectional(&mut client, &mut upstream).await;
        });
    }
}

async fn ask(proxy: SocketAddr, msg: &str) -> String {
    let mut stream = TcpStream::connect(proxy).await.unwrap();
    stream.write_all(format!("{msg}\n").as_bytes()).await.unwrap();
    let mut line = String::new();
    BufReader::new(&mut stream).read_line(&mut line).await.unwrap();
    if line.is_empty() { "(closed)".into() } else { line.trim().into() }
}

#[tokio::main]
async fn main() {
    let mut addrs = Vec::new();
    for name in 0..3 {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        addrs.push(listener.local_addr().unwrap());
        tokio::spawn(backend(name, listener));
    }
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let front = listener.local_addr().unwrap();
    tokio::spawn(proxy(listener, addrs));

    for i in 0..6 {
        println!("request {i} -> {}", ask(front, &format!("hello {i}")).await);
    }
}

Output

request 0 -> b0: hello 0
request 1 -> b1: hello 1
request 2 -> b2: hello 2
request 3 -> b0: hello 3
request 4 -> b1: hello 4
request 5 -> b2: hello 5

Requests go to backends 0, 1, 2, 0, 1, 2. The proxy is a byte forwarder, so it does not parse the lines.

Stage 2: Health checks on a watch channel

The up flag simulates a hung backend: a hung backend reads lines and ignores them. probe sends PING and needs PONG within 50 ms. health runs probes every 20 ms and publishes the vector of results with send_if_modified, which notifies receivers only when the vector changed. The proxy reads the latest vector with borrow on each accept. wait_for in main waits for the health task to notice a change.

Rust

use std::net::SocketAddr;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering::SeqCst};
use std::time::Duration;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, copy_bidirectional};
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::watch;
use tokio::time::{interval, timeout};

async fn backend(name: usize, listener: TcpListener, up: Arc<AtomicBool>) {
    loop {
        let (stream, _) = listener.accept().await.unwrap();
        let up = up.clone();
        tokio::spawn(async move {
            let (reader, mut writer) = stream.into_split();
            let mut lines = BufReader::new(reader).lines();
            while let Ok(Some(line)) = lines.next_line().await {
                if !up.load(SeqCst) { continue; }
                let reply = if line == "PING" { "PONG".to_string() } else { format!("b{name}: {line}") };
                if writer.write_all(format!("{reply}\n").as_bytes()).await.is_err() { break; }
            }
        });
    }
}

async fn probe(addr: SocketAddr) -> bool {
    let attempt = async {
        let mut stream = TcpStream::connect(addr).await.ok()?;
        stream.write_all(b"PING\n").await.ok()?;
        let mut line = String::new();
        BufReader::new(&mut stream).read_line(&mut line).await.ok()?;
        Some(line.trim() == "PONG")
    };
    timeout(Duration::from_millis(50), attempt).await.ok().flatten().unwrap_or(false)
}

async fn health(addrs: Vec<SocketAddr>, tx: watch::Sender<Vec<bool>>) {
    let mut tick = interval(Duration::from_millis(20));
    loop {
        tick.tick().await;
        let mut now = Vec::new();
        for addr in &addrs {
            now.push(probe(*addr).await);
        }
        tx.send_if_modified(|current| if *current != now { *current = now; true } else { false });
    }
}

async fn proxy(listener: TcpListener, addrs: Vec<SocketAddr>, status: watch::Receiver<Vec<bool>>) {
    let mut n = 0;
    loop {
        let (mut client, _) = listener.accept().await.unwrap();
        let healthy: Vec<usize> = status.borrow().iter().enumerate().filter(|(_, up)| **up).map(|(i, _)| i).collect();
        if healthy.is_empty() { continue; }
        let target = addrs[healthy[n % healthy.len()]];
        n += 1;
        tokio::spawn(async move {
            if let Ok(Ok(mut upstream)) = timeout(Duration::from_millis(100), TcpStream::connect(target)).await {
                let _ = copy_bidirectional(&mut client, &mut upstream).await;
            }
        });
    }
}

async fn ask(proxy: SocketAddr, msg: &str) -> String {
    let mut stream = TcpStream::connect(proxy).await.unwrap();
    stream.write_all(format!("{msg}\n").as_bytes()).await.unwrap();
    let mut line = String::new();
    BufReader::new(&mut stream).read_line(&mut line).await.unwrap();
    if line.is_empty() { "(closed)".into() } else { line.trim().into() }
}

#[tokio::main]
async fn main() {
    let flags: Vec<Arc<AtomicBool>> = (0..3).map(|_| Arc::new(AtomicBool::new(true))).collect();
    let mut addrs = Vec::new();
    for (name, up) in flags.iter().enumerate() {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        addrs.push(listener.local_addr().unwrap());
        tokio::spawn(backend(name, listener, up.clone()));
    }
    let (tx, rx) = watch::channel(vec![true; 3]);
    tokio::spawn(health(addrs.clone(), tx));
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let front = listener.local_addr().unwrap();
    tokio::spawn(proxy(listener, addrs, rx.clone()));

    for i in 0..3 {
        println!("request {i} -> {}", ask(front, &format!("hello {i}")).await);
    }

    flags[1].store(false, SeqCst);
    let mut watcher = rx.clone();
    watcher.wait_for(|status| !status[1]).await.unwrap();
    println!("health: {:?}", *rx.borrow());
    for i in 3..7 {
        println!("request {i} -> {}", ask(front, &format!("hello {i}")).await);
    }
}

Output

request 0 -> b0: hello 0
request 1 -> b1: hello 1
request 2 -> b2: hello 2
health: [true, false, true]
request 3 -> b2: hello 3
request 4 -> b0: hello 4
request 5 -> b2: hello 5
request 6 -> b0: hello 6

After backend 1 hangs, the health task marks it down and the proxy alternates between backends 0 and 2. The client never waits on the hung backend, because it is not chosen.

Stage 3: Recovery and no healthy backend

The last stage adds two phases: backend 1 recovers and rejoins, then every backend hangs and the proxy drops clients at once. The code below is the full program. The ask helper prints (closed) when the proxy drops the connection.

Rust

use std::net::SocketAddr;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering::SeqCst};
use std::time::Duration;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, copy_bidirectional};
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::watch;
use tokio::time::{interval, timeout};

async fn backend(name: usize, listener: TcpListener, up: Arc<AtomicBool>) {
    loop {
        let (stream, _) = listener.accept().await.unwrap();
        let up = up.clone();
        tokio::spawn(async move {
            let (reader, mut writer) = stream.into_split();
            let mut lines = BufReader::new(reader).lines();
            while let Ok(Some(line)) = lines.next_line().await {
                if !up.load(SeqCst) { continue; }
                let reply = if line == "PING" { "PONG".to_string() } else { format!("b{name}: {line}") };
                if writer.write_all(format!("{reply}\n").as_bytes()).await.is_err() { break; }
            }
        });
    }
}

async fn probe(addr: SocketAddr) -> bool {
    let attempt = async {
        let mut stream = TcpStream::connect(addr).await.ok()?;
        stream.write_all(b"PING\n").await.ok()?;
        let mut line = String::new();
        BufReader::new(&mut stream).read_line(&mut line).await.ok()?;
        Some(line.trim() == "PONG")
    };
    timeout(Duration::from_millis(50), attempt).await.ok().flatten().unwrap_or(false)
}

async fn health(addrs: Vec<SocketAddr>, tx: watch::Sender<Vec<bool>>) {
    let mut tick = interval(Duration::from_millis(20));
    loop {
        tick.tick().await;
        let mut now = Vec::new();
        for addr in &addrs {
            now.push(probe(*addr).await);
        }
        tx.send_if_modified(|current| if *current != now { *current = now; true } else { false });
    }
}

async fn proxy(listener: TcpListener, addrs: Vec<SocketAddr>, status: watch::Receiver<Vec<bool>>) {
    let mut n = 0;
    loop {
        let (mut client, _) = listener.accept().await.unwrap();
        let healthy: Vec<usize> = status.borrow().iter().enumerate().filter(|(_, up)| **up).map(|(i, _)| i).collect();
        if healthy.is_empty() { continue; }
        let target = addrs[healthy[n % healthy.len()]];
        n += 1;
        tokio::spawn(async move {
            if let Ok(Ok(mut upstream)) = timeout(Duration::from_millis(100), TcpStream::connect(target)).await {
                let _ = copy_bidirectional(&mut client, &mut upstream).await;
            }
        });
    }
}

async fn ask(proxy: SocketAddr, msg: &str) -> String {
    let mut stream = TcpStream::connect(proxy).await.unwrap();
    stream.write_all(format!("{msg}\n").as_bytes()).await.unwrap();
    let mut line = String::new();
    let _ = BufReader::new(&mut stream).read_line(&mut line).await;
    if line.is_empty() { "(closed)".into() } else { line.trim().into() }
}

#[tokio::main]
async fn main() {
    let flags: Vec<Arc<AtomicBool>> = (0..3).map(|_| Arc::new(AtomicBool::new(true))).collect();
    let mut addrs = Vec::new();
    for (name, up) in flags.iter().enumerate() {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        addrs.push(listener.local_addr().unwrap());
        tokio::spawn(backend(name, listener, up.clone()));
    }
    let (tx, rx) = watch::channel(vec![true; 3]);
    tokio::spawn(health(addrs.clone(), tx));
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let front = listener.local_addr().unwrap();
    tokio::spawn(proxy(listener, addrs, rx.clone()));
    let mut watcher = rx.clone();

    println!("phase 1: all up");
    for i in 0..3 {
        println!("  request {i} -> {}", ask(front, &format!("hello {i}")).await);
    }

    println!("phase 2: backend 1 hangs");
    flags[1].store(false, SeqCst);
    watcher.wait_for(|status| !status[1]).await.unwrap();
    println!("  health: {:?}", *rx.borrow());
    for i in 3..7 {
        println!("  request {i} -> {}", ask(front, &format!("hello {i}")).await);
    }

    println!("phase 3: backend 1 recovers");
    flags[1].store(true, SeqCst);
    watcher.wait_for(|status| status[1]).await.unwrap();
    println!("  health: {:?}", *rx.borrow());
    for i in 7..10 {
        println!("  request {i} -> {}", ask(front, &format!("hello {i}")).await);
    }

    println!("phase 4: every backend hangs");
    for flag in &flags {
        flag.store(false, SeqCst);
    }
    watcher.wait_for(|status| status.iter().all(|up| !up)).await.unwrap();
    println!("  health: {:?}", *rx.borrow());
    println!("  request 10 -> {}", ask(front, "hello 10").await);
}

Output

phase 1: all up
  request 0 -> b0: hello 0
  request 1 -> b1: hello 1
  request 2 -> b2: hello 2
phase 2: backend 1 hangs
  health: [true, false, true]
  request 3 -> b2: hello 3
  request 4 -> b0: hello 4
  request 5 -> b2: hello 5
  request 6 -> b0: hello 6
phase 3: backend 1 recovers
  health: [true, true, true]
  request 7 -> b1: hello 7
  request 8 -> b2: hello 8
  request 9 -> b0: hello 9
phase 4: every backend hangs
  health: [false, false, false]
  request 10 -> (closed)

The round-robin counter keeps counting while the healthy set changes, so the backend chosen after a change depends on the counter and the size of the healthy list. In phase 4 the proxy has no target and drops the client, which reads an empty line.

Part	Role
`backend` with `up` flag	echo server that can be made to hang without closing sockets
`probe`	connect, `PING`, expect `PONG`, all inside one `timeout`
`health` with `interval`	probes on a schedule and publishes only changes
`watch` channel	one writer, many readers, always holds the latest status
`proxy`	picks a healthy backend by counter and forwards bytes
`copy_bidirectional`	copies both directions until both sides close

Extension tasks:

Require two failed probes before marking a backend down, to avoid flapping.
Track connections per backend and choose the backend with the fewest (least connections).
Close in-flight client connections to a backend when it goes down, using a per-backend CancellationToken.

Project 20: Supervised actors with restart

An actor holds state and handles messages one at a time. If the actor panics, a supervisor starts a new one, waits longer after each failure, and gives up after a limit. Callers keep using the same handle.

Requirements:

The actor is a task with a mailbox (mpsc). Requests carry a oneshot sender for the reply.
A Boom message panics the actor.
The supervisor awaits the actor's JoinHandle, sees the panic as a JoinError, waits 100 ms, 200 ms, 400 ms, and restarts.
The mailbox survives restarts, so messages sent during the wait are handled by the new actor.
After three restarts the supervisor stops, and the handle reports that the actor is gone.

Stage 1: An actor that panics

Without a supervisor, a panic ends the task. The reply sender inside the message is dropped during unwinding, so the caller gets an error instead of waiting forever. The mailbox receiver is dropped with the task, so later sends fail. The hook that prints panic messages is silenced to keep the output fixed.

Rust

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

enum Cmd {
    Add(i64, oneshot::Sender<i64>),
    Boom(oneshot::Sender<i64>),
}

#[derive(Clone)]
struct Handle { tx: mpsc::Sender<Cmd> }

impl Handle {
    async fn call(&self, make: impl FnOnce(oneshot::Sender<i64>) -> Cmd) -> Result<i64, &'static str> {
        let (reply, answer) = oneshot::channel();
        self.tx.send(make(reply)).await.map_err(|_| "actor stopped")?;
        answer.await.map_err(|_| "no reply")
    }
    async fn add(&self, n: i64) -> Result<i64, &'static str> { self.call(|r| Cmd::Add(n, r)).await }
    async fn boom(&self) -> Result<i64, &'static str> { self.call(Cmd::Boom).await }
}

async fn actor(mut inbox: mpsc::Receiver<Cmd>) {
    let mut total = 0;
    while let Some(cmd) = inbox.recv().await {
        match cmd {
            Cmd::Add(n, reply) => { total += n; let _ = reply.send(total); }
            Cmd::Boom(_reply) => panic!("boom"),
        }
    }
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    std::panic::set_hook(Box::new(|_| {}));
    let (tx, rx) = mpsc::channel(8);
    let handle = Handle { tx };
    let task = tokio::spawn(actor(rx));

    println!("add 5 -> {:?}", handle.add(5).await);
    println!("add 3 -> {:?}", handle.add(3).await);
    println!("boom -> {:?}", handle.boom().await);
    let joined = task.await;
    println!("task panicked: {}", joined.unwrap_err().is_panic());
    println!("add 1 -> {:?}", handle.add(1).await);
}

Output

add 5 -> Ok(5)
add 3 -> Ok(8)
boom -> Err("no reply")
task panicked: true
add 1 -> Err("actor stopped")

The two error messages differ. no reply means the request was received but the actor died before answering. actor stopped means the mailbox is closed.

Stage 2: A supervisor that restarts

The receiver moves into Arc<tokio::sync::Mutex<Receiver>>. Each actor incarnation locks it for its whole life. The lock guard drops when the actor unwinds, and Tokio's mutex has no poisoning, so the next incarnation can lock it. The supervisor spawns the actor, awaits its JoinHandle, and on a panic sleeps and spawns again. State inside the actor (total) starts from zero each time.

Rust

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

enum Cmd {
    Add(i64, oneshot::Sender<i64>),
    Boom(oneshot::Sender<i64>),
}

#[derive(Clone)]
struct Handle { tx: mpsc::Sender<Cmd> }

impl Handle {
    async fn call(&self, make: impl FnOnce(oneshot::Sender<i64>) -> Cmd) -> Result<i64, &'static str> {
        let (reply, answer) = oneshot::channel();
        self.tx.send(make(reply)).await.map_err(|_| "actor stopped")?;
        answer.await.map_err(|_| "no reply")
    }
    async fn add(&self, n: i64) -> Result<i64, &'static str> { self.call(|r| Cmd::Add(n, r)).await }
    async fn boom(&self) -> Result<i64, &'static str> { self.call(Cmd::Boom).await }
}

async fn actor(inbox: Arc<Mutex<mpsc::Receiver<Cmd>>>) {
    let mut inbox = inbox.lock().await;
    let mut total = 0;
    while let Some(cmd) = inbox.recv().await {
        match cmd {
            Cmd::Add(n, reply) => { total += n; let _ = reply.send(total); }
            Cmd::Boom(_reply) => panic!("boom"),
        }
    }
}

async fn supervise(inbox: Arc<Mutex<mpsc::Receiver<Cmd>>>, start: Instant) {
    let mut restarts = 0u32;
    loop {
        match tokio::spawn(actor(inbox.clone())).await {
            Ok(()) => break,
            Err(e) if e.is_panic() => {
                let delay = Duration::from_millis(100 << restarts);
                restarts += 1;
                println!("[{:>3} ms] actor panicked, restart {restarts} in {delay:?}", start.elapsed().as_millis());
                sleep(delay).await;
            }
            Err(_) => break,
        }
    }
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    std::panic::set_hook(Box::new(|_| {}));
    let start = Instant::now();
    let (tx, rx) = mpsc::channel(8);
    let handle = Handle { tx };
    tokio::spawn(supervise(Arc::new(Mutex::new(rx)), start));

    println!("add 5 -> {:?}", handle.add(5).await);
    println!("add 3 -> {:?}", handle.add(3).await);
    println!("boom -> {:?}", handle.boom().await);
    println!("[{:>3} ms] add 1 -> {:?}", start.elapsed().as_millis(), handle.add(1).await);
    println!("[{:>3} ms] add 1 -> {:?}", start.elapsed().as_millis(), handle.add(1).await);
}

Output

add 5 -> Ok(5)
add 3 -> Ok(8)
[  0 ms] actor panicked, restart 1 in 100ms
boom -> Err("no reply")
[  0 ms] add 1 -> Ok(1)
[100 ms] add 1 -> Ok(2)

The add 1 call is sent right after the crash and waits in the mailbox. It is answered at 100 ms by the second incarnation, which started with total = 0.

Stage 3: Backoff limit and giving up

The supervisor stops after max_restarts. It returns, its Arc to the mailbox drops, the channel closes, and callers get actor stopped. The restart count is returned to main. This is the full program.

Rust

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

enum Cmd {
    Add(i64, oneshot::Sender<i64>),
    Boom(oneshot::Sender<i64>),
}

#[derive(Clone)]
struct Handle { tx: mpsc::Sender<Cmd> }

impl Handle {
    async fn call(&self, make: impl FnOnce(oneshot::Sender<i64>) -> Cmd) -> Result<i64, &'static str> {
        let (reply, answer) = oneshot::channel();
        self.tx.send(make(reply)).await.map_err(|_| "actor stopped")?;
        answer.await.map_err(|_| "no reply")
    }
    async fn add(&self, n: i64) -> Result<i64, &'static str> { self.call(|r| Cmd::Add(n, r)).await }
    async fn boom(&self) -> Result<i64, &'static str> { self.call(Cmd::Boom).await }
}

async fn actor(inbox: Arc<Mutex<mpsc::Receiver<Cmd>>>) {
    let mut inbox = inbox.lock().await;
    let mut total = 0;
    while let Some(cmd) = inbox.recv().await {
        match cmd {
            Cmd::Add(n, reply) => { total += n; let _ = reply.send(total); }
            Cmd::Boom(_reply) => panic!("boom"),
        }
    }
}

async fn supervise(inbox: Arc<Mutex<mpsc::Receiver<Cmd>>>, max_restarts: u32, start: Instant) -> u32 {
    let mut restarts = 0u32;
    loop {
        match tokio::spawn(actor(inbox.clone())).await {
            Ok(()) => return restarts,
            Err(e) if e.is_panic() => {
                if restarts == max_restarts {
                    println!("[{:>3} ms] actor panicked, giving up", start.elapsed().as_millis());
                    return restarts;
                }
                let delay = Duration::from_millis(100 << restarts);
                restarts += 1;
                println!("[{:>3} ms] actor panicked, restart {restarts} in {delay:?}", start.elapsed().as_millis());
                sleep(delay).await;
            }
            Err(_) => return restarts,
        }
    }
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    std::panic::set_hook(Box::new(|_| {}));
    let start = Instant::now();
    let (tx, rx) = mpsc::channel(8);
    let handle = Handle { tx };
    let supervisor = tokio::spawn(supervise(Arc::new(Mutex::new(rx)), 3, start));

    println!("add 5 -> {:?}", handle.add(5).await);
    for round in 1..=4 {
        println!("boom {round} -> {:?}", handle.boom().await);
    }
    println!("add 1 -> {:?}", handle.add(1).await);
    println!("restarts used: {}", supervisor.await.unwrap());
}

Output

add 5 -> Ok(5)
[  0 ms] actor panicked, restart 1 in 100ms
boom 1 -> Err("no reply")
[100 ms] actor panicked, restart 2 in 200ms
boom 2 -> Err("no reply")
[300 ms] actor panicked, restart 3 in 400ms
boom 3 -> Err("no reply")
[700 ms] actor panicked, giving up
boom 4 -> Err("no reply")
add 1 -> Err("actor stopped")
restarts used: 3

Each crash doubles the wait: 100, 200, 400 ms. The fourth crash exceeds the limit, so the supervisor stops at 700 ms, and every later call fails with actor stopped. Each boom call is answered with no reply at the moment of the crash, and the next boom is queued while the supervisor sleeps.

Part	Role
`Handle`	cloneable sender wrapper; `call` builds a message with a `oneshot` reply
`actor`	processes one message at a time and owns its state
`Arc<Mutex<Receiver>>`	keeps the mailbox alive across restarts
`supervise`	awaits the `JoinHandle`, tells a panic from a clean exit with `is_panic`
backoff `100 << restarts`	doubling delay before each restart
restart limit	turns a crash loop into a visible failure

Extension tasks:

Reset the restart counter when an incarnation has run for more than one second without a panic.
Keep the state across restarts by having the actor send total to the supervisor after every message and receive it back on start.
Supervise two actors with one supervisor using a JoinSet and restart only the one that failed.

Interview points

A Semaphore acquired inside each task bounds concurrency while the JoinSet still holds every task. Acquiring before spawn also bounds the number of tasks.
abort_all on a JoinSet cancels tasks that are running and tasks that are waiting for a permit. join_next reports them as cancelled JoinErrors.
A graceful shutdown has an order: stop accepting, close idle connections, drain running work with a deadline, then cancel what is left.
TaskTracker::wait completes only after close has been called and the count reaches zero.
select! on a token is safe only around cancel-safe futures such as next_line. Work started in a branch body is not cancelled by the token.
interval_at sets the first tick. MissedTickBehavior::Skip drops ticks that came due during a slow run.
Pool guards return their resource in Drop. Field order decides whether the resource is back before the slot is released.
A dropped timeout(get()) releases its place in the semaphore queue, so a timed-out caller holds nothing.
watch::send_if_modified wakes receivers only when the value changed. wait_for checks the current value first, so a change made before the call is not missed.
A health check needs its own timeout, because a hung peer accepts connections and never replies.
A supervisor sees a task panic as JoinError::is_panic. Keeping the mailbox outside the task lets callers keep one handle across restarts.

Summary

Project 15 combines JoinSet, Semaphore, and an error policy (collect all or fail fast with abort_all).
Project 16 shuts down in stages: cancel token, accept loop exit, TaskTracker drain, deadline with timeout, and a second token for the hard stop.
Project 17 runs jobs on interval_at with Skip, and stops between runs with a CancellationToken.
Project 18 builds a pool from Semaphore, Mutex<Vec<_>>, and an RAII guard. get_timeout and discard extend it.
Project 19 forwards bytes with copy_bidirectional, probes backends under timeout, and publishes status on a watch channel.
Project 20 restarts a panicked actor with doubling backoff and a restart limit, keeping the mailbox in an Arc<Mutex<Receiver>>.