Home

Async Python From the Event Loop Up

• 34 min read
Table of contents

Every example here was executed, and every block of output is the real thing — including the run where 5,000 OS threads refused to start and 50,000 coroutines didn't blink.

Asynchronous programming has a reputation for being hard, and most of that reputation is earned in the gap between knowing the syntax and knowing what the runtime does with it. You can memorise async and await in ten minutes. Understanding why adding them to the wrong function makes your program slower takes longer.

This article closes that gap from the bottom up. We start with a program that wastes three seconds doing nothing, work out exactly where the time goes, and rebuild it on top of an event loop. Along the way we cover coroutines, tasks, structured concurrency, cancellation, backpressure, and the three mistakes that account for most broken async code in production.

Every snippet below is a complete, runnable file. Every output block is captured from an actual run on CPython 3.14.6 on arm64 macOS — timings included. Where a number looks surprising, it surprised me too, and I've said so.

What you should already know

Contents

  1. The real cost of waiting
  2. Threads, processes, coroutines
  3. What async def actually returns
  4. Why await is not concurrency
  5. The event loop, traced
  6. Running many things at once
  7. Timeouts and cancellation
  8. Backpressure: semaphores and queues
  9. async with, async for, async generators
  10. The blocking-call trap
  11. Races at every await
  12. What async actually buys you
  13. Debugging async code
  14. Capstone: a fetcher shaped like production code
  15. Where to go next

1. The real cost of waiting

Here is a program that fetches three stock prices. The network call is faked with time.sleep(1), which is an honest stand-in: a real HTTP round trip is also a period where your CPU has nothing to do but wait for a socket.

# 01_blocking.py
import time

PRICES = {"AAPL": 227.5, "MSFT": 415.2, "NVDA": 118.9}

def fetch_price(symbol: str) -> float:
    time.sleep(1)                 # stands in for a network round trip
    return PRICES[symbol]

def main() -> None:
    start = time.perf_counter()
    prices = [fetch_price(s) for s in PRICES]
    print(f"prices  = {prices}")
    print(f"elapsed = {time.perf_counter() - start:.2f}s")

main()

$ python3 01_blocking.py

prices  = [227.5, 415.2, 118.9]
elapsed = 3.01s

Three seconds. The interesting question is not "why is it slow" — it's what the CPU was doing for those three seconds. The answer is: almost nothing. It issued a request, then sat parked inside a system call until the answer came back, three times in a row. The work was serial not because the tasks depend on each other, but because time.sleep gives the interpreter no way to do anything else.

Now the same program with an event loop underneath it. Two keywords change, and one function call replaces the list comprehension.

# 02_async_first.py
import asyncio, time

PRICES = {"AAPL": 227.5, "MSFT": 415.2, "NVDA": 118.9}

async def fetch_price(symbol: str) -> float:
    await asyncio.sleep(1)        # yields control instead of blocking
    return PRICES[symbol]

async def main() -> None:
    start = time.perf_counter()
    prices = await asyncio.gather(*(fetch_price(s) for s in PRICES))
    print(f"prices  = {prices}")
    print(f"elapsed = {time.perf_counter() - start:.2f}s")

asyncio.run(main())

$ python3 02_async_first.py

prices  = [227.5, 415.2, 118.9]
elapsed = 1.00s

Same results, one third of the wall clock, one thread, one CPU core. Nothing was made faster — the three waits simply happened during each other instead of after each other.

             0s          1s          2s          3s
             |-----------|-----------|-----------|

BLOCKING     [== AAPL ==][== MSFT ==][== NVDA ==]
                                                 -> 3.01s

EVENT LOOP   [== AAPL ==]
             [== MSFT ==]
             [== NVDA ==]
                         -> 1.00s

The key distinction. time.sleep(1) blocks the thread: nothing else in the process can run. await asyncio.sleep(1) suspends the coroutine and hands control back to the event loop, which is free to run other coroutines until the timer fires. Same duration, opposite consequences.

2. Threads, processes, coroutines: picking the right tool

Python gives you three concurrency models, and choosing wrongly is the most expensive mistake in this whole subject. The choice hinges on one question: is your program waiting, or is it computing?

A quick vocabulary check, because these two words get used interchangeably and they are not the same thing:

asyncio gives you concurrency, not parallelism. That distinction is the source of most of the disappointment people feel when they add async to CPU-heavy code and nothing gets faster — we measure exactly that in section 10.

MODEL            GOOD FOR                          UNIT COST   PARALLEL?   FAILURE MODE
---------------  --------------------------------  ----------  ----------  ---------------------------
asyncio          Thousands of concurrent I/O        ~1.4 KB     No          One blocking call
(coroutines)     waits: HTTP, sockets, DB, queues               one thread  freezes everything

threading        Blocking libraries you cannot      ~35 KB      I/O only    Hits an OS ceiling in
(OS threads)     rewrite; moderate concurrency                  (GIL)       the low thousands

multiprocessing  CPU-bound work: parsing,           MBs         Yes         Everything crossing the
(processes)      compression, numerics, images                  real cores  boundary must be picklable

The unit costs in that table are not folklore; they're measured in section 12.

The reason threads don't give you parallel Python bytecode is the Global Interpreter Lock — a mutex that lets only one thread execute Python bytecode at a time. Threads still help with I/O, because a thread blocked in a socket read releases the GIL and lets another thread run. They do not help with pure computation. (CPython 3.13+ ships an experimental free-threaded build without the GIL; the standard interpreter, including the 3.14.6 build used here, still has it.)

3. What async def actually returns

The first genuine surprise for most people is that calling a coroutine function does not run it. It builds an object and hands it back, inert. Nothing executes until something drives it.

# 03_coroutine_object.py
import asyncio, inspect

async def greet(name: str) -> str:
    return f"hello {name}"

coro = greet("world")                 # nothing has executed yet
print("type      :", type(coro))
print("iscoroutine:", inspect.iscoroutine(coro))
print("running   :", asyncio.run(coro))

print("function  :", type(greet))
print("iscorofunc:", inspect.iscoroutinefunction(greet))

$ python3 03_coroutine_object.py

type      : <class 'coroutine'>
iscoroutine: True
running   : hello world
function  : <class 'function'>
iscorofunc: True

Read that carefully. greet is an ordinary function. Calling it returns a coroutine — a suspendable object, closely related to a generator, that knows how to be started, stopped, resumed, and thrown into. It only runs when a driver steps it, and in practice that driver is the event loop started by asyncio.run().

This is why forgetting an await is such a quiet bug: you get a perfectly valid object that simply never runs. We'll see the warning Python emits for it in section 13.

4. Why await is not concurrency

Here is the misconception that costs people the most time. await does not mean "run this in the background". It means "suspend me here until this finishes, and let the loop do other work meanwhile". If you write two awaits in a row, the second one starts after the first one is done — exactly like synchronous code.

Concurrency comes from scheduling, and scheduling is what asyncio.create_task() does.

# 04_await_is_sequential.py
import asyncio, time

async def work(n: int) -> int:
    await asyncio.sleep(1)
    return n * n

async def sequential() -> None:
    start = time.perf_counter()
    a = await work(2)          # await #1 finishes ...
    b = await work(3)          # ... before await #2 starts
    print(f"sequential -> {a}, {b} in {time.perf_counter() - start:.2f}s")

async def concurrent() -> None:
    start = time.perf_counter()
    ta = asyncio.create_task(work(2))   # scheduled immediately
    tb = asyncio.create_task(work(3))   # scheduled immediately
    a, b = await ta, await tb
    print(f"concurrent -> {a}, {b} in {time.perf_counter() - start:.2f}s")

asyncio.run(sequential())
asyncio.run(concurrent())

$ python3 04_await_is_sequential.py

sequential -> 4, 9 in 2.00s
concurrent -> 4, 9 in 1.00s

Both functions are async. Both use await. One takes twice as long. The difference is that create_task hands the coroutine to the loop right away and returns a handle; the loop starts running it at the next opportunity, whether or not you've awaited the handle yet.

Common trap. Sprinkling async/await through a codebase and expecting a speedup. If every call site is await one_thing() followed by await the_next_thing(), you have written synchronous code with extra ceremony and a slightly slower interpreter path. Concurrency appears the moment you create tasks — or use gather / TaskGroup, which create them for you.

5. The event loop, traced

The event loop is a single-threaded scheduler running a simple cycle: take the next ready callback, run it until it suspends or returns, then look at what became ready in the meantime and queue it up. That's the whole idea.

   +------------------+        +---------------------+        +------------------+
   |   READY QUEUE    |  pops  |  RUNNING COROUTINE  | await  |   OS SELECTOR    |
   | callbacks to run | -----> | runs to next await  | -----> | kqueue / epoll   |
   |       now        |  one   |                     |suspends|  + timer heap    |
   +------------------+        +---------------------+        +------------------+
            ^                                                          |
            |          when it is ready, push the callback back        |
            +----------------------------------------------------------+

   One thread. Nothing is preempted -- a coroutine keeps the loop until it
   awaits or returns.

The await step is the only place control leaves your code. Everything a coroutine does between two awaits is uninterruptible, which is simultaneously why async code is easy to reason about and why one blocking call ruins it.

Let's watch the cycle happen. Three workers with different delays, each logging a timestamp relative to program start:

# 05_event_loop_trace.py
import asyncio, time

START = time.perf_counter()

def log(msg: str) -> None:
    print(f"[{time.perf_counter() - START:5.2f}s] {msg}")

async def worker(name: str, delay: float) -> None:
    log(f"{name}: start")
    await asyncio.sleep(delay)          # suspension point -> loop takes over
    log(f"{name}: resumed after {delay}s")
    await asyncio.sleep(delay)
    log(f"{name}: done")

async def main() -> None:
    await asyncio.gather(
        worker("A", 0.3),
        worker("B", 0.5),
        worker("C", 0.1),
    )
    log("main: all workers finished")

asyncio.run(main())

$ python3 05_event_loop_trace.py

[ 0.00s] A: start
[ 0.00s] B: start
[ 0.00s] C: start
[ 0.10s] C: resumed after 0.1s
[ 0.20s] C: done
[ 0.30s] A: resumed after 0.3s
[ 0.50s] B: resumed after 0.5s
[ 0.60s] A: done
[ 1.00s] B: done
[ 1.00s] main: all workers finished

This output is worth studying line by line, because it shows the scheduler's actual behaviour:

6. Running many things at once

There are three ways to run a group of coroutines concurrently, and they differ mainly in how they handle failure. That difference matters more than the syntax.

asyncio.gather — collect results in order

gather returns results in the order you passed the coroutines in, regardless of which finished first. Its default error behaviour catches people out:

# 06_gather_errors.py
import asyncio

async def ok(n: int) -> int:
    await asyncio.sleep(0.1)
    return n

async def boom() -> int:
    await asyncio.sleep(0.05)
    raise ValueError("upstream returned 500")

async def main() -> None:
    # 1. default: the first exception propagates, siblings keep running detached
    try:
        await asyncio.gather(ok(1), boom(), ok(3))
    except ValueError as exc:
        print("gather default      ->", type(exc).__name__, exc)

    # 2. return_exceptions=True: errors come back as values
    results = await asyncio.gather(ok(1), boom(), ok(3), return_exceptions=True)
    print("return_exceptions   ->", results)

asyncio.run(main())

$ python3 06_gather_errors.py

gather default      -> ValueError upstream returned 500
return_exceptions   -> [1, ValueError('upstream returned 500'), 3]

The subtle problem is in case 1. When boom() raises, gather propagates that exception to you immediately — but the two ok() tasks are not cancelled. They keep running, unowned, in the background. If they hold a database connection or write to a file, they will do it after you thought the operation had failed.

asyncio.TaskGroup — structured concurrency (3.11+)

A task group fixes exactly that. It's a scope: when the async with block exits, every task created inside it is guaranteed to be finished, and if any one of them fails, the rest are cancelled before the exception reaches you.

# 07_taskgroup.py
import asyncio

async def ok(n: int) -> int:
    try:
        await asyncio.sleep(1)
    except asyncio.CancelledError:
        print(f"  task {n}: cancelled by the group")
        raise
    return n

async def boom() -> int:
    await asyncio.sleep(0.1)
    raise ValueError("upstream returned 500")

async def main() -> None:
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(ok(1))
            tg.create_task(boom())
            tg.create_task(ok(3))
    except* ValueError as eg:
        print("ExceptionGroup     ->", eg.exceptions)

asyncio.run(main())

$ python3 07_taskgroup.py

  task 3: cancelled by the group
  task 1: cancelled by the group
ExceptionGroup     -> (ValueError('upstream returned 500'),)

Two things to notice. First, the sibling tasks received a real CancelledError and had the chance to clean up — the program never leaks a running task past the block. Second, errors arrive as an ExceptionGroup, caught with except*, because more than one task can fail at once and a single except clause cannot represent that honestly.

Rule of thumb. Reach for TaskGroup by default. Use gather(..., return_exceptions=True) when partial failure is a legitimate outcome — a scraper that wants the 37 pages it managed to fetch, for example. Use bare gather almost never.

asyncio.as_completed — react as results arrive

When you want to process results the moment each one lands rather than waiting for the slowest:

# 19_as_completed.py
import asyncio, time

START = time.perf_counter()

async def job(name: str, delay: float) -> str:
    await asyncio.sleep(delay)
    return name

async def main() -> None:
    coros = [job("slow", 0.6), job("fast", 0.1), job("medium", 0.3)]
    for future in asyncio.as_completed(coros):        # yields in completion order
        name = await future
        print(f"[{time.perf_counter() - START:4.2f}s] finished {name}")

asyncio.run(main())

$ python3 19_as_completed.py

[0.10s] finished fast
[0.30s] finished medium
[0.60s] finished slow

Submission order was slow, fast, medium. Completion order is fast, medium, slow. That's the whole point: the first result is available at 0.10s instead of 0.60s.

7. Timeouts and cancellation

Cancellation is the part of asyncio that people skip, and it's the part that separates a demo from a service. In asyncio, cancellation is implemented as an exception: the loop throws asyncio.CancelledError into the coroutine at its current suspension point. That means your try/finally and async with blocks run normally — connections get released, files get closed.

# 08_timeout_cancel.py
import asyncio, time

async def slow_query() -> str:
    try:
        await asyncio.sleep(5)
        return "rows"
    except asyncio.CancelledError:
        print("  slow_query: cancelled -> releasing connection")
        raise

async def main() -> None:
    start = time.perf_counter()

    # asyncio.timeout: a cancel scope, added in 3.11
    try:
        async with asyncio.timeout(0.5):
            await slow_query()
    except TimeoutError:
        print(f"timed out after {time.perf_counter() - start:.2f}s")

    # manual cancellation of a task
    task = asyncio.create_task(slow_query())
    await asyncio.sleep(0.2)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("task.cancelled() =", task.cancelled())

    # shielding critical cleanup from an outer cancel
    async def critical() -> str:
        await asyncio.sleep(0.3)
        return "committed"
    try:
        async with asyncio.timeout(0.1):
            print("shielded result  =", await asyncio.shield(critical()))
    except TimeoutError:
        print("outer timed out, but the shielded coroutine kept running")
        await asyncio.sleep(0.3)

asyncio.run(main())

$ python3 08_timeout_cancel.py

  slow_query: cancelled -> releasing connection
timed out after 0.50s
  slow_query: cancelled -> releasing connection
task.cancelled() = True
outer timed out, but the shielded coroutine kept running

Three mechanisms, three jobs:

Common trap. Swallowing CancelledError. Writing except Exception is safe (since 3.8 CancelledError inherits from BaseException), but except BaseException: pass around an await turns your task into one that cannot be shut down. Catch it only to clean up, then raise.

8. Backpressure: semaphores and queues

Unbounded concurrency is a bug that looks like a feature. gather over 10,000 URLs will happily open 10,000 sockets, exhaust your file descriptors, and earn you a rate-limit ban. You need a way to say "at most N of these at a time".

Semaphore: cap the concurrency

# 09_semaphore.py
import asyncio, time

START = time.perf_counter()
in_flight = 0
peak = 0

async def fetch(url_id: int, sem: asyncio.Semaphore) -> int:
    global in_flight, peak
    async with sem:                      # at most N concurrent bodies
        in_flight += 1
        peak = max(peak, in_flight)
        await asyncio.sleep(0.2)
        in_flight -= 1
        return url_id

async def main() -> None:
    sem = asyncio.Semaphore(4)           # the throttle
    results = await asyncio.gather(*(fetch(i, sem) for i in range(20)))
    print(f"completed  = {len(results)} requests")
    print(f"peak concurrency = {peak}")
    print(f"elapsed    = {time.perf_counter() - START:.2f}s")

asyncio.run(main())

$ python3 09_semaphore.py

completed  = 20 requests
peak concurrency = 4
elapsed    = 1.01s

The arithmetic confirms the throttle is real: 20 tasks through 4 slots, each holding a slot for 0.2s, is 5 rounds x 0.2s = 1.0s. The measured 1.01s leaves 10ms for everything else. All 20 tasks were created up front — the semaphore simply parks them at the async with until a slot frees up.

Queue: a pipeline with a bounded buffer

A semaphore caps concurrency. A queue additionally decouples the producer from the consumers and applies backpressure: once maxsize is reached, await queue.put() suspends the producer until a consumer catches up.

# 10_queue_pipeline.py
import asyncio, random, time

async def producer(queue: asyncio.Queue, n: int) -> None:
    for i in range(n):
        await queue.put(f"job-{i:02d}")
        await asyncio.sleep(0.01)
    print("producer: finished enqueuing")

async def consumer(name: str, queue: asyncio.Queue, done: list) -> None:
    while True:
        job = await queue.get()
        try:
            await asyncio.sleep(random.uniform(0.02, 0.06))   # process
            done.append((name, job))
        finally:
            queue.task_done()            # always mark, even on failure

async def main() -> None:
    random.seed(7)
    start = time.perf_counter()
    queue: asyncio.Queue[str] = asyncio.Queue(maxsize=5)   # backpressure
    done: list = []

    workers = [asyncio.create_task(consumer(f"w{i}", queue, done)) for i in range(3)]
    await producer(queue, 15)
    await queue.join()                   # wait until every job is task_done()

    for w in workers:                    # consumers loop forever -> cancel them
        w.cancel()
    await asyncio.gather(*workers, return_exceptions=True)

    per_worker = {w: sum(1 for n, _ in done if n == w) for w in ("w0", "w1", "w2")}
    print(f"processed  = {len(done)} jobs in {time.perf_counter() - start:.2f}s")
    print(f"per worker = {per_worker}")

asyncio.run(main())

$ python3 10_queue_pipeline.py

producer: finished enqueuing
processed  = 15 jobs in 0.19s
per worker = {'w0': 5, 'w1': 5, 'w2': 5}

The load split perfectly five-five-five because each worker takes the next job the instant it's free — no partitioning logic required. Note the shutdown dance at the end: consumers built as while True loops never return on their own, so you wait on queue.join() for the work to drain, then cancel the workers explicitly. Forgetting that step is a classic way to hang a program on exit.

Choosing a bound

9. async with, async for, and async generators

Python's protocols all have asynchronous counterparts, so resource management and iteration can suspend too:

# 11_async_protocols.py
import asyncio, time

class ConnectionPool:
    """Async context manager: __aenter__ / __aexit__."""
    async def __aenter__(self) -> "ConnectionPool":
        await asyncio.sleep(0.1)
        print("pool: opened")
        return self

    async def __aexit__(self, exc_type, exc, tb) -> bool:
        await asyncio.sleep(0.1)
        print(f"pool: closed (exc={exc_type.__name__ if exc_type else None})")
        return False

    async def rows(self, n: int):
        for i in range(n):
            await asyncio.sleep(0.05)     # a page fetched from the wire
            yield {"id": i, "ts": round(time.perf_counter(), 3)}

async def paginate(n: int):
    """Async generator -> supports `async for`."""
    for page in range(n):
        await asyncio.sleep(0.02)
        yield page

async def main() -> None:
    async with ConnectionPool() as pool:
        async for row in pool.rows(3):    # __aiter__ / __anext__
            print("row:", row["id"])

    pages = [p async for p in paginate(4)]   # async comprehension
    print("pages:", pages)

asyncio.run(main())

$ python3 11_async_protocols.py

pool: opened
row: 0
row: 1
row: 2
pool: closed (exc=None)
pages: [0, 1, 2, 3]

The last line uses an async comprehension — [p async for p in paginate(4)]. It's valid anywhere inside a coroutine and reads exactly like the synchronous form.

10. The blocking-call trap

This is the single most common way to ruin an async program, and it's invisible in code review because the offending line looks completely ordinary. Anything that blocks — time.sleep, requests.get, a synchronous database driver, hashlib.pbkdf2_hmac, resizing an image — stops the entire event loop. Not just the coroutine that called it. Every task in the process.

The demonstration below runs a heartbeat task that should print every 0.25s, alongside a one-second blocking call, twice: once called directly, once offloaded with asyncio.to_thread.

# 12_blocking_trap.py
import asyncio, time

START = time.perf_counter()

def log(msg: str) -> None:
    print(f"[{time.perf_counter() - START:5.2f}s] {msg}")

def legacy_sync_call() -> str:
    time.sleep(1)                 # a blocking DB driver, an image resize, bcrypt...
    return "legacy result"

async def heartbeat(tag: str) -> None:
    for _ in range(4):
        await asyncio.sleep(0.25)
        log(f"{tag}: heartbeat")

async def wrong() -> None:
    log("WRONG -- blocking call inside a coroutine")
    hb = asyncio.create_task(heartbeat("wrong"))
    legacy_sync_call()            # <-- the event loop cannot run anything else
    await hb

async def right() -> None:
    log("RIGHT -- offloaded to a worker thread")
    hb = asyncio.create_task(heartbeat("right"))
    result = await asyncio.to_thread(legacy_sync_call)
    log(f"right: got {result!r}")
    await hb

asyncio.run(wrong())
asyncio.run(right())

$ python3 12_blocking_trap.py

[ 0.00s] WRONG -- blocking call inside a coroutine
[ 1.26s] wrong: heartbeat
[ 1.51s] wrong: heartbeat
[ 1.76s] wrong: heartbeat
[ 2.01s] wrong: heartbeat
[ 2.01s] RIGHT -- offloaded to a worker thread
[ 2.27s] right: heartbeat
[ 2.52s] right: heartbeat
[ 2.77s] right: heartbeat
[ 3.02s] right: heartbeat
[ 3.02s] right: got 'legacy result'

Look at the timestamps in the first half. The heartbeat was supposed to fire at 0.25, 0.50, 0.75 and 1.00 seconds. Instead the first one appears at 1.26s — the loop was frozen for a full second and could not service its own timers. All four heartbeats then fire back-to-back, late.

In the second half the loop stays responsive: heartbeats land at 0.26, 0.51, 0.76 and 1.01 seconds after that section started, and the blocking work completes in the background at the same time. In a web service, the first half is every request in flight timing out because one handler called a synchronous client.

How to offload correctly

SITUATION                              USE                                       WHY
-------------------------------------  ----------------------------------------  --------------------------------
Blocking I/O (sync HTTP client,        await asyncio.to_thread(fn, *args)        The thread blocks; the loop does
legacy DB driver, file reads)                                                    not. GIL released during I/O.

CPU-bound work (parsing,               loop.run_in_executor(                     Needs a separate process to
compression, numerics)                     ProcessPoolExecutor(), fn, ...)       escape the GIL and use a core.

You control the library                A native async client                     No thread per call -- this is
                                                                                 what scales to thousands.

And to be unambiguous about what asyncio cannot do, here is CPU-bound work under all three models. count_primes is deliberately naive trial division:

# 17_cpu_bound.py
import asyncio, time
from concurrent.futures import ProcessPoolExecutor

def count_primes(limit: int) -> int:
    n = 0
    for x in range(2, limit):
        for d in range(2, int(x ** 0.5) + 1):
            if x % d == 0:
                break
        else:
            n += 1
    return n

LIMIT, JOBS = 800_000, 4

async def with_asyncio() -> None:
    async def job():
        return count_primes(LIMIT)      # no await -> nothing to interleave
    t = time.perf_counter()
    await asyncio.gather(*(job() for _ in range(JOBS)))
    print(f"asyncio.gather      : {time.perf_counter() - t:.2f}s")

def with_processes() -> None:
    t = time.perf_counter()
    with ProcessPoolExecutor(max_workers=JOBS) as pool:
        list(pool.map(count_primes, [LIMIT] * JOBS))
    print(f"ProcessPoolExecutor : {time.perf_counter() - t:.2f}s")

if __name__ == "__main__":
    t = time.perf_counter()
    for _ in range(JOBS):
        count_primes(LIMIT)
    print(f"sequential          : {time.perf_counter() - t:.2f}s")
    asyncio.run(with_asyncio())
    with_processes()

$ python3 17_cpu_bound.py

sequential          : 3.55s
asyncio.gather      : 3.58s
ProcessPoolExecutor : 1.03s

asyncio was 0.03s slower than doing nothing — the overhead of wrapping the work in coroutines, with zero benefit, because a coroutine with no await in its body never yields and therefore never interleaves. Processes gave a 3.4x speedup on four cores. If your profile is CPU-bound, no amount of async will help you; that's what multiprocessing is for.

11. Races still happen — at every await

A comforting half-truth about asyncio is "single-threaded, so no locks needed". The accurate version is: a coroutine is only atomic between awaits. Every await is a point where the loop may run another task that touches the same state.

# 20_await_race.py
import asyncio

balance = 100

async def withdraw_unsafe(amount: int) -> None:
    global balance
    if balance >= amount:            # check
        await asyncio.sleep(0)       # <-- await = a place another task can run
        balance -= amount            # act

async def withdraw_safe(amount: int, lock: asyncio.Lock) -> None:
    global balance
    async with lock:                 # check and act become atomic
        if balance >= amount:
            await asyncio.sleep(0)
            balance -= amount

async def main() -> None:
    global balance
    balance = 100
    await asyncio.gather(*(withdraw_unsafe(100) for _ in range(3)))
    print("without a lock:", balance)

    balance = 100
    lock = asyncio.Lock()
    await asyncio.gather(*(withdraw_safe(100, lock) for _ in range(3)))
    print("with a lock   :", balance)

asyncio.run(main())

$ python3 20_await_race.py

without a lock: -200
with a lock   : 0

Three withdrawals of 100 from a balance of 100 left the account at -200. All three coroutines passed the balance >= amount check before any of them reached the subtraction, because the await in between handed control away. Note that await asyncio.sleep(0) does no waiting at all — it's a pure yield to the scheduler — and it was still enough to break the invariant.

The fix is the same as in threaded code, with async spellings: asyncio.Lock, asyncio.Semaphore, asyncio.Event, asyncio.Condition. Use the asyncio versions, never threading.Lock — the latter blocks the loop rather than suspending the coroutine.

How to spot it. Scan for any await that sits between reading shared state and writing it. That's a check-then-act race, whether it involves a lock, a cache, a counter, or a "have I already fetched this URL?" set.

12. What async actually buys you

Fake latency with asyncio.sleep proves the concept but not the engineering. This benchmark runs real TCP connections against a real server — an asyncio server on a background thread with its own event loop, sleeping 50ms per request to imitate a slow upstream — and puts three client strategies against it.

# 15_benchmark_io.py
"""Sequential vs threads vs asyncio, measured against a real local TCP server."""
import asyncio, socket, threading, time
from concurrent.futures import ThreadPoolExecutor

LATENCY = 0.05          # 50 ms of server-side "work" per request
N = 200                 # number of requests per strategy

# ---------------------------------------------------------------- server
async def handle(reader, writer):
    await reader.read(1024)
    await asyncio.sleep(LATENCY)          # simulated upstream latency
    writer.write(b"pong")
    await writer.drain()
    writer.close()

def run_server(ready: threading.Event, box: dict) -> None:
    async def serve():
        server = await asyncio.start_server(handle, "127.0.0.1", 0)
        box["addr"] = server.sockets[0].getsockname()
        ready.set()
        async with server:
            await server.serve_forever()
    asyncio.run(serve())

# ---------------------------------------------------------------- clients
def sync_request(addr) -> bytes:
    with socket.create_connection(addr) as sock:
        sock.sendall(b"ping")
        return sock.recv(1024)

async def async_request(addr) -> bytes:
    reader, writer = await asyncio.open_connection(*addr)
    writer.write(b"ping")
    await writer.drain()
    data = await reader.read(1024)
    writer.close()
    return data

# ---------------------------------------------------------------- harness
async def main() -> None:
    ready, box = threading.Event(), {}
    threading.Thread(target=run_server, args=(ready, box), daemon=True).start()
    ready.wait()
    addr = box["addr"]
    rows = []

    t = time.perf_counter()
    for _ in range(N):
        sync_request(addr)
    rows.append(("sequential (blocking)", time.perf_counter() - t))

    with ThreadPoolExecutor(max_workers=32) as pool:
        t = time.perf_counter()
        list(pool.map(lambda _: sync_request(addr), range(N)))
        rows.append(("threads (32 workers)", time.perf_counter() - t))

    with ThreadPoolExecutor(max_workers=200) as pool:
        t = time.perf_counter()
        list(pool.map(lambda _: sync_request(addr), range(N)))
        rows.append(("threads (200 workers)", time.perf_counter() - t))

    t = time.perf_counter()
    await asyncio.gather(*(async_request(addr) for _ in range(N)))
    rows.append(("asyncio (1 thread, 200 tasks)", time.perf_counter() - t))

    print(f"{N} requests, {LATENCY * 1000:.0f} ms server latency each")
    print(f"theoretical floor with full concurrency: {LATENCY:.2f}s\n")
    print(f"{'strategy':<32}{'wall clock':>12}{'req/s':>10}{'speedup':>10}")
    print("-" * 64)
    baseline = rows[0][1]
    for label, elapsed in rows:
        print(f"{label:<32}{elapsed:>10.2f}s{N / elapsed:>10.0f}{baseline / elapsed:>9.1f}x")

asyncio.run(main())

$ python3 15_benchmark_io.py

200 requests, 50 ms server latency each
theoretical floor with full concurrency: 0.05s

strategy                          wall clock     req/s   speedup
----------------------------------------------------------------
sequential (blocking)                10.42s        19      1.0x
threads (32 workers)                  0.39s       510     26.6x
threads (200 workers)                 0.08s      2396    124.8x
asyncio (1 thread, 200 tasks)         0.08s      2443    127.3x

The honest reading of this result is important, because it's not the one people expect:

That last point is where the models actually diverge, so measure it directly. This script spawns N concurrent sleepers as tasks or as threads and reports spawn time and peak memory, each in a fresh process:

# 16_scaling_cost.py
"""What does one unit of concurrency cost?  Run as: python3 16_scaling_cost.py tasks|threads N"""
import asyncio, resource, sys, threading, time

def rss_mib() -> float:
    return resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024 ** 2   # macOS: bytes

async def run_tasks(n: int) -> None:
    base, t = rss_mib(), time.perf_counter()
    ts = [asyncio.create_task(asyncio.sleep(0.5)) for _ in range(n)]
    spawn = time.perf_counter() - t
    await asyncio.gather(*ts)
    report(f"{n:,} asyncio tasks", spawn, time.perf_counter() - t, rss_mib() - base)

def run_threads(n: int) -> None:
    base, t = rss_mib(), time.perf_counter()
    ths = [threading.Thread(target=time.sleep, args=(0.5,)) for _ in range(n)]
    for th in ths: th.start()
    spawn = time.perf_counter() - t
    for th in ths: th.join()
    report(f"{n:,} OS threads", spawn, time.perf_counter() - t, rss_mib() - base)

def report(label, spawn, total, mem):
    print(f"{label:<22} spawn {spawn * 1000:8.1f} ms   wall {total:5.2f}s   +peak RSS {mem:6.1f} MiB")

mode, n = sys.argv[1], int(sys.argv[2])
run_threads(n) if mode == "threads" else asyncio.run(run_tasks(n))

$ for n in 1000 5000; do python3 16_scaling_cost.py tasks $n; \
                        python3 16_scaling_cost.py threads $n; done

1,000 asyncio tasks    spawn      0.8 ms   wall  0.51s   +peak RSS    1.4 MiB
1,000 OS threads       spawn     23.9 ms   wall  0.54s   +peak RSS   34.8 MiB
5,000 asyncio tasks    spawn      4.7 ms   wall  0.55s   +peak RSS    6.5 MiB
Traceback (most recent call last):
  File "16_scaling_cost.py", line 26, in <module>
    run_threads(n) if mode == "threads" else asyncio.run(run_tasks(n))
    ~~~~~~~~~~~^^^
  File ".../threading.py", line 1005, in start
    _start_joinable_thread(self._bootstrap, handle=self._os_thread_handle,
    ~~~~~~~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
                           daemon=self.daemon)
                           ^^^^^^^^^^^^^^^^^^^
RuntimeError: can't start new thread

That traceback is the answer to "why bother with asyncio if threads are just as fast". At 5,000 concurrent units the thread version does not run at all — the OS refused to create the thread. The task version handled the same workload in 4.7 milliseconds of spawn time and 6.5 MiB.

Pushing further, well past where threads died:

$ python3 16_scaling_cost.py tasks 50000

50,000 asyncio tasks   spawn     56.3 ms   wall  0.75s   +peak RSS   67.3 MiB

CONCURRENCY      SPAWN TIME   PEAK RSS ADDED   PER UNIT   RESULT
---------------  -----------  ---------------  ---------  --------------
  1,000 tasks         0.8 ms          1.4 MiB   ~1.4 KB   ran
  1,000 threads      23.9 ms         34.8 MiB   ~35  KB   ran
  5,000 tasks         4.7 ms          6.5 MiB   ~1.3 KB   ran
  5,000 threads           --               --        --   RuntimeError
 50,000 tasks        56.3 ms         67.3 MiB   ~1.4 KB   ran

The actual value proposition

13. Debugging async code

Async bugs are mostly quiet. Nothing crashes; the program is just slow, or a task silently never ran. Two built-in tools catch the majority of them.

Debug mode finds whoever is blocking the loop

Pass debug=True to asyncio.run() (or set PYTHONASYNCIODEBUG=1) and asyncio will log any callback that hogs the loop for more than 100ms:

# 13_debug_mode.py
import asyncio, time

async def hog() -> None:
    time.sleep(0.4)               # blocks the loop for 400 ms

async def main() -> None:
    await hog()

asyncio.run(main(), debug=True)   # PYTHONASYNCIODEBUG=1 does the same

$ python3 13_debug_mode.py

Executing <Task finished name='Task-1' coro=<main() done, defined at
13_debug_mode.py:6> result=None created at .../asyncio/runners.py:110>
took 0.401 seconds

Turn this on in development permanently. It points straight at the offending coroutine, with the file and line, which is exactly what you need to find the sneaky synchronous call in the middle of a large codebase.

The forgotten await

# 14_forgotten_await.py
import asyncio

async def save(record: str) -> str:
    await asyncio.sleep(0.01)
    return f"saved {record}"

async def main() -> None:
    save("user-42")               # forgot the await -> never scheduled
    print("main finished")

asyncio.run(main())

$ python3 -W always 14_forgotten_await.py

14_forgotten_await.py:8: RuntimeWarning: coroutine 'save' was never awaited
  save("user-42")               # forgot the await -> never scheduled
RuntimeWarning: Enable tracemalloc to get the object allocation traceback
main finished

The record was never saved and the program exited successfully. In a test suite that runs with warnings suppressed, this bug ships. Run your async tests with -W error::RuntimeWarning and it becomes a hard failure instead.

The rest of the checklist

SYMPTOM                                LIKELY CAUSE                          FIX
-------------------------------------  ------------------------------------  ------------------------------
Async code is no faster than sync      Sequential awaits, or a blocking      create_task / TaskGroup;
                                       call inside a coroutine               debug=True to find the blocker

Work silently never happens            Missing await                         -W error::RuntimeWarning

"Task was destroyed but it is          A bare create_task whose handle       Keep a reference, or create it
pending!"                              was garbage collected                 inside a TaskGroup

Program hangs on exit                  A while True consumer that is         queue.join(), then cancel and
                                       never cancelled                       await the workers

Too many open files                    Unbounded gather over a large list    asyncio.Semaphore

Shutdown never completes               CancelledError caught and swallowed   Catch to clean up, then raise

Corrupted shared state                 Check-then-act across an await        asyncio.Lock

14. Capstone: a fetcher shaped like production code

Everything so far has been one idea at a time. Real async code combines them, and the combination is what actually looks like a service: bounded concurrency, a deadline per attempt, retries with backoff, a progress reporter running alongside the work, and a structured scope so nothing outlives the block.

The upstream here is deliberately hostile — 25% of calls raise ConnectionError, and one response in four takes 1.5 seconds, well past the 0.8s timeout.

# 18_capstone_scraper.py
"""Capstone: a bounded, timeout-aware, retrying async fetcher with live progress."""
import asyncio, random, time
from dataclasses import dataclass, field

MAX_CONCURRENCY = 8
PER_REQUEST_TIMEOUT = 0.8
MAX_ATTEMPTS = 3

@dataclass
class Stats:
    ok: int = 0
    failed: int = 0
    retries: int = 0
    latencies: list[float] = field(default_factory=list)

async def fake_http_get(url: str) -> str:
    """Flaky upstream: 25% transient failures, occasional very slow response."""
    delay = random.choice([0.05, 0.1, 0.15, 1.5])      # 1.5s -> will time out
    await asyncio.sleep(delay)
    if random.random() < 0.25:
        raise ConnectionError("connection reset by peer")
    return f"<html>{url}</html>"

async def fetch_one(url: str, sem: asyncio.Semaphore, stats: Stats) -> str | None:
    async with sem:                                    # bound concurrency
        for attempt in range(1, MAX_ATTEMPTS + 1):
            started = time.perf_counter()
            try:
                async with asyncio.timeout(PER_REQUEST_TIMEOUT):
                    body = await fake_http_get(url)
                stats.ok += 1
                stats.latencies.append(time.perf_counter() - started)
                return body
            except (ConnectionError, TimeoutError) as exc:
                if attempt == MAX_ATTEMPTS:
                    stats.failed += 1
                    print(f"  giving up on {url}: {type(exc).__name__}")
                    return None
                stats.retries += 1
                await asyncio.sleep(0.05 * 2 ** (attempt - 1))   # backoff

async def progress(total: int, stats: Stats, stop: asyncio.Event) -> None:
    while not stop.is_set():
        done = stats.ok + stats.failed
        print(f"  progress: {done:3d}/{total}  ok={stats.ok} "
              f"failed={stats.failed} retries={stats.retries}")
        try:
            await asyncio.wait_for(stop.wait(), timeout=0.4)
        except TimeoutError:
            pass

async def main() -> None:
    random.seed(11)
    urls = [f"https://example.test/page/{i}" for i in range(40)]
    sem, stats, stop = asyncio.Semaphore(MAX_CONCURRENCY), Stats(), asyncio.Event()
    start = time.perf_counter()

    reporter = asyncio.create_task(progress(len(urls), stats, stop))
    async with asyncio.TaskGroup() as tg:              # structured concurrency
        tasks = [tg.create_task(fetch_one(u, sem, stats)) for u in urls]
    stop.set()
    await reporter

    bodies = [t.result() for t in tasks if t.result()]
    avg = sum(stats.latencies) / len(stats.latencies)
    print(f"\nfetched   : {len(bodies)}/{len(urls)} pages")
    print(f"retries   : {stats.retries}   permanent failures: {stats.failed}")
    print(f"avg latency (successful): {avg * 1000:.0f} ms")
    print(f"wall clock: {time.perf_counter() - start:.2f}s")

asyncio.run(main())

$ python3 18_capstone_scraper.py

  progress:   0/40  ok=0 failed=0 retries=0
  progress:   7/40  ok=7 failed=0 retries=3
  progress:  14/40  ok=14 failed=0 retries=7
  giving up on https://example.test/page/3: ConnectionError
  progress:  29/40  ok=28 failed=1 retries=11
  progress:  33/40  ok=32 failed=1 retries=13
  giving up on https://example.test/page/1: ConnectionError
  progress:  36/40  ok=34 failed=2 retries=19
  progress:  38/40  ok=36 failed=2 retries=22
  progress:  38/40  ok=36 failed=2 retries=22
  giving up on https://example.test/page/39: TimeoutError

fetched   : 37/40 pages
retries   : 23   permanent failures: 3
avg latency (successful): 90 ms
wall clock: 3.16s

37 of 40 pages retrieved from an upstream that failed a quarter of all calls, in 3.16 seconds, on one thread. Three failures survived three attempts each — two connection resets and one request that hit the 0.8s deadline every time. The progress line kept updating throughout, because the reporter is an ordinary task sharing the loop with the fetchers.

The pieces map directly onto earlier sections:

LINE                                    MECHANISM                     WHAT IT PREVENTS
--------------------------------------  ----------------------------  ---------------------------------
async with sem                          Semaphore (sec. 8)            40 simultaneous sockets and a
                                                                      rate-limit ban

async with asyncio.timeout(...)         Cancel scope (sec. 7)         One slow response holding a slot
                                                                      forever

await asyncio.sleep(0.05 * 2 ** ...)    Exponential backoff           Hammering an upstream that is
                                                                      already struggling

async with asyncio.TaskGroup()          Structured concurrency (6)    Tasks outliving the function that
                                                                      started them

asyncio.Event + reporter task           Synchronisation (sec. 11)     A progress loop that never stops

Two things to add for real use. Retry only on idempotent operations, and add jitter to the backoff — delay * (1 + random.random()) — so a thousand clients recovering from the same outage don't retry in lockstep.

15. Where to go next

If you take five things from this article, take these:

  1. async is for waiting, not for computing. I/O-bound got a 100x win; CPU-bound got 0.03 seconds slower.
  2. await alone is not concurrency. Concurrency starts at create_task, gather, or TaskGroup.
  3. One blocking call freezes everything. Offload with asyncio.to_thread, and run development with debug=True to catch the ones you missed.
  4. Always bound your concurrency. A Semaphore or a bounded Queue is not optional at scale.
  5. Single-threaded does not mean race-free. Every await is a place another task can change your state.

Worth learning next

All 20 scripts in this article are complete and self-contained: copy any block into a .py file and run it with Python 3.11 or newer, no dependencies. Every output block was captured from a real run on CPython 3.14.6 (arm64 macOS). Timings will differ on your machine; the ratios should not.

Async Python From the Event Loop Up