Python Concurrency and Async for Interviews

In this lesson, we’ll work through the small, specific slice that most real-world concurrency comes down to. Four things:

• what the global interpreter lock (GIL) actually prevents,
• when to use threads versus processes versus asyncio,
• how each concurrency primitive works and which bug it exists to prevent,
• how to size a worker pool from a latency number instead of guessing.

We’ll build everything around one problem that shows up in many forms: call a slow, rate-limited, occasionally-failing service many times, quickly, without overloading it. By the end you’ll be able to write that program, find the four bugs it usually ships with, and size it with arithmetic.

Every program in this chapter has the same shape:

• In: a list of work items: 100 URLs, 500 record IDs, 40 files.
• Out: a list of results in a defined order, plus a defined answer to “what happened to the ones that failed.”

Concurrency changes how long that takes and how many things are in flight at once. It does not change what goes in or what comes out. A solution that quietly changes the output (drops failures, reorders results, returns partial data as if it were complete) has not solved the problem faster. It has solved a different problem.

What you need to know before reading this

You can write Python: functions, classes, exceptions, for loops, list comprehensions. You do not need prior experience with asyncio, threads, or multiprocessing.

Five words this chapter uses constantly

• Concurrency is dealing with many things at once: the program has 10 requests outstanding and handles whichever answer comes back first. It is about structure.
• Parallelism is doing many things at once: 8 CPU cores each executing your code simultaneously. It is about hardware.
• Blocking means a line of code that does nothing until something external finishes: time.sleep(1), a database query, an HTTP request.
• I/O (“input/output”) is any of those external things: network, disk, database. Code that spends its time on I/O is spending its time waiting.
• Latency is how long one operation takes, end to end. Throughput is how many operations finish per second. They are different numbers and move independently: a service can have 200 ms latency and 1,000 requests/second of throughput at the same time, if it handles 200 requests concurrently. The sizing section below is entirely about the relationship between them.

Blocking is what all of this exists to manage. A program waiting on the network is idle, and the goal is to get it doing something else during that idle time.

Concurrency without parallelism is the common and useful case, and it is what most of this chapter is about: one CPU core, a hundred outstanding network requests, all waiting at the same time instead of one after another.

The GIL, stated precisely

Almost every wrong answer in this area comes from getting the lock’s scope wrong, so pin the scope down first.

CPython (the standard Python interpreter) has a global interpreter lock (GIL): a single lock a thread must hold to execute Python bytecode (the low-level instructions your source is compiled into, which the interpreter actually steps through). One lock, one interpreter, so exactly one thread runs Python code at any instant, no matter how many cores you own.

The common wrong conclusion is “so Python threads are useless.” Here is why that is wrong, and the reason is the whole point:

A thread releases the GIL while it waits on I/O.

When a thread calls into the operating system to read a socket, it drops the lock before blocking and re-acquires it when the data arrives. So while thread A is waiting on the network, thread B holds the GIL and runs. Ten threads waiting on ten HTTP requests are all genuinely waiting at the same time.

I/O-bound versus CPU-bound

A workload is I/O-bound when the thing limiting it is waiting on the outside world, and CPU-bound when the thing limiting it is your processor doing arithmetic. This one distinction drives every later decision.

“I/O-bound or CPU-bound?” is the first question to ask about any workload. Answering it wrong makes the rest of a good design worthless.

Here is why threads lose on the CPU-bound row, concretely. Say four tasks each need one second of pure Python arithmetic.

• Serial: 1 + 1 + 1 + 1 = 4 seconds.
• Four threads: the GIL lets exactly one of them run at a time, so the same 4 seconds of work still has to happen one second at a time: 4 seconds, plus the cost of the interpreter switching between the threads every few milliseconds.

So you get the same total runtime plus context-switching overhead: measurably worse, not merely no better. The measurement further down confirms it.

Two footnotes worth having:

• Libraries written in C can release the GIL themselves. NumPy drops it around a big matrix multiply, so numeric code sometimes threads better than the rule predicts. This is a property of the library, not of Python.
• The GIL is an implementation detail of CPython, not of the language. Recent CPython versions ship an experimental build with no GIL at all. Nothing in this chapter depends on that; the sizing arithmetic in the sizing section is what survives either way.

Threads, processes, and async — choosing between them

Threads, processes, and asyncio all let you have many things in flight. They differ in what “a thing” is and who decides when to switch between them.

The decision tree

One question sorts every workload into one of three landing places.

flowchart TD
    Q{"Is the work waiting,<br/>or computing?"}
    Q -->|"waiting: network, disk"| IO{"How many at once?"}
    Q -->|"computing: parse, hash, resize"| CPU["multiprocessing<br/>one process per core"]
    IO -->|"hundreds or more"| ASY["asyncio<br/>one thread, one event loop"]
    IO -->|"a few dozen, or<br/>the library is blocking"| THR["ThreadPoolExecutor"]

The computing edge routes to multiprocessing. This work holds the GIL from start to finish. Separate processes have separate interpreters, and therefore separate GILs. That is the only way to use more than one core for Python code.

The waiting edge splits again on scale. This work spends its time idle, so the question becomes how many idle things you need at once.

Hundreds or more points to asyncio. The reason is memory per unit of concurrency. A coroutine (the object an async def function produces, defined properly in the event-loop section below) costs a few hundred bytes. A thread gets its own operating-system stack, with megabytes of address space reserved for it. Ten thousand coroutines is ordinary. Ten thousand threads is not.

A few dozen, or a blocking library, points to ThreadPoolExecutor. Threads work with unmodified blocking code. Async cannot: to use asyncio you need a library that was written for it, and if all you have is requests, threads are your answer.

Who decides when to switch

This is the deeper difference between threads and async, and it drives which bug you will spend your time on.

First, one term. A race condition is a bug whose outcome depends on the order in which concurrent operations happen to interleave. The classic one is two threads both doing counter += 1 and the counter only going up by one; the primitives section forces it to happen and shows the fix.

• Threads are preemptive: the interpreter can take control away from a thread whenever it likes. It can switch between almost any two bytecodes, including in the middle of counter += 1. You get concurrency for free and race conditions for free.
• Async is cooperative: a coroutine keeps the CPU until it volunteers to give it up. It runs until it hits await (the keyword meaning “suspend here and let something else run”, covered below), and only then may yield.

Because async switches only at points you can see in the source, the classic shared-state race mostly disappears. It is replaced by a different failure: one coroutine that never awaits blocks everything (bug 4 in the bugs section).

In short: threads give you concurrency you did not ask for and races you have to defend against. Async gives you concurrency only where you wrote await, which is safer to reason about and unforgiving if you block the loop. Processes give you real parallelism and make you pay for it by copying data between address spaces.

Measuring the CPU-bound claim

The claim that threads do not help CPU work is worth confirming with a measurement instead of taking on faith.

The trick is what to measure. Comparing threaded wall-clock against serial wall-clock is unreliable on a shared machine. A busy box does not slow a one-thread run and a four-thread run by the same amount. So measure two clocks inside the same run:

• time.perf_counter(): wall-clock time, the seconds a person would count on a stopwatch.
• time.process_time(): CPU time, summed across every thread of this process. Four threads genuinely running for one second of wall-clock burn four seconds of CPU time.

Divide CPU time by wall time and you get the parallelism actually achieved. Real parallelism across four threads gives a ratio near 4. The GIL forbids that, so the ratio sits near 1.

import time
from concurrent.futures import ThreadPoolExecutor

def burn(n):
    """Pure Python arithmetic: holds the GIL the entire time."""
    total = 0
    for i in range(n):
        total += i * i
    return total

N, TASKS, ROUNDS = 200_000, 4, 5

def best_of(fn):
    """Fastest of ROUNDS runs, measured two ways: wall-clock elapsed, and CPU
    time charged across all threads of this process. Comparing wall against CPU
    within one run is safe on a loaded box; comparing two runs' wall-clocks is
    not, because the machine does not slow a 1-thread and a 4-thread run alike."""
    best, out = (float("inf"), float("inf")), None
    for _ in range(ROUNDS):
        w0, c0 = time.perf_counter(), time.process_time()
        out = fn()
        elapsed = (time.perf_counter() - w0, time.process_time() - c0)
        best = min(best, elapsed, key=lambda pair: pair[0])
    return out, best

serial, (serial_wall, _) = best_of(lambda: [burn(N) for _ in range(TASKS)])

def threaded_run():
    with ThreadPoolExecutor(max_workers=TASKS) as pool:
        return list(pool.map(burn, [N] * TASKS))

threaded, (threaded_wall, threaded_cpu) = best_of(threaded_run)

assert serial == threaded                   # sanity check: burn() is deterministic

# CPU time / wall time IS the parallelism achieved. Under the GIL only one thread
# holds the interpreter at a time, so the ratio sits near 1, and both numbers
# come from the same run, so a loaded machine slows them together.
achieved_parallelism = threaded_cpu / threaded_wall
assert achieved_parallelism < 2.0, (
    f"threads achieved {achieved_parallelism:.1f}x parallelism on CPU-bound "
    f"Python, which the GIL forbids"
)
# The consequence a reader came for: four threads did not beat one.
assert threaded_wall > serial_wall / 2

The assertions are deliberately loose (timing on a shared machine is noisy) but strong enough to catch the thing that would be surprising: a 4× speedup. You will not get one, and knowing why before you run it is the point.

asyncio: the event loop in one picture

The measurement above closed the CPU case; the I/O case belongs to asyncio. What actually happens when you write await? Answer that precisely and the failure modes in the bugs section become predictable instead of mysterious.

Four terms, then the mechanism

• A coroutine is what async def creates, and calling it does not run it. It returns a coroutine object, the way calling a generator function returns a generator instead of running the body. This is the most common beginner surprise, and bug 1 below turns it into a bug.
• await means “suspend me here and let something else run until this finishes.” It is the only place a coroutine can yield control.
• The event loop is a single-threaded scheduler holding a queue of ready coroutines. It runs one until that coroutine awaits something unfinished, then picks the next.
• A Task is a coroutine handed to the loop to run concurrently, created by asyncio.create_task(...).

That last one carries the most weight, so state the difference explicitly:

• await coro means “run this now and wait for it.” Nothing else of yours starts until it finishes.
• asyncio.create_task(coro) means “start this, I’ll collect it later.” It returns immediately; the coroutine is now running alongside you.

That distinction is the difference between concurrent and sequential code, and getting it wrong is bug 2 below.

The same three fetches, drawn twice

Three fetches, each taking 100 ms, drawn as a timeline. The top half awaits each one as it is created; the bottom half creates all three as Tasks first, then awaits them. Same work, same results, different total.

SEQUENTIAL — await each one as you create it
  t=0ms    start A ....................... done at 100
  t=100    start B ....................... done at 200
  t=200    start C ....................... done at 300      total 300 ms

CONCURRENT — create all three as Tasks, then await them
  t=0ms    start A, B, C  (all three now waiting on the network)
  t=100    all three done                                    total 100 ms

Nothing ran in parallel. One thread, one core, the whole time. The gain is that the 100 ms of waiting overlapped instead of stacking: during A’s wait the loop had nothing else to do, so it started B, then C.

That is the entire value proposition of async. It also explains why async does nothing for CPU-bound work: if A is computing instead of waiting, it never reaches an await, so the loop never gets a chance to start B.

The same thing, as a runnable block

The code below uses 50 ms instead of 100 ms so the block runs quickly. The shape is identical: 3 × 50 = 150 ms stacked, against roughly 50 ms overlapped. The assertion bounds (> 0.14 and < 0.10) leave room for timer granularity and loop overhead; a timing assertion that demands the exact expected value is a flaky test.

import asyncio, time

async def fetch(name, delay=0.05):
    await asyncio.sleep(delay)            # stands in for a network round trip
    return name.upper()

async def sequential():
    return [await fetch(n) for n in ("a", "b", "c")]

async def concurrent():
    tasks = [asyncio.create_task(fetch(n)) for n in ("a", "b", "c")]
    return [await t for t in tasks]       # all three already running

async def main():
    t0 = time.perf_counter()
    seq = await sequential()
    seq_s = time.perf_counter() - t0

    t0 = time.perf_counter()
    con = await concurrent()
    con_s = time.perf_counter() - t0
    return seq, seq_s, con, con_s

seq, seq_s, con, con_s = asyncio.run(main())

assert seq == con == ["A", "B", "C"]      # identical output, both ways
assert seq_s > 0.14                       # three 50ms waits, stacked
assert con_s < 0.10                       # three 50ms waits, overlapped

The assertions prove more than a speedup: the outputs are identical and in the same order. Concurrency did not reorder anything, because [await t for t in tasks] collects results in the order the tasks were created, not the order they finished. Only the elapsed time changed.

That order-preserving property is what you want, and bug 3 below is what happens when you lose it by accident.

asyncio.gather, and how it handles failure

Writing create_task in a list comprehension and then awaiting each task is the long form. asyncio.gather is the shorthand for the same thing: “run all of these concurrently and give me the results in argument order.”

import asyncio

async def fetch(name, delay=0.01):
    await asyncio.sleep(delay)
    return name.upper()

async def main():
    return await asyncio.gather(*(fetch(n) for n in ("a", "b", "c")))

assert asyncio.run(main()) == ["A", "B", "C"]     # order follows the arguments

Two things about gather matter, and both are about failure instead of success.

The first concerns what happens when one of them raises. By default, the first exception propagates out of gather immediately, while the other tasks keep running in the background. That is how you end up with work still executing after the function that started it has already raised, and with 99 good results thrown away because of one bad URL.

Passing return_exceptions=True changes that: exceptions come back as ordinary items in the result list, in the same position as the input that produced them. For a batch job that is usually what you want. [result, result, ConnectionError(...), result] is a list you can act on; a raised exception and four discarded answers is not.

The second is that gather starts everything at once. All 500 of them, immediately. That is exactly the wrong thing to do to a rate-limited service, and bounding it is the subject of the sizing section.

The primitives, and the bug each one prevents

The task is rarely “what is a semaphore.” It is “here is a program that is wrong; fix it.” So learn each of the five synchronisation primitives by the failure it exists to stop.

Here is the whole set at a glance. The middle column is the one to memorise: if you can recall the bug, the primitive comes with it.

The Lock, and the race it prevents

counter += 1 looks like one operation. It is three:

1. read the current value into a temporary,
2. add one to the temporary,
3. write the temporary back.

A thread switch between step 1 and step 3 loses an update, because the second thread’s write is then overwritten by the first thread’s stale value. This is the canonical race.

Two notes before the code. counter = [0] is a one-element list instead of a plain integer because both threads need to point at the same mutable box; rebinding a plain counter = counter + 1 inside a function would only touch that function’s local name. And a threading.Event here is being used as a starting gun: has_read.wait() parks a thread until some other thread calls has_read.set().

The first half of the block forces the losing interleaving on purpose. The second half adds the lock. Compare the two counter values at the end.

import threading

# ---- the lost update, forced to happen -----------------------------------
# Rather than run millions of increments and hope the scheduler interleaves
# them, drive the exact interleaving with two events. This is the SAME bug you
# get by accident; here it is made to happen every single run.
counter = [0]
has_read = threading.Event()
other_finished = threading.Event()

def slow_thread():
    tmp = counter[0]            # 1. reads 0
    has_read.set()              #    the other thread is now allowed to run
    other_finished.wait()       #    ...and completes an entire increment here
    counter[0] = tmp + 1        # 2. writes 0 + 1 = 1, erasing that increment

def fast_thread():
    has_read.wait()             # begin only after the slow thread has read
    counter[0] += 1             # a full read-add-write: 0 -> 1
    other_finished.set()

a = threading.Thread(target=slow_thread)
b = threading.Thread(target=fast_thread)
a.start(); b.start(); a.join(); b.join()

# Two threads each added 1. The counter says 1, not 2. One update is gone, and
# nothing raised, nothing logged, nothing retried.
assert counter[0] == 1, "expected exactly one lost update"

# ---- the lock closes the window ------------------------------------------
def increment_safe(counter, n, lock):
    for _ in range(n):
        with lock:                   # the whole read-add-write is now atomic
            counter[0] += 1

N, THREADS = 50_000, 4
lock = threading.Lock()
safe = [0]
ts = [threading.Thread(target=increment_safe, args=(safe, N, lock)) for _ in range(THREADS)]
for t in ts: t.start()
for t in ts: t.join()
assert safe[0] == N * THREADS, "a lock must not lose updates"

The first half proves something stark: two threads each added one to a counter that started at zero, and it ended at one. No exception, no warning, no log line. An increment simply evaporated.

The lock in the second half fixes it by making the read, the add and the write inseparable: while one thread holds the lock, no other thread can slip between its steps.

Why the demonstration has to force the interleaving

That the code above uses events to drive the bad schedule is itself the lesson.

Left to the scheduler, the same unguarded code usually produces the right answer. Run four threads doing fifty thousand unguarded increments each and you will get exactly 200,000 most of the time.

A race condition is not a bug that happens. It is a bug that is permitted to happen, and it will pick production to exercise that permission. “It passed when I ran it” is worth nothing here.

That has a direct consequence for how you test this code, because an assertion that cannot fail is not a test. Asserting unsafe[0] <= N * THREADS looks like a race check. It is not. A lost update makes the counter smaller, never larger, so that assertion passes on correct code, on broken code, and on code that does nothing whatsoever. Either force the schedule, as above, or assert on an invariant recorded during the run, the technique in the testing section.

The Semaphore, the one this chapter is built on

Bounded concurrency is the answer to the rate-limit problem at the centre of the chapter, so this is the primitive to know cold.

A semaphore is a counter of permits. Semaphore(10) starts with ten. Every async with sem takes one and every exit returns one; the eleventh acquirer waits until somebody releases. It is a lock that admits N holders instead of one, and Semaphore(1) is just a lock.

In the block below, 50 tasks are launched at once against a Semaphore(5). in_flight counts how many are inside the semaphore right now, and peak records the highest that count ever reached. The assertion on peak is the one carrying the claim.

import asyncio

async def limited_worker(i, sem, in_flight, peak):
    async with sem:                             # blocks past the Nth holder
        in_flight[0] += 1
        peak[0] = max(peak[0], in_flight[0])
        await asyncio.sleep(0.01)               # the "request"
        in_flight[0] -= 1
        return i

async def main():
    sem = asyncio.Semaphore(5)
    in_flight, peak = [0], [0]
    results = await asyncio.gather(
        *(limited_worker(i, sem, in_flight, peak) for i in range(50))
    )
    return results, peak[0]

results, peak = asyncio.run(main())
assert results == list(range(50))               # all 50 done, in order
assert peak <= 5, f"semaphore let {peak} through, limit was 5"

Fifty tasks were created at once and all fifty completed, but never more than five were in flight.

One detail is worth pausing on. in_flight[0] += 1 is the exact read-add-write that lost an update three code blocks ago, and here it needs no lock. The reason is the one from the choosing section: this is async, so a switch can only happen at an await, and there is no await between the read and the write. The interpreter is not allowed to interrupt in the middle.

Write that same increment in the threaded version of this code and it is a race again. The primitive you need depends on the concurrency model, and this is the concrete example of why.

The Queue, and the problem a semaphore does not solve

A semaphore limits how many things run at once. A bounded queue limits how many are allowed to be waiting.

That distinction is called backpressure: when a system is saturated, the pressure has to propagate back to whoever is generating work, so they slow down too.

It matters when work arrives faster than you can do it:

• With a semaphore only: the extra tasks all exist; they are just parked. 100,000 pending coroutines sit in memory until the process dies.
• With a bounded queue: await q.put(item) suspends the producer once the queue is full. The producer literally cannot get ahead, so memory stays bounded.

Three things to read in the block below. maxsize=3 is the bound. None is a sentinel (a marker item meaning “no more work, shut down”) and there is one per consumer, so each consumer gets its own stop signal. q.task_done() marks an item as finished; it pairs with q.join() if you want to wait for a queue to drain (this block instead waits on the tasks themselves).

import asyncio

async def producer(q, n, log):
    for i in range(n):
        await q.put(i)                  # BLOCKS once the queue is full
        log.append(("put", i, q.qsize()))
    for _ in range(2):
        await q.put(None)               # one sentinel per consumer

async def consumer(q, done):
    while True:
        item = await q.get()
        if item is None:
            q.task_done()
            return
        await asyncio.sleep(0.001)      # the slow work
        done.append(item)
        q.task_done()

async def main():
    q = asyncio.Queue(maxsize=3)        # at most 3 items may WAIT
    log, done = [], []
    await asyncio.gather(producer(q, 20, log), consumer(q, done), consumer(q, done))
    return log, done, q.qsize()

log, done, left = asyncio.run(main())
assert sorted(done) == list(range(20))          # every item handled exactly once
assert left == 0                                # queue drained
assert max(size for _, _, size in log) <= 3, "maxsize was not enforced"

The maxsize=3 bound is the whole point. The producer wants to enqueue all 20 items immediately and cannot: once three are waiting, await q.put(i) suspends until a consumer takes one. Memory is bounded by the queue, not by how fast the producer runs.

Now the part that makes this a production bug instead of a textbook one: delete the maxsize and every assertion except the last one still passes. The output is correct either way. The only difference shows up under sustained load, in a memory graph, at 3 a.m.

Event and Condition

These two come up less often, but you should be able to name them.

An Event is a one-way flag. await event.wait() blocks every waiter until someone calls event.set(), and then they all proceed. Use it to hold workers back until initialisation finishes. Once set, it stays set.

A Condition is an Event you can re-arm: it is for waiting on a shared state to change, instead of on a one-time milestone. It exists so a worker can sleep until notified instead of spinning on while not ready: pass, which burns a full core doing nothing.

The four bugs this program always ships with

Knowing the primitives is no protection here: nearly every first draft of the fetch-many program ships the same four failures, and none of them is a missing lock. Each appears below as wrong code beside right code, because the skill that matters is recognising them by shape.

Bug 1 — calling a coroutine without awaiting it

You write record(log, i) instead of await record(log, i). The coroutine object gets created and immediately thrown away. It does not run, it does not raise, and it does nothing at all.

Python does emit a RuntimeWarning: coroutine ... was never awaited for this, but a warning is not an error: it goes to stderr, it does not fail a test, and in a noisy log you will not see it. The block below wraps the wrong version in warnings.catch_warnings() purely to silence that warning so the block’s own output stays clean. The assertion after it is the point.

import asyncio

async def record(log, item):
    log.append(item)

async def wrong(log):
    for i in range(3):
        record(log, i)              # ✗ creates a coroutine object and drops it
    return log

async def right(log):
    for i in range(3):
        await record(log, i)        # ✓ actually runs it
    return log

import warnings
with warnings.catch_warnings():
    warnings.simplefilter("ignore")          # silence "never awaited"
    assert asyncio.run(wrong([])) == []      # silently did nothing at all
assert asyncio.run(right([])) == [0, 1, 2]

Bug 2 — awaiting inside the loop

This is sequential code in async syntax. It is the most common performance bug in async Python, and because it produces correct output, no test catches it. Only a stopwatch does.

for url in urls:
    results.append(await fetch(url))     # ✗ each await finishes before the next starts

results = await asyncio.gather(*(fetch(u) for u in urls))   # ✓ all in flight

await means wait here. Writing it inside the loop means you wait for each one before starting the next, so 100 URLs at 200 ms each take 20 seconds, exactly as if you had never written async at all. You have a for loop with extra ceremony.

Concurrency requires creating all the awaitables first (as Tasks, or via gather) and then awaiting the collection.

Bug 3 — losing result order, or losing failures entirely

Two separate ways to silently change your output.

The first is order. asyncio.as_completed yields results in completion order, not input order. That is exactly what you want when streaming results to a user as they arrive, and it is a silent corruption when the caller expected result i to correspond to input i.

The second is failures. A bare gather raises on the first exception and discards the other results, as the event-loop section described. The three-line fix below keeps everything and lets the caller sort it out.

results = await asyncio.gather(*tasks, return_exceptions=True)
ok      = [r for r in results if not isinstance(r, Exception)]
failed  = [r for r in results if isinstance(r, Exception)]

The contract, then: results in input order, failures returned instead of raised, and the caller can split them into both lists. That is exactly the contract the assembled program implements. The important part is noticing that there is a decision here at all; either policy is defensible as long as it is documented.

Bug 4 — blocking the event loop

One synchronous call inside a coroutine freezes everything. There is one thread, and async is cooperative, so nothing can take control back from a coroutine that refuses to await.

The fragment below shows the two flavours of the bug and the three fixes.

The third fix is the one that needs a word. run_in_executor hands a plain blocking function off to a separate pool and gives you back something you can await, so the loop is free to run other coroutines meanwhile. Passing None as the first argument means “use the default thread pool”, which is right for blocking I/O. For work that is genuinely CPU-bound, pass a ProcessPoolExecutor instead, for the GIL reason: a thread would still be holding the GIL.

async def handler():
    time.sleep(1)                   # ✗ the whole loop stops for one second
    data = requests.get(url)        # ✗ same: a blocking library in async code

async def handler():
    await asyncio.sleep(1)          # ✓ yields to the loop
    data = await client.get(url)    # ✓ an async-native client
    rows = await asyncio.get_running_loop().run_in_executor(
        None, parse_big_csv, path   # ✓ CPU-bound work pushed to a thread
    )

The symptom is a program that gets slower as you add concurrency, while exactly one CPU core sits pegged at 100%. One core at 100% means real work is happening on the loop thread; more concurrency making it worse means the extra tasks are queued behind that work instead of overlapping with it.

Sizing the pool with arithmetic, not guesswork

Every semaphore so far has taken its limit as a given. Picking that limit from numbers you can measure is what separates a real answer from “I’d tune it,” and it needs a single result: Little’s Law.

The law, and what each letter means

L  =  λ · W

L = average number of requests in flight  (the pool size you are solving for)
λ = throughput, requests completed per second  (lambda, the arrival rate)
W = average latency of one request, in seconds

“Requests per second” is also written QPS (queries per second); the code below uses target_qps for λ. The units matter: W is in seconds, so 200 ms must go in as 0.2, not 200.

The estimate in one line

Suppose your service takes 0.2 seconds per request and you want to finish 50 requests every second. In any one-second window, 50 requests each spend 0.2 s inside your system: 50 × 0.2 = 10 request-seconds of work packed into one second of clock, which means about 10 requests are in progress at any instant. That is L.

To sustain a target throughput λ at measured latency W, you need λ · W concurrent slots. Round any fractional answer up. You cannot have 0.6 of a worker.

Read the second row against the first. Nothing about your goal changed (only the service got slower) and your pool has to grow 2.5×. Concurrency is not a property of your program alone. It is a property of your program and the thing it calls. A hard-coded worker count silently becomes the wrong number the moment a dependency degrades.

Running it backwards: the ceiling a pool imposes

Divide instead of multiply and the same equation tells you the most a given pool can do. With L slots and latency W:

max throughput = L / W = 10 / 0.2 = 50 req/s

So 10 slots against a 200 ms service can never exceed 50 req/s, no matter how much work you queue behind it. This is not a second law; it is L = λ · W rearranged.

Now put your ceiling next to the provider’s published rate limit. If a rate limit and your pool disagree, the smaller one wins and the other is decoration. A pool sized for 50 req/s pointed at a service that allows 30 req/s does not get 50; it gets 30, plus a stream of rejections. Size the semaphore from Little’s Law at the latency you measure, clamp it to the provider’s limit, and make it a config value instead of a constant, because the right number moves when their latency moves.

def slots_needed(target_qps, latency_s):
    """Little's Law: L = lambda * W, rounded up to whole workers."""
    import math
    return math.ceil(target_qps * latency_s)

def max_throughput(slots, latency_s):
    return slots / latency_s

assert slots_needed(50, 0.200) == 10
assert slots_needed(50, 0.500) == 25          # slower service, 2.5x the pool
assert slots_needed(200, 0.200) == 40
assert slots_needed(50, 0.020) == 1

assert max_throughput(10, 0.200) == 50.0      # the ceiling 10 slots imposes
assert max_throughput(25, 0.500) == 50.0      # same target, slower service

# The provider's limit is the real ceiling when it is lower than yours.
PROVIDER_LIMIT_QPS = 30
assert min(max_throughput(10, 0.200), PROVIDER_LIMIT_QPS) == 30

Testing concurrent code without flakiness

“It passed” is the weakest possible evidence about a race. The primitives section made the point directly: the unsafe counter usually produces the right answer, so a test that runs the code and checks the output passes on broken code most of the time. What you want is a test that fails reliably when the concurrency is wrong. Three techniques do that.

Assert on the invariant, not the outcome

An invariant is something that must be true at every instant while the code runs, not just at the end. Do not test “the result is correct.” Record the invariant during the run and assert on the recording.

The semaphore test earlier is exactly this. It tracks peak in-flight as the tasks execute and asserts it never exceeded the limit. The difference is what happens when you break the code:

• Delete the semaphore → peak jumps to 50 → the peak <= 5 assertion fails, every run, deterministically.
• Delete the semaphore → the returned results are still list(range(50)) → an output test passes, every run.

Make the schedule deterministic

Replace real I/O with a fake whose timing and failures you control, so “slow” and “fails twice” are inputs instead of accidents.

FakeService below takes two knobs: latency (how long each call sleeps) and fail_first (how many of the first N calls raise ConnectionError). It also records peak concurrency for technique 1. The one subtlety is that n = self.calls is captured before the await: after an await, other coroutines have run and self.calls has moved, so the check has to use the value from this call’s own turn.

import asyncio

class FakeService:
    """A stand-in whose latency and failures are inputs, not accidents."""

    def __init__(self, latency=0.001, fail_first=0):
        self.latency = latency
        self.fail_first = fail_first        # fail this many calls, then succeed
        self.calls = 0
        self.concurrent = 0
        self.peak = 0

    async def get(self, key):
        self.calls += 1
        n = self.calls          # capture BEFORE awaiting: self.calls moves under us
        self.concurrent += 1
        self.peak = max(self.peak, self.concurrent)
        try:
            await asyncio.sleep(self.latency)
            if n <= self.fail_first:
                raise ConnectionError(f"transient failure on call {self.calls}")
            return f"value-for-{key}"
        finally:
            self.concurrent -= 1

async def probe():
    svc = FakeService(fail_first=2)
    out = []
    for i in range(4):
        try:
            out.append(await svc.get(i))
        except ConnectionError as e:
            out.append(str(e))
    return out, svc.calls

out, calls = asyncio.run(probe())
assert calls == 4
assert out[0].startswith("transient failure") and out[1].startswith("transient failure")
assert out[2] == "value-for-2" and out[3] == "value-for-3"

The fake makes failure a parameter. fail_first=2 means calls 1 and 2 raise and calls 3 and 4 succeed, which is precisely what the four assertions check, and it lets a retry policy be tested in milliseconds without waiting for a real service to have a bad day.

Test the failure path explicitly

The bug in production is nearly always in the path you did not exercise. For this program that means:

• what happens on the third retry, when the budget runs out,
• what happens when the semaphore is full and a task is cancelled,
• what happens when one item in a batch of 100 fails.

Those are the tests worth writing, and the assembled program below writes all three.

The whole thing, assembled

The problem, in the form it is usually posed: fetch many keys from a service that is slow, rate-limited, and sometimes fails transiently. Be fast. Do not lose failures.

Every decision below traces to an earlier section. The contract comes before the code, because a concurrent function whose contract is implicit is a concurrent function nobody can safely call.

Contract. Returns one result per input key, in input order. Successes are values; failures are the final exception object, returned instead of raised, so the caller can split the list into successes and failures, and nothing is ever thrown away. Never more than limit requests in flight. Each key is retried up to retries times on a transient ConnectionError, with exponential backoff; any other exception is returned immediately without retrying.

The pieces fit together like this:

flowchart TD
    A["fetch_all(keys)"] --> B["gather every key concurrently<br/>return_exceptions=True"]
    B --> C["fetch_one(key)"]
    C --> D["acquire semaphore permit<br/>(bounds concurrency)"]
    D --> E["await service.get(key)"]
    E -->|"ConnectionError"| F{"retries left?"}
    F -->|"yes"| G["release permit,<br/>sleep base·2^attempt,<br/>retry"]
    G --> D
    F -->|"no"| H["return the last error"]
    E -->|"success"| I["return the value"]
    H --> J["one entry per key,<br/>in input order"]
    I --> J

Two terms in that contract

Exponential backoff means each retry waits longer than the last, doubling: base_delay × 2**attempt. With base_delay = 1s that is a wait of 1s, then 2s, then 4s. The alternative (a fixed delay, or no delay) is what causes a thundering herd: a thousand clients all fail at the same moment, all retry one second later at the same moment, and their synchronised second wave knocks the service over again exactly as it was recovering. Doubling spreads the waves out.

Transient means an error that might succeed if you try again:

• ConnectionError: a reset socket, a dropped connection. Retrying might work, so retry it.
• ValueError("400 malformed request"): your request is wrong and will be wrong every time. Retrying three times wastes the budget, adds latency, and delivers the same error later, so return it immediately.

Deciding which errors are worth retrying is a real design decision; “retry everything” is the wrong answer.

The code

The block is long, so here is its shape first.

• fetch_one handles one key: acquire the semaphore, call the service, catch ConnectionError, sleep, try again. It returns the last exception instead of raising it.
• fetch_all handles the batch: create the semaphore once, gather every key with return_exceptions=True.
• FakeService is the fake from the testing section, repeated so the block runs standalone.
• Then five tests, one per contract clause: happy path, retried failures, permanent failure, non-retryable error, and the backoff-placement timing check.

The one line to look hardest at is where await asyncio.sleep(...) sits relative to async with sem. It is outside, deliberately, and detail 1 below explains what that buys.

import asyncio, time

async def fetch_one(svc, key, sem, retries=3, base_delay=0.001):
    """One key: bounded by the semaphore, retried with exponential backoff."""
    last = None
    for attempt in range(retries + 1):
        async with sem:                       # bound the concurrency
            try:
                return await svc.get(key)
            except ConnectionError as e:
                last = e                      # transient -- worth retrying
        if attempt < retries:
            # Backoff OUTSIDE the semaphore: holding a permit while sleeping
            # would throttle everyone else for the duration of our retry.
            await asyncio.sleep(base_delay * (2 ** attempt))
    return last                               # exhausted: return the error, don't raise

async def fetch_all(svc, keys, limit=5, base_delay=0.001):
    sem = asyncio.Semaphore(limit)            # in production, sized from Little's Law
    return await asyncio.gather(              # everything in flight at once
        *(fetch_one(svc, k, sem, base_delay=base_delay) for k in keys),
        return_exceptions=True,               # safety net for non-ConnectionError
    )


class FakeService:
    def __init__(self, latency=0.001, fail_first=0):
        self.latency, self.fail_first = latency, fail_first
        self.calls = self.concurrent = self.peak = 0

    async def get(self, key):
        self.calls += 1
        n = self.calls          # capture BEFORE awaiting: self.calls moves under us
        self.concurrent += 1
        self.peak = max(self.peak, self.concurrent)
        try:
            await asyncio.sleep(self.latency)
            if n <= self.fail_first:
                raise ConnectionError("transient failure on call %d" % n)
            return "value-for-%s" % key
        finally:
            self.concurrent -= 1


# --- the happy path: order preserved, concurrency bounded ------------------
svc = FakeService()
out = asyncio.run(fetch_all(svc, list(range(20)), limit=5))
assert out == ["value-for-%d" % i for i in range(20)]   # INPUT order, not completion
assert svc.peak == 5, "the semaphore did not hold: peak was %d" % svc.peak
assert svc.calls == 20                                   # no retries needed

# --- transient failures are retried, not surfaced -------------------------
# fail_first counts CALLS, not keys: calls 1-3 fail whichever keys they belong to,
# so it is 5 first attempts + 3 retries = 8, regardless of scheduling order.
svc = FakeService(fail_first=3)
out = asyncio.run(fetch_all(svc, list(range(5)), limit=2))
assert all(isinstance(r, str) for r in out), "retries should have absorbed these"
assert svc.calls == 8
assert svc.peak == 2

# --- a permanent failure is RETURNED, never lost --------------------------
class AlwaysFails(FakeService):
    async def get(self, key):
        await asyncio.sleep(0.001)
        raise ConnectionError("permanent")

out = asyncio.run(fetch_all(AlwaysFails(), ["a", "b"], limit=2))
assert len(out) == 2                                     # one result per input
assert all(isinstance(r, ConnectionError) for r in out)  # returned, not raised

# --- a NON-retryable error is returned immediately, not retried -----------
# This is the line `return_exceptions=True` actually earns: fetch_one only
# catches ConnectionError, so anything else propagates out of the coroutine
# and gather would RAISE it -- discarding the other results -- without it.
class Malformed(FakeService):
    async def get(self, key):
        self.calls += 1
        await asyncio.sleep(0.001)
        raise ValueError("400 malformed request")

svc = Malformed()
out = asyncio.run(fetch_all(svc, ["a", "b"], limit=2))
assert all(isinstance(r, ValueError) for r in out)
assert svc.calls == 2, "a non-retryable error must not be retried"

# --- the backoff must NOT be held inside the semaphore --------------------
# 12 keys, 2 slots, 10ms latency, 50ms backoff. Sleeping outside the permit,
# retries overlap with other keys' work. Sleeping INSIDE, every retry blocks
# a slot that nobody can use, and wall-clock roughly triples.
svc = FakeService(latency=0.01, fail_first=4)
t0 = time.perf_counter()
asyncio.run(fetch_all(svc, list(range(12)), limit=2, base_delay=0.05))
elapsed = time.perf_counter() - t0
assert elapsed < 0.16, (
    "%.3fs -- too slow: the backoff is being held inside the semaphore" % elapsed
)

The four details that are easy to get wrong

Each is easy to get wrong in a way that still passes a naive test.

The first: the backoff sleep is outside the async with sem block. Sleeping while holding a permit means a retrying task occupies a slot it is not using, so the pool effectively shrinks exactly when the service is already struggling. This is the single most common flaw in an otherwise-correct answer.

It is also worth knowing precisely which test catches it, because the obvious one cannot. Asserting on peak is useless here: holding a permit while idle produces fewer requests in flight, never more, so peak reads 2 either way. Same for svc.calls: 16 either way. The only symptom is wall-clock, which is why the last test in the block is a timing assertion. Measured on the code above with 12 keys, 2 slots, 10 ms latency and 50 ms base delay:

Roughly 2.7× the time, with every other observable unchanged.

The second: return_exceptions=True earns its place only on the non-retryable path. Because fetch_one returns last instead of raising it, a ConnectionError never reaches gather at all. Delete the flag and the first three tests still pass. It matters when svc.get raises something fetch_one does not catch. Without the flag, gather re-raises immediately and the other results are discarded. The Malformed test is what makes the flag load-bearing instead of decorative.

The third: failures are returned, not raised. The caller gets one entry per input and decides what a failure means; this function does not decide for them. Splitting is theirs to do:

ok     = [r for r in out if not isinstance(r, Exception)]
failed = [r for r in out if isinstance(r, Exception)]

The fourth: exponential backoff, not a fixed delay, for the thundering-herd reason given above. A production version would also add jitter: a small random offset on each delay, so that clients which failed together do not retry together. It is left out of the code above only because randomness would make the timing test non-deterministic, which is itself the testing section’s point.

The same skeleton in an agent harness

The capstone above fetched keys from an anonymous service. Rename two things (the service is a model API, the keys are prompts) and it is the concurrency core of every agent harness in this track. The primitives show up in two specific places.

The first place is parallel tool calls inside the agent loop. One model response can carry several tool_use blocks at once (see the agent foundations chapter). The tools are I/O-bound (HTTP calls, file reads, database queries), which is the case the decision tree routes to asyncio. Run them concurrently with gather, then return all results in one user message; splitting them across messages is a silent 3× slowdown, and it is also bug 3 wearing agent clothes. return_exceptions=True plus zip is what keeps result i attached to call i and keeps one failed tool from discarding the others.

flowchart TD
    A["Model call"] --> B{"stop_reason<br/>== tool_use?"}
    B -->|no| C["Return the response"]
    B -->|yes| D["Run every tool_use block<br/>concurrently with gather"]
    D --> E["Append all tool_result blocks<br/>in one user message, in call order"]
    E --> A

import asyncio
import anthropic

client = anthropic.AsyncAnthropic()

async def run_tool(name: str, args: dict) -> str:
    ...  # dispatch to the real tool implementations

async def agent_turn(messages: list, tools: list):
    resp = await client.messages.create(
        model="claude-opus-5", max_tokens=4096,
        tools=tools, messages=messages,
    )
    messages.append({"role": "assistant", "content": resp.content})
    if resp.stop_reason != "tool_use":
        return resp

    calls = [b for b in resp.content if b.type == "tool_use"]
    results = await asyncio.gather(
        *(run_tool(b.name, b.input) for b in calls),
        return_exceptions=True,
    )
    messages.append({"role": "user", "content": [
        {
            "type": "tool_result",
            "tool_use_id": blk.id,
            "content": str(res),
            "is_error": isinstance(res, Exception),
        }
        for blk, res in zip(calls, results)
    ]})
    return resp

What carried over from the capstone is unchanged: a failed tool becomes a returned value (is_error: true) instead of a raised exception, because the model can read an error string and route around it, exactly as the capstone’s caller could. Raising would end the run over a fixable typo.

The second place is the bounded fan-out of model calls. An orchestrator handing briefs to workers (see the multi-agent chapter) is the capstone’s shape with the fetch replaced by an LLM call. The semaphore’s size comes from the sizing section’s closing rule: when the provider’s rate limit and your pool disagree, the smaller one wins, so the bound is the rate limit, never a CPU count, because these tasks hold no CPU while they wait.

sem = asyncio.Semaphore(4)   # the provider's rate limit, not a CPU count

async def worker(brief: str) -> str:
    async with sem:          # permit held for the request only
        resp = await client.messages.create(
            model="claude-haiku-4-5", max_tokens=1024,
            messages=[{"role": "user", "content": brief}],
        )
    return "".join(b.text for b in resp.content if b.type == "text")

async def fan_out(briefs: list) -> list:
    return await asyncio.gather(
        *(worker(b) for b in briefs),
        return_exceptions=True,   # one entry per brief, in input order
    )

Every detail from the capstone transfers: retries with backoff would sleep outside the async with sem block, gather preserves input order so brief i maps to report i with no bookkeeping, and the caller splits successes from failures itself. If briefs arrive continuously instead of as a fixed batch, that is the Queue from the primitives section, and the workers become long-lived consumers instead of one task per brief.

Deriving the numbers

If a figure in this chapter looked like it appeared from nowhere, it is derived here. (Timing constants inside the code blocks are chosen only to keep the blocks fast, and each is explained where it appears; they are the 50 ms fetch, the 0.14/0.10 bounds, maxsize=3, and limit=5.)

Conclusion

Every rule below exists because of a mechanism, not a convention. If you can recall the mechanism, you can rebuild the rule instead of memorising it.

The load-bearing takeaways: classify the work before choosing a tool; bound concurrency with a semaphore sized from Little’s Law and clamped to the provider’s limit; preserve input order and return failures instead of raising; and test on invariants recorded during the run, because a race that passes once tells you nothing.

Further reading

• Python asyncio documentation: the event loop, tasks, and gather.
• concurrent.futures documentation: ThreadPoolExecutor and ProcessPoolExecutor.
• David Beazley, “Understanding the Python GIL” (PyCon 2010): talk and slides on exactly why threads do not speed up CPU-bound Python.
• PEP 703 — Making the Global Interpreter Lock Optional in CPython: the design behind the experimental no-GIL build.
• Little’s Law: John D. C. Little, “A Proof for the Queuing Formula: L = λW,” Operations Research, 1961.