[Test] Split the serving perf tests by topic into basic_perf/ and route their thresholds through a kit (#40505)

python/sglang/test/kits/perf_bench_kit.py

@@ -0,0 +1,82 @@
+"""Report and bound the numbers a serving benchmark produces.
+
+Bounds are tuned per CI runner, so a local run prints them without asserting.
+"""
+
+from dataclasses import dataclass
+from typing import Callable, Optional, Sequence
+
+from sglang.test.test_utils import is_in_amd_ci, is_in_ci, write_github_step_summary
+
+
+@dataclass
+class Metric:
+    name: str
+    value: float
+    unit: str
+    bound: Optional[float] = None
+    amd_bound: Optional[float] = None
+    # unittest method that enforces `bound`, e.g. "assertLessEqual".
+    assertion: Optional[str] = None
+
+    def line(self) -> str:
+        return f"{self.name}: {self.value:.2f}" + (f" {self.unit}" if self.unit else "")
+
+    def check(self, test_case) -> None:
+        if self.bound is None:
+            return
+        limit = self.bound
+        if is_in_amd_ci() and self.amd_bound is not None:
+            limit = self.amd_bound
+        getattr(test_case, self.assertion)(self.value, limit)
+
+
+def at_least(name, value, bound, *, amd=None, unit="") -> Metric:
+    """A throughput-like number: the run passes when it reaches `bound`."""
+    return Metric(name, value, unit, bound, amd, "assertGreaterEqual")
+
+
+def at_most(name, value, bound, *, amd=None, unit="") -> Metric:
+    """A latency-like number: the run passes when it stays under `bound`."""
+    return Metric(name, value, unit, bound, amd, "assertLessEqual")
+
+
+def reported(name, value, *, unit="") -> Metric:
+    """A number worth printing that no threshold is attached to."""
+    return Metric(name, value, unit)
+
+
+def check_perf(test_case, *metrics: Metric, suffix: str = "") -> None:
+    """Report every metric under the caller's test name, then enforce in CI."""
+    label = test_case._testMethodName + suffix
+    report = f"### {label}\n" + "".join(m.line() + "\n" for m in metrics)
+    print(report, end="")
+    if not is_in_ci():
+        return
+    write_github_step_summary(report)
+    for m in metrics:
+        m.check(test_case)
+
+
+def check_batch_scaling(
+    test_case,
+    run_all: Callable[[Sequence[int]], Sequence[dict]],
+    bounds: Sequence[tuple],
+) -> None:
+    """Bound the latency at each batch size of one sweep.
+
+    `run_all` takes every size at once so they share one server. Each `bounds`
+    entry is `(batch_size, avg_ms, p95_ms, amd_avg_ms, amd_p95_ms)`.
+    """
+    results = run_all([b[0] for b in bounds])
+    for (batch_size, avg_ms, p95_ms, amd_avg_ms, amd_p95_ms), res in zip(
+        bounds, results
+    ):
+        test_case.assertEqual(res["successful_requests"], res["total_requests"])
+        check_perf(
+            test_case,
+            at_most("avg_latency_ms", res["avg_latency_ms"], avg_ms, amd=amd_avg_ms),
+            at_most("p95_latency_ms", res["p95_latency_ms"], p95_ms, amd=amd_p95_ms),
+            reported("throughput", res["throughput"], unit="req/s"),
+            suffix=f"_size_{batch_size}",
+        )

python/sglang/test/kits/vlm_perf_kit.py

@@ -0,0 +1,94 @@
+"""One VLM serving benchmark, run against a named attention backend.
+
+Pinned rather than inherited from `get_default_attn_backend`, so the file name
+keeps naming the right kernel after the default moves.
+"""
+
+import os
+from typing import Optional
+
+from sglang.test.kits.perf_bench_kit import at_least, at_most, check_perf, reported
+from sglang.test.test_utils import (
+    DEFAULT_SMALL_VLM_MODEL_NAME_FOR_TEST,
+    DEFAULT_URL_FOR_TEST,
+    auto_config_device,
+    get_benchmark_args,
+    run_bench_serving_multi,
+)
+
+
+def _local_tokenizer_path():
+    # The HF Hub API call can stall for minutes in CI; prefer a local snapshot.
+    try:
+        from sglang.srt.utils import find_local_repo_dir
+
+        local_dir = find_local_repo_dir(
+            DEFAULT_SMALL_VLM_MODEL_NAME_FOR_TEST, revision=None
+        )
+        if local_dir and os.path.isdir(local_dir):
+            return local_dir
+    except Exception:
+        pass
+    return None
+
+
+def check_vlm_serving_perf(
+    test_case,
+    attention_backend: str,
+    *,
+    e2e_ms: float,
+    ttft_ms: float,
+    itl_ms: float,
+    output_throughput: Optional[float] = None,
+):
+    """Offline then online against one server; bound both phases.
+
+    `output_throughput` unset means the offline number is reported, not bounded.
+    """
+    common = dict(
+        base_url=DEFAULT_URL_FOR_TEST,
+        dataset_name="mmmu",
+        dataset_path="",
+        tokenizer=_local_tokenizer_path(),
+        random_input_len=4096,
+        random_output_len=2048,
+        sharegpt_context_len=None,
+        disable_stream=False,
+        disable_ignore_eos=False,
+        seed=0,
+        device=auto_config_device(),
+        lora_name=None,
+    )
+    offline = get_benchmark_args(num_prompts=200, request_rate=float("inf"), **common)
+    # 50 is enough for a stable median against these loose ceilings.
+    online = get_benchmark_args(num_prompts=50, request_rate=1, **common)
+
+    (_, res_offline), (_, res_online) = run_bench_serving_multi(
+        DEFAULT_SMALL_VLM_MODEL_NAME_FOR_TEST,
+        DEFAULT_URL_FOR_TEST,
+        other_server_args=[
+            "--mem-fraction-static",
+            "0.7",
+            "--attention-backend",
+            attention_backend,
+        ],
+        benchmark_args=[offline, online],
+    )
+
+    throughput = res_offline["output_throughput"]
+    check_perf(
+        test_case,
+        (
+            at_least("output_throughput", throughput, output_throughput, unit="token/s")
+            if output_throughput is not None
+            else reported("output_throughput", throughput, unit="token/s")
+        ),
+        at_most(
+            "median_e2e_latency_ms",
+            res_online["median_e2e_latency_ms"],
+            e2e_ms,
+            unit="ms",
+        ),
+        at_most("median_ttft_ms", res_online["median_ttft_ms"], ttft_ms, unit="ms"),
+        at_most("median_itl_ms", res_online["median_itl_ms"], itl_ms, unit="ms"),
+    )

python/sglang/test/test_utils.py

@@ -61,7 +61,6 @@ DEFAULT_SMALL_MOE_MODEL_NAME_FOR_TEST_BASE = "Qwen/Qwen1.5-MoE-A2.7B"
 DEFAULT_SMALL_MOE_MODEL_NAME_FOR_TEST_CHAT = "Qwen/Qwen1.5-MoE-A2.7B-Chat"
 
 # MLA test models
-DEFAULT_SMALL_EMBEDDING_MODEL_NAME_FOR_TEST = "Alibaba-NLP/gte-Qwen2-1.5B-instruct"
 DEFAULT_SMALL_CROSS_ENCODER_MODEL_NAME_FOR_TEST = "cross-encoder/ms-marco-MiniLM-L6-v2"
 DEFAULT_MLA_MODEL_NAME_FOR_TEST = "deepseek-ai/DeepSeek-Coder-V2-Lite-Instruct"
 DEFAULT_MLA_FP8_MODEL_NAME_FOR_TEST = "neuralmagic/DeepSeek-Coder-V2-Lite-Instruct-FP8"
@@ -1167,6 +1166,29 @@ def run_score_benchmark(
     device="auto",
 ):
     """Score API benchmark function compatible with run_bench_serving pattern"""
+    return run_score_benchmark_multi(
+        model,
+        [batch_size],
+        num_requests=num_requests,
+        other_server_args=other_server_args,
+        need_warmup=need_warmup,
+        device=device,
+    )[0]
+
+
+def run_score_benchmark_multi(
+    model,
+    batch_sizes,
+    num_requests=100,
+    other_server_args=None,
+    need_warmup=False,
+    device="auto",
+):
+    """One server, one benchmark per batch size.
+
+    Batch size is a property of the request, not of the server, so the launch
+    is shared rather than repeated per size.
+    """
     if other_server_args is None:
         other_server_args = []
 
@@ -1182,7 +1204,7 @@ def run_score_benchmark(
         other_args=other_server_args,
     )
 
-    async def _run_benchmark():
+    async def _run_benchmark(batch_size, warmup):
         # Load tokenizer for generating test data
         from sglang.srt.utils.hf_transformers_utils import get_tokenizer
 
@@ -1205,7 +1227,7 @@ def run_score_benchmark(
                 )
             return text
 
-        if need_warmup:
+        if warmup:
             warmup_data = {
                 "query": generate_text_with_token_count(score_query_tokens),
                 "items": [
@@ -1253,12 +1275,16 @@ def run_score_benchmark(
         )
 
     try:
-        res = asyncio.run(_run_benchmark())
+        results = [
+            asyncio.run(_run_benchmark(bs, need_warmup and i == 0))
+            for i, bs in enumerate(batch_sizes)
+        ]
     finally:
         kill_process_tree(process.pid)
 
-    assert res["completed"] == res["successful_requests"]
-    return res
+    for res in results:
+        assert res["completed"] == res["successful_requests"]
+    return results
 
 
 def run_embeddings_benchmark(
@@ -1271,6 +1297,27 @@ def run_embeddings_benchmark(
     device="auto",
 ):
     """Embeddings API benchmark function compatible with run_bench_serving pattern"""
+    return run_embeddings_benchmark_multi(
+        model,
+        [batch_size],
+        num_requests=num_requests,
+        input_tokens=input_tokens,
+        other_server_args=other_server_args,
+        need_warmup=need_warmup,
+        device=device,
+    )[0]
+
+
+def run_embeddings_benchmark_multi(
+    model,
+    batch_sizes,
+    num_requests=100,
+    input_tokens=500,
+    other_server_args=None,
+    need_warmup=False,
+    device="auto",
+):
+    """One server, one benchmark per batch size. See run_score_benchmark_multi."""
     if other_server_args is None:
         other_server_args = []
 
@@ -1289,7 +1336,7 @@ def run_embeddings_benchmark(
         other_args=server_args,
     )
 
-    async def _run_benchmark():
+    async def _run_benchmark(batch_size, warmup):
 
         def generate_text_with_token_count(num_tokens):
             """Generate text with precise token count using special tokens."""
@@ -1300,7 +1347,7 @@ def run_embeddings_benchmark(
         # Generate input text
         input_text = generate_text_with_token_count(input_tokens)
 
-        if need_warmup:
+        if warmup:
             warmup_data = {
                 "input": input_text,
                 "model": model,
@@ -1340,12 +1387,16 @@ def run_embeddings_benchmark(
         )
 
     try:
-        res = asyncio.run(_run_benchmark())
+        results = [
+            asyncio.run(_run_benchmark(bs, need_warmup and i == 0))
+            for i, bs in enumerate(batch_sizes)
+        ]
     finally:
         kill_process_tree(process.pid)
 
-    assert res["completed"] == res["successful_requests"]
-    return res
+    for res in results:
+        assert res["completed"] == res["successful_requests"]
+    return results
 
 
 def run_bench_serving_multi(