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

[Config] Round 6.5: a namespace declares what it derives, next to what it derives it from (#38113)

Browse Source 
Fifth of five; stacked on #38049. The split gave every namespace a file, but
only for the half an operator types. This is the other half.

## The parallel quotients are declared, not written out

`attn_tp_size` and its five siblings were sixty lines of near-identical
properties in the runtime context, a file away from the leaves they are
quotients of, so reading `parallel.py` told you what you could set and nothing
about what that decides.

They are declared in `Parallel` now, in the same class as those leaves. They
carry no annotation, so they are not dataclass fields and
`collect_input_fields` never puts them on the record -- the same mechanism that
already keeps `_NS_PATH` off it. That is the right exclusion: a quotient has no
operator input to preserve, and the record is what crosses a process boundary,
where a stamped width is one an elastic scale-up will not refresh.

## A quotient is a value in the bag, like every other derived one

`_derived_width` answered from a stamp or, failing that, a live process group.
The group read could never disagree with the stamp:

- `initialize_model_parallel` stamps all six as its last statement,
  unconditionally;
- an elastic scale-up restamps `attn_dp_size` through
  `update_dp_attention_post_scale` -- the comment claiming it does *not* was
  wrong;
- no hardware backend builds groups of its own;
- `multimodal_gen`, which has its own `initialize_model_parallel` and does not
  stamp, never reads a quotient.

So a built group was always already stamped, and the group read goes -- and with
it the last reason for a quotient to be resolved on every read.

Every input to `derive_parallel_widths` is a record field. `dcp_enabled` is
`decode_context_parallel_size > 1`, not a fact about a built group; it was
spelled `_DCP is not None`, which is a longer way to say the same thing. So the
six are fixed once the configuration is fixed -- the same test every other
`Derived(fn=...)` in this PR passes. They are declared the same way and computed
the same way: once, at publish, into ordinary bag leaves.

What remains is override -> stamp -> published leaf. The stamp stays above the
leaf because an elastic scale-up restamps `attn_dp_size`; the override stays on
top because that is how a test names a width.

## One answer for the config-derived predicates

`enable_mamba_extra_buffer` and its lazy variant, `is_ep_joiner`,
`is_ep_scale_joiner`, `is_startup_weight_load_overlap`: each existed as a
`ServerArgs` member for the resolution pipeline and, for most of them, again as
a `runtime_context` function for readers after publish. Three places to keep
saying the same thing.

A `Derived(fn=...)` is a pure function of the published configuration, so
`publish` computes it once and stores it as an ordinary bag leaf -- a plain
attribute load, which is what a read inside compiled model code needs. The
function is handed the whole resolved config rather than the bag it lands in,
because a derivation is free to span namespaces and the mamba one does: it
reads `memory.disable_radix_cache` alongside its own `exec.mamba` strategy,
which is why it could never have been a method on either bag.

The pre-publish helpers stay -- resolution needs the predicate before there is
a bag to read -- and three readers keep them, because they run before their own
process publishes: `initialize_dp_attention`, which the weight-cache daemon
calls while building its groups thirty lines before its `publish`, and
`PortArgs.init_new`, a factory handed the record that already reads eighteen
other fields off it.

## Notes for a reviewer

**Overriding a leaf does not move its quotient.** `override(tp_size=2)` leaves
`attn_tp_size` where the published config put it, because nothing is recomputed
on read. A test states a topology by publishing a config -- which is what a
real process does -- or by naming the width it wants, `override(attn_tp_size=2)`.
Six tests say it that way now. This is the price of having one answer computed
once, and it is the same price every other derived value in the config already
carries.

A caller that reads a quotient without publishing or overriding now gets an
explicit error naming the field, instead of a default that an uninitialised
group happened to supply. One fixture was in that state --
`TestMlaWriteDoorsUnderDcp` built a bare pool and asked whether DCP was on --
and it publishes a config now, which is what the process it stands in for
does.

Eighteen sites read these predicates without calling them. That is correct --
they are properties -- but it is worth saying they were checked, because a
census that assumes otherwise reports eighteen always-true conditions.

## The skill that documents this subsystem is updated with it

`.claude/rules/modify-component-must-read.md` points at
`.claude/skills/sglang-runtime-context/SKILL.md` before anyone touches these
files, so a stale sentence there is a wrong instruction rather than a stale
note. Four of its load-bearing statements stopped being true across this series
and are corrected here: `NS(...)` is no longer how a field states its namespace
(the declaring class is); the DCP degrade rule is gone, because the quotients
are not live reads; `mamba_extra_buffer_enabled()` and the other predicate
functions it named as the shape to copy no longer exist; and the
namespace-coverage ratchet is described in terms of the marker. The docstring of
`test_server_args_namespaces.py` said the same thing and is fixed too.

The consequence a test author actually trips over is stated there as well:
overriding a leaf no longer moves its quotient, so a topology is stated by
publishing a config or by naming the width.

## Verification

A full registered-unit sweep (648 files) against this stack's merge-base:
19 failures on both sides, the same 19 -- AMD `gfx950`, `modelopt`,
`cuda_vmm`, `weight_checker` and friends, none of them config. The narrower 139-file config sweep used earlier in this series
does not contain the files this change reaches -- `test_kv_index_translator`
never names `get_parallel()`, it constructs an object that does -- which is why
the baseline differential over everything is what is quoted here.
This commit is contained in:
Cheng Wan
2026-09-06 21:44:24 -07:00
committed by GitHub
parent b99175dc7d
commit aaf9a95763
56 changed files with 1118 additions and 379 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test/registered/unit/constrained/test_grammar_manager.py
+68 -14
View File
@@ -26,10 +26,13 @@ from sglang.srt.constrained.base_grammar_backend import (
from sglang.srt.constrained.grammar_manager import GrammarManager
from sglang.srt.constrained.reasoner_grammar_backend import ReasonerGrammarObject
from sglang.srt.distributed.communication_tags import P2PTag
from sglang.srt.runtime_context import get_context, publish, reset_context
from sglang.srt.sampling.sampling_params import (
REQUEST_REASONING_END_TOKEN_IDS_KEY,
)
from sglang.srt.server_args import ServerArgs
from sglang.test.ci.ci_register import register_cpu_ci
from sglang.test.test_utils import enter_override
register_cpu_ci(2.0, "base-a-test-cpu")
@@ -38,13 +41,24 @@ register_cpu_ci(est_time=5, suite="stage-b-test-cpu-intel")
def _make_scheduler(grammar_backend_name="none", skip_tokenizer=False):
"""Create a mock scheduler with necessary attributes."""
"""Create a mock scheduler with necessary attributes.
The grammar manager reads its config from the bags, so the settings that
used to be hung off the mock are published instead. The caller resets the
context; every test here goes through `_GrammarFixture`.
"""
reset_context()
server_args = ServerArgs(
model_path="dummy",
grammar_backend=grammar_backend_name,
skip_tokenizer_init=skip_tokenizer,
reasoning_parser=None,
constrained_json_whitespace_pattern=None,
constrained_json_disable_any_whitespace=False,
)
publish(server_args, role="scheduler")
scheduler = MagicMock()
scheduler.server_args.grammar_backend = grammar_backend_name
scheduler.server_args.skip_tokenizer_init = skip_tokenizer
scheduler.server_args.reasoning_parser = None
scheduler.server_args.constrained_json_whitespace_pattern = None
scheduler.server_args.constrained_json_disable_any_whitespace = False
scheduler.server_args = server_args
scheduler.model_config.request_selectable_think_end_id_sequences = None
# Distributed group mocks
@@ -84,13 +98,19 @@ def _make_req(
class TestGrammarManagerInit(unittest.TestCase):
def setUp(self):
reset_context()
self.addCleanup(reset_context)
"""Test GrammarManager initialization."""
@patch("sglang.srt.constrained.grammar_manager.create_grammar_backend")
def test_init_with_backend(self, mock_create):
mock_create.return_value = MagicMock(spec=BaseGrammarBackend)
scheduler = _make_scheduler("xgrammar")
scheduler.server_args.skip_tokenizer_init = False
enter_override(
self, get_context().override_server_args(skip_tokenizer_init=False)
)
mgr = GrammarManager(scheduler)
self.assertIsNotNone(mgr.grammar_backend)
@@ -114,7 +134,9 @@ class TestGrammarManagerInit(unittest.TestCase):
mock_backend = MagicMock(spec=BaseGrammarBackend)
mock_create.return_value = mock_backend
scheduler = _make_scheduler()
scheduler.server_args.skip_tokenizer_init = False
enter_override(
self, get_context().override_server_args(skip_tokenizer_init=False)
)
mgr = GrammarManager(scheduler)
mgr.clear()
@@ -129,11 +151,17 @@ class TestGrammarManagerInit(unittest.TestCase):
class TestProcessReqWithGrammar(unittest.TestCase):
def setUp(self):
reset_context()
self.addCleanup(reset_context)
"""Test process_req_with_grammar dispatch and caching."""
def _make_mgr(self):
scheduler = _make_scheduler()
scheduler.server_args.skip_tokenizer_init = True
enter_override(
self, get_context().override_server_args(skip_tokenizer_init=True)
)
mgr = GrammarManager(scheduler)
mgr.grammar_backend = MagicMock(spec=BaseGrammarBackend)
return mgr
@@ -240,7 +268,9 @@ class TestProcessReqWithGrammar(unittest.TestCase):
def test_no_backend_aborts(self):
"""No grammar backend should abort request."""
scheduler = _make_scheduler()
scheduler.server_args.skip_tokenizer_init = True
enter_override(
self, get_context().override_server_args(skip_tokenizer_init=True)
)
mgr = GrammarManager(scheduler)
mgr.grammar_backend = None
@@ -357,11 +387,17 @@ class TestProcessReqWithGrammar(unittest.TestCase):
class TestAbortRequests(unittest.TestCase):
def setUp(self):
reset_context()
self.addCleanup(reset_context)
"""Test abort_requests handling."""
def _make_mgr_with_queue(self):
scheduler = _make_scheduler()
scheduler.server_args.skip_tokenizer_init = True
enter_override(
self, get_context().override_server_args(skip_tokenizer_init=True)
)
mgr = GrammarManager(scheduler)
mgr.grammar_backend = MagicMock(spec=BaseGrammarBackend)
return mgr
@@ -435,11 +471,17 @@ class TestAbortRequests(unittest.TestCase):
class TestGetReadyGrammarRequests(unittest.TestCase):
def setUp(self):
reset_context()
self.addCleanup(reset_context)
"""Test get_ready_grammar_requests polling and result handling."""
def _make_mgr(self):
scheduler = _make_scheduler()
scheduler.server_args.skip_tokenizer_init = True
enter_override(
self, get_context().override_server_args(skip_tokenizer_init=True)
)
mgr = GrammarManager(scheduler)
mgr.grammar_backend = MagicMock(spec=BaseGrammarBackend)
# Use very short poll interval for tests
@@ -710,11 +752,17 @@ class _FakePPGroup:
class TestGrammarManagerPPSync(unittest.TestCase):
def setUp(self):
reset_context()
self.addCleanup(reset_context)
"""Test PP synchronization of grammar ready/failed indexes."""
def _make_mgr_for_pp(self, pp_rank, pp_size, pp_group):
scheduler = _make_scheduler()
scheduler.server_args.skip_tokenizer_init = True
enter_override(
self, get_context().override_server_args(skip_tokenizer_init=True)
)
scheduler.ps.pp_rank = pp_rank
scheduler.ps.pp_size = pp_size
scheduler.pp_group = pp_group
@@ -772,11 +820,17 @@ class TestGrammarManagerPPSync(unittest.TestCase):
class TestStrictReasoningPaths(unittest.TestCase):
def setUp(self):
reset_context()
self.addCleanup(reset_context)
"""Test _enable_strict_thinking code paths in GrammarManager."""
def _make_mgr(self):
scheduler = _make_scheduler()
scheduler.server_args.skip_tokenizer_init = True
enter_override(
self, get_context().override_server_args(skip_tokenizer_init=True)
)
mgr = GrammarManager(scheduler)
mgr.grammar_backend = MagicMock(spec=BaseGrammarBackend)
mgr._enable_strict_thinking = True
test/registered/unit/entrypoints/openai/test_serving_chat.py
+27 -9
View File
@@ -6,7 +6,7 @@ or
python -m unittest discover -s tests -p "test_*unit.py" -v
"""
from sglang.test.test_utils import maybe_stub_sgl_kernel
from sglang.test.test_utils import enter_override, maybe_stub_sgl_kernel
maybe_stub_sgl_kernel() # must precede any import that pulls in sgl_kernel
@@ -3318,8 +3318,12 @@ class ServingChatTestCase(unittest.TestCase):
self.assertIsNone(response.sglext)
def test_non_streaming_ids_server_default_enables_flag(self):
self.tm.server_args.return_input_ids = True
self.tm.server_args.return_output_ids = True
enter_override(
self,
get_context().override_server_args(
return_input_ids=True, return_output_ids=True
),
)
req = ChatCompletionRequest(
model="x", messages=[{"role": "user", "content": "Hi?"}]
)
@@ -3371,7 +3375,12 @@ class ServingChatTestCase(unittest.TestCase):
):
"""Stream chunks with incremental or cumulative output_ids;
return parsed sglext chunks, or raw SSE strings when return_raw."""
self.tm.server_args.incremental_streaming_output = incremental
enter_override(
self,
get_context().override_server_args(
incremental_streaming_output=incremental
),
)
if framed:
self.fastapi_request.headers["x-sglext-ids-framed"] = "1"
@@ -3495,7 +3504,12 @@ class ServingChatTestCase(unittest.TestCase):
self, normal_chunks, abort_output_ids, incremental, abort_completion_tokens
):
"""Stream normal chunks then a graceful-abort chunk; return parsed sglext chunks."""
self.tm.server_args.incremental_streaming_output = incremental
enter_override(
self,
get_context().override_server_args(
incremental_streaming_output=incremental
),
)
async def _mock_generate():
generated = 0
@@ -3661,14 +3675,18 @@ class ServingChatTestCase(unittest.TestCase):
def test_continuous_usage_reports_cached_tokens(self):
"""continuous_usage_stats chunks include cached tokens when cache reporting is on."""
self.enterContext(get_context().override_server_args(enable_cache_report=True))
enter_override(
self, get_context().override_server_args(enable_cache_report=True)
)
usages = self._collect_continuous_usage(cached_tokens=6)
self.assertTrue(usages, "continuous_usage_stats attached no usage")
self.assertEqual(usages[0]["prompt_tokens_details"]["cached_tokens"], 6)
def test_continuous_usage_omits_cached_tokens_when_report_disabled(self):
"""With cache reporting off, continuous_usage_stats must not leak cached tokens."""
self.enterContext(get_context().override_server_args(enable_cache_report=False))
enter_override(
self, get_context().override_server_args(enable_cache_report=False)
)
usages = self._collect_continuous_usage(cached_tokens=6)
self.assertTrue(usages, "continuous_usage_stats attached no usage")
self.assertIsNone(usages[0].get("prompt_tokens_details"))
@@ -3684,8 +3702,8 @@ class ServingChatTestCase(unittest.TestCase):
Regression test for https://github.com/sgl-project/sglang/issues/22510.
"""
# Enable incremental_streaming_output on the mock
self.enterContext(
get_context().override_server_args(incremental_streaming_output=True)
enter_override(
self, get_context().override_server_args(incremental_streaming_output=True)
)
# Simulate incremental streaming: each yield has ONLY the new text (delta),
test/registered/unit/hardware_backend/mlx/test_tp_worker_routing.py
+9 -5
View File
@@ -226,14 +226,18 @@ class TestMlxExtendRouting(CustomTestCase):
MlxModelRunnerStub,
)
from sglang.srt.hardware_backend.mlx.tp_worker import MlxTpModelWorker
from sglang.srt.runtime_context import get_context
worker = MlxTpModelWorker.__new__(MlxTpModelWorker)
worker.server_args = SimpleNamespace(is_startup_weight_load_overlap=True)
with self.assertRaisesRegex(ValueError, "CUDA only"):
MlxModelRunnerStub.validate_startup_weight_load_mode(worker.server_args)
with self.assertRaisesRegex(ValueError, "CUDA only"):
worker._init_model_runner()
# The guard reads `get_model().is_startup_weight_load_overlap`, which is
# derived from `startup_weight_load_mode`. Stating it on a `server_args`
# of the worker's own no longer reaches it.
with get_context().override_server_args(startup_weight_load_mode="overlap"):
with self.assertRaisesRegex(ValueError, "CUDA only"):
MlxModelRunnerStub.validate_startup_weight_load_mode()
with self.assertRaisesRegex(ValueError, "CUDA only"):
worker._init_model_runner()
# ---------- the shared decision helper ----------
# The helper takes no seq_len: length cannot distinguish a 1-token
test/registered/unit/layers/moe/test_qwen35_flashinfer_fusion.py
+18 -5
View File
@@ -14,8 +14,14 @@ from sglang.srt.layers.moe.qwen35_flashinfer_fusion import (
is_supported_forward_mode,
resolve_max_m,
)
from sglang.srt.model_executor.cuda_graph_config import (
CudaGraphConfig,
PhaseConfig,
)
from sglang.srt.model_executor.forward_batch_info import ForwardMode
from sglang.srt.models.qwen3_5_text import Qwen3_5ForCausalLM
from sglang.srt.runtime_context import publish, reset_context
from sglang.srt.server_args import ServerArgs
from sglang.test.ci.ci_register import register_cpu_ci
register_cpu_ci(est_time=9, suite="base-a-test-cpu")
@@ -65,12 +71,19 @@ def test_supported_forward_modes(forward_mode, expected):
return_value=8192,
)
def test_framework_capacity_is_maximum_of_all_sources(_cutedsl_moe_max_num_tokens):
graph = SimpleNamespace(
decode=SimpleNamespace(max_bs=512, bs=[1, 64, 256]),
prefill=SimpleNamespace(max_bs=4096, bs=[1024, 2048, 4096]),
# The graph bounds are a bag leaf, so the test states them by publishing.
reset_context()
publish(
ServerArgs(
model_path="dummy",
cuda_graph_config=CudaGraphConfig(
decode=PhaseConfig(max_bs=512, bs=[1, 64, 256]),
prefill=PhaseConfig(max_bs=4096, bs=[1024, 2048, 4096]),
),
),
role="test",
)
server_args = SimpleNamespace(cuda_graph_config=graph)
runner = SimpleNamespace(server_args=server_args, max_running_requests=2048)
runner = SimpleNamespace(server_args=SimpleNamespace(), max_running_requests=2048)
assert resolve_max_m(runner) == 8192
test/registered/unit/layers/test_minicpm_sparse_cache.py
+12
View File
@@ -16,6 +16,8 @@ from sglang.srt.managers.scheduler_components.pool_stats_observer import (
)
from sglang.srt.mem_cache.allocator import BaseTokenToKVPoolAllocator
from sglang.srt.mem_cache.memory_pool import ReqToTokenPool
from sglang.srt.runtime_context import publish, reset_context
from sglang.srt.server_args import ServerArgs
from sglang.srt.session.streaming_session import SessionSlot, StreamingSession
from sglang.test.ci.ci_register import register_cpu_ci
@@ -91,6 +93,16 @@ def alloc_extend(pool, req_pool_idx: int, seq_len: int):
)
def setup_function(_):
# The cache reads its parallel topology from the context.
reset_context()
publish(ServerArgs(model_path="dummy"), role="test")
def teardown_function(_):
reset_context()
def test_extend_allocates_at_sparse_boundaries():
pool, _, req_pool_idx, allocator = make_pool_and_req()
cache = pool._aux_cache
test/registered/unit/managers/test_generation_auxiliary_output.py
+17 -1
View File
@@ -15,6 +15,8 @@ from sglang.srt.managers.scheduler_pp_mixin import PPBatchMetadata
from sglang.srt.managers.utils import GenerationBatchResult
from sglang.srt.model_executor.forward_batch_info import PPProxyTensors
from sglang.srt.model_executor.model_runner import ModelRunner
from sglang.srt.runtime_context import publish, reset_context
from sglang.srt.server_args import ServerArgs
from sglang.srt.speculative.spec_info import SpeculativeAlgorithm
from sglang.test.ci.ci_register import register_cpu_ci
@@ -69,13 +71,27 @@ def _model_runner_for_sampling_path(
spec_algorithm=SpeculativeAlgorithm.NONE,
dllm_algorithm=None,
):
# `supports_sampling_observer` reads the dLLM algorithm from the bags, so
# the path is stated by publishing it rather than by standing one in.
reset_context()
publish(ServerArgs(model_path="dummy", dllm_algorithm=dllm_algorithm), role="test")
runner = object.__new__(ModelRunner)
runner.server_args = SimpleNamespace(dllm_algorithm=dllm_algorithm)
runner.server_args = SimpleNamespace()
runner.spec_algorithm = spec_algorithm
runner._sampling_observer = None
return runner
def setup_function(_):
# The code under test reads its config from the bags.
reset_context()
publish(ServerArgs(model_path="dummy"), role="test")
def teardown_function(_):
reset_context()
def test_auxiliary_output_releases_device_holder_after_copy():
device_output = DeviceOutput(torch.tensor([1.0, 2.0]))
logits_output = LogitsProcessorOutput(
test/registered/unit/mem_cache/test_decode_radix_lock_ref.py
+22 -2
View File
@@ -20,7 +20,10 @@ Usage:
python -m pytest test/registered/unit/mem_cache/test_decode_radix_lock_ref.py -v
"""
from sglang.srt.runtime_context import get_context, publish, reset_context
from sglang.srt.server_args import ServerArgs
from sglang.test.ci.ci_register import register_cpu_ci
from sglang.test.test_utils import enter_override
register_cpu_ci(est_time=11, suite="base-a-test-cpu")
@@ -97,14 +100,26 @@ def _make_req(fill_ids, req_pool_idx=0, cache_protected_len=0, last_node=None):
class TestDecodeLockRefScenarios(unittest.TestCase):
def setUp(self):
# The decode queue reads its config from the bags.
reset_context()
self.addCleanup(reset_context)
publish(ServerArgs(model_path="dummy"), role="scheduler")
"""Test lock_ref balance across decode transfer scenarios."""
def test_swa_tail_len_keeps_page_aligned_matchable_window(self):
enter_override(
self,
get_context().override_server_args(
disaggregation_decode_enable_radix_cache=True
),
)
queue = DecodePreallocQueue.__new__(DecodePreallocQueue)
queue._uses_swa_tail_prealloc = MagicMock(return_value=True)
queue.scheduler = SimpleNamespace(
sliding_window_size=127,
server_args=SimpleNamespace(disaggregation_decode_enable_radix_cache=True),
server_args=SimpleNamespace(),
)
queue.token_to_kv_pool_allocator = MagicMock(page_size=64)
@@ -420,7 +435,12 @@ class TestDecodeLockRefScenarios(unittest.TestCase):
running_batch = MagicMock()
running_batch.reqs = []
server_args = MagicMock()
server_args.disaggregation_decode_enable_radix_cache = True
enter_override(
self,
get_context().override_server_args(
disaggregation_decode_enable_radix_cache=True
),
)
scheduler = MagicMock()
scheduler.running_batch = running_batch
scheduler.server_args = server_args
test/registered/unit/mem_cache/test_full_loc_fast_path.py
+11
View File
@@ -280,6 +280,17 @@ class TestMlaWriteDoorsUnderDcp(unittest.TestCase):
already collapsed -- so there is no single correct translation and refusing
is the contract."""
def setUp(self):
# `set_kv_buffer` asks the parallel context whether DCP is in play, and
# that is a value the configuration decides at publish -- so a case
# here states its topology the way a process does, by publishing one.
from sglang.srt.runtime_context import publish, reset_context
from sglang.srt.server_args import ServerArgs
reset_context()
self.addCleanup(reset_context)
publish(ServerArgs(model_path="dummy"), role="test")
def _bare_mla_pool(self):
from sglang.srt.mem_cache.memory_pool import MLATokenToKVPool
test/registered/unit/mem_cache/test_kv_index_translator.py
+44
View File
@@ -17,6 +17,8 @@ CPU-only: these exercise the builder's pure-torch reference path, not the
Triton kernel.
"""
from sglang.srt.runtime_context import publish, reset_context
from sglang.srt.server_args import ServerArgs
from sglang.test.ci.ci_register import register_cpu_ci
register_cpu_ci(est_time=10, suite="base-a-test-cpu")
@@ -115,6 +117,12 @@ def _reference_table(req_to_token, req_pool_indices, seq_lens, v2p, mult, ps, wi
class TestPassthrough(unittest.TestCase):
def setUp(self):
# The code under test reads its config from the bags.
reset_context()
self.addCleanup(reset_context)
publish(ServerArgs(model_path="dummy"), role="tokenizer")
def test_non_unified_returns_same_objects(self):
"""Strict passthrough: no copy, no branch. Any tensor op on the
non-unified path breaks byte-identity for every static-pool server."""
@@ -160,6 +168,12 @@ def _alloc_and_fill(allocator, ps, lens):
class TestReadTableBuild(unittest.TestCase):
def setUp(self):
# The code under test reads its config from the bags.
reset_context()
self.addCleanup(reset_context)
publish(ServerArgs(model_path="dummy"), role="tokenizer")
def test_read_table_matches_reference_across_multipliers(self):
"""Both read tables must equal the independent per-element derivation,
across page sizes and both multiplier regimes (MLA=1, MHA=2L); the swa
@@ -274,6 +288,12 @@ class TestBuildInto(unittest.TestCase):
FULL-side read-table entries -- the trtllm_mla / flashmla consumption route
(their rows ARE the read table's rows)."""
def setUp(self):
# The code under test reads its config from the bags.
reset_context()
self.addCleanup(reset_context)
publish(ServerArgs(model_path="dummy"), role="tokenizer")
def test_prefix_filled_tail_sentinel_preserved_width_capped(self):
"""The -1 tail sentinel belongs to the backend, and a table padded
WIDER than the req_to_token page span (trtllm's LCM alignment) must be
@@ -339,6 +359,12 @@ class TestPoolOwnership(unittest.TestCase):
to address a buffer with only num_slots rows.
"""
def setUp(self):
# The code under test reads its config from the bags.
reset_context()
self.addCleanup(reset_context)
publish(ServerArgs(model_path="dummy"), role="tokenizer")
def test_real_factory_bundle_satisfies_the_ownership_identity(self):
"""The guard rests on `allocator.get_kvcache() is token_to_kv_pool`, so
a factory returning a pool the allocator does not hold would silently
@@ -420,6 +446,12 @@ class TestPoolOwnership(unittest.TestCase):
class TestCaptureContract(unittest.TestCase):
def setUp(self):
# The code under test reads its config from the bags.
reset_context()
self.addCleanup(reset_context)
publish(ServerArgs(model_path="dummy"), role="tokenizer")
def test_caller_owned_table_is_returned_whole_and_filled_prefix_only(self):
ps = 4
allocator = _build_composite(ps)
@@ -503,6 +535,12 @@ class TestViewMemo(unittest.TestCase):
identity, so per-batch state stays out of the ForwardBatch while one
metadata build's many consumers still share one table build."""
def setUp(self):
# The code under test reads its config from the bags.
reset_context()
self.addCleanup(reset_context)
publish(ServerArgs(model_path="dummy"), role="tokenizer")
def _fb(self, allocator, ps, lens):
req_to_token, rows, seq_lens = _alloc_and_fill(allocator, ps, lens=lens)
fb = _FakeForwardBatch(
@@ -562,6 +600,12 @@ class TestWriteLoc(unittest.TestCase):
POINTWISE from the full-side values -- pads, slices, and fresh copies
included -- with no handover and no stored per-forward state."""
def setUp(self):
# The code under test reads its config from the bags.
reset_context()
self.addCleanup(reset_context)
publish(ServerArgs(model_path="dummy"), role="tokenizer")
def _built(self, ps=1, n=4):
allocator = _build_composite(ps)
req_to_token, rows, seq_lens = _alloc_and_fill(allocator, ps, lens=[max(n, 1)])
test/registered/unit/mem_cache/test_multi_ended_allocator.py
+68 -1
View File
@@ -44,7 +44,8 @@ from sglang.srt.mem_cache.unified_memory_pool import (
MLASubPoolSpec,
UnifiedKVPool,
)
from sglang.srt.runtime_context import get_parallel
from sglang.srt.runtime_context import get_parallel, publish, reset_context
from sglang.srt.server_args import ServerArgs
_DEV = "cpu"
@@ -103,6 +104,12 @@ class _RejectScalarIndexTensor:
class TestUnifiedKVPoolViews(unittest.TestCase):
def setUp(self):
# The code under test reads its config from the bags.
reset_context()
self.addCleanup(reset_context)
publish(ServerArgs(model_path="dummy"), role="tokenizer")
def test_min_slot_index_and_disjoint_bytes(self):
full = _make_mha_spec("full", "up", layer_num=4)
mamba = _make_mamba_spec("mamba", "down", layer_num=2)
@@ -173,6 +180,12 @@ class TestUnifiedKVPoolViews(unittest.TestCase):
class TestMultiEndedAllocator(unittest.TestCase):
def setUp(self):
# The code under test reads its config from the bags.
reset_context()
self.addCleanup(reset_context)
publish(ServerArgs(model_path="dummy"), role="tokenizer")
def _build_pair(self, n_full_slots=64, n_mamba_slots=16):
full = _make_mha_spec("full", "up", layer_num=2)
mamba = _make_mamba_spec("mamba", "down", layer_num=2)
@@ -526,6 +539,12 @@ class TestUnifiedSWATokenToKVPoolAllocator(unittest.TestCase):
"""The SWA composite: joint byte-budget, slot-conservation, `free_swa`
tombstone semantics, and divergent compaction of the two sub-pools."""
def setUp(self):
# The code under test reads its config from the bags.
reset_context()
self.addCleanup(reset_context)
publish(ServerArgs(model_path="dummy"), role="tokenizer")
def _build(
self,
n_full_slots=32,
@@ -891,6 +910,12 @@ class TestPagedMultiEndedAllocator(unittest.TestCase):
"""`MultiEndedAllocator(page_size=8)`: free-list, v2p/p2v and compaction are
page-granular, while the external API stays in token ids as at page_size 1."""
def setUp(self):
# The code under test reads its config from the bags.
reset_context()
self.addCleanup(reset_context)
publish(ServerArgs(model_path="dummy"), role="tokenizer")
PAGE_SIZE = 8
def _build(self, n_full_pages=16, n_swa_pages=8, full_layer_num=2, swa_layer_num=2):
@@ -1693,6 +1718,12 @@ class TestLazyCompaction(unittest.TestCase):
`_flush` only waits on `forward_stream` when one is passed, so these
allocators stay CPU-only."""
def setUp(self):
# The code under test reads its config from the bags.
reset_context()
self.addCleanup(reset_context)
publish(ServerArgs(model_path="dummy"), role="tokenizer")
def _make_full(self, *, lazy: bool, n_full_slots=64, n_mamba_slots=16):
full = _make_mha_spec("full", "up", layer_num=2)
mamba = _make_mamba_spec("mamba", "down", layer_num=2)
@@ -1969,6 +2000,12 @@ class TestO3FusedAllocBind(unittest.TestCase):
"""Fused take_physical_pages + bind_pages via `_alloc_bind_fast_or_slow`.
GPU-only: the fused kernel is Triton."""
def setUp(self):
# The code under test reads its config from the bags.
reset_context()
self.addCleanup(reset_context)
publish(ServerArgs(model_path="dummy"), role="tokenizer")
@classmethod
def setUpClass(cls):
if not torch.cuda.is_available():
@@ -2218,6 +2255,12 @@ class TestSWACompositeKernelIdSurface(unittest.TestCase):
`translate_kv_loc_for_kernel` / `full_v2p_page_table`, and every id must
follow `kernel_id(t) = v2p[t // ps] * (ps * mult) + t % ps`."""
def setUp(self):
# The code under test reads its config from the bags.
reset_context()
self.addCleanup(reset_context)
publish(ServerArgs(model_path="dummy"), role="tokenizer")
PS = 4
FULL_L = 4
SWA_L = 2
@@ -2328,6 +2371,12 @@ class TestPs64MLACompositeFeasibility(unittest.TestCase):
pages stress the sink-page floor, the per-layer-view tail pad and the
page-granular alloc at once."""
def setUp(self):
# The code under test reads its config from the bags.
reset_context()
self.addCleanup(reset_context)
publish(ServerArgs(model_path="dummy"), role="tokenizer")
PS = 64
LAYERS = 3
@@ -2424,6 +2473,12 @@ class TestChainFrontierWalk(unittest.TestCase):
"""N-pool chain walk: 2-pool byte-identity with the single-peer formulas,
transparent-middle skipping, and growth-side-neighbor credit routing."""
def setUp(self):
# The code under test reads its config from the bags.
reset_context()
self.addCleanup(reset_context)
publish(ServerArgs(model_path="dummy"), role="tokenizer")
def _build_pair(self):
full = _make_mha_spec("full", "up", layer_num=2)
mamba = _make_mamba_spec("mamba", "down", layer_num=2)
@@ -2552,6 +2607,12 @@ class TestFloatMultiEndedAllocator(unittest.TestCase):
larger-gap extension, boundary absorption with park-on-empty transparency,
and the on-demand movers `make_room` and `compact_holes`."""
def setUp(self):
# The code under test reads its config from the bags.
reset_context()
self.addCleanup(reset_context)
publish(ServerArgs(model_path="dummy"), role="tokenizer")
def _build_tri(self, n_state=8, n_float=32, n_full=32):
state = _make_mamba_spec("state", "up", layer_num=2)
fl = _make_mha_spec("swa", "float", layer_num=2)
@@ -2844,6 +2905,12 @@ class TestDcpWidening(unittest.TestCase):
"""`dcp_size > 1`: the alloc surface speaks a widened virtual id space while
the pool keeps storing one row per `dcp_size` logical ids."""
def setUp(self):
# The code under test reads its config from the bags.
reset_context()
self.addCleanup(reset_context)
publish(ServerArgs(model_path="dummy"), role="tokenizer")
@contextlib.contextmanager
def _dcp(self, dcp_size, dcp_rank=0):
"""The width comes from the parallel context, not a constructor
test/registered/unit/mem_cache/test_registry.py
+9 -8
View File
@@ -1,5 +1,6 @@
"""Unit tests for the radix-cache registry, routing, and selection chain."""
from sglang.srt.runtime_context import get_context
from sglang.test.ci.ci_register import register_cpu_ci
register_cpu_ci(est_time=11, suite="base-a-test-cpu")
@@ -16,7 +17,7 @@ from sglang.srt.mem_cache.registry import (
register_radix_cache_backend,
registered_radix_cache_backends,
)
from sglang.test.test_utils import CustomTestCase
from sglang.test.test_utils import CustomTestCase, enter_override
def _publish(testcase, **fields):
@@ -340,14 +341,14 @@ class TestDefaultRadixCacheFactory(CustomTestCase):
from sglang.srt.mem_cache.storage.umbp import umbp_direct_linker
ctx = _make_ctx(self)
object.__setattr__(
ctx.server_args, "enable_unified_cache_external_linker", True
# The factory reads the linker settings from the bags.
enter_override(
self,
get_context().override_server_args(
enable_unified_cache_external_linker=True,
unified_cache_external_linker_backend="mori",
),
)
object.__setattr__(
ctx.server_args, "unified_cache_external_linker_backend", "mori"
)
self.assertTrue(ctx.server_args.enable_unified_cache_external_linker)
self.assertEqual(ctx.server_args.unified_cache_external_linker_backend, "mori")
fake_components = MagicMock()
fake_components.ComponentType.FULL = "full"
fake_radix = MagicMock()
test/registered/unit/mem_cache/test_unified_byte_accounting.py
+20
View File
@@ -32,6 +32,8 @@ from test_multi_ended_allocator import (
)
from sglang.srt.mem_cache.allocator import unified_sub_pool as mea
from sglang.srt.runtime_context import publish, reset_context
from sglang.srt.server_args import ServerArgs
from sglang.test.ci.ci_register import register_cpu_ci
register_cpu_ci(est_time=10, suite="base-a-test-cpu")
@@ -51,6 +53,12 @@ def _paged_pair(lazy: bool):
class TestHealthyLifecycleReportsClean(unittest.TestCase):
def setUp(self):
# The code under test reads its config from the bags.
reset_context()
self.addCleanup(reset_context)
publish(ServerArgs(model_path="dummy"), role="tokenizer")
def test_lazy_end_pool_clean_through_free_and_flush(self):
full, _swa = _paged_pair(lazy=True)
self.assertEqual(full._byte_accounting_violations(), [])
@@ -68,6 +76,12 @@ class TestDriftReportsLoudly(unittest.TestCase):
passes every other test -- the pool still 'works', it just lies about
capacity."""
def setUp(self):
# The code under test reads its config from the bags.
reset_context()
self.addCleanup(reset_context)
publish(ServerArgs(model_path="dummy"), role="tokenizer")
def _lazy_full(self):
full, _swa = _paged_pair(lazy=True)
v = full.alloc(full.page_size * 4)
@@ -112,6 +126,12 @@ class TestDriftReportsLoudly(unittest.TestCase):
class TestChainFrontierOrder(unittest.TestCase):
def setUp(self):
# The code under test reads its config from the bags.
reset_context()
self.addCleanup(reset_context)
publish(ServerArgs(model_path="dummy"), role="tokenizer")
def test_overlapping_frontiers_report(self):
"""Both bands hold pages, then the up member's watermark is pushed
past the down member's LIVE low frontier. Both sides must be populated
test/registered/unit/mem_cache/test_unified_mha_views.py
+12
View File
@@ -367,6 +367,18 @@ class TestUnifiedMHATokenToKVPool(unittest.TestCase):
class TestFactoryViews(unittest.TestCase):
def setUp(self):
# `KVIndexTranslator.__init__` asks the parallel context for
# `attn_dcp_size`, which is a quotient of the configured leaves and is
# computed at publish. A bare process has none, so state one the way a
# real process does.
from sglang.srt.runtime_context import publish, reset_context
from sglang.srt.server_args import ServerArgs
reset_context()
self.addCleanup(reset_context)
publish(ServerArgs(model_path="dummy"), role="test")
"""Over the real SWA factory: matching kernel-facing multipliers in the
composite allocator, and a rebind that emits both write locs."""
test/registered/unit/mem_cache/test_unified_mla_gpu_parity.py
+8
View File
@@ -26,6 +26,8 @@ import unittest
import torch
from sglang.srt.runtime_context import publish, reset_context
from sglang.srt.server_args import ServerArgs
from sglang.test.ci.ci_register import register_cuda_ci
register_cuda_ci(est_time=10, stage="base-b", runner_config="1-gpu-small")
@@ -107,6 +109,12 @@ def _rand_locs(max_tokens: int, ps: int, n: int) -> torch.Tensor:
@unittest.skipUnless(_HAS_CUDA, "requires CUDA")
class TestUnifiedMLAPoolGPUParity(unittest.TestCase):
def setUp(self):
# The code under test reads its config from the bags.
reset_context()
self.addCleanup(reset_context)
publish(ServerArgs(model_path="dummy"), role="tokenizer")
def _assert_parity(self, unified, ref, locs, ps, layers=range(_L)):
for l in layers:
got = unified.get_key_buffer(l)[_kernel_id(locs, ps)]
test/registered/unit/model_executor/model_runner_components/test_startup_weight_load.py
+45 -5
View File
@@ -158,6 +158,9 @@ class _TiedWeightModel(nn.Module):
class TestStartupWeightLoadSelector(CustomTestCase):
def setUp(self):
reset_context()
self.addCleanup(reset_context)
publish(ServerArgs(model_path="dummy"), role="tokenizer")
self.load_config = LoadConfig(load_format=LoadFormat.SAFETENSORS)
self.loader = DefaultModelLoader(self.load_config)
self.device_config = DeviceConfig("cuda", 0)
@@ -324,6 +327,12 @@ class TestStartupWeightLoadSelector(CustomTestCase):
class TestStartupWeightLoadManager(CustomTestCase):
def setUp(self):
# The code under test reads its config from the bags.
reset_context()
self.addCleanup(reset_context)
publish(ServerArgs(model_path="dummy"), role="tokenizer")
def _manager(self, loader):
return StartupWeightLoadManager(
loader=loader,
@@ -512,6 +521,12 @@ class TestStartupWeightLoadManager(CustomTestCase):
class TestModelStorageManifest(CustomTestCase):
def setUp(self):
# The code under test reads its config from the bags.
reset_context()
self.addCleanup(reset_context)
publish(ServerArgs(model_path="dummy"), role="tokenizer")
def test_in_place_updates_preserve_the_manifest(self):
model = _TiedWeightModel()
manifest = ModelStorageManifest.capture(model)
@@ -554,6 +569,12 @@ class TestModelStorageManifest(CustomTestCase):
class TestCaptureSafeWeightInitialization(CustomTestCase):
def setUp(self):
# The code under test reads its config from the bags.
reset_context()
self.addCleanup(reset_context)
publish(ServerArgs(model_path="dummy"), role="tokenizer")
def test_only_parameters_are_filled(self):
model = _TiedWeightModel()
@@ -576,6 +597,12 @@ class _LifecycleRunner:
class TestStartupWeightLoadFanout(CustomTestCase):
def setUp(self):
# The code under test reads its config from the bags.
reset_context()
self.addCleanup(reset_context)
publish(ServerArgs(model_path="dummy"), role="tokenizer")
def test_primary_and_multi_runner_extras_are_started_once(self):
trace = []
primary = _LifecycleRunner("primary", trace)
@@ -629,6 +656,12 @@ class _RunnerStartupManager:
class TestModelRunnerStartupWeightLoadOwnership(CustomTestCase):
def setUp(self):
# The code under test reads its config from the bags.
reset_context()
self.addCleanup(reset_context)
publish(ServerArgs(model_path="dummy"), role="tokenizer")
@staticmethod
def _runner(manager):
runner = ModelRunner.__new__(ModelRunner)
@@ -689,14 +722,21 @@ class _SchedulerWorker:
class TestStartupWeightLoadSchedulerRouting(CustomTestCase):
@staticmethod
def _scheduler(worker, trace, *, mode, draft_worker=None):
def setUp(self):
reset_context()
self.addCleanup(reset_context)
def _scheduler(self, worker, trace, *, mode, draft_worker=None):
from sglang.srt.managers.scheduler import Scheduler
scheduler = Scheduler.__new__(Scheduler)
scheduler.server_args = SimpleNamespace(
is_startup_weight_load_overlap=mode == "overlap"
# The schedule reads the mode from the bags, so the test states it by
# publishing a record rather than by standing one in.
reset_context()
publish(
ServerArgs(model_path="dummy", startup_weight_load_mode=mode),
role="scheduler",
)
scheduler = Scheduler.__new__(Scheduler)
scheduler.init_tp_model_worker = lambda: setattr(scheduler, "tp_worker", worker)
scheduler.maybe_init_draft_worker = lambda: setattr(
scheduler, "draft_worker", draft_worker
test/registered/unit/server_args/test_resolution_reads_the_declarations.py
+3 -2
View File
@@ -472,8 +472,9 @@ class TestResolutionReadsTheDeclarations(CustomTestCase):
)
members = _record_members()
# The floor is here to catch the scan collapsing, not to pin the
# class's size.
self.assertGreater(len(members), 15, f"only {len(members)} members were found")
# class's size -- it drops as derived members move to their namespaces
# and become declarations rather than methods on the record.
self.assertGreater(len(members), 10, f"only {len(members)} members were found")
offenders = []
for name, fn in sorted(members.items()):
holders = _holders(fn) | {"self"}
test/registered/unit/spec/test_eagle_draft_cuda_graph_runner.py
+8
View File
@@ -19,6 +19,8 @@ from types import SimpleNamespace
import torch
from sglang.srt.runtime_context import publish, reset_context
from sglang.srt.server_args import ServerArgs
from sglang.srt.speculative.eagle_draft_cuda_graph_runner import (
EAGLEDraftCudaGraphRunner,
)
@@ -59,6 +61,12 @@ class _RecordingDraftBackend:
class TestEagleDraftCudaGraphRunner(CustomTestCase):
def setUp(self):
# The code under test reads its config from the bags.
reset_context()
self.addCleanup(reset_context)
publish(ServerArgs(model_path="dummy"), role="tokenizer")
def _build_runner(self, backend):
runner = EAGLEDraftCudaGraphRunner.__new__(EAGLEDraftCudaGraphRunner)
runner.deepep_adapter = SimpleNamespace(replay=lambda: None)
test/registered/unit/test_runtime_context.py
+202 -52
View File
@@ -53,21 +53,22 @@ _SRT = _pathlib.Path(next(iter(_sglang.__path__))).resolve() / "srt"
_PS = "sglang.srt.distributed.parallel_state"
_DP = "sglang.srt.layers.dp_attention"
# Ranks and the world size read the live group: they are not implied by
# anything, so there is nothing to derive them from. The quotients used to be
# in this table and are not any more -- `attn_tp_size` and its siblings are
# functions of the configured leaves, and `TestDerivedWidthsComeFromTheLeaves`
# is what pins them.
SIZE_RANK_DELEGATIONS = [
("world_size", f"{_PS}.get_world_size"),
("world_rank", f"{_PS}.get_world_rank"),
("tp_rank", f"{_PS}.get_tensor_model_parallel_rank"),
("dcp_rank", f"{_PS}.get_dcp_rank"),
("pp_rank", f"{_PS}.get_pipeline_model_parallel_rank"),
("moe_ep_size", f"{_PS}.get_moe_expert_parallel_world_size"),
("moe_ep_rank", f"{_PS}.get_moe_expert_parallel_rank"),
("moe_dp_rank", f"{_PS}.get_moe_data_parallel_rank"),
("moe_tp_size", f"{_PS}.get_moe_tensor_parallel_world_size"),
("moe_tp_rank", f"{_PS}.get_moe_tensor_parallel_rank"),
("attn_tp_size", f"{_PS}.get_attn_tensor_model_parallel_world_size"),
("attn_tp_rank", f"{_PS}.get_attn_tensor_model_parallel_rank"),
("attn_cp_rank", f"{_PS}.get_attn_context_model_parallel_rank"),
("attn_dp_size", f"{_DP}.get_attention_dp_size"),
("attn_dp_rank", f"{_DP}.get_attention_dp_rank"),
]
@@ -176,36 +177,49 @@ class TestParallelOverride(_IsolatedOverrides):
class TestParallelDCP(_IsolatedOverrides):
def test_attn_dcp_defaults_when_group_is_uninitialized(self):
"""The DCP width is a quotient; the DCP rank is a live reading.
They used to be tested the same way, by mocking the group getters, because
the width read the group too. It does not: `attn_dcp_size` is
`dcp_size if dcp_enabled else 1`, so the way to state it is to state the
leaves.
"""
def _published(self, **fields):
reset_context()
self.addCleanup(reset_context)
publish(ServerArgs(model_path="dummy", **fields), role="test")
return get_parallel()
def test_attn_dcp_is_one_when_dcp_is_off(self):
parallel = self._published(tp_size=8, dcp_size=1)
self.assertFalse(parallel.dcp_enabled)
self.assertEqual(parallel.attn_dcp_size, 1)
def test_attn_dcp_is_the_configured_width_when_on(self):
parallel = self._published(tp_size=8, dcp_size=8)
self.assertTrue(parallel.dcp_enabled)
self.assertEqual(parallel.attn_dcp_size, 8)
def test_the_dcp_rank_still_reads_the_group(self):
"""A rank is not implied by the configuration, so it reads the group --
gated on a width that is."""
with (
patch(f"{_PS}.get_dcp_group_no_assert", return_value=None),
patch(f"{_PS}.get_dcp_world_size", side_effect=AssertionError),
get_parallel().override(tp_size=8, dcp_size=8, dcp_enabled=False),
patch(f"{_PS}.get_dcp_rank", side_effect=AssertionError),
):
self.assertFalse(get_parallel().dcp_enabled)
self.assertEqual(get_parallel().attn_dcp_size, 1)
self.assertEqual(get_parallel().attn_dcp_rank, 0)
def test_attn_dcp_delegates_when_enabled(self):
with (
patch(f"{_PS}.get_dcp_group_no_assert", return_value=object()),
patch(f"{_PS}.get_dcp_world_size", return_value=8),
get_parallel().override(tp_size=8, dcp_size=8, dcp_enabled=True),
patch(f"{_PS}.get_dcp_rank", return_value=3),
):
self.assertTrue(get_parallel().dcp_enabled)
self.assertEqual(get_parallel().attn_dcp_size, 8)
self.assertEqual(get_parallel().attn_dcp_rank, 3)
def test_dcp_enablement_is_platform_agnostic(self):
with (
patch(f"{_PS}.get_dcp_group_no_assert", return_value=object()),
patch("sglang.srt.utils.is_cuda", return_value=False) as is_cuda,
patch(f"{_PS}.get_dcp_world_size", return_value=8),
patch(f"{_PS}.get_dcp_rank", return_value=3),
):
self.assertTrue(get_parallel().dcp_enabled)
self.assertEqual(get_parallel().attn_dcp_size, 8)
self.assertEqual(get_parallel().attn_dcp_rank, 3)
def test_the_width_does_not_consult_the_platform(self):
with patch("sglang.srt.utils.is_cuda", return_value=False) as is_cuda:
parallel = self._published(tp_size=8, dcp_size=8)
self.assertTrue(parallel.dcp_enabled)
self.assertEqual(parallel.attn_dcp_size, 8)
is_cuda.assert_not_called()
@@ -1150,27 +1164,32 @@ class TestDerivedPredicatesAgreeAcrossTiers(_IsolatedServerArgs):
_STRATEGIES = ("auto", "no_buffer", "extra_buffer", "extra_buffer_lazy")
def test_mamba_extra_buffer_matches_the_member(self):
from sglang.srt.runtime_context import (
mamba_extra_buffer_enabled,
mamba_extra_buffer_lazy_enabled,
)
def test_the_mamba_extra_buffer_predicate_has_one_answer(self):
"""It used to be asserted that two spellings agreed. There is one now:
the declaration computes it at publish, and the bag carries it."""
for disable_radix_cache in (False, True):
for strategy in self._STRATEGIES:
with self.subTest(radix=disable_radix_cache, strategy=strategy):
args = _FakeResolvedArgs(
disable_radix_cache=disable_radix_cache,
mamba_radix_cache_strategy=strategy,
reset_context()
publish(
ServerArgs(
model_path="dummy",
disable_radix_cache=disable_radix_cache,
mamba_radix_cache_strategy=strategy,
),
role="test",
)
get_context().set_server_args(args)
self.assertEqual(
ServerArgs.enable_mamba_extra_buffer(args),
mamba_extra_buffer_enabled(),
expected = disable_radix_cache is False and strategy in (
"extra_buffer",
"extra_buffer_lazy",
)
self.assertEqual(
ServerArgs.enable_mamba_extra_buffer_lazy(args),
mamba_extra_buffer_lazy_enabled(),
get_exec().mamba.enable_mamba_extra_buffer, expected
)
self.assertEqual(
get_exec().mamba.enable_mamba_extra_buffer_lazy,
disable_radix_cache is False
and strategy == "extra_buffer_lazy",
)
def test_prefill_buffer_ceiling_matches_the_member(self):
@@ -1446,6 +1465,62 @@ class TestDerivedWidths(_IsolatedOverrides):
)
)
def test_the_published_configuration_decides_the_widths(self):
"""The quotients are computed once, at publish, from the leaves.
Every input is a record field, so there is nothing to recompute on a
read: `publish` fills the bag and the bag is the answer.
"""
reset_context()
self.addCleanup(reset_context)
publish(
ServerArgs(
model_path="dummy", tp_size=8, dp_size=2, enable_dp_attention=True
),
role="test",
)
self.assertEqual(get_parallel().attn_tp_size, 4)
self.assertEqual(get_parallel().attn_dp_size, 2)
self.assertEqual(get_parallel().moe_tp_size, 8)
reset_context()
publish(
ServerArgs(model_path="dummy", tp_size=8, ep_size=4, moe_dp_size=2),
role="test",
)
self.assertEqual(get_parallel().moe_tp_size, 1)
def test_a_topology_is_stated_by_naming_the_width(self):
"""Overriding a leaf does not move the quotient -- the quotient is not
recomputed on read. Naming it is how a test states one."""
reset_context()
self.addCleanup(reset_context)
publish(ServerArgs(model_path="dummy", tp_size=8), role="test")
self.assertEqual(get_parallel().attn_tp_size, 8)
with get_parallel().override(tp_size=2):
self.assertEqual(get_parallel().attn_tp_size, 8)
with get_parallel().override(attn_tp_size=4):
self.assertEqual(get_parallel().attn_tp_size, 4)
def test_an_unstated_topology_still_fails(self):
"""Neutral leaves are for the dimensions a caller is not using, not for
a caller that stated nothing: every width would come back 1, which is a
plausible-looking number invented out of nothing."""
with self.assertRaises(RuntimeError) as caught:
get_parallel().attn_tp_size
self.assertIn("not available", str(caught.exception))
def test_a_stamp_and_a_live_group_both_win_over_the_leaves(self):
"""Order is stamp, then live group, then the leaves. Where a group
exists it is the truth -- elastic scale-up moves the group without
restamping -- so the leaf derivation only answers where there is none.
"""
parallel = get_parallel()
parallel.stamp_derived_widths(attn_tp_size=7)
self.addCleanup(parallel.clear_derived_widths)
with parallel.override(tp_size=8, attn_dp_size=2):
self.assertEqual(parallel.attn_tp_size, 7)
def test_the_quotients_come_from_the_leaves(self):
widths = derive_parallel_widths(
tp_size=8,
@@ -1497,10 +1572,24 @@ class TestDerivedWidths(_IsolatedOverrides):
self.assertEqual(parallel.attn_tp_size, 1)
self.assertEqual(parallel.attn_tp_size, 4)
def test_without_a_stamp_the_live_group_still_answers(self):
"""A process that installed groups by hand keeps working."""
with patch(f"{_PS}.get_attn_tensor_model_parallel_world_size", return_value=2):
self.assertEqual(get_parallel().attn_tp_size, 2)
def test_the_group_is_never_consulted(self):
"""There is no third source. A quotient comes from an override, a stamp
or the published leaf -- never from a group coordinator, which could
only ever agree, since `initialize_model_parallel` stamps as its last
statement."""
reset_context()
self.addCleanup(reset_context)
with patch(
f"{_PS}.get_attn_tensor_model_parallel_world_size",
side_effect=AssertionError("the group must not be consulted"),
):
publish(
ServerArgs(
model_path="dummy", tp_size=8, dp_size=2, enable_dp_attention=True
),
role="test",
)
self.assertEqual(get_parallel().attn_tp_size, 4)
def test_with_neither_the_failure_names_the_cause(self):
with patch(
@@ -1533,24 +1622,23 @@ class TestDerivedWidths(_IsolatedOverrides):
parallel.stamp_derived_widths(attn_dp_size=4)
self.assertEqual(parallel.attn_dp_size, 4)
parallel.clear_derived_widths()
with patch(f"{_DP}.get_attention_dp_size", return_value=1):
with parallel.override(tp_size=8, attn_dp_size=1):
self.assertEqual(parallel.attn_dp_size, 1)
def test_reset_context_drops_the_stamp(self):
"""The stamp belongs to the lifecycle that made it.
`_derived_width` prefers the stamp over the live group, so a stamp that
outlived `reset_context()` would let the next test read the previous
topology.
`_derived_width` prefers the stamp over the published leaf, so a stamp
that outlived `reset_context()` would let the next test read the
previous topology.
"""
from sglang.srt.runtime_context import reset_context
parallel = get_parallel()
parallel.stamp_derived_widths(attn_tp_size=4)
self.assertEqual(parallel.attn_tp_size, 4)
reset_context()
with patch(f"{_PS}.get_attn_tensor_model_parallel_world_size", return_value=1):
self.assertEqual(get_parallel().attn_tp_size, 1)
self.addCleanup(reset_context)
publish(ServerArgs(model_path="dummy", tp_size=1), role="test")
self.assertEqual(get_parallel().attn_tp_size, 1)
def test_the_arithmetic_has_one_home(self):
"""`parallel_state` builds its groups from the same dict it stamps, and
@@ -1589,5 +1677,67 @@ class TestDerivedWidths(_IsolatedOverrides):
self.assertEqual(attn_dp_size, widths["attn_dp_size"])
class TestTheDerivedHalfIsDeclared(CustomTestCase):
"""The quotients are declared beside the leaves, in the same class.
A namespace is one file and one class. `Parallel` says both what an
operator can set and what that decides; the quotients are unannotated, so
they are not dataclass fields and never reach the record.
`ParallelContext` installs a property per declaration rather than carrying
its own list, so the two cannot drift.
"""
def test_every_declared_quotient_has_a_property(self):
from sglang.srt.arg_groups.arg_utils import Derived
from sglang.srt.arg_groups.fields.parallel import Parallel
declared = {
name for name, value in vars(Parallel).items() if isinstance(value, Derived)
}
self.assertTrue(declared, "the derived half is empty")
for name in declared:
self.assertIsInstance(
getattr(type(get_context().parallel), name, None),
property,
f"{name} is declared but no property was installed",
)
def test_the_declared_set_is_what_derive_parallel_widths_produces(self):
"""The declaration is not a second list to keep in step: it names
exactly the quotients the derivation returns."""
from sglang.srt.arg_groups.arg_utils import Derived
from sglang.srt.arg_groups.fields.parallel import Parallel
declared = {
name for name, value in vars(Parallel).items() if isinstance(value, Derived)
}
produced = set(
derive_parallel_widths(
tp_size=8,
attn_cp_size=1,
attn_dp_size=2,
moe_ep_size=1,
moe_dp_size=1,
dcp_size=1,
dcp_enabled=False,
)
)
self.assertEqual(declared, produced)
def test_a_declared_quotient_is_not_a_record_field(self):
"""It has no operator input to preserve, and the record is what crosses
a process boundary."""
import dataclasses
from sglang.srt.arg_groups.arg_utils import Derived
from sglang.srt.arg_groups.fields.parallel import Parallel
from sglang.srt.server_args import ServerArgs
fields = {f.name for f in dataclasses.fields(ServerArgs)}
for name, value in vars(Parallel).items():
if isinstance(value, Derived):
self.assertNotIn(name, fields)
if __name__ == "__main__":
unittest.main()
test/registered/unit/test_server_args_migration.py
+7 -2
View File
@@ -8,6 +8,7 @@ import argparse
import unittest
from sglang.srt.arg_groups.overrides import resolution_result
from sglang.srt.runtime_context import get_model, publish, reset_context
from sglang.srt.server_args import ServerArgs
from sglang.srt.utils.common import configure_media_url_security
from sglang.test.ci.ci_register import register_cpu_ci
@@ -132,9 +133,13 @@ class TestServerArgsAnnotatedCli(CustomTestCase):
serial = self._parse([])
overlap = self._parse(["--startup-weight-load-mode", "overlap"])
self.assertEqual(serial.startup_weight_load_mode, "serial")
self.assertFalse(serial.is_startup_weight_load_overlap)
self.assertEqual(overlap.startup_weight_load_mode, "overlap")
self.assertTrue(overlap.is_startup_weight_load_overlap)
# The predicate over that leaf is a bag leaf now, computed at publish.
for record, expected in ((serial, False), (overlap, True)):
reset_context()
self.addCleanup(reset_context)
publish(record, role="test")
self.assertIs(get_model().is_startup_weight_load_overlap, expected)
with self.assertRaises(SystemExit):
self.parser.parse_args(
test/registered/unit/test_server_args_namespaces.py
+9 -4
View File
@@ -1,9 +1,14 @@
"""Coverage lint for the ServerArgs -> RuntimeContext namespace split.
Every ServerArgs field must carry an ``NS("<path>")`` marker in its ``Annotated``
metadata, and every path must be one of the known domains. This is the guardrail
that fails when an upstream PR adds a ServerArgs field without assigning it a
namespace (the property that retires the old hand-maintained mirror file).
Every ServerArgs field must resolve to a namespace, and every path must be one of
the known domains. A field gets its namespace from the ``arg_groups/fields/``
class that declares it -- each carries the ``_NS_PATH`` it stands for, so the
module a declaration lives in *is* the answer, and there is no per-field marker
to forget. (``NS("<path>")`` survives for the one shape a class cannot express:
an ad-hoc dataclass spanning namespaces, which the config-bag tests build.)
This is the guardrail that fails when an upstream PR adds a field to a namespace
class that has no ``_NS_PATH``, or adds one outside the taxonomy below.
"""
import dataclasses