[Benchmark] measure real decode speed when the server batches stream chunks (PaddlePaddle#8113)

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

Author: kevincheng2

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
 

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)")