The Trio Python async concurrency library

Following along with https://trio.readthedocs.io/en/stable/tutorial.html . This is not a tutorial.

Trio is a Python library for writing asynchronous applications — programs that want to do multiple things at the same time with parallelized I/O, web spiders, web servers, etc.¹

The tutorial is at a high level.

Async functions #

An async function is defined like a normal function, except with async def instead of def:

def regular_double(x):
    return 2 * x

async def async_double(x):
    return 2 * x

To call an async function you have to use the await keyword. You can only use the await keyword inside of async functions.

How is an application’s first async function called? #

When a program starts, it’s running regular synchronous code. How is the first async function called?

Trio, and it seems other libraries, use a special synchronous run function. trio.run() takes and calls an asynchronous function:

import trio


async def async_double_print(x):
    print(2 * x)


def main():
    trio.run(async_double_print, 3)


if __name__ == "__main__":
    main()

How Python interprets calls to await #

Python interprets await trio.sleep(3) as two things. The trio.sleep(3) part is a coroutine object:

In [1]: import trio

In [2]: trio.sleep(3)
Out[2]: <coroutine object sleep at 0x7fe41dc85220>

The coroutine gets passed to await. await runs the coroutine.

Running multiple async functions at the same time #

import trio


async def child1():
    print("    child1 started. sleeping now...")
    await trio.sleep(3)
    print("child1 exiting")


async def child2():
    print("    child2 started. sleeping now...")
    await trio.sleep(3)
    print("child2 exiting")


async def parent():
    print("parent started")
    async with trio.open_nursery() as nursery:
        print("parent spawning child1...")
        nursery.start_soon(child1)

        print("parent spawning child2...")
        nursery.start_soon(child2)

        print("parent waiting for children to finish")
    print("parent done")


def main():
    trio.run(parent)  # note that main doesn't have to be async


if __name__ == "__main__":
    main()

The async with trio.open_nursery() creates an “async context manager”.

Python context managers #

In Python, the statement with someobj: instructs the interpreter to call someobj.__enter__() at the beginning of the block, and someobj.__exit__() at the end of the block.

For async code: async with someobj calls await someobj.__aenter__() and await someobj.__aexit__().

Note that Python also has async for...

There’s a related note at notes / Running coroutines in Python

nursery.start_soon #

From googling, it seems that trio.nursery.start_soon() is similar to asyncio.TaskGroup.create_task().

multiple async functions can run at (seemingly) the same time #

The output of the code above is completed in 3 seconds, not 6 seconds. The child1 and child2 functions are run at (close to) the same time.

This is similar to what can be done with threads, but it’s done within a single thread.

In async programming, instead of spawning threads the code is spawning tasks.

The difference between threads and tasks #

• many tasks can take turns running on a single thread
• with threads, the Python interpreter can switch which thread is running at any given time. With tasks, the task that is being run can only be switched at designated “checkpoints”.

Task switching at designated checkpoints #

Async/await can be used to run lots of tasks simultaneously on a single thread. For example, on a web server one task could be sending an HTTP response at the same time as another task is waiting for new connections.

This is done by switching between tasks at appropriate places.

My sense is that task switching is generally handled by libraries like asyncio or Trio.

Trio has an Instrument API that can be used to get a high level idea of what’s going on. Here’s a Tracer class that implements the Instrument interface:

import trio


class Tracer(trio.abc.Instrument):
    def before_run(self):
        print("!!! run started")

    def _print_with_task(self, msg, task):
        # repr(task) could be used here (it's more noisy)
        print(f"{msg}: {task.name}")

    def task_spawned(self, task):
        self._print_with_task("### new task spawned", task)

    def task_scheduled(self, task):
        self._print_with_task("### new task scheduled", task)

    def before_task_step(self, task):
        self._print_with_task(">>> about to run one step of task", task)

    def after_task_step(self, task):
        self._print_with_task("<<< task step finished", task)

    def task_exited(self, task):
        self._print_with_task("### task exited", task)

    def before_io_wait(self, timeout):
        if timeout:
            print(f"### waiting for I/O for up to {timeout} seconds")
        else:
            print("### doing a quick check for I/O")
        self._sleep_time = trio.current_time()

    def after_io_wait(self, timeout):
        duration = trio.current_time() - self._sleep_time
        print(f"### finished I/O check (took {duration}) seconds")

    def after_run(self):
        print("!!! run finished")

A Tracer() instance can be passed to a call to trio.run():

async def child1():
    print("    child1 started. sleeping now...")
    await trio.sleep(3)
    print("child1 exiting")


async def child2():
    print("    child2 started. sleeping now...")
    await trio.sleep(3)
    print("child2 exiting")


async def parent():
    print("parent started")
    async with trio.open_nursery() as nursery:
        print("parent spawning child1...")
        nursery.start_soon(child1)

        print("parent spawning child2...")
        nursery.start_soon(child2)

        print("parent waiting for children to finish")
    print("parent done")


def main():
    trio.run(parent, instruments=[Tracer()])


if __name__ == "__main__":
    main()

Ignoring the initial Tracer() output, first a task is created and scheduled for the parent() function:

>>> about to run one step of task: <init>
### new task spawned: __main__.parent
### new task scheduled: __main__.parent
### new task spawned: <TrioToken.run_sync_soon task>
### new task scheduled: <TrioToken.run_sync_soon task>
<<< task step finished: <init>
### doing a quick check for I/O
### finished I/O check (took 3.141991328448057e-06) seconds
>>> about to run one step of task: __main__.parent
parent started

parent() then spawns its two child tasks. It pauses at the end of its async with block:

parent spawning child1...
### new task spawned: __main__.child1
### new task scheduled: __main__.child1
parent spawning child2...
### new task spawned: __main__.child2
### new task scheduled: __main__.child2
parent waiting for children to finish
<<< task step finished: __main__.parent

After the async with block, control goes back to trio.run():

>>> about to run one step of task: <TrioToken.run_sync_soon task>
<<< task step finished: <TrioToken.run_sync_soon task>
### doing a quick check for I/O
### finished I/O check (took 2.607994247227907e-06) seconds

Then the two child tasks are given the chance to run:

>>> about to run one step of task: __main__.child1
    child1 started. sleeping now...
<<< task step finished: __main__.child1
>>> about to run one step of task: __main__.child2
    child2 started. sleeping now...
<<< task step finished: __main__.child2

Each of the child tasks run until they hit the call to trio.sleep(), then control goes back to trio.run() so that trio.run() can decide what to run next.

This task switching works because trio.sleep() “has access to some special magic” that lets it pause itself and send “a note to trio.run()” requesting that it be woken after the sleep duration. The trio.sleep() task then suspends itself (that’s my imprecise understanding.) Python then gives control back to trio.run().

This is similar to how generators can suspend execution with yield (there’s overlap between the async and generator implementations).

Only async functions have access to the “special magic” for suspending tasks.

This execution switching would be relevant in a program where run() had some other work to do during the time that the tasks are suspended, for example, deal with some I/O operation.

Note that the calls to trio.sleep() are contrived. The “special magic” that it’s using isn’t clear to me. For example, what if instead of trio.sleep() the child functions were listing for requests to an HTTP route?

Further reading #

I’ll add these references here instead of to the references section. They are recommended in the Trio tutorial:

• How the heck does async/await work in Python 3.5?
• How to build a simple async I/O framework from the ground up

References #

python-trio.trio. “Trio.” Accessed on: February 4, 2026. https://github.com/python-trio/trio .

trio.readthedocs.io. “Tutorial.” Accessed on: February 4, 2026. https://trio.readthedocs.io/en/stable/tutorial.html .