Parsiad Azimzadeh

Much await about nothing: an intro to asyncio

The purpose of this article is to serve as a practical introduction to asyncio, a Python standard library package for concurrency. Concurrency (and, by extension, asyncio) is particularly useful when a program has multiple operations that spend time waiting for I/O. While one operation waits, another can make progress.

Concurrency vs. parallelism

People often confuse concurrency and parallelism. However, concurrency does not require simultaneous execution: operations can take turns making progress even on a single CPU. Let's define the two terms precisely:

Examples of concurrency and parallelism are given below.

Sequential (neither concurrent nor parallel)

Concurrent, not parallel

Concurrent and parallel

An example involving multiple HTTP requests

Suppose you have a list of webpages you want to fetch:

urls = [
    "https://www.example.com",
    "https://docs.python.org/3/library/asyncio.html",
]

You can fetch them sequentially with httpx as follows:

def fetch_page(client: httpx.Client, url: str) -> str:
    response = client.get(url)
    response.raise_for_status()  # Raise if the fetch failed
    return response.text


with httpx.Client() as client:
    results = [fetch_page(client, url) for url in urls]

This is a reasonable approach, especially when the number of pages is small. However, waiting for each response before starting the next request is wasteful, especially as the number of pages grows. Let's use concurrency instead:

async def async_fetch_page(client: httpx.AsyncClient, url: str) -> str:
    response = await client.get(url)
    response.raise_for_status()
    return response.text


client = httpx.AsyncClient()
results: list[str] = []
tasks: list[asyncio.Task[str]] = []
try:
    # Create one task per URL
    for url in urls:
        coro = async_fetch_page(client, url)
        task = asyncio.create_task(coro)
        tasks.append(task)
    # Await all tasks
    for task in tasks:
        result = await task
        results.append(result)
finally:
    await client.aclose()

There's a lot to unpack here if you are unfamiliar with asyncio, so let's proceed slowly:

await can only be used within a coroutine function. In a Jupyter notebook, the kernel detects when a cell requires async execution (for example, because it contains a top-level await) and handles it accordingly. That makes the code above valid for execution in a Jupyter cell. Trying to execute it in the Python interpreter yields the following error:

  File "<python-input-1>", line 18
    result = await task
             ^^^^^^^^^^
SyntaxError: 'await' outside function

To run it in the Python interpreter, wrap the top-level code in a coroutine function and call asyncio.run(my_coroutine()).

We can simplify the concurrent requests code by using list comprehensions. Doing so does not introduce any new asyncio concepts.

client = httpx.AsyncClient()
try:
    results = [
        await task
        for task in [asyncio.create_task(async_fetch_page(client, url)) for url in urls]
    ]
finally:
    await client.aclose()

We can make the code more idiomatic by passing the coroutines directly to asyncio.gather:

client = httpx.AsyncClient()
try:
    results = await asyncio.gather(*[async_fetch_page(client, url) for url in urls])
finally:
    await client.aclose()

asyncio.gather wraps each coroutine in a task and returns a special awaitable object. This object is neither a coroutine nor a task (more on it later). Awaiting on it is conceptually similar but not identical to naively awaiting on all of the tasks it generates. For example, asyncio.gather propagates an early failure immediately (without cancelling sibling tasks).

We can make the code even more idiomatic by using async with:

async with httpx.AsyncClient() as client:
    results = await asyncio.gather(*[async_fetch_page(client, url) for url in urls])

async with is analogous to with but the setup and cleanup methods are called with await:

SetupCleanup
with__enter__()__exit__(...)
async withawait __aenter__()await __aexit__(...)

In the above, we have arrived at a concurrent version that is nearly as concise as the original synchronous version of the code, while allowing requests to make independent progress.

Cooperative scheduling

asyncio uses cooperative scheduling. That is, the event loop never interrupts a task to give another task a turn. Instead, the currently executing task keeps control until it suspends or finishes.

An await expression is a possible suspension point, but it does not necessarily suspend. For example, directly awaiting a coroutine enters it immediately within the current task. This is made clear in the example below.

async def answer() -> int:
    return 42


async def announce() -> None:
    print("Hello, world")


task = asyncio.create_task(announce())
result = await answer()
print(result)
await task

42
Hello, world

With default task scheduling, creating the task schedules it but does not immediately execute it. Because answer() returns without suspending, awaiting it does not give the scheduled task a turn.

To reverse the order of execution, we can use await asyncio.sleep(0) to explicitly yield control:

task = asyncio.create_task(announce())
await asyncio.sleep(0)
result = await answer()
print(result)
await task

Hello, world
42

The rules are summarized below.

Await onDoes the current task suspend?
CoroutineIf execution inside it suspends
Task or futureIf it is pending
asyncio.sleep(0)Always

Note that unlike asyncio.sleep, an ordinary time.sleep call blocks without yielding control.

Cancellation

To request the cancellation of a task, call task.cancel(). Note that this does not immediately stop the task: asyncio arranges for CancelledError to be raised inside the coroutine at its next opportunity. This is made clear in the example below.

async def worker() -> None:
    try:
        print("Start work")
        await asyncio.sleep(60)
        print("End work")
    finally:
        print("Start cleanup")
        await asyncio.sleep(1)  # Simulate cleanup
        print("End cleanup")


task = asyncio.create_task(worker())
await asyncio.sleep(0)
print("Request cancellation")
task.cancel()
try:
    await task
except asyncio.CancelledError:
    print("Task cancelled")

Start work
Request cancellation
Start cleanup


End cleanup
Task cancelled

Often, several tasks form one operation: we want to wait for all of them, and if one fails, cancel the others and wait for their cleanup before continuing. This pattern is common enough that asyncio provides TaskGroup to handle it (available since Python 3.11).

As an example, here it is applied to the earlier example involving concurrent requests:

async with httpx.AsyncClient() as client:
    async with asyncio.TaskGroup() as group:
        tasks = [group.create_task(async_fetch_page(client, url)) for url in urls]

    results = [task.result() for task in tasks]

group.create_task(...) schedules a coroutine as a task and registers it with the group. On leaving the group's block, async with awaits its __aexit__ method, which waits for the group's tasks to finish.

If all tasks succeed, we collect their results in the original URL order. If one fails with an ordinary exception, the group requests cancellation of the others, waits for their cleanup, and raises an ExceptionGroup containing the failures. In that case, the results assignment is not reached.

Unlike the earlier asyncio.gather example, a failed request cannot leave sibling requests running after the HTTP client's context exits: the task group finishes first.

Futures

So far we have mainly dealt with tasks. In this section, we introduce futures. A plain future represents an eventual outcome supplied by other code. A task is a specialized future that also manages a coroutine's execution. This relationship is reflected in the class hierarchy:

issubclass(asyncio.Task, asyncio.Future)

True

We previously hinted at asyncio.gather returning a special awaitable object that is neither a coroutine nor a task. It turns out that this object is a future. To learn about how futures work, we introduce below a simplified implementation of asyncio.gather.

def my_gather(*awaitables: Awaitable[T]) -> asyncio.Future[list[T]]:
    loop = asyncio.get_running_loop()
    combined: asyncio.Future[list[T]] = loop.create_future()
    children: list[asyncio.Future[T]] = [asyncio.ensure_future(x) for x in awaitables]
    remaining = len(children)
    if remaining == 0:
        combined.set_result([])
        return combined

    def on_child_done(child: asyncio.Future[T]) -> None:
        nonlocal remaining
        remaining -= 1
        try:
            child.result()
        except BaseException as error:  # noqa: BLE001
            if not combined.done():
                combined.set_exception(error)
            return
        if remaining == 0 and not combined.done():
            results = [child.result() for child in children]
            combined.set_result(results)

    def on_combined_done(future: asyncio.Future[list[T]]) -> None:
        if future.cancelled():
            for child in children:
                child.cancel()

    combined.add_done_callback(on_combined_done)
    for child in children:
        child.add_done_callback(on_child_done)
    return combined