Add peak output tokens per second in bench_serving (#14165)

Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>

python/sglang/bench_serving.py

@@ -12,12 +12,14 @@ python3 -m sglang.bench_serving --backend sglang --dataset-name random --num-pro
 
 import argparse
 import asyncio
+import importlib.util
 import io
 import json
 import os
 import pickle
 import random
 import resource
+import shutil
 import sys
 import time
 import traceback
@@ -47,6 +49,10 @@ from transformers import (
 
 ASSISTANT_SUFFIX = "Assistant:"
 
+TERM_PLOTLIB_AVAILABLE = (importlib.util.find_spec("termplotlib") is not None) and (
+    shutil.which("gnuplot") is not None
+)
+
 global args
 
 
@@ -93,6 +99,7 @@ class RequestFuncOutput:
     prompt_len: int = 0
     error: str = ""
     output_len: int = 0
+    start_time: float = 0.0
 
     @staticmethod
     def init_new(request_func_input: RequestFuncInput):
@@ -230,6 +237,7 @@ async def async_request_openai_completions(
         output_len = request_func_input.output_len
         ttft = 0.0
         st = time.perf_counter()
+        output.start_time = st
         most_recent_timestamp = st
         try:
             async with session.post(
@@ -354,6 +362,7 @@ async def async_request_openai_chat_completions(
         output_len = request_func_input.output_len
         ttft = 0.0
         st = time.perf_counter()
+        output.start_time = st
         most_recent_timestamp = st
         try:
             async with session.post(
@@ -543,6 +552,7 @@ async def async_request_sglang_generate(
         output_len = request_func_input.output_len
         ttft = 0.0
         st = time.perf_counter()
+        output.start_time = st
         most_recent_timestamp = st
         last_output_len = 0
         try:
@@ -869,6 +879,8 @@ class BenchmarkMetrics:
     std_e2e_latency_ms: float
     p99_e2e_latency_ms: float
     concurrency: float
+    max_output_tokens_per_s: float = 0.0
+    max_concurrent_requests: int = 0
 
 
 SHAREGPT_URL = "https://huggingface.co/datasets/anon8231489123/ShareGPT_Vicuna_unfiltered/resolve/main/ShareGPT_V3_unfiltered_cleaned_split.json"
@@ -1666,6 +1678,7 @@ def calculate_metrics(
     tokenizer: PreTrainedTokenizerBase,
     backend: str,
     accept_length: Optional[float] = None,
+    plot_throughput: bool = False,
 ) -> Tuple[BenchmarkMetrics, List[int]]:
     output_lens: List[int] = []
     retokenized_output_lens: List[int] = []
@@ -1725,6 +1738,70 @@ def calculate_metrics(
             stacklevel=2,
         )
 
+    max_output_tokens_per_s = 0.0
+    max_concurrent_requests = 0
+
+    successful_outputs = [output for output in outputs if output.success]
+    if successful_outputs:
+        min_start_time = min(output.start_time for output in successful_outputs)
+        max_end_time = max(
+            output.start_time + output.latency for output in successful_outputs
+        )
+
+        duration_seconds = int(np.ceil(max_end_time - min_start_time)) + 1
+        tokens_per_second = np.zeros(duration_seconds)
+        concurrent_requests_per_second = np.zeros(duration_seconds)
+
+        for output in outputs:
+            if not output.success:
+                continue
+
+            token_times = [output.start_time + output.ttft]
+            current_time = token_times[0]
+            for itl_value in output.itl:
+                current_time += itl_value
+                token_times.append(current_time)
+
+            for token_time in token_times:
+                second_bucket = int(token_time - min_start_time)
+                if 0 <= second_bucket < duration_seconds:
+                    tokens_per_second[second_bucket] += 1
+
+            request_start_second = int(output.start_time - min_start_time)
+            request_end_second = int(
+                (output.start_time + output.latency) - min_start_time
+            )
+
+            for second in range(
+                request_start_second, min(request_end_second + 1, duration_seconds)
+            ):
+                concurrent_requests_per_second[second] += 1
+
+        if len(tokens_per_second) > 0:
+            max_output_tokens_per_s = float(np.max(tokens_per_second))
+            max_concurrent_requests = int(np.max(concurrent_requests_per_second))
+
+        if plot_throughput:
+            if TERM_PLOTLIB_AVAILABLE:
+                import termplotlib as tpl
+
+                fig = tpl.figure()
+                fig.plot(
+                    np.arange(len(tokens_per_second)),
+                    tokens_per_second,
+                    title="Output tokens per second",
+                    xlabel="Time (s)",
+                )
+                fig.plot(
+                    np.arange(len(concurrent_requests_per_second)),
+                    concurrent_requests_per_second,
+                    title="Concurrent requests per second",
+                    xlabel="Time (s)",
+                )
+                fig.show()
+            else:
+                print("tip: install termplotlib and gnuplot to plot the metrics")
+
     itls = retokenized_itls if use_retokenized_itl else itls
     metrics = BenchmarkMetrics(
         completed=completed,
@@ -1760,6 +1837,8 @@ def calculate_metrics(
         std_e2e_latency_ms=np.std(e2e_latencies) * 1000,
         p99_e2e_latency_ms=np.percentile(e2e_latencies, 99) * 1000,
         concurrency=np.sum(e2e_latencies) / dur_s,
+        max_output_tokens_per_s=max_output_tokens_per_s,
+        max_concurrent_requests=max_concurrent_requests,
     )
 
     return metrics, output_lens
@@ -2012,6 +2091,7 @@ async def benchmark(
         tokenizer=tokenizer,
         backend=backend,
         accept_length=accept_length,
+        plot_throughput=args.plot_throughput,
     )
 
     print("\n{s:{c}^{n}}".format(s=" Serving Benchmark Result ", n=50, c="="))
@@ -2055,6 +2135,16 @@ async def benchmark(
             "Output token throughput (tok/s):", metrics.output_throughput
         )
     )
+    print(
+        "{:<40} {:<10.2f}".format(
+            "Peak output token throughput (tok/s):", metrics.max_output_tokens_per_s
+        )
+    )
+    print(
+        "{:<40} {:<10}".format(
+            "Peak concurrent requests:", metrics.max_concurrent_requests
+        )
+    )
     print(
         "{:<40} {:<10.2f}".format(
             "Total token throughput (tok/s):", metrics.total_throughput
@@ -2142,6 +2232,8 @@ async def benchmark(
             "p99_itl_ms": metrics.p99_itl_ms,
             "concurrency": metrics.concurrency,
             "accept_length": accept_length,
+            "max_output_tokens_per_s": metrics.max_output_tokens_per_s,
+            "max_concurrent_requests": metrics.max_concurrent_requests,
         }
     else:
         print(f"Error running benchmark for request rate: {request_rate}")
@@ -2218,6 +2310,9 @@ def run_benchmark(args_: argparse.Namespace):
     if not hasattr(args, "tokenize_prompt"):
         args.tokenize_prompt = False
 
+    if not hasattr(args, "plot_throughput"):
+        args.plot_throughput = False
+
     if not hasattr(args, "use_trace_timestamps"):
         args.use_trace_timestamps = False
     if not hasattr(args, "mooncake_slowdown_factor"):
@@ -2609,6 +2704,11 @@ if __name__ == "__main__":
         help="Use Torch Profiler. The endpoint must be launched with "
         "SGLANG_TORCH_PROFILER_DIR to enable profiler.",
     )
+    parser.add_argument(
+        "--plot-throughput",
+        action="store_true",
+        help="Plot throughput and concurrent requests over time. Requires termplotlib and gnuplot.",
+    )
     # TODO unify all these
     parser.add_argument(
         "--profile-activities",