minke.yu/sglang
1
0
Fork 0
Code Issues Pull Requests Actions 11 Packages Projects Releases Wiki Activity

[Test] Drop cause-less disabled tests, fix XPU lane, demote quality gates off base-c (#40288)

Browse Source
This commit is contained in:
Liangsheng Yin
2026-09-20 14:36:02 -07:00
committed by GitHub
parent d229952e25
commit f6483e479f
14 changed files with 24 additions and 1286 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/pr-test-extra.yml
+16
View File
@@ -248,6 +248,21 @@ jobs:
rust_ext_artifact: ${{ needs.rust-ext-build.outputs.artifact_name }}
secrets: inherit
extra-b-test-8-gpu-b300:
needs: [check-changes, call-gate, sgl-kernel-build-wheels, rust-ext-build]
if: ${{ !failure() && !cancelled() && needs.check-changes.result == 'success' && (needs.call-gate.result == 'success' || needs.call-gate.result == 'skipped') }}
uses: ./.github/workflows/_pr-test-stage.yml
with:
self_name: extra-b-test-8-gpu-b300
runner_config: 8-gpu-b300
check_changes: ${{ toJson(needs.check-changes.outputs) }}
caller_inputs: ${{ toJson(inputs) }}
partitions: ${{ needs.check-changes.outputs.partitions }}
run_timeout_minutes: '60'
timeout_per_file: '3600'
rust_ext_artifact: ${{ needs.rust-ext-build.outputs.artifact_name }}
secrets: inherit
# =============================================== aggregator ====================================================
# Mirrors pr-test.yml's `pr-test-finish` so notify-pr-states below only
# depends on one job rather than re-listing every stage. Fails if any
@@ -266,6 +281,7 @@ jobs:
extra-b-test-4-gpu-h100,
extra-b-test-4-gpu-b200,
extra-b-test-8-gpu-h200,
extra-b-test-8-gpu-b300,
]
if: always()
runs-on: ubuntu-latest
test/registered/disaggregation/test_disaggregation_xpu.py
+2 -7
View File
@@ -18,18 +18,13 @@ import unittest
import requests
import torch
from sglang.test.ci.ci_register import register_cuda_ci
from sglang.test.ci.ci_register import register_xpu_ci
from sglang.test.server_fixtures.disaggregation_fixture import (
PDDisaggregationServerBase,
)
from sglang.test.test_utils import DEFAULT_SMALL_MODEL_NAME_FOR_TEST_QWEN
register_cuda_ci(
est_time=300,
stage="base-a",
runner_config="1-gpu-small",
disabled="Intel XPU only — not available in standard CUDA CI",
)
register_xpu_ci(est_time=300, suite="stage-b-test-1-gpu-xpu")
_XPU_AVAILABLE = torch.xpu.is_available()
test/registered/e2e/models/test_dsa_glm52_dp_mtp.py
+1 -1
View File
@@ -10,7 +10,7 @@ from sglang.test.server_fixtures.dsa_mtp_fixture import (
register_cuda_ci(
est_time=400,
stage="base-c",
stage="nightly",
runner_config="8-gpu-h200",
)
test/registered/e2e/models/test_dsa_glm52_tp_mtp.py
+1 -1
View File
@@ -10,7 +10,7 @@ from sglang.test.server_fixtures.dsa_mtp_fixture import (
register_cuda_ci(
est_time=400,
stage="base-c",
stage="nightly",
runner_config="8-gpu-h200",
)
test/registered/e2e/models/test_gpt_oss_4gpu_mxfp4.py
-1
View File
@@ -3,7 +3,6 @@ import unittest
from sglang.test.ci.ci_register import register_cuda_ci
from sglang.test.gpt_oss_common import BaseTestGptOss
register_cuda_ci(est_time=128, stage="base-c", runner_config="4-gpu-h100")
register_cuda_ci(est_time=119, stage="base-c", runner_config="4-gpu-b200")
test/registered/e2e/models/test_kimi_k3_b300_low_latency.py
+1 -1
View File
@@ -17,7 +17,7 @@ from sglang.test.test_utils import (
popen_launch_server,
)
register_cuda_ci(est_time=1472, stage="base-c", runner_config="8-gpu-b300")
register_cuda_ci(est_time=1472, stage="extra-b", runner_config="8-gpu-b300")
MODEL_PATH = "moonshotai/Kimi-K3"
DSPARK_DRAFT_MODEL = "RadixArk/Kimi-K3-DSpark"
test/registered/e2e/models/test_mimo_v2.py
+1 -1
View File
@@ -5,7 +5,7 @@ from sglang.test.ci.ci_register import register_cuda_ci
from sglang.test.kits.eval_accuracy_kit import GSM8KMixin
from sglang.test.server_fixtures.mmmu_fixture import MMMUServerBase
register_cuda_ci(est_time=317, stage="base-c", runner_config="8-gpu-h200")
register_cuda_ci(est_time=317, stage="extra-b", runner_config="8-gpu-h200")
MIMO_V2_MODEL = "XiaomiMiMo/MiMo-V2.5"
MIMO_V2_OTHER_ARGS = [
test/registered/layers/test_fla_layernorm_guard.py
-396
View File
@@ -1,396 +0,0 @@
from __future__ import annotations
import socket
import sys
from dataclasses import dataclass
import pytest
import torch
import torch.nn.functional as F
from sglang.kernels.ops.attention.fla.layernorm_gated import (
_layer_norm_fwd as layer_norm_fwd,
)
from sglang.kernels.ops.attention.fla.layernorm_gated import (
layernorm_fn,
rms_norm_ref,
)
from sglang.test.ci.ci_register import register_cuda_ci
register_cuda_ci(
est_time=60,
stage="base-b",
runner_config="2-gpu-large",
disabled="Temporarily disabled",
)
# Optional dependency in sglang repo; skip collection cleanly if absent.
custom_all_reduce_utils = pytest.importorskip(
"sglang.srt.distributed.device_communicators.custom_all_reduce_utils"
)
parallel_state = pytest.importorskip("sglang.srt.distributed.parallel_state")
update_environment_variables = custom_all_reduce_utils.update_environment_variables
init_distributed_environment = parallel_state.init_distributed_environment
initialize_model_parallel = parallel_state.initialize_model_parallel
NUM_GPUS = 2
def _find_free_port() -> int:
# Avoid hard-coded port collisions when pytest runs tests in parallel.
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.bind(("localhost", 0))
s.listen(1)
return int(s.getsockname()[1])
def _skip_if_no_cuda_or_not_enough_gpus(required_gpus: int = NUM_GPUS) -> None:
if not torch.cuda.is_available():
pytest.skip("CUDA device not available")
if torch.cuda.device_count() < required_gpus:
pytest.skip(f"Need >= {required_gpus} GPUs, got {torch.cuda.device_count()}")
def _skip_if_dtype_unsupported(dtype: torch.dtype) -> None:
if dtype is torch.bfloat16 and not torch.cuda.is_bf16_supported():
pytest.skip("bfloat16 not supported on this CUDA device")
def _setup_sglang_distributed(
local_rank: int,
world_size: int,
master_port: int,
dtype: torch.dtype,
) -> torch.device:
# Match sglang test style: set per-rank CUDA device + default dtype/device.
torch.manual_seed(0)
torch.cuda.manual_seed_all(0)
device = torch.device(f"cuda:{local_rank}")
torch.cuda.set_device(device)
if hasattr(torch, "set_default_device"):
torch.set_default_device(device)
if hasattr(torch, "set_default_dtype"):
torch.set_default_dtype(dtype)
update_environment_variables(
{
"RANK": str(local_rank),
"LOCAL_RANK": str(local_rank),
"WORLD_SIZE": str(world_size),
"MASTER_ADDR": "localhost",
"MASTER_PORT": str(master_port),
}
)
init_distributed_environment(
world_size=world_size, rank=local_rank, local_rank=local_rank
)
initialize_model_parallel(tensor_model_parallel_size=world_size)
return device
def layer_norm_ref(
x: torch.Tensor,
weight: torch.Tensor,
bias: torch.Tensor | None,
z: torch.Tensor | None = None,
eps: float = 1e-6,
group_size: int | None = None,
norm_before_gate: bool = True,
is_rms_norm: bool = False,
) -> torch.Tensor:
"""Reference implementation for both LayerNorm and RMSNorm (supports optional gate + group norm)."""
if is_rms_norm:
return rms_norm_ref(
x,
weight,
bias,
z=z,
eps=eps,
group_size=group_size,
norm_before_gate=norm_before_gate,
upcast=True,
)
dtype = x.dtype
x_f = x.float()
w_f = weight.float()
b_f = bias.float() if bias is not None else None
z_f = z.float() if z is not None else None
if z_f is not None and not norm_before_gate:
x_f = x_f * F.silu(z_f)
if group_size is None:
mean = x_f.mean(dim=-1, keepdim=True)
var = (x_f - mean).square().mean(dim=-1, keepdim=True)
rstd = torch.rsqrt(var + eps)
out = (x_f - mean) * rstd * w_f
if b_f is not None:
out = out + b_f
else:
hidden = x_f.shape[-1]
assert hidden % group_size == 0
ng = hidden // group_size
xg = x_f.view(*x_f.shape[:-1], ng, group_size)
mean = xg.mean(dim=-1, keepdim=True)
var = (xg - mean).square().mean(dim=-1, keepdim=True)
rstd = torch.rsqrt(var + eps)
xg = (xg - mean) * rstd
out = xg.reshape(*x_f.shape[:-1], hidden) * w_f
if b_f is not None:
out = out + b_f
if z_f is not None and norm_before_gate:
out = out * F.silu(z_f)
return out.to(dtype)
@dataclass(frozen=True)
class FwdCase:
name: str
with_gate: bool
norm_before_gate: bool
group_size: int | None
is_rms_norm: bool
CASES: list[FwdCase] = [
FwdCase(
"layernorm",
with_gate=False,
norm_before_gate=True,
group_size=None,
is_rms_norm=False,
),
FwdCase(
"rmsnorm",
with_gate=False,
norm_before_gate=True,
group_size=None,
is_rms_norm=True,
),
FwdCase(
"layernorm_gate_pre",
with_gate=True,
norm_before_gate=True,
group_size=None,
is_rms_norm=False,
),
FwdCase(
"layernorm_gate_post",
with_gate=True,
norm_before_gate=False,
group_size=None,
is_rms_norm=False,
),
FwdCase(
"rmsnorm_gate_pre",
with_gate=True,
norm_before_gate=True,
group_size=None,
is_rms_norm=True,
),
FwdCase(
"group_layernorm",
with_gate=False,
norm_before_gate=True,
group_size=128,
is_rms_norm=False,
),
FwdCase(
"group_rmsnorm",
with_gate=False,
norm_before_gate=True,
group_size=128,
is_rms_norm=True,
),
]
@pytest.mark.parametrize("num_tokens", [128])
@pytest.mark.parametrize("hidden_size", [256])
@pytest.mark.parametrize("dtype", [torch.bfloat16])
@pytest.mark.parametrize("case", CASES, ids=lambda c: c.name)
def test_layernorm_guard_fwd_spawn(
num_tokens: int,
hidden_size: int,
dtype: torch.dtype,
case: FwdCase,
device: str = "cuda",
):
_skip_if_no_cuda_or_not_enough_gpus(NUM_GPUS)
_skip_if_dtype_unsupported(dtype)
if case.group_size is not None and hidden_size % case.group_size != 0:
pytest.skip(
f"hidden_size {hidden_size} not divisible by group_size {case.group_size}"
)
master_port = _find_free_port()
world_size = NUM_GPUS
torch.multiprocessing.spawn(
_layernorm_guard_fwd_worker,
args=(
world_size,
master_port,
num_tokens,
hidden_size,
dtype,
case,
device,
),
nprocs=world_size,
join=True,
)
def _layernorm_guard_fwd_worker(
local_rank: int,
world_size: int,
master_port: int,
num_tokens: int,
hidden_size: int,
dtype: torch.dtype,
case: FwdCase,
device: str,
):
device = _setup_sglang_distributed(local_rank, world_size, master_port, dtype)
with torch.inference_mode():
torch.manual_seed(42 + local_rank)
torch.cuda.manual_seed_all(42 + local_rank)
x = torch.randn(num_tokens, hidden_size, dtype=dtype, device=device)
z = (
torch.randn(num_tokens, hidden_size, dtype=dtype, device=device)
if case.with_gate
else None
)
weight = torch.randn(hidden_size, dtype=dtype, device=device)
bias = (
None
if case.is_rms_norm
else torch.randn(hidden_size, dtype=dtype, device=device)
)
eps = 1e-6
out, mean, rstd = layer_norm_fwd(
x,
weight,
bias,
eps,
z=z,
group_size=case.group_size,
norm_before_gate=case.norm_before_gate,
is_rms_norm=case.is_rms_norm,
)
ref_out = layer_norm_ref(
x,
weight,
bias,
z=z,
eps=eps,
group_size=case.group_size,
norm_before_gate=case.norm_before_gate,
is_rms_norm=case.is_rms_norm,
)
assert out.shape == x.shape
assert out.dtype == x.dtype
torch.testing.assert_close(out, ref_out, atol=1e-2, rtol=1e-2)
# mean/rstd shape checks (same spirit as original vLLM tests)
if case.group_size is None:
if not case.is_rms_norm:
assert mean.shape == (num_tokens,)
assert rstd.shape == (num_tokens,)
else:
ngroups = hidden_size // case.group_size
if not case.is_rms_norm:
assert mean.shape == (ngroups * num_tokens,)
assert rstd.shape == (ngroups * num_tokens,)
@pytest.mark.parametrize("dtype", [torch.bfloat16])
def test_layernorm_guard_misc_spawn(dtype: torch.dtype, device: str = "cuda"):
_skip_if_no_cuda_or_not_enough_gpus(NUM_GPUS)
_skip_if_dtype_unsupported(dtype)
master_port = _find_free_port()
world_size = NUM_GPUS
torch.multiprocessing.spawn(
_layernorm_guard_misc_worker,
args=(world_size, master_port, dtype, device),
nprocs=world_size,
join=True,
)
def _layernorm_guard_misc_worker(
local_rank: int,
world_size: int,
master_port: int,
dtype: torch.dtype,
device: str,
):
device = _setup_sglang_distributed(local_rank, world_size, master_port, dtype)
with torch.inference_mode():
torch.manual_seed(123 + local_rank)
torch.cuda.manual_seed_all(123 + local_rank)
# 1) rows_per_block-like sizes
hidden_size = 1024
weight = torch.randn(hidden_size, dtype=dtype, device=device)
bias = torch.randn(hidden_size, dtype=dtype, device=device)
eps = 1e-6
for num_tokens in [513]:
x = torch.randn(num_tokens, hidden_size, dtype=dtype, device=device)
out, _, _ = layer_norm_fwd(x, weight, bias, eps, z=None, is_rms_norm=False)
ref = layer_norm_ref(x, weight, bias, z=None, eps=eps, is_rms_norm=False)
torch.testing.assert_close(out, ref, atol=1e-2, rtol=1e-2)
# 2) strided input (slice then contiguous)
num_tokens = 128
x_large = torch.randn(num_tokens, hidden_size * 2, dtype=dtype, device=device)
x = x_large[:, :hidden_size]
x_contig = x.contiguous()
out, _, _ = layer_norm_fwd(
x_contig, weight, bias, eps, z=None, is_rms_norm=False
)
ref = layer_norm_ref(x_contig, weight, bias, z=None, eps=eps, is_rms_norm=False)
torch.testing.assert_close(out, ref, atol=1e-2, rtol=1e-2)
# 3) provided output buffer
num_tokens = 256
x = torch.randn(num_tokens, hidden_size, dtype=dtype, device=device)
out_buf = torch.empty_like(x)
out, _, _ = layer_norm_fwd(
x, weight, bias, eps, z=None, out=out_buf, is_rms_norm=False
)
assert out.data_ptr() == out_buf.data_ptr()
ref = layer_norm_ref(x, weight, bias, z=None, eps=eps, is_rms_norm=False)
torch.testing.assert_close(out, ref, atol=1e-2, rtol=1e-2)
# 4) multidimensional input via autograd fn
for shape in [(4, 16, 1024)]:
hs = shape[-1]
x = torch.randn(*shape, dtype=dtype, device=device)
w = torch.randn(hs, dtype=dtype, device=device)
b = torch.randn(hs, dtype=dtype, device=device)
out = layernorm_fn(x, w, b, z=None, eps=eps)
ref = layer_norm_ref(x, w, b, z=None, eps=eps, is_rms_norm=False)
torch.testing.assert_close(out, ref, atol=1e-2, rtol=1e-2)
if __name__ == "__main__":
sys.exit(pytest.main([__file__]))
test/registered/perf/test_bench_serving_1gpu_part1.py
-33
View File
@@ -65,39 +65,6 @@ class TestBenchServing1GPUPart1(CustomTestCase):
else:
self.assertGreater(res["output_throughput"], 1050)
def test_offline_throughput_without_radix_cache(self):
res = run_bench_serving(
model=DEFAULT_MODEL_NAME_FOR_TEST,
num_prompts=500,
request_rate=float("inf"),
other_server_args=["--disable-radix-cache"],
)
if is_in_ci():
write_github_step_summary(
f"### test_offline_throughput_without_radix_cache\n"
f"Output throughput: {res['output_throughput']:.2f} token/s\n"
)
if is_in_amd_ci():
self.assertGreater(res["output_throughput"], 3050)
else:
self.assertGreater(res["output_throughput"], 3800)
def test_offline_throughput_without_chunked_prefill(self):
res = run_bench_serving(
model=DEFAULT_MODEL_NAME_FOR_TEST,
num_prompts=500,
request_rate=float("inf"),
other_server_args=["--chunked-prefill-size", "-1"],
)
if is_in_ci():
write_github_step_summary(
f"### test_offline_throughput_without_chunked_prefill\n"
f"Output throughput: {res['output_throughput']:.2f} token/s\n"
)
self.assertGreater(res["output_throughput"], 2600)
def test_offline_throughput_with_triton_attention_backend(self):
res = run_bench_serving(
model=DEFAULT_MODEL_NAME_FOR_TEST,
test/registered/perf/test_bench_serving_1gpu_part2.py
-26
View File
@@ -24,32 +24,6 @@ register_amd_ci(est_time=900, suite="stage-b-test-1-gpu-large-amd")
class TestBenchServing1GPUPart2(CustomTestCase):
@unittest.skip(
"Qwen2.5-VL server crashes with SIGBUS (exit code -7) on main; disable until fixed"
)
def test_vlm_offline_throughput(self):
res = run_bench_serving(
model=DEFAULT_SMALL_VLM_MODEL_NAME_FOR_TEST,
num_prompts=200,
request_rate=float("inf"),
other_server_args=[
"--mem-fraction-static",
"0.7",
],
dataset_name="mmmu",
)
if is_in_ci():
write_github_step_summary(
f"### test_vlm_offline_throughput\n"
f"Output throughput: {res['output_throughput']:.2f} token/s\n"
)
# relax for mi300x
if is_in_amd_ci():
self.assertGreater(res["output_throughput"], 900)
else:
self.assertGreater(res["output_throughput"], 2500)
def test_vlm_online_latency(self):
res = run_bench_serving(
model=DEFAULT_SMALL_VLM_MODEL_NAME_FOR_TEST,
test/registered/perf/test_bench_serving_2gpu.py
-18
View File
@@ -37,24 +37,6 @@ class TestBenchServing2GPU(CustomTestCase):
else:
self.assertGreater(res["output_throughput"], 2200)
def test_moe_offline_throughput_without_radix_cache(self):
res = run_bench_serving(
model=DEFAULT_MOE_MODEL_NAME_FOR_TEST,
num_prompts=300,
request_rate=float("inf"),
other_server_args=["--tp", "2", "--disable-radix-cache"],
)
if is_in_ci():
write_github_step_summary(
f"### test_moe_offline_throughput_without_radix_cache\n"
f"Output throughput: {res['output_throughput']:.2f} token/s\n"
)
if is_in_amd_ci():
self.assertGreater(res["output_throughput"], 2100)
else:
self.assertGreater(res["output_throughput"], 2200)
def test_pp_offline_throughput_default_decode(self):
res = run_bench_serving(
model=DEFAULT_MOE_MODEL_NAME_FOR_TEST,
test/registered/radix_cache/unified_radix_tree/test_unified_radix_cache_hicache_pp_kl.py
+1 -1
View File
@@ -19,7 +19,7 @@ from sglang.test.test_utils import (
terminate_and_kill_process_tree,
)
register_cuda_ci(est_time=738, stage="base-c", runner_config="4-gpu-h100")
register_cuda_ci(est_time=738, stage="nightly", runner_config="4-gpu-h100")
QWEN3_32B_MODEL = "Qwen/Qwen3-32B"
test/registered/scheduler/test_prefill_delayer.py
-800
View File
@@ -1,800 +0,0 @@
import asyncio
import os
import re
import time
import unittest
from dataclasses import dataclass
from types import SimpleNamespace
from typing import List, Optional
import openai
import requests
import torch
from sglang.benchmark.serving import run_benchmark
from sglang.srt.managers.prefill_delayer import PrefillDelayer
from sglang.srt.runtime_context import get_context
from sglang.srt.utils import kill_process_tree
from sglang.test.ci.ci_register import register_cuda_ci
from sglang.test.run_eval import run_eval
from sglang.test.test_utils import (
DEFAULT_MLA_MODEL_NAME_FOR_TEST,
DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
DEFAULT_URL_FOR_TEST,
CustomTestCase,
get_benchmark_args,
popen_launch_server,
run_distributed_test,
)
register_cuda_ci(
est_time=300,
stage="base-c",
runner_config="8-gpu-h200",
disabled="Temporarily disabled",
)
WORLD_SIZE = os.environ.get("SGLANG_TEST_WORLD_SIZE", "8")
# ============================ Unit Tests ============================
@dataclass
class NegotiateCall:
prefillable: List[bool]
token_usage: List[float]
# Optional scheduler state; when None, _run_negotiate_test does not pass
# the kwarg and the delayer falls back to the historical behavior of
# reading kwargs.get(..., 0).
running_batch: Optional[List[int]] = None
max_prefill_bs: Optional[List[int]] = None
waiting_queue_len: Optional[List[int]] = None
max_running_requests: Optional[int] = None
# Inter-call sleep (seconds). Used to exercise the queue-trigger
# wall-clock timeout.
sleep_before_s: float = 0.0
@dataclass
class NegotiateTestCase:
name: str
max_delay_passes: int
token_usage_low_watermark: Optional[float]
calls: List[NegotiateCall]
expected_allow: bool
expected_reason: str
# Queue-trigger knobs (new in the queue-based delayer). Leave both None
# to exercise the legacy slot-only code paths.
queue_min_ratio: Optional[float] = None
max_delay_ms: Optional[float] = None
prefill_max_requests: Optional[int] = None
# Expected accumulated wait surfaced on the final (release) outcome. When
# set, asserts the wait histograms would observe this value instead of 0.
expected_wait_forward_passes: Optional[int] = None
def _run_negotiate_test(rank, test_cases):
world_size = torch.distributed.get_world_size()
cpu_group = torch.distributed.new_group(backend="gloo")
for case in test_cases:
# The DP-attention gate is a published config leaf.
override = get_context().override_server_args(
enable_dp_attention=True,
prefill_delayer_queue_min_ratio=case.queue_min_ratio,
prefill_delayer_max_delay_ms=case.max_delay_ms,
prefill_max_requests=case.prefill_max_requests,
)
override.install()
delayer = PrefillDelayer(
dp_size=world_size,
attn_tp_size=1,
cpu_group=cpu_group,
max_delay_passes=case.max_delay_passes,
token_usage_low_watermark=case.token_usage_low_watermark,
)
for call in case.calls:
if call.sleep_before_s > 0:
time.sleep(call.sleep_before_s)
extra_kwargs = {}
if call.running_batch is not None:
extra_kwargs["running_batch"] = call.running_batch[rank]
if call.max_prefill_bs is not None:
extra_kwargs["max_prefill_bs"] = call.max_prefill_bs[rank]
if call.waiting_queue_len is not None:
extra_kwargs["waiting_queue_len"] = call.waiting_queue_len[rank]
if call.max_running_requests is not None:
extra_kwargs["max_running_requests"] = call.max_running_requests
result = delayer._negotiate_should_allow_prefill(
local_prefillable=call.prefillable[rank],
token_usage=call.token_usage[rank],
**extra_kwargs,
)
assert (result.output_allow, result.output_reason) == (
case.expected_allow,
case.expected_reason,
), f"Case {case.name} rank {rank}"
if case.expected_wait_forward_passes is not None:
assert result.wait_forward_passes == case.expected_wait_forward_passes, (
f"Case {case.name} rank {rank}: wait_forward_passes "
f"{result.wait_forward_passes} != {case.expected_wait_forward_passes}"
)
# On a release after a real wait, seconds must be observed too.
if case.expected_wait_forward_passes > 0:
assert result.wait_seconds > 0.0, (
f"Case {case.name} rank {rank}: wait_seconds not surfaced"
)
override.restore()
_NEGOTIATE_TEST_CASES = [
NegotiateTestCase(
name="all_prefillable",
max_delay_passes=100,
token_usage_low_watermark=0.8,
calls=[
NegotiateCall(
prefillable=[True, True, True, True],
token_usage=[0.9, 0.9, 0.9, 0.9],
)
],
expected_allow=True,
expected_reason="no_wait",
# No prior wait, so the histograms legitimately observe 0.
expected_wait_forward_passes=0,
),
NegotiateTestCase(
name="all_prefillable_with_previous_wait",
max_delay_passes=100,
token_usage_low_watermark=0.8,
calls=[
NegotiateCall(
prefillable=[True, False, True, False],
token_usage=[0.9, 0.9, 0.9, 0.9],
),
NegotiateCall(
prefillable=[True, True, True, True],
token_usage=[0.9, 0.9, 0.9, 0.9],
),
],
expected_allow=True,
expected_reason="wait_success",
# One mixed delay preceded the release, so the wait histograms must
# observe 1 forward pass (regression guard for #25949).
expected_wait_forward_passes=1,
),
NegotiateTestCase(
name="none_prefillable",
max_delay_passes=100,
token_usage_low_watermark=0.8,
calls=[
NegotiateCall(
prefillable=[False, False, False, False],
token_usage=[0.9, 0.9, 0.9, 0.9],
)
],
expected_allow=True,
expected_reason="",
),
NegotiateTestCase(
name="mixed_delay",
max_delay_passes=100,
token_usage_low_watermark=0.8,
calls=[
NegotiateCall(
prefillable=[True, False, True, False],
token_usage=[0.9, 0.9, 0.9, 0.9],
)
],
expected_allow=False,
expected_reason="delay",
),
NegotiateTestCase(
name="mixed_watermark_force_allow",
max_delay_passes=100,
token_usage_low_watermark=0.8,
calls=[
NegotiateCall(
prefillable=[True, False, True, False],
token_usage=[0.5, 0.9, 0.9, 0.9],
)
],
expected_allow=True,
expected_reason="token_watermark",
),
NegotiateTestCase(
name="mixed_watermark_disabled",
max_delay_passes=100,
token_usage_low_watermark=None,
calls=[
NegotiateCall(
prefillable=[True, False, True, False],
token_usage=[0.5, 0.9, 0.9, 0.9],
)
],
expected_allow=False,
expected_reason="delay",
),
NegotiateTestCase(
name="mixed_watermark_not_prefillable",
max_delay_passes=100,
token_usage_low_watermark=0.8,
calls=[
NegotiateCall(
prefillable=[False, False, True, False],
token_usage=[0.5, 0.9, 0.9, 0.9],
)
],
expected_allow=False,
expected_reason="delay",
),
NegotiateTestCase(
name="mixed_timeout",
max_delay_passes=3,
token_usage_low_watermark=0.8,
calls=[
NegotiateCall(
prefillable=[True, False, True, False],
token_usage=[0.9, 0.9, 0.9, 0.9],
),
NegotiateCall(
prefillable=[True, False, True, False],
token_usage=[0.9, 0.9, 0.9, 0.9],
),
NegotiateCall(
prefillable=[True, False, True, False],
token_usage=[0.9, 0.9, 0.9, 0.9],
),
],
expected_allow=True,
expected_reason="wait_timeout",
# Two delays accumulated before timing out; the timeout release must
# still surface that wait to the histograms.
expected_wait_forward_passes=2,
),
# Queue-based trigger: waiting queue below queue_min = min(running * R,
# max_prefill_bs) should defer prefill. With R=0.5, running=100 and
# max_prefill_bs=80, queue_min = min(50, 80) = 50, and queue_len=10 < 50.
NegotiateTestCase(
name="queue_trigger_delay",
max_delay_passes=100,
token_usage_low_watermark=0.8,
queue_min_ratio=0.5,
max_delay_ms=5000,
calls=[
NegotiateCall(
prefillable=[True, True, True, True],
token_usage=[0.9, 0.9, 0.9, 0.9],
running_batch=[100, 100, 100, 100],
max_prefill_bs=[80, 80, 80, 80],
waiting_queue_len=[10, 10, 10, 10],
max_running_requests=1024,
),
# skip_first_delayer consumes the first would-be delay; a second
# identical call must actually delay.
NegotiateCall(
prefillable=[True, True, True, True],
token_usage=[0.9, 0.9, 0.9, 0.9],
running_batch=[100, 100, 100, 100],
max_prefill_bs=[80, 80, 80, 80],
waiting_queue_len=[10, 10, 10, 10],
max_running_requests=1024,
),
],
expected_allow=False,
expected_reason="delay",
),
NegotiateTestCase(
name="queue_trigger_uses_admission_capacity",
max_delay_passes=100,
token_usage_low_watermark=0.8,
queue_min_ratio=0.02,
max_delay_ms=5000,
prefill_max_requests=128,
calls=[
NegotiateCall(
prefillable=[True, True, True, True],
token_usage=[0.9, 0.9, 0.9, 0.9],
running_batch=[500, 500, 500, 500],
max_prefill_bs=[1, 1, 1, 1],
waiting_queue_len=[1, 1, 1, 1],
max_running_requests=1024,
),
NegotiateCall(
prefillable=[True, True, True, True],
token_usage=[0.9, 0.9, 0.9, 0.9],
running_batch=[500, 500, 500, 500],
max_prefill_bs=[1, 1, 1, 1],
waiting_queue_len=[1, 1, 1, 1],
max_running_requests=1024,
),
],
expected_allow=False,
expected_reason="delay",
),
# Waiting queue at or above queue_min: queue trigger must not fire.
NegotiateTestCase(
name="queue_trigger_above_threshold",
max_delay_passes=100,
token_usage_low_watermark=0.8,
queue_min_ratio=0.5,
max_delay_ms=5000,
calls=[
NegotiateCall(
prefillable=[True, True, True, True],
token_usage=[0.9, 0.9, 0.9, 0.9],
running_batch=[100, 100, 100, 100],
max_prefill_bs=[80, 80, 80, 80],
waiting_queue_len=[64, 64, 64, 64],
max_running_requests=1024,
)
],
expected_allow=True,
expected_reason="no_wait",
),
# queue_min_ratio unset: queue trigger is opt-in and must stay disabled
# even when running_batch and queue_len would otherwise trigger it.
NegotiateTestCase(
name="queue_trigger_disabled_when_ratio_unset",
max_delay_passes=100,
token_usage_low_watermark=0.8,
queue_min_ratio=None,
max_delay_ms=None,
calls=[
NegotiateCall(
prefillable=[True, True, True, True],
token_usage=[0.9, 0.9, 0.9, 0.9],
running_batch=[100, 100, 100, 100],
max_prefill_bs=[80, 80, 80, 80],
waiting_queue_len=[1, 1, 1, 1],
max_running_requests=1024,
)
],
expected_allow=True,
expected_reason="no_wait",
),
# max_delay_ms wall-clock timeout: once a single queue-trigger delay
# exceeds the cap, prefill must be force-released.
# Call sequence:
# 1) queue_condition holds but skip_first_delayer consumes it
# (no state recorded, falls through to allow)
# 2) queue_condition holds -> delay, records start_time in state
# 3) after sleeping past max_delay_ms, elapsed >= cap -> force release
NegotiateTestCase(
name="queue_trigger_wall_clock_timeout",
max_delay_passes=100,
token_usage_low_watermark=0.8,
queue_min_ratio=0.5,
max_delay_ms=50,
calls=[
NegotiateCall(
prefillable=[True, True, True, True],
token_usage=[0.9, 0.9, 0.9, 0.9],
running_batch=[100, 100, 100, 100],
max_prefill_bs=[80, 80, 80, 80],
waiting_queue_len=[10, 10, 10, 10],
max_running_requests=1024,
),
NegotiateCall(
prefillable=[True, True, True, True],
token_usage=[0.9, 0.9, 0.9, 0.9],
running_batch=[100, 100, 100, 100],
max_prefill_bs=[80, 80, 80, 80],
waiting_queue_len=[10, 10, 10, 10],
max_running_requests=1024,
),
NegotiateCall(
prefillable=[True, True, True, True],
token_usage=[0.9, 0.9, 0.9, 0.9],
running_batch=[100, 100, 100, 100],
max_prefill_bs=[80, 80, 80, 80],
waiting_queue_len=[10, 10, 10, 10],
max_running_requests=1024,
sleep_before_s=0.2, # > max_delay_ms (50ms)
),
],
expected_allow=True,
expected_reason="wait_success",
# One queue-trigger delay was recorded before the wall-clock release.
expected_wait_forward_passes=1,
),
# slot_condition (all-branch) must not delay forever: with 128-100=28
# free slots < max_prefill_bs=80 the delay holds, but it must release
# with wait_timeout after max_delay_passes, like the mixed branch.
NegotiateTestCase(
name="slot_condition_pass_cap_timeout",
max_delay_passes=3,
token_usage_low_watermark=0.8,
calls=[
# skip_first_delayer consumes the first would-be delay.
NegotiateCall(
prefillable=[True, True, True, True],
token_usage=[0.9, 0.9, 0.9, 0.9],
running_batch=[100, 100, 100, 100],
max_prefill_bs=[80, 80, 80, 80],
waiting_queue_len=[10, 10, 10, 10],
max_running_requests=128,
),
NegotiateCall(
prefillable=[True, True, True, True],
token_usage=[0.9, 0.9, 0.9, 0.9],
running_batch=[100, 100, 100, 100],
max_prefill_bs=[80, 80, 80, 80],
waiting_queue_len=[10, 10, 10, 10],
max_running_requests=128,
),
NegotiateCall(
prefillable=[True, True, True, True],
token_usage=[0.9, 0.9, 0.9, 0.9],
running_batch=[100, 100, 100, 100],
max_prefill_bs=[80, 80, 80, 80],
waiting_queue_len=[10, 10, 10, 10],
max_running_requests=128,
),
NegotiateCall(
prefillable=[True, True, True, True],
token_usage=[0.9, 0.9, 0.9, 0.9],
running_batch=[100, 100, 100, 100],
max_prefill_bs=[80, 80, 80, 80],
waiting_queue_len=[10, 10, 10, 10],
max_running_requests=128,
),
],
expected_allow=True,
expected_reason="wait_timeout",
# Two slot-condition delays accumulated after the skip-first pass.
expected_wait_forward_passes=2,
),
]
class TestPrefillDelayerNegotiate(unittest.TestCase):
def test_negotiate(self):
run_distributed_test(
_run_negotiate_test,
world_size=4,
backend="gloo",
test_cases=_NEGOTIATE_TEST_CASES,
)
# ============================ E2E Tests ============================
class TestPrefillDelayerThroughputOnlineServing(CustomTestCase):
def test_throughput_comparison(self):
_run_throughput_comparison(
self,
test_name="online_serving",
other_launch_args=[
# Not really needed, only to test support non-FCFS algorithms
"--schedule-policy",
"lpm",
],
other_benchmark_args=dict(
num_prompts=500,
random_input_len=30000,
random_output_len=256,
request_rate=32,
),
# TODO: re-enable a throughput-improvement assertion once a
# workload that reliably exercises PrefillDelayer in online-
# serving mode is available. The current workload yields run-
# to-run noise on H200, while the offline test below shows the
# same code path is healthy (improvement ~+27%). We still
# validate functionality (server boot, benchmark completion,
# metrics emission).
min_improvement_pct=None,
)
class TestPrefillDelayerThroughputOfflineGen(CustomTestCase):
def test_throughput_comparison(self):
_run_throughput_comparison(
self,
test_name="offline_gen",
other_launch_args=["--max-total-tokens", "200000"],
other_benchmark_args=dict(
num_prompts=800,
random_input_len=30000,
random_output_len=500,
),
token_usage_low_watermark=0.8,
min_improvement_pct=20,
)
def _run_throughput_comparison(
test_case,
test_name: str,
other_launch_args,
other_benchmark_args,
min_improvement_pct: Optional[float],
token_usage_low_watermark: float = None,
):
common_kwargs = dict(
debug_name=test_name,
other_launch_args=other_launch_args,
other_benchmark_args=other_benchmark_args,
token_usage_low_watermark=token_usage_low_watermark,
)
res_enabled = _run_throughput_test(prefill_delayer=True, **common_kwargs)
res_disabled = _run_throughput_test(prefill_delayer=False, **common_kwargs)
_assert_throughput_improvement(
test_case,
test_name=test_name,
res_enabled=res_enabled,
res_disabled=res_disabled,
min_improvement_pct=min_improvement_pct,
)
def _run_throughput_test(
debug_name: str,
prefill_delayer: bool,
other_launch_args,
other_benchmark_args,
token_usage_low_watermark: float = None,
):
model = "Qwen/Qwen3-0.6B"
base_url = DEFAULT_URL_FOR_TEST
process = _launch_server(
prefill_delayer=prefill_delayer,
model=model,
base_url=base_url,
other_args=other_launch_args,
token_usage_low_watermark=token_usage_low_watermark,
)
try:
args = get_benchmark_args(
base_url=base_url,
dataset_name="random",
tokenizer=model,
**other_benchmark_args,
)
res = run_benchmark(args)
_print_prefill_delayer_metrics(base_url, expect_metrics=prefill_delayer)
finally:
kill_process_tree(process.pid)
print(f"=== {debug_name} ({prefill_delayer=}) ===")
res["total_throughput"] = res["input_throughput"] + res["output_throughput"]
print(f"Input throughput: {res['input_throughput']:.2f} token/s")
print(f"Output throughput: {res['output_throughput']:.2f} token/s")
print(f"Total throughput: {res['total_throughput']:.2f} token/s")
return res
def _assert_throughput_improvement(
test_case,
test_name: str,
res_enabled: dict,
res_disabled: dict,
min_improvement_pct: Optional[float],
):
test_case.assertEqual(
WORLD_SIZE,
"8",
f"This test requires 8 GPUs to properly measure throughput improvement, got {WORLD_SIZE}",
)
enabled = res_enabled["total_throughput"]
disabled = res_disabled["total_throughput"]
improvement_pct = (enabled - disabled) / disabled * 100
print(f"\n=== {test_name} Throughput Comparison ===")
print(
f"Total: enabled={enabled:.2f}, disabled={disabled:.2f}, improvement={improvement_pct:.2f}%"
)
if min_improvement_pct is None:
# Functionality-only mode: skip the perf assertion.
return
test_case.assertGreaterEqual(
improvement_pct,
min_improvement_pct,
f"{test_name}: Throughput improvement ({improvement_pct:.2f}%) < {min_improvement_pct}%",
)
class TestPrefillDelayerTokenUsageLowWatermark(CustomTestCase):
def test_1_with_low_watermark(self):
# The kv cache size here is deliberately small, thus we use smaller token usage
self._run(token_usage_low_watermark=0.5)
# TODO: re-enable once sglang/sglang#22511 (DP-attention detokenizer
# hang on H200 in CI) is fixed.
@unittest.skip("blocked by sgl-project/sglang#22511")
def test_2_without_low_watermark(self):
self._run(token_usage_low_watermark=None)
def _run(self, token_usage_low_watermark):
model = "Qwen/Qwen3-0.6B"
base_url = DEFAULT_URL_FOR_TEST
world_size = int(WORLD_SIZE)
process = _launch_server(
model=model,
base_url=base_url,
prefill_delayer=True,
other_args=["--max-total-tokens", "50000"],
# e.g. gen throughput is 370 tok/s on H200.
# Will need a different threshold on B200
max_delay_passes=3000,
token_usage_low_watermark=token_usage_low_watermark,
)
async def run_test():
client = openai.AsyncClient(base_url=f"{base_url}/v1", api_key="EMPTY")
long_prompt = "Hello " * 5000
async def send_blocking_request():
return await client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": long_prompt}],
max_tokens=10000,
extra_body={"data_parallel_rank": 0},
)
async def send_normal_request(dp_rank, req_idx):
start = time.time()
await client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": "Say hi"}],
max_tokens=10,
extra_body={"data_parallel_rank": dp_rank},
)
elapsed = time.time() - start
return dp_rank, req_idx, elapsed
asyncio.create_task(send_blocking_request())
await asyncio.sleep(3)
num_reqs_per_rank = 10
results = await asyncio.gather(
*[
send_normal_request(dp_rank, req_idx)
for dp_rank in range(1, world_size)
for req_idx in range(num_reqs_per_rank)
]
)
enabled = token_usage_low_watermark is not None
thresh = 5
for dp_rank, req_idx, elapsed in results:
print(f"DP rank {dp_rank} req {req_idx} completed in {elapsed:.2f}s")
self.assertTrue(
(elapsed < thresh) if enabled else (elapsed > thresh),
f"DP rank {dp_rank} req {req_idx}: elapsed={elapsed:.2f}s, thresh={thresh}, enabled={enabled}. "
f"Maybe you need a different `max_delay_passes` when using hardware other than H200.",
)
try:
asyncio.run(run_test())
metrics_text = _print_prefill_delayer_metrics(base_url, expect_metrics=True)
if token_usage_low_watermark is not None:
total = _sum_prometheus_metric_values(metrics_text, "token_watermark")
self.assertGreater(total, 0, "Expected token_watermark > 0")
print(f"total token_watermark: {total}")
finally:
kill_process_tree(process.pid)
class TestPrefillDelayerAccuracy(CustomTestCase):
def test_1_gsm8k_has_prefill_delayer(self):
self._run_accuracy_test(prefill_delayer=True)
def test_2_gsm8k_no_prefill_delayer(self):
self._run_accuracy_test(prefill_delayer=False)
def _run_accuracy_test(self, prefill_delayer: bool):
model = DEFAULT_MLA_MODEL_NAME_FOR_TEST
base_url = DEFAULT_URL_FOR_TEST
process = _launch_server(
prefill_delayer=prefill_delayer,
model=model,
base_url=base_url,
other_args=[
# Not really needed, only to test support non-FCFS algorithms
"--schedule-policy",
"lpm",
# Use this to ensure prefill delayer will be run
"--max-total-tokens",
"4096",
],
)
try:
args = SimpleNamespace(
base_url=base_url,
model=model,
eval_name="gsm8k",
num_examples=None,
num_threads=1024,
)
metrics = run_eval(args)
print(f"=== gsm8k ({prefill_delayer=}) ===")
print(f"{metrics=}")
self.assertGreater(metrics["score"], 0.57)
finally:
kill_process_tree(process.pid)
def _launch_server(
*,
model,
base_url,
prefill_delayer: bool,
other_args,
max_delay_passes: int = 100,
token_usage_low_watermark: float = None,
):
os.environ["SGLANG_PREFILL_DELAYER_DEBUG_LOG"] = "1"
return popen_launch_server(
model,
base_url,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
other_args=[
"--trust-remote-code",
"--tp",
WORLD_SIZE,
"--enable-dp-attention",
"--dp",
WORLD_SIZE,
"--chunked-prefill-size",
"131072",
"--mem-fraction-static",
"0.6",
"--enable-metrics",
*(["--enable-prefill-delayer"] if prefill_delayer else []),
"--prefill-delayer-max-delay-passes",
str(max_delay_passes),
*(
[
"--prefill-delayer-token-usage-low-watermark",
str(token_usage_low_watermark),
]
if token_usage_low_watermark is not None
else []
),
*(other_args or []),
],
)
def _print_prefill_delayer_metrics(base_url: str, expect_metrics: bool) -> str:
metrics_response = requests.get(f"{base_url}/metrics")
assert metrics_response.status_code == 200
metrics_text = metrics_response.text
prefill_delayer_metrics = [
line for line in metrics_text.split("\n") if "prefill_delayer" in line
]
print("=== PrefillDelayer Metrics ===")
for line in prefill_delayer_metrics:
print(line)
if expect_metrics:
assert "sglang:prefill_delayer_wait_forward_passes" in metrics_text
assert "sglang:prefill_delayer_wait_seconds" in metrics_text
assert "sglang:prefill_delayer_outcomes_total" in metrics_text
return metrics_text
def _sum_prometheus_metric_values(metrics_text: str, label_value: str) -> int:
matches = re.findall(rf'{label_value}".*?\}} (\d+)', metrics_text)
return sum(int(m) for m in matches)
if __name__ == "__main__":
unittest.main()
test/run_suite.py
+1
View File
@@ -101,6 +101,7 @@ PER_COMMIT_SUITES = {
"extra-b-test-4-gpu-h100",
"extra-b-test-4-gpu-b200",
"extra-b-test-8-gpu-h200",
"extra-b-test-8-gpu-b300",
],
HWBackend.NPU: [
"base-a-test-1-npu-a2",