Trio is a modern Python library designed for asynchronous I/O and concurrency. It prioritizes human usability and correctness over low-level flexibility, enforcing a concept known as structured concurrency. This approach ensures that asynchronous tasks are managed in predictable hierarchies, preventing common issues like orphaned tasks or race conditions that often plague complex asyncio implementations.

Core Architectural Concepts

Trio's design relies on three fundamental components: asynchronous functions (coroutines), cancellation scopes, and asynchronous context managers. Coroutines serve as the building blocks of concurrency, capable of suspending execution while awaiting I/O. Cancellation scopes provide a granular mechanism to terminate tasks and handle timeouts hierarchically. Furthermore, asynchronous context managers allow for the safe acquisition and release of resources within an asynchronous lifecycle using the async with statement.

TCP Network Communication

The following example demonstrates a basic TCP client. It establishes a connection to a local server, transmits a payload, and awaits a response. The open_tcp_stream function handles the connection logic, while the context manager ensures the socket is properly closed.

import trio

async def network_client(host, port):
    async with await trio.open_tcp_stream(host, port) as conn:
        payload = b"Ping from Trio client"
        await conn.send_all(payload)
        
        data = await conn.receive_some(1024)
        print(f"Server Reply: {data.decode()}")

async def main():
    await network_client("127.0.0.1", 8080)

if __name__ == "__main__":
    trio.run(main)

Asynchronous File Processing

File I/O operations can block the event loop if not handled correctly. Trio provides an asynchronous file interface that allows the event loop to proceed while waiting for disk operations. The example below reads a log file line by line without blocking other concurrent tasks.

import trio

async def process_log_file(file_path):
    async with await trio.open_file(file_path) as f:
        async for line in f:
            cleaned = line.strip()
            if cleaned:
                print(f"Log Entry: {cleaned}")

async def entry_point():
    await process_log_file("system.log")

trio.run(entry_point)

Concurrent Data Fetching

Managing multiple HTTP requests concurrently is a common use case. By utilizing a nursery, Trio can manage the lifecycle of several child tasks. In this example, we use the httpx library, which is compatible with Trio, to fetch data from multiple endpoints simultaneously.

import trio
import httpx

async def fetch_resource(client, url):
    response = await client.get(url)
    print(f"Completed {url} with status: {response.status_code}")

async def aggregate_requests():
    targets = [
        "https://www.python.org",
        "https://github.com"
    ]
    
    async with httpx.AsyncClient() as client:
        async with trio.open_nursery() as nursery:
            for target in targets:
                nursery.start_soon(fetch_resource, client, target)

trio.run(aggregate_requests)

Concurrency Limiting with Semaphores

When dealing with limited resources, such as database connections or API rate limits, it is crucial to restrict the number of concurrent operations. A Semaphore acts as a counter, allowing a defined number of tasks to proceed at once.

import trio

# Limit concurrent access to 3 tasks
rate_limiter = trio.Semaphore(3)

async def restricted_task(task_id):
    async with rate_limiter:
        print(f"Task {task_id} acquired lock")
        await trio.sleep(1)
        print(f"Task {task_id} released lock")

async def manager():
    async with trio.open_nursery() as nursery:
        for i in range(8):
            nursery.start_soon(restricted_task, i)

trio.run(manager)

Structured Task Management

Trio's open_nursery creates a scope where child tasks run concurrently. The nursery block does not exit until all child tasks have completed or raised an exception. This structure guarantees that no background task is left behind when the main function exits.

import trio

async def background_process():
    await trio.sleep(0.5)
    print("Background operation finalized")

async def primary_process():
    print("Primary operation started")
    await trio.sleep(1)
    print("Primary operation finished")

async def orchestrator():
    async with trio.open_nursery() as nursery:
        nursery.start_soon(background_process)
        await primary_process()

trio.run(orchestrator)