Async & concurrency

A model call spends almost all of its time waiting for a reply, so issuing them one after another leaves nearly all of your wall-clock on the table. This section is about running many at once without exhausting the provider's rate limit or your own memory: capping how many are in flight, collecting results as they finish rather than in order, and making sure one failure does not take the whole batch with it.

Requirement                                  What provides it
many slow calls overlapped                   asyncio.gather over coroutines
never more than N in flight                  asyncio.Semaphore(N), inside the task
one hung call cannot stall the job           asyncio.timeout around each call
one failure cannot lose the batch            return the exception as a value
a 429 does not kill the item                 retry with exponential backoff and jitter
a crash does not lose the work               append each result to disk as it lands
you can tell it apart from a hang            as_completed, print every N
you know what it cost before the bill        accumulate tokens per result

The shape you will write repeatedly#

Almost every eval harness, backfill and batch enrichment job has the same structure: many independent slow calls, a cap on how many run at once, a timeout on each, and a result set that tolerates some failures.

import asyncio

async def process_all(items, limit=8, timeout=60):
    sem = asyncio.Semaphore(limit)

    async def one(item):
        async with sem:
            try:
                async with asyncio.timeout(timeout):
                    return {"item": item, "ok": True, "result": await process(item)}
            except Exception as e:
                return {"item": item, "ok": False, "error": str(e)}

    return await asyncio.gather(*(one(i) for i in items))

Three details in that are load-bearing. The semaphore is acquired inside one, not around the gather, so a coroutine with no slot is parked rather than never created. The timeout is inside the semaphore, so a call that hangs gives its slot back after sixty seconds instead of holding it for ever. And the except returns rather than raises, which is what stops a single failure propagating out of gather and discarding every other result.

Retry the failures worth retrying#

Rate limits and transient server errors are the majority of failures in a large batch, and both succeed on a second attempt. Retry those and nothing else: a 400 for a malformed request will fail identically for ever, and retrying it three times wastes three times the latency to reach the same answer. Wait longer after each attempt, and add randomness, because otherwise every one of your eight workers fails at the same moment and retries at the same moment, which is the same burst again.

import random

RETRYABLE = (RateLimitError, APITimeoutError, InternalServerError)   # 429, 408, 5xx

async def with_retry(fn, *args, attempts=5, base=1.0):
    for attempt in range(attempts):
        try:
            return await fn(*args)
        except RETRYABLE:
            if attempt == attempts - 1:
                raise
            delay = base * (2 ** attempt)          # 1, 2, 4, 8 seconds
            await asyncio.sleep(delay * (0.5 + random.random()))   # jitter: 0.5x to 1.5x
    raise AssertionError("unreachable")

If the provider sends a retry-after header, obey it rather than your own schedule; it is the only number in the exchange that knows when the limit resets. And count retries separately from failures in whatever you print, because a job that succeeds after retrying four hundred times is telling you the concurrency limit is too high.

Progress on a long run#

A job that prints nothing for twenty minutes is indistinguishable from a hung one. as_completed lets you report as results arrive.

done = 0
for coro in asyncio.as_completed(tasks):
    result = await coro
    done += 1
    if done % 25 == 0:
        print(f"{done}/{len(tasks)}  cost so far ${total_cost:.2f}")

Print four things and you never have to guess what a running job is doing: how many are finished, how many failed, the elapsed time with a projection to the end, and the money spent so far. The projection is what makes you kill a job at minute two rather than discovering at minute ninety that it was never going to finish before the deadline.

start = time.perf_counter()
done = failed = 0
tokens_in = tokens_out = 0

for coro in asyncio.as_completed(tasks):
    r = await coro
    done += 1
    failed += not r["ok"]
    if r["ok"]:
        tokens_in += r["result"].usage.input_tokens
        tokens_out += r["result"].usage.output_tokens
    if done % 25 == 0:
        elapsed = time.perf_counter() - start
        rate = done / elapsed
        cost = tokens_in * 3e-6 + tokens_out * 15e-6
        remaining = (len(tasks) - done) / rate
        print(f"{done}/{len(tasks)}  {failed} failed  "
              f"{rate:.1f}/s  eta {remaining/60:.1f} min  ${cost:.2f}")

# 25/500  0 failed  4.1/s  eta 1.9 min  $0.41
# 50/500  2 failed  3.8/s  eta 2.0 min  $0.83

Make it resumable#

Long jobs get interrupted. Write each result as it completes rather than at the end, and skip anything already done on restart. This turns a crash from a lost afternoon into a lost minute.

done_ids = {json.loads(l)["id"] for l in open(out_path)} if exists(out_path) else set()
todo = [i for i in items if i.id not in done_ids]

with open(out_path, "a") as f:
    for coro in asyncio.as_completed([one(i) for i in todo]):
        result = await coro
        f.write(json.dumps(result) + "\n")
        f.flush()

Four things make this work and each is easy to leave out. The file is opened in append mode, so a restart adds rather than truncating. Each record carries its own id, because the order is completion order and position tells you nothing. flush is called after every line, or the last few thousand records sit in a buffer that a crash discards, which defeats the whole exercise. And the format is one JSON object per line, so a half-written final line from a kill in the middle of a write costs you exactly one record and the rest of the file still parses.

# Verified on 20 items, one of which always fails:
#   first run wrote 20 records in 0.07s
#   ok: 19  failed: 1
#   failures: [{'id': 7, 'ok': False, 'error': 'ValueError: bad item'}]
#   rerun processed: 0      -- every id was already in the file

# Skip the line a kill truncated rather than crashing on restart:
done_ids = set()
if exists(out_path):
    with open(out_path) as f:
        for line in f:
            try:
                done_ids.add(json.loads(line)["id"])
            except json.JSONDecodeError:
                pass        # a partial last line: it will simply be redone

Choosing the concurrency limit#

Too low and the job is slow. Too high and you are rate limited, which is slower still because of the backoff. Start at eight, watch for 429s, and adjust. The right number depends on your quota, not on your machine.

Providers usually limit two things at once: requests per minute and tokens per minute. A semaphore only caps requests, so a job sending short prompts can sit comfortably under the request limit and be throttled on tokens anyway. If that is happening, either lower the limit until it stops or track tokens yourself in a rolling window. The signal to watch is the proportion of calls that needed a retry: near zero means you can raise the limit, and consistently above about a tenth means you are already paying more in backoff than you gain in concurrency.

limit   wall clock   retries   what it means
    2       14m 20s         0      under-using the quota
    8        3m 40s         3      about right
   32        3m 10s       210      mostly backoff; barely faster, far noisier
  128        9m 50s     1,400+     slower than 8, and the provider notices

When async is not the answer#

Two other cases are worth recognising. If the work must happen in a fixed order, or each step needs the previous result, there is nothing to overlap and a plain loop is clearer. And if the job is large enough to run for hours, or needs to survive a machine restart, a queue with durable workers is the right shape rather than one script holding all the state in memory; the same batch of calls then restarts on a different machine without losing anything. Several providers also offer a batch endpoint that takes the whole job at a large discount and returns it within a day, which beats any amount of concurrency tuning when the answer is not needed now.

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.