| 1 | # SPDX-License-Identifier: Apache-2.0 |
| 2 | # SPDX-FileCopyrightText: Copyright contributors to the vLLM project |
| 3 | """Timing-accounting invariants for the streaming benchmark request functions. |
| 4 | |
| 5 | serve.py derives TPOT as (latency - ttft) / (output_len - 1), which only means |
| 6 | "average inter-token latency" if latency - ttft == sum(itl). These tests pin |
| 7 | that identity, and pin that the trailing choice-less usage chunk does not |
| 8 | extend the measured end of the request. |
| 9 | |
| 10 | Driven against a fake session and a scripted clock: no GPU, model, or socket. |
| 11 | """ |
| 12 | |
| 13 | import asyncio |
| 14 | import importlib.util |
| 15 | import json |
| 16 | from pathlib import Path |
| 17 | from typing import Any |
| 18 | |
| 19 | import aiohttp |
| 20 | import numpy as np |
| 21 | import pytest |
| 22 | |
| 23 | import vllm.benchmarks.lib.endpoint_request_func as request_func_module |
| 24 | from vllm.benchmarks.lib.endpoint_request_func import ( |
| 25 | RequestFuncInput, |
| 26 | async_request_openai_audio, |
| 27 | async_request_openai_chat_completions, |
| 28 | async_request_openai_completions, |
| 29 | ) |
| 30 | |
| 31 | pytestmark = pytest.mark.skip_global_cleanup |
| 32 | |
| 33 | AUDIO_SAMPLE_RATE = 16_000 |
| 34 | |
| 35 | |
| 36 | class _ScriptedClock: |
| 37 | """A deterministic stand-in for time.perf_counter. |
| 38 | |
| 39 | Each read advances the clock by exactly one tick, so what a code path |
| 40 | observes depends only on how many times it reads the clock. A stray or |
| 41 | misattributed read then shows up as a whole tick of drift rather than as |
| 42 | sub-microsecond jitter, which keeps these assertions free of flakiness. |
| 43 | """ |
| 44 | |
| 45 | def __init__(self, tick: float = 1.0, start: float = 1000.0) -> None: |
| 46 | self.tick = tick |
| 47 | self._now = start |
| 48 | |
| 49 | def __call__(self) -> float: |
| 50 | value = self._now |
| 51 | self._now += self.tick |
| 52 | return value |
| 53 | |
| 54 | |
| 55 | class _FakeTime: |
| 56 | """Namespace substituted for the module-level time import.""" |
| 57 | |
| 58 | def __init__(self, perf_counter) -> None: |
| 59 | self.perf_counter = perf_counter |
| 60 | |
| 61 | |
| 62 | def _sse(payload: dict[str, Any]) -> bytes: |
| 63 | return f"data: {json.dumps(payload)}\n\n".encode() |
| 64 | |
| 65 | |
| 66 | def _token_chunk(index: int, *, chat: bool) -> bytes: |
| 67 | if chat: |
| 68 | choice = {"index": 0, "delta": {"content": f"tok{index} "}} |
| 69 | else: |
| 70 | choice = {"index": 0, "text": f"tok{index} "} |
| 71 | return _sse({"id": "cmpl-test", "object": "x", "choices": [choice]}) |
| 72 | |
| 73 | |
| 74 | def _usage_chunk(n_tokens: int) -> bytes: |
| 75 | return _sse( |
| 76 | { |
| 77 | "id": "cmpl-test", |
| 78 | "object": "x", |
| 79 | "choices": [], |
| 80 | "usage": { |
| 81 | "prompt_tokens": 7, |
| 82 | "completion_tokens": n_tokens, |
| 83 | "total_tokens": 7 + n_tokens, |
| 84 | }, |
| 85 | } |
| 86 | ) |
| 87 | |
| 88 | |
| 89 | def _build_stream( |
| 90 | n_tokens: int, |
| 91 | *, |
| 92 | chat: bool, |
| 93 | ttft_delay: float = 0.0, |
| 94 | itl_delay: float = 0.0, |
| 95 | usage_delay: float = 0.0, |
| 96 | ) -> list[tuple[float, bytes]]: |
| 97 | """Build ``(delay_before_chunk, payload)`` pairs for one streamed response. |
| 98 | |
| 99 | ``vllm bench serve`` always requests ``stream_options.include_usage``, so a |
| 100 | choice-less usage chunk always trails the final token. |
| 101 | """ |
| 102 | chunks = [ |
| 103 | (ttft_delay if i == 0 else itl_delay, _token_chunk(i, chat=chat)) |
| 104 | for i in range(n_tokens) |
| 105 | ] |
| 106 | chunks.append((usage_delay, _usage_chunk(n_tokens))) |
| 107 | chunks.append((0.0, b"data: [DONE]\n\n")) |
| 108 | return chunks |
| 109 | |
| 110 | |
| 111 | class _FakeContent: |
| 112 | def __init__(self, chunks: list[tuple[float, bytes]]) -> None: |
| 113 | self._chunks = chunks |
| 114 | |
| 115 | async def iter_any(self): |
| 116 | for delay, payload in self._chunks: |
| 117 | if delay: |
| 118 | await asyncio.sleep(delay) |
| 119 | yield payload |
| 120 | |
| 121 | # The legacy script iterates ``response.content`` directly. |
| 122 | def __aiter__(self): |
| 123 | return self.iter_any() |
| 124 | |
| 125 | |
| 126 | class _FakeResponse: |
| 127 | def __init__(self, chunks: list[tuple[float, bytes]]) -> None: |
| 128 | self.status = 200 |
| 129 | self.reason = "OK" |
| 130 | self.content = _FakeContent(chunks) |
| 131 | |
| 132 | async def __aenter__(self) -> "_FakeResponse": |
| 133 | return self |
| 134 | |
| 135 | async def __aexit__(self, exc_type, exc, tb) -> bool: |
| 136 | return False |
| 137 | |
| 138 | |
| 139 | class _FakeSession: |
| 140 | """Returns a scripted SSE stream for any request, without touching a socket.""" |
| 141 | |
| 142 | def __init__(self, chunks: list[tuple[float, bytes]]) -> None: |
| 143 | self._chunks = chunks |
| 144 | |
| 145 | def post(self, **kwargs: Any) -> _FakeResponse: |
| 146 | del kwargs |
| 147 | return _FakeResponse(self._chunks) |
| 148 | |
| 149 | async def __aenter__(self) -> "_FakeSession": |
| 150 | return self |
| 151 | |
| 152 | async def __aexit__(self, exc_type, exc, tb) -> bool: |
| 153 | return False |
| 154 | |
| 155 | |
| 156 | # (id, request function, api url, is_chat_payload) |
| 157 | STREAMING_ENDPOINTS = [ |
| 158 | ( |
| 159 | "completions", |
| 160 | async_request_openai_completions, |
| 161 | "http://test/v1/completions", |
| 162 | False, |
| 163 | ), |
| 164 | ( |
| 165 | "chat", |
| 166 | async_request_openai_chat_completions, |
| 167 | "http://test/v1/chat/completions", |
| 168 | True, |
| 169 | ), |
| 170 | ( |
| 171 | "audio", |
| 172 | async_request_openai_audio, |
| 173 | "http://test/v1/audio/transcriptions", |
| 174 | True, |
| 175 | ), |
| 176 | ] |
| 177 | |
| 178 | |
| 179 | def _make_input(api_url: str, n_tokens: int, endpoint_id: str) -> RequestFuncInput: |
| 180 | multi_modal_content = None |
| 181 | if endpoint_id == "audio": |
| 182 | # A short silent clip; only its duration is read by the request func. |
| 183 | multi_modal_content = { |
| 184 | "audio": ( |
| 185 | np.zeros(AUDIO_SAMPLE_RATE // 10, dtype=np.float32), |
| 186 | AUDIO_SAMPLE_RATE, |
| 187 | ) |
| 188 | } |
| 189 | return RequestFuncInput( |
| 190 | prompt="hello", |
| 191 | api_url=api_url, |
| 192 | prompt_len=7, |
| 193 | output_len=n_tokens, |
| 194 | model="test-model", |
| 195 | multi_modal_content=multi_modal_content, |
| 196 | ) |
| 197 | |
| 198 | |
| 199 | def _run( |
| 200 | request_func, |
| 201 | api_url: str, |
| 202 | endpoint_id: str, |
| 203 | chunks, |
| 204 | n_tokens: int, |
| 205 | *, |
| 206 | expect_success: bool = True, |
| 207 | ): |
| 208 | output = asyncio.run( |
| 209 | request_func( |
| 210 | _make_input(api_url, n_tokens, endpoint_id), |
| 211 | _FakeSession(chunks), |
| 212 | ) |
| 213 | ) |
| 214 | if expect_success: |
| 215 | assert output.success, output.error |
| 216 | return output |
| 217 | |
| 218 | |
| 219 | @pytest.mark.parametrize( |
| 220 | "endpoint_id,request_func,api_url,chat", |
| 221 | STREAMING_ENDPOINTS, |
| 222 | ids=[e[0] for e in STREAMING_ENDPOINTS], |
| 223 | ) |
| 224 | def test_decode_span_identity_is_exact( |
| 225 | monkeypatch: pytest.MonkeyPatch, |
| 226 | endpoint_id: str, |
| 227 | request_func, |
| 228 | api_url: str, |
| 229 | chat: bool, |
| 230 | ) -> None: |
| 231 | """Latency - ttft must equal sum(itl) exactly. |
| 232 | |
| 233 | On a scripted clock this can only drift if a code path reads the clock an |
| 234 | extra time or attributes a read to the wrong event. |
| 235 | """ |
| 236 | tick = 1.0 |
| 237 | monkeypatch.setattr( |
| 238 | request_func_module, "time", _FakeTime(_ScriptedClock(tick=tick)) |
| 239 | ) |
| 240 | |
| 241 | n_tokens = 5 |
| 242 | chunks = _build_stream(n_tokens, chat=chat) |
| 243 | output = _run(request_func, api_url, endpoint_id, chunks, n_tokens) |
| 244 | |
| 245 | assert len(output.itl) == n_tokens - 1 |
| 246 | assert output.output_tokens == n_tokens |
| 247 | |
| 248 | residual = (output.latency - output.ttft) - sum(output.itl) |
| 249 | assert residual == pytest.approx(0.0, abs=tick / 1000), ( |
| 250 | f"{endpoint_id}: (latency - ttft) - sum(itl) = {residual!r}; " |
| 251 | f"TPOT is derived from latency - ttft, so it must match the measured " |
| 252 | f"inter-token latencies" |
| 253 | ) |
| 254 | |
| 255 | |
| 256 | @pytest.mark.parametrize( |
| 257 | "endpoint_id,request_func,api_url,chat", |
| 258 | STREAMING_ENDPOINTS, |
| 259 | ids=[e[0] for e in STREAMING_ENDPOINTS], |
| 260 | ) |
| 261 | def test_trailing_usage_chunk_does_not_extend_latency( |
| 262 | monkeypatch: pytest.MonkeyPatch, |
| 263 | endpoint_id: str, |
| 264 | request_func, |
| 265 | api_url: str, |
| 266 | chat: bool, |
| 267 | ) -> None: |
| 268 | """E2E latency must end at the last token, not at the usage trailer. |
| 269 | |
| 270 | The usage chunk carries no token, so counting it stretches E2EL (and |
| 271 | therefore TPOT) by the delivery time of one extra SSE frame. The [DONE] |
| 272 | sentinel is already excluded on the same grounds. |
| 273 | """ |
| 274 | tick = 1.0 |
| 275 | clock = _ScriptedClock(tick=tick) |
| 276 | monkeypatch.setattr(request_func_module, "time", _FakeTime(clock)) |
| 277 | |
| 278 | n_tokens = 5 |
| 279 | chunks = _build_stream(n_tokens, chat=chat) |
| 280 | output = _run(request_func, api_url, endpoint_id, chunks, n_tokens) |
| 281 | |
| 282 | # Clock reads: 1 for the start time, then 1 per non-[DONE] chunk. The last |
| 283 | # token is therefore read at start + n_tokens ticks; the usage chunk one |
| 284 | # tick later. |
| 285 | expected_latency = n_tokens * tick |
| 286 | assert output.latency == pytest.approx(expected_latency, abs=tick / 1000), ( |
| 287 | f"{endpoint_id}: latency {output.latency} should stop at the final " |
| 288 | f"token ({expected_latency}), not run on to the usage chunk " |
| 289 | f"({expected_latency + tick})" |
| 290 | ) |
| 291 | |
| 292 | |
| 293 | @pytest.mark.parametrize( |
| 294 | "endpoint_id,request_func,api_url,chat", |
| 295 | STREAMING_ENDPOINTS, |
| 296 | ids=[e[0] for e in STREAMING_ENDPOINTS], |
| 297 | ) |
| 298 | def test_decode_span_identity_holds_with_real_clock( |
| 299 | endpoint_id: str, |
| 300 | request_func, |
| 301 | api_url: str, |
| 302 | chat: bool, |
| 303 | ) -> None: |
| 304 | """The identity also holds against the real clock, with real gaps. |
| 305 | |
| 306 | The residual differences out the same recorded timestamps, so wall clock |
| 307 | jitter cancels and only float rounding remains; a microsecond bound is |
| 308 | ample rather than tight. |
| 309 | """ |
| 310 | n_tokens = 4 |
| 311 | chunks = _build_stream( |
| 312 | n_tokens, chat=chat, ttft_delay=0.02, itl_delay=0.01, usage_delay=0.02 |
| 313 | ) |
| 314 | output = _run(request_func, api_url, endpoint_id, chunks, n_tokens) |
| 315 | |
| 316 | residual = (output.latency - output.ttft) - sum(output.itl) |
| 317 | assert abs(residual) < 1e-6, ( |
| 318 | f"{endpoint_id}: (latency - ttft) - sum(itl) drifted by {residual:.9f}s" |
| 319 | ) |
| 320 | |
| 321 | |
| 322 | @pytest.mark.parametrize( |
| 323 | "endpoint_id,request_func,api_url,chat", |
| 324 | STREAMING_ENDPOINTS, |
| 325 | ids=[e[0] for e in STREAMING_ENDPOINTS], |
| 326 | ) |
| 327 | def test_usage_only_stream_is_not_reported_as_success( |
| 328 | endpoint_id: str, |
| 329 | request_func, |
| 330 | api_url: str, |
| 331 | chat: bool, |
| 332 | ) -> None: |
| 333 | """A stream with no token chunk must fail rather than report 0 ms. |
| 334 | |
| 335 | Only token chunks move the end of the request, so a degenerate stream |
| 336 | carrying nothing but a usage trailer leaves latency at zero. Reporting that |
| 337 | as a success would feed a zero-duration request into the benchmark |
| 338 | aggregates. |
| 339 | """ |
| 340 | chunks = [(0.0, _usage_chunk(0)), (0.0, b"data: [DONE]\n\n")] |
| 341 | output = _run(request_func, api_url, endpoint_id, chunks, 0, expect_success=False) |
| 342 | |
| 343 | assert not output.success |
| 344 | assert output.itl == [] |
| 345 | assert "TTFT" in output.error |
| 346 | |
| 347 | |
| 348 | # ``benchmarks/backend_request_func.py`` is a standalone script rather than a |
| 349 | # package module, so it is loaded by path. It also builds its own ClientSession, |
| 350 | # so the scripted stream is injected by swapping the module's aiohttp reference. |
| 351 | _LEGACY_PATH = Path(__file__).parents[2] / "benchmarks" / "backend_request_func.py" |
| 352 | |
| 353 | |
| 354 | class _FakeAiohttp: |
| 355 | """Stands in for the legacy module's aiohttp import.""" |
| 356 | |
| 357 | def __init__(self, chunks: list[tuple[float, bytes]]) -> None: |
| 358 | self._chunks = chunks |
| 359 | |
| 360 | def __getattr__(self, name: str) -> Any: |
| 361 | return getattr(aiohttp, name) |
| 362 | |
| 363 | def ClientSession(self, **kwargs: Any) -> _FakeSession: # noqa: N802 |
| 364 | del kwargs |
| 365 | return _FakeSession(self._chunks) |
| 366 | |
| 367 | |
| 368 | def _load_legacy(): |
| 369 | spec = importlib.util.spec_from_file_location( |
| 370 | "legacy_backend_request_func", _LEGACY_PATH |
| 371 | ) |
| 372 | assert spec is not None and spec.loader is not None |
| 373 | module = importlib.util.module_from_spec(spec) |
| 374 | spec.loader.exec_module(module) |
| 375 | return module |
| 376 | |
| 377 | |
| 378 | LEGACY_ENDPOINTS = [ |
| 379 | ( |
| 380 | "completions", |
| 381 | "async_request_openai_completions", |
| 382 | "http://test/v1/completions", |
| 383 | False, |
| 384 | ), |
| 385 | ( |
| 386 | "chat", |
| 387 | "async_request_openai_chat_completions", |
| 388 | "http://test/v1/chat/completions", |
| 389 | True, |
| 390 | ), |
| 391 | ( |
| 392 | "audio", |
| 393 | "async_request_openai_audio", |
| 394 | "http://test/v1/audio/transcriptions", |
| 395 | True, |
| 396 | ), |
| 397 | ] |
| 398 | |
| 399 | |
| 400 | def _run_legacy( |
| 401 | monkeypatch, func_name, api_url, endpoint_id, chunks, n_tokens, clock=None |
| 402 | ): |
| 403 | module = _load_legacy() |
| 404 | monkeypatch.setattr(module, "aiohttp", _FakeAiohttp(chunks)) |
| 405 | if clock is not None: |
| 406 | monkeypatch.setattr(module, "time", _FakeTime(clock)) |
| 407 | request_func = getattr(module, func_name) |
| 408 | return asyncio.run(request_func(_make_input(api_url, n_tokens, endpoint_id))) |
| 409 | |
| 410 | |
| 411 | @pytest.mark.parametrize( |
| 412 | "endpoint_id,func_name,api_url,chat", |
| 413 | LEGACY_ENDPOINTS, |
| 414 | ids=[e[0] for e in LEGACY_ENDPOINTS], |
| 415 | ) |
| 416 | def test_legacy_decode_span_identity_is_exact( |
| 417 | monkeypatch: pytest.MonkeyPatch, |
| 418 | endpoint_id: str, |
| 419 | func_name: str, |
| 420 | api_url: str, |
| 421 | chat: bool, |
| 422 | ) -> None: |
| 423 | """benchmarks/backend_request_func.py must honour the same identity.""" |
| 424 | tick = 1.0 |
| 425 | n_tokens = 5 |
| 426 | chunks = _build_stream(n_tokens, chat=chat) |
| 427 | output = _run_legacy( |
| 428 | monkeypatch, |
| 429 | func_name, |
| 430 | api_url, |
| 431 | endpoint_id, |
| 432 | chunks, |
| 433 | n_tokens, |
| 434 | clock=_ScriptedClock(tick=tick), |
| 435 | ) |
| 436 | |
| 437 | assert output.success, output.error |
| 438 | assert len(output.itl) == n_tokens - 1 |
| 439 | |
| 440 | residual = (output.latency - output.ttft) - sum(output.itl) |
| 441 | assert residual == pytest.approx(0.0, abs=tick / 1000), ( |
| 442 | f"legacy {endpoint_id}: (latency - ttft) - sum(itl) = {residual!r}" |
| 443 | ) |
| 444 | assert output.latency == pytest.approx(n_tokens * tick, abs=tick / 1000) |
| 445 | |
| 446 | |
| 447 | @pytest.mark.parametrize( |
| 448 | "endpoint_id,func_name,api_url,chat", |
| 449 | LEGACY_ENDPOINTS, |
| 450 | ids=[e[0] for e in LEGACY_ENDPOINTS], |
| 451 | ) |
| 452 | def test_legacy_usage_only_stream_is_not_reported_as_success( |
| 453 | monkeypatch: pytest.MonkeyPatch, |
| 454 | endpoint_id: str, |
| 455 | func_name: str, |
| 456 | api_url: str, |
| 457 | chat: bool, |
| 458 | ) -> None: |
| 459 | """A usage-only stream must fail rather than report a zero-duration success.""" |
| 460 | chunks = [(0.0, _usage_chunk(0)), (0.0, b"data: [DONE]\n\n")] |
| 461 | output = _run_legacy(monkeypatch, func_name, api_url, endpoint_id, chunks, 0) |
| 462 | |
| 463 | assert not output.success |
| 464 | assert output.itl == [] |