16 — Concurrency and async in Python, the way the interview actually tests it

A practical, non-LeetCode software-engineering interview tests a small, specific slice of concurrency. Four things:

• what the global interpreter lock (GIL) actually stops you doing,
• when to reach for 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.

It is built around one problem, which this style of interview keeps asking in different costumes: call a slow, rate-limited, occasionally-failing service many times, quickly, without melting it. By the end you will have written that program, found the four bugs it usually ships with, and be able to size it with arithmetic.

A common report from candidates who pass this kind of interview loop — the day of back-to-back sessions that decides an offer — is that grinding algorithm puzzles was the wrong preparation.

The problems look like work. Here is an API, here is a rate limit, here are some retries, make it fast and correct. There is no trick to spot.

What gets tested is three things: whether you know why your program is slow, whether you can name the primitive that fixes a race, and whether you notice that the obvious solution silently drops errors. That is what this chapter drills.

The shape of the thing, before any mechanism. Every program in this chapter has the same input and output.

• In goes a list of work items — 100 URLs, 500 record IDs, 40 files.
• Out comes 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.

Keep that fixed, because it is the standard by which a wrong answer gets caught. 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 to have used asyncio, threads, or multiprocessing before.

The five words this chapter uses constantly

• Concurrency is dealing with many things at once — the program has 10 requests outstanding and juggles 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 sits there doing 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 they 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. §6 is entirely about the relationship between them.

Blocking is the thing all of this exists to manage. A program waiting on the network is not busy. It is idle while pretending to be busy, and the whole game 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 of them waiting at the same time instead of one after another.

1. 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 you almost certainly run — has a global interpreter lock (GIL): a single lock that a thread must hold to execute Python bytecode (the low-level instructions your source is compiled into, and what 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 near-universal wrong conclusion is “so Python threads are useless.” That is wrong, and the reason it is wrong 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.

The one table this section exists to produce

Two terms first. 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.

“I/O-bound or CPU-bound?” is the first question to ask out loud in the interview. Every later decision follows from it, and answering it wrong makes the rest of a good answer 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. That result surprises people who have not internalised the rule above. Threads processes and async choosing between them measures it.

Two footnotes worth having, because interviewers sometimes reach for them:

• 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 Sizing the pool with arithmetic not guesswork is what survives either way.

2. 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 — and choosing among them should be a decision you can defend, not a habit.

The decision tree

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

flowchart TB
    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"]
    style CPU fill:#9d4edd,color:#fff
    style ASY fill:#2d6a4f,color:#fff
    style THR fill:#40916c,color:#fff

“Is the work waiting, or computing?” is The gil stated precisely’s question again, and it is the only one that has to be answered first.

The computing edge → 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 → split again on scale. This work spends its time idle, so the question becomes how many idle things you need at once.

Hundreds or more → asyncio. The reason is memory per unit of concurrency. A coroutine — the object an async def function produces, defined properly in Asyncio the event loop in one picture — 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 the library is blocking → 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 and the bug each one prevents 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 in Asyncio the event loop in one picture), 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 (The four bugs this program always ships with, bug 4).

Say it this way in an interview: “Threads give you concurrency you didn’t 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

“Threads don’t help CPU work” is the claim people want evidence for, so here it is as a runnable measurement.

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 the block measures 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 one by the other and you get the parallelism actually achieved. Real parallelism across four threads gives a ratio near 4. The GIL forbids that, so watch for the ratio sitting 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 to this process across all its threads.

    Taking the minimum handles ordinary noise. It does NOT make a
    cross-run comparison safe on a loaded box, which is the trap: the serial
    run has one runnable thread and the threaded run has four, so a busy
    machine does not penalise them equally, and `threaded_wall / serial_wall`
    can move in either direction for reasons that have nothing to do with the
    GIL. The assertion below therefore compares wall against CPU *within one
    run*, where both numbers come from the same four seconds of contention."""
    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)

# burn() is deterministic, so this is a sanity check only -- it proves nothing
# about threading. The timing assertion below is the one carrying the claim.
assert serial == threaded

# CPU time divided by wall time IS the parallelism achieved: four threads
# genuinely running at once burn four seconds of CPU per second of clock.
# Under the GIL only one of them holds the interpreter at a time, so the ratio
# sits at 1. Both numbers come from the same run, so a loaded machine slows
# them together and the ratio does not move -- which is what makes this safe
# on a busy CI box, and what the wall-clock version of this test got wrong.
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"
)
# And 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 they are strong enough to catch the thing that would actually be surprising: a 4× speedup. You will not get one, and knowing why before you run it is the point.

3. 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 four bugs this program always ships with become predictable rather than mysterious.

Four terms, then the mechanism

• A coroutine is what async def creates. 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 single most common beginner surprise, and The four bugs this program always ships with makes it 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 the second bug in The four bugs this program always ships with.

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 exactly why async does nothing for CPU-bound work: if A is computing rather than 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 rather than 100 ms so the block runs quickly. The shape is identical: 3 × 50 = 150 ms stacked, against roughly 50 ms overlapped.

The assertion bounds are > 0.14 and < 0.10, not > 0.15 and < 0.05, to 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 is the property you want, and The four bugs this program always ships with’s third bug is what happens when you lose it by accident.

asyncio.gather, and the two things it gets probed on

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 get probed, and both are about failure rather than success.

1. 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.

2. 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 Sizing the pool with arithmetic not guesswork.

4. The primitives, and the bug each one prevents

A primitive you cannot attach to a failure is a name you will forget — and the interview question 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, and it is worth being able to demonstrate on demand.

Two notes before the code. counter = [0] is a one-element list rather than 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.

That is the entire bug. 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, and it is the one people get wrong: an assertion that cannot fail is not a test.

Concretely: 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 — which is what Testing concurrent code without flakiness does.

The Semaphore, the one this interview actually cares about

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 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 Threads processes and async choosing between them gave: 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 fake 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"

maxsize=3 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 rather than 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, rather than 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.

5. 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 being tested 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.

Compare the two return values.

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 wearing 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.

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.

Failures. A bare gather raises on the first exception and discards the other results, as Asyncio the event loop in one picture 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)]

State the contract out loud: results are in input order, failures are returned rather than raised, and the caller can split them into both lists. That is exactly the contract The whole thing assembled implements.

An interviewer is usually more interested in whether you noticed there was a decision here than in which way you decided it.

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 stated precisely’s 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 tell is a program that gets slower as you add concurrency, with a CPU pegged at 100% of exactly one core.

That pair of symptoms is worth memorising, because it is the one you will actually be handed and asked to diagnose. 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 rather than overlapping with it.

6. 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 senior answer from “I’d tune it.”

Asked “how many workers?”, the weak answer is a number. The strong answer derives one — 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 Greek letter for 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.

Working it through once

The law is close to a tautology once you have seen a substitution.

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 start and 50 finish, and each of them spends 0.2 s inside your system. So the total time-in-system across that second is:

50 requests × 0.2 seconds each  =  10 request-seconds, spent inside 1 second of clock

Ten request-seconds of work packed into one second of clock means, on average, 10 requests are in progress at any instant. That is L. So:

L = λ · W = 50 × 0.2 = 10 slots

Stated as the rule you will use:

To sustain a target throughput λ at measured latency W, you need λ · W concurrent slots.

Four cases, worked

Each row is the same multiplication with different inputs. Round any fractional answer up — you cannot have 0.6 of a worker.

Read the second row against the first. 50 × 0.5 = 25, against 50 × 0.2 = 10. Nothing about your goal changed — only the service got slower — and your pool has to grow 2.5×.

That is the practical lesson. Concurrency is not a property of your program. 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.

The block below is those numbers as executable arithmetic. slots_needed is the law; max_throughput is the rearrangement; the last two lines are the clamp.

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

In an interview, say the last line out loud: “I’d size the semaphore from Little’s Law at the latency I measure, then clamp it to the provider’s published limit, and I’d make it a config value rather than a constant because the right number moves when their latency moves.”

7. Testing concurrent code without flakiness

“It passed” is the weakest possible evidence about a race. The primitives and the bug each one prevents 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 fix that. They are what you should reach for when asked “how would you test this?”

1. 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 in The primitives and the bug each one prevents 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.

2. Make the schedule deterministic

Replace real I/O with a fake whose timing and failures you control, so “slow” and “fails twice” are inputs rather than 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. Note n = self.calls 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.

3. 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 they are the ones a candidate usually forgets to mention. The whole thing assembled writes all three.

8. The whole thing, assembled

The problem, in the form the interview usually poses it: 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 rather than 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.

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. That distinction is the second half of the last clause, and it is a real design decision rather than boilerplate:

• ConnectionError — a reset socket, a dropped connection. Retrying might work. 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. Return it immediately.

Deciding which errors are worth retrying is itself a standard interview question, and “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 rather than raising it.
• fetch_all handles the batch: create the semaphore once, gather every key with return_exceptions=True.
• FakeService is Testing concurrent code without flakiness’s fake, 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:                       # §4: 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                               # §5 bug 3: exhausted -> return, don't raise

async def fetch_all(svc, keys, limit=5, base_delay=0.001):
    sem = asyncio.Semaphore(limit)            # §6: from slots_needed() in production
    return await asyncio.gather(              # §3: 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 the interview is looking for

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

1. 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, in principle and not just in practice: 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.

2. return_exceptions=True earns its place only on the non-retryable path. Because fetch_one returns last rather than 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 rather than decorative.

3. 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)]

4. Exponential backoff, not a fixed delay — for the thundering-herd reason given above.

Say out loud that you 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 Testing concurrent code without flakiness’s point.

The same skeleton in an agent harness

The capstone above fetched keys from an anonymous service, but 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, and both are fair game in an agentic-coding interview.

Place 1 — parallel tool calls inside the agent loop. One model response can carry several tool_use blocks at once (A worked trace). The tools are I/O-bound — HTTP calls, file reads, database queries — which is precisely the case the Threads processes and async choosing between them decision tree routes to asyncio. Run them concurrently with gather, then return all results in one user message; splitting them across messages is the silent 3× slowdown ch 01 warns about, and it is also The four bugs this program always ships with’s 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.

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

Note what carried over from the capstone unchanged: a failed tool becomes a returned value (is_error: true) rather than 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.

Place 2 — the bounded fan-out of model calls. An orchestrator handing briefs to workers (ch 06) is the capstone’s shape with the fetch replaced by an LLM call. The semaphore’s size comes from Sizing the pool with arithmetic not guesswork’s closing rule: 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 interview 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 the interviewer extends the problem — “now briefs arrive continuously” — that is the The primitives and the bug each one prevents Queue, and the workers become long-lived consumers instead of one task per brief.

9. 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 — the 50 ms fetch, the 0.14/0.10 bounds, maxsize=3, limit=5 — are chosen only to keep the blocks fast, and each is explained where it appears.)

The mechanism → rule map

Every rule on the right exists because of the mechanism on its left. If you can recall the left column, you can rebuild the right column instead of memorising it.

What interviewers probe

The question is in bold; the answer under it is what to say, compressed.

• “Will threads make this faster?” Always answer with the question: is it waiting or computing? Waiting, yes. Computing, no — the GIL is held, and you will add overhead for nothing.
• “You wrote await inside the loop. What does that do?” It makes the code sequential. The fix is to create the awaitables first and gather them.
• “How many workers?” L = λ·W from measured latency, clamped to the provider’s limit, and configurable rather than constant.
• “One of the 100 requests fails. What does your function return?” State the contract: one result per input, in input order, failures returned not raised. Any consistent answer beats not having noticed the question.
• “How would you test that the semaphore works?” Track peak in-flight during the run and assert on it. A test of the returned results passes with the semaphore deleted.
• “Your service is at 100% of one core and got slower when you raised concurrency.” Something blocking is on the event loop. Find it and move it to run_in_executor, or replace the library with an async-native one.
• “Why retry with exponential backoff rather than immediately?” Immediate retries from many clients re-create the load that caused the failure. Add jitter so the retries do not synchronise.
• “Where does your backoff sleep happen?” Outside the semaphore. Inside, a retrying task shrinks the pool for everyone precisely while the dependency is unhealthy.