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

config: the forwarding slots go; the dispatcher calls the family directly (#36792)

Browse Source 
Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Cheng Wan
2026-08-28 10:26:29 -07:00
committed by GitHub
co-authored by Claude Opus 5
parent 7bc3204117
commit ef20fab38a
23 changed files with 617 additions and 800 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test/registered/unit/server_args/test_model_config_reads_resolved_input.py
+27 -30
View File
@@ -337,38 +337,33 @@ def _pipeline():
methods = {
node.name: node for node in record.body if isinstance(node, ast.FunctionDef)
}
# The dispatcher calls its hooks by bare name, so the walk resolves those
# against `arg_groups/` alongside the record's own methods.
hooks = _hook_functions()
# Follow exactly one edge: the slot's own `from arg_groups.X import f` /
# `f(self)`. Merging every hook function by bare name would let the walk
# wander into families the slot never calls.
slot_target = {}
for name, node in methods.items():
imported = {
alias.asname or alias.name
for child in ast.walk(node)
if isinstance(child, ast.ImportFrom)
and child.module
and child.module.startswith("sglang.srt.arg_groups")
for alias in child.names
}
called = {
child.func.id
for child in ast.walk(node)
if isinstance(child, ast.Call) and isinstance(child.func, ast.Name)
}
for target in sorted(imported & called & set(hooks)):
slot_target.setdefault(name, target)
methods.update({name: hooks[name] for name in slot_target.values()})
methods.update({name: node for name, node in hooks.items() if name not in methods})
dispatch = methods["_run_resolution_pipeline"]
# A step is either a record method (`self._x()`) or a bare-name hook call.
steps = [
name
for _line, name in sorted(
(node.lineno, node.func.attr)
(
node.lineno,
(
node.func.attr
if isinstance(node.func, ast.Attribute)
else node.func.id
),
)
for node in ast.walk(dispatch)
if isinstance(node, ast.Call)
and isinstance(node.func, ast.Attribute)
and isinstance(node.func.value, ast.Name)
and node.func.value.id == "self"
and (
(
isinstance(node.func, ast.Attribute)
and isinstance(node.func.value, ast.Name)
and node.func.value.id == "self"
)
or (isinstance(node.func, ast.Name) and node.func.id in hooks)
)
)
]
@@ -387,20 +382,22 @@ def _pipeline():
and node.func.attr in methods
):
reaches(node.func.attr, seen)
target = slot_target.get(name)
if target is not None:
reaches(target, seen)
elif isinstance(node.func, ast.Name) and node.func.id in hooks:
reaches(node.func.id, seen)
return seen
step_lines = {}
for node in ast.walk(dispatch):
if not isinstance(node, ast.Call):
continue
if (
isinstance(node, ast.Call)
and isinstance(node.func, ast.Attribute)
isinstance(node.func, ast.Attribute)
and isinstance(node.func.value, ast.Name)
and node.func.value.id == "self"
):
step_lines.setdefault(node.func.attr, node.lineno)
elif isinstance(node.func, ast.Name) and node.func.id in hooks:
step_lines.setdefault(node.func.id, node.lineno)
return steps, methods, {name: reaches(name) for name in steps}, step_lines
test/registered/unit/server_args/test_page_major_backend_allowlist.py
+2 -1
View File
@@ -31,6 +31,7 @@ under its own default configuration.
import unittest
from sglang.srt.arg_groups.kv_cache_hook import handle_page_major_kv_layout
from sglang.srt.server_args import ServerArgs
from sglang.test.ci.ci_register import register_cpu_ci
@@ -68,7 +69,7 @@ def _accepts(
sa.use_mla_backend = lambda: use_mla
sa._resolved_attention_backends = lambda: [backend]
try:
ServerArgs._handle_page_major_kv_layout(sa)
handle_page_major_kv_layout(sa)
return True
except AssertionError:
return False
test/registered/unit/server_args/test_resolution_is_reproducible.py
+132
View File
@@ -1084,5 +1084,137 @@ class TestTheResolutionSeamHasOneCaller(CustomTestCase):
)
class TestResolutionStaysLazy(CustomTestCase):
"""Resolving a dummy model must not load the families it never reaches.
The forwarding slots imported their hook only when the step ran, so a
`ServerArgs(model_path="dummy")` resolution touched four hook modules. With
the slots gone the imports are function-local for the same reason, and a
module-level one costs every caller of the dummy boundary -- which is every
`override_server_args` in the test suite.
"""
def test_no_hook_module_imports_another_at_module_scope(self):
import ast
import sglang
srt = pathlib.Path(next(iter(sglang.__path__))).resolve() / "srt"
offenders = []
for path in sorted((srt / "arg_groups").glob("*.py")):
for node in ast.parse(path.read_text(encoding="utf-8-sig")).body:
if (
isinstance(node, ast.ImportFrom)
and node.module
and node.module.startswith("sglang.srt.arg_groups")
and node.module.endswith("_hook")
):
offenders.append(f"{path.name}:{node.lineno} -> {node.module}")
self.assertEqual(
offenders,
[],
"a hook module imports another at module scope, so loading one "
"family drags in a family it may never call. Import it inside the "
"function that calls it:\n " + "\n ".join(offenders),
)
def test_no_family_is_imported_before_the_step_that_calls_it(self):
"""Source-level, so it holds whatever else the process has imported.
Every hook import inside the dispatcher must come after the imports of
the families reached earlier and before its own first call -- what an
eager block at the top of the function breaks, and what a `sys.modules`
diff cannot see once another test has loaded those modules.
"""
import ast
import sglang
srt = pathlib.Path(next(iter(sglang.__path__))).resolve() / "srt"
tree = ast.parse((srt / "server_args.py").read_text(encoding="utf-8-sig"))
dispatch = next(
node
for node in ast.walk(tree)
if isinstance(node, ast.FunctionDef)
and node.name == "_run_resolution_pipeline"
)
early_return = min(
(
n.lineno
for n in ast.walk(dispatch)
if isinstance(n, ast.Return) and n.value is None
),
default=None,
)
self.assertIsNotNone(early_return, "the dummy short circuit is gone")
imported_early, called_early = set(), set()
for node in ast.walk(dispatch):
if (
isinstance(node, ast.ImportFrom)
and node.module
and node.module.endswith("_hook")
and node.lineno < early_return
):
imported_early.add(node.module.rsplit(".", 1)[-1])
if (
isinstance(node, ast.Call)
and isinstance(node.func, ast.Name)
and node.lineno < early_return
):
called_early.add(node.func.id)
hooks = {}
for path in sorted((srt / "arg_groups").glob("*_hook.py")):
for node in ast.parse(path.read_text(encoding="utf-8-sig")).body:
if isinstance(node, ast.FunctionDef):
hooks[node.name] = path.stem
needed_early = {hooks[name] for name in called_early if name in hooks}
self.assertEqual(
imported_early - needed_early,
set(),
"the dispatcher imports a hook family before the dummy short "
"circuit without calling it there, so every dummy resolution pays "
"for a family it never reaches",
)
def test_a_dummy_resolution_loads_only_what_it_reaches(self):
"""The same claim measured, in an interpreter of its own.
In-process this would be vacuous: another test that resolved a real
model has already imported the late families, and the `sys.modules`
diff comes back empty.
"""
import subprocess
import sys
import sglang
probe = (
"import sys\n"
"from sglang.srt.server_args import ServerArgs\n"
"before = set(sys.modules)\n"
"ServerArgs(model_path='dummy').resolve_once()\n"
"print(','.join(sorted(m.rsplit('.', 1)[-1] for m in set(sys.modules) - before"
" if '.arg_groups.' in m and m.endswith('_hook'))))\n"
)
env = dict(os.environ)
env["PYTHONPATH"] = str(
pathlib.Path(next(iter(sglang.__path__))).resolve().parent
)
out = subprocess.run(
[sys.executable, "-c", probe],
capture_output=True,
text=True,
timeout=300,
env=env,
)
self.assertEqual(out.returncode, 0, out.stderr[-2000:])
loaded = [name for name in out.stdout.strip().split(",") if name]
self.assertTrue(loaded, f"the probe reported nothing:\n{out.stdout}")
for late in ("model_hook", "cuda_graph_hook", "attention_hook", "lora_hook"):
self.assertNotIn(late, loaded)
if __name__ == "__main__":
unittest.main()
test/registered/unit/server_args/test_resolution_reads_the_declarations.py
+19 -3
View File
@@ -471,8 +471,22 @@ class TestResolutionReadsTheDeclarations(CustomTestCase):
def test_no_handler_reads_a_field_off_self(self):
handlers = _resolution_handlers()
self.assertGreater(
len(handlers), 50, f"only {len(handlers)} handlers were reached"
# What the dispatcher reaches inside the class is these read wrappers;
# the package side is covered by
# `test_no_hook_reads_a_field_off_the_record`. Pinned rather than
# counted: a walk that collapsed to the wrappers would clear any floor
# low enough to admit them.
self.assertEqual(
set(handlers),
{
"_run_resolution_pipeline",
"_handle_hardware_runtime_validation",
"_handle_page_size",
"_handle_pipeline_parallelism",
"_handle_sampling_backend",
},
f"the walk reached {sorted(handlers)}; if the dispatcher grew or "
"lost a handler, add it here after checking it reads the view",
)
offenders = []
for name, fn in sorted(handlers.items()):
@@ -492,7 +506,9 @@ class TestResolutionReadsTheDeclarations(CustomTestCase):
len(decided), 100, f"the declaration set derived only {len(decided)} fields"
)
members = _record_members()
self.assertGreater(len(members), 100, f"only {len(members)} members were found")
# The floor is here to catch the scan collapsing, not to pin the
# class's size.
self.assertGreater(len(members), 40, f"only {len(members)} members were found")
offenders = []
for name, fn in sorted(members.items()):
holders = _holders(fn) | {"self"}
test/registered/unit/server_args/test_server_args.py
+166 -123
View File
@@ -10,8 +10,51 @@ from unittest.mock import MagicMock, patch
import sglang.srt.server_args as server_args_module
from sglang.srt.arg_groups import parallel_hook, pd_disaggregation_hook, serving_hook
from sglang.srt.arg_groups.attention_hook import (
handle_attention_backend_compatibility,
handle_deterministic_inference,
)
from sglang.srt.arg_groups.cuda_graph_hook import (
disable_tc_piecewise_cudagraph_if_incompatible,
handle_cuda_graph_config,
)
from sglang.srt.arg_groups.hicache_hook import (
handle_hicache,
handle_hicache_ratio_default,
)
from sglang.srt.arg_groups.hisparse_hook import (
validate_hisparse_dsa_backend,
validate_hisparse_kv_cache_dtype,
)
from sglang.srt.arg_groups.kv_cache_hook import (
handle_cache_compatibility,
validate_prefill_only_disable_kv_cache_args,
)
from sglang.srt.arg_groups.mamba_hook import handle_mamba_backend
from sglang.srt.arg_groups.model_path_hook import handle_load_format
from sglang.srt.arg_groups.moe_hook import (
handle_a2a_moe,
validate_deepep_v2_dispatch_token_budget,
validate_deepep_v2_speculative_draft,
)
from sglang.srt.arg_groups.overrides import resolution_result
from sglang.srt.arg_groups.parallel_hook import (
handle_context_parallelism,
handle_data_parallelism,
handle_legacy_cp_arguments,
)
from sglang.srt.arg_groups.pd_disaggregation_hook import handle_pd_disaggregation
from sglang.srt.arg_groups.serving_hook import (
handle_crash_dump_env,
handle_deprecated_args,
handle_load_balance_method,
handle_missing_default_values,
handle_multimodal_feature_transport,
handle_ssl_validation,
handle_tokenizer_batching,
)
from sglang.srt.arg_groups.speculative_hook import handle_speculative_decoding
from sglang.srt.arg_groups.validation_hook import check_two_batch_overlap
from sglang.srt.entrypoints.sidecar import (
SGLANG_GRPC_ENDPOINT_ENV,
Sidecar,
@@ -56,7 +99,7 @@ class TestPrepareServerArgs(CustomTestCase):
# This validation runs before model construction and should allow the
# daemon to build the same static EPLB layout as the engine.
args._handle_load_format()
handle_load_format(args)
def test_enable_w4a4_mxfp4_megamoe_sets_deepgemm_env(self):
deepgemm_env = {
@@ -174,7 +217,7 @@ class TestPrepareServerArgs(CustomTestCase):
def test_draft_quantization_explicitness_survives_asdict_round_trip(self):
inherited = ServerArgs(model_path="dummy", quantization="modelopt_fp4")
inherited._handle_missing_default_values()
handle_missing_default_values(inherited)
self.assertEqual(
resolution_result(inherited, "speculative_draft_model_quantization"),
"modelopt_fp4",
@@ -186,7 +229,7 @@ class TestPrepareServerArgs(CustomTestCase):
)
reconstructed = ServerArgs(**dataclasses.asdict(inherited))
reconstructed._handle_missing_default_values()
handle_missing_default_values(reconstructed)
self.assertFalse(
resolution_result(
@@ -226,7 +269,7 @@ class TestMmEncoderDataParallelLogging(CustomTestCase):
)
with self.assertLogs(parallel_hook.logger, level="WARNING") as logs:
server_args._handle_data_parallelism()
handle_data_parallelism(server_args)
self.assertIn("TP=1", logs.output[0])
self.assertIn("no data-parallel work", logs.output[0])
@@ -237,7 +280,7 @@ class TestMmEncoderDataParallelLogging(CustomTestCase):
)
with self.assertLogs(parallel_hook.logger, level="INFO") as logs:
server_args._handle_data_parallelism()
handle_data_parallelism(server_args)
self.assertIn("TP=4", logs.output[0])
self.assertIn("high-resolution or multi-image", logs.output[0])
@@ -247,7 +290,7 @@ class TestImageProcessorBackend(CustomTestCase):
def test_new_backend_does_not_set_legacy_flag(self):
server_args = ServerArgs(model_path="dummy", image_processor_backend="pil")
server_args._handle_deprecated_args()
handle_deprecated_args(server_args)
self.assertEqual(
resolution_result(server_args, "image_processor_backend"), "pil"
@@ -258,7 +301,7 @@ class TestImageProcessorBackend(CustomTestCase):
server_args = ServerArgs(model_path="dummy", disable_fast_image_processor=True)
with self.assertLogs(serving_hook.logger, level="WARNING") as logs:
server_args._handle_deprecated_args()
handle_deprecated_args(server_args)
self.assertEqual(
resolution_result(server_args, "image_processor_backend"), "pil"
@@ -279,7 +322,7 @@ class TestImageProcessorBackend(CustomTestCase):
)
with self.assertRaisesRegex(ValueError, "conflicts.*torchvision"):
server_args._handle_deprecated_args()
handle_deprecated_args(server_args)
class TestMultimodalFeatureTransport(CustomTestCase):
@@ -298,7 +341,7 @@ class TestMultimodalFeatureTransport(CustomTestCase):
with patch.dict(os.environ, {"SGLANG_USE_CUDA_IPC_TRANSPORT": "0"}):
with self.assertLogs(serving_hook.logger, level="INFO") as logs:
server_args._handle_multimodal_feature_transport()
handle_multimodal_feature_transport(server_args)
self.assertEqual(
resolution_result(server_args, "mm_feature_transport"), "cuda_ipc"
@@ -315,7 +358,7 @@ class TestMultimodalFeatureTransport(CustomTestCase):
with patch.dict(os.environ, {"SGLANG_USE_CUDA_IPC_TRANSPORT": "0"}):
with self.assertLogs(serving_hook.logger, level="WARNING") as logs:
server_args._handle_multimodal_feature_transport()
handle_multimodal_feature_transport(server_args)
self.assertEqual(
resolution_result(server_args, "mm_feature_transport"), "cuda_ipc"
@@ -335,7 +378,7 @@ class TestMultimodalFeatureTransport(CustomTestCase):
)
with self.assertRaisesRegex(ValueError, "conflicts.*cuda_vmm"):
server_args._handle_multimodal_feature_transport()
handle_multimodal_feature_transport(server_args)
@patch("sglang.srt.arg_groups.serving_hook.is_cuda", return_value=True)
def test_explicit_cpu_overrides_legacy_environment(self, _mock_is_cuda):
@@ -343,7 +386,7 @@ class TestMultimodalFeatureTransport(CustomTestCase):
with patch.dict(os.environ, {"SGLANG_USE_CUDA_IPC_TRANSPORT": "1"}):
with self.assertLogs(serving_hook.logger, level="WARNING") as logs:
server_args._handle_multimodal_feature_transport()
handle_multimodal_feature_transport(server_args)
self.assertEqual(
resolution_result(server_args, "mm_feature_transport"), "cpu"
@@ -356,7 +399,7 @@ class TestMultimodalFeatureTransport(CustomTestCase):
server_args = ServerArgs(model_path="dummy")
with patch.dict(os.environ, {"SGLANG_USE_CUDA_IPC_TRANSPORT": "0"}):
server_args._handle_multimodal_feature_transport()
handle_multimodal_feature_transport(server_args)
self.assertEqual(
resolution_result(server_args, "mm_feature_transport"), "cpu"
@@ -371,7 +414,7 @@ class TestMultimodalFeatureTransport(CustomTestCase):
with patch.dict(os.environ, {}, clear=False):
envs.SGLANG_USE_CUDA_IPC_TRANSPORT.clear()
with self.assertNoLogs(server_args_module.logger, level="INFO"):
server_args._handle_multimodal_feature_transport()
handle_multimodal_feature_transport(server_args)
self.assertEqual(
resolution_result(server_args, "mm_feature_transport"), "cpu"
@@ -386,7 +429,7 @@ class TestMultimodalFeatureTransport(CustomTestCase):
with patch.dict(os.environ, {}, clear=False):
envs.SGLANG_USE_CUDA_IPC_TRANSPORT.clear()
with self.assertNoLogs(server_args_module.logger, level="INFO"):
server_args._handle_multimodal_feature_transport()
handle_multimodal_feature_transport(server_args)
self.assertEqual(
resolution_result(server_args, "mm_feature_transport"), "cpu"
@@ -415,7 +458,7 @@ class TestMultimodalFeatureTransport(CustomTestCase):
with patch.dict(os.environ, {}, clear=False):
envs.SGLANG_USE_CUDA_IPC_TRANSPORT.clear()
with self.assertLogs(serving_hook.logger, level="INFO") as logs:
server_args._handle_multimodal_feature_transport()
handle_multimodal_feature_transport(server_args)
self.assertEqual(
resolution_result(server_args, "mm_feature_transport"), "cuda_vmm"
@@ -446,7 +489,7 @@ class TestMultimodalFeatureTransport(CustomTestCase):
self._set_model_type(server_args, is_multimodal=True)
with self.assertLogs(serving_hook.logger, level="INFO") as logs:
server_args._handle_multimodal_feature_transport()
handle_multimodal_feature_transport(server_args)
self.assertEqual(resolution_result(server_args, "mm_feature_transport"), "cpu")
self.assertIn("has not opted into CUDA VMM", "\n".join(logs.output))
@@ -465,7 +508,7 @@ class TestMultimodalFeatureTransport(CustomTestCase):
with patch.dict(os.environ, {}, clear=False):
envs.SGLANG_USE_CUDA_IPC_TRANSPORT.clear()
with self.assertLogs(serving_hook.logger, level="INFO") as logs:
server_args._handle_multimodal_feature_transport()
handle_multimodal_feature_transport(server_args)
self.assertEqual(
resolution_result(server_args, "mm_feature_transport"), "cpu"
@@ -485,7 +528,7 @@ class TestMultimodalFeatureTransport(CustomTestCase):
with patch.dict(os.environ, {}, clear=False):
envs.SGLANG_USE_CUDA_IPC_TRANSPORT.clear()
server_args._handle_multimodal_feature_transport()
handle_multimodal_feature_transport(server_args)
self.assertEqual(
resolution_result(server_args, "mm_feature_transport"), "cpu"
@@ -499,7 +542,7 @@ class TestMultimodalFeatureTransport(CustomTestCase):
with patch.dict(os.environ, {}, clear=False):
envs.SGLANG_USE_CUDA_IPC_TRANSPORT.clear()
server_args._handle_multimodal_feature_transport()
handle_multimodal_feature_transport(server_args)
self.assertEqual(
resolution_result(server_args, "mm_feature_transport"), "cpu"
@@ -511,7 +554,7 @@ class TestMultimodalFeatureTransport(CustomTestCase):
server_args = ServerArgs(model_path="dummy", mm_feature_transport="cuda_ipc")
with self.assertRaisesRegex(ValueError, "requires NVIDIA CUDA"):
server_args._handle_multimodal_feature_transport()
handle_multimodal_feature_transport(server_args)
@patch("sglang.srt.arg_groups.serving_hook.is_cuda", return_value=True)
def test_cuda_ipc_rejects_multi_node(self, _mock_is_cuda):
@@ -520,7 +563,7 @@ class TestMultimodalFeatureTransport(CustomTestCase):
)
with self.assertRaisesRegex(ValueError, "single node"):
server_args._handle_multimodal_feature_transport()
handle_multimodal_feature_transport(server_args)
@patch("sglang.srt.arg_groups.serving_hook.is_cuda", return_value=True)
def test_cuda_vmm_is_explicit_and_uses_shared_budget(self, _mock_is_cuda):
@@ -536,7 +579,7 @@ class TestMultimodalFeatureTransport(CustomTestCase):
envs.SGLANG_MM_FEATURE_CACHE_MB.override(256),
):
with self.assertLogs(serving_hook.logger, level="INFO") as logs:
server_args._handle_multimodal_feature_transport()
handle_multimodal_feature_transport(server_args)
self.assertEqual(
resolution_result(server_args, "mm_feature_transport"), "cuda_vmm"
@@ -554,7 +597,7 @@ class TestMultimodalFeatureTransport(CustomTestCase):
server_args = ServerArgs(model_path="dummy", mm_feature_transport="cuda_vmm")
with self.assertRaisesRegex(ValueError, "requires NVIDIA CUDA"):
server_args._handle_multimodal_feature_transport()
handle_multimodal_feature_transport(server_args)
@patch("sglang.srt.arg_groups.serving_hook.is_cuda", return_value=True)
def test_cuda_vmm_rejects_rust_server(self, _mock_is_cuda):
@@ -564,7 +607,7 @@ class TestMultimodalFeatureTransport(CustomTestCase):
envs.SGLANG_RUST_SERVER.override(True),
self.assertRaisesRegex(ValueError, "SGLANG_RUST_SERVER"),
):
server_args._handle_multimodal_feature_transport()
handle_multimodal_feature_transport(server_args)
@patch("sglang.srt.arg_groups.serving_hook.is_cuda", return_value=True)
def test_cuda_vmm_rejects_pipeline_parallelism(self, _mock_is_cuda):
@@ -573,7 +616,7 @@ class TestMultimodalFeatureTransport(CustomTestCase):
)
with self.assertRaisesRegex(ValueError, "pipeline parallelism"):
server_args._handle_multimodal_feature_transport()
handle_multimodal_feature_transport(server_args)
class TestMambaCacheStochasticRounding(unittest.TestCase):
@@ -585,7 +628,7 @@ class TestMambaCacheStochasticRounding(unittest.TestCase):
)
with self.assertRaisesRegex(ValueError, "--mamba-ssm-dtype float16"):
server_args._handle_mamba_backend()
handle_mamba_backend(server_args)
@patch("sglang.srt.arg_groups.mamba_hook.is_cuda", return_value=False)
def test_rejects_non_cuda(self, _mock_is_cuda):
@@ -596,7 +639,7 @@ class TestMambaCacheStochasticRounding(unittest.TestCase):
)
with self.assertRaisesRegex(ValueError, "NVIDIA CUDA"):
server_args._handle_mamba_backend()
handle_mamba_backend(server_args)
@patch("sglang.srt.arg_groups.mamba_hook.is_cuda", return_value=True)
@patch("sglang.srt.arg_groups.mamba_hook.is_sm100_supported", return_value=False)
@@ -609,14 +652,14 @@ class TestMambaCacheStochasticRounding(unittest.TestCase):
)
with self.assertRaisesRegex(ValueError, "requires SM100"):
server_args._handle_mamba_backend()
handle_mamba_backend(server_args)
class TestLoadBalanceMethod(unittest.TestCase):
def _load_balance_args(self, **kwargs):
server_args = ServerArgs(model_path="dummy", **kwargs)
server_args._handle_pd_disaggregation()
server_args._handle_load_balance_method()
handle_pd_disaggregation(server_args)
handle_load_balance_method(server_args)
return server_args
def test_non_pd_defaults_to_round_robin(self):
@@ -645,7 +688,7 @@ class TestLoadBalanceMethod(unittest.TestCase):
dcp_size=4,
)
with self.assertLogs(pd_disaggregation_hook.logger, level="WARNING") as logs:
server_args._handle_pd_disaggregation()
handle_pd_disaggregation(server_args)
self.assertIn("without improving prefill performance", "\n".join(logs.output))
def test_pd_decode_dcp_forces_chunk_cache(self):
@@ -666,7 +709,7 @@ class TestLoadBalanceMethod(unittest.TestCase):
with self.assertRaisesRegex(
ValueError, "mooncake, nixl, or fake for synthetic benchmarking"
):
server_args._handle_pd_disaggregation()
handle_pd_disaggregation(server_args)
def test_pd_decode_dcp_allows_fake_transfer_backend(self):
server_args = self._load_balance_args(
@@ -685,7 +728,7 @@ class TestLoadBalanceMethod(unittest.TestCase):
dcp_size=4,
)
with self.assertRaisesRegex(ValueError, "currently requires chunk cache"):
server_args._handle_pd_disaggregation()
handle_pd_disaggregation(server_args)
def test_pd_decode_dcp_rejects_hierarchical_cache(self):
server_args = ServerArgs(
@@ -696,7 +739,7 @@ class TestLoadBalanceMethod(unittest.TestCase):
dcp_size=4,
)
with self.assertRaisesRegex(ValueError, "--enable-hierarchical-cache"):
server_args._handle_pd_disaggregation()
handle_pd_disaggregation(server_args)
def test_pd_decode_radix_cache_rejects_hisparse(self):
server_args = ServerArgs(
@@ -707,7 +750,7 @@ class TestLoadBalanceMethod(unittest.TestCase):
enable_hisparse=True,
)
with self.assertRaises(ValueError) as context:
server_args._handle_pd_disaggregation()
handle_pd_disaggregation(server_args)
self.assertIn(
"--disaggregation-decode-enable-radix-cache is incompatible with "
@@ -723,7 +766,7 @@ class TestLoadBalanceMethod(unittest.TestCase):
disaggregation_transfer_backend="fake",
)
with self.assertRaises(ValueError) as context:
server_args._handle_pd_disaggregation()
handle_pd_disaggregation(server_args)
self.assertIn(
"--disaggregation-decode-enable-radix-cache is incompatible "
@@ -754,7 +797,7 @@ class TestSkipTokenizerInit(unittest.TestCase):
detokenizer_worker_num=3,
)
server_args._handle_tokenizer_batching()
handle_tokenizer_batching(server_args)
# Tokenizer fanout preserved; detokenizer coerced to 1 (no decode work).
self.assertEqual(resolution_result(server_args, "tokenizer_worker_num"), 4)
@@ -830,8 +873,8 @@ class TestHiSparseDsaBackendPolicy(unittest.TestCase):
dsa_decode_backend="flashinfer_sparse_mla",
)
server_args._validate_hisparse_dsa_backend("dsa_prefill_backend", "prefill")
server_args._validate_hisparse_dsa_backend("dsa_decode_backend", "decode")
validate_hisparse_dsa_backend(server_args, "dsa_prefill_backend", "prefill")
validate_hisparse_dsa_backend(server_args, "dsa_decode_backend", "decode")
@patch("sglang.srt.server_args.is_hip", return_value=True)
def test_hisparse_defaults_to_tilelang_on_rocm(self, _mock_is_hip):
@@ -859,8 +902,8 @@ class TestHiSparseDsaBackendPolicy(unittest.TestCase):
dsa_decode_backend="aiter",
)
server_args._validate_hisparse_dsa_backend("dsa_prefill_backend", "prefill")
server_args._validate_hisparse_dsa_backend("dsa_decode_backend", "decode")
validate_hisparse_dsa_backend(server_args, "dsa_prefill_backend", "prefill")
validate_hisparse_dsa_backend(server_args, "dsa_decode_backend", "decode")
@patch("sglang.srt.server_args.is_hip", return_value=True)
def test_hisparse_rejects_cuda_backend_on_rocm(self, _mock_is_hip):
@@ -872,7 +915,7 @@ class TestHiSparseDsaBackendPolicy(unittest.TestCase):
)
with self.assertRaisesRegex(ValueError, "tilelang"):
server_args._validate_hisparse_dsa_backend("dsa_prefill_backend", "prefill")
validate_hisparse_dsa_backend(server_args, "dsa_prefill_backend", "prefill")
@patch("sglang.srt.server_args.is_hip", return_value=False)
def test_hisparse_rejects_rocm_backend_on_cuda(self, _mock_is_hip):
@@ -884,7 +927,7 @@ class TestHiSparseDsaBackendPolicy(unittest.TestCase):
)
with self.assertRaisesRegex(ValueError, "flashmla_sparse"):
server_args._validate_hisparse_dsa_backend("dsa_decode_backend", "decode")
validate_hisparse_dsa_backend(server_args, "dsa_decode_backend", "decode")
def test_hisparse_accepts_bfloat16_kv_cache_dtype(self):
server_args = ServerArgs(
@@ -893,7 +936,7 @@ class TestHiSparseDsaBackendPolicy(unittest.TestCase):
kv_cache_dtype="bfloat16",
)
server_args._validate_hisparse_kv_cache_dtype()
validate_hisparse_kv_cache_dtype(server_args)
def test_hisparse_accepts_fp8_e4m3_kv_cache_dtype(self):
server_args = ServerArgs(
@@ -902,7 +945,7 @@ class TestHiSparseDsaBackendPolicy(unittest.TestCase):
kv_cache_dtype="fp8_e4m3",
)
server_args._validate_hisparse_kv_cache_dtype()
validate_hisparse_kv_cache_dtype(server_args)
def test_hisparse_rejects_unsupported_kv_cache_dtype(self):
server_args = ServerArgs(
@@ -912,7 +955,7 @@ class TestHiSparseDsaBackendPolicy(unittest.TestCase):
)
with self.assertRaisesRegex(ValueError, r"fp8_e4m3"):
server_args._validate_hisparse_kv_cache_dtype()
validate_hisparse_kv_cache_dtype(server_args)
class TestFa4PageSizeAutoForce(CustomTestCase):
@@ -942,7 +985,7 @@ class TestFa4PageSizeAutoForce(CustomTestCase):
# `--attention-backend fa4` (combined): prefill/decode fields stay None.
args = self._make_args(attention_backend="fa4")
args._handle_attention_backend_compatibility()
handle_attention_backend_compatibility(args)
from sglang.srt.arg_groups.overrides import resolved_view
@@ -955,7 +998,7 @@ class TestFa4PageSizeAutoForce(CustomTestCase):
# `--prefill-attention-backend fa4`: the previously-covered path.
args = self._make_args(attention_backend=None, prefill="fa4", page_size=1)
args._handle_attention_backend_compatibility()
handle_attention_backend_compatibility(args)
from sglang.srt.arg_groups.overrides import resolved_view
@@ -1002,7 +1045,7 @@ class TestContextParallelServerArgs(CustomTestCase):
cp_strategy=resolution_result(args, "cp_strategy"),
)
with self.assertRaisesRegex(ValueError, "--cp-strategy"):
server_args._handle_context_parallelism()
handle_context_parallelism(server_args)
def test_deprecated_dsa_cp_mode_maps_to_unified_strategy(self):
args = self.parser.parse_args(
@@ -1021,7 +1064,7 @@ class TestContextParallelServerArgs(CustomTestCase):
dsa_prefill_cp_mode=resolution_result(args, "dsa_prefill_cp_mode"),
)
server_args._handle_legacy_cp_arguments()
handle_legacy_cp_arguments(server_args)
self.assertTrue(resolution_result(server_args, "enable_prefill_cp"))
self.assertEqual(resolution_result(server_args, "cp_strategy"), "interleave")
@@ -1036,8 +1079,8 @@ class TestContextParallelServerArgs(CustomTestCase):
attention_backend="dsa",
)
server_args._handle_legacy_cp_arguments()
server_args._handle_context_parallelism()
handle_legacy_cp_arguments(server_args)
handle_context_parallelism(server_args)
self.assertTrue(
resolution_result(server_args, "enable_dsa_prefill_context_parallel")
@@ -1060,7 +1103,7 @@ class TestContextParallelServerArgs(CustomTestCase):
tp_size=2,
)
server_args._handle_context_parallelism()
handle_context_parallelism(server_args)
self.assertTrue(is_cp_enabled())
self.assertTrue(is_interleave())
@@ -1132,8 +1175,8 @@ class TestContextParallelServerArgs(CustomTestCase):
with self.subTest(name=name):
server_args = self._new_cp_args(**overrides)
server_args._handle_legacy_cp_arguments()
server_args._handle_context_parallelism()
handle_legacy_cp_arguments(server_args)
handle_context_parallelism(server_args)
self.assertTrue(resolution_result(server_args, "enable_prefill_cp"))
self.assertEqual(
@@ -1303,7 +1346,7 @@ class TestPortArgs(unittest.TestCase):
class TestSSLArgs(unittest.TestCase):
def _validate_ssl(self, **kwargs):
server_args = ServerArgs(model_path="dummy", **kwargs)
server_args._handle_ssl_validation()
handle_ssl_validation(server_args)
return server_args
def test_ssl_keyfile_without_certfile_raises(self):
@@ -1413,7 +1456,7 @@ class TestHiCacheArgs(unittest.TestCase):
# so `_handle_hicache` would never run. Its one prerequisite (the
# host/device ratio default) is run by hand.
args = ServerArgs(model_path="dummy", **overrides)
args._handle_hicache_ratio_default()
handle_hicache_ratio_default(args)
return args
def _assert_hicache_fields(
@@ -1495,7 +1538,7 @@ class TestHiCacheArgs(unittest.TestCase):
for case in cases:
with self.subTest(case=case["name"]):
args = self._make_args(**case["overrides"])
args._handle_hicache()
handle_hicache(args)
self._assert_hicache_fields(
args,
expected_io_backend=case["expected_io_backend"],
@@ -1510,7 +1553,7 @@ class TestHiCacheArgs(unittest.TestCase):
attention_backend="fa3",
decode_attention_backend=None,
)
args._handle_hicache()
handle_hicache(args)
self.assertEqual(resolution_result(args, "hicache_io_backend"), "kernel")
self.assertEqual(resolution_result(args, "hicache_mem_layout"), "page_first")
@@ -1525,7 +1568,7 @@ class TestHiCacheArgs(unittest.TestCase):
)
with self.assertRaisesRegex(ValueError, "mutually exclusive"):
args._handle_cache_compatibility()
handle_cache_compatibility(args)
def test_decode_offload_allows_cpu_tensor_retraction(self):
args = self._make_args(
@@ -1535,7 +1578,7 @@ class TestHiCacheArgs(unittest.TestCase):
disaggregation_decode_retraction_backup="cpu_tensor",
)
args._handle_cache_compatibility()
handle_cache_compatibility(args)
class TestNgramExternalSamArgs(CustomTestCase):
@@ -1659,7 +1702,7 @@ class TestWaterfillArgs(CustomTestCase):
disable_shared_experts_fusion=True,
)
# dummy-model path short-circuits __post_init__; invoke the handler directly.
server_args._handle_a2a_moe()
handle_a2a_moe(server_args)
from sglang.srt.arg_groups.overrides import resolved_view
@@ -1674,7 +1717,7 @@ class TestWaterfillArgs(CustomTestCase):
enable_waterfill=True,
)
# dummy-model path short-circuits __post_init__; invoke the handler directly.
server_args._handle_a2a_moe()
handle_a2a_moe(server_args)
from sglang.srt.arg_groups.overrides import resolved_view
@@ -1690,7 +1733,7 @@ class TestWaterfillArgs(CustomTestCase):
disable_shared_experts_fusion=True,
)
# dummy-model path short-circuits __post_init__; invoke the handler directly.
server_args._handle_a2a_moe()
handle_a2a_moe(server_args)
from sglang.srt.arg_groups.overrides import resolved_view
@@ -1706,7 +1749,7 @@ class TestWaterfillArgs(CustomTestCase):
deepep_mode="low_latency",
)
# dummy-model path short-circuits __post_init__; invoke the handler directly.
server_args._handle_a2a_moe()
handle_a2a_moe(server_args)
self.assertEqual(resolution_result(server_args, "deepep_mode"), "low_latency")
self.assertFalse(resolution_result(server_args, "disable_cuda_graph"))
@@ -1739,8 +1782,8 @@ class TestPrefillOnlyDisableKvCache(unittest.TestCase):
def _validate_prefill_only_args(self, **overrides):
sa = ServerArgs(**self._base_kwargs(**overrides))
sa._handle_legacy_cp_arguments()
sa._validate_prefill_only_disable_kv_cache_args()
handle_legacy_cp_arguments(sa)
validate_prefill_only_disable_kv_cache_args(sa)
return sa
def test_valid_minimal_config_constructs(self):
@@ -1829,7 +1872,7 @@ class TestCudaGraphDisaggregationRoles(CustomTestCase):
patch("sglang.srt.utils.is_cuda", return_value=True),
patch.object(ServerArgs, "use_mla_backend", return_value=False),
):
args._handle_cuda_graph_config()
handle_cuda_graph_config(args)
return args
def test_cuda_graph_prefill_role_defaults_disable_decode_graph(self):
@@ -1902,7 +1945,7 @@ class TestPrefillCudaGraphLoRACompatibility(CustomTestCase):
patch("sglang.srt.utils.is_cuda", return_value=True),
patch.object(ServerArgs, "use_mla_backend", return_value=False),
):
args._handle_cuda_graph_config()
handle_cuda_graph_config(args)
return args
def test_enable_lora_keeps_breakable_prefill_graph(self):
@@ -1941,7 +1984,7 @@ class TestPrefillCudaGraphLoRACompatibility(CustomTestCase):
patch("sglang.srt.arg_groups.cuda_graph_hook.is_mps", return_value=False),
patch("sglang.srt.arg_groups.cuda_graph_hook.is_xpu", return_value=False),
):
args._disable_tc_piecewise_cudagraph_if_incompatible()
disable_tc_piecewise_cudagraph_if_incompatible(args)
self.assertEqual(
resolution_result(args, "cuda_graph_config").prefill.backend,
@@ -1966,7 +2009,7 @@ class TestBreakableCudaGraphMultimodalAllowlist(CustomTestCase):
patch("sglang.srt.utils.is_cuda", return_value=True),
patch.object(ServerArgs, "use_mla_backend", return_value=False),
):
args._handle_cuda_graph_config()
handle_cuda_graph_config(args)
return args
def test_multimodal_arch_disables_prefill_breakable(self):
@@ -2163,7 +2206,7 @@ class TestDeepEPv2Args(CustomTestCase):
):
args = self._args(moe_runner_backend="deep_gemm")
args._model_config.hf_config.architectures = [architecture]
args._handle_a2a_moe()
handle_a2a_moe(args)
def test_unvalidated_and_missing_architectures_rejected(self):
for architectures in (
@@ -2175,7 +2218,7 @@ class TestDeepEPv2Args(CustomTestCase):
args = self._args(moe_runner_backend="deep_gemm")
args._model_config.hf_config.architectures = architectures
with self.assertRaisesRegex(ValueError, "not validated"):
args._handle_a2a_moe()
handle_a2a_moe(args)
def test_instance_connector_rejected(self):
args = self._args(
@@ -2183,7 +2226,7 @@ class TestDeepEPv2Args(CustomTestCase):
moe_runner_backend="deep_gemm",
)
with self.assertRaisesRegex(ValueError, "instance connector"):
args._handle_a2a_moe()
handle_a2a_moe(args)
def test_deterministic_inference_rejected(self):
args = self._args(
@@ -2191,7 +2234,7 @@ class TestDeepEPv2Args(CustomTestCase):
enable_deterministic_inference=True,
)
with self.assertRaisesRegex(ValueError, "deterministic sorting"):
args._handle_a2a_moe()
handle_a2a_moe(args)
def test_rl_on_policy_deterministic_inference_rejected(self):
args = self._args(
@@ -2205,9 +2248,9 @@ class TestDeepEPv2Args(CustomTestCase):
envs.SGLANG_ENABLE_DETERMINISTIC_INFERENCE.get()
),
):
args._handle_deterministic_inference()
handle_deterministic_inference(args)
with self.assertRaisesRegex(ValueError, "deterministic sorting"):
args._handle_a2a_moe()
handle_a2a_moe(args)
def test_deterministic_inference_does_not_affect_legacy_deepep(self):
args = self._args(
@@ -2215,7 +2258,7 @@ class TestDeepEPv2Args(CustomTestCase):
moe_runner_backend="deep_gemm",
enable_deterministic_inference=True,
)
args._handle_a2a_moe()
handle_a2a_moe(args)
def test_runner_restored_by_declaration_fails_fast(self):
# Validate the declaration-resolved runner rather than the raw field.
@@ -2224,13 +2267,13 @@ class TestDeepEPv2Args(CustomTestCase):
("test_mxfp8", {"moe_runner_backend": "flashinfer_trtllm"})
]
with self.assertRaises(ValueError):
args._handle_a2a_moe()
handle_a2a_moe(args)
def test_declarations_resolve_ep_size_and_fusion(self):
from sglang.srt.arg_groups.overrides import resolved_view
args = self._args(moe_runner_backend="auto", tp_size=2)
args._handle_a2a_moe()
handle_a2a_moe(args)
self.assertEqual(resolved_view(args).ep_size, args.tp_size)
self.assertTrue(resolved_view(args).disable_shared_experts_fusion)
@@ -2238,23 +2281,23 @@ class TestDeepEPv2Args(CustomTestCase):
from sglang.srt.arg_groups.overrides import resolved_view
args = self._args(moe_runner_backend="auto")
args._handle_a2a_moe()
handle_a2a_moe(args)
self.assertEqual(resolved_view(args).moe_runner_backend, "deep_gemm")
def test_unsupported_runner_rejected(self):
args = self._args(moe_runner_backend="flashinfer_trtllm")
with self.assertRaises(ValueError):
args._handle_a2a_moe()
handle_a2a_moe(args)
def test_triton_runner_rejected(self):
args = self._args(moe_runner_backend="triton")
with self.assertRaises(ValueError):
args._handle_a2a_moe()
handle_a2a_moe(args)
def test_decode_graph_stays_enabled_in_both_comm_modes(self):
for mode in ("direct", "hybrid"):
args = self._args(moe_runner_backend="deep_gemm", deepep_v2_mode=mode)
args._handle_a2a_moe()
handle_a2a_moe(args)
declared = resolution_result(args, "cuda_graph_config")
self.assertEqual(declared.decode.backend, Backend.FULL)
self.assertEqual(declared.prefill.backend, Backend.DISABLED)
@@ -2262,7 +2305,7 @@ class TestDeepEPv2Args(CustomTestCase):
def test_two_batch_overlap_rejected(self):
args = self._args(moe_runner_backend="deep_gemm", enable_two_batch_overlap=True)
with self.assertRaises(ValueError):
args._handle_a2a_moe()
handle_a2a_moe(args)
def test_speculative_draft_backend_rejected(self):
for main_backend in ("none", "deepep", "deepep_v2"):
@@ -2272,7 +2315,7 @@ class TestDeepEPv2Args(CustomTestCase):
speculative_moe_a2a_backend="deepep_v2",
)
with self.assertRaisesRegex(ValueError, "speculative draft backend"):
args._validate_deepep_v2_speculative_draft()
validate_deepep_v2_speculative_draft(args)
def test_inherited_speculative_draft_backend_rejected(self):
args = self._args(
@@ -2280,14 +2323,14 @@ class TestDeepEPv2Args(CustomTestCase):
speculative_algorithm="EAGLE",
)
with self.assertRaisesRegex(ValueError, "speculative draft backend"):
args._validate_deepep_v2_speculative_draft()
validate_deepep_v2_speculative_draft(args)
def test_ngram_does_not_inherit_a_draft_backend(self):
args = self._args(
moe_runner_backend="deep_gemm",
speculative_algorithm="NGRAM",
)
args._validate_deepep_v2_speculative_draft()
validate_deepep_v2_speculative_draft(args)
def test_explicit_legacy_speculative_backend_allowed(self):
args = self._args(
@@ -2295,7 +2338,7 @@ class TestDeepEPv2Args(CustomTestCase):
speculative_algorithm="EAGLE",
speculative_moe_a2a_backend="deepep",
)
args._validate_deepep_v2_speculative_draft()
validate_deepep_v2_speculative_draft(args)
def test_resolved_legacy_speculative_backend_allowed(self):
args = self._args(
@@ -2308,18 +2351,18 @@ class TestDeepEPv2Args(CustomTestCase):
{"speculative_moe_a2a_backend": "deepep"},
)
]
args._validate_deepep_v2_speculative_draft()
validate_deepep_v2_speculative_draft(args)
def test_prefill_chunk_exceeding_cap_rejected(self):
args = self._args(moe_runner_backend="deep_gemm", chunked_prefill_size=2048)
with envs.SGLANG_DEEPEP_V2_NUM_MAX_DISPATCH_TOKENS_PER_RANK.override(1024):
with self.assertRaisesRegex(ValueError, "NUM_MAX_DISPATCH_TOKENS_PER_RANK"):
args._validate_deepep_v2_dispatch_token_budget()
validate_deepep_v2_dispatch_token_budget(args)
def test_prefill_chunk_at_cap_boundary_accepted(self):
args = self._args(moe_runner_backend="deep_gemm", chunked_prefill_size=1024)
with envs.SGLANG_DEEPEP_V2_NUM_MAX_DISPATCH_TOKENS_PER_RANK.override(1024):
args._validate_deepep_v2_dispatch_token_budget()
validate_deepep_v2_dispatch_token_budget(args)
def test_dynamic_chunking_probe_is_included(self):
args = self._args(
@@ -2331,7 +2374,7 @@ class TestDeepEPv2Args(CustomTestCase):
)
with envs.SGLANG_DEEPEP_V2_NUM_MAX_DISPATCH_TOKENS_PER_RANK.override(1024):
with self.assertRaisesRegex(ValueError, "required=1280"):
args._validate_deepep_v2_dispatch_token_budget()
validate_deepep_v2_dispatch_token_budget(args)
def test_disabled_chunking_uses_max_prefill_tokens(self):
for disabled in (None, 0, -1):
@@ -2342,7 +2385,7 @@ class TestDeepEPv2Args(CustomTestCase):
)
with envs.SGLANG_DEEPEP_V2_NUM_MAX_DISPATCH_TOKENS_PER_RANK.override(128):
with self.assertRaisesRegex(ValueError, "required=1024"):
args._validate_deepep_v2_dispatch_token_budget()
validate_deepep_v2_dispatch_token_budget(args)
def test_decode_role_skips_prefill_capacity(self):
args = self._args(
@@ -2352,7 +2395,7 @@ class TestDeepEPv2Args(CustomTestCase):
dp_size=1,
)
with envs.SGLANG_DEEPEP_V2_NUM_MAX_DISPATCH_TOKENS_PER_RANK.override(128):
args._validate_deepep_v2_dispatch_token_budget()
validate_deepep_v2_dispatch_token_budget(args)
def test_decode_graph_capacity_boundaries(self):
for max_bs, raises in ((128, False), (129, True)):
@@ -2364,9 +2407,9 @@ class TestDeepEPv2Args(CustomTestCase):
with envs.SGLANG_DEEPEP_V2_NUM_MAX_DISPATCH_TOKENS_PER_RANK.override(128):
if raises:
with self.assertRaisesRegex(ValueError, "decode CUDA graph"):
args._validate_deepep_v2_dispatch_token_budget()
validate_deepep_v2_dispatch_token_budget(args)
else:
args._validate_deepep_v2_dispatch_token_budget()
validate_deepep_v2_dispatch_token_budget(args)
def test_dp_attention_divides_max_running_requests_per_rank(self):
args = self._args(
@@ -2377,7 +2420,7 @@ class TestDeepEPv2Args(CustomTestCase):
enable_dp_attention=True,
)
with envs.SGLANG_DEEPEP_V2_NUM_MAX_DISPATCH_TOKENS_PER_RANK.override(128):
args._validate_deepep_v2_dispatch_token_budget()
validate_deepep_v2_dispatch_token_budget(args)
def test_tp_only_max_running_requests_is_not_divided(self):
args = self._args(
@@ -2389,7 +2432,7 @@ class TestDeepEPv2Args(CustomTestCase):
)
with envs.SGLANG_DEEPEP_V2_NUM_MAX_DISPATCH_TOKENS_PER_RANK.override(128):
with self.assertRaisesRegex(ValueError, "decode CUDA graph"):
args._validate_deepep_v2_dispatch_token_budget()
validate_deepep_v2_dispatch_token_budget(args)
def test_memory_derived_eager_pool_remains_runtime_validated(self):
args = self._args(
@@ -2398,7 +2441,7 @@ class TestDeepEPv2Args(CustomTestCase):
)
args.cuda_graph_config.decode.backend = Backend.DISABLED
with envs.SGLANG_DEEPEP_V2_NUM_MAX_DISPATCH_TOKENS_PER_RANK.override(1):
args._validate_deepep_v2_dispatch_token_budget()
validate_deepep_v2_dispatch_token_budget(args)
def test_speculative_decode_width_is_included(self):
args = self._args(
@@ -2411,7 +2454,7 @@ class TestDeepEPv2Args(CustomTestCase):
)
with envs.SGLANG_DEEPEP_V2_NUM_MAX_DISPATCH_TOKENS_PER_RANK.override(128):
with self.assertRaisesRegex(ValueError, "tokens/request=8"):
args._validate_deepep_v2_dispatch_token_budget()
validate_deepep_v2_dispatch_token_budget(args)
def test_adaptive_speculative_uses_widest_candidate(self):
args = self._args(
@@ -2430,7 +2473,7 @@ class TestDeepEPv2Args(CustomTestCase):
):
with envs.SGLANG_DEEPEP_V2_NUM_MAX_DISPATCH_TOKENS_PER_RANK.override(128):
with self.assertRaisesRegex(ValueError, "tokens/request=16"):
args._validate_deepep_v2_dispatch_token_budget()
validate_deepep_v2_dispatch_token_budget(args)
def test_prefill_role_skips_decode_capacity(self):
args = self._args(
@@ -2439,7 +2482,7 @@ class TestDeepEPv2Args(CustomTestCase):
max_running_requests=8192,
)
with envs.SGLANG_DEEPEP_V2_NUM_MAX_DISPATCH_TOKENS_PER_RANK.override(128):
args._validate_deepep_v2_dispatch_token_budget()
validate_deepep_v2_dispatch_token_budget(args)
def test_other_backend_skips_capacity_validation(self):
args = self._args(
@@ -2448,13 +2491,13 @@ class TestDeepEPv2Args(CustomTestCase):
max_running_requests=4096,
)
with envs.SGLANG_DEEPEP_V2_NUM_MAX_DISPATCH_TOKENS_PER_RANK.override(1):
args._validate_deepep_v2_dispatch_token_budget()
validate_deepep_v2_dispatch_token_budget(args)
def test_capacity_validation_uses_resolved_backend(self):
args = self._args(chunked_prefill_size=4096)
args._resolved_overrides = [("test", {"moe_a2a_backend": "deepep"})]
with envs.SGLANG_DEEPEP_V2_NUM_MAX_DISPATCH_TOKENS_PER_RANK.override(1):
args._validate_deepep_v2_dispatch_token_budget()
validate_deepep_v2_dispatch_token_budget(args)
class TestHandleCrashDumpEnv(CustomTestCase):
@@ -2474,7 +2517,7 @@ class TestHandleCrashDumpEnv(CustomTestCase):
for key in self._COREDUMP_ENV_KEYS:
if key not in (preset_env or {}):
os.environ.pop(key, None)
ServerArgs._handle_crash_dump_env(server_args)
handle_crash_dump_env(server_args)
def test_creates_coredump_dir_when_auto_set(self):
with tempfile.TemporaryDirectory() as tmp:
@@ -2504,7 +2547,7 @@ class TestGrpcServerArgs(CustomTestCase):
alongside HTTP; --smg-grpc-mode (and the deprecated --grpc-mode) select the
legacy SMG server. Worker-threads / max-prefill-tokens are env-only knobs.
The gRPC setup lives in ServerArgs._handle_deprecated_args, which
The gRPC setup lives in `serving_hook.handle_deprecated_args`, which
__post_init__ skips for dummy models, so these tests build a dummy
ServerArgs and invoke that handler directly (mirroring the real flow for a
concrete model path).
@@ -2516,20 +2559,20 @@ class TestGrpcServerArgs(CustomTestCase):
def test_http_only_high_port_does_not_derive_grpc_port(self):
sa = self._args(port=56000)
sa._handle_deprecated_args()
handle_deprecated_args(sa)
self.assertIsNone(resolution_result(sa, "grpc_port"))
def test_grpc_port_enables_native_and_env_knobs(self):
sa = self._args(grpc_port=50051)
with envs.SGLANG_GRPC_WORKER_THREADS.override(8):
sa._handle_deprecated_args()
handle_deprecated_args(sa)
self.assertEqual(resolution_result(sa, "grpc_port"), 50051)
self.assertEqual(resolution_result(sa, "grpc_worker_threads"), 8)
def test_env_grpc_port_enables_native(self):
sa = self._args(port=30000)
with envs.SGLANG_GRPC_PORT.override(45000):
sa._handle_deprecated_args()
handle_deprecated_args(sa)
self.assertEqual(resolution_result(sa, "grpc_port"), 45000)
@staticmethod
@@ -2602,17 +2645,17 @@ class TestGrpcServerArgs(CustomTestCase):
def test_sidecar_requires_native_grpc(self):
sa = self._args(sidecar="example.sidecar")
with self.assertRaisesRegex(ValueError, "requires --grpc-port"):
sa._handle_deprecated_args()
handle_deprecated_args(sa)
def test_sidecar_rejects_legacy_grpc(self):
sa = self._args(sidecar="example.sidecar", smg_grpc_mode=True)
with self.assertRaisesRegex(ValueError, "native gRPC server"):
sa._handle_deprecated_args()
handle_deprecated_args(sa)
def test_sidecar_rejects_empty_value(self):
sa = self._args(sidecar="", grpc_port=50051)
with self.assertRaisesRegex(ValueError, "must not be empty"):
sa._handle_deprecated_args()
handle_deprecated_args(sa)
def test_sidecar_sets_endpoint_env_before_import_and_calls_main(self):
main = MagicMock()
@@ -2666,37 +2709,37 @@ class TestGrpcServerArgs(CustomTestCase):
def test_legacy_smg_derives_grpc_port_from_http_port(self):
sa = self._args(port=30000, smg_grpc_mode=True)
sa._handle_deprecated_args()
handle_deprecated_args(sa)
self.assertEqual(resolution_result(sa, "grpc_port"), 40000)
def test_grpc_mode_is_deprecated_alias_for_smg_grpc_mode(self):
sa = self._args(grpc_mode=True)
with self.assertLogs(serving_hook.logger, level="WARNING") as cm:
sa._handle_deprecated_args()
handle_deprecated_args(sa)
self.assertTrue(resolution_result(sa, "smg_grpc_mode"))
self.assertTrue(any("--grpc-mode is deprecated" in line for line in cm.output))
def test_legacy_smg_takes_precedence_over_grpc_port(self):
sa = self._args(grpc_port=50051, smg_grpc_mode=True)
sa._handle_deprecated_args()
handle_deprecated_args(sa)
self.assertTrue(resolution_result(sa, "smg_grpc_mode"))
self.assertEqual(resolution_result(sa, "grpc_port"), 50051)
def test_native_grpc_rejects_multi_tokenizer(self):
sa = self._args(grpc_port=40000, tokenizer_worker_num=2)
with self.assertRaises(ValueError):
sa._handle_deprecated_args()
handle_deprecated_args(sa)
def test_native_grpc_rejects_http_auth(self):
sa = self._args(grpc_port=40000, api_key="secret")
with self.assertRaises(ValueError):
sa._handle_deprecated_args()
handle_deprecated_args(sa)
def test_invalid_grpc_worker_threads_rejected(self):
sa = self._args(grpc_port=40000)
with envs.SGLANG_GRPC_WORKER_THREADS.override(0):
with self.assertRaises(ValueError):
sa._handle_deprecated_args()
handle_deprecated_args(sa)
def test_start_server_call_site_matches_native_signature(self):
"""Regression for the startup blocker: the native start_server binding
@@ -2764,19 +2807,19 @@ class TestTwoBatchOverlapBackend(CustomTestCase):
def test_no_a2a_without_dp_attention_raises(self):
args = self._args(enable_dp_attention=False)
with self.assertRaisesRegex(ValueError, "enable-dp-attention"):
args._check_two_batch_overlap()
check_two_batch_overlap(args)
def test_no_a2a_with_dp_attention_ok(self):
# DP TBO path is valid: --enable-dp-attention + --enable-two-batch-overlap
# with a2a backend 'none' must NOT raise (no SGLANG_ENABLE_DP_TBO needed).
args = self._args(enable_dp_attention=True)
args._check_two_batch_overlap()
check_two_batch_overlap(args)
def test_ep_a2a_backend_ok_without_dp_attention(self):
# EP a2a path (e.g. deepep) overlaps dispatch/combine; the guard does not
# require dp-attention there.
args = self._args(moe_a2a_backend="deepep", enable_dp_attention=False)
args._check_two_batch_overlap()
check_two_batch_overlap(args)
class TestDcpKvEventContract(CustomTestCase):