Event Loop & Coroutines

Most programs spend far more time waiting than computing: for a web service to answer, a database to reply, a socket to deliver bytes. If your code sits idle during every wait, a server handling thousands of connections needs thousands of threads, and each one costs megabytes of stack. asyncio takes a different route. It runs many tasks on one thread and lets each task hand control back whenever it has to wait, so the waiting itself is almost free.

This is cooperative concurrency: a task keeps running until it reaches an await, and only there can another task take over. That makes switch points visible in your code and removes many of the races that plague threads. It also comes with a catch that this chapter keeps coming back to. A task that never awaits, such as a time.sleep() call or a tight CPU loop, freezes everything on the loop, and asyncio does not make CPU-bound work any faster.

By the end of the chapter you will be able to write coroutines with async / await, start a program with asyncio.run, and tell coroutines, Tasks and Futures apart. You will see how an await suspends and resumes a coroutine, and run work concurrently. You will also keep blocking calls off the loop with to_thread and run_in_executor, and pick the right tool (asyncio, threads or processes) for a given workload.

Two words get mixed up all the time. Concurrency means juggling many tasks so that their progress overlaps in time: while one is waiting, another moves forward. Parallelism means tasks literally running at the same instant, each on its own CPU core. You can have the first without the second. A single cook who starts the rice, chops vegetables while it simmers, and checks the oven is concurrent. Three cooks working at three stoves are parallel.

asyncio gives you concurrency only. Everything runs on one thread, so at any single moment exactly one piece of your Python code is executing. The speedup comes from not wasting the time a task spends waiting.

The way asyncio achieves this is called cooperative multitasking. A task keeps the thread until it chooses to give it up, and the place where it chooses is an await. The diagram below follows one task through that cycle.

Threads are preemptive: the operating system may pause one at almost any bytecode boundary and run another, whether the code is ready or not. You cannot point at a line and say where the switch will happen. asyncio works the opposite way. A switch can only happen at an await that you can see written in the source, and every other line runs straight through without interruption.

async def handle(conn):
    req = await conn.read()    # may switch here
    n = count(req)             # cannot switch: no await
    await db.save(n)           # may switch here
    await conn.write(n)        # may switch here

Only the three awaits are switch points; count(req) always runs to completion.

Here is the idea in a program you can run. Two workers start together, but each hands control back while it sleeps. Worker B has the shorter sleep, so it finishes first even though A started first.

Both workers print start before either prints done, because A reached its await and let B begin. The whole run takes about 0.2 seconds, not 0.3.

asyncio suits I/O-bound workloads: calls to other services over HTTP, database queries that wait for a reply, raw sockets, chat servers and websockets. The classic case is a server holding thousands of mostly idle connections, each of which spends nearly all its time waiting for the next message.

The reason it works so well is that waiting costs almost nothing. When a coroutine awaits a socket, it is parked. No thread is blocked and no CPU is burned. The event loop simply runs whichever other coroutines are ready, and when the data arrives the parked one resumes right after its await.

The second saving is memory. An operating system thread reserves roughly 1 to 8 MB of stack space up front, while a coroutine object is only about 1 to 2 KB. That difference is what makes ten thousand simultaneous connections practical.

To see this, the program below starts ten thousand tasks that each wait a tenth of a second. Because the waiting overlaps, the whole thing finishes in roughly a tenth of a second rather than seventeen minutes.

Because switches happen only at an await, the code between two awaits is effectively atomic: no other task can interrupt it. That removes most of the race conditions you would worry about with threads. It does not remove all of them. If a task reads shared state, awaits, and then writes the state back, other tasks can change it during that await and the write is based on stale data. Updates that span an await need an asyncio.Lock.

Two withdrawals of 10 from 100 should leave 80. The unsafe version loses one of them because both tasks read 100 before either wrote back. The lock makes the read, the await and the write happen as one unit.

Now compare the three ways of getting concurrency in Python. Each trades something different for what it gives you.

Function colouring is the catch in that last row. A plain function cannot await, so calling a coroutine from sync code just creates a coroutine object that never runs. Once one function in a call chain becomes async def, its callers must become async too, all the way up. One blocking call hidden inside stalls every task, so you also need async libraries all the way down.

Writing async def in front of a function changes what calling it means. The result is a coroutine function, and when you call it, Python does not run a single line of the body. Instead the call hands back a coroutine object, a suspendable computation that is waiting to be started.

The example below defines a coroutine function, calls it, and prints nothing from the body until the coroutine is actually run. asyncio.run is used here only as a way to start the coroutine. The event loop behind it gets its own section later.

import asyncio

async def greet():
    print('hi')
    return 'done'

c = greet()
print(type(c))
print(asyncio.run(c))

The call alone prints nothing from the body

Notice the order. The line print(type(c)) ran first and showed a coroutine object, and the word hi appeared only once asyncio.run started the coroutine. A coroutine object runs in exactly two situations: another coroutine awaits it, or it is wrapped in a Task and scheduled on the event loop.

Inside a coroutine, await expr pauses the current coroutine until the awaitable expr has finished. The whole await expression then evaluates to the awaitable's result. If the awaitable failed, the exception is re-raised at the await line, so you can wrap it in an ordinary try and except.

Because await can pause a function, Python only accepts it in places that are allowed to pause. That means the body of an async def, or the top level of the python -m asyncio REPL. Anywhere else it is a SyntaxError, even in a module-level script.

Here is the classic shape. fetch waits one second and returns a value, and a second coroutine awaits it to get that value. Only a coroutine can await another coroutine, which is why the caller is itself an async def.

Coroutines are built on the same machinery as generators, which is why they can pause and resume. Calling send(None) on a coroutine object advances it to its next suspension point. When the body finishes, the coroutine raises StopIteration, and the return value is stored in its value attribute. The event loop does exactly this for you behind the scenes. Doing it by hand shows the mechanism.

async def add():
    return 1 + 2

c = add()
try:
    c.send(None)
except StopIteration as e:
    print(e.value)

Driving a coroutine by hand, the way the loop does

This coroutine never suspends, so a single send(None) ran it to the end. A coroutine that awaits something pending would stop at that point instead, and the loop would send again later to resume it.

Ordinary with and for statements call methods that cannot pause. When setting up a resource or fetching the next item needs to wait on I/O, Python offers async versions of both. They use their own special methods, which are themselves coroutines.

An async generator is simply an async def that uses yield. You cannot loop over it with a plain for. You consume it with async for, which awaits each next value, and that in turn can only happen inside another coroutine. The example combines an async with block and an async for loop over an async generator.

The open line comes from __aenter__ and the close line from __aexit__, which runs even if the loop body raises. Between them, each pass of async for waits for the next value from ticks.

async def main():
    fetch()        # RuntimeWarning: coroutine 'fetch' was never awaited
    await fetch()  # this one actually runs

The first line only builds an object

The opposite mistake is awaiting something that is not awaitable. If you await the result of an ordinary function, you are awaiting its return value, such as an int, and Python refuses with a TypeError.

When you are unsure what you are holding, ask Python. inspect.iscoroutinefunction takes the function itself, the thing you wrote after async def. inspect.iscoroutine takes the object you get back from calling it. asyncio.iscoroutine answers the same question as inspect.iscoroutine. The code below runs both checks on both objects.

import asyncio
import inspect

async def fetch():
    return 'data'

c = fetch()
print(inspect.iscoroutinefunction(fetch))
print(inspect.iscoroutine(fetch))
print(inspect.iscoroutine(c))
print(asyncio.iscoroutine(c))
c.close()

c.close() quietly discards the unused coroutine

The table lines up each check with the two things you might pass to it. Each column is named after what you hand to the check. Mixing the two up gives False.

Every await you have written so far hands control to something, and that something is the event loop. It is asyncio's scheduler: one object, running on one thread, that decides which piece of your program runs next. It does not run anything in parallel. It simply keeps picking the next piece of work and running it to its next suspension point.

To do that job the loop keeps three collections. Think of them as the loop's whole memory.

Everything asyncio does, from sleep to network reads, comes down to putting entries into these three places and reacting when they become ready. A coroutine that is waiting is not stored in a busy list that the loop polls. It is parked behind a timer or a file descriptor, and it costs nothing until that event fires.

A thread can have at most one running loop at a time. asyncio.get_running_loop() returns the loop that is running in the current thread, and it raises RuntimeError when there is none. The first snippet below shows both cases.

import asyncio

try:
    asyncio.get_running_loop()
except RuntimeError as e:
    print('outside:', e)

async def main():
    loop = asyncio.get_running_loop()
    print('inside:', loop is not None)

asyncio.run(main())

The same call fails outside a loop and succeeds inside one

The loop is just a cycle. Each trip around it is called an iteration, and every iteration does the same four things in the same order.

The timeout step is what keeps the loop cheap. If the ready queue is empty and the nearest timer is 2 seconds away, the loop sleeps in the selector for at most 2 seconds, and wakes earlier if a socket becomes readable. If callbacks are already waiting, the timeout is zero and the loop does not block at all.

The selector wait is not a loop of polling calls in Python. It is a single call into an operating system facility that watches many file descriptors at once and returns when any of them is ready. Which facility is used depends on the platform.

A Task is not a thread and not a special object that the loop understands. It is driven by ordinary callbacks. When a task is created, the loop puts a call to Task.__step() on the ready queue. When that callback runs, it advances the coroutine by exactly one send(), which runs your code up to its next suspension point. Then control returns to the loop.

When whatever the task was waiting for finishes, another callback puts Task.__step() back on the ready queue, and the cycle repeats. A coroutine with five suspension points is therefore advanced by five separate callbacks.

You rarely touch the ready queue directly, but the loop exposes the same mechanism that tasks use. Three methods put a plain function on the loop's schedule. They are all called from the loop's own thread.

The first example shows an important detail about call_soon. The call comes before the print in main, but it only queues the callback. Nothing else can run while main is executing, so the callback has to wait until main suspends at its await.

import asyncio

async def main():
    loop = asyncio.get_running_loop()
    loop.call_soon(print, 'callback: ran from the ready queue')
    print('main: before the first await')
    await asyncio.sleep(0)
    print('main: after the first await')

asyncio.run(main())

call_soon is queued first, yet it prints second

Timers work the same way, except the callback waits on the heap instead of the ready queue. call_at takes an absolute time on the loop's own clock, so build it from loop.time(), never from time.time(). Here the callbacks fire in time order, not in the order they were scheduled.

The loop is not thread-safe. Its ready queue and timer heap are plain Python structures with no locking, and the loop may be in the middle of using them when another thread touches them. A plain call_soon from a worker thread can corrupt the queue, or leave the loop asleep in the selector with no wake-up. call_soon_threadsafe fixes both: it queues the callback safely and wakes the loop.

In this example, a worker thread hands a print to the loop, then signals an asyncio.Event the same way. Note that Event.set must also go through call_soon_threadsafe, because the event belongs to the loop.

When the thread needs a result back, use run_coroutine_threadsafe. It returns a concurrent.futures.Future, and calling result(timeout) on it blocks the worker thread, not the loop, until the coroutine finishes.

Because the loop runs one callback at a time, any callback that runs for a long time delays everything else. Debug mode makes these problems visible. Turn it on with asyncio.run(main(), debug=True) or by setting the environment variable PYTHONASYNCIODEBUG=1. It logs every callback that runs longer than 100 ms, and it reports coroutines that were created but never awaited.

PYTHONASYNCIODEBUG=1 python app.py

Same effect as passing debug=True to asyncio.run

The default loop is written in Python on top of the selector. uvloop is a drop-in replacement built on libuv, the same C library that powers Node.js. For network-heavy servers it is often about 2-4x faster. With version 0.18 or later you use it by calling uvloop.run(main()) instead of asyncio.run(main()). It supports Linux and macOS only, not Windows.

A Python program starts in ordinary synchronous code, but a coroutine can only run on an event loop. asyncio.run(main()) is the standard bridge between the two. Plain sync code calls it, hands it a coroutine, and gets that coroutine's result back as if it had made a normal function call. It is normally called once per program, at the very top level, and everything else in your app runs inside the loop it creates.

The first step builds a brand-new loop and makes it the current loop for the calling thread, so asyncio.get_running_loop() works inside your coroutines. The second step runs main() until it finishes. Whatever main() returns becomes the return value of asyncio.run, and if main() raises, the same exception comes out of asyncio.run into your sync code.

The usual shape of a script is an async def main() that holds the program, followed by a guarded call at the bottom of the file. The if __name__ == '__main__': guard keeps the loop from starting when another module merely imports your file.

import asyncio

async def work():
    await asyncio.sleep(0.1)
    return 'done'

async def main():
    result = await work()
    return result.upper()

if __name__ == '__main__':
    value = asyncio.run(main())   # once, at the top
    print(value)

The result of main() comes back as a plain value

The third step is cleanup, and it runs whether main() returned normally or raised. asyncio.run does not simply drop the loop. It works through a short list of chores so that nothing is left half-finished before the loop is closed.

Exceptions travel the same road as results. The loop is already current while main() runs, and an error raised inside it reaches your calling code only after cleanup has finished.

import asyncio

async def main():
    loop = asyncio.get_running_loop()
    print('running:', loop.is_running())
    raise ValueError('boom')

try:
    asyncio.run(main())
except ValueError as e:
    print('caught', e)

A failure in main() is re-raised by asyncio.run

The cancellation step has a consequence that surprises many people. When main() returns, every task still pending is cancelled, so fire-and-forget work is killed silently at shutdown. In the next example the background task asks for ten seconds, but main() only waits a tenth of a second.

import asyncio

async def background():
    try:
        await asyncio.sleep(10)
        print('finished')
    except asyncio.CancelledError:
        print('background cancelled')
        raise

async def main():
    asyncio.create_task(background())
    await asyncio.sleep(0.1)
    print('main returns')

asyncio.run(main())

The task never gets to print 'finished'

Without that print in the except block you would see nothing at all, which is why this failure is so quiet. The fix is to keep a reference to the task and await it before main() returns.

import asyncio

async def background():
    await asyncio.sleep(0.2)
    print('finished')

async def main():
    task = asyncio.create_task(background())
    print('main waiting')
    await task
    print('main returns')

asyncio.run(main())

Awaiting the task lets it complete before cleanup

asyncio.run always wants to create a fresh loop, so it refuses to start when a loop is already running in the current thread. This happens in Jupyter, where the notebook has its own loop, and also if you call it from inside another coroutine. Either way it raises RuntimeError with the message asyncio.run() cannot be called from a running event loop. In those places the loop already exists, so you simply write await main() instead.

Sometimes a sync program really does need several separate entry points, for example a setup step followed by the main work, and both must live on the same loop. Python 3.11 added asyncio.Runner(), a context manager that creates one loop and lets you call runner.run(coro) as many times as you like. The loop is closed, with the same cleanup as before, when the with block exits. The example below records the loop seen by each call and compares them.

import asyncio

loops = []

async def step(name):
    loops.append(asyncio.get_running_loop())
    return name

with asyncio.Runner() as runner:
    print(runner.run(step('setup')))
    print(runner.run(step('main')))
print('runner same loop:', loops[0] is loops[1])

loops.clear()
asyncio.run(step('a'))
asyncio.run(step('b'))
print('run twice same loop:', loops[0] is loops[1])

Runner reuses one loop; two run() calls make two loops

Most asyncio.run problems come from treating it like an ordinary function call that is cheap to repeat. Every call builds a loop, runs the full cleanup sequence and closes the loop again, so it only belongs at the outer edge of your program.

The second mistake follows from the first. Objects such as a Lock, a Queue or a client session belong to the loop they were used on, and they cannot be moved to a different loop. If a second asyncio.run call finds an object that an earlier call tied to its own loop, the object fails. The example below uses a module-level lock that really gets contended in each run.

import asyncio

lock = asyncio.Lock()

async def use():
    async def holder():
        async with lock:
            await asyncio.sleep(0.05)
    t = asyncio.create_task(holder())
    await asyncio.sleep(0)
    async with lock:      # has to wait, so it binds to this loop
        pass
    await t

asyncio.run(use())
print('first run ok')
try:
    asyncio.run(use())
except RuntimeError as e:
    print('second run failed:', 'different event loop' in str(e))

The lock stays tied to the first run's loop

The await keyword does not accept just anything. It accepts an awaitable: an object the event loop knows how to wait on. There are three built-in kinds, and a fourth escape hatch for your own classes.

A coroutine is the plainest awaitable, and it is lazy. Calling step() only builds a coroutine object. Its body starts when something awaits it, and it then runs sequentially inside the caller: the caller waits until the coroutine finishes before moving to its next line. Awaiting a coroutine does not create any concurrency by itself.

The program below awaits two coroutines one after the other. Each one finishes completely before the next one starts, even though both are async def functions.

The fourth kind is any object whose __await__ method returns an iterator. Libraries use this to make their own types awaitable. The class below delegates to asyncio.sleep(0) so that awaiting it yields to the loop once, then hands back a stored value.

A Future is a low-level placeholder for a result that does not exist yet. Code that needs the value awaits the Future. Other code, such as a timer, a socket callback or a library internal, fills the slot later. Application code rarely creates one; you meet Futures mostly inside libraries.

A Future is always in one of three states: pending (no outcome yet), done (a result or an exception was set) or cancelled. You move it out of pending with set_result, set_exception or cancel, and you register add_done_callback to be told when that happens.

The example creates a Future by hand with loop.create_future(), attaches a callback, and schedules a timer that fills the slot after a tenth of a second. The await suspends main until then. Done callbacks run through the loop, not inline, which is why the callback prints before the awaiting line resumes.

The second await fut returned instantly with the same value. Awaiting a finished Future is cheap and can be repeated, and many tasks can wait on the same Future at once. Each of them is woken when the result arrives. A bare coroutine cannot do this: it can be awaited only once, and a second await raises RuntimeError.

A Task is a Future subclass that wraps a coroutine and schedules it on the event loop immediately. Because it is a Future, you can await it, ask whether it is done, and collect its result. Because it wraps a coroutine, the loop drives the coroutine for you. This makes the Task the unit of concurrency in asyncio.

You make one with asyncio.create_task(coro, name=...). The coroutine starts running concurrently at the next point where the current code yields to the loop. Later, await task retrieves the result. In the example, users-fetch makes progress while main awaits a different coroutine.

Once a Task exists you can inspect it. task.done() is true once it has finished or been cancelled. task.result() returns the value, or re-raises the exception the coroutine ended with. task.exception() returns that exception, or None if there was none. Calling task.cancel() injects CancelledError at the point where the task is currently awaiting.

Putting the three kinds side by side shows why each exists. Write coroutines, start them as Tasks when you want concurrency, and leave raw Futures to library code.

Tasks are easy to start and easy to lose track of. Two failures come up again and again, and both are silent.

The fix is to hold a strong reference for as long as the task runs. The usual pattern keeps tasks in a set and removes each one when it finishes. A TaskGroup does the same bookkeeping for you, since it keeps strong references to every task created through it.

Retrieve every task's outcome. Either await the task, or attach a done callback that looks at the exception. The callback below first skips cancelled tasks, because calling exception() on a cancelled task raises CancelledError.

An await does not always pause anything. Whether control returns to the event loop depends on what sits at the very bottom of the chain of awaits. To see why, follow one Task as it drives a stack of nested coroutines.

The Task starts the chain by calling send(None) on its top coroutine. Each await on a coroutine simply runs that coroutine's body, so the chain goes down through main(), then fetch(), then read(). None of these steps suspends anything. The descent stops only when it reaches a Future that is still pending, meaning its result has not arrived yet.

The key detail is what Future.__await__ does at that point. It does not return, and it does not block. It yields the Future object itself. A yield inside a coroutine chain passes straight up through every await above it, so the Future travels back through read(), fetch() and main() until it comes out of the send() call in the Task. As a result, the Task always learns exactly which Future the whole stack is waiting on.

You can watch this happen without a Task. The example below drives a two-level chain by hand. The first send(None) returns the very same Future object that the innermost await was waiting on. After the Future gets a result, a second send(None) resumes the chain right after the await and finishes it.

import asyncio

async def read(fut):
    return await fut

async def fetch(fut):
    return await read(fut)

async def main():
    loop = asyncio.get_running_loop()
    fut = loop.create_future()
    coro = fetch(fut)
    yielded = coro.send(None)
    print(yielded is fut)
    print(fut.done())
    fut.set_result(42)
    try:
        coro.send(None)
    except StopIteration as e:
        print(e.value)

asyncio.run(main())

Playing the Task's role by hand

Once the Task holds the pending Future, it calls add_done_callback(self.__wakeup) on it and then returns to the event loop. Returning is the moment the suspension happens. The Python frames of main(), fetch() and read() stay frozen in memory, and the loop is free to run any other ready task.

The Future does not call the Task directly. When set_result runs, the done callback is queued with the loop, and the loop runs Task.__step on its next pass. The step calls send(), the innermost await returns the result x, and the code continues from exactly where it stopped.

The reverse case matters just as much. If you await a coroutine whose body never reaches a pending Future, nothing is ever yielded. The await behaves like an ordinary function call, running the other coroutine to completion before the caller moves on. No other task gets a turn in the meantime.

The next example shows this. main creates a background task and then awaits inner twice. Those awaits never suspend, so the background task has to wait until main reaches a real yield, await asyncio.sleep(0).

Each time you reach an await, you can decide whether it suspends by asking one question about what the chain ends in.

Sometimes you want to give up control even though you are not waiting on anything. await asyncio.sleep(0) is the standard idiom for this. Internally it is a bare yield: it hands control back to the Task without any Future attached, so the Task goes straight back onto the ready queue. Every other task that is already ready gets a turn first, and then this one continues.

Here two workers each print three lines and call await asyncio.sleep(0) after each one. Because every await suspends, the two tasks alternate.

Now remove the await from the loop. The coroutine is still async def, but nothing in it ever suspends. Task A runs to the end without being interrupted, and only then does B get its first turn.

The same rule has a serious consequence. The loop can run only one thing at a time, so a coroutine that holds the thread without suspending blocks every other task. This happens in two ways: a tight loop with no await, or a loop whose awaits all run straight through without suspending.

To find the culprit, run with asyncio.run(main(), debug=True). Debug mode logs any callback that holds the loop for more than 100 ms. The usual fix is to add await asyncio.sleep(0) every so many iterations. For heavy work, move it off the loop, as the later sections on to_thread and process pools show.

Cancellation depends on suspension points in the same way. task.cancel() does not interrupt running code. It arranges for CancelledError to be thrown into the coroutine at the await where it is currently suspended. A coroutine that suspends often is cancelled promptly, as in the example below.

Now suppose the loop body were only n += 1 with no await. The task would never reach a suspension point, so the loop would never get a chance to deliver the cancellation. The task would keep running, and the whole program would hang.

Putting async on a function does not make anything overlap. A coroutine runs inside whoever awaits it, so await a() followed by await b() finishes a completely before b even starts. The total is time(a) + time(b), exactly as if the code were synchronous.

To overlap the waits you hand several awaitables to the loop at once. asyncio.gather(a(), b()) wraps each coroutine in a Task, lets them run side by side, and gives you one list back. The total drops to about max(time(a), time(b)), because the loop runs the other task while one is waiting. The list holds the results in argument order, not in the order the tasks finished.

import asyncio, time

async def a():
    await asyncio.sleep(0.2)
    return 'A'

async def b():
    await asyncio.sleep(0.4)
    return 'B'

async def sequential():
    start = time.perf_counter()
    ra = await a()
    rb = await b()
    print('sequential', [ra, rb], f'{time.perf_counter() - start:.1f}s')

async def concurrent():
    start = time.perf_counter()
    results = await asyncio.gather(b(), a())
    print('gather    ', results, f'{time.perf_counter() - start:.1f}s')

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

a finishes first, but gather(b(), a()) still returns b's result first

By default, if one awaitable raises, gather re-raises that first exception to you straight away. The other awaitables are not cancelled; they keep running in the background, and you no longer see their results. Passing return_exceptions=True changes this: every exception is placed in the result list as an ordinary value, and gather itself never raises for them. You then sort the failures from the successes yourself.

Python 3.11 added asyncio.TaskGroup, which ties the lifetime of your tasks to a block of code. You open it with async with, start work through tg.create_task(...), and the block does not finish until every task inside it is done. Nothing can outlive the block, which is why this style is called structured concurrency.

Each create_task call returns a Task. After the block exits you read its outcome with t.result(). This is how you get values out, since a TaskGroup does not build a result list for you.

The failure behaviour is where TaskGroup differs most. When one task raises, the group cancels all its sibling tasks, waits for them to finish cancelling, and then raises the collected errors together as an ExceptionGroup. You catch them with except*, which matches by exception type inside the group.

import asyncio

async def slow():
    try:
        await asyncio.sleep(10)
    except asyncio.CancelledError:
        print('slow cancelled')
        raise

async def fail():
    await asyncio.sleep(0.1)
    raise ValueError('boom')

async def with_failure():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(slow())
            tg.create_task(fail())
    except* ValueError as eg:
        print('caught', [str(e) for e in eg.exceptions])

asyncio.run(with_failure())

fail() raises after 0.1 s, so the group cancels slow() instead of waiting 10 s

Sometimes you do not want to wait for everything. asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED) returns as soon as one task finishes, handing back two sets: done and pending. It does not raise the tasks' errors; you inspect each task in done yourself, and any task still pending is yours to cancel or keep waiting on. asyncio.as_completed() instead gives you the awaitables in the order they finish, so you can act on the fastest result first.

Two forms put a time limit on work. async with asyncio.timeout(5): (3.11+) limits a whole block of awaits, while await asyncio.wait_for(coro, 5) limits a single awaitable. When time runs out, both cancel the inner work and then raise TimeoutError to the caller. Catch that error outside the timeout block, because code inside the block only sees the cancellation.

Concurrency is only useful up to a point. If you build 10,000 coroutines and gather them, you may try to open 10,000 sockets at once. A Semaphore(n) is a counter that lets only n tasks into a guarded section at a time; the rest wait their turn. Wrap the expensive call in async with sem: and everything else stays the same.

import asyncio

active = 0
peak = 0
sem = asyncio.Semaphore(3)

async def fetch(i):
    global active, peak
    async with sem:
        active += 1
        peak = max(peak, active)
        await asyncio.sleep(0.05)
        active -= 1
        return i

async def main():
    results = await asyncio.gather(*(fetch(i) for i in range(10)))
    print(len(results), 'done, peak', peak)

asyncio.run(main())

ten tasks are started, but never more than three are inside the semaphore

Two mistakes account for most disappointing asyncio programs. The first silently throws away the concurrency you wanted. The second breaks the machinery that cancellations and timeouts depend on.

import asyncio

async def worker():
    try:
        await asyncio.sleep(10)
    except asyncio.CancelledError:
        print('cleaning up')
        raise

async def main():
    t = asyncio.create_task(worker())
    await asyncio.sleep(0.1)
    t.cancel()
    try:
        await t
    except asyncio.CancelledError:
        print('cancelled', t.cancelled())

asyncio.run(main())

the re-raise is what lets the task finish as cancelled

The event loop runs on a single thread, and it can only switch between tasks when a task reaches an await that actually suspends. A blocking call never gives the thread back. time.sleep, requests.get, a synchronous database driver and a read of a very large file all hold the thread until they return. While one of them runs, no other task moves, no timer fires and no socket is serviced.

The example below makes this visible. A ticker task wants to print every 0.1 seconds. We let it start, then run a coroutine that calls time.sleep(0.35). The ticker is due again after 0.1 seconds, but the loop cannot run it until the sleep returns.

import asyncio
import time

async def ticker():
    for i in range(3):
        print('tick', i)
        await asyncio.sleep(0.1)

async def blocker():
    print('blocking start')
    time.sleep(0.35)  # holds the only thread
    print('blocking end')

async def main():
    t = asyncio.create_task(ticker())
    await asyncio.sleep(0)  # let the ticker start
    await blocker()
    await t

asyncio.run(main())

The ticker is frozen for the whole blocking call

Notice that tick 1 only appears after blocking end. The ticker was ready to run long before that, but nothing could schedule it. In a real server the frozen ticker would be every other client, every heartbeat and every timeout.

The rule that follows is simple. Inside async code, either use a library that is built for asyncio, or move the blocking call off the loop thread so the loop stays free.

The simplest way to offload a blocking function is await asyncio.to_thread(fn, *args, **kwargs), available since Python 3.9. It runs fn in the loop's default thread pool and suspends your coroutine until the function finishes. While the worker thread is busy, the loop is free to run every other task. The await evaluates to the function's return value, or re-raises the exception it raised.

Here is the earlier example fixed. The blocking time.sleep now happens in a worker thread, and the ticker keeps its rhythm.

import asyncio
import time

async def ticker():
    for i in range(3):
        print('tick', i)
        await asyncio.sleep(0.1)

async def offloaded():
    print('offloaded start')
    await asyncio.to_thread(time.sleep, 0.35)
    print('offloaded end')

async def main():
    t = asyncio.create_task(ticker())
    await asyncio.sleep(0)
    await offloaded()
    await t

asyncio.run(main())

Same blocking sleep, but the loop stays responsive

to_thread has two conveniences. Keyword arguments pass straight through to your function, so await asyncio.to_thread(requests.get, url, timeout=5) just works. It also copies the current contextvars into the worker thread, so things like a request id set before the call are still visible inside it.

The lower-level tool is the loop's own method. You get the running loop and call await loop.run_in_executor(None, fn, arg). Passing None as the first argument selects the loop's default ThreadPoolExecutor. This method accepts positional arguments only. If your function needs keywords, bind them first with functools.partial(fn, key=val). Passing keywords directly raises a TypeError.

The next example shows both calls on the same function. It also shows a difference that matters: to_thread carries the contextvar into the thread, while run_in_executor does not.

import asyncio
import contextvars
import functools

request_id = contextvars.ContextVar('request_id', default='none')

def read_config(path, encoding='utf-8'):
    return f'{path} ({encoding}) rid={request_id.get()}'

async def main():
    request_id.set('r-42')
    a = await asyncio.to_thread(read_config, 'app.cfg', encoding='latin-1')
    loop = asyncio.get_running_loop()
    job = functools.partial(read_config, 'app.cfg', encoding='latin-1')
    b = await loop.run_in_executor(None, job)
    print(a)
    print(b)

asyncio.run(main())

Keywords go through partial; contextvars are copied only by to_thread

The first argument of run_in_executor is the reason to use it over to_thread. Instead of None, you can pass any executor you built. Passing a ProcessPoolExecutor runs the function in a separate Python process, so CPU-heavy work really does run in parallel on several cores. The price is that the function, its arguments and its result all travel between processes by pickling. That means the function must be defined at module top level, not as a lambda, and the entry point should sit behind if __name__ == '__main__':.

import asyncio
from concurrent.futures import ProcessPoolExecutor

def crunch(n):
    return sum(i * i for i in range(n))

async def main():
    loop = asyncio.get_running_loop()
    with ProcessPoolExecutor(max_workers=2) as pool:
        results = await asyncio.gather(
            loop.run_in_executor(pool, crunch, 1000),
            loop.run_in_executor(pool, crunch, 2000),
        )
    print(results)

if __name__ == '__main__':
    asyncio.run(main())

CPU work in two real processes; the loop only waits for the results

You may wonder how the loop can await something that runs in another thread or process. The executor's submit returns a concurrent.futures.Future, which is a thread-world object that the loop cannot await. run_in_executor wraps it with asyncio.wrap_future, which creates an asyncio Future tied to the loop. When the worker finishes, the result is handed back to the loop thread safely and your coroutine resumes.

The default thread pool is not unlimited. Its size is min(32, os.cpu_count() + 4) workers. If you start more simultaneous blocking calls than that, the extra ones wait in the executor's queue until a worker is free. Their coroutines are simply suspended, so the loop stays healthy, but the work itself is delayed. For heavy fan-out, pass your own ThreadPoolExecutor(max_workers=...) to run_in_executor, or cap concurrency with a semaphore.

Cancelling a task normally stops its coroutine at the current await. A thread is different: Python offers no safe way to kill a running thread. If you cancel a task that is awaiting to_thread, only the waiting side is cancelled. The function keeps running in its worker thread until it returns on its own, and its result is thrown away.

import asyncio
import time

def slow():
    time.sleep(0.3)
    print('thread finished anyway')

async def main():
    task = asyncio.create_task(asyncio.to_thread(slow))
    await asyncio.sleep(0.05)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print('awaiting side cancelled')
    await asyncio.sleep(0.5)  # outlive the thread

asyncio.run(main())

The cancelled await returns at once, the thread does not

Everything in this chapter so far has made waiting cheap. A coroutine that waits on a socket is parked, and the loop runs something else. That only helps when there is something to wait for. asyncio runs every coroutine on one thread, so at any moment exactly one piece of Python code is executing. The loop can interleave tasks, but it can never run two of them at the same instant.

CPU-bound code (hashing, parsing, number crunching) never waits on anything. It has no await that actually suspends, so it never hands control back. Putting it inside async def changes nothing about how it runs. You only add the overhead of creating a coroutine and a task.

The next program makes four coroutines burn CPU and starts them together with gather. Each one prints a line when it starts and when it finishes. If they overlapped, you would see all four starts first. The order shows they do not.

import asyncio

def burn(n):
    total = 0
    for i in range(n):
        total += i * i
    return total

async def crunch(name):
    print(name, 'start')
    burn(2_000_000)  # pure CPU, no await
    print(name, 'done')

async def main():
    await asyncio.gather(crunch('A'), crunch('B'), crunch('C'), crunch('D'))

asyncio.run(main())

Four async functions, all CPU work

Each coroutine ran from start to finish before the next one could begin, because nothing in burn ever gave the loop a chance to switch. Now scale the work so each call takes about 1 s of CPU. The gather of four takes about 4 s, not 1 s. For that whole time the loop is frozen: no timers fire, no other client is served, and no task can be cancelled.

The usual escape from a blocked loop is to_thread, which you met in the previous section. For blocking I/O it works well, because a thread waiting on the network lets go of the interpreter. For pure-Python CPU work it does not help. On standard CPython the GIL (global interpreter lock) lets only one thread execute Python bytecode at a time. Four threads doing arithmetic take turns, and the total time is about the same as running them one after another.

There are two exceptions. Some C extensions release the GIL while they work in native code: NumPy on large arrays, hashlib on large inputs, and zlib. Several to_thread calls into these can run on separate cores at once. The other exception is the free-threaded CPython build (3.13t and later), which has no GIL and allows real thread parallelism. It is opt-in and not yet the default, so you cannot assume it on a user's machine.

The general fix is a ProcessPoolExecutor driven through loop.run_in_executor. Each worker process has its own interpreter and its own GIL, so the pool scales with the number of cores. The loop only awaits the futures, so it stays responsive while the workers crunch. The example below sends four jobs to four processes and prints a small checksum of each result.

import asyncio
from concurrent.futures import ProcessPoolExecutor

def burn(n):
    total = 0
    for i in range(n):
        total += i * i
    return total

async def main():
    loop = asyncio.get_running_loop()
    with ProcessPoolExecutor(max_workers=4) as pool:
        futs = [loop.run_in_executor(pool, burn, 1_000_000 + i)
                for i in range(4)]
        results = await asyncio.gather(*futs)
    for r in results:
        print('checksum', r % 1000)

if __name__ == '__main__':
    asyncio.run(main())

Four CPU jobs in four processes

Two rules come with processes. The worker function must be a top-level def, not a lambda, so it can be pickled. And the entry point needs the if __name__ == '__main__': guard, because the child processes import your module.

Processes are not free. Each one costs startup time, and every argument and result is pickled and sent between processes. That overhead is fixed per task, so the pool only pays off when each task is chunky.

Sometimes you cannot or do not want to use processes, but you still must not freeze the loop. The middle ground is to split a long CPU loop into slices and put await asyncio.sleep(0) between them. sleep(0) is a bare yield: it lets every other ready task run once and then resumes you. Heartbeats, timeouts and other clients get a turn between slices.

import asyncio

def burn(n):
    total = 0
    for i in range(n):
        total += i * i
    return total

async def heartbeat():
    for i in range(3):
        print('heartbeat', i)
        await asyncio.sleep(0)

async def crunch_chunked(chunks):
    for i in range(chunks):
        burn(200_000)  # one CPU slice
        print('chunk', i)
        await asyncio.sleep(0)

async def main():
    await asyncio.gather(heartbeat(), crunch_chunked(3))

asyncio.run(main())

A heartbeat keeps running between CPU slices

The heartbeat now runs between the slices instead of waiting for all the crunching to end. The work is responsive, but not faster. The total CPU time is the same, or slightly more because of the extra switches. Chunking fixes the frozen loop, not the speed.

Choosing a tool comes down to what the work is actually doing. Ask what the code spends its time on, in this order.

Real programs often mix these. A server may accept thousands of connections, which is I/O, and also need to parse each payload, which is CPU. The answer is an asyncio front end with a process pool behind it. The loop handles the connections and hands each heavy job to the pool. Create the pool once and reuse it.

This section is a quick reference for everything the chapter covered. Each entry has a short explanation and a snippet you can copy. We start with the basic moves: define a coroutine, run it, and decide whether things happen one after another or together.

You define work with async def. Calling that function does not run it. It only gives you a coroutine object. The bridge from ordinary code into the event loop is asyncio.run(main()). Call it once, at the top level of your program, and let everything else happen inside main().

Inside main() you choose how the work is arranged. Two await lines in a row run strictly one after the other, even when both functions are async. To overlap them, hand both coroutines to asyncio.gather. It starts them together and returns their results in the order you passed them, not the order they finished.

import asyncio

async def a():
    await asyncio.sleep(0.2)
    print('a done')
    return 'A'

async def b():
    await asyncio.sleep(0.1)
    print('b done')
    return 'B'

async def main():
    first = [await a(), await b()]
    print('sequential:', first)
    second = await asyncio.gather(a(), b())
    print('together:', second)

asyncio.run(main())

Sequential finishes a then b. Gather lets the shorter b finish first, yet the results stay in argument order.

Sometimes you want work to start now and collect its result later. asyncio.create_task(coro()) wraps the coroutine in a Task and schedules it on the loop right away. You carry on with other awaits, and when you need the answer you await the task. The loop keeps only weak references to tasks, so keep your own reference to every task you create. A task you do not hold can be garbage-collected in the middle of its run.

On Python 3.11 and newer, asyncio.TaskGroup is the structured way to do this. Tasks created through the group are held for you. The async with block does not exit until every task has finished. If one task fails, its siblings are cancelled and the errors arrive together as an ExceptionGroup, which you catch with except*.

import asyncio

async def job(name, delay):
    await asyncio.sleep(delay)
    print(name, 'finished')
    return name.upper()

async def main():
    t = asyncio.create_task(job('report', 0.1))
    print('doing other work')
    result = await t
    print('got', result)

    async with asyncio.TaskGroup() as tg:
        t1 = tg.create_task(job('one', 0.2))
        t2 = tg.create_task(job('two', 0.1))
    print(t1.result(), t2.result())

asyncio.run(main())

The task starts at create_task, not at await. The group block waits for both tasks.

Network calls can hang, so give them a deadline. async with asyncio.timeout(s): limits a whole block of awaits. await asyncio.wait_for(coro, s) limits a single awaitable. Both cancel the inner work when time runs out and then raise TimeoutError. Since 3.11, asyncio.TimeoutError is simply the built-in TimeoutError.

import asyncio

async def slow():
    await asyncio.sleep(1)
    return 'slow result'

async def main():
    try:
        async with asyncio.timeout(0.1):
            await slow()
    except TimeoutError:
        print('block timed out')
    try:
        await asyncio.wait_for(slow(), 0.1)
    except TimeoutError:
        print('wait_for timed out')
    await asyncio.sleep(0)
    print('still running')

asyncio.run(main())

Catch TimeoutError outside the timeout block. Code inside the block only ever sees CancelledError.

The last line of that example used await asyncio.sleep(0). This is the way to yield to the loop on purpose. It suspends the current task for one turn so other ready tasks can run, then resumes it. Use it every so often inside a long loop that has no other await. The opposite of this is time.sleep(). It blocks the only thread and freezes every task, so it never belongs in async code. Use await asyncio.sleep(s) instead.

When you must call blocking synchronous code, move it off the loop thread. await asyncio.to_thread(fn, *args) runs fn in the default thread pool, passes keyword arguments straight through, and gives you the return value or re-raises the error. For CPU-heavy work, threads do not help in standard CPython, so use a process pool: await loop.run_in_executor(ProcessPoolExecutor(), fn, x). The function and its arguments must be picklable, so use a top-level def, not a lambda. run_in_executor takes positional arguments only, so wrap keywords with functools.partial(fn, k=v).

import asyncio
import time

def blocking(n):
    time.sleep(0.1)
    return n * 2

async def ticker():
    for _ in range(3):
        print('tick')
        await asyncio.sleep(0.03)

async def main():
    result, _ = await asyncio.gather(asyncio.to_thread(blocking, 21), ticker())
    print(result)

asyncio.run(main())

The blocking call runs in a worker thread, so the ticker keeps going.

Inside a coroutine, get the running loop with asyncio.get_running_loop(). The loop is not thread-safe, so a plain worker thread must not touch it directly. From another thread, hand a coroutine over with asyncio.run_coroutine_threadsafe(coro, loop). It returns a concurrent future, and calling .result(timeout) on it blocks that thread until the loop has produced the answer.

Three small primitives handle most coordination. A Semaphore caps how many tasks do something at once. A Lock protects state that is read, then awaited on, then written, because another task can slip in at the await. A Queue passes items from producers to consumers.

import asyncio

sem = asyncio.Semaphore(2)
lock = asyncio.Lock()
active = 0
peak = 0
balance = 100

async def fetch(i):
    global active, peak
    async with sem:
        active += 1
        peak = max(peak, active)
        await asyncio.sleep(0.05)
        active -= 1

async def withdraw(n):
    global balance
    async with lock:
        current = balance
        await asyncio.sleep(0.01)
        balance = current - n

async def main():
    await asyncio.gather(*(fetch(i) for i in range(6)))
    print('peak', peak)
    await asyncio.gather(*(withdraw(10) for _ in range(3)))
    print('balance', balance)

asyncio.run(main())

Six fetches run, never more than two at once. Without the lock the balance would end at 90.

import asyncio

async def producer(q):
    for i in range(3):
        await q.put(i)
    await q.put(None)

async def consumer(q):
    while (item := await q.get()) is not None:
        print('consumed', item)

async def main():
    q = asyncio.Queue()
    await asyncio.gather(producer(q), consumer(q))

asyncio.run(main())

None is used here as a stop signal.

Most asyncio bugs fall into a few familiar patterns. Learn to spot these five on sight. You can also run asyncio.run(main(), debug=True). Debug mode logs coroutines that were never awaited and any callback that runs longer than 100 ms.

Work out each answer on paper first, then open the answer. These questions test what the code does, not what the chapter said.

• asyncio gives you concurrency on one thread, never parallelism, and switches tasks only at an await that actually suspends.
• Calling an async def function returns a coroutine object. It runs only when awaited or wrapped in a Task.
• asyncio.run(main()) creates the loop, runs main, cleans up and closes. Call it once at the top level, never inside a running loop.
• A coroutine is lazy and sequential, a Task is scheduled at once and runs concurrently, and a Future is a raw result slot. Keep a reference to every Task.
• gather or a TaskGroup runs awaitables together. Awaiting them one by one adds their times.
• A blocking call or a CPU loop freezes every task, so use to_thread for blocking libraries and a process pool for CPU-bound work.