Backpressure and Load Shedding

A service that accepts work faster than it finishes it builds a queue. Backpressure slows the producer down when the consumer cannot keep up. Load shedding rejects some work so the rest is served within its time limit. This lesson shows both, the arithmetic for sizing a limit, and what an unbounded queue does under overload.

Idea	Tokio tool
bounded queue	`mpsc::channel(capacity)`, `send().await` waits when full
shed by rejecting	`try_send`, `try_reserve`, `Semaphore::try_acquire_owned`
limit work in progress	`Semaphore` permits
time budget	`timeout`, `timeout_at`, a deadline passed to every call
rate limit	token bucket, `interval`
queue-length reasoning	Little's law, `L = λ · W`

All programs run on the paused clock (start_paused = true, explained in Time: sleep, timeout, and interval), so timings in the output are exact virtual milliseconds and repeat on every run. Queue sizes are printed as item counts, never as bytes.

Example 1: A bounded queue says "full"

A channel created with mpsc::channel(3) holds at most three items. try_send returns at once. When the queue is full it gives the item back inside TrySendError::Full. Receiver::len reports how many items are queued.

Rust

use tokio::sync::mpsc::{self, error::TrySendError};

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

    for item in 1..=5 {
        match tx.try_send(item) {
            Ok(()) => println!("item {item}: queued, queue length = {}", rx.len()),
            Err(TrySendError::Full(item)) => println!("item {item}: rejected, queue is full"),
            Err(TrySendError::Closed(item)) => println!("item {item}: rejected, channel closed"),
        }
    }
}

Output

The queue length never went above 3. The rejected items were handed back to the caller, which can drop them, retry later, or report an error. A queue with a fixed limit turns overload into a decision that the code has to make.

Example 2: send waits, and that is backpressure

send(item).await on a full bounded channel waits until the receiver takes an item. The producer is paused for as long as the consumer is behind. Here the consumer takes 10 ms per item and the queue holds 2 items. The program prints when the producer's send returns.

Rust

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

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let (tx, mut rx) = mpsc::channel::<u32>(2);
    let start = Instant::now();

    let consumer = tokio::spawn(async move {
        let mut handled = 0;
        while let Some(_item) = rx.recv().await {
            sleep(Duration::from_millis(10)).await;
            handled += 1;
        }
        handled
    });

    for item in 1..=6 {
        tx.send(item).await.unwrap();
        println!("item {item} accepted at {} ms", start.elapsed().as_millis());
    }
    drop(tx);
    println!("consumer handled {} items", consumer.await.unwrap());
}

Output

After the first burst, the producer's pace matches the consumer's: one new item every 10 ms. The producer did not need to know the consumer's speed. The full queue passed that information back through send.

Example 3: An unbounded queue grows without limit

unbounded_channel has no capacity, so send never waits and never fails while the receiver exists. If the producer is faster than the consumer, the queue grows for as long as the overload lasts. In this program the producer adds one item every 10 ms and the consumer needs 20 ms per item. A counter tracks the queue length, and a sampler prints it every 50 ms (offset by 5 ms so it never ties with another event).

Rust

use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Duration;
use tokio::sync::mpsc;
use tokio::time::{sleep, Instant};

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let (tx, mut rx) = mpsc::unbounded_channel::<(u32, Instant)>();
    let queued = Arc::new(AtomicUsize::new(0));
    let start = Instant::now();

    let producer = {
        let queued = queued.clone();
        tokio::spawn(async move {
            for item in 0..20 {
                queued.fetch_add(1, Ordering::SeqCst);
                tx.send((item, Instant::now())).unwrap();
                sleep(Duration::from_millis(10)).await;
            }
        })
    };

    let consumer = {
        let queued = queued.clone();
        tokio::spawn(async move {
            let mut waits = Vec::new();
            while let Some((item, enqueued_at)) = rx.recv().await {
                queued.fetch_sub(1, Ordering::SeqCst);
                waits.push((item, enqueued_at.elapsed().as_millis()));
                sleep(Duration::from_millis(20)).await;
            }
            waits
        })
    };

    sleep(Duration::from_millis(5)).await;
    for _ in 0..4 {
        println!(
            "at {:>3} ms: queue length = {}",
            start.elapsed().as_millis(),
            queued.load(Ordering::SeqCst)
        );
        sleep(Duration::from_millis(50)).await;
    }

    producer.await.unwrap();
    println!("producer finished at {} ms, queue length = {}", start.elapsed().as_millis(), queued.load(Ordering::SeqCst));

    let waits = consumer.await.unwrap();
    for (item, waited) in waits.iter().filter(|(item, _)| item % 5 == 0 || *item == 19) {
        println!("item {item} waited {waited} ms in the queue");
    }
}

Output

The arrival rate was 100 items per second and the service rate was 50 items per second, so the queue gained 50 items per second. Two things grew together: the queue length, and the time each item waited (the wait of item i was 10 * i ms). A queue that keeps growing has two costs. Memory use grows with the number of queued items. Every new item also waits behind all the earlier ones, so the latency of the whole service rises even though each item is processed quickly once it reaches the front. If the overload lasts, the process runs out of memory before the queue empties.

Example 4: The same load with a bounded queue

The same producer and consumer with mpsc::channel(4). The producer's send waits when the queue is full, so the queue length stays at or below 4 and the producer finishes later.

Rust

use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Duration;
use tokio::sync::mpsc;
use tokio::time::{sleep, Instant};

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let (tx, mut rx) = mpsc::channel::<(u32, Instant)>(4);
    let queued = Arc::new(AtomicUsize::new(0));
    let peak = Arc::new(AtomicUsize::new(0));
    let start = Instant::now();

    let producer = {
        let (queued, peak) = (queued.clone(), peak.clone());
        tokio::spawn(async move {
            for item in 0..20 {
                tx.send((item, Instant::now())).await.unwrap();
                let now = queued.fetch_add(1, Ordering::SeqCst) + 1;
                peak.fetch_max(now, Ordering::SeqCst);
                sleep(Duration::from_millis(10)).await;
            }
        })
    };

    let consumer = {
        let queued = queued.clone();
        tokio::spawn(async move {
            let mut longest_wait = 0;
            while let Some((_item, enqueued_at)) = rx.recv().await {
                queued.fetch_sub(1, Ordering::SeqCst);
                longest_wait = longest_wait.max(enqueued_at.elapsed().as_millis());
                sleep(Duration::from_millis(20)).await;
            }
            longest_wait
        })
    };

    producer.await.unwrap();
    println!("producer finished at {} ms", start.elapsed().as_millis());
    let longest_wait = consumer.await.unwrap();
    println!("peak queue length: {}", peak.load(Ordering::SeqCst));
    println!("longest wait in the queue: {longest_wait} ms");
}

Output

The peak stayed near the capacity of 4, and the longest wait stopped growing with the number of items. The producer took longer to send its 20 items, which is the cost: the delay moved from the queue to the producer. Whether that is acceptable depends on who the producer is. A producer that is a client waiting for a reply can wait. A producer that reads a network socket stops reading, and the sender's own buffers fill (the same effect one level down).

Little's law

Little's law relates three averages of a system in which items arrive, spend time inside, and leave:

Text

L = λ · W

Symbol	Meaning	Unit
`L`	average number of items inside the system (queued and being served)	items
`λ` (lambda)	average arrival rate, which equals the departure rate when the system is stable	items per second
`W`	average time an item spends in the system, from arrival to departure	seconds

The law holds for any arrival pattern and any service order, as long as the averages are taken over a period in which the system starts and ends empty, or in a stable long run. It gives the third quantity when two are known.

Example 5: Little's law computed by a program

Twenty requests arrive 10 ms apart. Each request stays in the system for 45 ms (it is served at once, in parallel). The program records the arrival time and departure time of every request. It computes W as the mean time in the system, λ as requests divided by the length of the observation window, and L independently by adding up how many requests were inside at each moment (the area under the "requests inside" curve, divided by the window length).

Rust

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

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let start = Instant::now();
    let events = Arc::new(Mutex::new(Vec::<(u128, i64)>::new()));
    let mut set = JoinSet::new();

    for i in 0..20u64 {
        let events = events.clone();
        set.spawn(async move {
            sleep(Duration::from_millis(10 * i)).await;
            let arrived = start.elapsed();
            events.lock().unwrap().push((arrived.as_micros(), 1));

            sleep(Duration::from_millis(45)).await;
            let left = start.elapsed();
            events.lock().unwrap().push((left.as_micros(), -1));

            left - arrived
        });
    }

    let mut total_time = Duration::ZERO;
    let mut requests = 0u32;
    while let Some(joined) = set.join_next().await {
        total_time += joined.unwrap();
        requests += 1;
    }

    // L: integrate the number of requests inside over the window.
    let mut events = events.lock().unwrap().clone();
    events.sort();
    let window_start = events.first().unwrap().0;
    let window_end = events.last().unwrap().0;
    let mut inside = 0i64;
    let mut last = window_start;
    let mut area = 0i128; // request-microseconds
    for (time, change) in events {
        area += i128::from(inside) * (time - last) as i128;
        inside += change;
        last = time;
    }

    let window = (window_end - window_start) as f64 / 1_000_000.0;
    let lambda = f64::from(requests) / window;
    let w = total_time.as_secs_f64() / f64::from(requests);
    let l = area as f64 / 1_000_000.0 / window;

    println!("requests:                     {requests}");
    println!("window:                       {:.0} ms", window * 1000.0);
    println!("lambda = requests / window:   {lambda:.3} per second");
    println!("W = mean time in system:      {:.1} ms", w * 1000.0);
    println!("lambda * W:                   {:.4}", lambda * w);
    println!("L measured from the events:   {l:.4}");
}

Output

λ · W and the separately measured L agree. Over the whole window the requests were in the system for 20 × 45 ms in total, which is the area under the curve, so the identity is exact here.

Steady-state reading of the same numbers: when a request arrives every 10 ms, λ is 100 per second. With W = 0.045 s, L = 100 × 0.045 = 4.5. On average 4.5 requests are inside the system at once. The rest of this lesson uses the law in two directions.

Question	Known	Compute
How many requests run at once?	`λ`, `W`	`L = λ · W`
How many workers or permits are needed?	`λ`, `W`	at least `⌈λ · W⌉`
What is the average latency at a given depth?	`L`, `λ`	`W = L / λ`
What arrival rate can be sustained?	permits `L`, `W`	`λ = L / W`

Example 6: Sizing a concurrency limit with Little's law

With λ = 100 requests per second and W = 45 ms, L = 4.5, so at least 5 requests must be able to run at once. The program computes that number, then runs 100 requests (one every 10 ms) through a Semaphore of 4 permits and through one of 5 permits, and reports how many requests were waiting for a permit.

Rust

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

async fn run(permits: usize) -> (usize, usize) {
    let semaphore = Arc::new(Semaphore::new(permits));
    let waiting = Arc::new(AtomicUsize::new(0));
    let peak_waiting = Arc::new(AtomicUsize::new(0));
    let mut set = JoinSet::new();

    for _ in 0..100 {
        let (semaphore, waiting, peak_waiting) = (semaphore.clone(), waiting.clone(), peak_waiting.clone());
        set.spawn(async move {
            let now = waiting.fetch_add(1, Ordering::SeqCst) + 1;
            peak_waiting.fetch_max(now, Ordering::SeqCst);
            let permit = semaphore.acquire_owned().await.unwrap();
            waiting.fetch_sub(1, Ordering::SeqCst);
            sleep(Duration::from_millis(45)).await;
            drop(permit);
        });
        sleep(Duration::from_millis(10)).await;
    }
    let waiting_at_last_arrival = waiting.load(Ordering::SeqCst);
    while set.join_next().await.is_some() {}
    (waiting_at_last_arrival, peak_waiting.load(Ordering::SeqCst))
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let lambda = 1000.0 / 10.0; // arrivals per second
    let w = 0.045; // seconds per request
    let l: f64 = lambda * w;
    println!("lambda = {lambda} per second, W = {w} s, L = lambda * W = {l}");
    println!("permits needed: at least {}", l.ceil());

    for permits in [4, 5] {
        let (at_end, peak) = run(permits).await;
        println!("{permits} permits: {at_end} requests waiting after the last arrival, peak waiting = {peak}");
    }
}

Output

With 5 permits, no request waited. With 4 permits, capacity was 4 / 0.045 ≈ 88.9 requests per second, less than the 100 per second arriving, so the waiting line grew for as long as the arrivals continued. A limit below λ · W cannot keep up, however long the queue is allowed to be.

Example 7: Admission control with try_acquire

Admission control decides at the door whether to accept a request. A Semaphore with 3 permits allows 3 requests in progress. try_acquire_owned returns an error at once when no permit is free, and the request is shed instead of queued. Requests arrive every 10 ms and take 45 ms.

Rust

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

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let semaphore = Arc::new(Semaphore::new(3));
    let start = Instant::now();
    let mut set = JoinSet::new();
    let (mut accepted, mut shed) = (0, 0);

    for id in 0..10 {
        let at = start.elapsed().as_millis();
        match semaphore.clone().try_acquire_owned() {
            Ok(permit) => {
                accepted += 1;
                println!("request {id} at {at:>3} ms: accepted");
                set.spawn(async move {
                    sleep(Duration::from_millis(45)).await;
                    drop(permit);
                });
            }
            Err(_) => {
                shed += 1;
                println!("request {id} at {at:>3} ms: rejected (overloaded)");
            }
        }
        sleep(Duration::from_millis(10)).await;
    }
    while set.join_next().await.is_some() {}
    println!("accepted = {accepted}, shed = {shed}");
}

Output

Requests 3 and 4 found all three permits taken and were rejected without waiting. Request 5 arrived after the first request finished and was accepted. Rejected requests cost almost nothing, and the accepted requests ran at their normal speed. A server usually turns the rejection into an "overloaded" reply (HTTP status 503) so the client can retry later or elsewhere.

Example 8: Dropping the newest or the oldest

When a queue is full, the item to drop is a choice. try_send drops the newest item (the one that just arrived). A queue that drops the oldest item keeps the most recent data, which suits values where old ones are stale, such as sensor readings. The second queue below is a VecDeque with a manual size check.

Rust

use std::collections::VecDeque;
use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    // Drop the newest.
    let (tx, mut rx) = mpsc::channel::<u32>(4);
    let mut dropped = 0;
    for item in 0..10 {
        if tx.try_send(item).is_err() {
            dropped += 1;
        }
    }
    drop(tx);
    let mut kept = Vec::new();
    while let Some(item) = rx.recv().await {
        kept.push(item);
    }
    println!("drop newest: kept {kept:?}, dropped {dropped}");

    // Drop the oldest.
    let mut queue = VecDeque::new();
    let mut dropped = 0;
    for item in 0..10 {
        if queue.len() == 4 {
            queue.pop_front();
            dropped += 1;
        }
        queue.push_back(item);
    }
    println!("drop oldest: kept {queue:?}, dropped {dropped}");
}

Output

Both queues hold 4 items and dropped 6. They kept different items.

Policy	Keeps	Fits
block the producer (`send().await`)	everything, in order	work that cannot be lost and a producer that can wait
drop the newest (`try_send`)	the oldest items	requests where a client retry is cheap, fairness to work already accepted
drop the oldest	the newest items	data where only the latest value matters (readings, screen updates)
unbounded buffer	everything	short bursts of a known maximum size, otherwise unsafe under overload

Example 9: Reserve a slot before doing the work

try_reserve claims one slot of capacity without sending anything yet. The caller does the expensive work only after the slot is secured, so rejected requests skip that work entirely. Permit::send then puts the value in the reserved slot. A dropped permit gives the slot back. mpsc Channels covers reserve.

Rust

use tokio::sync::mpsc;

fn expensive_prepare(id: u32) -> String {
    println!("  preparing request {id}");
    format!("payload-{id}")
}

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

    for id in 1..=3 {
        match tx.try_reserve() {
            Ok(permit) => {
                let payload = expensive_prepare(id);
                permit.send(payload);
                println!("request {id}: queued");
            }
            Err(_) => println!("request {id}: shed before any work was done"),
        }
        if id == 2 {
            println!("consumer takes {:?}", rx.recv().await);
        }
    }
}

Output

Request 2 was shed without calling expensive_prepare. After the consumer took the first item, request 3 could reserve a slot again.

Example 10: A timeout on waiting in the queue

A time budget bounds how long a request may wait. A timeout around the wait for a permit turns "wait indefinitely" into "wait at most 30 ms, then fail". Four requests arrive at once at a service that handles one request at a time in 20 ms.

Rust

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

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let semaphore = Arc::new(Semaphore::new(1));
    let mut handles = Vec::new();

    for id in 0..4 {
        let semaphore = semaphore.clone();
        handles.push(tokio::spawn(async move {
            let permit = match timeout(Duration::from_millis(30), semaphore.acquire_owned()).await {
                Ok(permit) => permit.unwrap(),
                Err(_) => return format!("request {id}: gave up after waiting 30 ms"),
            };
            sleep(Duration::from_millis(20)).await;
            drop(permit);
            format!("request {id}: done")
        }));
    }

    for handle in handles {
        println!("{}", handle.await.unwrap());
    }
}

Output

Request 0 ran from 0 to 20 ms and request 1 from 20 to 40 ms. Requests 2 and 3 would have started at 40 and 60 ms, later than their 30 ms budget, so they gave up. Waiting longer would have produced answers the client had already stopped waiting for. A timed-out request also frees its place in the queue for requests that can still succeed.

Example 11: A deadline passed through every call

Per-call timeouts add up. If each of three stages allows 100 ms, a request can take 300 ms. A deadline is one fixed point in time for the whole request. Each stage receives the deadline, and timeout_at(deadline, future) fails the stage when that point passes. The program compares a 100 ms deadline shared by three stages with a separate 100 ms timeout for each stage.

Rust

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

async fn stage_with_deadline(name: &str, work_ms: u64, deadline: Instant) -> Result<(), String> {
    let left = deadline.saturating_duration_since(Instant::now());
    println!("  {name}: starts with {} ms left", left.as_millis());
    timeout_at(deadline, sleep(Duration::from_millis(work_ms)))
        .await
        .map_err(|_| format!("{name}: deadline reached"))
}

async fn stage_with_own_timeout(name: &str, work_ms: u64) -> Result<(), String> {
    timeout(Duration::from_millis(100), sleep(Duration::from_millis(work_ms)))
        .await
        .map_err(|_| format!("{name}: timed out"))
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let works = [("parse", 40), ("lookup", 35), ("render", 50)];

    println!("one deadline of 100 ms:");
    let start = Instant::now();
    let deadline = start + Duration::from_millis(100);
    let mut result = Ok(());
    for (name, ms) in works {
        result = stage_with_deadline(name, ms, deadline).await;
        if result.is_err() {
            break;
        }
    }
    println!("  result: {result:?} after {} ms", start.elapsed().as_millis());

    println!("a 100 ms timeout per stage:");
    let start = Instant::now();
    let mut result = Ok(());
    for (name, ms) in works {
        result = stage_with_own_timeout(name, ms).await;
        if result.is_err() {
            break;
        }
    }
    println!("  result: {result:?} after {} ms", start.elapsed().as_millis());
}

Output

With the shared deadline, the last stage started with 25 ms left, needed 50 ms, and was cut off at exactly 100 ms. With separate timeouts, every stage passed and the request took 125 ms, longer than the caller's budget. Across a network, the deadline is sent to the next service as the remaining time, so the next service does not start work that the caller will have abandoned.

Example 12: A token bucket rate limiter

A rate limiter caps how often an action may happen. A token bucket holds up to capacity tokens. Each allowed action removes one token. One token is added every interval, up to the capacity. Full buckets allow a burst of capacity actions. Afterwards the sustained rate is one action per interval. The Instant type here is Tokio's, so the paused clock controls it.

Rust

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

struct TokenBucket {
    capacity: u32,
    tokens: u32,
    interval: Duration,
    last_refill: Instant,
}

impl TokenBucket {
    fn new(capacity: u32, interval: Duration) -> Self {
        TokenBucket { capacity, tokens: capacity, interval, last_refill: Instant::now() }
    }

    fn try_acquire(&mut self) -> bool {
        let now = Instant::now();
        if self.tokens == self.capacity {
            // A full bucket does not bank time.
            self.last_refill = now;
        }
        let elapsed = now.duration_since(self.last_refill);
        let new_tokens = (elapsed.as_millis() / self.interval.as_millis()) as u32;
        if new_tokens > 0 {
            self.tokens = (self.tokens + new_tokens).min(self.capacity);
            self.last_refill += self.interval * new_tokens;
        }
        if self.tokens > 0 {
            self.tokens -= 1;
            true
        } else {
            false
        }
    }
}

fn burst(bucket: &mut TokenBucket, count: u32) -> String {
    (0..count).map(|_| if bucket.try_acquire() { 'Y' } else { 'N' }).collect()
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    // Capacity 3, one token every 100 ms (10 per second sustained).
    let mut bucket = TokenBucket::new(3, Duration::from_millis(100));

    println!("burst of 5 at 0 ms:          {}", burst(&mut bucket, 5));

    sleep(Duration::from_millis(250)).await;
    println!("burst of 4 at 250 ms:        {}", burst(&mut bucket, 4));

    sleep(Duration::from_secs(2)).await;
    println!("burst of 5 after 2 s idle:   {}", burst(&mut bucket, 5));

    let mut allowed = 0;
    for _ in 0..200 {
        sleep(Duration::from_millis(10)).await;
        if bucket.try_acquire() {
            allowed += 1;
        }
    }
    println!("200 requests over 2 s, one every 10 ms: {allowed} allowed");
}

Output

The first burst used the 3 stored tokens (Y allowed, N denied). After 250 ms, two tokens had been added. After a long idle time the bucket refilled to its capacity of 3 and no further, so an idle period does not allow a larger burst later. Under steady demand of 100 requests per second, the limiter let through about 10 per second, its sustained rate. The last try_acquire timing detail (allowed count) comes from the exact refill boundaries, which the paused clock makes repeatable.

try_acquire sheds excess requests. A limiter can instead delay them.

Example 13: Delaying instead of rejecting with interval

interval(period) yields a tick every period. Waiting for a tick before each job spaces the jobs out. This is shaping: the work is delayed, not dropped. The first tick completes at once.

Rust

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

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let start = Instant::now();
    let mut ticks = interval(Duration::from_millis(100));

    for job in 1..=4 {
        ticks.tick().await;
        println!("job {job} started at {} ms", start.elapsed().as_millis());
    }
}

Output

Shaping keeps every request and adds delay, so it needs a bound on how many requests may wait. Without that bound it is the unbounded queue of Example 3 again.

Example 14: Backpressure through a pipeline

Bounded channels between stages pass the slowness of the last stage back to the first. The pipeline has a producer, a middle stage, and a consumer that takes 10 ms per item. Each channel holds 2 items. The producer sends as fast as it is allowed. At 105 ms the program counts how many items were produced and consumed.

Rust

use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Duration;
use tokio::sync::mpsc;
use tokio::time::sleep;

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let produced = Arc::new(AtomicUsize::new(0));
    let consumed = Arc::new(AtomicUsize::new(0));
    let (tx1, mut rx1) = mpsc::channel::<u64>(2);
    let (tx2, mut rx2) = mpsc::channel::<u64>(2);

    let producer = {
        let produced = produced.clone();
        tokio::spawn(async move {
            for item in 0.. {
                if tx1.send(item).await.is_err() {
                    break;
                }
                produced.fetch_add(1, Ordering::SeqCst);
            }
        })
    };
    let middle = tokio::spawn(async move {
        while let Some(item) = rx1.recv().await {
            if tx2.send(item * 2).await.is_err() {
                break;
            }
        }
    });
    let consumer = {
        let consumed = consumed.clone();
        tokio::spawn(async move {
            while let Some(_item) = rx2.recv().await {
                sleep(Duration::from_millis(10)).await;
                consumed.fetch_add(1, Ordering::SeqCst);
            }
        })
    };

    sleep(Duration::from_millis(105)).await;
    let produced_count = produced.load(Ordering::SeqCst);
    let consumed_count = consumed.load(Ordering::SeqCst);
    println!("produced = {produced_count}, consumed = {consumed_count}");
    println!("items inside the pipeline = {}", produced_count - consumed_count);

    producer.abort();
    middle.abort();
    consumer.abort();
}

Output

The producer would generate items without end, yet only a small, fixed number of items were inside the pipeline: the two channel buffers, one item held by the middle stage while it waits to send, and one item being processed by the consumer. Without the bounds, the producer would have produced items continuously, at whatever speed the CPU allows, and the difference between produced and consumed would keep growing.

Example 15: Bounding latency with a limit on items in the system

Little's law also works from the queue side. If the system never holds more than L items, and it completes λ items per second, the time in the system is at most L / λ. The program uses two semaphores. The first admits at most 10 items (queue plus workers). The second is the 2 worker slots. Each item needs 10 ms of a worker. Thirty items arrive at once.

Rust

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

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let system_limit = 10; // items allowed inside: waiting plus in service
    let worker_count = 2;
    let service_ms = 10.0;

    let admit = Arc::new(Semaphore::new(system_limit));
    let workers = Arc::new(Semaphore::new(worker_count));
    let start = Instant::now();
    let mut set = JoinSet::new();
    let mut shed = 0;

    for _ in 0..30 {
        match admit.clone().try_acquire_owned() {
            Ok(admitted) => {
                let workers = workers.clone();
                set.spawn(async move {
                    let worker = workers.acquire_owned().await.unwrap();
                    sleep(Duration::from_millis(10)).await;
                    drop(worker);
                    drop(admitted);
                    start.elapsed().as_millis()
                });
            }
            Err(_) => shed += 1,
        }
    }

    let mut latencies = Vec::new();
    while let Some(joined) = set.join_next().await {
        latencies.push(joined.unwrap());
    }
    latencies.sort();

    let throughput_per_ms = worker_count as f64 / service_ms;
    let predicted = system_limit as f64 / throughput_per_ms;
    println!("admitted = {}, shed = {shed}", latencies.len());
    println!("throughput = {:.0} items per second", throughput_per_ms * 1000.0);
    println!("predicted time to clear a full system: L / throughput = {predicted:.0} ms");
    println!("measured completion times (ms): {latencies:?}");
}

Output

The last admitted item finished at the predicted 50 ms. The 20 shed items were rejected at once and never affected the latency of the 10 admitted ones. Setting the admission limit is therefore also setting the worst latency: worst latency ≈ limit / throughput. To meet a latency target of 100 ms at this throughput, the limit would be 0.1 s × 200 per second = 20. This arithmetic assumes the service time stays at 10 ms. A slower dependency raises W, so the same limit gives a longer worst latency, and monitoring must watch W as well as L.

Sizing rules

Question	Rule
How many concurrent requests does the service need to support?	`λ · W`, rounded up
How large may the queue be?	`latency target × throughput − requests in service`
What should happen at the limit?	reject quickly (shed), or wait with a time budget
Where should the limit sit?	at the entry point, before any expensive work
What number should be watched in production?	queue length and time in queue, not only the average latency

Interview points

Backpressure slows the producer when the consumer is behind. A bounded channel implements it: send().await waits when the queue is full.
Load shedding rejects work to protect the work already accepted. try_send, try_reserve, and try_acquire return an error instead of waiting.
An unbounded queue under sustained overload grows in length and in waiting time until memory runs out.
Little's law: L = λ · W. L is the average number of items in the system, λ the arrival rate in items per second, W the average time in the system in seconds.
A concurrency limit below λ · W cannot keep up, and a queue in front of it grows.
A limit of L items in the system with throughput λ bounds the time in the system at about L / λ.
Timeouts bound the wait for one step. A deadline bounds the whole request, and each step gets the remaining time.
A token bucket allows a burst up to its capacity and a sustained rate of one token per interval. A full bucket does not bank extra time.
Drop the newest to keep old work in order, drop the oldest to keep fresh data, or block when nothing may be lost.
try_reserve claims capacity before the expensive work, so rejected requests cost almost nothing.

Summary

Bounded queues turn overload into waiting (send().await) or an error (try_send).
Unbounded queues never push back; their length and delay grow with the overload.
Admission control with Semaphore::try_acquire_owned limits work in progress and sheds the rest.
L = λ · W gives the number of concurrent requests to support, and L / λ gives the latency bound for a system limit.
timeout bounds one wait; timeout_at with a shared deadline bounds a whole request.
A token bucket limits rate with a burst allowance; interval shapes work by delaying it.
In a pipeline of bounded channels, the slowest stage sets the pace of all earlier stages.
Shedding at the entry point is cheaper than failing deep inside the system.