One request should not start a thousand tasks at once

Article · Interrupted execution

An aggregation endpoint receives a list of records and starts one task for each record. In a small test, it is impressively fast.

Then a customer sends a much larger list. The service creates thousands of tasks, each holding an input and waiting for an upstream connection. Adding a semaphore inside every task limits the calls, but all those waiting tasks still exist.

I would put the limit earlier. Admit more work as previous work finishes, instead of creating everything and asking most of it to wait.

Keep the set of tasks bounded

Tokio's JoinSet owns a collection of tasks and returns their results as they complete. This Rust excerpt keeps only a bounded set outstanding:

async fn fanout(
    input: std::ops::Range<usize>,
    limit: usize,
    gate: Arc<Semaphore>,
    started: mpsc::UnboundedSender<usize>,
    retained: Arc<AtomicUsize>,
) -> Vec<usize> {
    assert!(limit > 0);
    struct Input(Arc<AtomicUsize>);
    impl Drop for Input {
        fn drop(&mut self) {
            self.0.fetch_sub(1, Ordering::SeqCst);
        }
    }
    let mut tasks = JoinSet::new();
    let mut output = Vec::new();
    for id in input {
        if tasks.len() >= limit {
            output.push(tasks.join_next().await.unwrap().unwrap());
        }
        let (gate, started, retained) = (gate.clone(), started.clone(), retained.clone());
        retained.fetch_add(1, Ordering::SeqCst);
        let input = Input(retained);
        tasks.spawn(async move {
            let _input = input;
            started.send(id).unwrap();
            let permit = gate.acquire().await.unwrap();
            permit.forget();
            id
        });
    }
    while let Some(result) = tasks.join_next().await {
        output.push(result.unwrap());
    }
    output
}

This is the test scheduler, with its required imports in the linked fixture. It validates a nonzero limit and counts input owners. The semaphore is a one-way test gate: consuming its permits with forget deliberately controls progress. A production concurrency limiter normally drops its permit after the operation, returning capacity. The output consumer must also be bounded. Collecting every large result in memory can create a second problem.

Results arrive in completion order. If the response needs input order, attach an index or stable identity and restore order deliberately. Do not accidentally pair a fast result with the wrong input.

The Tokio runtime article explains where spawned work runs. Cheap task creation is useful, but it does not remove the memory and upstream work retained by each task.

Per-request limits are only one layer

A limit of eight tasks per request can still produce eight hundred active calls across a hundred requests. Add a service-wide or provider-account limit when that is the resource you need to protect.

Be careful how those limits interact. A task waiting on a global permit still occupies its per-request slot. That can be fine, provided it is intentional and the wait has a deadline. Avoid holding an unrelated database connection during that wait.

Reject or page an input list that is too large before materializing expensive per-item data. A bounded scheduler cannot undo an unbounded request-body allocation.

Observe cancellation and failures

Decide whether one failed fetch cancels the batch or produces a per-item error. Dropping a JoinSet aborts its tasks, but an external operation may already have happened. Write endpoints need stronger recovery rules than read-only aggregation.

Test using an upstream that waits on a signal. Track the highest active-call count and how many tasks have been created before the signal is released. The bounded design should obey both the active-work policy and its outstanding-task bound.

Then cancel the parent request and check which work remains. A concurrency limit is incomplete when it only describes the happy path.

Spawning everything can be reasonable for a small, strictly bounded input. Put that bound in the API contract. For customer-sized lists, I prefer a scheduler whose resource use follows an explicit limit rather than the length of the customer's data.

Count inputs before releasing the upstream

The runtime reproduction drives a lazy range of twenty input identities through a JoinSet limited to three. The controlled upstream starts with no available permits. Three tasks report that they started; a fourth start must not be present, and the retained-input counter must be three.

Every admitted input owns a drop guard before its task is spawned. This order matters. A task cancelled before its first poll still owns that guard inside its future, so destroying the future releases the count. Constructing the guard only after the task starts would miss that cancellation path.

One run releases the upstream and checks all twenty returned identities. It sorts them only for the assertion; the scheduler itself produces completion order. The other run cancels the parent while its three tasks are waiting. Dropping the parent's JoinSet aborts those tasks, and the observer waits until all three input owners have been destroyed.

The controlled semaphore is a test gate. The fixture's per-request limit does not establish a complete provider-account rate policy, and a count of input owners does not measure their heap size. Add a service-wide limit around the actual shared dependency when several requests compete for it.

The input iterator in this lab is lazy and cheap. A real handler that first parses an enormous request into a Vec has already paid that allocation cost. Bound request size and per-item construction separately. The useful scheduler property is that creating the next expensive owner follows a free slot, with completion and cancellation both returning that slot.