Module

_concurrency

Shared structured-concurrency helpers.

Single source of truth for theFIRST_COMPLETEDrace that several request/connection handlers run: the ASGI app task is raced against a companion task (disconnect monitor, body reader, frame reader, ...) and whichever finishes first wins. The losing task(s) are always cancelled and awaited so no orphaned task is leaked back into the event loop.

Callers keep their own divergent winner handling — these helpers only own the part that is byte-identical everywhere (theasyncio.waitplus the cancel-and-drain of losing tasks) and hand back the(done, pending)split so each site can inspect which task won.

3Functions

Functions

cancel_and_drain 1 None ▼
Cancel every task in *tasks* and await it, swallowing cancellation. Guarantees…
async
async def cancel_and_drain(tasks: Iterable[asyncio.Task[object]]) -> None

Cancel every task in tasks and await it, swallowing cancellation.

Guarantees each task is cancelled and awaited under contextlib.suppress(asyncio.CancelledError)so no losing task is left dangling on the event loop. Safe to call with an empty iterable.

Parameters
Name Type Description
tasks Iterable[asyncio.Task[object]]
race_first_completed 1 tuple[set[asyncio.Task[o… ▼
Race *tasks* until the first one completes, then drain the losers. Awaits `wai…
async
async def race_first_completed(*tasks: asyncio.Task[object]) -> tuple[set[asyncio.Task[object]], set[asyncio.Task[object]]]

Race tasks until the first one completes, then drain the losers.

Awaitswait() with return_when=FIRST_COMPLETEDand then guarantees that every still-pending (losing) task is cancelled and awaited via cancel_and_drain(), so no task is left dangling on the loop.

The returnedpendingset therefore contains tasks that are already finished (cancelled and drained); callers must not re-await them or call .result() on them. Winner handling should operate on the doneset.

This is the right helper when every loser should be abandoned (the WebSocket and HTTP/2 WebSocket bridges). When a losing task must instead be allowed to run to completion (e.g. the ASGI app must finish its response after the request body has been fully read), use wait_first_completed() together with cancel_and_drain() so the caller controls exactly which tasks are cancelled.

Parameters
Name Type Description
*tasks asyncio.Task[object]
Returns
tuple[set[asyncio.Task[object]], set[asyncio.Task[object]]]
wait_first_completed 1 tuple[set[asyncio.Task[o… ▼
Await `wait`() with ``FIRST_COMPLETED`` and return the split. A thin wrapper t…
async
async def wait_first_completed(*tasks: asyncio.Task[object]) -> tuple[set[asyncio.Task[object]], set[asyncio.Task[object]]]

Awaitwait() with FIRST_COMPLETEDand return the split.

A thin wrapper that does not touch the pending tasks, for callers whose winner handling needs to inspectdone/pendingand selectively let a losing task finish before draining the rest (via cancel_and_drain()).

Parameters
Name Type Description
*tasks asyncio.Task[object]
Returns
tuple[set[asyncio.Task[object]], set[asyncio.Task[object]]]