How asyncio actually works

Almost everything an AI service does is wait: for a model, for a database, for another service. asyncio is Python's way of overlapping those waits on a single thread, so fifty calls that each take a second finish in about a second rather than fifty. This section covers how the event loop manages that, how to put timeouts and cancellation around it, and the one mistake that undoes all of it, which is a blocking call inside a coroutine freezing every other request until it returns.

The idea in one example#

import asyncio, time

async def fake_call(n):
    await asyncio.sleep(1)       # waiting, like a network call
    return n

async def main():
    start = time.perf_counter()
    results = await asyncio.gather(*(fake_call(i) for i in range(10)))
    print(results, f"{time.perf_counter() - start:.1f}s")   # about 1.0s, not 10

asyncio.run(main())

One thread runs an event loop. When a coroutine reaches await on something slow, it gives control back to the loop, which runs another coroutine until that one waits too. Ten one-second waits overlap and finish in about one second. Nothing runs in parallel; the waiting does.

What the event loop actually is#

Strip the vocabulary away and the loop is a while loop over a queue of things that are ready to run. It takes the first one, runs it until it hits an await on something not yet finished, and at that moment the function returns control to the loop while keeping its own local variables and its position in the code. The loop asks the operating system which of the sockets it is waiting on have data, moves anything that is ready back onto the queue, and takes the next item.

ready queue: [A, B, C]

run A  -> A awaits a socket read     A parked, loop registers the socket
run B  -> B awaits a socket read     B parked
run C  -> C awaits a socket read     C parked
ready queue empty; ask the OS which sockets have data, and wait

   B's reply arrives -> B back on the queue
run B  -> B finishes

Nothing was running while all three waited. That waiting is what overlapped.

Three pieces of vocabulary follow from that. An async def function is a coroutine function, and calling it does not run anything — it returns a coroutine object, which is why a forgotten await produces no error and no effect, only a RuntimeWarning that a coroutine was never awaited and a result that is mysteriously absent. await runs it and gives the loop permission to run something else in the meantime. And a Task is a coroutine handed to the loop to get on with immediately, which is what makes it run concurrently with the line after it rather than instead of it.

async def fetch(url): ...

c = fetch("a")           # nothing has happened. c is a coroutine object.
r = await c              # now it runs, and this line waits for it

# Sequential: the second starts when the first has finished. 2 seconds.
a = await fetch("a")
b = await fetch("b")

# Concurrent: both are handed to the loop, then both are waited on. 1 second.
ta = asyncio.create_task(fetch("a"))
tb = asyncio.create_task(fetch("b"))
a, b = await ta, await tb

gather, as_completed and TaskGroup#

These are three ways to run many coroutines at once, and they differ in when you get results and what happens when one fails. gather returns a list in the order you passed them in, once all of them are done. as_completed yields each one as it finishes, in completion order, which is what lets a long job report progress. A TaskGroup is gather with strict failure handling.

# gather: results in input order, when everything has finished.
results = await asyncio.gather(*(fetch(u) for u in urls))

# One failure cancels nothing and loses everything else:
async def boom(i):
    await asyncio.sleep(0.1)
    if i == 2:
        raise ValueError(f"item {i} failed")
    return i

await asyncio.gather(*(boom(i) for i in range(4)))
# ValueError: item 2 failed        -- items 0, 1 and 3 succeeded and you have none of them

await asyncio.gather(*(boom(i) for i in range(4)), return_exceptions=True)
# [0, 1, ValueError('item 2 failed'), 3]

# as_completed: results as they arrive, so you can report progress.
tasks = [slow(i) for i in range(5)]
for coro in asyncio.as_completed(tasks):
    result = await coro
    print("done:", result)
# done: 2
# done: 1
# done: 0
# done: 4
# done: 3        -- completion order, not input order

async with asyncio.TaskGroup() as tg:          # Python 3.11+
    t1 = tg.create_task(fetch("a"))
    t2 = tg.create_task(fetch("b"))
print(t1.result(), t2.result())    # only reached if both succeeded

If one task fails, a TaskGroup cancels its siblings, waits for them to finish unwinding, and raises an ExceptionGroup containing everything that went wrong. Catch it with except* rather than except. That is the behaviour you want for a request handler, where a partial result is worse than a clean failure; gather with return_exceptions is the behaviour you want for a batch, where it is better.

try:
    async with asyncio.TaskGroup() as tg:
        t1 = tg.create_task(ok(1))       # takes 0.3s
        t2 = tg.create_task(fail())      # raises after 0.05s
except* RuntimeError as eg:
    print([str(e) for e in eg.exceptions])   # ['nope']
    print(t1.cancelled())                    # True -- the sibling was cancelled

Limiting how many run at once#

gather over ten thousand items starts ten thousand requests, which exhausts your rate limit, your file handles and your memory at the same time. A semaphore is a counter with a limit: acquiring it when the count is at the limit suspends the coroutine until somebody releases. Wrapping the work in one caps how many are in flight without changing anything else about the code.

sem = asyncio.Semaphore(3)

async def one(i):
    async with sem:                 # waits here if 3 are already inside
        await asyncio.sleep(0.1)
    return i

await asyncio.gather(*(one(i) for i in range(12)))
# peak in flight: 3
# elapsed: 0.41s      -- four waves of three, not one wave of twelve

Every call needs a timeout#

async def call_with_timeout(prompt):
    try:
        async with asyncio.timeout(30):            # Python 3.11+
            return await client.generate(prompt)
    except TimeoutError:
        return None                                # decide what a timeout means here

Without one, a request that never completes holds its slot in the semaphore for ever, and a batch job that is ninety-nine percent finished sits at ninety-nine percent until somebody kills it. The timeout applies to everything inside the block, so it bounds a retry loop as a whole rather than each attempt, which is usually what you want; asyncio.wait_for wraps a single awaitable and is the older spelling. Note that since Python 3.11 asyncio.TimeoutError is the builtin TimeoutError, so catching the builtin is correct.

A timeout works by cancelling the coroutine, which raises CancelledError at the await it was suspended on. That gives your code a chance to clean up, and it means two things about how you write it: never swallow CancelledError, because catching it and continuing leaves the loop believing a task it cancelled is still running, and put cleanup in a finally rather than after the await.

async def worker():
    try:
        await asyncio.sleep(5)
    except asyncio.CancelledError:
        release_the_slot()          # clean up
        raise                       # then re-raise. Always.
    finally:
        await conn.close()

t = asyncio.create_task(worker())
await asyncio.sleep(0.05)
t.cancel()
try:
    await t
except asyncio.CancelledError:
    pass                            # awaiting a cancelled task re-raises it

The call that freezes everything#

If any coroutine calls something slow that does not await, such as time.sleep, a synchronous HTTP library, or heavy CPU work, the whole loop stops. Every other request waits. Move blocking work off the loop.

This is the failure worth being able to recognise from a description, because it presents as a mystery: latency for every user spikes at the same moment, the slow endpoint is not the one anybody complained about, and the CPU graph shows one core busy. There is one thread. Whatever it is doing, it is not serving anyone else.

async def good(i):
    await asyncio.sleep(0.5)
    return i

async def bad():
    time.sleep(2.0)          # blocking: no await, so the loop cannot leave
    return "bad"

await asyncio.gather(bad(), *(good(i) for i in range(3)))
# 2.5s      the three 0.5s waits did not start until bad() returned

await asyncio.gather(asyncio.to_thread(time.sleep, 2.0), *(good(i) for i in range(3)))
# 2.0s      the loop stayed free, so the 0.5s waits overlapped with the 2s

# Bad: blocks every other request for the duration
text = extract_pdf_text(path)

# Good: runs in a worker thread, loop stays free
text = await asyncio.to_thread(extract_pdf_text, path)

# For CPU-bound work, a thread does not help. Use a process pool.
loop = asyncio.get_running_loop()
with ProcessPoolExecutor() as pool:
    result = await loop.run_in_executor(pool, heavy_numeric_function, data)

• requests, and any library built on it, is blocking. Use httpx or aiohttp in async code.
• time.sleep blocks; asyncio.sleep does not. The same distinction applies to every wait in the standard library.
• A synchronous database driver blocks. psycopg with an async connection, or asyncpg, does not.
• open() and file reads block, briefly, and on a slow network filesystem not briefly at all.
• json.loads on a twenty megabyte response blocks for a noticeable time, and so does any parsing or encryption of something large.
• Run the loop with debug=True during development: it logs any callback that occupied the loop for more than 100 ms, which finds these without guessing.

Threads, processes, async#

Waiting on network or disk, many at once   asyncio (or threads)
Blocking library you cannot change          threads
Heavy CPU work in Python                    processes
Heavy numeric work                          NumPy, PyTorch: they release the GIL internally

In standard CPython, the global interpreter lock lets only one thread run Python code at a time, so threads do not speed up CPU-bound Python. Recent Python versions offer an optional build without the lock, but most production code still assumes it.