Module

async_pool

AsyncPool — dedicated event loop for streaming and WebSocket connections.

Receives handoffs from SyncWorkers when the ASGI app returns a streaming response (more_body=True) or WebSocket upgrade. Wraps the socket in asyncio streams and continues the ASGI lifecycle.

3Classes1Function

Classes

StreamingHandoff 6 ▼
Handoff for HTTP streaming response (more_body=True).

Handoff for HTTP streaming response (more_body=True).

Attributes

Name Type Description
conn socket.socket —
scope dict[str, Any] —
body bytes —
request_id str | None —
worker_id int —
connection_id int —
WebSocketHandoff 5 ▼
Handoff for WebSocket upgrade.

Handoff for WebSocket upgrade.

Attributes

Name Type Description
conn socket.socket —
request RequestReceived —
client tuple[str, int] —
server tuple[str, int] —
scope dict[str, Any] —
AsyncPool 13 ▼
Dedicated event loop for streaming/SSE/WebSocket connections. Accepts handoffs from SyncWorker thr…

Dedicated event loop for streaming/SSE/WebSocket connections.

Accepts handoffs from SyncWorker threads via a thread-safe queue. Wraps sockets in asyncio streams and runs the ASGI app.

Methods

set_lifespan_state 1 ▼
Set the lifespan state dict shared with all requests.
def set_lifespan_state(self, state: dict[str, Any]) -> None
Parameters
Name Type Description
state —
set_app 1 ▼
Swap the ASGI app used for future handoffs. Safe to call while the pool is run…
def set_app(self, app: ASGIApp) -> None

Swap the ASGI app used for future handoffs.

Safe to call while the pool is running:self._appis read once per handoff (in _handle_streaming_handoff / _handle_websocket_handoff), not captured at loop start. Used by graceful reload to point a freshly built pool at the reimported app (issue #102).

Parameters
Name Type Description
app —
request_shutdown 0 ▼
Signal this pool's serve loop to stop accepting new handoffs. Distinct from th…
def request_shutdown(self) -> None

Signal this pool's serve loop to stop accepting new handoffs.

Distinct from the shared_ext_shutdown: retires only this pool so a graceful reload can drain the old generation without tearing down the whole server (issue #102).

accept_handoff 1 ▼
Accept a handoff from a SyncWorker (thread-safe).
def accept_handoff(self, handoff: HandoffRequest) -> None
Parameters
Name Type Description
handoff —
run 0 ▼
Run the event loop until shutdown (blocking).
def run(self) -> None
Internal Methods 8 ▼
__init__ 5 ▼
def __init__(self, config: ServerConfig, app: ASGIApp, *, shutdown_event: threading.Event | None = None, ssl_context: ssl.SSLContext | None = None, lifecycle_collector: LifecycleCollector | None = None) -> None
Parameters
Name Type Description
config —
app —
shutdown_event — Default:None
ssl_context — Default:None
lifecycle_collector — Default:None
_serve 0 ▼
Event loop: process handoffs until shutdown. The pool retires on its OWN ``_po…
async
async def _serve(self) -> None

Event loop: process handoffs until shutdown.

The pool retires on its OWN_pool_shutdownevent, NOT on the shared _ext_shutdown(#104). On a FULL shutdown the supervisor sets _ext_shutdown immediately but only calls request_shutdown() (which sets_pool_shutdown) AFTER the sync workers have drained — because a draining worker can still be SERVING an in-flight streaming/WebSocket request that it hands off to this pool. Retiring on _ext_shutdownwould close the pool before that late handoff arrives, orphaning the connection (empty response). A safety deadline keyed off _ext_shutdownguarantees the pool still self-retires even if request_shutdown() is never delivered (e.g. a supervisor fault).

_start_handoff 1 ▼
Schedule a handoff as a tracked task.
def _start_handoff(self, handoff: HandoffRequest) -> None
Parameters
Name Type Description
handoff —
_drain_queued_handoffs 1 ▼
Pull and start any handoffs that land during the drain window. On a FULL shutd…
async
async def _drain_queued_handoffs(self, timeout: float) -> None

Pull and start any handoffs that land during the drain window.

On a FULL shutdown the sync workers retire concurrently and hand off their last in-flight streaming/WebSocket requests to this pool. The pool may observe shutdown and exit its main loop before those handoffs are enqueued, so keep polling the queue here until it has stayed empty AND no handoff task is in flight for a sustained idle interval (the workers have finished enqueuing). Bounded by timeout so this can never spin forever; in-flight tasks themselves are awaited by _drain_handoffs.

Parameters
Name Type Description
timeout —
_drain_handoffs 1 ▼
Await in-flight handoff tasks up to *timeout*, then cancel stragglers.
async
async def _drain_handoffs(self, timeout: float) -> None
Parameters
Name Type Description
timeout —
_handle_handoff_async 1 ▼
Handle a handoff (async task).
async
async def _handle_handoff_async(self, handoff: HandoffRequest) -> None
Parameters
Name Type Description
handoff —
_handle_streaming_handoff 1 ▼
Handle HTTP streaming handoff: wrap socket, run app from scratch.
async
async def _handle_streaming_handoff(self, handoff: StreamingHandoff) -> None
Parameters
Name Type Description
handoff —
_handle_websocket_handoff 1 ▼
Handle WebSocket handoff: wrap socket, run handle_websocket.
async
async def _handle_websocket_handoff(self, handoff: WebSocketHandoff) -> None
Parameters
Name Type Description
handoff —

Functions

_create_h1_protocol 1 H1Protocol ▼
Create an HTTP/1.1 protocol handler.
def _create_h1_protocol(*, max_incomplete_event_size: int | None = None) -> H1Protocol
Parameters
Name Type Description
max_incomplete_event_size int | None Default:None
Returns
H1Protocol