Request Queuing for LLM Backends
LLM request queuing is the layer that sits between your API and your model server, holding incoming prompts, batching them intelligently, and releasing them to the GPU in an order that maximizes throughput without starving any single caller. Without it, a burst of traffic either crashes your inference process with out-of-memory errors or leaves your GPU idle between one-off requests that never get batched together. This article walks through building a real LLM request queue in Python, from a naive in-memory version to a Redis-backed distributed queue with priority tiers, backpressure, and batching tuned for continuous-batching engines like vLLM.
Why LLM Request Queuing Matters
A single GPU serving an LLM can process many prompts in parallel far more efficiently than one at a time, because the bottleneck is memory bandwidth, not raw compute. Batch eight requests together and the marginal cost of the ninth token across all eight is nearly free compared to running them sequentially. But requests do not arrive in neat batches, they arrive as a Poisson-ish stream from users, agents, and background jobs, so someone has to hold them, group them, and dispatch them.
Without a queue you end up with one of two failure modes. Either you accept every request immediately and let the inference engine's internal scheduler deal with it, which works until concurrency spikes and you blow past your GPU's KV-cache memory, or you reject requests past a hard concurrency limit, which throws errors at real users during traffic spikes instead of just making them wait. A queue with defined semantics (max depth, timeout, priority, backpressure signal) turns an unpredictable failure into a predictable, observable wait time.
Request queuing also decouples your API's scaling from your model server's scaling. Your FastAPI or Express layer can accept thousands of connections, while the actual token-generation capacity stays fixed to what your GPUs can do. The queue is the shock absorber between those two different scaling curves.
How LLM Backends Handle Concurrent Requests
Modern inference servers like vLLM, TGI (Text Generation Inference), and TensorRT-LLM already do continuous batching internally: they interleave decode steps for many concurrent sequences on the same GPU, adding new requests to the batch as old ones finish. That is a form of queuing, but it operates at the token level and has no concept of your business logic (which customer gets priority, what happens after N seconds of waiting, how to shed load gracefully).
That is why most production stacks run two layers of queuing:
- Inference-level batching inside vLLM or TGI, which manages GPU memory and decode scheduling automatically.
- Application-level queuing in front of the inference server, which manages admission control, priority, retries, and backpressure before a request ever reaches the GPU.
This article focuses on the second layer, the one you actually own and can customize. The rest of this piece builds that layer step by step.
Building a Simple In-Memory Queue with Python asyncio
Start with the simplest version that solves a real problem: capping concurrency so you never send more requests to your model server than it can handle at once.
import asyncio
import time
from dataclasses import dataclass, field
from typing import Any
@dataclass
class LLMRequest:
prompt: str
future: asyncio.Future = field(default_factory=asyncio.Future)
enqueued_at: float = field(default_factory=time.monotonic)
class SimpleLLMQueue:
def __init__(self, max_concurrency: int, call_model):
self.queue: asyncio.Queue[LLMRequest] = asyncio.Queue()
self.semaphore = asyncio.Semaphore(max_concurrency)
self.call_model = call_model
self._workers_started = False
async def submit(self, prompt: str) -> str:
req = LLMRequest(prompt=prompt)
await self.queue.put(req)
return await req.future
async def _worker(self):
while True:
req = await self.queue.get()
async with self.semaphore:
try:
result = await self.call_model(req.prompt)
req.future.set_result(result)
except Exception as exc:
req.future.set_exception(exc)
finally:
self.queue.task_done()
async def start(self, num_workers: int = 4):
if self._workers_started:
return
self._workers_started = True
for _ in range(num_workers):
asyncio.create_task(self._worker())This gives you a bounded number of in-flight calls to your model server, and every caller gets a future they can await for the result. It is enough to stop a burst of requests from overwhelming a single-instance inference server, and it costs almost nothing to add. The gap: everything is FIFO, there is no priority, and state disappears if the process restarts.
Adding Priority and Rate Limiting
Not every request deserves the same treatment. An interactive chat request from a paying user should usually jump ahead of a bulk summarization job running in the background. Swap the plain asyncio.Queue for a PriorityQueue and attach a tier to each request.
import asyncio
import heapq
import itertools
import time
from dataclasses import dataclass, field
_counter = itertools.count()
@dataclass(order=True)
class PriorityLLMRequest:
priority: int
sequence: int = field(compare=True)
prompt: str = field(compare=False)
future: asyncio.Future = field(compare=False, default_factory=asyncio.Future)
enqueued_at: float = field(compare=False, default_factory=time.monotonic)
class PriorityLLMQueue:
# Lower number = higher priority. 0 = interactive, 5 = background.
def __init__(self, max_concurrency: int, call_model):
self.queue: asyncio.PriorityQueue = asyncio.PriorityQueue()
self.semaphore = asyncio.Semaphore(max_concurrency)
self.call_model = call_model
async def submit(self, prompt: str, priority: int = 3) -> str:
req = PriorityLLMRequest(
priority=priority, sequence=next(_counter), prompt=prompt
)
await self.queue.put(req)
return await req.future
async def _worker(self):
while True:
req: PriorityLLMRequest = await self.queue.get()
wait_time = time.monotonic() - req.enqueued_at
if wait_time > 30:
req.future.set_exception(TimeoutError("queued too long"))
self.queue.task_done()
continue
async with self.semaphore:
try:
result = await self.call_model(req.prompt)
req.future.set_result(result)
except Exception as exc:
req.future.set_exception(exc)
finally:
self.queue.task_done()
async def start(self, num_workers: int = 4):
for _ in range(num_workers):
asyncio.create_task(self._worker())The sequence counter breaks ties between equal-priority requests so it stays FIFO within a tier, and the timeout check drops requests that waited too long instead of running a stale generation nobody wants anymore. Set the priority based on your own signal, subscription tier, request type, or an explicit X-Priority header from an internal caller.
For rate limiting per tenant on top of this, add a token bucket keyed by API key before the request ever reaches submit:
class TokenBucket:
def __init__(self, rate_per_sec: float, burst: int):
self.rate = rate_per_sec
self.capacity = burst
self.tokens = burst
self.last_refill = time.monotonic()
def allow(self) -> bool:
now = time.monotonic()
elapsed = now - self.last_refill
self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
self.last_refill = now
if self.tokens >= 1:
self.tokens -= 1
return True
return FalseCheck allow() before calling queue.submit, and reject with a 429 immediately if it returns False. This keeps one noisy tenant from filling the whole queue and starving everyone else.
Using Redis for Distributed Request Queuing
An in-process asyncio.Queue disappears the moment your API process restarts, and it cannot be shared across multiple API replicas behind a load balancer. Once you run more than one API instance, move the queue itself into Redis so every replica pushes into and pulls from the same structure.
Redis Streams work well here because they give you consumer groups, at-least-once delivery, and a durable log you can replay if a worker crashes mid-request.
import redis.asyncio as redis
import json
import uuid
STREAM = "llm:requests"
GROUP = "llm-workers"
async def setup_stream(r: redis.Redis):
try:
await r.xgroup_create(STREAM, GROUP, id="0", mkstream=True)
except redis.ResponseError as exc:
if "BUSYGROUP" not in str(exc):
raise
async def submit_request(r: redis.Redis, prompt: str, priority: int = 3) -> str:
request_id = str(uuid.uuid4())
await r.xadd(
STREAM,
{"id": request_id, "prompt": prompt, "priority": priority},
)
return request_id
async def consume_loop(r: redis.Redis, consumer_name: str, call_model):
await setup_stream(r)
while True:
entries = await r.xreadgroup(
GROUP, consumer_name, {STREAM: ">"}, count=1, block=5000
)
if not entries:
continue
for _, messages in entries:
for message_id, fields in messages:
prompt = fields["prompt"]
try:
result = await call_model(prompt)
await r.set(f"llm:result:{fields['id']}", json.dumps(result), ex=600)
finally:
await r.xack(STREAM, GROUP, message_id)The API process writes to llm:requests and polls (or subscribes via pub/sub) for llm:result:<id> to know when a response is ready. Worker processes, which can run on the same boxes as your inference server or separately, pull entries off the stream with xreadgroup, call the model, and acknowledge. If a worker dies before acking, the message stays pending and another consumer in the group can claim it with xclaim after a timeout.
For a simpler setup that does not need Redis Streams' durability guarantees, a plain sorted set (ZADD with a priority score, ZPOPMIN to dequeue) is easier to reason about and works fine for most teams under a few thousand requests per minute.
Batching Requests to Maximize GPU Throughput
If you are running your own inference server rather than calling a hosted API, you can go further and batch prompts explicitly before sending them to the model, especially useful for embedding generation or any workload where the engine does not already do continuous batching for you.
import asyncio
import time
class BatchingQueue:
def __init__(self, call_model_batch, max_batch_size=16, max_wait_ms=20):
self.pending: list[LLMRequest] = []
self.lock = asyncio.Lock()
self.call_model_batch = call_model_batch
self.max_batch_size = max_batch_size
self.max_wait_ms = max_wait_ms
self._flush_scheduled = False
async def submit(self, prompt: str):
req = LLMRequest(prompt=prompt)
async with self.lock:
self.pending.append(req)
should_flush_now = len(self.pending) >= self.max_batch_size
if not self._flush_scheduled and not should_flush_now:
self._flush_scheduled = True
asyncio.create_task(self._delayed_flush())
if should_flush_now:
await self._flush()
return await req.future
async def _delayed_flush(self):
await asyncio.sleep(self.max_wait_ms / 1000)
await self._flush()
async def _flush(self):
async with self.lock:
if not self.pending:
self._flush_scheduled = False
return
batch = self.pending
self.pending = []
self._flush_scheduled = False
prompts = [r.prompt for r in batch]
try:
results = await self.call_model_batch(prompts)
for req, result in zip(batch, results):
req.future.set_result(result)
except Exception as exc:
for req in batch:
req.future.set_exception(exc)This is the classic "collect for N milliseconds or until the batch is full, whichever comes first" pattern. max_wait_ms is the knob that trades latency for throughput: 20ms adds a barely noticeable delay to any single request but lets the GPU process a full batch instead of one prompt at a time. Tune it down for latency-sensitive chat traffic and up for bulk jobs like re-ranking or embedding a document corpus.
If you are running vLLM, you often do not need this layer at all, since its continuous batching already groups concurrent requests at the token level. Explicit batching earns its keep for embedding models, classification heads, or any custom inference loop where the framework does not batch for you.
Handling Backpressure and Timeouts
A queue without a depth limit is not protection, it is just a slower way to run out of memory. Cap the queue and reject new work once it is full, returning a clear signal the caller can act on.
class BoundedLLMQueue(PriorityLLMQueue):
def __init__(self, max_concurrency: int, max_queue_depth: int, call_model):
super().__init__(max_concurrency, call_model)
self.max_queue_depth = max_queue_depth
async def submit(self, prompt: str, priority: int = 3) -> str:
if self.queue.qsize() >= self.max_queue_depth:
raise QueueFullError(
f"queue depth {self.queue.qsize()} exceeds limit {self.max_queue_depth}"
)
return await super().submit(prompt, priority)
class QueueFullError(Exception):
passMap QueueFullError to an HTTP 503 with a Retry-After header at your API layer, not a generic 500. Callers, whether that is a frontend, a retry-aware SDK, or another service, need to distinguish "the server is overloaded, back off" from "something is broken." Pick max_queue_depth from your acceptable worst-case wait time: if a request in position N takes roughly N / throughput seconds to get served, set the cap so the tail wait stays under whatever SLA you promised.
Combine this with a per-request timeout on the caller's side too. asyncio.wait_for(queue.submit(prompt), timeout=25) ensures a request that has been queued too long fails fast on the client rather than hanging a connection indefinitely.
Monitoring Queue Depth and Latency
You cannot tune backpressure limits, batch windows, or worker counts without visibility into what the queue is actually doing. Track at minimum:
- Queue depth over time, sampled every few seconds, to see how close you are running to your cap.
- Time-in-queue per request (the
enqueued_atdelta from the examples above), broken out by priority tier. - Batch fill rate, meaning the average number of prompts per dispatched batch versus your
max_batch_size, which tells you whether your wait window is too short or too long. - Rejection rate from
QueueFullErroror 429s, which tells you when you need more GPU capacity rather than a tuning tweak.
Export these as Prometheus counters and histograms and put them on the same dashboard as GPU utilization. A queue that is always near-empty next to a GPU running at 40% utilization usually means your batch window is too short. A queue that is consistently deep next to a GPU pinned at 100% means you need more capacity, not more queue tuning.
from prometheus_client import Histogram, Gauge
QUEUE_DEPTH = Gauge("llm_queue_depth", "Current number of requests waiting")
QUEUE_WAIT_SECONDS = Histogram(
"llm_queue_wait_seconds", "Time spent waiting in queue before dispatch"
)
# inside the worker, right before calling the model:
QUEUE_WAIT_SECONDS.observe(time.monotonic() - req.enqueued_at)
QUEUE_DEPTH.set(self.queue.qsize())Common Pitfalls in LLM Request Queuing
- No timeout at all. A request that queues forever because a downstream worker died silently is worse than an explicit error. Always set a max wait and fail loudly past it.
- One priority tier for everyone. Treating an internal batch job the same as an interactive user request means your best customers wait behind your cron jobs during peak load.
- Ignoring token count when sizing batches. A batch of 16 short prompts and a batch of 16 long ones consume very different amounts of KV-cache memory. If you batch explicitly, cap by estimated token budget, not just request count.
- Retrying inside the queue instead of at the edge. Automatic retries on a request that already timed out in the queue just adds more load to an already-overloaded system. Retry at the client with backoff, not inside the worker loop.
- Losing requests on deploys. An in-memory queue drops everything on restart. If you cannot tolerate that, move to Redis Streams or another durable queue before you scale past a single instance.
- Tuning batch windows once and forgetting them. Traffic patterns change. Revisit
max_wait_msandmax_batch_sizewhenever you change models, GPUs, or see a shift in your batch-fill-rate metric.
FAQ
What is LLM request queuing? It is the practice of holding incoming prompt requests in a managed queue in front of your model server, controlling how many run concurrently, in what order, and how they get batched, instead of sending every request straight to the GPU as it arrives.
Do I need a queue if I use vLLM or another engine with continuous batching? Usually yes, at the application level. The engine batches at the token level once requests reach it, but it has no concept of per-tenant priority, admission control, or backpressure. Put a lightweight queue in front of it for those concerns, and let the engine handle GPU-level scheduling.
Should I use Redis or an in-memory queue? Start in-memory if you run a single API process. Move to Redis Streams (or a similar durable, shared queue) as soon as you run more than one API replica, or the moment you need requests to survive a process restart.
How do I pick the batch wait window? Start around 15 to 30 milliseconds for interactive chat traffic and measure your batch-fill-rate metric. If batches are consistently full before the window expires, shrink the window. If they are consistently underfilled, extend it, but watch added latency on the p99.
What should happen when the queue is full? Reject immediately with a 503 and a Retry-After header rather than accepting the request and letting it wait indefinitely. Callers should treat this as a signal to back off and retry, not as a bug to route around with an aggressive retry loop.
How do I prioritize requests fairly without starving low-priority traffic? Use weighted priority instead of strict priority: dispatch from the high-priority tier most of the time but reserve a small percentage of worker capacity for lower tiers so background jobs still make progress instead of waiting forever behind a constant stream of interactive traffic.
BootcampA 30-day guided bootcamp: build, harden and ship a production autonomous agent from scratch.
AI AgentsUnderstand how AI agents really work: the loop, the tools, the memory, and why most agent projects fail.