@tduncan / Nvidia-Optimization-Stack-VLLM / commits / dafbef15a1

[Core] Add `max_num_queued_reqs` and `max_num_queued_tokens` for queue size management (#49445)

Signed-off-by: NickLucche <nicolo.lucchesi@mistral.ai>

Nicolò Lucchesi committed Aug 31, 2026 at 16:26 UTC dafbef15a1c879c64ebb99427917e4ca8d5bca1e
18 files changed +644 -35
tests/entrypoints/openai/chat_completion/test_serving_chat.py
+33 -1
@@ -42,7 +42,7 @@ from vllm.entrypoints.openai.models.serving import (
42 )
43 from vllm.entrypoints.openai.parser.harmony_utils import get_encoding
44 from vllm.entrypoints.serve.engine.protocol import ErrorResponse
45 -from vllm.exceptions import VLLMValidationError
45 +from vllm.exceptions import QueueOverflowError, VLLMValidationError
46 from vllm.inputs import TokensPrompt
47 from vllm.multimodal.inputs import PlaceholderRange
48 from vllm.outputs import CompletionOutput, RequestOutput
@@ -848,6 +848,9 @@ class MockEngine:
848 renderer: MagicMock = field(default_factory=MagicMock)
849 errored: bool = False
850
851 + def check_admission(self, n: int = 1, request_id: str | None = None) -> None:
852 + pass
853 +
854
855 async def _async_serving_chat_init():
856 engine = MockEngine()
@@ -944,6 +947,35 @@ async def test_serving_chat_returns_correct_model_name():
947 assert await serving_chat.create_chat_completion(req) == MODEL_NAME
948
949
950 +@pytest.mark.asyncio
951 +@pytest.mark.parametrize("stream", [True, False])
952 +async def test_admission_rejection_escapes_before_response_starts(stream):
953 + """Overload rejections must propagate out of create_chat_completion.
954 +
955 + For streaming this is the only chance to return a real HTTP status: once
956 + StreamingResponse is constructed the 200 has already been sent and the
957 + rejection would degrade into an in-band SSE error chunk.
958 + """
959 + mock_engine = MagicMock(spec=AsyncLLM)
960 + mock_engine.errored = False
961 + mock_engine.model_config = MockModelConfig()
962 + mock_engine.input_processor = MagicMock()
963 + mock_engine.renderer = _build_renderer(mock_engine.model_config)
964 + mock_engine.check_admission.side_effect = QueueOverflowError()
965 +
966 + serving_chat = _build_serving_chat(mock_engine)
967 + req = ChatCompletionRequest(
968 + model=MODEL_NAME,
969 + messages=[{"role": "user", "content": "what is 1+1?"}],
970 + stream=stream,
971 + )
972 +
973 + with pytest.raises(QueueOverflowError):
974 + await serving_chat.create_chat_completion(req)
975 +
976 + mock_engine.generate.assert_not_called()
977 +
978 +
979 @pytest.mark.asyncio
980 async def test_serving_chat_should_set_correct_max_tokens():
981 mock_engine = MagicMock(spec=AsyncLLM)
tests/entrypoints/speech_to_text/test_speech_to_text_cancellation.py
+2
@@ -38,6 +38,7 @@ async def test_non_streaming_cancel_aborts_engine_requests(
38 ):
39 engine_client = SimpleNamespace(
40 errored=False,
41 + check_admission=Mock(),
42 generate=Mock(side_effect=lambda *_args, **_kwargs: _never_finishes()),
43 abort=AsyncMock(),
44 is_tracing_enabled=AsyncMock(return_value=False),
@@ -104,6 +105,7 @@ async def test_non_streaming_cancel_advances_all_chunk_generators():
105 started_request_ids: list[str] = []
106 engine_client = SimpleNamespace(
107 errored=False,
108 + check_admission=Mock(),
109 generate=Mock(
110 side_effect=lambda *_args, **_kwargs: _records_start_then_never_finishes(
111 started_request_ids, _args[2]
tests/v1/engine/test_admission_control.py new
+384
@@ -0,0 +1,384 @@
1 +# SPDX-License-Identifier: Apache-2.0
2 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3 +
4 +"""Unit tests for admission control (max_num_queued_reqs / max_num_queued_tokens).
5 +
6 +These tests cover:
7 +- OutputProcessor.get_num_queued_tokens() token counting
8 +- AsyncLLM.check_admission() admission control logic
9 +- Exception classes (GracefulHTTPError, QueueOverflowError, MaxQueuedTokensError)
10 +- create_error_response() mapping GracefulHTTPError to HTTP 503
11 +- SchedulerConfig field defaults and validation
12 +- human_readable_int CLI notation for max_num_queued_tokens
13 +"""
14 +
15 +import argparse
16 +import asyncio
17 +from http import HTTPStatus
18 +from types import SimpleNamespace
19 +from unittest.mock import AsyncMock, MagicMock
20 +
21 +import pytest
22 +from pydantic import ValidationError
23 +
24 +from vllm.config.scheduler import SchedulerConfig
25 +from vllm.entrypoints.serve.exception_handling.error_response import (
26 + create_error_response,
27 +)
28 +from vllm.exceptions import (
29 + GracefulHTTPError,
30 + MaxQueuedTokensError,
31 + QueueOverflowError,
32 + VLLMError,
33 +)
34 +from vllm.pooling_params import PoolingParams
35 +from vllm.sampling_params import SamplingParams
36 +from vllm.utils.argparse_utils import human_readable_int
37 +from vllm.v1.engine.async_llm import AsyncLLM
38 +from vllm.v1.engine.output_processor import OutputProcessor
39 +
40 +pytestmark = pytest.mark.cpu_test
41 +
42 +# ---------------------------------------------------------------------------
43 +# Helpers
44 +# ---------------------------------------------------------------------------
45 +
46 +
47 +def _make_req_state(prompt_len: int, is_prefilling: bool = True):
48 + return SimpleNamespace(
49 + prompt_len=prompt_len,
50 + is_prefilling=is_prefilling,
51 + )
52 +
53 +
54 +def _make_async_llm(
55 + max_num_queued_reqs: int | None = None,
56 + max_num_queued_tokens: int | None = None,
57 + num_unfinished: int = 0,
58 + num_queued_tokens: int = 0,
59 +) -> AsyncLLM:
60 + """Create a bare AsyncLLM with just the attributes needed for scheduling."""
61 + llm = AsyncLLM.__new__(AsyncLLM)
62 + llm.scheduler_config = SimpleNamespace(
63 + max_num_queued_reqs=max_num_queued_reqs,
64 + max_num_queued_tokens=max_num_queued_tokens,
65 + )
66 + llm.output_processor = MagicMock()
67 + llm.output_processor.get_num_unfinished_requests.return_value = num_unfinished
68 + llm.output_processor.get_num_queued_tokens.return_value = num_queued_tokens
69 + return llm
70 +
71 +
72 +def _make_output_processor(**request_states) -> OutputProcessor:
73 + op = OutputProcessor.__new__(OutputProcessor)
74 + op.request_states = request_states
75 + return op
76 +
77 +
78 +def _make_scheduler_config(**kwargs) -> SchedulerConfig:
79 + return SchedulerConfig(
80 + runner_type="generate",
81 + max_model_len=4096,
82 + is_encoder_decoder=False,
83 + **kwargs,
84 + )
85 +
86 +
87 +# ---------------------------------------------------------------------------
88 +# Exception classes
89 +# ---------------------------------------------------------------------------
90 +
91 +
92 +def test_graceful_http_error_carries_status_and_message():
93 + err = GracefulHTTPError("custom message", HTTPStatus.SERVICE_UNAVAILABLE)
94 + assert err.message == "custom message"
95 + assert err.http_status == HTTPStatus.SERVICE_UNAVAILABLE
96 + assert str(err) == "custom message"
97 +
98 +
99 +def test_graceful_http_error_is_vllm_error():
100 + err = GracefulHTTPError("msg", HTTPStatus.TOO_MANY_REQUESTS)
101 + assert isinstance(err, VLLMError)
102 +
103 +
104 +def test_queue_overflow_error():
105 + err = QueueOverflowError()
106 + assert err.http_status == HTTPStatus.SERVICE_UNAVAILABLE
107 + assert isinstance(err, GracefulHTTPError)
108 + assert "busy" in err.message.lower() or "try again" in err.message.lower()
109 +
110 +
111 +def test_max_queued_tokens_error():
112 + err = MaxQueuedTokensError()
113 + assert err.http_status == HTTPStatus.SERVICE_UNAVAILABLE
114 + assert isinstance(err, GracefulHTTPError)
115 + assert "backlog" in err.message.lower() or "try again" in err.message.lower()
116 +
117 +
118 +@pytest.mark.parametrize("exc_cls", [QueueOverflowError, MaxQueuedTokensError])
119 +def test_admission_exceptions_are_vllm_errors(exc_cls):
120 + """Admission rejections must reach the VLLMError HTTP handler."""
121 + assert issubclass(exc_cls, VLLMError)
122 +
123 +
124 +# ---------------------------------------------------------------------------
125 +# OutputProcessor.get_num_queued_tokens
126 +# ---------------------------------------------------------------------------
127 +
128 +
129 +def test_queued_tokens_empty():
130 + assert _make_output_processor().get_num_queued_tokens() == 0
131 +
132 +
133 +def test_queued_tokens_sums_prefilling_requests():
134 + op = _make_output_processor(r1=_make_req_state(100), r2=_make_req_state(200))
135 + assert op.get_num_queued_tokens() == 300
136 +
137 +
138 +def test_queued_tokens_excludes_non_prefilling():
139 + op = _make_output_processor(
140 + r1=_make_req_state(100, is_prefilling=True),
141 + r2=_make_req_state(200, is_prefilling=False),
142 + r3=_make_req_state(50, is_prefilling=True),
143 + )
144 + assert op.get_num_queued_tokens() == 150
145 +
146 +
147 +def test_queued_tokens_all_non_prefilling():
148 + op = _make_output_processor(
149 + r1=_make_req_state(100, is_prefilling=False),
150 + r2=_make_req_state(200, is_prefilling=False),
151 + )
152 + assert op.get_num_queued_tokens() == 0
153 +
154 +
155 +# ---------------------------------------------------------------------------
156 +# AsyncLLM.check_admission
157 +# ---------------------------------------------------------------------------
158 +
159 +
160 +def test_admission_no_limits_allows_everything():
161 + llm = _make_async_llm(num_unfinished=999, num_queued_tokens=999)
162 + llm.check_admission()
163 +
164 +
165 +# -- max_num_queued_reqs ----------------------------------------------------
166 +
167 +
168 +def test_admission_reqs_allows_when_under_limit():
169 + llm = _make_async_llm(max_num_queued_reqs=10, num_unfinished=5)
170 + llm.check_admission()
171 +
172 +
173 +def test_admission_reqs_rejects_at_limit():
174 + llm = _make_async_llm(max_num_queued_reqs=10, num_unfinished=10)
175 + with pytest.raises(QueueOverflowError):
176 + llm.check_admission()
177 +
178 +
179 +def test_admission_reqs_rejects_with_n():
180 + llm = _make_async_llm(max_num_queued_reqs=10, num_unfinished=8)
181 + with pytest.raises(QueueOverflowError):
182 + llm.check_admission(3)
183 +
184 +
185 +def test_admission_reqs_allows_n_at_boundary():
186 + llm = _make_async_llm(max_num_queued_reqs=10, num_unfinished=7)
187 + llm.check_admission(3)
188 +
189 +
190 +def test_admission_reqs_rejects_when_zero_limit():
191 + llm = _make_async_llm(max_num_queued_reqs=0, num_unfinished=0)
192 + with pytest.raises(QueueOverflowError):
193 + llm.check_admission()
194 +
195 +
196 +# -- max_num_queued_tokens --------------------------------------------------
197 +
198 +
199 +def test_admission_tokens_allows_when_under_limit():
200 + llm = _make_async_llm(max_num_queued_tokens=1000, num_queued_tokens=500)
201 + llm.check_admission()
202 +
203 +
204 +def test_admission_tokens_rejects_at_limit():
205 + llm = _make_async_llm(max_num_queued_tokens=1000, num_queued_tokens=1000)
206 + with pytest.raises(MaxQueuedTokensError):
207 + llm.check_admission()
208 +
209 +
210 +def test_admission_tokens_rejects_over_limit():
211 + llm = _make_async_llm(max_num_queued_tokens=1000, num_queued_tokens=1500)
212 + with pytest.raises(MaxQueuedTokensError):
213 + llm.check_admission()
214 +
215 +
216 +def test_admission_tokens_rejects_when_zero_limit():
217 + llm = _make_async_llm(max_num_queued_tokens=0, num_queued_tokens=0)
218 + with pytest.raises(MaxQueuedTokensError):
219 + llm.check_admission()
220 +
221 +
222 +# -- interaction between both limits ----------------------------------------
223 +
224 +
225 +def test_admission_both_limits_checked_independently():
226 + llm = _make_async_llm(
227 + max_num_queued_reqs=100,
228 + max_num_queued_tokens=1000,
229 + num_unfinished=5,
230 + num_queued_tokens=1000,
231 + )
232 + with pytest.raises(MaxQueuedTokensError):
233 + llm.check_admission()
234 +
235 +
236 +def test_admission_req_limit_checked_before_token_limit():
237 + llm = _make_async_llm(
238 + max_num_queued_reqs=10,
239 + max_num_queued_tokens=1000,
240 + num_unfinished=10,
241 + num_queued_tokens=1000,
242 + )
243 + with pytest.raises(QueueOverflowError):
244 + llm.check_admission()
245 +
246 +
247 +# -- n derived from params at the add_request call site ---------------------
248 +
249 +
250 +@pytest.mark.parametrize("params", [SamplingParams(), PoolingParams()])
251 +def test_admission_params_without_explicit_n_count_as_one_slot(params):
252 + """PoolingParams has no ``.n``; add_request must fall back to 1."""
253 + llm = _make_async_llm(max_num_queued_reqs=10, num_unfinished=9)
254 + llm.check_admission(getattr(params, "n", 1) or 1)
255 +
256 + llm = _make_async_llm(max_num_queued_reqs=10, num_unfinished=10)
257 + with pytest.raises(QueueOverflowError):
258 + llm.check_admission(getattr(params, "n", 1) or 1)
259 +
260 +
261 +@pytest.mark.asyncio
262 +async def test_concurrent_single_request_admission_respects_limit():
263 + """Concurrent n=1 admissions cannot consume the same slot."""
264 + llm = _make_async_llm(max_num_queued_reqs=1)
265 + request_states: dict[str, SimpleNamespace] = {}
266 + llm.output_processor.get_num_unfinished_requests.side_effect = lambda: len(
267 + request_states
268 + )
269 + llm.output_processor.has_request.side_effect = request_states.__contains__
270 +
271 + def add_request(request, *_args):
272 + request_states[request.request_id] = _make_req_state(
273 + len(request.prompt_token_ids)
274 + )
275 +
276 + llm.output_processor.add_request.side_effect = add_request
277 +
278 + async def add_request_async(_request):
279 + await asyncio.sleep(0)
280 +
281 + llm.engine_core = SimpleNamespace(
282 + add_request_async=AsyncMock(side_effect=add_request_async),
283 + shutdown=MagicMock(),
284 + )
285 + llm.log_requests = False
286 +
287 + requests = [
288 + SimpleNamespace(request_id=f"request-{idx}", prompt_token_ids=[idx])
289 + for idx in range(2)
290 + ]
291 + results = await asyncio.gather(
292 + *(
293 + llm._add_request(request, None, None, 0, MagicMock())
294 + for request in requests
295 + ),
296 + return_exceptions=True,
297 + )
298 +
299 + assert sum(result is None for result in results) == 1
300 + assert sum(isinstance(result, QueueOverflowError) for result in results) == 1
301 + assert len(request_states) == 1
302 +
303 +
304 +# ---------------------------------------------------------------------------
305 +# create_error_response integration
306 +# ---------------------------------------------------------------------------
307 +
308 +
309 +def test_queue_overflow_maps_to_503():
310 + resp = create_error_response(QueueOverflowError())
311 + assert resp.error.code == HTTPStatus.SERVICE_UNAVAILABLE.value
312 + assert resp.error.type == HTTPStatus.SERVICE_UNAVAILABLE.phrase
313 + assert resp.error.param is None
314 + msg = resp.error.message.lower()
315 + assert "busy" in msg or "try again" in msg
316 +
317 +
318 +def test_max_queued_tokens_maps_to_503():
319 + resp = create_error_response(MaxQueuedTokensError())
320 + assert resp.error.code == HTTPStatus.SERVICE_UNAVAILABLE.value
321 + assert resp.error.type == HTTPStatus.SERVICE_UNAVAILABLE.phrase
322 + msg = resp.error.message.lower()
323 + assert "backlog" in msg or "try again" in msg
324 +
325 +
326 +def test_custom_graceful_error_maps_to_its_status():
327 + err = GracefulHTTPError("custom", HTTPStatus.SERVICE_UNAVAILABLE)
328 + resp = create_error_response(err)
329 + assert resp.error.code == HTTPStatus.SERVICE_UNAVAILABLE.value
330 + assert resp.error.type == HTTPStatus.SERVICE_UNAVAILABLE.phrase
331 +
332 +
333 +# ---------------------------------------------------------------------------
334 +# SchedulerConfig field defaults
335 +# ---------------------------------------------------------------------------
336 +
337 +
338 +def test_scheduler_config_defaults_are_none():
339 + config = _make_scheduler_config()
340 + assert config.max_num_queued_reqs is None
341 + assert config.max_num_queued_tokens is None
342 +
343 +
344 +def test_scheduler_config_accepts_explicit_values():
345 + config = _make_scheduler_config(
346 + max_num_queued_reqs=100,
347 + max_num_queued_tokens=32000,
348 + )
349 + assert config.max_num_queued_reqs == 100
350 + assert config.max_num_queued_tokens == 32000
351 +
352 +
353 +@pytest.mark.parametrize("field", ["max_num_queued_reqs", "max_num_queued_tokens"])
354 +def test_scheduler_config_rejects_negative(field):
355 + with pytest.raises(ValidationError):
356 + _make_scheduler_config(**{field: -1})
357 +
358 +
359 +# ---------------------------------------------------------------------------
360 +# human_readable_int for CLI notation
361 +# ---------------------------------------------------------------------------
362 +
363 +
364 +@pytest.mark.parametrize(
365 + "input_str, expected",
366 + [
367 + ("32k", 32_000),
368 + ("1k", 1_000),
369 + ("1K", 1_024),
370 + ("1m", 1_000_000),
371 + ("1M", 1_048_576),
372 + ("100", 100),
373 + ("2.5k", 2_500),
374 + ("0", 0),
375 + ],
376 +)
377 +def test_human_readable_int_parses_notation(input_str: str, expected: int):
378 + assert human_readable_int(input_str) == expected
379 +
380 +
381 +@pytest.mark.parametrize("invalid", ["abc", "1x", "", "k", "1.5K"])
382 +def test_human_readable_int_rejects_invalid(invalid: str):
383 + with pytest.raises((argparse.ArgumentTypeError, ValueError)):
384 + human_readable_int(invalid)
vllm/config/scheduler.py
+31
@@ -71,6 +71,37 @@ class SchedulerConfig:
71 """For chunked prefill, a request is considered long if the prompt is
72 longer than this number of tokens. 0 disables the cap (default)."""
73
74 + max_num_queued_reqs: int | None = Field(default=None, ge=0)
75 + """Maximum number of requests that can be in-flight (waiting or running)
76 + at the same time, or None for no limit. When the limit is reached, new
77 + requests are rejected with HTTP 503 so the client can retry on another
78 + instance. This bounds vLLM's otherwise unbounded request queue and is
79 + primarily a coarse capacity valve."""
80 +
81 + max_num_queued_tokens: int | None = Field(default=None, ge=0)
82 + """Maximum total prompt tokens of requests currently in the prefill
83 + phase, or None for no limit. When the limit is reached, new requests
84 + are rejected with HTTP 503.
85 +
86 + This is a TTFT QoS mechanism: by setting it to
87 + ``target_TTFT * prefill_throughput`` you reject requests when the
88 + prefill backlog would exceed the latency target. In a disaggregated
89 + prefill-decode setup this maps directly to the prefill pool's
90 + capacity.
91 +
92 + Note: the count is conservative. A partially prefilled request
93 + still contributes its full ``prompt_len`` until it transitions out
94 + of the prefill phase, because the scheduler's per-iteration
95 + ``num_computed_tokens`` progress is not propagated to the API
96 + server process during prefill (``EngineCoreOutput`` is only
97 + emitted once the request starts producing tokens). Similarly,
98 + prefix-cache hits (``num_cached_tokens``) are only known to the
99 + OutputProcessor after prefill completes. This overestimates the
100 + real backlog, causing earlier rejection than strictly necessary
101 + — the safe direction for QoS. The impact is limited to long
102 + prompts under chunked prefill; short prompts that prefill in a
103 + single iteration are unaffected."""
104 +
105 enable_chunked_prefill: bool = True
106 """If True, prefill requests can be chunked based
107 on the remaining `max_num_batched_tokens`.
vllm/engine/arg_utils.py
+12
@@ -367,6 +367,7 @@ def _compute_kwargs(cls: ConfigType) -> dict[str, dict[str, Any]]:
367 "max_num_scheduled_tokens",
368 "kv_cache_memory_bytes",
369 "safetensors_prefetch_block_size",
370 + "max_num_queued_tokens",
371 }
372 if name == "max_model_len":
373 kwargs[name]["type"] = human_readable_int_or_auto
@@ -544,6 +545,8 @@ class EngineArgs:
545 max_num_scheduled_tokens: int | None = None
546 long_prefill_token_threshold: int = SchedulerConfig.long_prefill_token_threshold
547 max_num_seqs: int | None = None
548 + max_num_queued_reqs: int | None = None
549 + max_num_queued_tokens: int | None = None
550 max_logprobs: int = ModelConfig.max_logprobs
551 logprobs_mode: LogprobsMode = ModelConfig.logprobs_mode
552 use_fp64_gumbel: bool = ModelConfig.use_fp64_gumbel
@@ -1554,6 +1557,13 @@ class EngineArgs:
1557 "default": None,
1558 },
1559 )
1560 + scheduler_group.add_argument(
1561 + "--max-num-queued-reqs", **scheduler_kwargs["max_num_queued_reqs"]
1562 + )
1563 + scheduler_group.add_argument(
1564 + "--max-num-queued-tokens",
1565 + **scheduler_kwargs["max_num_queued_tokens"],
1566 + )
1567 scheduler_group.add_argument(
1568 "--long-prefill-token-threshold",
1569 **scheduler_kwargs["long_prefill_token_threshold"],
@@ -2348,6 +2358,8 @@ class EngineArgs:
2358 max_num_batched_tokens=self.max_num_batched_tokens,
2359 max_num_scheduled_tokens=self.max_num_scheduled_tokens,
2360 max_num_seqs=self.max_num_seqs,
2361 + max_num_queued_reqs=self.max_num_queued_reqs,
2362 + max_num_queued_tokens=self.max_num_queued_tokens,
2363 max_model_len=model_config.max_model_len,
2364 enable_chunked_prefill=self.enable_chunked_prefill,
2365 disable_chunked_mm_input=self.disable_chunked_mm_input,
vllm/engine/protocol.py
+17
@@ -62,6 +62,23 @@ class EngineClient(ABC):
62 @abstractmethod
63 def dead_error(self) -> BaseException: ...
64
65 + def check_admission( # noqa: B027
66 + self, n: int = 1, request_id: str | None = None
67 + ) -> None:
68 + """Reject the request up front if it would exceed queue limits.
69 +
70 + Called before a response is started so that overload rejections can
71 + carry an HTTP status, which is not possible once a streaming response
72 + has begun. Engines without admission control accept everything.
73 +
74 + Args:
75 + n: Number of sequences the request will occupy.
76 + request_id: Request id, used for logging only.
77 +
78 + Raises:
79 + GracefulHTTPError: If the request cannot be admitted.
80 + """
81 +
82 @abstractmethod
83 def generate(
84 self,
vllm/entrypoints/generate/base/serving.py
+13
@@ -171,6 +171,19 @@ class GenerateBaseServing(BaseServing, BeamSearchOnlineMixin):
171 # Never fail server startup over the fingerprint.
172 self.system_fingerprint = None
173
174 + def _preflight(self, n: int = 1) -> None:
175 + """Engine checks that must run before a response is started.
176 +
177 + This is required for the streaming case, where we return a success
178 + status before we actually start generating text :).
179 +
180 + Args:
181 + n: Number of sequences the request will occupy.
182 + """
183 + if self.engine_client.errored:
184 + raise self.engine_client.dead_error
185 + self.engine_client.check_admission(n)
186 +
187 def create_streaming_error_response(
188 self,
189 message: str | Exception,
vllm/entrypoints/generate/generative_scoring/serving.py
+1
@@ -195,6 +195,7 @@ class ServingGenerativeScoring(BaseServing):
195 # Check if engine is alive
196 if self.engine_client.errored:
197 raise self.engine_client.dead_error
198 + self.engine_client.check_admission(len(request.items))
199
200 # Get tokenizer
201 tokenizer = self.renderer.tokenizer
vllm/entrypoints/openai/chat_completion/batch_serving.py
+1 -2
@@ -56,8 +56,7 @@ class OpenAIServingChatBatch(OpenAIServingChat):
56 logger.error("Error with model %s", error_check_ret)
57 return error_check_ret
58
59 - if self.engine_client.errored:
60 - raise self.engine_client.dead_error
59 + self._preflight()
60
61 renderer = self.online_renderer
62
vllm/entrypoints/openai/chat_completion/serving.py
+1 -5
@@ -235,11 +235,7 @@ class OpenAIServingChat(GenerateBaseServing):
235 logger.error("Error with model %s", error_check_ret)
236 return error_check_ret
237
238 - # If the engine is dead, raise the engine's DEAD_ERROR.
239 - # This is required for the streaming case, where we return a
240 - # success status before we actually start generating text :).
241 - if self.engine_client.errored:
242 - raise self.engine_client.dead_error
238 + self._preflight(request.n or 1)
239
240 return await self.online_renderer.render_chat(request)
241
vllm/entrypoints/openai/completion/serving.py
+1 -5
@@ -102,11 +102,7 @@ class OpenAIServingCompletion(GenerateBaseServing):
102 if error_check_ret is not None:
103 return error_check_ret
104
105 - # If the engine is dead, raise the engine's DEAD_ERROR.
106 - # This is required for the streaming case, where we return a
107 - # success status before we actually start generating text :).
108 - if self.engine_client.errored:
109 - raise self.engine_client.dead_error
105 + self._preflight(request.n or 1)
106
107 return await self.online_renderer.render_completion(request)
108
vllm/entrypoints/openai/responses/serving.py
+1 -5
@@ -350,11 +350,7 @@ class OpenAIServingResponses(GenerateBaseServing):
350 if maybe_validation_error is not None:
351 return maybe_validation_error
352
353 - # If the engine is dead, raise the engine's DEAD_ERROR.
354 - # This is required for the streaming case, where we return a
355 - # success status before we actually start generating text :).
356 - if self.engine_client.errored:
357 - raise self.engine_client.dead_error
353 + self._preflight()
354
355 if request.store and not self.enable_store:
356 # Disable the store option.
vllm/entrypoints/scale_out/token_in_token_out/serving.py
+1 -5
@@ -110,11 +110,7 @@ class ServingTokens(GenerateBaseServing):
110 logger.error("Error with model %s", error_check_ret)
111 return error_check_ret
112
113 - # If the engine is dead, raise the engine's DEAD_ERROR.
114 - # This is required for the streaming case, where we return a
115 - # success status before we actually start generating text :).
116 - if self.engine_client.errored:
117 - raise self.engine_client.dead_error
113 + self._preflight()
114
115 lora_request = None
116 lora_request = self._maybe_get_adapters(request, supports_default_mm_loras=True)
vllm/entrypoints/serve/exception_handling/error_response.py
+6 -1
@@ -26,6 +26,7 @@ def create_error_response(
26 )
27
28 from vllm.exceptions import (
29 + GracefulHTTPError,
30 VLLMClientError,
31 VLLMNotFoundError,
32 VLLMServerError,
@@ -33,7 +34,11 @@ def create_error_response(
34 VLLMValidationError,
35 )
36
36 - if isinstance(exc, VLLMValidationError):
37 + if isinstance(exc, GracefulHTTPError):
38 + err_type = HTTPStatus(exc.http_status).phrase
39 + status_code = exc.http_status
40 + param = None
41 + elif isinstance(exc, VLLMValidationError):
42 err_type = "BadRequestError"
43 status_code = HTTPStatus.BAD_REQUEST
44 param = exc.parameter
vllm/entrypoints/speech_to_text/base/serving.py
+1 -5
@@ -440,11 +440,7 @@ class SpeechToTextBaseServing(GenerateBaseServing):
440 if not request.model:
441 request.model = self.models.model_name()
442
443 - # If the engine is dead, raise the engine's DEAD_ERROR.
444 - # This is required for the streaming case, where we return a
445 - # success status before we actually start generating text :).
446 - if self.engine_client.errored:
447 - raise self.engine_client.dead_error
443 + self._preflight()
444
445 if request.response_format not in [
446 "text",
vllm/exceptions.py
+45
@@ -125,3 +125,48 @@ class GenerationError(VLLMServerError):
125 def __init__(self, message: str = "Internal server error"):
126 super().__init__(message)
127 self.status_code = HTTPStatus.INTERNAL_SERVER_ERROR
128 +
129 +
130 +class GracefulHTTPError(VLLMError):
131 + """Exception that should be translated into an HTTP error response.
132 +
133 + These are expected to occur during normal operation (e.g. admission
134 + control rejections) and should be surfaced to the client with the
135 + explicit HTTP status code they carry, rather than being mapped to a
136 + generic 4xx/5xx by the client/server split.
137 + """
138 +
139 + def __init__(self, message: str, http_status: HTTPStatus):
140 + super().__init__(message)
141 + self.message = message
142 + self.http_status = http_status
143 +
144 +
145 +class QueueOverflowError(GracefulHTTPError):
146 + """Raised when admitting a request would exceed the request queue limit.
147 +
148 + Returns HTTP 503 (Service Unavailable) so that load balancers and
149 + client SDKs retry the request on a different instance.
150 + """
151 +
152 + def __init__(self):
153 + super().__init__(
154 + "The engine is currently busy and cannot accept new requests. "
155 + "Please try again later or on a different instance.",
156 + HTTPStatus.SERVICE_UNAVAILABLE,
157 + )
158 +
159 +
160 +class MaxQueuedTokensError(GracefulHTTPError):
161 + """Raised when the pending prefill tokens exceed the configured limit.
162 +
163 + Returns HTTP 503 (Service Unavailable) so that load balancers and
164 + client SDKs retry the request on a different instance.
165 + """
166 +
167 + def __init__(self):
168 + super().__init__(
169 + "The engine has reached its prefill token backlog limit. "
170 + "Please try again later or on a different instance.",
171 + HTTPStatus.SERVICE_UNAVAILABLE,
172 + )
vllm/v1/engine/async_llm.py
+80 -6
@@ -21,7 +21,13 @@ from vllm.distributed.weight_transfer.base import (
21 from vllm.engine.arg_utils import AsyncEngineArgs
22 from vllm.engine.protocol import EngineClient, StreamingInput
23 from vllm.entrypoints.serve.elastic_ep.middleware import set_scaling_elastic_ep
24 -from vllm.exceptions import VLLMClientError, VLLMValidationError
24 +from vllm.exceptions import (
25 + GracefulHTTPError,
26 + MaxQueuedTokensError,
27 + QueueOverflowError,
28 + VLLMClientError,
29 + VLLMValidationError,
30 +)
31 from vllm.inputs import EngineInput, PromptType
32 from vllm.logger import init_logger
33 from vllm.lora.request import LoRARequest
@@ -112,6 +118,7 @@ class AsyncLLM(EngineClient):
118 self.vllm_config = vllm_config
119 self._elastic_ep_lock = asyncio.Lock()
120 self.model_config = vllm_config.model_config
121 + self.scheduler_config = vllm_config.scheduler_config
122 self.observability_config = vllm_config.observability_config
123
124 tracing_endpoint = self.observability_config.otlp_traces_endpoint
@@ -273,6 +280,64 @@ class AsyncLLM(EngineClient):
280 if handler is not None:
281 cancel_task_threadsafe(handler)
282
283 + def get_num_unfinished_requests(self) -> int:
284 + return self.output_processor.get_num_unfinished_requests()
285 +
286 + def get_num_queued_tokens(self) -> int:
287 + return self.output_processor.get_num_queued_tokens()
288 +
289 + def check_admission(self, n: int = 1, request_id: str | None = None) -> None:
290 + """Reject the request if it would exceed queue limits.
291 +
292 + Both limits return HTTP 503 (Service Unavailable) so that load
293 + balancers and client SDKs retry on a different instance.
294 +
295 + - ``max_num_queued_reqs``: hard cap on the number of unfinished requests
296 + (waiting + running). A request with ``n > 1`` counts as ``n`` slots.
297 + - ``max_num_queued_tokens``: TTFT QoS — cap on the total prompt
298 + tokens of requests still in prefill.
299 +
300 + Note: ``get_num_queued_tokens`` uses ``prompt_len`` for all prefilling requests.
301 + Chunked prefill progress and prefix-cache hits are not subtracted because the
302 + scheduler's ``num_computed_tokens`` and ``num_cached_tokens`` are only
303 + propagated to the API server after prefill completes. The overestimation is
304 + conservative — earlier rejection, preserving TTFT targets.
305 +
306 + Args:
307 + n: Number of sequences the request will occupy.
308 + request_id: Request id, used for logging only.
309 +
310 + Raises:
311 + QueueOverflowError: If ``max_num_queued_reqs`` would be exceeded.
312 + MaxQueuedTokensError: If ``max_num_queued_tokens`` would be exceeded.
313 + """
314 + max_num_reqs = self.scheduler_config.max_num_queued_reqs
315 + if max_num_reqs is not None:
316 + current = self.get_num_unfinished_requests()
317 + if current + n > max_num_reqs:
318 + logger.info(
319 + "Request queue full - rejecting request %s "
320 + "(current=%d, n=%d, max=%d).",
321 + request_id,
322 + current,
323 + n,
324 + max_num_reqs,
325 + )
326 + raise QueueOverflowError()
327 +
328 + max_queued_tokens = self.scheduler_config.max_num_queued_tokens
329 + if max_queued_tokens is not None:
330 + current_tokens = self.get_num_queued_tokens()
331 + if current_tokens >= max_queued_tokens:
332 + logger.info(
333 + "Max queued tokens reached - rejecting request %s "
334 + "(current_tokens=%d, max=%d).",
335 + request_id,
336 + current_tokens,
337 + max_queued_tokens,
338 + )
339 + raise MaxQueuedTokensError()
340 +
341 async def get_supported_tasks(self) -> tuple[SupportedTask, ...]:
342 if not hasattr(self, "_supported_tasks"):
343 # Cache the result
@@ -317,6 +382,10 @@ class AsyncLLM(EngineClient):
382 "prompt logprobs"
383 )
384
385 + if isinstance(params, SamplingParams) and params.n > 1:
386 + # TODO (NickLucche) Batch check admission check for all n requests
387 + self.check_admission(params.n, request_id)
388 +
389 if isinstance(prompt, AsyncGenerator):
390 if reasoning_ended is not None or reasoning_parser_kwargs is not None:
391 raise NotImplementedError
@@ -430,7 +499,12 @@ class AsyncLLM(EngineClient):
499 index: int,
500 queue: RequestOutputCollector,
501 ):
433 - # Add the request to OutputProcessor (this process).
502 + if parent_req is None and not self.output_processor.has_request(
503 + request.request_id
504 + ):
505 + self.check_admission(request_id=request.request_id)
506 +
507 + # Register locally before the first await so concurrent tasks see this request.
508 self.output_processor.add_request(request, prompt, parent_req, index, queue)
509
510 # Add the EngineCoreRequest to EngineCore (separate process).
@@ -630,8 +704,8 @@ class AsyncLLM(EngineClient):
704 logger.info("Request %s failed (engine dead).", request_id)
705 raise
706
633 - # Request validation error.
634 - except VLLMClientError as e:
707 + # Request validation error or admission control rejection.
708 + except (VLLMClientError, GracefulHTTPError) as e:
709 if self.log_requests:
710 logger.info("Request %s failed (bad request): %s.", request_id, e)
711 raise
@@ -907,8 +981,8 @@ class AsyncLLM(EngineClient):
981 logger.info("Request %s failed (engine dead).", request_id)
982 raise
983
910 - # Request validation error.
911 - except VLLMClientError:
984 + # Request validation error or admission control rejection.
985 + except (VLLMClientError, GracefulHTTPError):
986 if self.log_requests:
987 logger.info("Request %s failed (bad request).", request_id)
988 raise
vllm/v1/engine/output_processor.py
+14
@@ -467,6 +467,20 @@ class OutputProcessor:
467 def get_num_unfinished_requests(self):
468 return len(self.request_states)
469
470 + def has_request(self, request_id: str) -> bool:
471 + return request_id in self.request_states
472 +
473 + def get_num_queued_tokens(self) -> int:
474 + """Total prompt tokens of requests currently in the prefill phase.
475 +
476 + Uses ``prompt_len`` rather than remaining prefill work because the
477 + scheduler's ``num_computed_tokens`` is not propagated to the API
478 + server until prefill completes. See ``SchedulerConfig`` docs.
479 + """
480 + return sum(
481 + req.prompt_len for req in self.request_states.values() if req.is_prefilling
482 + )
483 +
484 def has_unfinished_requests(self) -> bool:
485 return len(self.request_states) > 0
486