Python's asyncio Concurrency Control for Inference Replicas

Semaphore, bounded queues, and token-bucket backpressure — admission control inside one replica.

§ IFrame

The autoscaler in today's Ops lesson works in whole replicas. It can give an endpoint one more copy of the model or take one away, and it reads the queue across all copies to decide. What it cannot do is reach inside a single replica and tell it to accept fewer requests at once. That decision lives in the process, in Python, in three small primitives that decide how much work one replica admits before it pushes back.

A replica without admission control accepts every request the network hands it. Under a burst, it accepts a thousand concurrent calls, allocates a coroutine and a slice of GPU memory for each, and discovers at request seven hundred that the GPU is out of memory and every in-flight request fails together. The replica that accepted everything served nobody. Admission control is the replica's own version of load-shedding: bound what you take, queue a little, and refuse the rest cleanly.

Python's asyncio gives three tools for exactly this. A Semaphore bounds how many requests run concurrently. A bounded Queue holds a small overflow and applies backpressure when it fills. A token bucket caps the rate, which is how the cost-per-token budget becomes a runtime control rather than a monthly surprise. Build them in order; each handles a failure the others let through.

§ IILanguage Idiom

asyncio.Semaphore is a counter with two operations, acquire and release, and an async with form that pairs them so a release cannot be forgotten. Create it with a count, and at most that many coroutines hold it at once; the rest suspend at acquire until a holder releases. This is the concurrency bound. It does not slow any single request; it limits how many run together, which is precisely the GPU-memory constraint a replica lives under.

asyncio.Queue(maxsize=N) is the backpressure tool. A producer calling await queue.put(item) on a full queue does not raise and does not drop the item; it suspends until a consumer makes room. That suspension travels: the producer that cannot put is itself slowed, so the slowness propagates back toward the source instead of piling unbounded work in memory. Designing Data-Intensive Applications (Ch. 11, pp. 449–452) frames this as the central choice in any messaging system — when the consumer cannot keep up, you block, you drop, or you buffer without bound, and only the first two are safe. A bounded queue chooses to block. The unbounded queue, the Python default, chooses to buffer without bound, which is the choice that runs the process out of memory.

The token bucket is not in the standard library, but it is a dozen lines. A bucket holds tokens up to a capacity and refills at a fixed rate. A request takes a token before it proceeds; if the bucket is empty, the request waits for a refill or is refused. Size the bucket by the budget: tokens-per-second set to the cost-per-token ceiling divided by the average tokens per request gives a rate that holds spend under the line. Python for DevOps (Ch. 15, pp. 635–636) shows the queue-fronted async ingestion shape this sits inside — a rate-limited producer feeding a bounded buffer feeding a pool of workers.

§ IIICode Worked Example

Start with the admission gate. The semaphore bounds concurrency; the async with form guarantees the release.

import asyncio

class InferenceReplica:
    def __init__(self, max_concurrent: int):
        self._gate = asyncio.Semaphore(max_concurrent)

    async def infer(self, request):
        async with self._gate:
            return await self._run_model(request)

The replica now never runs more than max_concurrent forward passes at once. Request max_concurrent + 1 suspends at the async with line until a predecessor finishes. No GPU-memory blowout, no thousand-way fan-out. The bound is a single number, sized to what one GPU can batch.

The semaphore alone has a gap. A request that suspends at the gate waits without limit, and a planner upstream waiting on it has no idea whether it is queued for one second or sixty. The bounded queue closes that gap by making the wait explicit and finite.

class AdmissionController:
    def __init__(self, max_concurrent: int, max_queued: int):
        self._gate = asyncio.Semaphore(max_concurrent)
        self._queue = asyncio.Queue(maxsize=max_queued)

    async def submit(self, request):
        try:
            self._queue.put_nowait(request)
        except asyncio.QueueFull:
            raise AtCapacity(request.endpoint)
        return await self._serve()

The put_nowait is the load-shedding decision in one call. When the queue is full, it raises QueueFull immediately rather than suspending, and the controller turns that into a typed AtCapacity the caller understands. This is the clean refusal the Ops lesson insisted on: the replica says no in milliseconds, the planner reads the typed signal, and it backs off instead of retrying into a wall. A request admitted to the queue is one the replica has committed to serving within a bounded wait.

The token bucket caps the budget. It refills on a schedule and hands out permits; a request with no permit waits for the next refill.

import time

class TokenBucket:
    def __init__(self, rate_per_sec: float, capacity: float):
        self._rate = rate_per_sec
        self._capacity = capacity
        self._tokens = capacity
        self._updated = time.monotonic()
        self._lock = asyncio.Lock()

    async def take(self, cost: float):
        async with self._lock:
            now = time.monotonic()
            self._tokens = min(self._capacity, self._tokens + (now - self._updated) * self._rate)
            self._updated = now
            if self._tokens < cost:
                deficit = cost - self._tokens
                await asyncio.sleep(deficit / self._rate)
                self._tokens = 0.0
            else:
                self._tokens -= cost

The bucket refills lazily. It computes elapsed time on each take and credits tokens for the gap, so no background timer is needed. The lock keeps two coroutines from double-spending the same tokens. Cost is passed per request, so a large-context call takes more tokens than a short one, which ties the rate to actual token spend rather than to request count. Set rate_per_sec from the endpoint's dollar budget and the model's price, and the bucket holds the cost-per-token line by making expensive bursts wait.

The three compose into one admission path. A request takes a budget permit, enters the bounded queue or is refused, then passes the concurrency gate to run.

class GovernedReplica:
    def __init__(self, max_concurrent, max_queued, rate_per_sec, capacity):
        self._gate = asyncio.Semaphore(max_concurrent)
        self._queue = asyncio.Queue(maxsize=max_queued)
        self._bucket = TokenBucket(rate_per_sec, capacity)

    async def infer(self, request):
        await self._bucket.take(request.estimated_tokens)
        try:
            self._queue.put_nowait(request)
        except asyncio.QueueFull:
            raise AtCapacity(request.endpoint)
        async with self._gate:
            self._queue.get_nowait()
            return await self._run_model(request)

§ IVConnection to Today's Ops Lesson

The Ops lesson built capacity governance at the fleet scale: scale on the line, keep a warm floor against the cold-start tax, shed by class, and let a budget bound the autoscaler's ceiling. Every one has a per-replica twin in this code. Scale-on-the-line is the fleet reading queue depth across replicas; the bounded queue is one replica reading its own depth. Shed-by-class is the fleet refusing low-priority work; the AtCapacity raise is one replica refusing when its queue is full. The budget that bounds the autoscaler's ceiling is the same dollar figure the token bucket converts into a refill rate.

The two scales need each other. The autoscaler alone cannot stop a single replica from over-admitting and dying on a memory blowout; the semaphore can, but it cannot summon a second replica when one is genuinely overloaded. Horizontal scaling adds lines; in-process admission control bounds the length of each line. Run both, and a burst meets a fleet that adds capacity and replicas that refuse cleanly while it arrives.

§ VPrior-Lesson Reach

The 06-05 lesson built asyncio TaskGroups and cancellation scopes for live risk monitors, with the watchdog and kill-switch disciplines. The kill switch there cancels in-flight work on a hard signal; the admission controller here refuses new work on a soft one. They are the two ends of the same lifecycle — refuse at the door under load, cancel inside under emergency. A replica wants both: semaphore and queue at admission, TaskGroup and cancellation for the work that got in.

The 06-08 lesson on Protocols and generics gives the controller its type discipline. The request flowing through this admission path is a Protocol with endpoint and estimated_tokens; making it a typed Protocol rather than a loose dict means the token-bucket cost and the typed AtCapacity are checked at the boundary, not discovered at runtime. The 05-21 contextvars lesson carries the request's identity through the suspension: when a request waits at the gate, its context — trace id, tenant, priority — travels with the coroutine, so the shed decision can read the priority the Ops lesson sheds by.

§ VIClosing

A replica is judged by how it behaves at its limit, and the limit always arrives. The semaphore decides how many requests run together, sized to what the GPU can batch. The bounded queue decides how many may wait, and turns the overflow into a clean typed refusal instead of an out-of-memory death. The token bucket decides how fast tokens may be spent, and turns the budget into a control that acts during the burst rather than a number read after it.

Build them in order, because each fails a different way. Then read this code beside the Ops lesson and see one principle at two scales. Bound what you take, queue a little, refuse the rest cleanly. The fleet does it in replicas; the replica does it in coroutines.