iHore/vLLM
8
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

[Bugfix][Benchmark] Make streaming TTFT/E2E latency accounting consistent across endpoints (#55508)

Browse Source 
Signed-off-by: surajm20061998 <surajm20061998@gmail.com>
Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Suraj
2026-09-13 05:20:54 +00:00
committed by GitHub
co-authored by Claude Opus 5
parent 2671fedfc7
commit 93e051e290
3 changed files with 505 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
benchmarks/backend_request_func.py
+33 -17
View File
@@ -105,7 +105,7 @@ async def async_request_tgi(
timestamp = time.perf_counter()
# First token
if ttft == 0.0:
ttft = time.perf_counter() - st
ttft = timestamp - st
output.ttft = ttft
# Decoding phase
@@ -330,8 +330,7 @@ async def async_request_openai_completions(
# First token
if not first_chunk_received:
first_chunk_received = True
ttft = time.perf_counter() - st
output.ttft = ttft
output.ttft = timestamp - st
# Decoding phase
else:
@@ -416,7 +415,7 @@ async def async_request_openai_chat_completions(
output.prompt_len = request_func_input.prompt_len
generated_text = ""
ttft = 0.0
first_chunk_received = False
st = time.perf_counter()
most_recent_timestamp = st
try:
@@ -443,22 +442,30 @@ async def async_request_openai_chat_completions(
if choices := data.get("choices"):
content = choices[0]["delta"].get("content")
# First token
if ttft == 0.0:
ttft = timestamp - st
output.ttft = ttft
if not first_chunk_received:
first_chunk_received = True
output.ttft = timestamp - st
# Decoding phase
else:
output.itl.append(timestamp - most_recent_timestamp)
generated_text += content or ""
# Only token chunks advance the request end;
# the trailing usage chunk carries no token.
most_recent_timestamp = timestamp
elif usage := data.get("usage"):
output.output_tokens = usage.get("completion_tokens")
most_recent_timestamp = timestamp
output.generated_text = generated_text
output.success = True
if first_chunk_received:
output.success = True
else:
output.success = False
output.error = (
"Never received a valid chunk to calculate TTFT."
"This response will be marked as failed!"
)
output.latency = most_recent_timestamp - st
else:
output.error = response.reason or ""
@@ -530,7 +537,7 @@ async def async_request_openai_audio(
output.prompt_len = request_func_input.prompt_len
generated_text = ""
ttft = 0.0
first_chunk_received = False
st = time.perf_counter()
most_recent_timestamp = st
try:
@@ -551,9 +558,9 @@ async def async_request_openai_audio(
if choices := data.get("choices"):
content = choices[0]["delta"].get("content")
# First token
if ttft == 0.0:
ttft = timestamp - st
output.ttft = ttft
if not first_chunk_received:
first_chunk_received = True
output.ttft = timestamp - st
# Decoding phase
else:
@@ -562,15 +569,24 @@ async def async_request_openai_audio(
)
generated_text += content or ""
# Only token chunks advance the request
# end; the trailing usage chunk carries no
# token.
most_recent_timestamp = timestamp
elif usage := data.get("usage"):
output.output_tokens = usage.get(
"completion_tokens"
)
most_recent_timestamp = timestamp
output.generated_text = generated_text
output.success = True
if first_chunk_received:
output.success = True
else:
output.success = False
output.error = (
"Never received a valid chunk to calculate TTFT."
"This response will be marked as failed!"
)
output.latency = most_recent_timestamp - st
else:
output.error = response.reason or ""
tests/benchmarks/test_endpoint_request_func_timing.py
+464
View File
@@ -0,0 +1,464 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
"""Timing-accounting invariants for the streaming benchmark request functions.
serve.py derives TPOT as (latency - ttft) / (output_len - 1), which only means
"average inter-token latency" if latency - ttft == sum(itl). These tests pin
that identity, and pin that the trailing choice-less usage chunk does not
extend the measured end of the request.
Driven against a fake session and a scripted clock: no GPU, model, or socket.
"""
import asyncio
import importlib.util
import json
from pathlib import Path
from typing import Any
import aiohttp
import numpy as np
import pytest
import vllm.benchmarks.lib.endpoint_request_func as request_func_module
from vllm.benchmarks.lib.endpoint_request_func import (
RequestFuncInput,
async_request_openai_audio,
async_request_openai_chat_completions,
async_request_openai_completions,
)
pytestmark = pytest.mark.skip_global_cleanup
AUDIO_SAMPLE_RATE = 16_000
class _ScriptedClock:
"""A deterministic stand-in for time.perf_counter.
Each read advances the clock by exactly one tick, so what a code path
observes depends only on how many times it reads the clock. A stray or
misattributed read then shows up as a whole tick of drift rather than as
sub-microsecond jitter, which keeps these assertions free of flakiness.
"""
def __init__(self, tick: float = 1.0, start: float = 1000.0) -> None:
self.tick = tick
self._now = start
def __call__(self) -> float:
value = self._now
self._now += self.tick
return value
class _FakeTime:
"""Namespace substituted for the module-level time import."""
def __init__(self, perf_counter) -> None:
self.perf_counter = perf_counter
def _sse(payload: dict[str, Any]) -> bytes:
return f"data: {json.dumps(payload)}\n\n".encode()
def _token_chunk(index: int, *, chat: bool) -> bytes:
if chat:
choice = {"index": 0, "delta": {"content": f"tok{index} "}}
else:
choice = {"index": 0, "text": f"tok{index} "}
return _sse({"id": "cmpl-test", "object": "x", "choices": [choice]})
def _usage_chunk(n_tokens: int) -> bytes:
return _sse(
{
"id": "cmpl-test",
"object": "x",
"choices": [],
"usage": {
"prompt_tokens": 7,
"completion_tokens": n_tokens,
"total_tokens": 7 + n_tokens,
},
}
)
def _build_stream(
n_tokens: int,
*,
chat: bool,
ttft_delay: float = 0.0,
itl_delay: float = 0.0,
usage_delay: float = 0.0,
) -> list[tuple[float, bytes]]:
"""Build ``(delay_before_chunk, payload)`` pairs for one streamed response.
``vllm bench serve`` always requests ``stream_options.include_usage``, so a
choice-less usage chunk always trails the final token.
"""
chunks = [
(ttft_delay if i == 0 else itl_delay, _token_chunk(i, chat=chat))
for i in range(n_tokens)
]
chunks.append((usage_delay, _usage_chunk(n_tokens)))
chunks.append((0.0, b"data: [DONE]\n\n"))
return chunks
class _FakeContent:
def __init__(self, chunks: list[tuple[float, bytes]]) -> None:
self._chunks = chunks
async def iter_any(self):
for delay, payload in self._chunks:
if delay:
await asyncio.sleep(delay)
yield payload
# The legacy script iterates ``response.content`` directly.
def __aiter__(self):
return self.iter_any()
class _FakeResponse:
def __init__(self, chunks: list[tuple[float, bytes]]) -> None:
self.status = 200
self.reason = "OK"
self.content = _FakeContent(chunks)
async def __aenter__(self) -> "_FakeResponse":
return self
async def __aexit__(self, exc_type, exc, tb) -> bool:
return False
class _FakeSession:
"""Returns a scripted SSE stream for any request, without touching a socket."""
def __init__(self, chunks: list[tuple[float, bytes]]) -> None:
self._chunks = chunks
def post(self, **kwargs: Any) -> _FakeResponse:
del kwargs
return _FakeResponse(self._chunks)
async def __aenter__(self) -> "_FakeSession":
return self
async def __aexit__(self, exc_type, exc, tb) -> bool:
return False
# (id, request function, api url, is_chat_payload)
STREAMING_ENDPOINTS = [
(
"completions",
async_request_openai_completions,
"http://test/v1/completions",
False,
),
(
"chat",
async_request_openai_chat_completions,
"http://test/v1/chat/completions",
True,
),
(
"audio",
async_request_openai_audio,
"http://test/v1/audio/transcriptions",
True,
),
]
def _make_input(api_url: str, n_tokens: int, endpoint_id: str) -> RequestFuncInput:
multi_modal_content = None
if endpoint_id == "audio":
# A short silent clip; only its duration is read by the request func.
multi_modal_content = {
"audio": (
np.zeros(AUDIO_SAMPLE_RATE // 10, dtype=np.float32),
AUDIO_SAMPLE_RATE,
)
}
return RequestFuncInput(
prompt="hello",
api_url=api_url,
prompt_len=7,
output_len=n_tokens,
model="test-model",
multi_modal_content=multi_modal_content,
)
def _run(
request_func,
api_url: str,
endpoint_id: str,
chunks,
n_tokens: int,
*,
expect_success: bool = True,
):
output = asyncio.run(
request_func(
_make_input(api_url, n_tokens, endpoint_id),
_FakeSession(chunks),
)
)
if expect_success:
assert output.success, output.error
return output
@pytest.mark.parametrize(
"endpoint_id,request_func,api_url,chat",
STREAMING_ENDPOINTS,
ids=[e[0] for e in STREAMING_ENDPOINTS],
)
def test_decode_span_identity_is_exact(
monkeypatch: pytest.MonkeyPatch,
endpoint_id: str,
request_func,
api_url: str,
chat: bool,
) -> None:
"""latency - ttft must equal sum(itl) exactly.
On a scripted clock this can only drift if a code path reads the clock an
extra time or attributes a read to the wrong event.
"""
tick = 1.0
monkeypatch.setattr(
request_func_module, "time", _FakeTime(_ScriptedClock(tick=tick))
)
n_tokens = 5
chunks = _build_stream(n_tokens, chat=chat)
output = _run(request_func, api_url, endpoint_id, chunks, n_tokens)
assert len(output.itl) == n_tokens - 1
assert output.output_tokens == n_tokens
residual = (output.latency - output.ttft) - sum(output.itl)
assert residual == pytest.approx(0.0, abs=tick / 1000), (
f"{endpoint_id}: (latency - ttft) - sum(itl) = {residual!r}; "
f"TPOT is derived from latency - ttft, so it must match the measured "
f"inter-token latencies"
)
@pytest.mark.parametrize(
"endpoint_id,request_func,api_url,chat",
STREAMING_ENDPOINTS,
ids=[e[0] for e in STREAMING_ENDPOINTS],
)
def test_trailing_usage_chunk_does_not_extend_latency(
monkeypatch: pytest.MonkeyPatch,
endpoint_id: str,
request_func,
api_url: str,
chat: bool,
) -> None:
"""E2E latency must end at the last token, not at the usage trailer.
The usage chunk carries no token, so counting it stretches E2EL (and
therefore TPOT) by the delivery time of one extra SSE frame. The [DONE]
sentinel is already excluded on the same grounds.
"""
tick = 1.0
clock = _ScriptedClock(tick=tick)
monkeypatch.setattr(request_func_module, "time", _FakeTime(clock))
n_tokens = 5
chunks = _build_stream(n_tokens, chat=chat)
output = _run(request_func, api_url, endpoint_id, chunks, n_tokens)
# Clock reads: 1 for the start time, then 1 per non-[DONE] chunk. The last
# token is therefore read at start + n_tokens ticks; the usage chunk one
# tick later.
expected_latency = n_tokens * tick
assert output.latency == pytest.approx(expected_latency, abs=tick / 1000), (
f"{endpoint_id}: latency {output.latency} should stop at the final "
f"token ({expected_latency}), not run on to the usage chunk "
f"({expected_latency + tick})"
)
@pytest.mark.parametrize(
"endpoint_id,request_func,api_url,chat",
STREAMING_ENDPOINTS,
ids=[e[0] for e in STREAMING_ENDPOINTS],
)
def test_decode_span_identity_holds_with_real_clock(
endpoint_id: str,
request_func,
api_url: str,
chat: bool,
) -> None:
"""The identity also holds against the real clock, with real gaps.
The residual differences out the same recorded timestamps, so wall clock
jitter cancels and only float rounding remains; a microsecond bound is
ample rather than tight.
"""
n_tokens = 4
chunks = _build_stream(
n_tokens, chat=chat, ttft_delay=0.02, itl_delay=0.01, usage_delay=0.02
)
output = _run(request_func, api_url, endpoint_id, chunks, n_tokens)
residual = (output.latency - output.ttft) - sum(output.itl)
assert abs(residual) < 1e-6, (
f"{endpoint_id}: (latency - ttft) - sum(itl) drifted by {residual:.9f}s"
)
@pytest.mark.parametrize(
"endpoint_id,request_func,api_url,chat",
STREAMING_ENDPOINTS,
ids=[e[0] for e in STREAMING_ENDPOINTS],
)
def test_usage_only_stream_is_not_reported_as_success(
endpoint_id: str,
request_func,
api_url: str,
chat: bool,
) -> None:
"""A stream with no token chunk must fail rather than report 0 ms.
Only token chunks move the end of the request, so a degenerate stream
carrying nothing but a usage trailer leaves latency at zero. Reporting that
as a success would feed a zero-duration request into the benchmark
aggregates.
"""
chunks = [(0.0, _usage_chunk(0)), (0.0, b"data: [DONE]\n\n")]
output = _run(request_func, api_url, endpoint_id, chunks, 0, expect_success=False)
assert not output.success
assert output.itl == []
assert "TTFT" in output.error
# ``benchmarks/backend_request_func.py`` is a standalone script rather than a
# package module, so it is loaded by path. It also builds its own ClientSession,
# so the scripted stream is injected by swapping the module's aiohttp reference.
_LEGACY_PATH = Path(__file__).parents[2] / "benchmarks" / "backend_request_func.py"
class _FakeAiohttp:
"""Stands in for the legacy module's aiohttp import."""
def __init__(self, chunks: list[tuple[float, bytes]]) -> None:
self._chunks = chunks
def __getattr__(self, name: str) -> Any:
return getattr(aiohttp, name)
def ClientSession(self, **kwargs: Any) -> _FakeSession: # noqa: N802
del kwargs
return _FakeSession(self._chunks)
def _load_legacy():
spec = importlib.util.spec_from_file_location(
"legacy_backend_request_func", _LEGACY_PATH
)
assert spec is not None and spec.loader is not None
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
LEGACY_ENDPOINTS = [
(
"completions",
"async_request_openai_completions",
"http://test/v1/completions",
False,
),
(
"chat",
"async_request_openai_chat_completions",
"http://test/v1/chat/completions",
True,
),
(
"audio",
"async_request_openai_audio",
"http://test/v1/audio/transcriptions",
True,
),
]
def _run_legacy(
monkeypatch, func_name, api_url, endpoint_id, chunks, n_tokens, clock=None
):
module = _load_legacy()
monkeypatch.setattr(module, "aiohttp", _FakeAiohttp(chunks))
if clock is not None:
monkeypatch.setattr(module, "time", _FakeTime(clock))
request_func = getattr(module, func_name)
return asyncio.run(request_func(_make_input(api_url, n_tokens, endpoint_id)))
@pytest.mark.parametrize(
"endpoint_id,func_name,api_url,chat",
LEGACY_ENDPOINTS,
ids=[e[0] for e in LEGACY_ENDPOINTS],
)
def test_legacy_decode_span_identity_is_exact(
monkeypatch: pytest.MonkeyPatch,
endpoint_id: str,
func_name: str,
api_url: str,
chat: bool,
) -> None:
"""benchmarks/backend_request_func.py must honour the same identity."""
tick = 1.0
n_tokens = 5
chunks = _build_stream(n_tokens, chat=chat)
output = _run_legacy(
monkeypatch,
func_name,
api_url,
endpoint_id,
chunks,
n_tokens,
clock=_ScriptedClock(tick=tick),
)
assert output.success, output.error
assert len(output.itl) == n_tokens - 1
residual = (output.latency - output.ttft) - sum(output.itl)
assert residual == pytest.approx(0.0, abs=tick / 1000), (
f"legacy {endpoint_id}: (latency - ttft) - sum(itl) = {residual!r}"
)
assert output.latency == pytest.approx(n_tokens * tick, abs=tick / 1000)
@pytest.mark.parametrize(
"endpoint_id,func_name,api_url,chat",
LEGACY_ENDPOINTS,
ids=[e[0] for e in LEGACY_ENDPOINTS],
)
def test_legacy_usage_only_stream_is_not_reported_as_success(
monkeypatch: pytest.MonkeyPatch,
endpoint_id: str,
func_name: str,
api_url: str,
chat: bool,
) -> None:
"""A usage-only stream must fail rather than report a zero-duration success."""
chunks = [(0.0, _usage_chunk(0)), (0.0, b"data: [DONE]\n\n")]
output = _run_legacy(monkeypatch, func_name, api_url, endpoint_id, chunks, 0)
assert not output.success
assert output.itl == []
vllm/benchmarks/lib/endpoint_request_func.py
+8 -6
View File
@@ -235,8 +235,7 @@ async def async_request_openai_completions(
# First token
if not first_chunk_received:
first_chunk_received = True
ttft = time.perf_counter() - st
output.ttft = ttft
output.ttft = timestamp - st
# Decoding phase
else:
@@ -415,13 +414,14 @@ async def async_request_openai_chat_completions(
output.itl.append(timestamp - most_recent_timestamp)
generated_text += content or ""
# Only token chunks advance the request end;
# the trailing usage chunk carries no token.
most_recent_timestamp = timestamp
elif usage := data.get("usage"):
output.output_tokens = usage.get("completion_tokens")
if (pt := usage.get("prompt_tokens")) is not None:
output.prompt_len = pt
most_recent_timestamp = timestamp
output.generated_text = generated_text
if first_chunk_received:
output.success = True
@@ -540,13 +540,15 @@ async def async_request_openai_audio(
)
generated_text += content or ""
# Only token chunks advance the request
# end; the trailing usage chunk carries no
# token.
most_recent_timestamp = timestamp
elif usage := data.get("usage"):
output.output_tokens = usage.get(
"completion_tokens"
)
most_recent_timestamp = timestamp
output.generated_text = generated_text
if first_chunk_received:
output.success = True