Asyncio in Python

Note · Published on Nov 22, 2025

Contents

Does await always let other tasks run? It does not. When I orchestrated async RL training, with samplers, a trainer and evals running together, I needed to know exactly when a task runs and when it gives up control. This note is the mental model I built to answer that, with two use cases: calling LLM APIs and async RL training.

Asyncio vs Multi-Thread vs Multi-Process#

Python has three concurrency models: asyncio, multi-thread, and multi-process.

	asyncio	multi-thread	multi-process
Best for	I/O-bound tasks	I/O-bound tasks	CPU-intensive tasks
Who hands over control	your code, at an `await` that has to wait	the OS	the OS
Cost	low	a separate stack per thread	a separate interpreter per process

For CPU-intensive tasks, use multi-process, so several processes split the work. Multi-thread cannot help because of the GIL: only one thread can run Python code at a time. Neither can asyncio.
For I/O-bound tasks, use asyncio or multi-thread. While one task waits on I/O, another one runs. A large batch of API calls is the typical case. If your library has an async client, use asyncio.

How Await Gives Up Control#

Underneath, everything a task waits on, such as a network response or a timer, is a future. That is how a single thread switches between tasks.

The mental model is that an event loop keeps track of every submitted task [1]. When a task runs await coro, the coroutine runs until it reaches a pending future. The task then gives control to the event loop, which picks the next task to run. When the future finishes, the event loop resumes the task at the same line. In other words, await x never moves to the next line until x has its result.

Concept	What it is
Coroutine	The object returned by calling an async function. It runs only when awaited or wrapped in a task.
Future	A result that is not ready yet, such as a network response. Awaiting a pending future pauses the current task.
Task	A future whose result comes from running a coroutine. A plain future is completed by someone else, such as an async library; a task completes itself when its coroutine returns.

A small example:

async def child(name):
    print(name)

async def main():
    t = asyncio.create_task(child("B"))
    await child("A")
    print("main: next line")
    await t

asyncio.create_task(coro) wraps a coroutine in a task and schedules it on the event loop. It returns at once. The task starts when the current task gives up control.

The output is A, main: next line, B. B was scheduled first, but await child("A") never reaches a pending future, so main does not give up control until await t.

Applying the Model#

Calling APIs#

An example with OpenAI API calls:

client = AsyncOpenAI()
sem = asyncio.Semaphore(50)

async def ask(prompt):
    # at most 50 calls at once; give up on one after 30 seconds
    async with sem, asyncio.timeout(30):
        return await client.chat.completions.create(model=MODEL, messages=[...])

async def main(prompts):
    # a failed call comes back as an exception in the list
    return await asyncio.gather(*[ask(p) for p in prompts], return_exceptions=True)

asyncio.gather(*aws, return_exceptions=False) runs the awaitables concurrently and returns their results in input order. By default it raises the first exception. With return_exceptions=True, exceptions come back as items in the list.

Orchestrating Work in RL Training#

Sampler tasks capture the latest published policy and its version, run a rollout, and put the result with its version in the queue. queue.put waits while the queue is full.
The train step collects a batch with queue.get(), drops results whose version is too old, does one update, and publishes the new policy.
Eval tasks run every EVAL_EVERY steps on a captured policy, so training does not wait for them.
The training loop in train() waits on all of them with asyncio.wait and moves on only when the batch is ready. If a sampler or eval fails, the finally block cancels the rest.

Samplers fill a bounded queue, the train step takes a fresh batch and publishes a new policy, and the training loop moves on only when the batch is ready or a task fails.

async def sampler(queue):
    while True:
        # capture the latest published policy and its version
        p, v = policy, version
        # waits when the queue is full
        await queue.put((v, await rollout(p)))

async def train():
    queue = asyncio.Queue(maxsize=QUEUE_CAPACITY)
    samplers = [asyncio.create_task(sampler(queue)) for _ in range(NUM_WORKERS)]
    evals = []
    try:
        for step in range(1, NUM_STEPS + 1):
            batch_task = asyncio.create_task(collect_fresh_batch(queue))
            # wake up when any task finishes; move on once the batch is ready
            while not batch_task.done():
                done, _ = await asyncio.wait([batch_task, *samplers, *evals],
                                             return_when=asyncio.FIRST_COMPLETED)
                for t in done:
                    # raises if a sampler or eval failed
                    t.result()
                    # drop a finished eval so wait does not return it again
                    if t in evals:
                        evals.remove(t)
            await update(batch_task.result())
            await publish(version + 1)
            if step % EVAL_EVERY == 0:
                evals.append(asyncio.create_task(evaluate(policy, version)))
        await asyncio.gather(*evals)
    finally:
        # stop the samplers, and on error the evals too
        for t in [*samplers, *evals]:
            t.cancel()

References#

Python documentation. A Conceptual Overview of asyncio.