Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 4 additions & 2 deletions benchmarks/router/common.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,8 +10,6 @@
import logging
import os

from prefix_data_generator.synthesizer import Synthesizer

# Default values
DEFAULT_MODEL = "deepseek-ai/DeepSeek-R1-Distill-Llama-8B"
DEFAULT_URL = "http://localhost:8000"
Expand Down Expand Up @@ -326,6 +324,10 @@ def prepare_trace_dataset(args, output_dir, logger):
)
logger.info(f" Random seed: {args.seed}")

# Synthetic generation has optional dependencies absent from slim CPU test images.
# Keep the request helpers and trace-file path usable without those dependencies.
from prefix_data_generator.synthesizer import Synthesizer

synthesizer = Synthesizer(
args.input_dataset,
block_size=args.block_size,
Expand Down
29 changes: 20 additions & 9 deletions lib/kv-router/src/scheduling/AGENTS.md
Original file line number Diff line number Diff line change
Expand Up @@ -16,30 +16,41 @@ sequenceDiagram
end
participant S as WorkerSelector

H->>A: Enqueue(request)
A->>A: Resolve class_index
alt Immediate path
H->>H: Resolve class_index, sample snapshot, stamp arrival
H->>A: Enqueue(request, metadata)
alt Direct path (never-queues class or zero workers)
A->>S: select_worker(request)
S-->>A: Selected worker
A->>A: Reserve worker capacity and respond
else Queued path
A->>P: enqueue(class_index, request)
else Policy-queue path (every other request)
A->>P: enqueue(metadata, request)
P->>C: Push into the selected class queue
H->>A: Update after capacity changes
A->>P: pop_next()
A->>P: pop_next() in the same actor turn
P->>C: Get one candidate per dispatchable class
C-->>P: Dispatch candidate
P-->>A: DRR winner
A->>S: select_worker(request)
S-->>A: Selected worker
A->>A: Reserve worker capacity and respond
end
H->>A: Update after capacity changes drains the remaining backlog
```

- `SchedulerQueue` is the public handle that sends commands to the actor.
- `SchedulerQueueActor` classifies each request and chooses the immediate or queued path.
- `SchedulerQueue` is the public handle. It resolves the class index, samples the
cache-overlap snapshot, and stamps the arrival offset on the caller task before
sending the Enqueue command.
- `SchedulerQueueActor` routes every request through the policy queue and drains the
ready backlog in the same actor turn, so freed capacity can go to another class's
queued head under DRR rather than to the arriving request. The direct path remains
only for classes that can never hold a queued entry (any zero per-worker limit) and
for zero-discovered-worker windows, which surface `NoEndpoints` from selection.
- `PolicyQueue` does not classify requests. It owns all class queues and uses deficit round robin (DRR) to give each class weighted turns.
- Each `PolicyClassQueue` owns ordering and accounting for one class such as `latency`, `agents`, or `batch`.
- Within one strict-priority tier, deadline-bearing entries order earliest-due-first
ahead of all non-deadline entries before the FCFS/LCFS/WSPT policy score applies.
Deadlines are indexed in an ordered due set and a conditional actor timer rejects
expired entries; all deadline math uses `tokio::time::Instant` so paused-clock tests
stay coherent.
- `SchedulerQueueActor::admit_one` performs final worker selection and reserves worker capacity after either path.
- A single-class profile still uses `PolicyQueue`, but DRR has no cross-class effect.

Expand Down
9 changes: 9 additions & 0 deletions lib/kv-router/src/scheduling/policy_config.rs
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,15 @@ impl PolicyClassConfig {
self.prefill_busy_threshold.is_some() || self.prefill_busy_threshold_frac.is_some()
}

/// Any zero per-worker limit means `queue_rejection` rejects every queued
/// entry (`current >= limit` holds at zero usage), so the class can never
/// hold a queued request and must keep the direct admission path.
pub fn never_queues(&self) -> bool {
Comment thread
ishandhanani marked this conversation as resolved.
self.request_queue_limit_per_worker == Some(0)
|| self.raw_isl_token_queue_limit_per_worker == Some(0)
|| self.cached_token_queue_limit_per_worker == Some(0)
}

pub fn worker_is_busy(&self, active_tokens: usize, max_batched_tokens: u64) -> bool {
let absolute_busy = self
.prefill_busy_threshold
Expand Down
Loading
Loading