Priority request queue with batch formation

In production LLM serving, you almost always have multiple tiers of traffic. Interactive chat users need low latency. Background summarization jobs need throughput. Evaluation pipelines need to run but should never block user-facing traffic. A simple FIFO queue treats them all the same, which means a burst of batch processing requests can spike latency for your paying users.

Today I will build a priority request queue that respects these tiers and also forms batches intelligently, grouping requests with similar prompt lengths to reduce padding waste and improve GPU utilization.

The priority queue

The core data structure is a priority queue where each request carries a priority level. Higher-priority requests are dequeued first, regardless of arrival order. Within the same priority level, requests are served in FIFO order.

import heapq
import time
from dataclasses import dataclass, field
from enum import IntEnum

class Priority(IntEnum):
    CRITICAL = 0    # System health checks, canary requests
    HIGH = 1        # Interactive users, paid tier
    NORMAL = 2      # Standard API requests
    LOW = 3         # Batch processing, eval pipelines
    BACKGROUND = 4  # Backfill, non-urgent tasks

@dataclass(order=True)
class PrioritizedRequest:
    priority: int
    arrival_time: float = field(compare=True)
    request_id: str = field(compare=False)
    prompt_len: int = field(compare=False)
    max_tokens: int = field(compare=False)
    payload: dict = field(compare=False, default_factory=dict)

class PriorityQueue:
    def __init__(self, max_queue_size: int = 1000):
        self.heap = []
        self.max_queue_size = max_queue_size
        self.dropped = 0

    def enqueue(self, request: PrioritizedRequest) -> bool:
        if len(self.heap) >= self.max_queue_size:
            # Shed load: drop the lowest-priority request
            if self.heap and self.heap[-1].priority > request.priority:
                # New request is higher priority; drop something lower
                self._drop_lowest()
            else:
                self.dropped += 1
                return False
        heapq.heappush(self.heap, request)
        return True

    def dequeue(self) -> PrioritizedRequest:
        return heapq.heappop(self.heap)

    def _drop_lowest(self):
        """Remove the lowest-priority (highest number) request."""
        # Find and remove the max-priority item
        if not self.heap:
            return
        worst_idx = max(range(len(self.heap)),
                        key=lambda i: (self.heap[i].priority, self.heap[i].arrival_time))
        self.heap[worst_idx] = self.heap[-1]
        self.heap.pop()
        heapq.heapify(self.heap)
        self.dropped += 1

    def __len__(self):
        return len(self.heap)

Why batch formation matters

Continuous batching (as I covered on day 50) lets requests enter and leave the batch at every decode step. But the prefill phase still benefits from grouping. When the scheduler runs prefill for multiple new requests simultaneously, having similar prompt lengths reduces padding waste in the attention computation.

Consider two scenarios for prefilling 4 requests:

• Ungrouped: Prompt lengths [128, 2048, 256, 1024]. Padded to 2048 for all. Wasted compute: (2048-128) + (2048-256) + (2048-1024) = 4,736 tokens of padding.
• Grouped: Batch 1 with [128, 256], padded to 256. Batch 2 with [1024, 2048], padded to 2048. Wasted compute: (256-128) + (2048-1024) = 1,152 tokens. That is 4x less waste.

In practice, modern systems like vLLM use ragged (unpadded) prefill, so the padding waste is less severe. But grouping still helps because similar-length sequences have more uniform memory requirements, making KV cache block allocation more predictable.

The batch former

class BatchFormer:
    def __init__(self, max_batch_size: int = 8,
                 max_wait_ms: float = 50.0,
                 length_buckets: list[int] = None):
        self.max_batch_size = max_batch_size
        self.max_wait_ms = max_wait_ms
        # Buckets for grouping similar prompt lengths
        self.length_buckets = length_buckets or [256, 512, 1024, 2048, 4096, 8192]

    def _bucket_for_length(self, prompt_len: int) -> int:
        """Find the smallest bucket that fits this prompt length."""
        for bucket in self.length_buckets:
            if prompt_len <= bucket:
                return bucket
        return self.length_buckets[-1]

    def form_batch(self, queue: PriorityQueue,
                   current_batch_size: int = 0) -> list[PrioritizedRequest]:
        """Form a batch from the priority queue.

        Returns a list of requests to prefill together.
        Prioritizes: (1) priority level, (2) length similarity, (3) wait time.
        """
        available_slots = self.max_batch_size - current_batch_size
        if available_slots <= 0 or len(queue) == 0:
            return []

        batch = []
        skipped = []
        target_bucket = None
        now = time.time()

        while len(batch) < available_slots and len(queue) > 0:
            request = queue.dequeue()

            # Always admit high-priority requests regardless of bucketing
            if request.priority <= Priority.HIGH:
                batch.append(request)
                continue

            bucket = self._bucket_for_length(request.prompt_len)

            # First non-critical request sets the target bucket
            if target_bucket is None:
                target_bucket = bucket
                batch.append(request)
                continue

            # Same bucket or waited too long: add to batch
            wait_ms = (now - request.arrival_time) * 1000
            if bucket == target_bucket or wait_ms > self.max_wait_ms:
                batch.append(request)
            else:
                skipped.append(request)

        # Put skipped requests back
        for req in skipped:
            queue.enqueue(req)

        return batch

Putting it together: the request handler

class InferenceRouter:
    def __init__(self):
        self.queue = PriorityQueue(max_queue_size=500)
        self.batch_former = BatchFormer(max_batch_size=16)
        self.active_batch_size = 0

    def handle_request(self, request_id: str, prompt: str,
                       max_tokens: int, priority: str = "normal"):
        """Accept an incoming request and add it to the priority queue."""
        pri = getattr(Priority, priority.upper(), Priority.NORMAL)
        prompt_len = len(prompt.split())  # approximate token count

        req = PrioritizedRequest(
            priority=pri,
            arrival_time=time.time(),
            request_id=request_id,
            prompt_len=prompt_len,
            max_tokens=max_tokens,
            payload={"prompt": prompt},
        )

        accepted = self.queue.enqueue(req)
        if not accepted:
            return {"error": "Queue full", "status": 429}

        return {"status": "queued", "position": len(self.queue)}

    def next_batch(self) -> list[PrioritizedRequest]:
        """Called by the inference engine when it has capacity."""
        batch = self.batch_former.form_batch(
            self.queue,
            current_batch_size=self.active_batch_size,
        )
        self.active_batch_size += len(batch)
        return batch

    def on_request_complete(self, request_id: str):
        """Called when a request finishes generation."""
        self.active_batch_size -= 1

Load shedding and fairness

The priority queue naturally handles load shedding: when the queue is full, the lowest-priority requests are dropped first. But we also need fairness guarantees to prevent starvation. A constant stream of HIGH priority requests should not indefinitely block NORMAL requests.

The max_wait_ms parameter in the batch former provides soft fairness: if a normal-priority request has been waiting more than 50 milliseconds, it gets admitted to the next batch regardless of bucketing. For hard fairness, you can implement priority aging:

def age_priorities(self, aging_rate_ms: float = 100.0):
    """Promote requests that have waited too long.

    Every aging_rate_ms of wait time, promote by one priority level.
    This prevents starvation of low-priority requests.
    """
    now = time.time()
    for req in self.queue.heap:
        wait_ms = (now - req.arrival_time) * 1000
        promotions = int(wait_ms / aging_rate_ms)
        if promotions > 0:
            req.priority = max(0, req.priority - promotions)
    heapq.heapify(self.queue.heap)

Integration with vLLM and SGLang

Both vLLM and SGLang have their own internal schedulers, so this priority queue sits in front of the engine, not inside it. The integration point is an API gateway (like FastAPI) that accepts requests, enqueues them, and forwards them to the engine when capacity is available.

vLLM's AsyncLLMEngine exposes an add_request method that accepts individual requests. Your gateway calls next_batch() on the priority queue and feeds each request to the engine via add_request. The engine's internal continuous batching scheduler handles the rest.

# Simplified integration with vLLM's AsyncLLMEngine
from vllm.engine.async_llm_engine import AsyncLLMEngine

async def scheduler_loop(router: InferenceRouter, engine: AsyncLLMEngine):
    while True:
        batch = router.next_batch()
        for req in batch:
            await engine.add_request(
                request_id=req.request_id,
                prompt=req.payload["prompt"],
                sampling_params=SamplingParams(max_tokens=req.max_tokens),
            )
        await asyncio.sleep(0.01)  # 10ms scheduling interval

A priority queue is the simplest form of QoS (quality of service) for LLM serving. It does not require changes to the inference engine, just a thin layer between the load balancer and the engine that decides which requests go first.

This wraps up the initial production systems block. We have gone from building a container to balancing load to managing request priorities. Next, I will look at multi-GPU tensor parallel benchmarks to understand how scaling across GPUs affects the throughput and latency characteristics we have been measuring.