Build a queue-based admission control system in Python

Implement a simple bounded-queue admission controller that accepts or rejects incoming requests based on current queue capacity.

Easy Python 3.9+ Aug 9, 2026 Reliability & rate limiting 16 views 0 copies

admission-control queue rate-limiting collections-deque simulation

Python code

40 lines

Python 3.9+

from collections import deque
import time


class AdmissionControl:
    """Simple admission control using a bounded queue.

    Requests arrive at the queue; they are admitted in FIFO order.
    If the queue is full, the incoming request is rejected.
    """

    def __init__(self, capacity: int):
        self.capacity = capacity
        self.queue = deque()

    def admit(self, request_id: str) -> str:
        if len(self.queue) >= self.capacity:
            return f"REJECTED: {request_id} (queue full)"
        # Simulate arrival timestamp in seconds
        self.queue.append((request_id, time.time()))
        return f"ADMITTED: {request_id}"

    def process_one(self) -> str:
        if not self.queue:
            return "NO_REQUESTS"
        request_id, arrival_time = self.queue.popleft()
        return f"PROCESSED: {request_id} (arrived at {arrival_time:.2f})"

    def pending_count(self) -> int:
        return len(self.queue)


if __name__ == "__main__":
    controller = AdmissionControl(capacity=3)
    for req in ["A", "B", "C", "D", "E"]:
        print(controller.admit(req))
    print(f"Pending: {controller.pending_count()}")
    for _ in range(4):
        print(controller.process_one())
    print(f"Pending after processing: {controller.pending_count()}")

Output

stdout

ADMITTED: A
ADMITTED: B
ADMITTED: C
REJECTED: D (queue full)
REJECTED: E (queue full)
Pending: 3
PROCESSED: A (arrived at 123456.78)
PROCESSED: B (arrived at 123456.78)
PROCESSED: C (arrived at 123456.78)
NO_REQUESTS
Pending after processing: 0

How it works

This implementation uses a collections.deque as a FIFO queue, which gives O(1) appends and pops from both ends. The capacity check before appending ensures that the queue never exceeds the configured bound, and any request that arrives when the queue is full is immediately rejected. Arrival timestamps are captured with time.time() to simulate real-world admission timing. Processing dequeues items from the left, preserving the order in which they were admitted. The pending_count method provides visibility into current load, which is useful for monitoring and autoscaling decisions.

Common mistakes

Using a regular list instead of deque – list pop(0) is O(n) and becomes slow under load.
Forgetting to check capacity before adding, leading to unbounded queue growth and memory pressure.
Not simulating arrival time, losing visibility into latency and queueing delay.
Assuming that admission control alone prevents overload – you still need processing capacity tuning.

Variations

Replace the deque with `asyncio.Queue` and run admission control inside an async event loop.
Add a time-to-live (TTL) so requests expire from the queue if they wait too long.
Introduce weighted priorities instead of strict FIFO by using a `heapq` or priority queue.

Real-world use cases

Rate limiting API calls by queueing requests per tenant up to a fixed capacity to protect backend services.
Bounding the number of in-flight tasks in a worker pool so that slow consumers don't accumulate unbounded load.
Simulating admission control for a load-testing tool where incoming concurrency is capped before hitting a service.

Build a queue-based admission control system in Python

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Reliability & rate limiting

Tutorials

Quizzes

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Reliability & rate limiting

Keep learning

Tutorials

Quizzes