Execute an action on a dirty app (async/non-blocking). Args: app_path: Import path of the dirty app action:

Signal that the connection has been closed.

Load a dirty app class from an import path. Args: import_path: String in format 'module.path:ClassName' Returns: An ins

Stream results from a dirty app action (async). This method returns an async iterator that yields chunks from a streaming re

Write one message to a socket. Args: sock: Socket to write to data: Message dictionary to send

Get HTTP reason phrase for status code.

Drain the request body.

Create an error response message. Args: request_id: Request identifier being responded to error: Error message Returns:

Parse a dirty app specification. Supports two formats: - ``"module:Class"`` - standard format, all workers load the app - ``"module:

Read one message from an async reader. Args: reader: asyncio StreamReader Returns: Decoded message

\ Initialize the arbiter. Start listening and set pidfile if needed.

Update stream priority from PRIORITY frame. Args: weight: Priority weight (1-256), higher = more resources depends_on

Build ASGI HTTP scope from HTTP/2 request.

Close the transport safely. Calls write_eof() first if supported to signal end of writing, which helps ensure buffered data is flushe

Get or create a thread-local stash client.

Delete all entries.

Establish sync socket connection to arbiter. Raises: DirtyConnectionError: If connection fails

Build a stash operation message dict. Args: request_id: Unique request identifier (int or str) op: Stash operation code (STA

ASGI receive callable - returns body chunks or disconnect.

Send command and wait for response. Args: command: Command string (e.g., "show workers") args: Optional addi

Run lifespan startup and wait for completion. Raises: RuntimeError: If startup fails or app doesn't support lifespan

Execute a stash operation. Args: op: Operation code (STASH_OP_*) table: Table name key: Optional

Measure request duration request_time is a datetime.timedelta

Encode the 16-byte message header. Args: msg_type: Message type (MSG_TYPE_REQUEST, etc.) request_id: Unique

Discard any unread body of the current message. Called before returning a keepalive connection to the poller so the socket does not a

Create HTTP/2 client.

Get PIDs of gunicorn processes. Uses ps output with proctitle if available, otherwise falls back to process tree analysis.

Handle a single request message. Supports both regular (non-streaming) and streaming responses. For streaming, detects if th

Route a request to an available dirty worker via queue. Each worker has a dedicated queue and consumer task. Requests are su

\ Stop workers :attr graceful: boolean, If True (the default) workers will be killed gracefully (ie. trying to wait for the

Iterate over values.

Write a message to async stream. Accepts dict format for backwards compatibility. Args: writer: asyncio StreamW

Build minimal WSGI-like environ dict for access logging.

Check if request is a WebSocket upgrade. Per RFC 6455 Section 4.1, the opening handshake requires: - HTTP method MUST be GET

Read exactly size bytes from the unreader.

Close the connection gracefully with GOAWAY. Args: error_code: HTTP/2 error code (default: NO_ERROR) last_stream_id:

Cleanup on shutdown.

Create a test arbiter with mocked components.

Create a client with a mock socket returning the given messages.

Drain the request body.

Get a header value by name. Args: name: Header name (case-insensitive) Returns: str: Header value, or None i

Read data from the body. Args: size: Number of bytes to read, or None for all remaining Returns: bytes: The

Mark body as complete (called when message ends).

Signal handler - NO LOGGING, just queue the signal.

Signal that the client transport has gone away.

Start the dirty arbiter and workers.

Stop the dirty arbiter.

Build ASGI WebSocket scope from parsed request.

Clean up after a worker exits. Saves the dead worker's app list to pending respawns so the replacement worker gets the same

Create a mock HTTP request.

Create an ASGIProtocol instance for testing.

Create a WebSocketProtocol instance for testing.

Determine which apps a new worker should load. Returns a list of import paths for apps that need more workers. Apps with wor

Get an available worker PID using round-robin selection. If app_path is provided, only returns workers that have loaded that

Calculate minimum number of workers required by app specs. Returns the maximum worker_count across all apps that have limits.

Close the sync connection.

Cleanup resources. Called when the dirty worker is shutting down. Use this to release resources like database connections, u

Connect to control socket. Raises: ControlClientError: If connection fails

Create a client with mock async reader/writer.

Create a worker instance for testing.

Execute an action on a dirty app (sync/blocking). Args: app_path: Import path of the dirty app (e.g., 'myapp.ml:MLApp')

Make HTTP GET request to the container.

\ Your worker subclass must arrange to have this method called once every ``self.timeout`` seconds. If you fail in accomplishing

Read data from unreader and append to bytearray buffer.

Stream response chunks from URL.

Close connection to a worker.

Create a LifespanManager instance.

Create a mock WebSocket upgrade request.

Create a mock HTTP request.

Create an ASGIProtocol instance for testing.

Create an ASGIProtocol instance for testing.

Handle a single HTTP request.

Normalize socket address to ASGI-compatible (host, port) tuple. ASGI spec requires server/client to be (host, port) tuples. IPv6 sockets retu

Handle signals.