From 0ce2c1312ad8f0c2f2717fd887d4172f31e9fb1b Mon Sep 17 00:00:00 2001 From: kevincheng2 Date: Mon, 24 Aug 2026 16:27:35 +0800 Subject: [PATCH] [Benchmark] measure real decode speed when the server batches stream chunks A server-side reasoning / tool-call parser only emits an SSE chunk once it has user-visible text, so decode steps whose text is still buffered are invisible to the client while their tokens are already counted in usage.completion_tokens. Taking TTFT from the first delta-carrying chunk then charges those tokens to TTFT and shrinks the decode window, which inflates the reported decode speed. - count a chunk whose delta is empty but whose usage.completion_tokens grew as a real token arrival, so TTFT/ITL see every generated token - credit the tokens that only show up in the trailing usage chunk (e.g. tool-call closing markers) to ITL, so the numerator and the denominator cover the same tokens - derive per-request decode speed from len(itl)/sum(itl); the previous (output_tokens - 1) form assumes the first chunk carries exactly one token, which is false under speculative decoding - drop the "TPOT < 1ms is unreliable" filter and its duplicated report block: with the window measured correctly there is nothing to filter --- benchmarks/backend_request_func.py | 32 ++++++++--- benchmarks/benchmark_serving.py | 88 +++++++----------------------- 2 files changed, 45 insertions(+), 75 deletions(-) diff --git a/benchmarks/backend_request_func.py b/benchmarks/backend_request_func.py index d1ef342e827..be30eeb2bcd 100644 --- a/benchmarks/backend_request_func.py +++ b/benchmarks/backend_request_func.py @@ -352,6 +352,9 @@ async def async_request_eb_openai_chat_completions( payload["stream_options"] = { "include_usage": True, "continuous_usage_stats": True, + # sglang 扩展：解析器攒批时补发只带 usage 的空 delta 包，否则这些 + # token 会被算进 TTFT，解码速度被高估。不支持的服务端会忽略该字段。 + "step_usage_chunks": "first_token", } if request_func_input.json_data: json_data = request_func_input.json_data @@ -496,7 +499,18 @@ async def async_request_eb_openai_chat_completions( reason_content = choices[0]["delta"].get("reasoning_content") tool_calls = choices[0]["delta"].get("tool_calls") completion_token_ids = choices[0]["delta"].get("completion_token_ids", []) + # 服务端已生成的累计 token 数：优先用 usage(continuous_usage_stats)， + # 否则退化为按 token_ids 长度推断。 + cur_completion_tokens = (data.get("usage") or {}).get("completion_tokens") + if cur_completion_tokens is None and completion_token_ids: + cur_completion_tokens = len(output.output_ids) + len(completion_token_ids) has_token_chunk = bool(content or reason_content or tool_calls or completion_token_ids) + # reasoning/tool-call parser 攒批时 delta 为空，服务端只发 usage 心跳包 + # （sglang 侧开关 stream_options.step_usage_chunks）。token 数增长 + # 就是一次真实的 token 到达，必须计入 TTFT/ITL，否则被攒批吞掉的 token + # 会被算进 TTFT，解码区间被压缩、解码速度虚高。 + if not has_token_chunk and cur_completion_tokens is not None: + has_token_chunk = cur_completion_tokens > last_output_len if tool_calls: for tc in tool_calls: idx = tc.get("index", 0) @@ -536,10 +550,6 @@ async def async_request_eb_openai_chat_completions( output.prompt_len = 0 # 首 token 也要更新 last_output_len，用于后续 burst 摊分 - cur_completion_tokens = (data.get("usage") or {}).get("completion_tokens") - if cur_completion_tokens is None and completion_token_ids: - # 没有 usage 时退化为按 token_ids 长度推断 - cur_completion_tokens = len(output.output_ids) + len(completion_token_ids) if cur_completion_tokens is not None: last_output_len = cur_completion_tokens else: @@ -550,10 +560,6 @@ async def async_request_eb_openai_chat_completions( # buffer burst 修正：如果服务端把多个 token 合并到同一个流式 chunk 里， # 直接把整段间隔记成单个 ITL 会高估解码间隔。这里参考 sglang 官方 # bench_serving 的做法：用真实 token 增量摊分该 chunk 的等待时间。 - cur_completion_tokens = (data.get("usage") or {}).get("completion_tokens") - if cur_completion_tokens is None and completion_token_ids: - cur_completion_tokens = len(output.output_ids) + len(completion_token_ids) - chunk_gap = timestamp - most_recent_timestamp if cur_completion_tokens is not None: num_new_tokens = cur_completion_tokens - last_output_len @@ -596,6 +602,16 @@ async def async_request_eb_openai_chat_completions( prompt_tokens_details = usage.get("prompt_tokens_details") or {} if output.prompt_len == 0: output.prompt_len = prompt_tokens_details.get("cached_tokens", 0) + # 收尾 usage 包：把最后一批"生成了但没随 delta 下发"的 token 补进 ITL， + # 否则分子(总 token)覆盖不到分母(ITL 区间)，解码速度会被高估。 + tail_tokens = output.output_tokens - last_output_len + if tail_tokens > 0 and ttft > 0.0: + tail_gap = timestamp - most_recent_timestamp + if tail_gap > 0: + output.itl.extend([tail_gap / tail_tokens] * tail_tokens) + last_output_len = output.output_tokens + most_recent_timestamp = timestamp + token_timestamps.append(wall_timestamp) last_chunk_timestamp = timestamp diff --git a/benchmarks/benchmark_serving.py b/benchmarks/benchmark_serving.py index 33158f8c29f..530b73a12be 100644 --- a/benchmarks/benchmark_serving.py +++ b/benchmarks/benchmark_serving.py @@ -65,12 +65,8 @@ class BenchmarkMetrics: input_throughput: float output_throughput: float total_token_throughput: float - # 解码速度(过滤 TPOT<1ms 的请求后) - s_decode_filtered_mean: float # tok/s - s_decode_filtered_median: float # tok/s - n_decode_total: int # 参与统计的总请求数 - n_decode_filtered: int # 被过滤的请求数(TPOT<1ms) - n_decode_reliable: int # 可信请求数(TPOT>=1ms) + # 解码速度：首 token 之后每秒真实到达的 token 数，全量统计不过滤 + n_decode_total: int # 参与统计的请求数 mean_s_decode: float median_s_decode: float std_s_decode: float @@ -259,10 +255,19 @@ def calculate_metrics( else: # sglang等无arrival_time场景fallback if outputs[i].output_tokens > 1: - decode_time = outputs[i].latency - outputs[i].ttft + # ITL 条目数 = 首个可见 chunk 之后真实到达的 token 数（按服务端 usage 增量摊分）， + # sum(itl) = 对应的完整解码区间。用 len/sum 而不是 (output_tokens-1)/sum： + # 首个 chunk 在 MTP 下可能一次带 1~3 个 token，减 1 会有系统偏差。 + # 也不用 latency-ttft：解析器攒批时被吞掉的 token 会被计进 TTFT，解码速度虚高。 + if outputs[i].itl: + decode_time = sum(outputs[i].itl) + decode_tokens = len(outputs[i].itl) + else: + decode_time = outputs[i].latency - outputs[i].ttft + decode_tokens = outputs[i].output_tokens - 1 if decode_time > 0: - s_decodes.append((outputs[i].output_tokens - 1) / decode_time) + s_decodes.append(decode_tokens / decode_time) else: s_decodes.append(0) else: @@ -309,25 +314,10 @@ def calculate_metrics( total_input += prefill_only_input print(f"零输出(prefill-only)请求: {prefill_only_reqs} 条, " f"补入 ITPS 的输入 token: {prefill_only_input}") - # === 解码速度过滤：TPOT < 1ms 的请求视为不可信(引擎批量flush伪象) === - MIN_TPOT_S = 0.001 # 1ms - reliable_s_decodes = [] - n_decode_total = 0 - n_decode_filtered = 0 - for o in outputs: - if not o.success or o.output_tokens <= 1: - continue - decode_time = sum(o.itl) if o.itl else 0 - if decode_time <= 0: - continue - n_decode_total += 1 - tokens = o.output_tokens - 1 - tpot = decode_time / tokens - if tpot >= MIN_TPOT_S: - reliable_s_decodes.append(tokens / decode_time) - else: - n_decode_filtered += 1 - n_decode_reliable = len(reliable_s_decodes) + # 解码速度只有一份口径：上面循环里按 len(itl)/sum(itl) 逐请求算好的 s_decodes。 + # ITL 由服务端 usage 的 token 增量摊分得到（含被解析器攒批吞掉的 token 与收尾 token）， + # 所以 sum(itl) 就是真实解码区间，不需要"TPOT<1ms 视为不可信"这类兜底过滤。 + n_decode_total = len([s for s in s_decodes if s > 0]) metrics = BenchmarkMetrics( completed=completed, @@ -338,11 +328,7 @@ def calculate_metrics( input_throughput=total_input / dur_s, output_throughput=sum(actual_output_lens) / dur_s, total_token_throughput=(total_input + sum(actual_output_lens)) / dur_s, - s_decode_filtered_mean=float(np.mean(reliable_s_decodes)) if reliable_s_decodes else 0.0, - s_decode_filtered_median=float(np.median(reliable_s_decodes)) if reliable_s_decodes else 0.0, n_decode_total=n_decode_total, - n_decode_filtered=n_decode_filtered, - n_decode_reliable=n_decode_reliable, mean_s_decode=np.mean(s_decodes or 0) * 1, # ttfts is empty if streaming is not supported by backend std_s_decode=np.std(s_decodes or 0) * 1, median_s_decode=np.median(s_decodes or 0) * 1, @@ -808,11 +794,7 @@ async def limited_request_func_per_ip(req_input, semaphore, pbar): "reasoning_contents": [output.reasoning_content for output in outputs], "errors": [output.error for output in outputs], "metrics": [output.metrics for output in outputs], - "s_decode_filtered_mean": metrics.s_decode_filtered_mean, - "s_decode_filtered_median": metrics.s_decode_filtered_median, "n_decode_total": metrics.n_decode_total, - "n_decode_filtered": metrics.n_decode_filtered, - "n_decode_reliable": metrics.n_decode_reliable, } def process_one_metric( @@ -961,22 +943,8 @@ def process_one_length( print("{:<40} {:<10.2f}".format(f"P{p_word} {metric_name}:", value)) result[f"p{p_word}_{metric_attribute_name}"] = value - print("{s:{c}^{n}}".format(s="解码速度 (过滤TPOT<1ms)", n=50, c="-")) - _n_total = max(metrics.n_decode_total, 1) - print("{:<40} {:<10d}".format("Total requests:", metrics.n_decode_total)) - print( - "{:<40} {:<10d} ({:.2f}%)".format( - "Filtered (TPOT<1ms):", metrics.n_decode_filtered, 100 * metrics.n_decode_filtered / _n_total - ) - ) - print( - "{:<40} {:<10d} ({:.2f}%)".format( - "Reliable (TPOT>=1ms):", metrics.n_decode_reliable, 100 * metrics.n_decode_reliable / _n_total - ) - ) - print("{:<40} {:<10.2f}".format("Mean Decode (tok/s):", metrics.s_decode_filtered_mean)) - print("{:<40} {:<10.2f}".format("Median Decode (tok/s):", metrics.s_decode_filtered_median)) - process_one_length("s_decode", "Decode", "解码速度(tok/s)") + print("{:<40} {:<10d}".format("Requests in decode stats:", metrics.n_decode_total)) + process_one_length("s_decode", "Decode", "解码速度(tok/s, 首token之后)") process_one_metric("ttft", "TTFT", "Time to First Token") process_one_metric("s_ttft", "S_TTFT", "Infer Time to First Token") process_one_metric("res_ttft", "Response TTFT", "包含思考首token耗时") @@ -1164,22 +1132,8 @@ def process_one_length( print("{:<40} {:<10.2f}".format(f"P{p_word} {metric_name}:", value)) result[f"p{p_word}_{metric_attribute_name}"] = value - print("{s:{c}^{n}}".format(s="解码速度 (过滤TPOT<1ms)", n=50, c="-")) - _n_total = max(metrics.n_decode_total, 1) - print("{:<40} {:<10d}".format("Total requests:", metrics.n_decode_total)) - print( - "{:<40} {:<10d} ({:.2f}%)".format( - "Filtered (TPOT<1ms):", metrics.n_decode_filtered, 100 * metrics.n_decode_filtered / _n_total - ) - ) - print( - "{:<40} {:<10d} ({:.2f}%)".format( - "Reliable (TPOT>=1ms):", metrics.n_decode_reliable, 100 * metrics.n_decode_reliable / _n_total - ) - ) - print("{:<40} {:<10.2f}".format("Mean Decode (tok/s):", metrics.s_decode_filtered_mean)) - print("{:<40} {:<10.2f}".format("Median Decode (tok/s):", metrics.s_decode_filtered_median)) - process_one_length("s_decode", "Decode", "解码速度(tok/s)") + print("{:<40} {:<10d}".format("Requests in decode stats:", metrics.n_decode_total)) + process_one_length("s_decode", "Decode", "解码速度(tok/s, 首token之后)") process_one_metric("ttft", "TTFT", "Time to First Token") process_one_metric("s_ttft", "S_TTFT", "Infer Time to First Token") process_one_metric("tpot", "TPOT", "Time per Output Token (excl. 1st token)")