Tokio Backpressure: Bounded Channels and Semaphore-Limited HTTP Fan-Out

What you will build

One Cargo project, five programs, one idea: pressure must have somewhere to go, and the program — not the operating system — decides where.

| Program | What it proves | | src/main.rs | The two semaphore placements, side by side, 50 tasks, 5 permits | | examples/01-bounded-channel.rs | A full bounded channel parks the sender; try_send refuses at once; reserve() holds a slot | | examples/03-head-of-line.rs | Fairness cuts both ways: a big request at the front blocks a small one behind it | | examples/04-cancel-close.rs | Cancelling an acquire loses queue place; close() ends a semaphore; the limits | | examples/05-http-fanout.rs | Real HTTP fan-out: reqwest against a local test server, at most 5 requests in flight |

The Cargo.toml:

[package]
name = "tokio-backpressure"
version = "0.1.0"
edition = "2021"

[dependencies]
tokio = { version = "1.53.2", features = ["rt-multi-thread", "macros", "sync", "time", "net"] }
reqwest = { version = "0.12.28", default-features = false, features = ["rustls-tls"] }

Tested with cargo 1.99.1, rustc 1.99.1, Tokio 1.53.2, and reqwest 0.12.28 on x86_64 Linux (WSL). Every command and code listing on this page was compiled and run on those versions, and every error quoted below is a real one. MSVC is not applicable here; no other Rust toolchain was tested.

The toolchain reality first

Two facts about the build machine shape everything below. First, the Rust toolchain is rustup-managed and is not on PATH in a shell with no terminal attached: a bare which cargo says command not found on a healthy toolchain. Export the cargo bin directory before concluding anything is missing:

export PATH="$HOME/.cargo/bin:$PATH"
cargo --version   # cargo 1.99.1 (5f94df478 2026-08-27)
rustc --version   # rustc 1.99.1 (58cf8f9c3 2026-10-13)

Second, the project builds offline. The registry cache on the build machine holds Tokio 1.53.2 (and 1.50.0, 1.52.3, 1.53.0) and reqwest 0.12.28, so the pin follows the cache, not the newest release on crates.io:

cargo build --offline
cargo run --offline --example 01-bounded-channel

That has one consequence worth stating plainly: adding a dependency the cache does not hold fails an offline build. If you follow along on your own machine with network access, plain cargo build resolves the same versions from crates.io. The Tokio API documentation this guide cites is the docs.rs page for Tokio 1.53.2 — the version actually built — not the 1.48.0 pages an earlier search turned up.

The bounded channel: the sender parks

A bounded channel is the first backpressure valve. mpsc::channel(2) buffers up to two messages; once the buffer is full, send().await waits until a message is received. The module documentation states it directly: "if this limit is reached, trying to send another message will wait until a message is received from the channel. In other words, the channel provides backpressure." Values arrive in the order they were sent, and there is only one receiver. The unbounded variant, unbounded_channel, has no such limit — send always completes immediately, and memory grows without bound if the producer outruns the consumer. That is the tradeoff the bounded channel exists to avoid.

examples/01-bounded-channel.rs makes the park observable with a clock:

// 01-bounded-channel — a full bounded channel parks the sender.
// Run: cargo run --offline --example 01-bounded-channel
use std::sync::{
    atomic::{AtomicBool, Ordering},
    Arc,
};
use std::time::Instant;
use tokio::sync::mpsc;
use tokio::time::{sleep, Duration};

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

    // Two sends fill the buffer of 2; both complete immediately.
    tx.send(0).await.unwrap();
    tx.send(1).await.unwrap();
    println!(
        "sent 0, 1 at {:?}; capacity left: {}",
        start.elapsed(),
        tx.capacity()
    );

    // The buffer is full, so try_send is refused at once...
    match tx.try_send(99) {
        Err(tokio::sync::mpsc::error::TrySendError::Full(v)) => {
            println!("try_send({v}) refused: channel full at {:?}", start.elapsed());
        }
        other => panic!("unexpected try_send outcome: {other:?}"),
    }

    // ...and send().await parks until the consumer takes a value.
    let parked = Arc::new(AtomicBool::new(false));
    let parked2 = parked.clone();
    let tx2 = tx.clone();
    let sender = tokio::spawn(async move {
        tx2.send(2).await.unwrap();
        parked2.store(true, Ordering::SeqCst);
        println!("send 2 completed at {:?}", start.elapsed());
    });
    sleep(Duration::from_millis(200)).await;
    println!(
        "after 200ms with no consumer, send 2 parked: {}",
        !parked.load(Ordering::SeqCst)
    );
    assert!(!parked.load(Ordering::SeqCst));

    let v = rx.recv().await.unwrap();
    println!("consumer received {v} at {:?}", start.elapsed());
    sender.await.unwrap();
    assert!(parked.load(Ordering::SeqCst));

    // Drain what is left, then reserve() holds a slot before the value exists.
    assert_eq!(rx.recv().await, Some(1));
    assert_eq!(rx.recv().await, Some(2));
    let permit = tx.reserve().await.unwrap();
    println!("reserved a slot; capacity left: {}", tx.capacity());
    permit.send(3);
    assert_eq!(rx.recv().await, Some(3));
    println!("bounded-channel OK");
}

The output:

sent 0, 1 at 5.113µs; capacity left: 0
try_send(99) refused: channel full at 31.138µs
after 200ms with no consumer, send 2 parked: true
consumer received 0 at 201.104337ms
send 2 completed at 201.173237ms
reserved a slot; capacity left: 1
bounded-channel OK

Three observations, all measured rather than asserted. The first two sends complete in microseconds and leave zero capacity. The third send is still parked after 200 milliseconds with no consumer, and completes within a tenth of a millisecond of the consumer taking a value. And reserve() is the escape hatch for the case where the value does not exist yet: it waits for a slot, then hands back a permit that guarantees capacity, so the message can be built after the wait instead of before it.

Two shutdown details from the same module page, stated here as documented behavior rather than tested code: when all senders are dropped, recv returns None once the buffer drains — that is the channel's termination event. And a clean shutdown calls Receiver::close first, which stops new sends, then drains the buffered values before dropping the receiver; dropping the receiver outright discards unread messages.

The two placements: where you acquire decides what is bounded

This is the spine of the guide, and it comes straight from the Tokio Semaphore documentation, which shows both placements as separate examples. One warning about that page before using it: it disagrees with itself. The prose for the outgoing-requests example says the semaphore has 10 permits; the code underneath creates 5 (Semaphore::new(5)). This guide uses 5, the code's figure, everywhere.

Placement A — acquire inside each spawned task. Spawn all 50 tasks, and let each one acquire before it works. This bounds concurrent requests only. All 50 tasks exist, waiting.

Placement B — acquire before spawning. Acquire an owned permit first, then spawn the task holding it. The spawning loop itself blocks on the sixth task until one of the first five finishes, so at most 5 tasks exist at any moment.

src/main.rs runs both with an atomic in-flight counter — increment before the work, decrement after, and record the maximum — so the bound is measured, not assumed:

// tokio-backpressure — main binary: the two semaphore placements, side by side.
// Placement A: acquire INSIDE each spawned task — bounds concurrent work only.
// Placement B: acquire_owned BEFORE spawn — bounds the number of spawned tasks.
// (Extended from the 2026-11-05 probe: 6 tasks / 2 permits -> 50 tasks / 5 permits.)
use std::sync::{
    atomic::{AtomicUsize, Ordering},
    Arc,
};
use tokio::sync::{mpsc, Semaphore};
use tokio::time::{sleep, Duration};

static IN_FLIGHT: AtomicUsize = AtomicUsize::new(0);
static MAX_IN_FLIGHT: AtomicUsize = AtomicUsize::new(0);
static STARTED: AtomicUsize = AtomicUsize::new(0);

fn record_start() {
    let n = IN_FLIGHT.fetch_add(1, Ordering::SeqCst) + 1;
    MAX_IN_FLIGHT.fetch_max(n, Ordering::SeqCst);
}

#[tokio::main]
async fn main() {
    // Placement A: 50 tasks, semaphore of 5, acquire inside the task.
    let sem = Arc::new(Semaphore::new(5));
    let (tx, mut rx) = mpsc::channel::<u32>(8);
    for i in 0..50u32 {
        let sem = sem.clone();
        let tx = tx.clone();
        tokio::spawn(async move {
            STARTED.fetch_add(1, Ordering::SeqCst);
            let _permit = sem.acquire().await.expect("semaphore open");
            record_start();
            sleep(Duration::from_millis(20)).await;
            IN_FLIGHT.fetch_sub(1, Ordering::SeqCst);
            tx.send(i).await.expect("receiver alive");
        });
    }
    drop(tx);
    // All 50 tasks exist almost at once, but only 5 ever work at once:
    // STARTED counts tasks that began (before acquiring), IN_FLIGHT
    // counts tasks holding a permit.
    sleep(Duration::from_millis(10)).await;
    println!(
        "placement-A after 10ms: started={} in_flight={} (50 tasks exist, 5 work)",
        STARTED.load(Ordering::SeqCst),
        IN_FLIGHT.load(Ordering::SeqCst)
    );
    assert_eq!(STARTED.load(Ordering::SeqCst), 50);
    let mut sum = 0;
    while let Some(v) = rx.recv().await {
        sum += v;
    }
    println!(
        "placement-A: received 50 values, sum={} max_in_flight={} (semaphore permits=5)",
        sum,
        MAX_IN_FLIGHT.load(Ordering::SeqCst)
    );
    assert_eq!(sum, (0..50u32).sum::<u32>());
    assert_eq!(MAX_IN_FLIGHT.load(Ordering::SeqCst), 5);

    // Placement B: acquire_owned BEFORE spawn — the loop itself blocks on
    // the 6th task until one of the first 5 finishes, so at most 5 tasks
    // exist at any moment.
    IN_FLIGHT.store(0, Ordering::SeqCst);
    MAX_IN_FLIGHT.store(0, Ordering::SeqCst);
    let sem_b = Arc::new(Semaphore::new(5));
    let spawned_b = Arc::new(AtomicUsize::new(0));
    let spawned_b2 = spawned_b.clone();
    let producer = tokio::spawn(async move {
        let mut handles = Vec::new();
        for i in 0..50u32 {
            let permit = sem_b.clone().acquire_owned().await.expect("permit");
            spawned_b2.fetch_add(1, Ordering::SeqCst);
            handles.push(tokio::spawn(async move {
                record_start();
                sleep(Duration::from_millis(20)).await;
                IN_FLIGHT.fetch_sub(1, Ordering::SeqCst);
                drop(permit);
                i
            }));
        }
        handles
    });
    sleep(Duration::from_millis(10)).await;
    println!(
        "placement-B after 10ms: spawned={} (permits=5, 6th spawn waits for a permit)",
        spawned_b.load(Ordering::SeqCst)
    );
    assert_eq!(spawned_b.load(Ordering::SeqCst), 5);
    let handles = producer.await.expect("producer");
    let mut done = 0;
    for h in handles {
        h.await.expect("task");
        done += 1;
    }
    println!(
        "placement-B: spawned 50 tasks one permit at a time, completed={} max_in_flight={}",
        done,
        MAX_IN_FLIGHT.load(Ordering::SeqCst)
    );
    assert_eq!(done, 50);
    assert_eq!(MAX_IN_FLIGHT.load(Ordering::SeqCst), 5);
    println!("placements OK");
}

The output:

placement-A after 10ms: started=50 in_flight=5 (50 tasks exist, 5 work)
placement-A: received 50 values, sum=1225 max_in_flight=5 (semaphore permits=5)
placement-B after 10ms: spawned=5 (permits=5, 6th spawn waits for a permit)
placement-B: spawned 50 tasks one permit at a time, completed=50 max_in_flight=5
placements OK

The contrast is in the two "after 10ms" lines. In placement A, all 50 tasks have started within 10 milliseconds and 5 hold permits; the other 45 are tasks waiting on the semaphore, each one holding a future, a stack, and a place in the queue. In placement B, only 5 tasks exist after 10 milliseconds, because the sixth acquire_owned cannot complete until one of the first five tasks drops its permit. Both placements end with the same totals — 50 values, sum 1225, maximum 5 in flight — but they bound different things, and only placement B bounds the number of tasks.

The first version of this program measured the wrong thing, and the mistake is worth keeping: the original counter incremented after acquiring the permit, so placement A reported spawned=5 after 10 milliseconds — true of tasks holding permits, false of tasks existing. The counter has to increment before the acquire, at the top of the spawned closure, to count tasks that exist. A counter placed after the thing it measures measures the wrong thing.

Why acquire_owned exists

acquire returns a SemaphorePermit, which borrows the semaphore. acquire_owned requires the semaphore wrapped in an Arc and returns an OwnedSemaphorePermit, which owns its share of that Arc and can be moved into a spawned task. The difference is not stylistic; the compiler enforces it. This program does not compile:

use tokio::sync::Semaphore;

#[tokio::main]
async fn main() {
    let sem = Semaphore::new(1);
    let permit = sem.acquire().await.unwrap();
    tokio::spawn(async move {
        drop(permit);
    })
    .await
    .unwrap();
}

The compiler says, verbatim:

error[E0597]: `sem` does not live long enough
   --> src/main.rs:6:18
    |
  5 |       let sem = Semaphore::new(1);
    |           --- binding `sem` declared here
  6 |       let permit = sem.acquire().await.unwrap();
    |                    ^^^ borrowed value does not live long enough
  7 | /     tokio::spawn(async move {
  8 | |         drop(permit);
  9 | |     })
    | |______- argument requires that `sem` is borrowed for `'static`
...
 12 |   }
    |   - `sem` dropped here while still borrowed
    |
note: requirement that the value outlives `'static` introduced here
   --> /home/droopy/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/tokio-1.53.2/src/task/spawn.rs:176:28
    |
176 |         F: Future + Send + 'static,
    |                            ^^^^^^^

tokio::spawn requires a 'static future, and a borrowed permit cannot satisfy that, because the semaphore it borrows might be dropped while the task is still running. acquire_owned is the fix: the permit carries an Arc to the semaphore, so the semaphore lives as long as the permit does. The same rule is why the docs' incoming-connections example acquires with acquire_owned before accepting a socket — the permit has to be movable into the spawned task. A borrowed permit is fine when the work stays in the same task; the moment the permit crosses a spawn boundary, it must be owned.

Both permit types also support forget, which drops the permit without returning it to the semaphore, permanently reducing the available count; split, which carves permits off a permit into a second one; and merge, which folds two permits from the same semaphore into one. Those are API surface from the permit documentation pages, not behavior this guide executed — the programs above never need them.

Fairness and its cost: head-of-line blocking

A Tokio semaphore is a fair permit queue: permits are handed out in the order they were requested. Fairness sounds unambiguously good until a large request reaches the front of the queue. The Semaphore documentation states the consequence plainly: if a call to acquire_many at the front of the queue requests more permits than are available, it can prevent a later acquire from completing, even when the semaphore has enough permits to complete that call. examples/03-head-of-line.rs demonstrates it rather than asserting it:

// 03-head-of-line — fairness cuts both ways: a big request at the front
// of the queue blocks a small request behind it, even when permits are free.
// Run: cargo run --offline --example 03-head-of-line
use std::sync::{atomic::{AtomicBool, Ordering}, Arc, Mutex};
use tokio::sync::Semaphore;
use tokio::time::{sleep, Duration};

#[tokio::main]
async fn main() {
    let sem = Arc::new(Semaphore::new(3));
    // Hold 1 of 3 permits, so only 2 are free.
    let held = sem.acquire().await.unwrap();
    assert_eq!(sem.available_permits(), 2);

    // Waiter 1 queues first and asks for 3 — more than the 2 free.
    let order = Arc::new(Mutex::new(Vec::new()));
    let sem1 = sem.clone();
    let order1 = order.clone();
    let w1 = tokio::spawn(async move {
        let _p = sem1.acquire_many(3).await.unwrap();
        order1.lock().unwrap().push("w1");
    });
    sleep(Duration::from_millis(50)).await; // let w1 reach the front of the queue

    // Waiter 2 asks for 1. Two permits are free — but it still waits.
    let done2 = Arc::new(AtomicBool::new(false));
    let sem2 = sem.clone();
    let done2b = done2.clone();
    let order2 = order.clone();
    let w2 = tokio::spawn(async move {
        let _p = sem2.acquire().await.unwrap();
        done2b.store(true, Ordering::SeqCst);
        order2.lock().unwrap().push("w2");
    });
    sleep(Duration::from_millis(100)).await;
    println!(
        "w2 done while 2 permits sit free: {}",
        done2.load(Ordering::SeqCst)
    );
    assert!(
        !done2.load(Ordering::SeqCst),
        "head-of-line: w2 must wait behind w1"
    );

    drop(held); // 3 free now: w1 takes all 3, then w2 waits for w1 to drop.
    w1.await.unwrap();
    w2.await.unwrap();
    println!("completion order: {:?}", order.lock().unwrap());
    assert_eq!(*order.lock().unwrap(), vec!["w1", "w2"]);
    println!("head-of-line-blocking OK");
}

The output:

w2 done while 2 permits sit free: false
completion order: ["w1", "w2"]
head-of-line-blocking OK

Waiter 2 asks for one permit while two sit free, and still waits, because waiter 1 asked first and asked for three. The practical rule: keep acquire_many requests small relative to the semaphore, or a single greedy waiter stalls every small waiter behind it. If one part of the program needs many permits at once and another needs one, they should not share a semaphore.

Cancellation loses queue place; close() ends the semaphore

Two more behaviors from the same documentation page, both executed in examples/04-cancel-close.rs. First, canceling an acquire loses the caller's place in the queue — the docs say so for acquire, acquire_many, and both _owned variants. Second, close() prevents new permits and wakes pending waiters with an error, and the error type depends on how you ask: acquire after close returns AcquireError, while try_acquire returns TryAcquireError::Closed, which is distinct from TryAcquireError::NoPermits (permits exhausted, semaphore still open).

// 04-cancel-close — canceling an acquire loses queue place; close() ends
// the semaphore with AcquireError for waiters and Closed for try_acquire.
// Also prints MAX_PERMITS and shows mpsc::channel(0) panicking.
// Run: cargo run --offline --example 04-cancel-close
use std::sync::{
    atomic::{AtomicBool, Ordering},
    Arc,
};
use tokio::sync::{Semaphore, TryAcquireError};
use tokio::time::{sleep, Duration};

#[tokio::main]
async fn main() {
    // --- cancellation loses queue place ---
    let sem = Arc::new(Semaphore::new(1));
    let held = sem.acquire().await.unwrap();

    let got_a = Arc::new(AtomicBool::new(false));
    let got_a2 = got_a.clone();
    let sem_a = sem.clone();
    let h_a = tokio::spawn(async move {
        let _p = sem_a.acquire().await.unwrap();
        got_a2.store(true, Ordering::SeqCst);
    });
    sleep(Duration::from_millis(50)).await; // A is queued
    h_a.abort(); // canceling the acquire drops A's queue place
    let _ = h_a.await; // JoinError: task canceled

    let got_b = Arc::new(AtomicBool::new(false));
    let got_b2 = got_b.clone();
    let sem_b = sem.clone();
    let h_b = tokio::spawn(async move {
        let _p = sem_b.acquire().await.unwrap();
        got_b2.store(true, Ordering::SeqCst);
        sleep(Duration::from_millis(10)).await;
    });
    sleep(Duration::from_millis(50)).await; // B is queued where A was
    drop(held); // the permit goes to B, not to the canceled A
    h_b.await.unwrap();
    println!(
        "after cancel: A acquired={} B acquired={}",
        got_a.load(Ordering::SeqCst),
        got_b.load(Ordering::SeqCst)
    );
    assert!(!got_a.load(Ordering::SeqCst));
    assert!(got_b.load(Ordering::SeqCst));

    // --- NoPermits vs Closed ---
    let sem2 = Arc::new(Semaphore::new(1));
    let _p = sem2.try_acquire().unwrap();
    assert_eq!(sem2.try_acquire().err(), Some(TryAcquireError::NoPermits));
    println!("try_acquire on exhausted open semaphore: NoPermits");
    drop(_p);

    let sem3 = Arc::new(Semaphore::new(1));
    // Hold the only permit first, so the waiter is genuinely queued
    // when close() wakes it — otherwise it would acquire at once.
    let _held3 = sem3.try_acquire().unwrap();
    let sem3b = sem3.clone();
    // The waiter must use acquire_owned: a borrowed SemaphorePermit cannot
    // cross the spawn boundary (see the "why acquire_owned exists" section).
    let waiter = tokio::spawn(async move { sem3b.acquire_owned().await });
    sleep(Duration::from_millis(50)).await;
    sem3.close();
    assert!(waiter.await.unwrap().is_err());
    println!("waiter woken by close() got AcquireError");
    assert_eq!(sem3.try_acquire().err(), Some(TryAcquireError::Closed));
    println!("try_acquire after close(): Closed");
    assert!(sem3.is_closed());

    // --- limits ---
    println!("Semaphore::MAX_PERMITS = {}", Semaphore::MAX_PERMITS);
    assert_eq!(Semaphore::MAX_PERMITS, usize::MAX >> 3);
    // Silence the panic hook so the caught panic below does not print:
    // its message is "mpsc bounded channel requires buffer > 0".
    std::panic::set_hook(Box::new(|_| {}));
    let panicked = std::panic::catch_unwind(|| tokio::sync::mpsc::channel::<u32>(0));
    assert!(panicked.is_err());
    println!("mpsc::channel(0) panics (capacity must be at least 1)");
    println!("cancel-close-limits OK");
}

The output:

after cancel: A acquired=false B acquired=true
try_acquire on exhausted open semaphore: NoPermits
waiter woken by close() got AcquireError
try_acquire after close(): Closed
Semaphore::MAX_PERMITS = 2305843009213693951
mpsc::channel(0) panics (capacity must be at least 1)
cancel-close-limits OK

Three details in that program deserve a second look. The canceled task A is aborted with JoinHandle::abort, which drops its acquire future; when the permit is released, it goes to B, the next waiter in line. A task that re-queues after being canceled goes to the back of the queue — there is no way to resume the old place. The close test holds the only permit before spawning the waiter, because a waiter that is not actually waiting cannot be woken: the first version of this test forgot that and the waiter acquired the free permit immediately, so the assertion failed. And the channel limit is enforced by panic, not by an error: mpsc::channel(0) panics with mpsc bounded channel requires buffer > 0, and the documented maximum capacity is Semaphore::MAX_PERMITS — usize::MAX >> 3, which printed as 2305843009213693951 on this 64-bit machine. Exceeding the semaphore limit via new or add_permits panics the same way.

The HTTP fan-out: a real client against a local server

The semaphore half is only useful if the thing it guards is real work. examples/05-http-fanout.rs points reqwest at a local test server — a TcpListener on an ephemeral port that reads a request, waits 50 milliseconds, and answers 200 ok — and fans out 50 requests through a 5-permit semaphore, acquiring inside each spawned task. The server delay is what makes the observation possible: without it, requests finish faster than tasks start and the counter never climbs. The client is one reqwest::Client, created once and cloned into each task. The docs say the client holds a connection pool internally and should be created once and reused, and that it does not need wrapping in an Arc because it already uses one internally. No timeout, retry, or connection-limit behavior is configured or claimed here — this guide sets none of those, and the program measures only what the semaphore controls.

// 05-http-fanout — real HTTP fan-out: reqwest against a local test server.
// The server delays every response 50ms; the semaphore (5 permits) is
// acquired INSIDE each spawned task, so all 50 tasks exist but at most
// 5 requests are in flight. Run: cargo run --offline --example 05-http-fanout
use std::sync::{
    atomic::{AtomicUsize, Ordering},
    Arc,
};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpListener;
use tokio::sync::Semaphore;
use tokio::time::{sleep, Duration};

static IN_FLIGHT: AtomicUsize = AtomicUsize::new(0);
static MAX_IN_FLIGHT: AtomicUsize = AtomicUsize::new(0);

async fn serve_forever(listener: TcpListener) {
    loop {
        let (mut socket, _) = match listener.accept().await {
            Ok(pair) => pair,
            Err(_) => return,
        };
        tokio::spawn(async move {
            let mut buf = [0u8; 1024];
            // Read the request headers, then answer after a fixed delay.
            let mut got = 0;
            while got < buf.len() {
                match socket.read(&mut buf[got..]).await {
                    Ok(0) | Err(_) => return,
                    Ok(n) => {
                        got += n;
                        if buf[..got].windows(4).any(|w| w == b"\r\n\r\n") {
                            break;
                        }
                    }
                }
            }
            sleep(Duration::from_millis(50)).await;
            let body = b"ok";
            let resp = format!(
                "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
                body.len()
            );
            let _ = socket.write_all(resp.as_bytes()).await;
            let _ = socket.write_all(body).await;
        });
    }
}

#[tokio::main]
async fn main() {
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let port = listener.local_addr().unwrap().port();
    tokio::spawn(serve_forever(listener));

    // One client, reused for all 50 requests: reqwest holds a connection
    // pool internally, so there is no need to wrap it in an Arc.
    let client = reqwest::Client::new();
    let sem = Arc::new(Semaphore::new(5));
    let mut handles = Vec::new();
    for i in 0..50u32 {
        let client = client.clone();
        let sem = sem.clone();
        handles.push(tokio::spawn(async move {
            let _permit = sem.acquire().await.unwrap();
            let n = IN_FLIGHT.fetch_add(1, Ordering::SeqCst) + 1;
            MAX_IN_FLIGHT.fetch_max(n, Ordering::SeqCst);
            let url = format!("http://127.0.0.1:{port}/{i}");
            let status = client.get(&url).send().await.unwrap().status();
            IN_FLIGHT.fetch_sub(1, Ordering::SeqCst);
            assert!(status.is_success());
            i
        }));
    }
    let mut ok = 0;
    for h in handles {
        h.await.unwrap();
        ok += 1;
    }
    println!(
        "http-fanout: {ok} requests, all 200; max_in_flight={} (permits=5)",
        MAX_IN_FLIGHT.load(Ordering::SeqCst)
    );
    assert_eq!(MAX_IN_FLIGHT.load(Ordering::SeqCst), 5);
    println!("http-fanout OK");
}

The output:

http-fanout: 50 requests, all 200; max_in_flight=5 (permits=5)
http-fanout OK

Fifty requests, all answered 200, and the in-flight counter never exceeded 5. Note what this placement does not do: it does not bound the number of tasks. All 50 tasks exist from the start, each holding a request URL and waiting on the semaphore — exactly placement A from the main binary, now with real HTTP underneath. If the tasks themselves were the expensive resource — if each one held a large buffer or a database connection — placement B, acquiring before spawn, would be the right shape here too. The semaphore bounds the requests; only the placement bounds the tasks.

What breaks, and the fix

| Error | Cause | Fix | | command not found for cargo | Rustup toolchain not on PATH in a shell with no terminal attached | export PATH="$HOME/.cargo/bin:$PATH" | | error[E0597]: 'sem' does not live long enough | A borrowed SemaphorePermit moved into tokio::spawn, which requires a 'static future | Use acquire_owned so the permit owns its Arc<Semaphore> | | w2 never completes while permits sit free | Head-of-line blocking: a queued acquire_many(3) is ahead of it | Keep acquire_many requests small, or give the two sizes separate semaphores | | Waiter not woken by close() | The waiter was not actually waiting — a permit was free | Hold the permit first, then spawn the waiter, then close | | try_acquire returns NoPermits when you expected Closed | The semaphore is exhausted but still open | NoPermits means out of permits; Closed means close() was called — check is_closed() | | mpsc bounded channel requires buffer > 0 | mpsc::channel(0) | Capacity must be at least 1 | | Panic from Semaphore::new or add_permits | More than MAX_PERMITS (usize::MAX >> 3) | Stay below the limit; the panic is the enforcement | | Offline build fails resolving a new dependency | The dependency is not in the registry cache | Add it while online, or pick a version the cache holds |

Frequently asked questions

Can I use the semaphore as a rate limiter?

No — not as shown in the docs' own token-bucket example, and the docs say so themselves: that implementation "is suboptimal when the duration is small, because it consumes a lot of cpu constantly looping and sleeping." A semaphore limits how many things happen at once; a rate limiter limits how many happen per second. Those are different guarantees, and a semaphore with a timer task refilling permits conflates them. If you need a rate — requests per second against an API quota — use a real rate-limiting mechanism, not a semaphore with add_permits on a timer.

Should the channel be bounded or unbounded?

Bounded, unless you can prove the producer cannot outrun the consumer. An unbounded channel's send always completes immediately, which is convenient and is also the failure mode: memory grows with the backlog and nothing pushes back on the producer. A bounded channel turns that growth into waiting, which is observable, measurable, and fixable. Pick the capacity from the backlog you are willing to hold — a channel of capacity 2 holds two items of slack, no more — and treat a parked sender as the system working, not as a bug.

When do I acquire before spawn, and when inside the task?

Acquire before spawn when the task itself is the resource: each task holds memory, a connection, or a file handle, and 50 waiting tasks cost more than 5 working ones. Acquire inside the task when the guarded resource is downstream — the network, an API, a disk — and the tasks are cheap futures waiting their turn. The cost of getting it wrong is concrete: placement A with 50 heavy tasks holds 50 tasks' worth of resources while 5 work; placement B with a slow producer starves the consumer, because the producer cannot even create the next task until a permit frees.

Does canceling a send lose the message?

This guide does not answer that, and deliberately so: the semaphore cancel-safety claims above are documented for acquire and its variants, and the channel's send-cancellation behavior was not verified against the mpsc documentation at build time. Do not carry a cancel-safety claim for send().await from this page.

Why is Semaphore::const_new not the default in the examples?

const_new builds a semaphore usable in a static, which the docs' file-writing example does. The same page notes that a semaphore created with const_new is not instrumented for tokio-console under the unstable tracing feature; Semaphore::new is the instrumented path. The programs here create semaphores at runtime, so new is the right call either way.

Sources

• Tokio, Semaphore — fair permit queue, head-of-line blocking, cancel safety, both placements, close() semantics, MAX_PERMITS, the token-bucket warning.
• Tokio, tokio::sync::mpsc — bounded vs unbounded channels, backpressure, disconnection, clean shutdown, blocking_* sync bridge.
• Tokio, mpsc::channel — buffer semantics, capacity at least 1, maximum Semaphore::MAX_PERMITS, FIFO order.
• Tokio, SemaphorePermit — forget, merge, split, num_permits.
• Tokio, OwnedSemaphorePermit — forget, merge, split, semaphore().
• reqwest, Client — one reusable client, internal connection pool, no Arc needed, Client::new() and get().

Facts verified 2026-11-05. Every command and code listing on this page was compiled and run with cargo 1.99.1, rustc 1.99.1, Tokio 1.53.2, and reqwest 0.12.28 on x86_64 Linux (WSL), except the combinations labeled as untested and the permit forget / split / merge API surface, which is documented but not executed here. Failures are quoted verbatim from the actual compiler and program output. No search-demand evidence exists for this topic; no demand is claimed.