Concurrency

async and await

Asynchronous code runs many waiting tasks in one thread. A coroutine pauses at await, the event loop runs something else, and thousands of tasks become practical.

On this page (14)

The idea

import asyncio
import time


async def task(name, seconds):
    print(f"{name} starting")
    await asyncio.sleep(seconds)          # pause, let others run
    print(f"{name} finished")
    return f"{name} result"


async def main():
    start = time.perf_counter()
    results = await asyncio.gather(
        task("A", 2),
        task("B", 2),
        task("C", 2),
    )
    print(results)
    print(f"total {time.perf_counter() - start:.2f}s")


asyncio.run(main())

A starting
B starting
C starting
A finished
B finished
C finished
['A result', 'B result', 'C result']
total 2.00s

Six seconds of waiting in two, using a single thread and no locks. Nothing runs in parallel; the tasks simply take turns while waiting.

Coroutines

import asyncio


async def greet(name):
    return f"Hello, {name}"


coroutine = greet("Meera")
print(type(coroutine))            # <class 'coroutine'> - nothing has run yet

print(asyncio.run(greet("Meera")))

Calling an async def function does not run it; it creates a coroutine object. It runs only when awaited, or when scheduled on an event loop.

import asyncio


async def inner():
    await asyncio.sleep(0.1)
    return "inner done"


async def outer():
    result = await inner()        # await: run it and wait for the result
    return f"outer got: {result}"


print(asyncio.run(outer()))

await is only legal inside an async def function.
You can only await something awaitable: a coroutine, a task or a future.
Forgetting await gives you a coroutine object and a "never awaited" warning.

import asyncio


async def main():
    result = inner()              # WRONG - no await
    print(result)                 # <coroutine object>

    result = await inner()        # correct
    print(result)


async def inner():
    return 42


asyncio.run(main())

Running things at the same time

import asyncio
import time


async def work(name, seconds):
    await asyncio.sleep(seconds)
    return name


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


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


async def with_tasks():
    start = time.perf_counter()
    a = asyncio.create_task(work("a", 1))     # scheduled immediately
    b = asyncio.create_task(work("b", 1))
    print(await a, await b)
    print(f"tasks:      {time.perf_counter() - start:.2f}s")


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

A row of await statements is sequential. Concurrency needs gather, create_task or a task group. This is the single most common misunderstanding about async code.

Task groups

import asyncio


async def work(name, seconds):
    await asyncio.sleep(seconds)
    if name == "bad":
        raise ValueError("task failed")
    return name


async def main():
    try:
        async with asyncio.TaskGroup() as group:      # Python 3.11+
            a = group.create_task(work("a", 1))
            b = group.create_task(work("bad", 0.5))
    except* ValueError as errors:
        for error in errors.exceptions:
            print("caught:", error)


asyncio.run(main())

A task group waits for every task and cancels the rest if one fails. It is the modern replacement for bare gather, because it never leaves an orphaned task running.

import asyncio


async def work(n):
    await asyncio.sleep(0.1)
    if n == 2:
        raise ValueError("two is bad")
    return n


async def main():
    results = await asyncio.gather(work(1), work(2), work(3),
                                   return_exceptions=True)
    for result in results:
        if isinstance(result, Exception):
            print("failed:", result)
        else:
            print("ok:", result)


asyncio.run(main())

Timeouts and cancellation

import asyncio


async def slow():
    await asyncio.sleep(5)
    return "finished"


async def main():
    try:
        result = await asyncio.wait_for(slow(), timeout=1)
        print(result)
    except asyncio.TimeoutError:
        print("timed out")

    task = asyncio.create_task(slow())
    await asyncio.sleep(0.1)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("cancelled")


asyncio.run(main())

import asyncio


async def cleanly_cancellable():
    try:
        await asyncio.sleep(10)
    except asyncio.CancelledError:
        print("cleaning up")
        raise                     # always re-raise CancelledError
    finally:
        print("finished cleanup")

Async iteration and context managers

import asyncio


class Ticker:
    def __init__(self, count):
        self.count = count

    def __aiter__(self):
        self.n = 0
        return self

    async def __anext__(self):
        if self.n >= self.count:
            raise StopAsyncIteration
        await asyncio.sleep(0.1)
        self.n += 1
        return self.n


async def numbers(count):
    for i in range(count):
        await asyncio.sleep(0.1)
        yield i                       # an async generator


async def main():
    async for value in Ticker(3):
        print("ticker:", value)

    async for value in numbers(3):
        print("generator:", value)

    print([v async for v in numbers(3)])      # an async comprehension


asyncio.run(main())

import asyncio
from contextlib import asynccontextmanager


@asynccontextmanager
async def connection(name):
    print(f"opening {name}")
    await asyncio.sleep(0.1)
    try:
        yield name
    finally:
        print(f"closing {name}")
        await asyncio.sleep(0.1)


async def main():
    async with connection("db") as conn:
        print("using", conn)


asyncio.run(main())

Blocking code ruins everything

import asyncio
import time


async def blocking():
    time.sleep(2)               # WRONG - blocks the whole event loop
    return "done"


async def correct():
    await asyncio.sleep(2)      # yields control while waiting
    return "done"


async def offloaded():
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(None, time.sleep, 2)     # runs in a thread


async def main():
    start = time.perf_counter()
    await asyncio.gather(correct(), correct(), correct())
    print(f"async sleep:  {time.perf_counter() - start:.2f}s")   # about 2s

    start = time.perf_counter()
    await asyncio.gather(blocking(), blocking(), blocking())
    print(f"blocking:     {time.perf_counter() - start:.2f}s")   # about 6s


asyncio.run(main())

One blocking call freezes every task on the loop. Anything slow that is not async - a database driver, a heavy computation, a legacy library - must be pushed onto a thread or process with run_in_executor or asyncio.to_thread.

import asyncio
import time


async def main():
    result = await asyncio.to_thread(time.sleep, 1)     # Python 3.9+
    print("offloaded to a thread")


asyncio.run(main())

Limiting concurrency

import asyncio

limit = asyncio.Semaphore(3)


async def fetch(name):
    async with limit:                    # at most three at a time
        print(f"fetching {name}")
        await asyncio.sleep(0.5)
        return f"{name} done"


async def main():
    results = await asyncio.gather(*(fetch(f"page-{i}") for i in range(10)))
    print(len(results), "completed")


asyncio.run(main())

Async queues

import asyncio


async def producer(queue, count):
    for i in range(count):
        await queue.put(i)
        await asyncio.sleep(0.05)
    for _ in range(3):
        await queue.put(None)


async def consumer(queue, name):
    while True:
        item = await queue.get()
        if item is None:
            queue.task_done()
            break
        await asyncio.sleep(0.1)
        print(f"{name} handled {item}")
        queue.task_done()


async def main():
    queue = asyncio.Queue(maxsize=5)
    await asyncio.gather(
        producer(queue, 9),
        *(consumer(queue, f"c{i}") for i in range(3)),
    )


asyncio.run(main())

Choosing async

Use asyncio when	Do not when
Thousands of concurrent I/O operations	The work is pure computation
The libraries you need are async	Your libraries are all blocking
Long lived connections, sockets, streams	A handful of tasks would do with threads
You control the whole call chain	You are adding it to a small script

Async is not faster than threads for four downloads. It becomes decisive at ten thousand connections, where a thread each is not practical. It is also all or nothing in a call chain: an async function can only be awaited from another one.

Common mistakes

Forgetting await and getting a coroutine object.
Awaiting sequentially and expecting concurrency.
Calling time.sleep or any blocking function inside a coroutine.
Calling asyncio.run more than once, or inside a running loop.
Creating a task and never awaiting it, so it is garbage collected mid-flight.
Swallowing CancelledError instead of re-raising it.
Using async for CPU bound work.

Best practices

Use asyncio.run(main()) once, as the single entry point.
Use TaskGroup, or gather, for anything that should overlap.
Keep a reference to every task you create.
Push blocking calls to asyncio.to_thread.
Bound concurrency with a Semaphore when calling an external service.
Always re-raise CancelledError.

Practice

Write three coroutines and run them sequentially and then with gather, comparing the times.
Demonstrate that a time.sleep inside a coroutine blocks every other task.
Add a timeout to a slow coroutine and handle the timeout cleanly.
Build a producer and consumer pipeline with asyncio.Queue.
Limit ten simulated requests to three at a time using a semaphore.

Conclusion

Async runs many waiting tasks in one thread by letting each one yield at await. Use gather or a task group to get concurrency, never block the loop, and reach for it when the number of concurrent I/O operations is large enough that threads would not scale.

Written by Lorens Mishra

Default administrator account created by the installer.