Workshop 6: Runtime Internals

Two projects apply the runtime-internals section. Project 24 builds time-based combinators by hand, using Future, Stream, and pin projection from Writing Futures and Streams by Hand. Project 25 runs a scheduler experiment whose printed counts follow from the constants in Runtime Builder and the Scheduler.

Project	Uses
24. Timeout, debounce, and retry combinators	`impl Future`, `impl Stream`, `pin-project-lite`, `Sleep::reset`, paused time
25. Scheduler experiment with runtime metrics	`Builder`, `Handle::metrics`, `global_queue_depth`, `worker_park_count`, `spawn_blocking`, `max_blocking_threads`

Both programs print only values that do not change between runs: times come from the paused clock, and scheduler results are counts.

Project 24: Timeout, debounce, and retry combinators

Problem. A search box sends a query to a flaky backend. Keystrokes arrive in bursts, so only the last value of a burst should be sent (debounce: emit an item only after the input has been quiet for a set period). Each backend call must finish within a deadline (timeout), and failed calls are repeated with growing delays (retry with exponential backoff).

Requirements

timeout(d, fut) returns Result<F::Output, Elapsed>; written as a struct that implements Future.
Debounce<S, T> implements Stream; it emits the latest item after quiet without a new item, and flushes a pending item at once when the input ends.
retry(max_attempts, base, op) calls op(attempt) up to max_attempts times, sleeping base * 2^(attempt - 1) after each failure.
Runs on a current-thread runtime with start_paused = true, and prints the paused-clock time of each event.

Stage 1: the timeout future

The struct holds the inner future and a Sleep, both pinned. poll tries the inner future first, so a future that is ready at the deadline still wins.

Rust

use pin_project_lite::pin_project;
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll, ready};
use std::time::Duration;
use tokio::time::{Instant, Sleep, sleep};

#[derive(Debug)]
struct Elapsed;

pin_project! {
    struct Timeout<F> {
        #[pin]
        fut: F,
        #[pin]
        delay: Sleep,
    }
}

fn timeout<F: Future>(d: Duration, fut: F) -> Timeout<F> {
    Timeout { fut, delay: sleep(d) }
}

impl<F: Future> Future for Timeout<F> {
    type Output = Result<F::Output, Elapsed>;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let this = self.project();
        if let Poll::Ready(v) = this.fut.poll(cx) {
            return Poll::Ready(Ok(v));
        }
        ready!(this.delay.poll(cx));
        Poll::Ready(Err(Elapsed))
    }
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let start = Instant::now();
    let ms = move || start.elapsed().as_millis();

    let fast = async { sleep(Duration::from_millis(30)).await; "fast" };
    println!("timeout fast: {:?} at {}ms", timeout(Duration::from_millis(100), fast).await, ms());

    let slow = async { sleep(Duration::from_millis(300)).await; "slow" };
    println!("timeout slow: {:?} at {}ms", timeout(Duration::from_millis(100), slow).await, ms());
}

Output

timeout fast: Ok("fast") at 30ms
timeout slow: Err(Elapsed) at 130ms

The second call starts at 30 ms and gives up at 130 ms. Returning Err drops the Timeout, which drops the inner future and its 300 ms Sleep, removing that timer from the wheel.

Stage 2: debounce and retry (full program)

Debounce drains every item that is ready, keeps the newest, and resets its Sleep to now + quiet for each one. Sleep::reset moves the existing timer entry instead of creating a new one. Retry is an async fn: it needs no pinned fields, so a hand-written state machine gains nothing here.

Rust

use pin_project_lite::pin_project;
use std::future::Future;
use std::pin::{Pin, pin};
use std::task::{Context, Poll, ready};
use std::time::Duration;
use tokio::sync::mpsc;
use tokio::time::{Instant, Sleep, sleep};
use tokio_stream::wrappers::ReceiverStream;
use tokio_stream::{Stream, StreamExt};

#[derive(Debug)]
struct Elapsed;

pin_project! {
    struct Timeout<F> {
        #[pin]
        fut: F,
        #[pin]
        delay: Sleep,
    }
}

fn timeout<F: Future>(d: Duration, fut: F) -> Timeout<F> {
    Timeout { fut, delay: sleep(d) }
}

impl<F: Future> Future for Timeout<F> {
    type Output = Result<F::Output, Elapsed>;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let this = self.project();
        if let Poll::Ready(v) = this.fut.poll(cx) {
            return Poll::Ready(Ok(v));
        }
        ready!(this.delay.poll(cx));
        Poll::Ready(Err(Elapsed))
    }
}

pin_project! {
    struct Debounce<S, T> {
        #[pin]
        inner: S,
        #[pin]
        delay: Sleep,
        pending: Option<T>,
        quiet: Duration,
        inner_done: bool,
    }
}

impl<S: Stream<Item = T>, T> Debounce<S, T> {
    fn new(inner: S, quiet: Duration) -> Self {
        Debounce { inner, delay: sleep(quiet), pending: None, quiet, inner_done: false }
    }
}

impl<S: Stream<Item = T>, T> Stream for Debounce<S, T> {
    type Item = T;

    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<T>> {
        let mut this = self.project();
        while !*this.inner_done {
            match this.inner.as_mut().poll_next(cx) {
                Poll::Ready(Some(item)) => {
                    *this.pending = Some(item);
                    let deadline = Instant::now() + *this.quiet;
                    this.delay.as_mut().reset(deadline);
                }
                Poll::Ready(None) => *this.inner_done = true,
                Poll::Pending => break,
            }
        }
        if this.pending.is_none() {
            return if *this.inner_done { Poll::Ready(None) } else { Poll::Pending };
        }
        if !*this.inner_done {
            ready!(this.delay.poll(cx));
        }
        Poll::Ready(this.pending.take())
    }
}

async fn retry<F, Fut, T>(max_attempts: u32, base: Duration, mut op: F) -> Result<T, String>
where
    F: FnMut(u32) -> Fut,
    Fut: Future<Output = Result<T, String>>,
{
    let mut attempt = 1;
    loop {
        match op(attempt).await {
            Ok(v) => return Ok(v),
            Err(e) if attempt == max_attempts => return Err(e),
            Err(e) => {
                let backoff = base * 2u32.pow(attempt - 1);
                println!("  attempt {attempt} failed: {e}; retry in {}ms", backoff.as_millis());
                sleep(backoff).await;
                attempt += 1;
            }
        }
    }
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let start = Instant::now();
    let ms = move || start.elapsed().as_millis();

    // 1. timeout
    let fast = async { sleep(Duration::from_millis(30)).await; "fast" };
    println!("timeout fast: {:?} at {}ms", timeout(Duration::from_millis(100), fast).await, ms());
    let slow = async { sleep(Duration::from_millis(300)).await; "slow" };
    println!("timeout slow: {:?} at {}ms", timeout(Duration::from_millis(100), slow).await, ms());

    // 2. retry, each attempt limited to 50 ms
    let result = retry(3, Duration::from_millis(100), |attempt| async move {
        println!("retry: attempt {attempt} at {}ms", ms());
        let call = async move {
            match attempt {
                1 => { sleep(Duration::from_millis(80)).await; Ok("data") }
                2 => Err("refused".to_string()),
                _ => { sleep(Duration::from_millis(10)).await; Ok("data") }
            }
        };
        match timeout(Duration::from_millis(50), call).await {
            Ok(r) => r,
            Err(Elapsed) => Err("timed out".to_string()),
        }
    }).await;
    println!("retry result: {result:?} at {}ms", ms());

    // 3. debounce keystrokes with a 200 ms quiet period
    let (tx, rx) = mpsc::channel(8);
    tokio::spawn(async move {
        for (text, pause) in [("h", 50), ("he", 50), ("hel", 400), ("hell", 40), ("hello", 300)] {
            tx.send(text).await.unwrap();
            sleep(Duration::from_millis(pause)).await;
        }
    });
    let mut typed = pin!(Debounce::new(ReceiverStream::new(rx), Duration::from_millis(200)));
    while let Some(text) = typed.next().await {
        println!("debounced: {text} at {}ms", ms());
    }
    println!("stream ended at {}ms", ms());
}

Output

timeout fast: Ok("fast") at 30ms
timeout slow: Err(Elapsed) at 130ms
retry: attempt 1 at 130ms
  attempt 1 failed: timed out; retry in 100ms
retry: attempt 2 at 280ms
  attempt 2 failed: refused; retry in 200ms
retry: attempt 3 at 480ms
retry result: Ok("data") at 490ms
debounced: hel at 790ms
debounced: hello at 1230ms
stream ended at 1330ms

Reading the timeline:

Retry starts at 130 ms. Attempt 1 needs 80 ms but is cut at 50 ms (180 ms), then waits 100 ms. Attempt 2 fails at once at 280 ms and waits 200 ms. Attempt 3 finishes 10 ms after 480 ms.
Keystrokes arrive at 490, 540, 590, 990, and 1030 ms. After hel at 590 there is a 400 ms gap, so hel is emitted at 590 + 200 = 790 ms. hello at 1030 is emitted at 1230 ms. The producer drops tx at 1330 ms, which ends the stream.
h and he are never emitted: each was replaced within 200 ms.

Part	What it does
`Timeout<F>`	polls the inner future, then the pinned `Sleep`; the inner future wins a tie
`Debounce::poll_next` loop	drains every ready item so a burst costs one wake, keeps the newest
`delay.as_mut().reset(...)`	moves the timer entry to the new deadline
`inner_done` flush	emits the pending item as soon as the input ends
`retry`	calls a closure per attempt, because a future can be polled to completion only once
paused clock	the runtime advances time to the next timer when every task is idle, so times are exact

Extensions

Add a max_wait to Debounce so a continuous stream of keystrokes still emits at least once per max_wait.
Make retry take a predicate that decides whether an error is retryable, and stop at once on a fatal error.
Write a Throttle<S> stream that emits at most one item per period and drops the rest, and test it with the paused clock.

Project 25: Scheduler experiment with runtime metrics

Problem. Confirm three scheduler facts from the source by measurement, using only stable metrics: where spawns go, what happens when a worker's local queue overflows, and how max_blocking_threads limits spawn_blocking.

Requirements

One multi-thread runtime with worker_threads(1) and max_blocking_threads(2). A single worker means no stealing, so counts are exact.
Report global_queue_depth() after spawning from main while the worker is busy, and after spawning 8 and 300 tasks from inside a task.
Run 6 blocking jobs and report the highest number that ran at the same time.
Report whether the worker has parked (worker_park_count).
Print counts only.

Expected counts, derived from the constants:

Case	Expected global queue depth	Reason
8 spawns from `main`	8	a non-worker thread always uses the global queue
8 spawns from a task	0	LIFO slot plus local queue, far below 256
300 spawns from a task	129	spawn 258 pushes the 257th task into a full 256-slot queue; 128 queued tasks plus that task move to the global queue

Stage 1: queue placement

Rust

use tokio::runtime::{Builder, Handle, Runtime};

fn spawns_from_main(rt: &Runtime) -> usize {
    let (started_tx, started_rx) = std::sync::mpsc::channel();
    let (release_tx, release_rx) = std::sync::mpsc::channel::<()>();
    // Occupy the only worker on purpose so the spawned tasks stay queued.
    let blocker = rt.spawn(async move {
        started_tx.send(()).unwrap();
        release_rx.recv().unwrap();
    });
    started_rx.recv().unwrap();

    let handles: Vec<_> = (0..8).map(|i| rt.spawn(async move { i })).collect();
    let depth = rt.handle().metrics().global_queue_depth();

    release_tx.send(()).unwrap();
    rt.block_on(async {
        blocker.await.unwrap();
        for h in handles {
            h.await.unwrap();
        }
    });
    depth
}

fn spawns_from_task(rt: &Runtime, n: usize) -> usize {
    rt.block_on(async move {
        tokio::spawn(async move {
            let handles: Vec<_> = (0..n).map(|i| tokio::spawn(async move { i })).collect();
            let depth = Handle::current().metrics().global_queue_depth();
            for h in handles {
                h.await.unwrap();
            }
            depth
        })
        .await
        .unwrap()
    })
}

fn main() {
    let rt = Builder::new_multi_thread().worker_threads(1).build().unwrap();
    println!("workers: {}", rt.handle().metrics().num_workers());
    println!("8 spawned from main while the worker is busy: global depth {}", spawns_from_main(&rt));
    println!("8 spawned from a task: global depth {}", spawns_from_task(&rt, 8));
    println!("300 spawned from a task: global depth {}", spawns_from_task(&rt, 300));
}

Output

workers: 1
8 spawned from main while the worker is busy: global depth 8
8 spawned from a task: global depth 0
300 spawned from a task: global depth 129

The parent in spawns_from_task runs on the worker, so each child goes through schedule_local: the new child takes the LIFO slot and the previous one moves to the local queue. The depth is read before the parent yields, so no child has run yet.

Stage 2: blocking pool and parking (full program)

Rust

use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Duration;
use tokio::runtime::{Builder, Handle, Runtime};

fn spawns_from_main(rt: &Runtime) -> usize {
    let (started_tx, started_rx) = std::sync::mpsc::channel();
    let (release_tx, release_rx) = std::sync::mpsc::channel::<()>();
    let blocker = rt.spawn(async move {
        started_tx.send(()).unwrap();
        release_rx.recv().unwrap();
    });
    started_rx.recv().unwrap();

    let handles: Vec<_> = (0..8).map(|i| rt.spawn(async move { i })).collect();
    let depth = rt.handle().metrics().global_queue_depth();

    release_tx.send(()).unwrap();
    rt.block_on(async {
        blocker.await.unwrap();
        for h in handles {
            h.await.unwrap();
        }
    });
    depth
}

fn spawns_from_task(rt: &Runtime, n: usize) -> usize {
    rt.block_on(async move {
        tokio::spawn(async move {
            let handles: Vec<_> = (0..n).map(|i| tokio::spawn(async move { i })).collect();
            let depth = Handle::current().metrics().global_queue_depth();
            for h in handles {
                h.await.unwrap();
            }
            depth
        })
        .await
        .unwrap()
    })
}

fn blocking_peak(rt: &Runtime, jobs: usize) -> usize {
    let running = Arc::new(AtomicUsize::new(0));
    let peak = Arc::new(AtomicUsize::new(0));
    rt.block_on(async {
        let mut handles = Vec::new();
        for _ in 0..jobs {
            let (running, peak) = (running.clone(), peak.clone());
            handles.push(tokio::task::spawn_blocking(move || {
                let now = running.fetch_add(1, Ordering::SeqCst) + 1;
                peak.fetch_max(now, Ordering::SeqCst);
                std::thread::sleep(Duration::from_millis(50));
                running.fetch_sub(1, Ordering::SeqCst);
            }));
        }
        for h in handles {
            h.await.unwrap();
        }
    });
    peak.load(Ordering::SeqCst)
}

fn main() {
    let rt = Builder::new_multi_thread()
        .worker_threads(1)
        .max_blocking_threads(2)
        .build()
        .unwrap();
    let metrics = rt.handle().metrics();

    println!("workers: {}", metrics.num_workers());
    println!("8 spawned from main while the worker is busy: global depth {}", spawns_from_main(&rt));
    println!("8 spawned from a task: global depth {}", spawns_from_task(&rt, 8));
    println!("300 spawned from a task: global depth {}", spawns_from_task(&rt, 300));
    println!("6 blocking jobs, max_blocking_threads 2: peak running {}", blocking_peak(&rt, 6));
    println!("worker 0 has parked: {}", metrics.worker_park_count(0) > 0);
    println!("global depth at the end: {}", metrics.global_queue_depth());
}

Output

workers: 1
8 spawned from main while the worker is busy: global depth 8
8 spawned from a task: global depth 0
300 spawned from a task: global depth 129
6 blocking jobs, max_blocking_threads 2: peak running 2
worker 0 has parked: true
global depth at the end: 0

The six jobs share two blocking threads; the other four wait in the blocking pool's own queue, which has no limit. Worker threads are not counted against max_blocking_threads. The worker parked while main waited for the blocking jobs, because it had no tasks.

Part	What it does
`worker_threads(1)`	removes stealing, so queue counts are exact
blocker task with a `std` channel	holds the worker so remote spawns stay in the global queue
`Handle::current().metrics()`	reads metrics from inside a task
`spawns_from_task(&rt, 300)`	overflows the 256-slot local queue once
`blocking_peak`	counts concurrent closures with `fetch_add` and `fetch_max`
`worker_park_count`	a stable per-worker counter; only its sign is printed

Extensions

Spawn 600 tasks from one task, predict the global queue depth from the overflow rule before running it, then compare.
Repeat Stage 1 on a current-thread runtime and explain why tasks spawned from main also enter the global queue there.
Add worker_total_busy_duration and print only whether it grew between two measurements taken before and after a CPU-bound task.

Interview points

A timeout combinator polls the inner future before the timer, so a future that is ready at the deadline is not reported as timed out.
Debounce resets one Sleep per item with Sleep::reset instead of allocating a new timer.
Retry takes a closure that builds a new future per attempt, because a completed future cannot be polled again.
Spawns from a non-worker thread go to the global queue; spawns from a worker go to its LIFO slot and local queue.
A full 256-slot local queue sends 128 tasks plus the new one to the global queue.
max_blocking_threads limits blocking threads only; extra blocking jobs wait in an unbounded queue.

Summary

Project 24 built Timeout, Debounce, and retry and checked their timing with the paused clock.
Project 25 measured queue placement, local-queue overflow (129 tasks), and the blocking-pool limit with global_queue_depth, num_workers, and worker_park_count.
Both programs print times from the paused clock or counts, never wall-clock durations.