Module

_request_queue

Request queuing and load shedding for pounce.

Implements application-level request queueing with bounded capacity to gracefully handle server overload.

2Classes1Function

Classes

RequestQueue 5 ▼
Bounded request queue for load shedding. Queues incoming requests when workers are busy. Returns 5…

Bounded request queue for load shedding.

Queues incoming requests when workers are busy. Returns 503 when queue is full to shed load and prevent resource exhaustion.

Thread-safe for concurrent request handling.

Methods

acquire 0 bool ▼
Try to acquire a slot in the queue.
async
async def acquire(self) -> bool
Returns
bool True if acquired, False if queue is full
release 0 ▼
Release a slot in the queue.
def release(self) -> None
get_depth 0 int ▼
Get current queue depth.
def get_depth(self) -> int
Returns
int Number of requests currently queued
get_max_depth 0 int ▼
Get maximum queue depth.
def get_max_depth(self) -> int
Returns
int Maximum queue capacity (0 = unlimited)
Internal Methods 1 ▼
__init__ 1 ▼
Initialize request queue.
def __init__(self, max_depth: int) -> None
Parameters
Name Type Description
max_depth —

Maximum number of queued requests (0 = unlimited)
QueueMetrics 4 ▼
Metrics for request queue monitoring. Tracks queue depth, wait times, and rejection rate.

Metrics for request queue monitoring.

Tracks queue depth, wait times, and rejection rate.

Methods

record_queued 1 ▼
Record a queued request.
async
async def record_queued(self, wait_time_ms: float) -> None
Parameters
Name Type Description
wait_time_ms —

Time spent waiting in queue (milliseconds)
record_rejected 0 ▼
Record a rejected request (queue full).
async
async def record_rejected(self) -> None
get_stats 0 dict[str, Any] ▼
Get queue statistics.
def get_stats(self) -> dict[str, Any]
Returns
dict[str, Any] Dictionary with queue metrics
Internal Methods 1 ▼
__init__ 0 ▼
Initialize queue metrics.
def __init__(self) -> None

Functions

create_queue_wrapper 3 Callable ▼
Wrap an ASGI app with request queueing. Buffers requests when workers are busy…
def create_queue_wrapper(app: Callable, queue: RequestQueue, metrics: QueueMetrics | None = None) -> Callable

Wrap an ASGI app with request queueing.

Buffers requests when workers are busy. Returns 503 when queue is full.

Parameters
Name Type Description
app Callable

Original ASGI app
queue RequestQueue

RequestQueue instance
metrics QueueMetrics | None

Optional QueueMetrics instance for monitoring

Default:None
Returns
Callable