asyncio in Practice: Tasks, Concurrency, and I/O-Bound Workloads

Tasks and Structured Concurrency

Awaiting a coroutine directly runs it to completion before the next line executes — useful, but sequential. To get concurrency you must schedule coroutines as Tasks: independent units of work the event loop can interleave. A Task is a subclass of Future that wraps a coroutine and drives it forward on the loop, running it "in the background" relative to whatever created it.

create_task vs bare coroutines

The primitive for scheduling is asyncio.create_task(coro). It hands the coroutine to the loop immediately and returns a Task handle you can later await for its result. In sequential(), the first await fully completes before the second begins. In concurrent(), both tasks are launched onto the loop before either is awaited, so their sleep intervals overlap and the whole thing finishes in roughly one second. This is the essential move: schedule first, await later. For our gateway, this is how one inbound Slack event can trigger a database lookup, a user-profile fetch, and an AI call that all wait simultaneously.

asyncio.TaskGroup (3.11+) for structured concurrency

Scheduling loose tasks works, but raises a governance problem: who guarantees they all finish, and what happens if one fails? Structured concurrency is the discipline that answers this — the rule that concurrent tasks are bounded to a visible region of code, so they cannot outlive the block that spawned them. Python 3.11 introduced asyncio.TaskGroup, an asynchronous context manager that enforces exactly this: every task created inside the async with block is guaranteed to complete (or be cancelled) before the block exits. You enter with async with asyncio.TaskGroup() as tg: and schedule work via tg.create_task(coro). When the block exits, it implicitly awaits every task in the group — no manual bookkeeping, no forgotten handles.

The safety guarantee is what makes TaskGroup the recommended default. If any task raises an exception other than CancelledError, the group cancels all remaining tasks, waits for them to unwind, and then re-raises. The collected failures (excluding cancellations) are combined into an ExceptionGroup per PEP 654, which you handle with the new except* syntax. KeyboardInterrupt and SystemExit are special-cased — re-raised immediately rather than being bundled into the group.

Figure 3.1: TaskGroup structured-concurrency lifecycle

Fire-and-forget pitfalls and keeping strong references to tasks

There is a sharp edge with detached tasks. The event loop keeps only a weak reference to a Task. If you call create_task but do not store the returned handle anywhere, the task can be garbage-collected mid-flight, and it may simply vanish — silently, with no error. The idiom is to add each task to a module-level set and remove it via add_done_callback when it completes. For most gateway work, though, the better answer is to not fire-and-forget at all: wrap the work in a TaskGroup, which holds strong references for you and surfaces exceptions instead of eating them. Truly detached background tasks (a periodic metrics flush, say) are the exception, not the rule.

Coordinating Many Coroutines

TaskGroups excel when you want all-or-nothing completion. But often you need finer control: collect results in order, react to whichever finishes first, or process a hundred items while never exceeding ten in flight. asyncio offers gather, wait, as_completed, and the Semaphore for these patterns.

asyncio.gather for parallel fan-out

asyncio.gather(*aws) runs many awaitables concurrently and returns their results as a list, in the same order you passed them — regardless of which finished first. This is the workhorse for fan-out: our gateway broadcasting one AI response to a user's Slack, Discord, and WebEx channels at once.

Its behavior on failure is the critical detail. By default (return_exceptions=False), the first exception propagates to the caller immediately — but the other awaitables keep running in the background; gather does not cancel them. This is precisely the safety gap that TaskGroup closes. With return_exceptions=True, exceptions are instead returned as elements of the result list alongside successful values, so one failed delivery does not abort the others.

Table 3.1: gather vs. TaskGroup

as_completed and wait for streaming results

Sometimes you do not want to wait for the slowest task — you want to react to results the moment each arrives. asyncio.as_completed(aws) yields awaitables in completion order, so a slow AI call does not hold up processing of the fast ones (as of Python 3.13 it is also directly async-iterable). asyncio.wait(tasks, ...) is a lower-level tool that returns a tuple of two sets, (done, pending), and — importantly — does not raise on timeout and does not cancel the pending tasks for you; you inspect and manage them yourself. It shines with return_when=asyncio.FIRST_COMPLETED for "wake me when any of these finishes" designs, such as racing a primary websocket read against a shutdown signal.

Bounding concurrency with Semaphores

Fan-out has a danger: launching one task per inbound message is fine at ten messages and catastrophic at ten thousand, where you would open ten thousand simultaneous connections and get rate-limited or memory-crushed. An asyncio.Semaphore solves this. It holds an internal counter; Semaphore(n) permits at most n coroutines to hold it at once. Acquiring decrements the counter (and blocks at zero); releasing increments it and wakes a waiter. Used as an async context manager, acquire and release happen automatically — even on exceptions. A thousand tasks may be scheduled, but only five ever execute the protected region at any instant; the rest queue at the async with for a free slot. This is the canonical way to respect an upstream's concurrency budget.

Figure 3.3: Semaphore-bounded fan-out

One subtlety worth internalizing: prefer asyncio.BoundedSemaphore. A plain Semaphore lets you call release() more often than acquire(), quietly raising the ceiling above its intended limit and permitting more concurrency than you designed for. BoundedSemaphore raises ValueError on such an over-release, turning a silent bug into a loud one.

Timeouts, Cancellation, and Synchronization

An AI backend that hangs must not hang the user. Every I/O operation in a gateway needs a bounded lifetime, and bounding it means learning how cancellation flows through asyncio. asyncio also runs coroutines cooperatively on one thread, so you never see torn-write races — but you can still get logical races when a coroutine reads shared state, awaits, then writes.

asyncio.timeout and wait_for

Python 3.11 added asyncio.timeout(delay), an async context manager that bounds the time spent inside its block. When the deadline passes, it cancels the block's current task and transforms the resulting CancelledError into a TimeoutError, which you catch outside the async with. These context managers nest safely, and a sibling, asyncio.timeout_at(when), takes an absolute deadline on the loop's clock. The older asyncio.wait_for(aw, timeout) wraps a single awaitable, cancels it on expiry, and raises TimeoutError directly. The distinction: timeout() scopes a region and lets code after the block continue; wait_for() targets one awaitable and raises immediately. To protect a critical task from being cancelled by an outer timeout, wrap it in asyncio.shield().

How cancellation propagates via CancelledError

Both timeouts and explicit task.cancel() work through one mechanism: asyncio.CancelledError. When a task is cancelled, this exception is raised at its next await point, unwinding the coroutine stack like any exception. The design detail that makes cancellation robust is that CancelledError subclasses BaseException, not Exception — so a broad except Exception: handler will not accidentally swallow a cancellation. Cancellation is cooperative: a coroutine that never awaits cannot be cancelled, because there is no yield point at which to inject the exception.

Figure 3.4: Cancellation and timeout propagation

Writing cancellation-safe cleanup with try/finally

Because a CancelledError can fire at any await, resources acquired before that point must be released even when the coroutine is torn down mid-flight. The rule is: use try/finally for cleanup, and if you catch CancelledError to do work, re-raise it so the cancelled state propagates. This matters doubly in structured code: TaskGroup and asyncio.timeout are implemented using cancellation internally. A coroutine that catches CancelledError and silently swallows it can break these mechanisms — the timeout never "takes," the group never unwinds. For our gateway, try/finally around a websocket consumer is what guarantees connections close and in-flight messages flush when systemd sends SIGTERM.

asyncio.Lock, Event, and Condition

An asyncio.Lock guarantees mutual exclusion across an await: only one coroutine holds it at a time, so a read-modify-write sequence stays atomic. Without the lock, two coroutines could both read count as 5 across an await and both write 6, losing an update. An asyncio.Event is a one-to-many signal: coroutines await event.wait() and all wake when someone calls event.set() — perfect for broadcasting "the connection is ready" or "shutdown requested." An asyncio.Condition combines a lock with waiting, letting consumers block until a predicate holds and be woken by notify(). Crucially, all of these yield to the loop while waiting, so a coroutine parked on a lock never blocks other coroutines.

asyncio.Queue for in-process producer/consumer pipelines

The most important primitive for a gateway is asyncio.Queue, the standard implementation of the producer/consumer pattern. Ingress coroutines (producers) put inbound events; a fixed pool of worker coroutines (consumers) get and process them. This decouples fast receipt from slow processing. A bounded queue (maxsize=N) additionally applies backpressure, blocking producers when full so a burst cannot exhaust memory. Two coordination methods make graceful shutdown possible: a consumer calls queue.task_done() after finishing each item, and queue.join() blocks until every put item has a matching task_done(). Because the workers block forever on get(), cancelling them after join() is the standard drain-then-shutdown sequence.

Figure 3.2: Producer/consumer pipeline via asyncio.Queue

Avoiding race conditions without blocking the loop

The golden rule ties this chapter to the last: never protect shared state with a blocking primitive. A threading.Lock or time.sleep inside a coroutine freezes the entire loop and every other coroutine with it. Always reach for the asyncio.* primitives, which cooperate with the loop. And favor bounded designs — a Semaphore or a maxsize queue — over unbounded fan-out, so a traffic spike degrades into a growing queue you can observe rather than an out-of-memory crash you cannot. For genuinely CPU-bound work that must not stall the loop, offload it with run_in_executor or to_thread rather than a lock.