Async with asyncio: coroutines, await, gather

Understand asyncio's cooperative concurrency: lazy coroutines, suspension points, gather, TaskGroup and the pitfalls of blocking calls.

10 min · 3 questions

asyncio implements cooperative concurrency on a single thread: an event loop orchestrates coroutines that voluntarily yield control at every await. A fundamental point, often misunderstood: calling an async def function runs nothing. It builds an inert coroutine object whose body only advances when it is awaited with await or scheduled as a task on the loop.

import asyncio

async def greet() -> str:
    return "hello"

coro = greet()
print(type(coro).__name__)

async def main() -> None:
    print(await coro)

asyncio.run(main())

await marks a suspension point: the coroutine freezes there, the loop takes back control and advances another ready task. Between two awaits, your code runs without interruption - no locks needed for most structures. To launch several coroutines in parallel (in the I/O sense), wrap them in asyncio.gather: three one-second waits finish in one second total, not three.

import asyncio
import time

async def fetch(name: str, delay: float) -> str:
    await asyncio.sleep(delay)
    return f"{name} done"

async def main() -> None:
    start = time.perf_counter()
    results = await asyncio.gather(
        fetch("a", 1.0),
        fetch("b", 1.0),
        fetch("c", 1.0),
    )
    elapsed = time.perf_counter() - start
    print(results)
    print(f"total: {elapsed:.1f} s")

asyncio.run(main())

By default, gather propagates the first exception raised but lets the other tasks keep running in the background - rarely what you want. With return_exceptions=True, exceptions are returned as results, to be sorted out afterward. Since Python 3.11, prefer asyncio.TaskGroup: at the first error, all sibling tasks are cancelled cleanly, and any multiple exceptions bubble up grouped in an ExceptionGroup, to be caught with except*.

import asyncio
import time

def blocking_io() -> str:
    time.sleep(1)
    return "report written"

async def main() -> None:
    results = await asyncio.gather(
        asyncio.to_thread(blocking_io),
        asyncio.sleep(1, result="tick"),
    )
    print(results)

asyncio.run(main())

Knowledge check

Make sure you remember the key points of this lesson.

1. What does the call fetch('a', 1.0) written on its own, with no await or create_task, produce?
   • An inert coroutine object: not a single line of the function's body runs
   • The immediate, synchronous execution of the coroutine
   • A task automatically scheduled on the event loop
   • A SyntaxError, because await is mandatory
2. Three asyncio.sleep(1) launched together via asyncio.gather: what total duration?
   • About 3 seconds, since a single thread runs the code
   • Impossible to predict without knowing the number of cores
   • About 1 second: the waits overlap, the loop switches from one task to another
3. Why is time.sleep(2) banned inside a coroutine?
   • It raises an exception in an async context
   • It blocks the event loop thread: no other task advances for 2 seconds
   • It is simply less precise than asyncio.sleep
   • It uses twice as much memory