Running Concurrent Requests Safely

Unit 12, Lesson 6 showed the basic pattern of firing off multiple requests with asyncio.gather and bounding them with a single asyncio.Semaphore. That pattern works for a demonstration of a few dozen calls. It starts to break down once you're processing tens of thousands of records, need partial results as they complete rather than all at once, need to cancel cleanly on a fatal error, or need to control memory usage when the full list of records doesn't comfortably fit as pending tasks in memory at once. This lesson covers the concurrency patterns that make bulk request processing robust rather than merely functional.

gather vs. as_completed: Two Different Needs

asyncio.gather(*coroutines) runs everything concurrently and returns a list of results only once every single one has finished, in the same order they were passed in. This is the right tool when you need all results together before continuing, and when the input list is small enough to hold entirely in memory as scheduled tasks.

asyncio.as_completed(coroutines) also runs everything concurrently, but instead gives you an iterator that yields each result as soon as it finishes, in completion order rather than input order. This matters for large batches: you can start acting on results (saving them, updating progress) immediately rather than waiting for the single slowest item to finish before processing anything.

import asyncio


async def fetch_with_delay(n: int) -> int:
    await asyncio.sleep(0.1 * (5 - n))  # later items finish sooner
    return n


async def demo_as_completed():
    coros = [fetch_with_delay(n) for n in range(5)]
    for coro in asyncio.as_completed(coros):
        result = await coro
        print(f"finished: {result}")  # prints in completion order, not 0,1,2,3,4


asyncio.run(demo_as_completed())

For a pipeline's Collect stage (Lesson 2), as_completed is usually the better fit, because it lets you write each result to durable storage the moment it's ready — which directly supports the progress tracking in Lesson 6 and the resumability design in Lesson 9, both of which depend on results being persisted incrementally rather than all at the end.

Bounding Concurrency with a Worker Pool Pattern

A single semaphore wrapped around individual coroutines (as in Unit 12) works, but at high volumes a producer/consumer worker pool built on asyncio.Queue gives you more control: a fixed number of worker coroutines pull items from a shared queue and process them, so memory usage stays bounded by the queue size rather than by the number of tasks you've created up front, and you can add or remove workers without changing how records are enqueued.

import asyncio
from openai import AsyncOpenAI


async def worker(name: str, queue: asyncio.Queue, client: AsyncOpenAI, results: list):
    while True:
        record = await queue.get()
        if record is None:  # sentinel value signals "no more work"
            queue.task_done()
            break
        try:
            response = await client.responses.create(
                model="gpt-5.6-terra",
                input=record.prompt,
            )
            record.model_response = response.output_text
            record.status = RecordStatus.SUCCEEDED
        except Exception as exc:
            record.error = str(exc)
            record.status = RecordStatus.FAILED
        results.append(record)
        queue.task_done()


async def run_worker_pool(records: list, concurrency: int) -> list:
    client = AsyncOpenAI()
    queue: asyncio.Queue = asyncio.Queue()
    results: list = []

    for record in records:
        queue.put_nowait(record)

    workers = [
        asyncio.create_task(worker(f"worker-{i}", queue, client, results))
        for i in range(concurrency)
    ]

    await queue.join()  # wait until every enqueued item has been processed

    for _ in workers:
        queue.put_nowait(None)  # tell each worker to stop
    await asyncio.gather(*workers)

    return results

This pattern has several properties worth calling out explicitly. The number of worker coroutines (concurrency) directly and predictably bounds how many requests are in flight at once — there is no way for the system to accidentally schedule more concurrent calls than that, unlike a naive gather over thousands of coroutines where every one of them starts immediately unless separately throttled. The None sentinel is a standard idiom for telling a fixed pool of consumers to stop: each worker exits its loop the first time it dequeues a None, so you enqueue exactly as many None values as there are workers. queue.join() blocks until every item put on the queue has had task_done() called for it, which is how the pool knows all real work items (not just the sentinels) have been processed before it starts shutting workers down.

Isolating Failures So One Bad Item Doesn't Take Down the Rest

A concurrency bug that is easy to introduce and expensive to have in production: letting one coroutine's unhandled exception propagate up through gather and cancel every other in-flight task. By default, asyncio.gather without return_exceptions=True does exactly this — the moment one coroutine raises, gather raises too, and any tasks still running are left in a cancelled state, silently discarding whatever work they had done.

async def risky(n: int) -> int:
    if n == 3:
        raise ValueError("simulated failure")
    await asyncio.sleep(0.05)
    return n


async def demo_failure_isolation():
    coros = [risky(n) for n in range(5)]

    # Without return_exceptions=True, one failure cancels the whole batch.
    results = await asyncio.gather(*coros, return_exceptions=True)
    for n, result in enumerate(results):
        if isinstance(result, Exception):
            print(f"item {n} failed: {result}")
        else:
            print(f"item {n} succeeded: {result}")


asyncio.run(demo_failure_isolation())

return_exceptions=True changes gather's behavior so that a failing coroutine's exception is captured and placed in the results list at its corresponding position instead of being re-raised — every other coroutine keeps running to completion regardless. The worker pool pattern above achieves the same isolation more directly, since each worker already catches exceptions per item inside its own try/except and never lets one item's failure affect the loop that processes the next one. Either approach is valid; the worker pool additionally isolates failures within a single long-running coroutine (the worker itself never crashes), while gather(..., return_exceptions=True) isolates failures across a fixed, one-shot list of coroutines. Lesson 7 goes further into what to actually do with these captured failures at the pipeline level.

Structured Concurrency with TaskGroup

Python 3.11 introduced asyncio.TaskGroup, which addresses a subtler problem with plain gather: if you cancel or the program shuts down while tasks are running, gather gives you no built-in guarantee that every child task is cleanly awaited or cancelled before your code continues. TaskGroup is a context manager that guarantees all tasks created within it are complete (or cancelled together) before the async with block exits, which makes cleanup and cancellation behavior predictable:

async def demo_task_group():
    results = []

    async def process(n: int):
        await asyncio.sleep(0.05)
        results.append(n * n)

    async with asyncio.TaskGroup() as tg:
        for n in range(5):
            tg.create_task(process(n))
    # By this point, every task has finished (or the group raised together).

    print(sorted(results))

Unlike gather, if any task inside a TaskGroup raises an unhandled exception, the group cancels every other task in the group and raises a single ExceptionGroup containing all the failures once everything has actually stopped — there is no window where some tasks are left dangling. For new code targeting Python 3.11 or later, TaskGroup is generally the more robust default over gather for exactly this reason, though gather(..., return_exceptions=True) remains the more portable choice for anything running on 3.10 or earlier.

Note: asyncio.TaskGroup requires Python 3.11+. Confirm the Python version your production environment targets before relying on it, and check current documentation for its exact exception-grouping behavior.

Testing Concurrency Logic Without Real Calls

Because the worker pool and failure-isolation logic above are pure asyncio control flow, they can be tested with a fake client that never touches the network:

class FakeAsyncClient:
    class responses:
        @staticmethod
        async def create(model, input):
            class FakeResponse:
                output_text = f"fake response to: {input}"
            if "fail" in input:
                raise RuntimeError("simulated API error")
            return FakeResponse()


async def test_worker_pool_isolates_failures():
    records = [
        PipelineRecord(record_id=str(i), source_data={}, prompt=f"item {i}")
        for i in range(4)
    ]
    records[2].prompt = "please fail this one"

    queue: asyncio.Queue = asyncio.Queue()
    for r in records:
        queue.put_nowait(r)
    results: list = []
    workers = [
        asyncio.create_task(worker(f"w{i}", queue, FakeAsyncClient(), results))
        for i in range(2)
    ]
    await queue.join()
    for _ in workers:
        queue.put_nowait(None)
    await asyncio.gather(*workers)

    succeeded = [r for r in results if r.status == RecordStatus.SUCCEEDED]
    failed = [r for r in results if r.status == RecordStatus.FAILED]
    assert len(succeeded) == 3
    assert len(failed) == 1
    print("PASS: worker pool isolates one failure without affecting others")


asyncio.run(test_worker_pool_isolates_failures())

FakeAsyncClient mimics only the shape the worker function needs — a responses.create coroutine — which is enough to exercise the real control flow (queueing, dispatching to workers, catching exceptions, recording status) without any dependency on the actual OpenAI service, cost, or network reliability.

Common Mistakes

• Creating unbounded concurrency by calling gather over every record at once with no limit. Ten thousand simultaneous coroutines each opening a connection will exhaust file descriptors, overwhelm the API's rate limits (Lesson 5), and consume far more memory than a bounded worker pool.
• Letting one failing coroutine cancel the entire batch by omitting return_exceptions=True on gather, or by not catching exceptions inside a long-running worker loop — either mistake can silently discard completed-but-uncollected work.
• Forgetting to send the correct number of sentinel values when shutting down a worker pool, which leaves some workers permanently blocked waiting on an empty queue and the program never exits.

Best Practices

• Prefer a bounded worker pool or a semaphore-limited gather over unbounded concurrency for any batch of more than a few dozen items, and choose the pool when you also need incremental result handling or when the record list is very large.
• Always decide explicitly how failures propagate — either isolate them per item (inside a worker's try/except) or capture them explicitly with return_exceptions=True — rather than leaving Python's default propagation behavior to determine it implicitly.
• Reach for TaskGroup on Python 3.11+ for new code where clean, guaranteed cancellation semantics matter, and reserve gather for simpler, smaller-scale cases or environments still on older Python versions.