codeloopback
Guide contents

Lesson 12 of 44

Workshop 1: Tasks and Time

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

The projects use only what the earlier lessons of this course introduced: tokio::spawn and JoinHandle (Spawning Tasks), join_all (join!, try_join!, and Running Work Concurrently), abort (Cancellation, abort, and Panics), select! (select!: Racing Futures), spawn_blocking (Blocking and CPU-Bound Work), and sleep, timeout, interval, and the paused clock (Time: sleep, timeout, and interval).

Projects 1, 3, and 4 run under the paused clock (start_paused = true). Their times are exact and their output does not depend on the machine. Project 2 does real computation and prints only values that depend on the data.

Project 1: Concurrent countdown timers

A kitchen has several timers. Each counts down at its own pace, and all must run at once. One timer can be cancelled before it finishes.

Requirements:

  • A countdown has a name, a start value, and a tick length.
  • Each countdown prints its remaining value at every tick and liftoff at the end, with the time in milliseconds since the program started.
  • All countdowns run concurrently, each as its own task.
  • The program waits for every timer and reports how each ended.
  • One timer is cancelled part way through, and the report shows that it was cancelled.

Stage 1: One countdown

A countdown is an async fn with a loop and a sleep. The start instant is passed in so that every line shows the time since the program began.

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

async fn countdown(name: &str, from: u32, tick: Duration, start: Instant) {
    for remaining in (1..=from).rev() {
        println!("[{:>4} ms] {name}: {remaining}", start.elapsed().as_millis());
        sleep(tick).await;
    }
    println!("[{:>4} ms] {name}: liftoff", start.elapsed().as_millis());
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let start = Instant::now();
    countdown("tea", 3, Duration::from_millis(300), start).await;
}
Output
[   0 ms] tea: 3
[ 300 ms] tea: 2
[ 600 ms] tea: 1
[ 900 ms] tea: liftoff

The clock is paused, so the three sleeps of 300 ms finish instantly in real time and the times are exact.

Stage 2: Run several countdowns at once

Awaiting countdowns one after another would take the sum of their times. Spawning each one makes them run concurrently. A spawned task must own its data ('static), so the name is a &'static str and the other values are Copy. The tick lengths differ, and no two ticks fall on the same instant after the start, so the interleaving is fixed.

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

async fn countdown(name: &'static str, from: u32, tick: Duration, start: Instant) {
    for remaining in (1..=from).rev() {
        println!("[{:>4} ms] {name}: {remaining}", start.elapsed().as_millis());
        sleep(tick).await;
    }
    println!("[{:>4} ms] {name}: liftoff", start.elapsed().as_millis());
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let start = Instant::now();
    let timers = [("tea", 3, 300), ("eggs", 4, 250), ("toast", 2, 700)];

    let mut handles = Vec::new();
    for (name, from, tick_ms) in timers {
        handles.push(tokio::spawn(countdown(
            name,
            from,
            Duration::from_millis(tick_ms),
            start,
        )));
    }

    for handle in handles {
        handle.await.unwrap();
    }
    println!("[{:>4} ms] all timers finished", start.elapsed().as_millis());
}
Output
[   0 ms] tea: 3
[   0 ms] eggs: 4
[   0 ms] toast: 2
[ 250 ms] eggs: 3
[ 300 ms] tea: 2
[ 500 ms] eggs: 2
[ 600 ms] tea: 1
[ 700 ms] toast: 1
[ 750 ms] eggs: 1
[ 900 ms] tea: liftoff
[1000 ms] eggs: liftoff
[1400 ms] toast: liftoff
[1400 ms] all timers finished

The three first lines share the time 0 and appear in spawn order, because the current_thread runtime polls new tasks in the order they were spawned. After that the lines follow the deadlines. The whole run took 1400 ms of virtual time, the length of the longest timer, not the sum of all three.

Stage 3: Results, cancellation, and a report

The final program gives each countdown a return value (the number of ticks it completed), adds a fourth timer that is cancelled at 1100 ms, and reports each outcome. A cancelled task returns Err(JoinError) from its handle, and is_cancelled() tells that apart from a panic. The tick count lives in a variable that the task returns only when it finishes, so a cancelled timer has no count.

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

struct Timer {
    name: &'static str,
    from: u32,
    tick_ms: u64,
}

async fn countdown(timer: Timer, start: Instant) -> u32 {
    let tick = Duration::from_millis(timer.tick_ms);
    let mut ticks = 0;
    for remaining in (1..=timer.from).rev() {
        println!(
            "[{:>4} ms] {}: {remaining}",
            start.elapsed().as_millis(),
            timer.name
        );
        sleep(tick).await;
        ticks += 1;
    }
    println!(
        "[{:>4} ms] {}: liftoff",
        start.elapsed().as_millis(),
        timer.name
    );
    ticks
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let start = Instant::now();
    let timers = [
        Timer { name: "tea", from: 3, tick_ms: 300 },
        Timer { name: "eggs", from: 4, tick_ms: 250 },
        Timer { name: "toast", from: 2, tick_ms: 700 },
        Timer { name: "pasta", from: 5, tick_ms: 400 },
    ];

    let mut running: Vec<(&'static str, JoinHandle<u32>)> = Vec::new();
    for timer in timers {
        let name = timer.name;
        running.push((name, tokio::spawn(countdown(timer, start))));
    }

    sleep(Duration::from_millis(1100)).await;
    println!("[{:>4} ms] cancelling pasta", start.elapsed().as_millis());
    running[3].1.abort();

    println!("--- report ---");
    for (name, handle) in running {
        match handle.await {
            Ok(ticks) => println!("{name}: finished after {ticks} ticks"),
            Err(e) if e.is_cancelled() => println!("{name}: cancelled"),
            Err(e) => println!("{name}: failed ({e})"),
        }
    }
}
Output
[   0 ms] tea: 3
[   0 ms] eggs: 4
[   0 ms] toast: 2
[   0 ms] pasta: 5
[ 250 ms] eggs: 3
[ 300 ms] tea: 2
[ 400 ms] pasta: 4
[ 500 ms] eggs: 2
[ 600 ms] tea: 1
[ 700 ms] toast: 1
[ 750 ms] eggs: 1
[ 800 ms] pasta: 3
[ 900 ms] tea: liftoff
[1000 ms] eggs: liftoff
[1100 ms] cancelling pasta
--- report ---
tea: finished after 3 ticks
eggs: finished after 4 ticks
[1400 ms] toast: liftoff
toast: finished after 2 ticks
pasta: cancelled

abort takes effect at the timer's next .await, which is the sleep. The pasta task never printed liftoff. The tea, eggs, and toast tasks had already finished when their handles were awaited, and the awaits returned at once.

What each part does:

Part Role
Timer plain data describing one countdown, moved into its task
countdown the loop of prints and sleeps; returns the tick count
tokio::spawn starts each countdown as its own task
running keeps the name next to each JoinHandle so the report can name it
abort cancels one task at its next .await
match handle.await separates normal finish, cancellation, and failure

Extension tasks:

  1. Add a Duration limit for the whole run with timeout and cancel every timer that has not finished when it expires.
  2. Print the report in finishing order instead of spawn order by awaiting the handles with select! or by having each task print its own result.
  3. Give each timer a label from user input, for example std::env::args, and report the sum of all completed ticks.

Project 2: Parallel checksums of generated data

A program must compute a checksum for each of several large data blocks. The computation is CPU work, so it must not run on the async worker threads. The results must come out in block order even though the jobs finish in any order.

Requirements:

  • Blocks of pseudo-random bytes are generated from a seed, so the data is the same on every run.
  • The checksum is FNV-1a with 64 bits, printed as 16 hexadecimal digits.
  • Each block is processed on the blocking pool with spawn_blocking.
  • Results are reported in block order.
  • The parallel result must equal a sequential computation.
  • A block of length zero is an error that the job returns; a job that panics or is cancelled is reported separately.

Stage 1: Sequential baseline

Two plain functions do the work. generate uses a linear congruential generator (a formula state = state * a + c that produces a repeatable sequence) and takes the top byte of each state. fnv1a folds each byte into a 64-bit hash with an XOR and a multiplication. Both use wrapping arithmetic so overflow is defined.

Rust
fn generate(seed: u64, len: usize) -> Vec<u8> {
    let mut state = seed;
    (0..len)
        .map(|_| {
            state = state
                .wrapping_mul(6364136223846793005)
                .wrapping_add(1442695040888963407);
            (state >> 56) as u8
        })
        .collect()
}

fn fnv1a(data: &[u8]) -> u64 {
    let mut hash: u64 = 0xcbf29ce484222325;
    for &byte in data {
        hash ^= u64::from(byte);
        hash = hash.wrapping_mul(0x100000001b3);
    }
    hash
}

fn main() {
    let sizes = [200_000usize, 300_000, 100_000, 250_000];
    for (index, &len) in sizes.iter().enumerate() {
        let data = generate(index as u64 + 1, len);
        println!("block {index}: {len} bytes, checksum {:016x}", fnv1a(&data));
    }
}
Output
block 0: 200000 bytes, checksum c49e58a4b7700cf7
block 1: 300000 bytes, checksum 6aa498484c375022
block 2: 100000 bytes, checksum 779d9df934c91277
block 3: 250000 bytes, checksum 9c8d410f99496824

This program has no runtime. It defines the values that the parallel version must reproduce.

Stage 2: One blocking job per block

spawn_blocking moves each job to the blocking pool. Generation and hashing both happen inside the closure, so the async side only starts jobs and waits. The handles are awaited in the order they were created, so the printed order matches the block order.

Rust
fn generate(seed: u64, len: usize) -> Vec<u8> {
    let mut state = seed;
    (0..len)
        .map(|_| {
            state = state
                .wrapping_mul(6364136223846793005)
                .wrapping_add(1442695040888963407);
            (state >> 56) as u8
        })
        .collect()
}

fn fnv1a(data: &[u8]) -> u64 {
    let mut hash: u64 = 0xcbf29ce484222325;
    for &byte in data {
        hash ^= u64::from(byte);
        hash = hash.wrapping_mul(0x100000001b3);
    }
    hash
}

#[tokio::main]
async fn main() {
    let sizes = [200_000usize, 300_000, 100_000, 250_000];

    let mut handles = Vec::new();
    for (index, &len) in sizes.iter().enumerate() {
        handles.push(tokio::task::spawn_blocking(move || {
            let data = generate(index as u64 + 1, len);
            fnv1a(&data)
        }));
    }

    for (index, handle) in handles.into_iter().enumerate() {
        let checksum = handle.await.unwrap();
        println!("block {index}: checksum {checksum:016x}");
    }
}
Output
block 0: checksum c49e58a4b7700cf7
block 1: checksum 6aa498484c375022
block 2: checksum 779d9df934c91277
block 3: checksum 9c8d410f99496824

The checksums equal those of Stage 1. The jobs ran on up to four blocking threads at the same time, and awaiting the handles in order restored the block order.

Stage 3: All handles at once, errors, and verification

The full program awaits all handles together with futures::future::join_all, which returns the results in input order. A job returns Result<u64, ChecksumError> so that a bad block is an ordinary error. Each handle can also fail with a JoinError, so the result of one handle is Result<Result<u64, ChecksumError>, JoinError>. The program combines the block checksums into one value and compares it with a sequential computation on the same data.

Rust
use futures::future::join_all;
use std::fmt;

#[derive(Debug)]
enum ChecksumError {
    EmptyBlock,
}

impl fmt::Display for ChecksumError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            ChecksumError::EmptyBlock => write!(f, "block is empty"),
        }
    }
}

fn generate(seed: u64, len: usize) -> Vec<u8> {
    let mut state = seed;
    (0..len)
        .map(|_| {
            state = state
                .wrapping_mul(6364136223846793005)
                .wrapping_add(1442695040888963407);
            (state >> 56) as u8
        })
        .collect()
}

fn fnv1a(data: &[u8]) -> u64 {
    let mut hash: u64 = 0xcbf29ce484222325;
    for &byte in data {
        hash ^= u64::from(byte);
        hash = hash.wrapping_mul(0x100000001b3);
    }
    hash
}

fn checksum_block(seed: u64, len: usize) -> Result<u64, ChecksumError> {
    if len == 0 {
        return Err(ChecksumError::EmptyBlock);
    }
    Ok(fnv1a(&generate(seed, len)))
}

fn combine(checksums: &[u64]) -> u64 {
    let bytes: Vec<u8> = checksums.iter().flat_map(|c| c.to_le_bytes()).collect();
    fnv1a(&bytes)
}

#[tokio::main]
async fn main() {
    let sizes = [200_000usize, 300_000, 0, 100_000, 250_000];

    let handles: Vec<_> = sizes
        .iter()
        .enumerate()
        .map(|(index, &len)| {
            tokio::task::spawn_blocking(move || checksum_block(index as u64 + 1, len))
        })
        .collect();
    let results = join_all(handles).await;

    let mut good = Vec::new();
    for (index, result) in results.into_iter().enumerate() {
        match result {
            Ok(Ok(checksum)) => {
                println!("block {index}: checksum {checksum:016x}");
                good.push(checksum);
            }
            Ok(Err(e)) => println!("block {index}: error: {e}"),
            Err(e) => println!("block {index}: job failed: {e}"),
        }
    }

    let sequential: Vec<u64> = sizes
        .iter()
        .enumerate()
        .filter_map(|(index, &len)| checksum_block(index as u64 + 1, len).ok())
        .collect();

    println!("blocks ok: {}", good.len());
    println!("combined: {:016x}", combine(&good));
    println!("matches sequential: {}", good == sequential);
}
Output
block 0: checksum c49e58a4b7700cf7
block 1: checksum 6aa498484c375022
block 2: error: block is empty
block 3: checksum 6aec435e35154a8f
block 4: checksum 0717ef2d6692177f
blocks ok: 4
combined: 9f9d3763fd4a7fb0
matches sequential: true

join_all waits for every handle, so one bad block does not hide the others. The order of results follows the order of handles, whatever order the jobs finished in.

What each part does:

Part Role
generate repeatable pseudo-random bytes from a seed
fnv1a 64-bit FNV-1a hash of a byte slice
checksum_block one job: validate, generate, hash
spawn_blocking runs a job on the blocking pool
join_all collects all results in input order
combine hashes the little-endian bytes of the block checksums into one value
matches sequential checks the parallel run against the single-threaded code

Extension tasks:

  1. Limit the work to four jobs at a time when there are many blocks, using the batching idea: spawn four handles, await them, then spawn the next four.
  2. Replace the inner hash loop with rayon parallel iterators over 64 KiB pieces of one large block (Blocking and CPU-Bound Work, Example 12).
  3. Add a job that panics for one input and print the resulting JoinError with is_panic().

Project 3: Retry with exponential backoff and timeout

A client calls an unreliable operation. A call can fail or hang. The client must limit each call with a timeout, wait longer after each failure, and stop after a fixed number of attempts.

Requirements:

  • Each attempt has a time limit; an attempt that exceeds it counts as a failed attempt.
  • After a failed attempt the client waits base_delay * 2^(failures - 1), but never more than max_delay.
  • After max_attempts failures the last error is returned.
  • The retry function works with any operation that returns a future of Result<T, String>.
  • The program prints the time of each event so the backoff is visible.

Stage 1: A fixed delay between attempts

The first version retries an operation that fails twice and then succeeds. The delay is constant. attempt is passed to the operation so that it can decide how to behave.

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

async fn flaky(attempt: u32) -> Result<&'static str, String> {
    sleep(Duration::from_millis(20)).await;
    if attempt < 3 {
        Err(format!("attempt {attempt} failed"))
    } else {
        Ok("payload")
    }
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let start = Instant::now();
    let mut attempt = 0;
    loop {
        attempt += 1;
        match flaky(attempt).await {
            Ok(value) => {
                println!("[{:>4} ms] success: {value}", start.elapsed().as_millis());
                break;
            }
            Err(e) => {
                println!("[{:>4} ms] error: {e}", start.elapsed().as_millis());
                sleep(Duration::from_millis(100)).await;
            }
        }
    }
}
Output
[  20 ms] error: attempt 1 failed
[ 140 ms] error: attempt 2 failed
[ 260 ms] success: payload

Each cycle takes the 20 ms of the call plus a 100 ms pause. This loop retries forever if the operation never succeeds.

Stage 2: Exponential backoff with a cap and a limit

A RetryPolicy holds the settings. delay_for(failures) computes base_delay * 2^(failures - 1) and caps it at max_delay. Duration supports multiplication by a u32, and checked_mul and min keep the result bounded. retry stops after max_attempts.

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

struct RetryPolicy {
    max_attempts: u32,
    base_delay: Duration,
    max_delay: Duration,
}

impl RetryPolicy {
    fn delay_for(&self, failures: u32) -> Duration {
        let factor = 2u32.saturating_pow(failures - 1);
        self.base_delay
            .checked_mul(factor)
            .unwrap_or(self.max_delay)
            .min(self.max_delay)
    }
}

async fn always_fails(attempt: u32) -> Result<(), String> {
    sleep(Duration::from_millis(20)).await;
    Err(format!("attempt {attempt} failed"))
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let policy = RetryPolicy {
        max_attempts: 6,
        base_delay: Duration::from_millis(100),
        max_delay: Duration::from_millis(500),
    };

    for failures in 1..policy.max_attempts {
        println!("delay after failure {failures}: {:?}", policy.delay_for(failures));
    }

    let start = Instant::now();
    let mut attempt = 0;
    let last_error = loop {
        attempt += 1;
        let error = match always_fails(attempt).await {
            Ok(()) => return,
            Err(e) => e,
        };
        println!("[{:>4} ms] {error}", start.elapsed().as_millis());
        if attempt == policy.max_attempts {
            break error;
        }
        sleep(policy.delay_for(attempt)).await;
    };
    println!("gave up: {last_error}");
}
Output
delay after failure 1: 100ms
delay after failure 2: 200ms
delay after failure 3: 400ms
delay after failure 4: 500ms
delay after failure 5: 500ms
[  20 ms] attempt 1 failed
[ 140 ms] attempt 2 failed
[ 360 ms] attempt 3 failed
[ 780 ms] attempt 4 failed
[1300 ms] attempt 5 failed
[1820 ms] attempt 6 failed
gave up: attempt 6 failed

The delays double from 100 ms and stop at the 500 ms cap. The program stopped after six attempts and returned the last error.

Stage 3: Per-attempt timeout and a generic retry function

The full program adds timeout around each attempt and turns the loop into a generic function. retry takes a closure that builds a new future for each attempt. The error type AttemptError separates an operation error from a timeout. The first scenario uses an operation that fails, hangs, fails again, and then succeeds. The second scenario never succeeds.

Rust
use std::fmt;
use std::future::Future;
use std::time::Duration;
use tokio::time::{sleep, timeout, Instant};

#[derive(Debug)]
enum AttemptError {
    TimedOut,
    Failed(String),
}

impl fmt::Display for AttemptError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            AttemptError::TimedOut => write!(f, "timed out"),
            AttemptError::Failed(msg) => write!(f, "failed: {msg}"),
        }
    }
}

struct RetryPolicy {
    max_attempts: u32,
    base_delay: Duration,
    max_delay: Duration,
    attempt_timeout: Duration,
}

impl RetryPolicy {
    fn delay_for(&self, failures: u32) -> Duration {
        let factor = 2u32.saturating_pow(failures - 1);
        self.base_delay
            .checked_mul(factor)
            .unwrap_or(self.max_delay)
            .min(self.max_delay)
    }
}

async fn retry<T, F, Fut>(
    policy: &RetryPolicy,
    start: Instant,
    mut operation: F,
) -> Result<T, AttemptError>
where
    F: FnMut(u32) -> Fut,
    Fut: Future<Output = Result<T, String>>,
{
    let mut attempt = 0;
    loop {
        attempt += 1;
        let error = match timeout(policy.attempt_timeout, operation(attempt)).await {
            Ok(Ok(value)) => return Ok(value),
            Ok(Err(message)) => AttemptError::Failed(message),
            Err(_elapsed) => AttemptError::TimedOut,
        };
        println!(
            "[{:>4} ms] attempt {attempt} {error}",
            start.elapsed().as_millis()
        );
        if attempt == policy.max_attempts {
            return Err(error);
        }
        let delay = policy.delay_for(attempt);
        println!("           waiting {delay:?}");
        sleep(delay).await;
    }
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let policy = RetryPolicy {
        max_attempts: 5,
        base_delay: Duration::from_millis(100),
        max_delay: Duration::from_millis(400),
        attempt_timeout: Duration::from_millis(200),
    };

    println!("== scenario 1: recovers ==");
    let start = Instant::now();
    let result = retry(&policy, start, |attempt| async move {
        match attempt {
            1 => {
                sleep(Duration::from_millis(20)).await;
                Err("connection refused".to_string())
            }
            2 => {
                sleep(Duration::from_secs(10)).await;
                Ok("too late")
            }
            3 => Err("service unavailable".to_string()),
            _ => {
                sleep(Duration::from_millis(30)).await;
                Ok("payload")
            }
        }
    })
    .await;
    match result {
        Ok(value) => println!("[{:>4} ms] success: {value}", start.elapsed().as_millis()),
        Err(e) => println!("[{:>4} ms] gave up: {e}", start.elapsed().as_millis()),
    }

    println!("== scenario 2: gives up ==");
    let short = RetryPolicy { max_attempts: 3, ..policy };
    let start = Instant::now();
    let result: Result<(), AttemptError> = retry(&short, start, |attempt| async move {
        sleep(Duration::from_millis(10)).await;
        Err(format!("error {attempt}"))
    })
    .await;
    if let Err(e) = result {
        println!("[{:>4} ms] gave up: {e}", start.elapsed().as_millis());
    }
}
Output
== scenario 1: recovers ==
[  20 ms] attempt 1 failed: connection refused
           waiting 100ms
[ 320 ms] attempt 2 timed out
           waiting 200ms
[ 520 ms] attempt 3 failed: service unavailable
           waiting 400ms
[ 950 ms] success: payload
== scenario 2: gives up ==
[  10 ms] attempt 1 failed: error 1
           waiting 100ms
[ 120 ms] attempt 2 failed: error 2
           waiting 200ms
[ 330 ms] attempt 3 failed: error 3
[ 330 ms] gave up: failed: error 3

Attempt 2 would have taken ten seconds. The 200 ms timeout dropped it, and the client moved on. The waits between attempts were 100 ms, 200 ms, and 400 ms (the cap). In scenario 2 the last error is returned unchanged after the third attempt, without an extra wait.

What each part does:

Part Role
AttemptError distinguishes a timeout from an error returned by the operation
RetryPolicy::delay_for exponential delay with a cap
timeout(policy.attempt_timeout, ...) bounds one attempt and cancels it when the limit passes
retry runs the operation until success or max_attempts
closure ` attempt
paused clock makes every printed time exact

Extension tasks:

  1. Add random jitter to each delay (a random fraction of the delay) using the rand crate, and make the test deterministic by injecting the random source as a parameter.
  2. Add a retryable check so that some errors (for example "invalid request") stop the loop at once.
  3. Add an overall deadline for all attempts together with timeout_at, and return a separate error when it passes.

Project 4: Heartbeat monitor that can be stopped

A service reports that it is alive by sending a heartbeat at a fixed rate. The monitor must keep its schedule even when a check runs long, and it must be possible to stop it two ways: politely, by a signal it can react to, and forcibly, by cancelling its task.

Requirements:

  • A monitor beats at a fixed period using interval.
  • A beat runs a check that takes some time; when the check makes the monitor late, missed beats are skipped, not sent in a burst.
  • A monitor stops when a stop future completes, returns the number of beats it sent, and reports that it stopped.
  • A monitor can also be aborted; its state is still cleaned up by Drop.
  • The program runs two monitors with different periods and stops them in the two ways.

Stage 1: A heartbeat loop stopped with abort

The simplest stoppable monitor is a task with an endless loop. abort ends it at the next .await, which is tick(). The program lets it run for 350 ms.

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

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let start = Instant::now();

    let monitor = tokio::spawn(async move {
        let mut ticker = interval(Duration::from_millis(100));
        let mut beats = 0u32;
        loop {
            ticker.tick().await;
            beats += 1;
            println!("[{:>3} ms] heartbeat {beats}", start.elapsed().as_millis());
        }
    });

    sleep(Duration::from_millis(350)).await;
    monitor.abort();
    let result = monitor.await;
    println!("cancelled = {}", result.unwrap_err().is_cancelled());
}
Output
[  0 ms] heartbeat 1
[100 ms] heartbeat 2
[200 ms] heartbeat 3
[300 ms] heartbeat 4
cancelled = true

The first beat came at 0 ms because the first tick() completes immediately. Abort gives no chance to report the final count: the loop's variable beats was dropped with the task.

Stage 2: A polite stop with select!

The monitor takes a stop future. In each loop turn, select! waits for either the next tick or the stop future. When the stop future completes, the loop ends and returns the beat count. stop is pinned with tokio::pin! and polled as &mut stop, so it keeps its progress between loop turns (select!: Racing Futures). Any future can serve as the stop signal; here it is a sleep.

Rust
use std::future::Future;
use std::time::Duration;
use tokio::time::{interval, sleep, Instant};

async fn monitor(name: &'static str, period: Duration, stop: impl Future<Output = ()>, start: Instant) -> u32 {
    tokio::pin!(stop);
    let mut ticker = interval(period);
    let mut beats = 0;
    loop {
        tokio::select! {
            _ = ticker.tick() => {
                beats += 1;
                println!("[{:>3} ms] {name}: heartbeat {beats}", start.elapsed().as_millis());
            }
            _ = &mut stop => {
                println!("[{:>3} ms] {name}: stop requested", start.elapsed().as_millis());
                return beats;
            }
        }
    }
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let start = Instant::now();
    let beats = monitor(
        "db",
        Duration::from_millis(100),
        sleep(Duration::from_millis(350)),
        start,
    )
    .await;
    println!("db sent {beats} heartbeats");
}
Output
[  0 ms] db: heartbeat 1
[100 ms] db: heartbeat 2
[200 ms] db: heartbeat 3
[300 ms] db: heartbeat 4
[350 ms] db: stop requested
db sent 4 heartbeats

The monitor returned its count, which abort could not provide. tick is cancel safe, so a tick that loses the select! race is not consumed.

Stage 3: Two monitors, skipped beats, and cleanup

The full program runs two monitors as tasks. Each beat runs a check, modeled by a sleep of check_cost. The cache monitor has a check that takes longer than its period, so it misses ticks, and its interval uses MissedTickBehavior::Skip. The db monitor is stopped with a stop future. The cache monitor is aborted while a check is running. A guard value with a Drop implementation shows that the monitor's state is released on abort.

The cache interval starts 25 ms late (interval_at) so that no two monitors ever act at the same instant.

Rust
use std::future::Future;
use std::time::Duration;
use tokio::time::{interval_at, sleep, Instant, MissedTickBehavior};

struct Cleanup(&'static str);

impl Drop for Cleanup {
    fn drop(&mut self) {
        println!("{}: monitor state released", self.0);
    }
}

async fn monitor(
    name: &'static str,
    first_tick: Duration,
    period: Duration,
    check_cost: Duration,
    stop: impl Future<Output = ()>,
    start: Instant,
) -> u32 {
    let _cleanup = Cleanup(name);
    tokio::pin!(stop);
    let mut ticker = interval_at(start + first_tick, period);
    ticker.set_missed_tick_behavior(MissedTickBehavior::Skip);
    let mut beats = 0;
    loop {
        tokio::select! {
            _ = ticker.tick() => {}
            _ = &mut stop => {
                println!("[{:>3} ms] {name}: stop requested", start.elapsed().as_millis());
                return beats;
            }
        }
        beats += 1;
        println!("[{:>3} ms] {name}: heartbeat {beats}", start.elapsed().as_millis());
        sleep(check_cost).await;
    }
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let start = Instant::now();

    let db = tokio::spawn(monitor(
        "db",
        Duration::ZERO,
        Duration::from_millis(100),
        Duration::from_millis(30),
        sleep(Duration::from_millis(450)),
        start,
    ));
    let cache = tokio::spawn(monitor(
        "cache",
        Duration::from_millis(25),
        Duration::from_millis(150),
        Duration::from_millis(200),
        std::future::pending::<()>(),
        start,
    ));

    sleep(Duration::from_millis(700)).await;
    println!("[{:>3} ms] aborting cache", start.elapsed().as_millis());
    cache.abort();

    let db_beats = db.await.unwrap();
    println!("db sent {db_beats} heartbeats");
    match cache.await {
        Ok(beats) => println!("cache sent {beats} heartbeats"),
        Err(e) => println!("cache cancelled = {}", e.is_cancelled()),
    }
}
Output
[  0 ms] db: heartbeat 1
[ 25 ms] cache: heartbeat 1
[100 ms] db: heartbeat 2
[200 ms] db: heartbeat 3
[225 ms] cache: heartbeat 2
[300 ms] db: heartbeat 4
[400 ms] db: heartbeat 5
[425 ms] cache: heartbeat 3
[450 ms] db: stop requested
db: monitor state released
[625 ms] cache: heartbeat 4
[700 ms] aborting cache
db sent 5 heartbeats
cache: monitor state released
cache cancelled = true

The db monitor beat at 0, 100, 200, 300, and 400 ms, saw the stop future complete at 450 ms, and returned. The cache monitor beat at 25 ms, and its 200 ms check ended at 225 ms. The ticks scheduled for 175 ms (and none other before 225 ms) were missed. With Skip the next tick was the next multiple of 150 ms after the start, so the following beat came at 325 ms. It did not fire a burst of missed ticks. The abort at 700 ms arrived during the check that began at 625 ms, and the guard printed its message when the task's state was dropped.

std::future::pending::<()>() is a future that never completes, so the cache monitor has no polite stop and can end only through abort.

What each part does:

Part Role
interval_at starts the schedule at a chosen instant so the two monitors do not tick together
MissedTickBehavior::Skip drops ticks missed while a check runs long
select! on tick and &mut stop reacts to the stop signal at every turn
stop: impl Future<Output = ()> accepts any future as the stop signal
Cleanup and Drop shows that abort releases the task's state
abort forced stop; the task returns no value
JoinHandle<u32> returns the beat count of a monitor that stopped by itself

Extension tasks:

  1. Add a third monitor that stops when either of two futures completes: a deadline and a shared AtomicBool flag polled every 10 ms.
  2. Change Skip to Burst and Delay in turn and compare the beat times of the cache monitor.
  3. Make the monitor count the beats that were skipped and return the count with the beat total.

Interview points

  • A spawned task needs owned ('static) data. A countdown can take &'static str names and Copy numbers, or it can own a struct.
  • Awaiting spawned handles in order gives results in order even if the tasks finish in another order. join_all does the same in one call.
  • CPU-bound jobs belong on the blocking pool. Results returned from spawn_blocking come as Result<T, JoinError>, so a job that returns its own Result gives a nested result.
  • A retry loop needs three limits: a cap on attempts, a cap on the delay, and a timeout per attempt.
  • timeout around an operation drops the operation on expiry; work already sent to another system is not undone.
  • interval keeps its schedule; sleep in a loop adds the work time to every period. MissedTickBehavior chooses how a late interval catches up.
  • abort stops a task at its next .await and returns no value; a stop future checked in select! lets the task clean up and return a result.
  • The paused clock makes tests of delays exact and instant, because the runtime moves the clock to the next timer when idle.

Summary

  • Project 1: spawn, sleep, and JoinHandle run several countdowns at once; abort cancels one and is_cancelled reports it.
  • Project 2: spawn_blocking runs CPU work on the blocking pool; join_all returns results in input order; the parallel result equals the sequential one.
  • Project 3: timeout bounds each attempt, delay_for doubles the wait up to a cap, and a generic retry takes a closure that builds each attempt's future.
  • Project 4: interval_at, MissedTickBehavior::Skip, and select! on a tick and a stop future build a monitor that stops by signal or by abort.
  • All timing output in these projects is exact because the examples use the paused clock.