diff --git a/python/sglang/benchmark/one_batch.py b/python/sglang/benchmark/one_batch.py index 83a2e42ab..309add1b9 100644 --- a/python/sglang/benchmark/one_batch.py +++ b/python/sglang/benchmark/one_batch.py @@ -327,7 +327,6 @@ def load_model(server_args, port_args, gpu_id, tp_rank): server_args=server_args, ) - # Phase two: this entry has no scheduler to run it. bootstrap.init_parallel_runtime( server_args=server_args, model_config=model_config, diff --git a/python/sglang/multimodal_gen/runtime/distributed/parallel_state.py b/python/sglang/multimodal_gen/runtime/distributed/parallel_state.py index f3be2f255..6dad38a9b 100644 --- a/python/sglang/multimodal_gen/runtime/distributed/parallel_state.py +++ b/python/sglang/multimodal_gen/runtime/distributed/parallel_state.py @@ -102,8 +102,6 @@ def _sync_srt_world_group() -> None: if srt_parallel_state._WORLD is None: srt_parallel_state._WORLD = _WORLD if srt_parallel_state._WORLD is _WORLD: - # On the context too: that is where a handle is read from, and - # assigning the module global above does not reach it. get_parallel().override_permanently(world_group=_WORLD) @@ -115,18 +113,11 @@ def _clear_srt_world_group() -> None: def _sync_srt_tp_group() -> None: - """Lend this package's TP group to `srt`, and state the widths it implies. + """Expose this package's TP group, widths, and ranks to shared SRT layers. - Shared `srt` layers run in this package and ask `get_parallel()` for how to - shard -- `srt/layers/attention/vision.py` reads `attn_tp_size`. The - published `srt` config cannot answer: `gpu_worker.py` publishes a dummy - carrying *this* package's `tp_size`, which a sequence-parallel launch sets - to 1 while the group lent here is as wide as the world. So the widths are - permanently overridden alongside the group -- this runs with no `srt` - config published at all, which is exactly why it cannot go through - `RuntimeContext.override` (it requires one). - - Only tensor parallelism folds this way, so every other dimension is one. + Use the group's actual width: sequence parallelism can make it wider than + the TP size in the dummy SRT configuration. Other parallel dimensions are + one. Overrides also work before SRT configuration is published. """ import sglang.srt.distributed.parallel_state as srt_parallel_state from sglang.srt.runtime_context import derive_parallel_widths, get_parallel @@ -137,21 +128,9 @@ def _sync_srt_tp_group() -> None: srt_parallel_state._ATTN_TP = _TP if srt_parallel_state._ATTN_TP is _TP: get_parallel().override_permanently( - # The group itself, because that is what the `srt` context answers - # a handle with -- assigning the module global above does not reach - # it. `tp_size` comes with them: the group is as wide as the world - # while the dummy carries this package's, and the widths below are - # quotients of one number, so stating a subset would describe a - # layout that does not exist. tp_group=_TP, attn_tp_group=_TP, tp_size=_TP.world_size, - # The ranks too. The shared layers shard by them -- `vision.py` - # reads `attn_tp_rank`, every `srt` linear built without an - # explicit rank reads `tp_rank` -- and this package publishes no - # rank bundle, so nothing else writes one. The draft has no - # pipeline, context or expert dimension of its own, so those - # positions are zero. tp_rank=_TP.rank_in_group, attn_tp_rank=_TP.rank_in_group, moe_tp_rank=_TP.rank_in_group, @@ -178,8 +157,7 @@ def _clear_srt_tp_group() -> None: srt_parallel_state._ATTN_TP = None get_parallel().clear_stamp() if srt_parallel_state._WORLD is not None: - # `clear_stamp` drops every stamped name; the WORLD group this - # package lent is still built, so hand it back. + # Restore the still-active WORLD handle after clearing TP overrides. get_parallel().override_permanently(world_group=srt_parallel_state._WORLD) if srt_parallel_state._TP is _TP: srt_parallel_state._TP = None diff --git a/python/sglang/multimodal_gen/test/unit/test_component_accuracy_parallel_runtime.py b/python/sglang/multimodal_gen/test/unit/test_component_accuracy_parallel_runtime.py index 9539f9e87..30502af55 100644 --- a/python/sglang/multimodal_gen/test/unit/test_component_accuracy_parallel_runtime.py +++ b/python/sglang/multimodal_gen/test/unit/test_component_accuracy_parallel_runtime.py @@ -148,13 +148,8 @@ def test_srt_attention_tp_group_tracks_diffusion_tp_group(): assert srt_parallel_state._TP is tp_group assert srt_parallel_state._ATTN_TP is tp_group assert get_parallel().attn_tp_size == 2 - # The handle too: assigning the module global does not reach the `srt` - # context, which is what the shared layers ask for a group. assert get_parallel().tp_group is tp_group assert get_parallel().attn_tp_group is tp_group - # And the ranks the shared layers shard by. Nothing else writes one - # here: this package publishes no rank bundle, so a handle without a - # rank leaves every `srt` linear unable to say which shard it is. assert get_parallel().tp_rank == 1 assert get_parallel().attn_tp_rank == 1 diff --git a/python/sglang/srt/arg_groups/arg_utils.py b/python/sglang/srt/arg_groups/arg_utils.py index 168c5edad..5e00fa566 100644 --- a/python/sglang/srt/arg_groups/arg_utils.py +++ b/python/sglang/srt/arg_groups/arg_utils.py @@ -96,40 +96,18 @@ _NO_DEFAULT = object() class Derived(msgspec.Struct, frozen=True): - """Metadata for a field the configuration implies, not one anyone types. + """Metadata for namespace fields that are not CLI inputs or record fields. - The other half of a namespace. An ``Arg`` field is the operator's input and - is collected into ``ServerArgs``; a ``Derived`` field carries no annotation, - so it is not a dataclass field and never reaches the record -- which is - right, because it has no input to preserve and the record is what crosses a - process boundary. - - ``fn`` names what computes it, as a dotted path resolved lazily so that a - declaration module stays free of runtime imports. Such a field is a pure - function of the published configuration, so it is computed once at - ``publish`` and stored as an ordinary bag leaf -- a plain attribute load, - which is what a read inside compiled model code needs. - - A declaration with no ``fn`` is one nothing can compute: a rank, or a - process group. Those are written into the namespace at runtime -- by - ``publish`` from the spawn bundle, or by the build that creates the group -- - and until then the name has no answer. - - Most declarations carry ``fn``, the parallel quotients included: - they are a function of the configured leaves, so they are computed at - publish like the rest. What is special about them is not how they are - computed but that a stamp can move one afterwards -- ``initialize_dp_attention`` - restamps ``attn_dp_size`` -- which ``ParallelContext`` answers above the - published leaf. + ``fn`` is a lazily resolved dotted function path. It computes a value from + resolved configuration once at publication. Fields without ``fn``, such as + ranks and group handles, are set at runtime. Parallel overrides take + precedence over published values. """ doc: str = "" fn: str = "" - # For a declaration with no ``fn`` whose absence is itself an answer: - # ``gpu_id`` is ``None`` in a process that runs on no device, and a reader - # wants that rather than an error. A rank has no such value -- the wrong - # one is a hang in a collective -- so it carries no default and a read - # before the write says so. + # Default for runtime-only fields, e.g. ``gpu_id=None`` without a device. + # Fields without a default raise if read before initialization. default: Any = _NO_DEFAULT diff --git a/python/sglang/srt/arg_groups/fields/parallel.py b/python/sglang/srt/arg_groups/fields/parallel.py index 378a52220..a2bfd2a84 100644 --- a/python/sglang/srt/arg_groups/fields/parallel.py +++ b/python/sglang/srt/arg_groups/fields/parallel.py @@ -280,18 +280,7 @@ class Parallel(msgspec.Struct): "Maximum EP size the server can scale to at runtime. Pre-allocates active-rank state and backend buffers to this size. Defaults to the launch-time world size.", ] = None - # ---- derived: the quotients of the leaves above ------------------------- - # - # Declared here, beside what they are computed from, because a namespace is - # one file and one class. They are not annotated, so they are not dataclass - # fields and `collect_input_fields` does not put them on the record -- which - # is right: a quotient has no operator input to preserve, and the record is - # what crosses a process boundary, so a width put there would be a stale - # copy the moment an elastic scale-up restamps one. Every input is a leaf - # above, so all six are fixed once the configuration is: `publish` computes - # them through `parallel_widths_of` and stores them as ordinary bag leaves, - # and `ParallelContext` answers with the stamp when a scale-up has moved - # one. + # Derived fields are computed at publication and are not stored in ServerArgs. attn_tp_size = Derived( fn="sglang.srt.runtime_context.attn_tp_size_of", doc="Attention tensor-parallel width: `tp_size` divided by the " @@ -321,13 +310,7 @@ class Parallel(msgspec.Struct): "wider than one rank, which is exactly when the group gets built.", ) - # -- written at runtime, not carried by any configuration -------------- - # - # No `fn`: nothing here is a function of the leaves above. A rank is - # written by `publish` from the spawn bundle; a group by the build that - # creates it. Until one of them has run there is no answer, and a read - # says so rather than deriving something that would answer a different - # question. + # Runtime fields: publish sets ranks; distributed initialization sets groups. tp_rank = Derived(doc="This process's place in the tensor-parallel group.") pp_rank = Derived(doc="This process's place in the pipeline group.") moe_ep_rank = Derived(doc="This process's place in the expert-parallel group.") diff --git a/python/sglang/srt/disaggregation/encoder/server.py b/python/sglang/srt/disaggregation/encoder/server.py index 67f1abde0..40a4e6710 100644 --- a/python/sglang/srt/disaggregation/encoder/server.py +++ b/python/sglang/srt/disaggregation/encoder/server.py @@ -589,12 +589,7 @@ class MMEncoder: distributed_init_method=dist_init_method, local_rank=rank, ) - # The encoder serves the vision tower on a world of its own: `tp_size` - # ranks wide, with no pipeline, no expert or MoE-DP dimension and no - # decode context parallelism, whatever the generation side published. - # That has always been the layout it builds; stating it is what stops - # the context from answering with the other side's topology while these - # groups answer with this one. + # The encoder uses a separate WORLD with tensor and attention-CP parallelism. parallel = get_parallel() attn_cp_size = parallel.attn_cp_size attn_tp_size = parallel.tp_size // attn_cp_size diff --git a/python/sglang/srt/distributed/bootstrap.py b/python/sglang/srt/distributed/bootstrap.py index 0fedb3e0a..db988f3a4 100644 --- a/python/sglang/srt/distributed/bootstrap.py +++ b/python/sglang/srt/distributed/bootstrap.py @@ -59,23 +59,18 @@ _is_cpu_arm64 = is_host_cpu_arm64() _TP_ALL_TO_ALL_WARMUP_BYTES_PER_PEER = 4 << 20 -#: Set by `init_parallel`; `destroy_model_parallel` clears it, so a test that -#: tears the groups down can build them again. +#: Cleared by `destroy_model_parallel`. _PARALLEL_INITIALISED = False def reset_parallel_initialised() -> None: - """Forget that the groups were built. Paired with tearing them down.""" + """Reset the initialization guard when model-parallel groups are destroyed.""" global _PARALLEL_INITIALISED _PARALLEL_INITIALISED = False def _bind_threads_if_cpu(*, device: str) -> "Optional[List[int]]": - """Pin OpenMP threads to this process's NUMA node, on CPU. - - A precondition of the CPU group build, which reads the binding, so it is - done here rather than left for a caller to remember. - """ + """Bind OpenMP threads to the NUMA node before CPU group initialization.""" if device != "cpu": return None from sglang.srt.utils import numa_utils @@ -99,23 +94,11 @@ def init_parallel_runtime( device: str, dist_port: int, ) -> None: - """Phase two of startup: bring the parallel runtime up, once. + """Initialize the parallel runtime once, after publishing configuration. - Publish says what the topology is; this makes it exist. Nothing returns, - because the groups are read through the runtime context -- a caller that - wants one asks `get_parallel()`, in this process or any later phase. - - "Runtime" rather than "groups": two things have to be in place before the - groups can be built, and they are done here rather than left for every - entry to remember. The OpenMP/NUMA binding is what the CPU group build - reads, and the shared Mooncake transfer engine is what the Mooncake - process-group backend asks for -- create that one late and a second engine - appears. Both are preconditions of the build, not separate work. - - Runs on the target worker only. A draft worker shares its target's groups, - which is why this is a phase the entry runs rather than something a runner - does on its way up: whether the groups exist must not depend on which - runner happened to be constructed first. + Set up CPU thread binding, the current device, and the shared Mooncake + engine before creating process groups. Draft workers reuse their target's + groups and must not call this function. """ global _PARALLEL_INITIALISED if _PARALLEL_INITIALISED: @@ -140,9 +123,7 @@ def init_parallel_runtime( _set_all_reduce_flags() local_omp_cpuid = _bind_threads_if_cpu(device=device) - # Everything below allocates on the current device -- the NCCL warm-up, the - # mooncake all-reduce buffer -- and without this every rank on a node would - # pick device 0, because the default is not to reindex the visible set. + # Select the local device before communicator allocation and warmup. try: torch.get_device_module(device).set_device(get_device().gpu_id) except Exception: @@ -199,11 +180,9 @@ def init_parallel_runtime( def measure_pre_model_load_memory(*, device: str, is_draft_worker: bool) -> float: - """Available memory after the groups exist and before the model loads. + """Measure available memory for KV-cache sizing before this runner loads. - Sized into the KV cache later, so it has to be taken at exactly this point - -- which is why it stays with the runner rather than moving into the - parallel phase. + Call after parallel initialization and before allocating model weights. """ before_avail_memory = get_available_gpu_memory(device, get_device().gpu_id) diff --git a/python/sglang/srt/distributed/parallel_state.py b/python/sglang/srt/distributed/parallel_state.py index fcc38511d..873633b1c 100644 --- a/python/sglang/srt/distributed/parallel_state.py +++ b/python/sglang/srt/distributed/parallel_state.py @@ -2156,12 +2156,10 @@ _PDMUX_PREFILL_TP_GROUP: Optional[GroupCoordinator] = None @contextmanager def pdmux_prefill_tp_group(): - """Run on the prefill stream's own tensor-parallel communicator. + """Use the duplicate TP communicator for the prefill stream. - PD multiplexing builds a duplicate TP group -- the same ranks, a second - communicator -- so prefill and decode can occupy separate streams without - serialising on one. Nothing about the topology differs, so the scope states - the handle and nothing else. + PD multiplexing keeps prefill and decode on separate communicators with + the same ranks. Only the TP handle changes within this scope. """ assert _PDMUX_PREFILL_TP_GROUP is not None, ( "tensor model parallel group for PD-Multiplexing Prefill is not initialized" @@ -2503,9 +2501,7 @@ def init_distributed_environment( assert _WORLD.world_size == torch.distributed.get_world_size(), ( "world group already initialized with a different world size" ) - # Stated here rather than with the groups below it: WORLD is built in this - # function, and every group `initialize_model_parallel` builds is placed by - # reading it back. + # Publish WORLD before model-parallel initialization reads its local rank. get_parallel().override_permanently(world_group=_WORLD) @@ -2520,17 +2516,8 @@ def initialize_model_parallel( """ Initialize model parallel groups at the published widths. - Every width comes from the runtime context rather than from an argument: - the configuration already says how wide each dimension is, and a caller - that translates it again is a second place for the two to disagree. A - process that needs a narrower layout than the one it published -- the - media encoder is the case in the tree -- states that layout on the context - first, so what it builds and what it answers stay the same thing. - - The remaining arguments are not topology. `backend` is decided by the - device, `duplicate_tp_group` and `enable_symm_mem` by other namespaces, and - `recovered_rank` / `rank_offset` / `max_world_size` describe this - particular join rather than the layout being joined. + Read topology widths from ``get_parallel()``. Callers needing a different + layout must override the context before building groups. The widths this reads: tp_size: GPUs used for tensor model parallelism. @@ -2935,19 +2922,8 @@ def initialize_model_parallel( max_world_size=max_world_size, ) - # The groups just built and the configuration they were built from are two - # accounts of one layout, and this is where they meet: stating a group - # checks the identities, so a group built on the wrong peers is refused - # here rather than hanging in a collective later. - # - # A dimension this configuration does not have is left unstated -- `_DCP` - # is None without decode context parallelism -- so reading it says the - # group was never built, which is what these getters have always said, - # rather than handing back a None to fail on at the collective. - # - # WORLD is not here: it is built and stated by - # `init_distributed_environment`, which is what lets every build above - # place its group by reading `get_world_group().local_rank`. + # Validate group widths against the context. Leave disabled groups unset + # so reading them raises. WORLD was published by distributed initialization. built = { "tp_group": _TP, "pp_group": _PP, @@ -3061,8 +3037,6 @@ def patch_pipeline_parallel_group(pp_group: GroupCoordinator): old_pp_group = _PP _PP = pp_group try: - # `pp_size` is a configured leaf: unlike the rank and the handle it - # does not follow the group being swapped, so the scope has to name it. with get_parallel().override( pp_size=pp_group.world_size, pp_rank=pp_group.rank_in_group, @@ -3076,36 +3050,12 @@ def patch_pipeline_parallel_group(pp_group: GroupCoordinator): @contextmanager def patch_tensor_parallel_group(tp_group: GroupCoordinator, *, owns_attention: bool): - """Run under a different tensor-parallel group until this scope ends. + """Temporarily replace the TP group and its runtime-context values. - This is for draft workers of speculative decoding, which run the draft model - at the target's attention-TP width rather than its global TP width. - - The scope replaces both the module global that ``get_tp_group()`` reads and - the members the runtime context answers with. - - Which members depends on what the draft is, and only the worker knows: the - same call site hands over an attention-TP slice for one draft and the - target's whole TP group for another, so this cannot be read off the group. - - ``owns_attention`` says which. A draft that owns its attention topology - runs the whole model on the group being installed -- there is no - attention-DP replica inside it, so its attention identity is the group - itself, one replica, one context shard, and no expert dimension either. - Leaving those names on the target's answers is what lets a draft read - report a replica count the draft does not have. - - A draft that does not own it was built outside any scope and keeps the - target's layout: the process is still one of several attention-DP replicas - and still gathers with them. Claiming one replica there is the same error - in the other direction, and the reader that acts on it is a collective -- a - DP gather takes its buffer size from the replica count and its communicator - from this group, so the two stop agreeing. - - Args: - tp_group (GroupCoordinator): the tp group coordinator - owns_attention (bool): whether the draft's attention topology is this - group, decided by the worker where it builds its draft runner + For speculative drafts with ``owns_attention=True``, the installed group + is the draft's attention-TP group, with attention-DP, attention-CP, and + MoE-DP/EP widths set to one. Otherwise, retain the target's attention + topology. The worker must specify this based on how it constructed the draft. """ global _TP_STATE_PATCHED @@ -3458,20 +3408,9 @@ def monkey_patch_vllm_parallel_state(reverse: bool = False): setattr(vllm_parallel_state, "get_world_group", get_world_group) -# --- deprecation --------------------------------------------------------- -# -# These getters are the definition of a name, not a second spelling of it. -# Business code asks `get_parallel()`, which answers by calling them and which -# a scope can redirect; a call that arrives here directly cannot be redirected, -# so a draft worker's scope does not reach it. The package that defines them -# keeps calling them -- a read there would go through the context back into -# itself -- so the warning fires only for callers outside it, and once per -# name, because the point is to name the replacement rather than to fill a log. +# Use `get_parallel()` outside this package. Warn once per deprecated getter. _EXEMPT_CALLERS = ("sglang.srt.distributed.",) -# Which context name each getter here answers. The shim's own bookkeeping -- -# what a getter was replaced by is of no interest to whoever declares the field -# -- so it is written next to the warning that uses it. _CONTEXT_NAME_OF = { "get_world_group": "world_group", "get_tp_group": "tp_group", @@ -3494,13 +3433,8 @@ _CONTEXT_NAME_OF = { "get_attn_context_model_parallel_rank": "attn_cp_rank", "get_dcp_rank": "dcp_rank", } -# The width getters read a built group; the context answers the same names from -# the configuration. Those are one answer rather than two only for the groups -# the build checks against the configuration -- `_WIDTH_AND_GROUP` in -# `runtime_context` -- so only those are listed here. `moe_dp`, `moe_tp` and -# `dcp` are not on that list and are deliberately absent: the MoE-DP group is -# the attention-CP group when the latter is wider, and the other two are simply -# not pinned yet. +# Only deprecate width getters whose group widths are validated against +# configuration by `_WIDTH_AND_GROUP` in `runtime_context`. _CONTEXT_NAME_OF["get_tensor_model_parallel_world_size"] = "tp_size" _CONTEXT_NAME_OF["get_attn_tensor_model_parallel_world_size"] = "attn_tp_size" _CONTEXT_NAME_OF["get_attn_context_model_parallel_world_size"] = "attn_cp_size" @@ -3540,9 +3474,7 @@ del _name, _replacement, _fn # What `from sglang.srt.distributed import *` re-exports: everything public -# except the deprecated getters. Business code reaches them through -# `get_parallel()`, and the package that defines them imports them from this -# module by name, so nothing needs the package path to reach one. +# except the deprecated getters. __all__ = [ _public for _public in list(globals()) diff --git a/python/sglang/srt/layers/dp_attention.py b/python/sglang/srt/layers/dp_attention.py index 7e947f44d..2693d4f8b 100644 --- a/python/sglang/srt/layers/dp_attention.py +++ b/python/sglang/srt/layers/dp_attention.py @@ -45,12 +45,10 @@ if TYPE_CHECKING: def deployment_attn_dp_size() -> int: - """Attention-DP replicas in the deployment, which no draft scope narrows. + """Return the deployment's attention-DP replica count. - A draft runs on one attention-DP replica and its scope says so, but the - metadata a draft gathers is shaped by the replicas it gathers *with* -- - the target's. Those come from the configuration, which the scope leaves - alone, so this answers the same number inside the scope and outside it. + Draft scopes retain this count because their metadata gathers include + the target's replicas. """ parallel = get_parallel() attn_dp_size, _ = derive_attention_widths( @@ -63,25 +61,20 @@ def deployment_attn_dp_size() -> int: def dp_gather_width() -> int: - """How many replicas the DP sync gathers over. + """Return the DP gather width. - The attention-DP replicas, except after an elastic-EP scale-up, when the - gather spans the expanded WORLD -- whose width is the `dp_size` the - scale-up published. Read from the context either way: a scoped width has - to reach this, which is the whole reason the name has one home. + After elastic scale-up, the gather spans the expanded WORLD; otherwise + it spans the attention-DP replicas. """ parallel = get_parallel() return parallel.dp_size if world_dp_gather_enabled() else parallel.attn_dp_size def dp_gather_slot() -> int: - """This process's index in the list the DP sync just gathered. + """Return this process's index in the DP gather. - The gather spans the attention-DP replicas, except after an elastic-EP - scale-up, when it spans the expanded WORLD and the joining cohort is - numbered from its offset. Which list was gathered is what the flag below - says, so the index is read from there rather than kept as a second name on - the topology. + After elastic scale-up, use the TP rank plus the join offset; otherwise + use the attention-DP rank. """ parallel = get_parallel() if world_dp_gather_enabled(): @@ -100,12 +93,10 @@ def enable_joiner_all_gather(): def update_dp_attention_post_scale(new_dp_size: int, new_dp_rank: int): - """Point the DP gather at the expanded WORLD. + """Switch DP gathers to the expanded WORLD. - The widths themselves are not written here: the caller scales `dp_size` on - the published bag, and the gather reads its width and this process's slot - from there. The arguments are the values the caller is about to publish, - kept so the log says which scale-up this was. + The caller updates the configured widths; these arguments identify the + scale-up in the log. """ get_flags().dp.use_world_group_for_gather = True logger.debug( @@ -433,9 +424,6 @@ def initialize_dp_attention( if ep_scale_joiner_of(resolving_view(server_args)): dp.joiner_skip_all_gather = True - # Stamped together, after the elastic adjustment: the width and the rank - # describe one topology, and a reader that caught them mid-update would - # see this process placed in a group it is not in. get_parallel().override_permanently( attn_dp_size=attn_dp_size, attn_dp_rank=attn_dp_rank ) @@ -457,9 +445,6 @@ def is_allocation_symmetric() -> bool: def get_dp_local_info(forward_batch: ForwardBatch) -> Tuple[torch.Tensor, torch.Tensor]: # `get_dp_local_info` is only called in global DP gather and scatter. We use global DP rank here. - # The slot in the list that was gathered. A scale-up widens that list - # to WORLD, and this process's index in it is not its index among the - # launch replicas. dp_rank = dp_gather_slot() if forward_batch.dp_local_start_pos is None: @@ -484,9 +469,6 @@ def get_dp_local_slice_cpu( # CPU (start, length) slice for DP-local data in a rank-padded buffer. # Returns Python ints (no D2H sync) and handles the cuda-graph-padded layout. global_num_tokens = forward_batch.global_num_tokens_cpu - # The slot in the list that was gathered. A scale-up widens that list - # to WORLD, and this process's index in it is not its index among the - # launch replicas. dp_rank = dp_gather_slot() local_num_tokens = global_num_tokens[dp_rank] if can_run_graph: diff --git a/python/sglang/srt/layers/sampler.py b/python/sglang/srt/layers/sampler.py index 341073e16..acd35db2c 100644 --- a/python/sglang/srt/layers/sampler.py +++ b/python/sglang/srt/layers/sampler.py @@ -112,10 +112,7 @@ class Sampler(nn.Module): self.cp_sync_group = None if is_dp_attention_enabled(): self.tp_sync_group = get_parallel().attn_tp_group.device_group - # Only when there is more than one context shard to reconcile. The - # sync below already short-circuits on that, and a model running on - # one shard -- a speculative draft, under the scope that says so -- - # has no context-parallel communicator to name. + # Single-shard drafts may have no context-parallel group. if get_parallel().attn_cp_size > 1: self.cp_sync_group = get_parallel().attn_cp_group.device_group diff --git a/python/sglang/srt/managers/scheduler.py b/python/sglang/srt/managers/scheduler.py index 51d3c4633..e368797dd 100644 --- a/python/sglang/srt/managers/scheduler.py +++ b/python/sglang/srt/managers/scheduler.py @@ -517,8 +517,6 @@ class Scheduler( # Init ZBAL, switch allocator should before any torch alloc action self.init_zbal_on_npu() - # The groups are the first thing that allocates, so this comes after the - # allocator switch above and before anything that reads a group. bootstrap.init_parallel_runtime( server_args=server_args, model_config=self.model_config, @@ -5910,7 +5908,6 @@ def dispatch_event_loop(scheduler: Scheduler): def _dispatch_event_loop_once(scheduler: Scheduler): - # The live PP property asserts before torch.distributed init (MLX stub). disaggregation_mode: DisaggregationMode = scheduler.disaggregation_mode if disaggregation_mode == DisaggregationMode.NULL: if scheduler.enable_pdmux: @@ -5940,15 +5937,8 @@ def _dispatch_event_loop_once(scheduler: Scheduler): def resolve_spawn_dp_rank(dp_rank: Optional[int]) -> Optional[int]: - """The `dp_rank` this process was spawned with, in either of its two forms. - - A router does not pass it as an argument, it sets `SGLANG_DP_RANK`. Both - forms are the launcher naming this process's place, so both have to be in - hand before `publish` records the placement -- resolving one of them after - would leave the context answering `None` for a process that has a rank. - """ + """Resolve the launcher DP rank, falling back to ``SGLANG_DP_RANK``.""" if dp_rank is None and "SGLANG_DP_RANK" in os.environ: - # [For Router] if env var "SGLANG_DP_RANK" exist, set dp_rank to the value of the env var return int(os.environ["SGLANG_DP_RANK"]) return dp_rank @@ -6034,10 +6024,6 @@ def run_scheduler_process( # Load plugins so hooks can override Scheduler and its dependencies. load_plugins() dp_rank = resolve_spawn_dp_rank(dp_rank) - # Publish before anything in this process reads configuration, with the - # placement the launcher decided: from here on a rank read is answered - # without a process group, which is what every reader needs before - # `init_torch_distributed` has run. publish( server_args, role="scheduler", diff --git a/python/sglang/srt/managers/scheduler_components/dp_attn.py b/python/sglang/srt/managers/scheduler_components/dp_attn.py index ae1069b13..cc6c8aa94 100644 --- a/python/sglang/srt/managers/scheduler_components/dp_attn.py +++ b/python/sglang/srt/managers/scheduler_components/dp_attn.py @@ -59,9 +59,7 @@ def _resolve_elastic_world_dp_size( live_dp_size = dp_gather_width() effective_ep_size = ElasticEPStateManager.get_effective_ep_size() - # The group's own membership, not the width it was built at: this is the - # one number an out-of-process join moves, and it is the upper bound the - # served width has to stay under. + # Query live membership because elastic joins can expand WORLD. world_size = torch.distributed.get_world_size(group) if live_dp_size != effective_ep_size: diff --git a/python/sglang/srt/mem_cache/kv_cache_builder.py b/python/sglang/srt/mem_cache/kv_cache_builder.py index e0529c281..f403e7cd4 100644 --- a/python/sglang/srt/mem_cache/kv_cache_builder.py +++ b/python/sglang/srt/mem_cache/kv_cache_builder.py @@ -267,8 +267,6 @@ def build_kv_cache( enable_hierarchical_cache: bool, hicache_draft_plan: Optional[HiCacheDraftPlan] = None, ) -> KVCacheBuildResult: - # Built from the scheduler loop, outside any draft scope, so the context - # answers for the process this cache belongs to. parallel = get_parallel() sliding_window_size: Optional[int] = None full_tokens_per_layer: Optional[int] = None diff --git a/python/sglang/srt/mem_cache/kv_cache_configurator.py b/python/sglang/srt/mem_cache/kv_cache_configurator.py index ef86199ba..97382225b 100644 --- a/python/sglang/srt/mem_cache/kv_cache_configurator.py +++ b/python/sglang/srt/mem_cache/kv_cache_configurator.py @@ -259,8 +259,7 @@ class _PoolSizes(msgspec.Struct, frozen=True, kw_only=True): class KVCacheConfigurator: device: str gpu_id: int - # Frozen at construction, not asked for later: this configurator is built - # inside the scope that describes a draft runner and used outside it. + # Capture draft placement at construction; the configurator outlives the scope. attn_dp_size: int pp_size: int pp_group: Any diff --git a/python/sglang/srt/mem_cache/pool_host/base.py b/python/sglang/srt/mem_cache/pool_host/base.py index 924429484..8c4a6674e 100644 --- a/python/sglang/srt/mem_cache/pool_host/base.py +++ b/python/sglang/srt/mem_cache/pool_host/base.py @@ -46,13 +46,9 @@ def host_memory_budget_scope(budget_bytes: int): def ranks_per_host() -> int: - """Number of ranks of this job running on the same machine as this one. + """Return the launch ranks per host, assuming uniform placement. - Derived as the launch width // nnodes: the launcher slices ranks - uniformly across nodes (resolution asserts divisibility), so no hostname - collective is needed — a collective here would have to be issued the same - number of times on every rank, and ranks build different numbers of host - pools. + Avoid a collective: ranks may construct different numbers of host pools. """ if not (torch.distributed.is_available() and torch.distributed.is_initialized()): return 1 diff --git a/python/sglang/srt/model_executor/model_runner.py b/python/sglang/srt/model_executor/model_runner.py index aba114416..9f7a69c7b 100644 --- a/python/sglang/srt/model_executor/model_runner.py +++ b/python/sglang/srt/model_executor/model_runner.py @@ -1157,11 +1157,7 @@ class ModelRunner: self.pre_model_load_memory = bootstrap.measure_pre_model_load_memory( device=self.device, is_draft_worker=self.is_draft_worker ) - # Read once, here: a draft runner is constructed inside the scope that - # states its topology and used outside it, so what it holds has to be - # the placement it was built for rather than whatever the context - # answers later. Groups and widths alike -- a runner asked about its own - # shape after the scope has closed must still describe itself. + # Capture draft placement at construction; the runner outlives the scope. parallel = get_parallel() self.tp_group = parallel.tp_group self.pp_group = parallel.pp_group diff --git a/python/sglang/srt/models/inkling_common/moe.py b/python/sglang/srt/models/inkling_common/moe.py index b7644bb4c..6713f5bd3 100644 --- a/python/sglang/srt/models/inkling_common/moe.py +++ b/python/sglang/srt/models/inkling_common/moe.py @@ -671,12 +671,8 @@ class InklingSharedFusedMoE(FusedMoE): quant_config: QuantizationConfig | None, inference_moe_w13_interleaved: bool, ) -> None: - # FusedMoE.__init__ reads get_parallel() once and caches it on self, so - # scoping the override to just this call is sufficient for the module's lifetime. - # The shared experts are replicated rather than sharded, so there is no - # expert-parallel communication here and no group to name: a width of - # one with the wider group still installed would describe a layout that - # does not exist. + # FusedMoE caches this topology at construction. Shared experts are + # replicated, so they need no expert-parallel group. with get_parallel().override( moe_ep_size=1, moe_ep_rank=0, diff --git a/python/sglang/srt/runtime_context.py b/python/sglang/srt/runtime_context.py index 77e67b2c4..f44b0928f 100644 --- a/python/sglang/srt/runtime_context.py +++ b/python/sglang/srt/runtime_context.py @@ -13,13 +13,10 @@ # ============================================================================== """A single structured accessor for process-static runtime state. -``get_parallel()`` returns a ``ParallelContext``. Ranks and process-group handles -read through **live** to the canonical getter in ``distributed.parallel_state`` / -``layers.dp_attention`` — exactly what those getters return, a read-through -wrapper and not a cache. Every other name, the sizes included, is a leaf of the -published ``parallel`` bag. It gives call-sites one import and one naming scheme -in place of a dozen free functions, plus an ``override()`` hook to force a -topology without monkeypatching the underlying getters. +``get_parallel()`` returns a ``ParallelContext`` for configuration, ranks, and +process-group handles. Reads use scoped overrides, then permanent overrides, +then the published configuration. ``publish`` records ranks from ``SpawnRanks``; +distributed initialization records group handles. ``get_server_args()`` returns the process-wide ``ServerArgs``. This is the user's raw input, kept **read-only** for debug and reproduction; what @@ -72,13 +69,7 @@ _PARALLEL_STATE = None def _ps(): - """The module every rank and group read ends at. - - Cached because the import statement dominated the read: a group read is - two attribute lookups plus this, and it runs per row-linear on an eager - forward. The getter is still resolved by name on the returned module, so - a test that patches `parallel_state.get_tp_group` is still seen. - """ + """Lazily import and cache the parallel-state module.""" global _PARALLEL_STATE if _PARALLEL_STATE is None: from sglang.srt.distributed import parallel_state @@ -95,11 +86,7 @@ def _dp(): @functools.lru_cache(maxsize=1) def _parallel_config_leaves() -> frozenset: - """Names under the ``parallel`` namespace, for the unpublished error path. - - Read from the field metadata rather than the bag, which is what does not - exist yet when this is needed. - """ + """Return configured parallel field names, including before publication.""" from sglang.srt.arg_groups.arg_utils import namespace_of from sglang.srt.server_args import ServerArgs @@ -110,31 +97,12 @@ def _parallel_config_leaves() -> frozenset: ) -# Ranks and group handles: the names no configuration carries. This table is -# their declaration, the way `arg_groups/fields/parallel.py` is the leaves' and -# `Derived` is the widths'. A group handle names the getter that owns it, -# because the module that builds the groups is where it lives; a rank is a -# position in one of those groups, so it is read off the handle. `None` marks a -# name only a stamp can answer: no coordinator knows this process's -# attention-DP rank. _MISSING_READ = object() @functools.lru_cache(maxsize=1) def _parallel_fields() -> frozenset: - """Every name `ParallelContext` answers for, read from the declarations. - - Three sources, because the namespace has three kinds of name and each one - declares itself somewhere already: - - * configured leaves -- the `parallel` namespace of the record; - * derived widths -- the `Derived` declarations beside those leaves; - * ranks and group handles -- declared beside the leaves with no `fn`, - because nothing computes them; they are written at runtime. - - The set is the union of those three, so `override()` cannot refuse a name - the class answers for. - """ + """Return configured and derived parallel names accepted by overrides.""" from sglang.srt.arg_groups.arg_utils import Derived from sglang.srt.arg_groups.fields.parallel import Parallel @@ -147,12 +115,7 @@ def _parallel_fields() -> frozenset: def derive_attention_widths( *, tp_size: int, attn_cp_size: int, dp_size: int, enable_dp_attention: bool ) -> tuple: - """(attn_dp_size, attn_tp_size) from the leaves. - - Split out because the rank computation in - `dp_attention.compute_dp_attention_world_info` needs the same two numbers - and must not carry a second copy of the arithmetic. - """ + """Return (attn_dp_size, attn_tp_size) from the configured widths.""" attn_dp_size = dp_size if enable_dp_attention else 1 return attn_dp_size, tp_size // attn_dp_size // attn_cp_size @@ -160,18 +123,12 @@ def derive_attention_widths( def derive_attention_ranks( *, tp_rank: int, attn_tp_size: int, attn_cp_size: int, enable_dp_attention: bool ) -> tuple: - """(attn_tp_rank, attn_dp_rank) for a process at `tp_rank`. + """Return (attn_tp_rank, attn_dp_rank) for a process at ``tp_rank``. - The rank layout is (dp, cp, tp) with tp the fastest-changing dimension:: + The rank layout is (dp, cp, tp), with tp changing fastest:: tp_rank = (attn_dp_rank * attn_cp_size + attn_cp_rank) * attn_tp_size + attn_tp_rank - - Split out beside `derive_attention_widths` because two places need it from - different inputs: `publish` has this process's `tp_rank` from the spawn and - the widths from the configuration, while `initialize_dp_attention` has them - from the groups it just built. They must not carry separate copies of the - arithmetic -- the point of computing it at publish is that the two agree. """ attn_tp_rank = tp_rank % attn_tp_size if not enable_dp_attention: @@ -180,13 +137,9 @@ def derive_attention_ranks( def spawn_world_rank(server_args, *, tp_rank: int, pp_rank: int) -> int: - """This process's place in WORLD, from the ranks its entry was given. + """Compute the WORLD rank from launcher TP and PP ranks. - The inverse of `derive_spawn_ranks`, for the entries that have the pieces - but not the whole: the same expression `bootstrap` hands to - `init_distributed_environment` when it builds the group. - - Reads a resolving view because it runs before `publish`. + Uses a resolving view because callers may run before ``publish``. """ from sglang.srt.arg_groups.model_override_base import resolving_view @@ -204,17 +157,10 @@ def derive_spawn_ranks( moe_dp_size: int, moe_ep_size: int, ) -> dict: - """Every rank a process group would answer, from its place in WORLD. + """Derive TP, PP, attention, and MoE ranks without process groups. - WORLD is laid out `rank = ep_join_rank_offset + tp_size * pp_rank + - tp_rank`: `initialize_model_parallel` builds tensor-parallel groups as - contiguous blocks of `tp_size` and pipeline groups strided by it, so the - map is a bijection and this is its inverse. The attention and MoE ranks - are then positions inside the tensor-parallel block, which is what the - launcher computes when it decides what to spawn. - - Pure arithmetic over the published widths: no group is consulted, which is - the point -- this runs at publish, before any of them exist. + WORLD uses ``rank = ep_join_rank_offset + tp_size * pp_rank + tp_rank``. + TP groups are contiguous blocks; PP groups are strided by ``tp_size``. """ local = world_rank - ep_join_rank_offset tp_rank = local % tp_size @@ -242,27 +188,10 @@ def derive_parallel_widths( dcp_size: int, dcp_enabled: bool, ) -> dict: - """The parallel widths no flag sets, from the leaves that do. - - `tp_size` and its siblings are configured; these are quotients of them, so - the arithmetic lives here rather than being read back off the group - coordinators. - - The world widths are not among them: neither is a quotient, and they are - not one number. `launch_world_size` is what the WORLD group was built at - and is frozen there -- `GroupCoordinator.world_size` is `len(ranks)`, so it - does not move when mooncake admits ranks into an expandable WORLD; - `max_world_size` is what that group has room for. How much of that room is - serving right now is elastic-EP state and is asked of its owner. Deriving - either from the leaves would be wrong in a further way: on a scale joiner it - would answer with the joining cohort's own `tp * pp`, while that process's - WORLD spans `ep_join_rank_offset + tp * pp`. - """ + """Derive attention and MoE widths and DCP settings from configuration.""" return { "attn_dp_size": attn_dp_size, - # `attn_dp_size` is already the effective width (1 when DP attention is - # off), so the flag is spent here; a caller passing the raw `dp_size` - # leaf with the attention disabled would get tp/dp/cp instead of tp/1/cp. + # `attn_dp_size` already accounts for disabled DP attention. "attn_tp_size": derive_attention_widths( tp_size=tp_size, attn_cp_size=attn_cp_size, @@ -277,14 +206,7 @@ def derive_parallel_widths( def parallel_widths_of(cfg: Any) -> dict: - """The six quotients, from a resolved config. - - Every input is a record field, so this is a function of the configuration - and nothing else -- which is why the six are declared `Derived(fn=...)` and - computed once at publish rather than on every read. `dcp_enabled` is - `dcp_size > 1` because that is exactly when `initialize_model_parallel` - builds the group. - """ + """Return derived parallel settings from resolved configuration.""" attn_dp_size, _ = derive_attention_widths( tp_size=cfg.tp_size, attn_cp_size=cfg.attn_cp_size, @@ -303,20 +225,15 @@ def parallel_widths_of(cfg: Any) -> dict: def launch_world_size_of(cfg: Any): - """`launch_world_size`, computed at publish. + """Return the initial WORLD width, including existing ranks below a joiner. - The width `bootstrap` builds the WORLD at: one rank per pipeline stage of - each tensor-parallel group, above the offset a scale joiner comes in at -- - zero for everyone else, which is the convention `spawn_world_rank` uses for - the same arithmetic. A scale-up does not move it, which is the point of the - name. + This value remains fixed after elastic scale-up. """ return cfg.ep_join_rank_offset + cfg.tp_size * cfg.pp_size def max_world_size_of(cfg: Any): - """`max_world_size`, computed at publish. The ceiling the group is - pre-allocated to: `--max-ep-size` when set, otherwise the launch width.""" + """Return the WORLD capacity: ``max_ep_size`` or the launch width.""" return cfg.max_ep_size or launch_world_size_of(cfg) @@ -351,25 +268,12 @@ def dcp_enabled_of(cfg: Any): class SpawnRanks(msgspec.Struct, frozen=True): - """Where the launcher put this process, in the two numbers only it knows. + """Process placement supplied by the launcher. - `world_rank` is this process's place in the WORLD group, which fixes every - other rank: the groups are laid out from the published widths, so - `tp_rank`, `pp_rank` and the attention / MoE ranks are functions of it (see - `derive_spawn_ranks`). Passing them separately would be passing the same - fact five more times, with five more ways for an entry to contradict - itself. - - `dp_rank` is the exception, because data-parallel replicas are separate - WORLD groups: with `--dp-size 2` each replica holds ranks `0 .. n-1`, so - the rank cannot say which replica this is. `None` means "no controller", - which is an answer rather than an absence, and it is recorded as one. - - `gpu_id` is the device the parent picked for this process. It is not a - position in any group -- reindexing narrows the visible devices before the - spawn, and Ray allocates from its own pool -- but it is the same kind of - fact: something only the entry that spawned the process can state. `None` - for a process that runs on no device. + ``world_rank`` determines TP, PP, attention, and MoE ranks from the + configured widths. ``dp_rank`` identifies the replica across separate + WORLD groups; ``None`` means no data-parallel controller. ``gpu_id`` is + the assigned device index, or ``None`` for a process without a device. """ world_rank: int @@ -386,10 +290,8 @@ _RANK_AND_WIDTH = ( ("moe_ep_rank", "moe_ep_size"), ) -# `moe_dp` is absent because `initialize_model_parallel` aliases the MoE-DP -# group to the attention-CP group when the latter is wider: there the group and -# the name are two facts, which is the same reason `moe_dp_rank` is left off the -# record at publish. +# MoE-DP may alias a wider attention-CP group, so its configured width +# need not match the group width. _WIDTH_AND_GROUP = ( ("tp_size", "tp_group"), ("pp_size", "pp_group"), @@ -402,25 +304,14 @@ _UNREADABLE = object() def _validate_parallel(parallel, source: str) -> None: - """Fail on a topology that cannot describe a real process layout. + """Check rank bounds, topology factorizations, and group widths. - Every identity holds unconditionally: a width and a rank are both - plausible small integers whichever way they are wrong, so an inconsistent - set is not caught by anything downstream -- it surfaces as a hang or a - wrong answer in a collective, far from the write. Stating one leaf without - the quotients that follow from it leaves the namespace describing no real - layout, and the caller that did so is the one that has to say what it meant. - - Names that cannot be read are skipped rather than treated as zero: a - process that has published nothing can still stamp a rank, and a group that - has not been built answers nothing at all. + Skip unavailable or non-integer values so partially initialized contexts + can be validated. """ def read(name): - """A width or a rank, or `_UNREADABLE` for anything these identities - cannot be stated about -- an absent name, `None`, or a stand-in a test - put in a group's place. Booleans are integers in Python and are not - widths, so they are out too.""" + """Return an integer topology value, or ``_UNREADABLE``; exclude booleans.""" try: value = getattr(parallel, name) except Exception: @@ -500,24 +391,12 @@ def _validate_parallel(parallel, source: str) -> None: class ParallelContext: - """Parallel-topology namespace: one spelling per name. + """Parallel configuration, process ranks, and group handles. - Every name is answered by a lookup, never by asking a process group. A - configured leaf and a width derived from one come off the published - ``parallel`` bag; a rank is written by ``publish`` from the spawn bundle, - and a group handle by ``initialize_model_parallel`` as it builds them. A - read before the write that answers it says which write is missing rather - than deriving a number from whatever is installed -- the two answer - different questions, and a plausible wrong rank surfaces as a hang in a - collective far from here. - - That makes a scope a matter of stating names: ``patch_tensor_parallel_group`` - runs a draft worker under a different TP group by overriding the members it - changes, and PD multiplexing points ``tp_group`` at the prefill - communicator the same way. Elastic EP needs no rule at all: it scales - ``ep_size`` / ``dp_size`` on the published bag while ``launch_world_size`` - keeps the width the groups were built at, so the two are different names - rather than two answers to one name. + Configured and derived widths come from the published configuration. + ``publish`` records ranks from the launcher; distributed initialization + records group handles. Scoped overrides take precedence over permanent + overrides and configuration. Uninitialized runtime fields raise on read. """ __slots__ = ("_overrides", "_stamp", "_config") @@ -536,17 +415,7 @@ class ParallelContext: return self._read(name) def _read(self, name): - """The one read path, for every kind of name in the namespace. - - Scoped override, then the permanent stamp, then the published bag. - A rank or a group handle is on no bag, so once those three are out the - name has not been written yet and the read says so. - - The two override maps stay separate because they are taken down by - different things -- a `with` block and `clear_stamp()` -- and - merging them would let a teardown of one drop the other, and would - turn "which wins" into whichever was written last. - """ + """Read a scoped override, permanent override, or published value, in order.""" overrides = self._overrides if name in overrides: return overrides[name] @@ -577,20 +446,10 @@ class ParallelContext: raise AttributeError(f"ParallelContext has no {name!r}") def override_permanently(self, **values) -> None: - """Permanently record a width or rank the published bag can't answer - or no longer answers correctly -- not `RuntimeContext.override`, - because neither is a resolved config leaf and this must work with no - config published at all (`multimodal_gen` lends a TP group to `srt` - layers with no `srt` config to publish against). + """Set parallel values until ``clear_stamp`` or ``reset_context``. - Widths are quotients of the configured leaves, so the bag can usually - answer and this only corrects it; a rank is a per-process fact the - configuration never carries, so for those this is the only source. - - Lives beside, not inside, the `@contextmanager` `override` below -- a - name it cannot also have on this class -- because these are permanent - for the process, not scoped to a `with` block: none of the real - callers ever restore the value they set here. + Works without published configuration. Validate the combined topology + and restore the previous values if validation fails. """ unknown = set(values) - _parallel_fields() if unknown: @@ -628,7 +487,7 @@ class ParallelContext: def _derived_widths() -> dict: - """The declared quotients, by name -- `{name: Derived}`.""" + """Return parallel ``Derived`` declarations, including ranks and groups.""" from sglang.srt.arg_groups.arg_utils import Derived from sglang.srt.arg_groups.fields.parallel import Parallel @@ -638,18 +497,9 @@ def _derived_widths() -> dict: def _install_parallel_properties() -> None: - """Give `ParallelContext` a property per name that is not a config leaf. + """Expose declared parallel fields as documented properties. - Every name the namespace answers that is not a plain leaf: the quotients, - and the ranks and group handles declared beside them with no `fn`. - - Properties rather than names left to `__getattr__` because the class - surface is what the guards introspect -- `hasattr(ParallelContext, - "tp_size")` and `vars(ParallelContext)` are how the tests read the set from - the class side -- and because each one carries its `Derived.doc`. - - Every one of them resolves through `_read`, so there is a single priority - chain rather than one per kind of name. + Properties support class-level introspection; all reads use ``_read``. """ for name, decl in _derived_widths().items(): @@ -755,10 +605,10 @@ class MoeFlags(_FlagGroupBase): class DpFlags(_FlagGroupBase): - """DP-attention runtime flags, materialized by ``initialize_dp_attention`` - (after distributed setup; reads the model config). The topology values it - also computes -- the attention-DP width and rank -- are stamped on - ``get_parallel()``, not kept here.""" + """DP-attention runtime flags set by ``initialize_dp_attention``. + + Attention-DP width and rank are stored on ``get_parallel()``. + """ enabled: bool = False use_world_group_for_gather: bool = False @@ -1134,9 +984,7 @@ def _install_derived_leaves(tops: dict, server_args: Any) -> None: if not isinstance(decl, Derived): continue if not decl.fn: - # Nothing computes it. Seed the ones whose absence is itself an - # answer, so a process that never states one still reads it; - # the rest stay unwritten and say so when read. + # Initialize runtime-only fields that declare a default. if decl.default is not _NO_DEFAULT: bag = tops.get(path.split(".")[0]) for segment in path.split(".")[1:]: @@ -1262,10 +1110,7 @@ class RuntimeContext: # Snapshot resolved config into the namespace bags (the single source of # truth for config reads). Placed by `namespace_of`; a mock/partial # config that declares no namespace yields an empty tree (no bags). - # A name the configuration does not carry survives the re-projection. - # `gpu_id` is stated by the spawn, not derived, so rebuilding the bags - # from the record must not unwrite it -- the record has no field for it - # to be rebuilt from. + # Preserve the launcher-assigned device when rebuilding config bags. stated = {} if self._config_bags is not None: device = self._config_bags.get("device") @@ -1805,11 +1650,9 @@ def publish( namespace-read enforcement (``record`` audits the reads instead). ``hf_config`` is accepted for forward-compat and currently unused. - ``ranks`` is who this process is, from the entry that spawned it. It is - optional because most roles are not placed in the topology at all -- a - tokenizer has no ``tp_rank`` -- and those processes raise on a rank read - exactly as they do today, with a message naming the missing bundle rather - than an absent process group. + ``ranks`` supplies launcher placement for roles that participate in the + parallel topology. Without it, rank reads require an explicit override, + except for ``attn_dcp_rank=0`` when DCP is disabled. A process holds at most one live config: the bags always describe the engine running now. Re-publish is allowed and is **last-publish-wins** @@ -1836,26 +1679,14 @@ def publish( ), ) _CONTEXT._publish_role = role - # Zero for every process when decode context parallelism is off, which is a - # fact about the configuration and not about the spawn -- so it answers - # without a rank bundle, the way it did when it stood for a group that was - # never built. With DCP on it is a position, and the bundle below states it. + # Disabled DCP has rank zero even in processes without a rank bundle. if not _CONTEXT.parallel.dcp_enabled: _CONTEXT.parallel.override_permanently(attn_dcp_rank=0) - # Stated on the bag directly: `gpu_id` is declared but not configured, so - # it is not a leaf `override` can route, and the spawn is the only thing - # that knows it. Written whatever it is, `None` included -- most roles run - # on no device, and that is an answer rather than a name nobody wrote. + # The device is assigned by the launcher; it is not a config field. _CONTEXT.config_bag("device")._set( "gpu_id", ranks.gpu_id if ranks is not None else None ) if ranks is not None: - # The placement, worked out here rather than carried: the widths are on - # the bag a moment ago, and `world_rank` fixes the rest. A read of any - # of these then needs no process group, which is the point -- they are - # read long before one exists. The scoped overrides that swap a group - # for a draft worker sit above the record in the read chain, so a - # scope still wins. parallel = _CONTEXT.parallel placement = derive_spawn_ranks( world_rank=ranks.world_rank, @@ -1866,30 +1697,18 @@ def publish( moe_dp_size=parallel.moe_dp_size, moe_ep_size=parallel.moe_ep_size, ) - # `initialize_model_parallel` aliases the MoE-DP group to the - # attention-CP one when the CP dimension is the wider of the two, so - # this process's place in it is its CP index rather than the MoE-DP - # index the arithmetic above gives. + # MoE-DP aliases the attention-CP group when CP is wider. if parallel.moe_dp_size < parallel.attn_cp_size: placement["moe_dp_rank"] = placement["attn_cp_rank"] - # `dp_rank` is recorded whatever it is, None included: replicas are - # separate WORLD groups, so no rank implies it and `None` is the answer - # "no controller" rather than an absence. + # `None` means no data-parallel controller. placement["dp_rank"] = ranks.dp_rank placement["launch_world_rank"] = ranks.world_rank placement.update(_attention_ranks(parallel, placement["tp_rank"])) - # A DCP group is a contiguous slice of a TP group, so this process's - # place in one is its TP rank folded by that width. `attn_dcp_rank` is - # the same number, and zero where decode context parallelism is off, so - # a reader does not have to ask whether it is on first. + # DCP groups are contiguous slices of a TP group. if parallel.dcp_enabled: placement["dcp_rank"] = placement["tp_rank"] % parallel.dcp_size placement["attn_dcp_rank"] = placement.get("dcp_rank", 0) - # One stamp, not two: the identities are checked on every write, and a - # half-placed process satisfies none of them. parallel.override_permanently(**placement) - # Publish established the whole layout, so every identity applies here, - # not just the ones the stamp happened to name. _validate_parallel(parallel, "publish") if _ROLE_NS_MODE == "record": # The '-' marker distinguishes a zero-read role from a process where @@ -1906,18 +1725,7 @@ def publish( def _attention_ranks(parallel, tp_rank: int) -> dict: - """Place this process in the attention topology, from the configuration. - - The widths are already on the bag -- `publish` computed them a moment ago -- - and the rank comes from the spawn, so the position is known here, before any - process group exists. That is the point: a rank read then works in a process - that never initialises distributed, which is what the per-runner record used to provide - by being a plain frozen record. - - These are stamped rather than written as bag leaves because they are - per-process facts, and nothing about the configuration distinguishes one - rank from another. - """ + """Derive attention ranks from the configured widths and the TP rank.""" attn_tp_rank, attn_dp_rank = derive_attention_ranks( tp_rank=tp_rank, attn_tp_size=parallel.attn_tp_size, @@ -2065,13 +1873,9 @@ def restore_context(state: dict[str, Any]) -> None: def reset_context() -> None: - """Clear the context-owned store (unit-test teardown): drop the published - ``server_args`` and install fresh ``Flags`` and ``Resources``. + """Clear published configuration, parallel overrides, flags, and resources. - ``parallel`` holds the permanently-overridden derived widths, which go - with the lifecycle that set them: `_read` prefers them over the - published leaves, so leaving one behind lets the next test read the - previous topology. + Used for test teardown and runtime lifecycle reset. """ _CONTEXT._server_args = None _CONTEXT._config_bags = None diff --git a/python/sglang/srt/speculative/dflash_worker_v2.py b/python/sglang/srt/speculative/dflash_worker_v2.py index 58348432f..c55007f3d 100644 --- a/python/sglang/srt/speculative/dflash_worker_v2.py +++ b/python/sglang/srt/speculative/dflash_worker_v2.py @@ -396,11 +396,7 @@ class DFlashWorkerV2(BaseSpecWorker): self.draft_tp_context = ( draft_tp_context if get_parallel().enable_dp_attention else empty_context ) - # One decision, used twice: whether the draft runs on an attention-TP - # slice of its own. It picks how the runner is built, and then what the - # scope may say about attention every time it is entered -- a draft - # built outside the scope keeps the target's replica count and still - # gathers with it. + # Use the same attention topology during draft construction and execution. self.draft_owns_attention = get_parallel().enable_dp_attention if self.draft_owns_attention: draft_init_ctx = draft_tp_context( diff --git a/python/sglang/srt/speculative/eagle_worker_v2.py b/python/sglang/srt/speculative/eagle_worker_v2.py index 836878007..e6506504f 100644 --- a/python/sglang/srt/speculative/eagle_worker_v2.py +++ b/python/sglang/srt/speculative/eagle_worker_v2.py @@ -260,12 +260,7 @@ class EagleDraftWorker(EagleDraftWorkerBase): self._rebuild_topk1_chain_buffers() - # Load draft model weights only. - # One decision, used twice: whether the draft runs on an attention-TP - # slice of its own. It picks how the runner is built, and then what the - # scope may say about attention every time it is entered -- a draft - # built outside the scope keeps the target's replica count and still - # gathers with it. + # Use the same attention topology during draft construction and execution. self.draft_owns_attention = ( get_parallel().enable_dp_attention and self.speculative_algorithm.is_eagle3() diff --git a/python/sglang/srt/speculative/frozen_kv_mtp_worker_v2.py b/python/sglang/srt/speculative/frozen_kv_mtp_worker_v2.py index 883975ef8..bf31d835c 100644 --- a/python/sglang/srt/speculative/frozen_kv_mtp_worker_v2.py +++ b/python/sglang/srt/speculative/frozen_kv_mtp_worker_v2.py @@ -160,9 +160,7 @@ class FrozenKVMTPDraftWorker(EagleDraftWorkerBase, TpModelWorker): self.kv_context: Optional[FrozenKVMTPContext] = None - # Built above under the pipeline scope only, so this runner carries the - # target's attention topology: entering the tensor scope later swaps the - # communicator without giving the draft a replica of its own. + # Retain the target's attention topology when swapping TP groups. self.draft_owns_attention = False self.draft_tp_context = ( draft_tp_context if get_parallel().enable_dp_attention else empty_context diff --git a/python/sglang/srt/speculative/multi_layer_eagle_worker_v2.py b/python/sglang/srt/speculative/multi_layer_eagle_worker_v2.py index 67342a701..1a9d74be5 100644 --- a/python/sglang/srt/speculative/multi_layer_eagle_worker_v2.py +++ b/python/sglang/srt/speculative/multi_layer_eagle_worker_v2.py @@ -188,10 +188,7 @@ class MultiLayerEagleDraftWorker(EagleDraftWorkerBase): "InklingForConditionalGenerationMTP", "GigaChat35ForCausalLMNextN", ] - # The draft runner is built outside any tensor-parallel scope, so it - # carries the target's topology: entering the scope later swaps the - # communicator without making this process a draft with an attention - # replica of its own. It still gathers with the target's replicas. + # Retain the target's attention topology when swapping TP groups. self.draft_owns_attention = False self.draft_tp_context = ( draft_tp_context if get_parallel().enable_dp_attention else empty_context diff --git a/python/sglang/srt/speculative/standalone_worker_v2.py b/python/sglang/srt/speculative/standalone_worker_v2.py index 508f4e598..3122d98ac 100644 --- a/python/sglang/srt/speculative/standalone_worker_v2.py +++ b/python/sglang/srt/speculative/standalone_worker_v2.py @@ -88,10 +88,7 @@ class StandaloneDraftWorker(EagleDraftWorker): # Alias for better readability self.draft_runner = self.draft_worker.model_runner - # The draft runner is built outside any tensor-parallel scope, so it - # carries the target's topology: entering the scope later swaps the - # communicator without making this process a draft with an attention - # replica of its own. It still gathers with the target's replicas. + # Retain the target's attention topology when swapping TP groups. self.draft_owns_attention = False self.draft_tp_context = ( draft_tp_context if get_parallel().enable_dp_attention else empty_context diff --git a/python/sglang/srt/utils/weight_checker.py b/python/sglang/srt/utils/weight_checker.py index 4b554c6a8..80a76a431 100644 --- a/python/sglang/srt/utils/weight_checker.py +++ b/python/sglang/srt/utils/weight_checker.py @@ -70,10 +70,7 @@ def _is_non_persistent_buffer_name(name: str) -> bool: class WeightChecker: def __init__(self, *, get_model: Callable[[], Any]): self._get_model = get_model - # A check is served on demand from the scheduler loop, which is outside - # the scope that describes a draft runner. The report has to name the - # runner it was built for, so the placement is read here, at - # construction, rather than asked for when the request arrives. + # Capture the runner placement before its draft scope exits. parallel = get_parallel() self._placement = ParallelismInfo( tp_rank=parallel.tp_rank, @@ -176,9 +173,7 @@ class WeightChecker: return info.model_dump() def _parallelism_info(self) -> ParallelismInfo: - # The WORLD position is asked for now rather than frozen: unlike the - # runner's placement it is a property of the process, and an elastic - # scale-up moves it. + # Read the current WORLD rank because elastic scale-up can change it. return self._placement.model_copy( update={ "rank": dist.get_rank() if dist.is_initialized() else 0, diff --git a/python/sglang/test/kits/attention_unittest/attention_methods/dense_attention.py b/python/sglang/test/kits/attention_unittest/attention_methods/dense_attention.py index 8a2e1c6f6..0bff81879 100644 --- a/python/sglang/test/kits/attention_unittest/attention_methods/dense_attention.py +++ b/python/sglang/test/kits/attention_unittest/attention_methods/dense_attention.py @@ -22,11 +22,7 @@ from sglang.srt.model_executor.model_runner import ModelRunner from sglang.srt.runtime_context import get_context, get_parallel from sglang.srt.speculative.spec_info import SpeculativeAlgorithm -# Unit tests run without distributed initialization. Backends that size buffers by -# attention tensor-parallel degree should see the single-rank default, and a -# backend that places itself in the decode context-parallel group needs a -# position: nothing publishes here, so there is no configuration to derive the -# zero these tests run at from. +# Use single-rank attention without distributed initialization. _parallel_override = get_parallel().override(attn_tp_size=1, attn_dcp_rank=0) _parallel_override.__enter__() diff --git a/python/sglang/test/kits/attention_unittest/attention_methods/mamba2_attention.py b/python/sglang/test/kits/attention_unittest/attention_methods/mamba2_attention.py index de378e9ad..7b8c3675f 100644 --- a/python/sglang/test/kits/attention_unittest/attention_methods/mamba2_attention.py +++ b/python/sglang/test/kits/attention_unittest/attention_methods/mamba2_attention.py @@ -5,9 +5,7 @@ import torch import torch.nn.functional as F from torch import nn -# State the topology before importing modules that read it at __init__. The -# group is stated too: `RowParallelLinear.forward` asks for it to manage -# symmetric memory, and `world_size=1` short-circuits that. +# Set the single-rank topology before importing the attention implementation. from sglang.srt.runtime_context import get_context, get_parallel _parallel_override = get_parallel().override( diff --git a/python/sglang/test/test_utils.py b/python/sglang/test/test_utils.py index dac26d3e1..404c40e46 100644 --- a/python/sglang/test/test_utils.py +++ b/python/sglang/test/test_utils.py @@ -2056,19 +2056,10 @@ def maybe_stub_sgl_kernel(): @contextlib.contextmanager def published_topology(role: str = "test", *, ranks=None, **server_args_fields): - """Publish a record describing the parallel topology a test wants. + """Publish a test topology, defaulting to WORLD rank zero. - Replaces standing a per-process parallel record into the object under - test. The widths arrive the way production gets them -- from published - configuration -- and the per-process ranks the way a spawned process gets - them, so a rank read is answered without building a process group. Stating - the topology through the same door production uses also keeps the derived - widths honest: a hand-built double can claim an `attn_tp_size` the - configuration would never produce. - - `ranks` overrides the spawn identities; by default this process is rank - zero of the world, which fixes every other rank. The context is reset on exit, including when the - test fails. + ``ranks`` overrides the launcher placement. Reset the context before + publication and on exit, including when the test fails. """ from sglang.srt.runtime_context import SpawnRanks, publish, reset_context from sglang.srt.server_args import ServerArgs @@ -2085,15 +2076,10 @@ def published_topology(role: str = "test", *, ranks=None, **server_args_fields): def publish_build_topology(*, world_rank: int = 0, **server_args_fields): - """State the widths `initialize_model_parallel` is about to build at. + """Publish the topology for a subsequent ``initialize_model_parallel`` call. - The build reads every width from the runtime context, so a test that wants - a particular topology publishes it here rather than passing it in -- the - same door production uses, which also keeps the derived widths honest. - - Unlike `published_topology` this is not a scope: the groups it is about to - build outlive any block, so the configuration describing them has to as - well. Callers that tear the groups down are already resetting the process. + Preserve an existing WORLD group across the context reset. The caller is + responsible for tearing down groups and resetting the context afterward. """ from sglang.srt.distributed import parallel_state from sglang.srt.runtime_context import ( @@ -2110,11 +2096,7 @@ def publish_build_topology(*, world_rank: int = 0, **server_args_fields): role="test", ranks=SpawnRanks(world_rank=world_rank), ) - # Callers that go on to build groups have already run - # `init_distributed_environment`, which states the WORLD group -- and the - # build below places every group it creates by reading that back. The reset - # above drops it, so hand it over again: publishing a configuration does not - # unbuild a process group. + # Restore the existing WORLD handle after resetting the context. if parallel_state._WORLD is not None: get_parallel().override_permanently(world_group=parallel_state._WORLD) diff --git a/test/registered/dcp/test_reduce_scatter_along_dim.py b/test/registered/dcp/test_reduce_scatter_along_dim.py index 9bc6fa74b..b006792a3 100644 --- a/test/registered/dcp/test_reduce_scatter_along_dim.py +++ b/test/registered/dcp/test_reduce_scatter_along_dim.py @@ -131,8 +131,6 @@ def init_distributed(): local_rank=local_rank, backend="nccl", ) - # The context answers a handle from what was stated on it, not from - # this module global, so a rank stood up by hand says so itself. get_parallel().override_permanently(world_group=coord) cpu_group = coord.cpu_group diff --git a/test/registered/kernels/benchmark/communication/bench_custom_all_reduce.py b/test/registered/kernels/benchmark/communication/bench_custom_all_reduce.py index 71de22850..9e8bcb7c4 100644 --- a/test/registered/kernels/benchmark/communication/bench_custom_all_reduce.py +++ b/test/registered/kernels/benchmark/communication/bench_custom_all_reduce.py @@ -62,8 +62,6 @@ def _init_cpu_group() -> dist.ProcessGroup: local_rank=local_rank, backend="nccl", ) - # The context answers a handle from what was stated on it, not from - # this module global, so a rank stood up by hand says so itself. get_parallel().override_permanently(world_group=coord) atexit.register(dist.destroy_process_group) torch.cuda.set_stream(torch.cuda.Stream()) diff --git a/test/registered/kernels/benchmark/communication/bench_symm_mem_all_gather.py b/test/registered/kernels/benchmark/communication/bench_symm_mem_all_gather.py index 4d3e23892..d68396444 100644 --- a/test/registered/kernels/benchmark/communication/bench_symm_mem_all_gather.py +++ b/test/registered/kernels/benchmark/communication/bench_symm_mem_all_gather.py @@ -78,8 +78,6 @@ def _init_cpu_group() -> dist.ProcessGroup: local_rank=local_rank, backend="nccl", ) - # The context answers a handle from what was stated on it, not from - # this module global, so a rank stood up by hand says so itself. get_parallel().override_permanently(world_group=ps._WORLD) atexit.register(dist.destroy_process_group) logging.disable(logging.INFO) diff --git a/test/registered/kernels/benchmark/communication/bench_tp_qknorm.py b/test/registered/kernels/benchmark/communication/bench_tp_qknorm.py index 0d0452749..68bc89067 100644 --- a/test/registered/kernels/benchmark/communication/bench_tp_qknorm.py +++ b/test/registered/kernels/benchmark/communication/bench_tp_qknorm.py @@ -109,8 +109,6 @@ def _init_cpu_group() -> dist.ProcessGroup: local_rank=local_rank, backend="nccl", ) - # The context answers a handle from what was stated on it, not from - # this module global, so a rank stood up by hand says so itself. get_parallel().override_permanently(world_group=coord) atexit.register(dist.destroy_process_group) logging.disable(logging.INFO) diff --git a/test/registered/kernels/ops/communication/test_custom_all_reduce.py b/test/registered/kernels/ops/communication/test_custom_all_reduce.py index fd7984128..2e3787a51 100644 --- a/test/registered/kernels/ops/communication/test_custom_all_reduce.py +++ b/test/registered/kernels/ops/communication/test_custom_all_reduce.py @@ -130,8 +130,6 @@ def _init_cpu_group_once() -> dist.ProcessGroup: local_rank=local_rank, backend="nccl", ) - # The context answers a handle from what was stated on it, not from - # this module global, so a rank stood up by hand says so itself. get_parallel().override_permanently(world_group=coord) atexit.register(dist.destroy_process_group) cpu_group = coord.cpu_group diff --git a/test/registered/kernels/ops/communication/test_symm_mem_all_gather.py b/test/registered/kernels/ops/communication/test_symm_mem_all_gather.py index 2a2f48d5e..75f0aaf9e 100644 --- a/test/registered/kernels/ops/communication/test_symm_mem_all_gather.py +++ b/test/registered/kernels/ops/communication/test_symm_mem_all_gather.py @@ -71,8 +71,6 @@ def _init_cpu_group_once() -> dist.ProcessGroup: local_rank=local_rank, backend="nccl", ) - # The context answers a handle from what was stated on it, not from - # this module global, so a rank stood up by hand says so itself. get_parallel().override_permanently(world_group=ps._WORLD) atexit.register(dist.destroy_process_group) logging.disable(logging.INFO) diff --git a/test/registered/kernels/ops/communication/test_tp_qknorm.py b/test/registered/kernels/ops/communication/test_tp_qknorm.py index a70c6b211..deed03f49 100644 --- a/test/registered/kernels/ops/communication/test_tp_qknorm.py +++ b/test/registered/kernels/ops/communication/test_tp_qknorm.py @@ -93,8 +93,6 @@ def _init_cpu_group_once() -> dist.ProcessGroup: local_rank=local_rank, backend="nccl", ) - # The context answers a handle from what was stated on it, not from - # this module global, so a rank stood up by hand says so itself. get_parallel().override_permanently(world_group=coord) atexit.register(dist.destroy_process_group) cpu_group = coord.cpu_group diff --git a/test/registered/kernels/ops/kimi_k3/test_ar_fusion.py b/test/registered/kernels/ops/kimi_k3/test_ar_fusion.py index 229bad61d..eac0d2bd9 100644 --- a/test/registered/kernels/ops/kimi_k3/test_ar_fusion.py +++ b/test/registered/kernels/ops/kimi_k3/test_ar_fusion.py @@ -66,8 +66,6 @@ def _init_world(): local_rank=local_rank, backend="nccl", ) - # The context answers a handle from what was stated on it, not from - # this module global, so a rank stood up by hand says so itself. get_parallel().override_permanently(world_group=coord) atexit.register(dist.destroy_process_group) logging.disable(logging.INFO) diff --git a/test/registered/kernels/ops/kimi_k3/test_collectives.py b/test/registered/kernels/ops/kimi_k3/test_collectives.py index 1ccb32fb9..66d228e66 100644 --- a/test/registered/kernels/ops/kimi_k3/test_collectives.py +++ b/test/registered/kernels/ops/kimi_k3/test_collectives.py @@ -54,8 +54,6 @@ def _init_world(): local_rank=local_rank, backend="nccl", ) - # The context answers a handle from what was stated on it, not from - # this module global, so a rank stood up by hand says so itself. get_parallel().override_permanently(world_group=coord) atexit.register(dist.destroy_process_group) cpu_group = coord.cpu_group diff --git a/test/registered/kernels/ops/layernorm/test_mhc_kernels.py b/test/registered/kernels/ops/layernorm/test_mhc_kernels.py index 47e0c181b..b2010cd6c 100644 --- a/test/registered/kernels/ops/layernorm/test_mhc_kernels.py +++ b/test/registered/kernels/ops/layernorm/test_mhc_kernels.py @@ -13,12 +13,7 @@ register_cuda_ci(est_time=10, stage="base-b", runner_config="1-gpu-large") @pytest.fixture def stated_tp_group(): - """A TP group for a test that runs in a process without one. - - The production call passes the group *into* `use_symmetric_memory`, so - stubbing that context manager does not stop the read -- the argument is - evaluated first. Stating it on the context answers every spelling. - """ + """Provide a TP-group placeholder for kernels with mocked symmetric memory.""" from sglang.srt.runtime_context import get_parallel with get_parallel().override(tp_group=None): @@ -35,10 +30,7 @@ def test_mhc_fused_post_pre_matches_unfused( pytest.skip("CUDA is required for TileLang mHC kernels") monkeypatch.setattr(mhc, "is_dsa_prefill_cp_interleave", lambda: False) - # This is a single-process kernel unit test with no TP group initialized. - # mhc_pre / mhc_fused_post_pre allocate the MoE input in the symmetric-memory - # pool, which asks for the TP group; bypassing the allocation is enough, and - # then nothing asks. Mirrors the workaround in test_mxfp4_sm90_cutlass.py. + # Disable symmetric-memory allocation for this single-process kernel test. monkeypatch.setattr(mhc, "use_symmetric_memory", lambda *a, **kw: nullcontext()) monkeypatch.setattr(mhc, "is_allocation_symmetric", lambda: False) torch.manual_seed(0) diff --git a/test/registered/kernels/ops/moe/test_minimax_quant_scatter.py b/test/registered/kernels/ops/moe/test_minimax_quant_scatter.py index e905cfe9f..4b9f3621e 100644 --- a/test/registered/kernels/ops/moe/test_minimax_quant_scatter.py +++ b/test/registered/kernels/ops/moe/test_minimax_quant_scatter.py @@ -37,12 +37,7 @@ dev = "cuda" @pytest.fixture def stated_tp_group(): - """A TP group for a test that runs in a process without one. - - The production call passes the group *into* `use_symmetric_memory`, so - stubbing that context manager does not stop the read -- the argument is - evaluated first. Stating it on the context answers every spelling. - """ + """Provide a TP-group placeholder for kernels with mocked symmetric memory.""" from sglang.srt.runtime_context import get_parallel with get_parallel().override(tp_group=None): diff --git a/test/registered/layers/mamba/test_mamba2_mixer.py b/test/registered/layers/mamba/test_mamba2_mixer.py index 771536a40..463118595 100644 --- a/test/registered/layers/mamba/test_mamba2_mixer.py +++ b/test/registered/layers/mamba/test_mamba2_mixer.py @@ -129,9 +129,6 @@ def mixer2_gated_norm_tensor_parallel( ) mixer.weight.weight_loader(mixer.weight, weight) - # m2 reads tp via get_parallel().tp_size/rank — state a single-rank topology - # through the context. Every width that follows from `tp_size` is named: - # narrowing one leaf and leaving the quotients behind describes no layout. with get_parallel().override( tp_size=1, tp_rank=0, diff --git a/test/registered/ops/test_aiter_greedy_sample_amd.py b/test/registered/ops/test_aiter_greedy_sample_amd.py index a479b497c..6757dbdc6 100644 --- a/test/registered/ops/test_aiter_greedy_sample_amd.py +++ b/test/registered/ops/test_aiter_greedy_sample_amd.py @@ -37,10 +37,7 @@ def _mock_global_server_args(backend="pytorch"): class _DummyTPGroup: device_group = None - # `Sampler.__init__` asks the context for the group; state one for the rest - # of the process, since this process has no distributed init. Not the scoped - # `override()`: its context manager would be collected here and take the - # value back down with it. + # Provide a TP group for sampler initialization without distributed setup. get_parallel().override_permanently(tp_group=_DummyTPGroup()) from sglang.srt.runtime_context import get_flags diff --git a/test/registered/unit/constrained/test_grammar_manager.py b/test/registered/unit/constrained/test_grammar_manager.py index 55c824b55..9a6a96def 100644 --- a/test/registered/unit/constrained/test_grammar_manager.py +++ b/test/registered/unit/constrained/test_grammar_manager.py @@ -47,12 +47,9 @@ 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 and publish its configuration and placement. - The grammar manager reads its config and its place in the pipeline from - the context, 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`. + The caller must reset the context during teardown. """ reset_context() server_args = ServerArgs( @@ -778,9 +775,7 @@ class TestGrammarManagerPPSync(unittest.TestCase): enter_override( self, get_context().override_server_args(skip_tokenizer_init=True) ) - # After that override, not before: installing a server-args override - # re-resolves the parallel bag from defaults, which puts `pp_size` - # back to 1 whatever was published. + # Override ranks after the server-args override rebuilds the config bags. enter_scope(self, get_parallel().override(pp_size=pp_size, pp_rank=pp_rank)) scheduler.pp_group = pp_group mgr = GrammarManager(scheduler) diff --git a/test/registered/unit/disaggregation/test_register_to_bootstrap.py b/test/registered/unit/disaggregation/test_register_to_bootstrap.py index c7692cbac..312040e8d 100644 --- a/test/registered/unit/disaggregation/test_register_to_bootstrap.py +++ b/test/registered/unit/disaggregation/test_register_to_bootstrap.py @@ -204,8 +204,6 @@ class TestRegisterToBootstrap(CustomTestCase): def test_rust_attention_dp_replicates_complete_topology_across_hosts( self, mock_put ): - # The consumer reads the group through `get_parallel()`, so the - # stub is stated there rather than in the module the build writes. mock_world_group = MagicMock() success_resp = MagicMock() success_resp.status_code = 200 diff --git a/test/registered/unit/entrypoints/test_rust_server_dp_ports.py b/test/registered/unit/entrypoints/test_rust_server_dp_ports.py index 92f0ab590..a834d7b08 100644 --- a/test/registered/unit/entrypoints/test_rust_server_dp_ports.py +++ b/test/registered/unit/entrypoints/test_rust_server_dp_ports.py @@ -45,9 +45,6 @@ def test_dp_leaders_reuse_node_local_ports( server_args=SimpleNamespace(), model_config=SimpleNamespace(is_multimodal=False), ) - # Where this rank sits, stated whole: the attention rank follows - # from the TP rank and the attention-TP width, and the identities - # refuse the combination if it describes no real layout. with parallel.override( tp_rank=tp_rank, attn_dp_rank=dp_rank, diff --git a/test/registered/unit/layers/attention/test_flashattention_pa_swa_prefill_lens_size.py b/test/registered/unit/layers/attention/test_flashattention_pa_swa_prefill_lens_size.py index 2ecf5011d..43e66379d 100644 --- a/test/registered/unit/layers/attention/test_flashattention_pa_swa_prefill_lens_size.py +++ b/test/registered/unit/layers/attention/test_flashattention_pa_swa_prefill_lens_size.py @@ -93,8 +93,6 @@ def _make_prefill_aware_swa_runner( page_size=1, attn_cp_size=1, tp_size=1, - # The backend still reads the runner's frozen record for these two; - # same single-rank placement, stated where it looks for it. ps=SimpleNamespace(attn_cp_size=1, tp_size=1), is_draft_worker=False, server_args=server_args, diff --git a/test/registered/unit/layers/quantization/test_mxfp4_sm100_trtllm_gen.py b/test/registered/unit/layers/quantization/test_mxfp4_sm100_trtllm_gen.py index a47428b38..60e8f4e36 100644 --- a/test/registered/unit/layers/quantization/test_mxfp4_sm100_trtllm_gen.py +++ b/test/registered/unit/layers/quantization/test_mxfp4_sm100_trtllm_gen.py @@ -33,12 +33,7 @@ GROUP_SIZE = 32 # MXFP4 block size @pytest.fixture def stated_tp_group(): - """A TP group for a test that runs in a process without one. - - The production call passes the group *into* `use_symmetric_memory`, so - stubbing that context manager does not stop the read -- the argument is - evaluated first. Stating it on the context answers every spelling. - """ + """Provide a TP-group placeholder for kernels with mocked symmetric memory.""" from sglang.srt.runtime_context import get_parallel with get_parallel().override(tp_group=None): diff --git a/test/registered/unit/layers/quantization/test_mxfp4_sm120_cutlass.py b/test/registered/unit/layers/quantization/test_mxfp4_sm120_cutlass.py index 043642587..e358e2411 100644 --- a/test/registered/unit/layers/quantization/test_mxfp4_sm120_cutlass.py +++ b/test/registered/unit/layers/quantization/test_mxfp4_sm120_cutlass.py @@ -19,12 +19,7 @@ register_cuda_ci(est_time=14, stage="base-b", runner_config="1-gpu-small") @pytest.fixture def stated_tp_group(): - """A TP group for a test that runs in a process without one. - - The production call passes the group *into* `use_symmetric_memory`, so - stubbing that context manager does not stop the read -- the argument is - evaluated first. Stating it on the context answers every spelling. - """ + """Provide a TP-group placeholder for kernels with mocked symmetric memory.""" from sglang.srt.runtime_context import get_parallel with get_parallel().override(tp_group=None): diff --git a/test/registered/unit/layers/quantization/test_mxfp4_sm90_cutlass.py b/test/registered/unit/layers/quantization/test_mxfp4_sm90_cutlass.py index 9fc43d277..62696f011 100644 --- a/test/registered/unit/layers/quantization/test_mxfp4_sm90_cutlass.py +++ b/test/registered/unit/layers/quantization/test_mxfp4_sm90_cutlass.py @@ -65,12 +65,7 @@ GROUP_SIZE = 32 # MXFP4 block size @pytest.fixture def stated_tp_group(): - """A TP group for a test that runs in a process without one. - - The production call passes the group *into* `use_symmetric_memory`, so - stubbing that context manager does not stop the read -- the argument is - evaluated first. Stating it on the context answers every spelling. - """ + """Provide a TP-group placeholder for kernels with mocked symmetric memory.""" from sglang.srt.runtime_context import get_parallel with get_parallel().override(tp_group=None): diff --git a/test/registered/unit/managers/test_disagg_idle_step_counters.py b/test/registered/unit/managers/test_disagg_idle_step_counters.py index 8352498a1..90a769d53 100644 --- a/test/registered/unit/managers/test_disagg_idle_step_counters.py +++ b/test/registered/unit/managers/test_disagg_idle_step_counters.py @@ -64,8 +64,6 @@ def load_mlx_scheduler_module(): class TestSchedulerIdleStepCounters(CustomTestCase): def setUp(self): super().setUp() - # The loop asks the context where this process sits; nothing here - # builds a process group, so the placement arrives by publishing one. enter_scope(self, published_topology(role="scheduler")) @parameterized.expand( @@ -184,11 +182,7 @@ class TestSchedulerIdleStepCounters(CustomTestCase): f"{PDMUX_MODULE}.torch.cuda.stream", side_effect=lambda stream: nullcontext(), ), - # The prefill section runs under the duplicate communicator - # `--enable-pdmux` builds, in place of the module flag this - # replaces. The loop has no process groups at all, so stand - # one in: the scope refuses to open without it rather than - # letting prefill quietly share the decode communicator. + # PD multiplexing requires a separate prefill communicator. patch.object( parallel_state, "_PDMUX_PREFILL_TP_GROUP", diff --git a/test/registered/unit/managers/test_loadstat_wire.py b/test/registered/unit/managers/test_loadstat_wire.py index 689b21636..02a55f664 100644 --- a/test/registered/unit/managers/test_loadstat_wire.py +++ b/test/registered/unit/managers/test_loadstat_wire.py @@ -97,12 +97,10 @@ class TestLoadPublisherGating(CustomTestCase): """ def _build(self, *, config=ZMQ_ENDPOINT, explicit="auto", ranks=None, **topology): - """Construct a publisher with the socket bind stubbed out, returning - (publisher, captured _open_pub_socket mock). Opts in via explicit="auto" - by default (the feature is off without it). The topology is published - rather than overridden, so the ranks the publisher reads are the ones a - layout of that shape actually produces; every read happens in the - constructor.""" + """Return a publisher and its mocked socket factory under a published topology. + + ``explicit="auto"`` enables load publication by default. + """ with ( published_topology(ranks=ranks, **topology), patch( diff --git a/test/registered/unit/managers/test_output_streamer_customized_info.py b/test/registered/unit/managers/test_output_streamer_customized_info.py index 6614b9d71..d6b438f2e 100644 --- a/test/registered/unit/managers/test_output_streamer_customized_info.py +++ b/test/registered/unit/managers/test_output_streamer_customized_info.py @@ -103,7 +103,6 @@ class TestOutputStreamerCustomizedInfo(unittest.TestCase): ) serving_patch.start() observability_patch.start() - # The streamer asks the context which rank it is streaming from. enter_scope(self, published_topology(ranks={"dp_rank": 0})) self.addCleanup(serving_patch.stop) self.addCleanup(observability_patch.stop) diff --git a/test/registered/unit/managers/test_pp_cp_rank_offsets.py b/test/registered/unit/managers/test_pp_cp_rank_offsets.py index 0b5ca84ff..6cbf68877 100644 --- a/test/registered/unit/managers/test_pp_cp_rank_offsets.py +++ b/test/registered/unit/managers/test_pp_cp_rank_offsets.py @@ -23,11 +23,9 @@ register_cpu_ci(est_time=11, suite="base-a-test-cpu") def _published_topology(): - """The topology these tests run in. + """Publish WORLD rank 12 with TP=8 and PP=2. - World rank 12 of a `tp=8, pp=2` world is `tp_rank=4` on the second stage, - which puts this process at `attn_dp_rank=1` with `attn_tp_rank=0`: the - context derives all of them from that one number and the widths. + This gives TP rank 4, PP rank 1, attention-DP rank 1, and attention-TP rank 0. """ return published_topology( role="scheduler", diff --git a/test/registered/unit/managers/test_scheduler_chunked_abort_race.py b/test/registered/unit/managers/test_scheduler_chunked_abort_race.py index 1f4862e87..b848b69d9 100644 --- a/test/registered/unit/managers/test_scheduler_chunked_abort_race.py +++ b/test/registered/unit/managers/test_scheduler_chunked_abort_race.py @@ -50,7 +50,6 @@ def _make_scheduler(pending_req, *, chunked_req, running_reqs) -> Scheduler: class TestPendingChunkedAbortRace(CustomTestCase): def setUp(self): - # The abort path asks the context for the pipeline width. enter_scope(self, published_topology()) def test_req_left_chunked_slot_is_aborted(self): diff --git a/test/registered/unit/managers/test_scheduler_timeouts.py b/test/registered/unit/managers/test_scheduler_timeouts.py index 242d7bc73..a6c152e32 100644 --- a/test/registered/unit/managers/test_scheduler_timeouts.py +++ b/test/registered/unit/managers/test_scheduler_timeouts.py @@ -131,7 +131,6 @@ class TestWaitingTimeout(CustomTestCase): class TestRunningTimeout(CustomTestCase): def setUp(self): - # The poll asks the context for the pipeline width. enter_scope(self, published_topology()) def test_emits_only_stale_unfinished_reqs_without_marking(self): diff --git a/test/registered/unit/mem_cache/test_mamba_donated_alloc_ratio.py b/test/registered/unit/mem_cache/test_mamba_donated_alloc_ratio.py index ce4205096..b628d176b 100644 --- a/test/registered/unit/mem_cache/test_mamba_donated_alloc_ratio.py +++ b/test/registered/unit/mem_cache/test_mamba_donated_alloc_ratio.py @@ -297,7 +297,6 @@ class TestPPMambaPoolSizing(unittest.TestCase): server_args=SimpleNamespace(), spec_algorithm=SimpleNamespace(is_none=lambda: True), layer_info=SimpleNamespace(start_layer=start, end_layer=end), - # The runner carries its placement as plain attributes. attn_dp_size=1, pp_size=pp_size, hybrid_gdn_config=None, diff --git a/test/registered/unit/mem_cache/test_mem_pool_host.py b/test/registered/unit/mem_cache/test_mem_pool_host.py index 99ec163af..bcc54d0fb 100644 --- a/test/registered/unit/mem_cache/test_mem_pool_host.py +++ b/test/registered/unit/mem_cache/test_mem_pool_host.py @@ -302,10 +302,6 @@ class TestHostMemoryBudget(CustomTestCase): ) def test_ranks_per_host_divides_world_size_by_nodes(self): - # The launcher slices ranks uniformly across nodes, so the co-located - # rank count is world_size // nnodes — no hostname collective. - # tp_size=16 states the launch width the count divides -- the - # published configuration is where ranks_per_host reads it from. with ( get_context().override_server_args(nnodes=2, tp_size=16), unittest.mock.patch.object( diff --git a/test/registered/unit/model_executor/test_pool_configurator.py b/test/registered/unit/model_executor/test_pool_configurator.py index a6ff40f26..26fc85b22 100644 --- a/test/registered/unit/model_executor/test_pool_configurator.py +++ b/test/registered/unit/model_executor/test_pool_configurator.py @@ -35,8 +35,6 @@ def mock_cpu_env(kv_size=2, tp_size=1, swa_eviction_interval=4): with ( patch("torch._utils._element_size", return_value=kv_size), - # The whole attention triple, not just one leaf: a width that does not - # factor describes no layout. get_parallel().override( tp_size=tp_size, attn_tp_size=tp_size, diff --git a/test/registered/unit/model_loader/test_transformers_fallback.py b/test/registered/unit/model_loader/test_transformers_fallback.py index 36de833ee..c18d20dd0 100644 --- a/test/registered/unit/model_loader/test_transformers_fallback.py +++ b/test/registered/unit/model_loader/test_transformers_fallback.py @@ -55,8 +55,6 @@ class TestTransformersFallbackSkipSubstrs(CustomTestCase): pass with ( - # `__init__` only stashes the pipeline group, so an empty - # stand-in carries it past the read. get_parallel().override(pp_group=SimpleNamespace()), patch( "sglang.srt.models.transformers.get_hf_text_config", diff --git a/test/registered/unit/models/test_glm5_next_bfg_fusion.py b/test/registered/unit/models/test_glm5_next_bfg_fusion.py index 272c52b1e..d3b3411a5 100644 --- a/test/registered/unit/models/test_glm5_next_bfg_fusion.py +++ b/test/registered/unit/models/test_glm5_next_bfg_fusion.py @@ -83,9 +83,6 @@ class TestGlm5NextBfgFusion(unittest.TestCase): for attn_tp, rank in ((1, 0), (2, 0), (2, 1)): with ( self.subTest(route=expected_route, attn_tp=attn_tp, rank=rank), - # A width is a whole topology: the attention triple has - # to factor `tp_size`, and this process has to sit where - # the triple puts it. get_parallel().override( tp_size=4, tp_rank=rank, diff --git a/test/registered/unit/observability/test_forward_pass_metrics.py b/test/registered/unit/observability/test_forward_pass_metrics.py index 1442b4848..1f8b1e171 100644 --- a/test/registered/unit/observability/test_forward_pass_metrics.py +++ b/test/registered/unit/observability/test_forward_pass_metrics.py @@ -64,12 +64,7 @@ class _DummyPublisherThread: def _publish_server_args(test, **fields): - """Publish a config for the reporter under test and return the instance. - - The collector asks the context where this process sits, so the ranks are - stated too: without them a rank read falls through to a process group that - a unit test has not built. - """ + """Install reporter configuration and rank overrides, with test cleanup.""" fields.setdefault("decode_log_interval", 40) override = get_context().override_server_args(**fields) server_args = override.install() @@ -297,8 +292,6 @@ class TestForwardPassMetrics(unittest.TestCase): forward_pass_metrics_ipc_name=None, kv_events_config=None, ) - # The reporter asks the context whether this is the last stage, and - # which replica it is reporting for. enter_scope(self, get_parallel().override(pp_rank=0, pp_size=1, dp_rank=2)) scheduler.enable_kv_cache_events = False @@ -336,7 +329,6 @@ class TestForwardPassMetrics(unittest.TestCase): forward_pass_metrics_ipc_name=None, kv_events_config=None, ) - # The reporter asks the context whether this is the last stage. enter_scope(self, get_parallel().override(pp_rank=0, pp_size=2)) scheduler.enable_kv_cache_events = False diff --git a/test/registered/unit/server_args/test_resolution_declarations.py b/test/registered/unit/server_args/test_resolution_declarations.py index 8a95bb41d..9fe1f63e1 100644 --- a/test/registered/unit/server_args/test_resolution_declarations.py +++ b/test/registered/unit/server_args/test_resolution_declarations.py @@ -132,15 +132,7 @@ def _stash_overlay(server_args): def _live_topology_leaves(): - """Names `ParallelContext` answers from a runtime write, not the config. - - Read off the declarations that carry no `fn`, which is what those are. - Inferring them from "did the read raise" is wrong -- it raises only while - nothing has written the name, so in a process where an earlier test stated - one the property answers that value and a leaf check reads it as a config - mismatch (`parallel.tp_size: bag=1 resolution=2`). Whether a name is - shadowed is a property of the declaration, not of the process. - """ + """Return runtime-only parallel fields, identified by declarations without ``fn``.""" from sglang.srt.runtime_context import _derived_widths return frozenset(n for n, d in _derived_widths().items() if not d.fn) diff --git a/test/registered/unit/spec/test_dflash_logits.py b/test/registered/unit/spec/test_dflash_logits.py index b01ce6572..f635a7b7f 100644 --- a/test/registered/unit/spec/test_dflash_logits.py +++ b/test/registered/unit/spec/test_dflash_logits.py @@ -249,7 +249,6 @@ def test_worker_folds_a_gate_admitted_quantized_selector_head(monkeypatch): block_size=8, selector=object(), model_runner=SimpleNamespace(tp_rank=0), - # The worker rank-gates its logging on its own frozen record. ps=SimpleNamespace(tp_rank=0), draft_model=SimpleNamespace(lm_head=None), device="cpu", @@ -285,7 +284,6 @@ def test_worker_warns_once_when_selector_sampling_is_disabled(monkeypatch): _selector_sampling_enabled=False, _warned_sampling_fallback=False, model_runner=SimpleNamespace(tp_rank=0), - # The worker rank-gates its logging on its own frozen record. ps=SimpleNamespace(tp_rank=0), ) batch = SimpleNamespace(sampling_info=SimpleNamespace(is_all_greedy=False)) diff --git a/test/registered/unit/test_runtime_context.py b/test/registered/unit/test_runtime_context.py index 6cf582ad9..b7bd5faa9 100644 --- a/test/registered/unit/test_runtime_context.py +++ b/test/registered/unit/test_runtime_context.py @@ -1,4 +1,4 @@ -"""Unit tests for runtime_context: delegation, singletons, and override().""" +"""Unit tests for runtime configuration, process placement, and overrides.""" from sglang.test.ci.ci_register import register_cpu_ci @@ -66,12 +66,10 @@ _PACKAGE = _pathlib.Path(next(iter(_sglang.__path__))).resolve() def _sources(): - """Every Python file this checkout ships, package and siblings alike. + """Yield Python sources in the package and checkout siblings. - The package alone is the wrong subject set for anything about entries or - public names: `benchmark/`, `examples/` and the top-level `test/` call the - same doors and are not covered by any suite that would notice them break. - An installed package has no siblings, and then this is the package alone.""" + Installed packages without a checkout scan only the package. + """ roots = [_PACKAGE] checkout = _PACKAGE.parents[1] roots += [ @@ -85,13 +83,7 @@ def _sources(): def _scope_entries_that_say_nothing(paths): - """Draft-scope entries that do not state `owns_attention`, as `path:line`. - - The scope either narrows the draft's attention and expert identity or - leaves the target's in place, and only the worker knows which -- so the - keyword has no default. Omitting it is a `TypeError`, but only on the path - that runs, and those paths want a GPU and a draft model. - """ + """Return ``path:line`` for draft scopes missing ``owns_attention``.""" import ast missing = [] @@ -123,9 +115,7 @@ def _parallel_state(): _DP = "sglang.srt.layers.dp_attention" -#: The groups `initialize_model_parallel` states on the context, by the module -#: global it builds each one into. WORLD is not among them: it is built and -#: stated by `init_distributed_environment`, one call earlier. +# Model-parallel group names and their module globals. WORLD is initialized separately. GROUP_STAMPS = { "tp_group": "_TP", "dcp_group": "_DCP", @@ -140,12 +130,7 @@ GROUP_STAMPS = { def _groups_the_build_states() -> dict: - """What `initialize_model_parallel` hands the context, read out of its - source: `{context name: the module global it passes}`. - - Out of the source because that is the only place the whole set appears at - once -- calling the function needs ten live process groups. - """ + """Extract context-to-module group mappings from initialization source.""" import ast import inspect import textwrap @@ -193,9 +178,7 @@ class _IsolatedOverrides(CustomTestCase): class TestTheBuildStatesEveryGroup(_IsolatedOverrides): - """Nothing derives a group, so one the build forgets to state is a name - that answers "not written" for the rest of the process -- and the reader - that finds out is a model layer, a long way from here.""" + """Group initialization publishes every available handle to the context.""" def test_the_build_states_every_group_the_context_declares(self): from sglang.srt.runtime_context import _parallel_fields @@ -204,9 +187,6 @@ class TestTheBuildStatesEveryGroup(_IsolatedOverrides): self.assertEqual(declared, set(GROUP_STAMPS) | {"world_group"}) def test_the_world_group_is_stated_where_it_is_built(self): - """`initialize_model_parallel` places every group it builds by reading - `get_world_group().local_rank`, so WORLD has to be answerable before it - runs -- one function earlier, where it is constructed.""" import ast import inspect import textwrap @@ -227,11 +207,6 @@ class TestTheBuildStatesEveryGroup(_IsolatedOverrides): self.assertIn("world_group", stated) def test_nothing_reads_a_name_this_build_has_not_stated_yet(self): - """The stamp is at the end, so a getter called before it answers a name - nothing has written -- a crash at startup, in a process no unit test - runs. What the function may read is what a *previous* call stated, and - that is WORLD alone: a rank belongs to the spawn, and this build does - not get to assume the spawn ran first.""" import ast import inspect import textwrap @@ -254,9 +229,6 @@ class TestTheBuildStatesEveryGroup(_IsolatedOverrides): self.assertEqual(_groups_the_build_states(), GROUP_STAMPS) def test_a_dimension_the_configuration_has_not_got_is_left_unstated(self): - """`_DCP` is None without decode context parallelism, and every one of - these getters has always refused to answer for a group that was never - built rather than handing back a None to fail on at the collective.""" import ast import inspect import textwrap @@ -287,15 +259,7 @@ class TestParallelDelegation(_IsolatedOverrides): class TestTheTwoWorldWidths(_IsolatedOverrides): - """Two questions about the WORLD group: what it was launched at, and what - it has room for. - - Both are arithmetic over the configured leaves and are worked out at - publish, so they answer in a process that never builds the group -- which - is where a good half of the readers are. How much of that room is serving - after a scale-up is neither of them: that is elastic-EP state, asked of the - manager that owns it rather than mirrored onto this namespace. - """ + """Launch width and WORLD capacity are distinct configuration values.""" def _published(self, **fields): reset_context() @@ -307,8 +271,6 @@ class TestTheTwoWorldWidths(_IsolatedOverrides): self.assertEqual(self._published(tp_size=4, pp_size=2).launch_world_size, 8) def test_the_launch_width_spans_the_ranks_a_joiner_came_in_above(self): - """A scale joiner lays its own groups out at `tp * pp`, while its WORLD - spans the cohort already running underneath it as well.""" parallel = self._published(tp_size=4, pp_size=1, ep_join_rank_offset=8) self.assertEqual(parallel.launch_world_size, 12) @@ -321,7 +283,6 @@ class TestTheTwoWorldWidths(_IsolatedOverrides): self.assertEqual(parallel.max_world_size, 8) def test_each_width_can_be_stated_on_its_own(self): - """Stating one must not answer for the other: they are two names.""" parallel = self._published(tp_size=8) with parallel.override(launch_world_size=2): self.assertEqual(parallel.launch_world_size, 2) @@ -332,12 +293,7 @@ class TestTheTwoWorldWidths(_IsolatedOverrides): class TestSpawnIdentities(_IsolatedOverrides): - """`dp_rank` and `gpu_id` come from the spawn, because nothing else has them. - - Both vary per process while the record is identical across them, and - neither is a position in any process group -- no group has one member per - data-parallel replica. So the process entry states them at publish. - """ + """Publish records the launcher-assigned ranks and device.""" def setUp(self): super().setUp() @@ -353,9 +309,6 @@ class TestSpawnIdentities(_IsolatedOverrides): self.addCleanup(reset_context) def test_one_rank_fixes_the_rest(self): - """Every other rank is a position in a group laid out from the widths, - so `world_rank` is the whole placement: rank 5 of a `tp=4, pp=2` world - is the second stage's second device.""" publish( ServerArgs(model_path="dummy", tp_size=4, pp_size=2), role="test", @@ -368,9 +321,6 @@ class TestSpawnIdentities(_IsolatedOverrides): self.assertEqual(parallel.dp_rank, 2) def test_the_spawn_states_the_device_and_the_record_stays_clean(self): - """The parent picks the device, so it arrives with the rest of the - placement. It is stamped onto the bag: the record is the startup - input and stays as the caller handed it over.""" server_args = ServerArgs(model_path="dummy") publish( server_args, @@ -378,17 +328,12 @@ class TestSpawnIdentities(_IsolatedOverrides): ranks=SpawnRanks(world_rank=0, gpu_id=3), ) self.assertEqual(get_device().gpu_id, 3) - # Not on the record at all. An `Arg` is the operator's input and is - # collected into `ServerArgs`; nobody types this one, so it is - # declared rather than carried, and the startup input has no field - # for the spawn to have to leave alone. + # `gpu_id` is a runtime field, not a `ServerArgs` input. self.assertNotIn( "gpu_id", {f.name for f in msgspec.structs.fields(type(server_args))} ) def test_a_process_on_no_device_is_told_nothing(self): - """Most roles run on no device at all, so the bundle leaves it out and - the bag keeps the declared default rather than inventing a zero.""" publish( ServerArgs(model_path="dummy"), role="test", @@ -397,9 +342,6 @@ class TestSpawnIdentities(_IsolatedOverrides): self.assertIsNone(get_device().gpu_id) def test_no_controller_is_an_answer_not_a_failure(self): - """`dp_rank=None` means "not under a data parallel controller", which - is a fact about the deployment, unlike never having been told. The - replicas are separate WORLD groups, so no rank implies it.""" publish( ServerArgs(model_path="dummy", tp_size=2), role="test", @@ -414,7 +356,6 @@ class TestSpawnIdentities(_IsolatedOverrides): self.assertIn("rank bundle", str(caught.exception)) def test_the_attention_rank_keeps_its_own_explanation(self): - """Two stamp-only names, two different reasons to be missing.""" publish(ServerArgs(model_path="dummy", tp_size=2), role="test") with self.assertRaises(RuntimeError) as caught: get_parallel().attn_dp_rank @@ -422,13 +363,7 @@ class TestSpawnIdentities(_IsolatedOverrides): class TestAttentionRanksComeFromPublish(_IsolatedOverrides): - """With a spawn bundle, a rank read works before any group exists. - - This is what the per-runner record provided by being a plain frozen object, and - what the topology init could not: it needs the groups. Deriving at publish - is what lets a reader ask the context in a process that never initialises - distributed -- every unit test that builds a scheduler component, for one. - """ + """Published ranks are available before distributed initialization.""" def setUp(self): super().setUp() @@ -444,11 +379,6 @@ class TestAttentionRanksComeFromPublish(_IsolatedOverrides): self.addCleanup(reset_context) def test_it_matches_the_topology_init_for_every_shape(self): - """Cross-checked against the function the groups use, not restated. - - Same inputs, two callers: one has them from the configuration and the - spawn, the other from the groups it just built. - """ from sglang.srt.layers.dp_attention import compute_dp_attention_world_info shapes = [ @@ -480,7 +410,6 @@ class TestAttentionRanksComeFromPublish(_IsolatedOverrides): self.assertEqual(get_parallel().attn_dp_rank, want_dp, msg) def test_the_rank_reads_without_a_process_group(self): - """No distributed init, no patching of any getter.""" publish( ServerArgs( model_path="dummy", tp_size=8, dp_size=2, enable_dp_attention=True @@ -493,9 +422,6 @@ class TestAttentionRanksComeFromPublish(_IsolatedOverrides): self.assertEqual(get_parallel().attn_dp_rank, 1) def test_without_a_bundle_a_rank_read_says_what_is_missing(self): - """There is nothing to fall back to. Deriving one from whatever group - happens to be installed would answer a different question -- where this - process sits in that group, not where the launcher put it.""" publish(ServerArgs(model_path="dummy", tp_size=8), role="test") with self.assertRaises(RuntimeError) as caught: get_parallel().attn_tp_rank @@ -505,12 +431,7 @@ class TestAttentionRanksComeFromPublish(_IsolatedOverrides): class TestStampedRanks(_IsolatedOverrides): - """`attn_dp_rank` comes from the stamp, and says so when there is none. - - It is the one rank no group answers with: `initialize_dp_attention` - computes it from this process's `tp_rank`. Falling back to anything would - be inventing a placement for this process. - """ + """Attention-DP rank reads require an explicit runtime value.""" def setUp(self): super().setUp() @@ -528,7 +449,6 @@ class TestStampedRanks(_IsolatedOverrides): parallel = get_parallel() parallel.override_permanently(attn_dp_rank=3) self.assertEqual(parallel.attn_dp_rank, 3) - # An elastic scale-up restamps it; the newest stamp wins. parallel.override_permanently(attn_dp_rank=9) self.assertEqual(parallel.attn_dp_rank, 9) @@ -545,13 +465,6 @@ class TestStampedRanks(_IsolatedOverrides): self.assertIn("initialize_dp_attention", str(caught.exception)) def test_a_stated_width_reaches_the_padding_mode(self): - """The reason this PR exists, from a reader's side. - - `get_dp_padding_mode` reads the attention-DP width. Before the width - had one home, a scoped `override` moved the context and left the - module global answering, so stating a topology moved only half the - runtime: this asserted `SUM_LEN` with the width stated as 1. - """ from sglang.srt.layers.dp_attention import DpPaddingMode with get_parallel().override(attn_dp_size=1): @@ -560,7 +473,6 @@ class TestStampedRanks(_IsolatedOverrides): ) self.assertIs(mode, DpPaddingMode.MAX_LEN) - # And the branch it would have taken with the target's width. with get_parallel().override(attn_dp_size=2): mode = DpPaddingMode.get_dp_padding_mode( is_extend_in_batch=True, global_num_tokens=[3, 5] @@ -568,10 +480,6 @@ class TestStampedRanks(_IsolatedOverrides): self.assertIs(mode, DpPaddingMode.SUM_LEN) def test_the_gather_slot_follows_the_list_that_was_gathered(self): - """The DP sync gathers over the attention-DP replicas, or over the - expanded WORLD once a scale-up has moved the gather there. The index - into that list is a property of the gather, so it is read beside the - flag that says which one happened rather than kept on the topology.""" from sglang.srt.layers.dp_attention import dp_gather_slot self.addCleanup(reset_context) @@ -605,13 +513,10 @@ class TestStampedRanks(_IsolatedOverrides): dp_flags.joiner_skip_all_gather = False parallel.override_permanently(ep_join_rank_offset=8) self.assertEqual(dp_gather_slot(), 8 + parallel.tp_rank) - # and the topology it was read off is untouched self.assertEqual(parallel.attn_dp_size, 8) self.assertEqual(parallel.tp_size, 8) def test_a_scale_up_writes_no_width(self): - """The identities stay unconditional because nothing overrides them: - the scale-up only points the gather at the expanded WORLD.""" from sglang.srt.layers.dp_attention import update_dp_attention_post_scale dp_flags = get_flags().dp @@ -635,13 +540,7 @@ class TestStampedRanks(_IsolatedOverrides): class TestEveryDeclaredParallelNameIsStatable(_IsolatedOverrides): - """The overridable set is read from the declarations, not maintained by hand. - - A hand-kept list can hold a name the class does not answer, or miss one it - does; either way `override()` refuses or accepts the wrong thing with - nothing to say so. The three tests below check the set against the - declarations from both sides. - """ + """Overrides accept exactly the declared parallel fields.""" def test_every_declared_name_can_be_stated_and_reads_back(self): from sglang.srt.runtime_context import _parallel_fields @@ -656,12 +555,6 @@ class TestEveryDeclaredParallelNameIsStatable(_IsolatedOverrides): self.assertIs(getattr(parallel, name), sentinel, msg=name) def test_every_name_the_class_answers_for_is_in_the_set(self): - """Cross-check from the other side: the class's own surface. - - Derived from the class rather than from the same declarations the set - is built from, so a source dropped out of `_parallel_fields` shows up - here instead of agreeing with itself. - """ from sglang.srt.runtime_context import _parallel_fields answered = { @@ -673,13 +566,6 @@ class TestEveryDeclaredParallelNameIsStatable(_IsolatedOverrides): self.assertEqual(answered - _parallel_fields(), set()) def test_a_live_name_is_never_also_answered_from_the_bag(self): - """The two answer differently, so a name in both would make the read - order -- not the declaration -- decide which one a caller gets. - - The bag carries the declared quotients as well as the operator's - leaves, and both are ahead of the live getter once a configuration is - published: a declared name that is also a leaf would answer - from the getter before publish and from the bag after.""" from sglang.srt.runtime_context import ( _derived_widths, _parallel_config_leaves, @@ -694,13 +580,9 @@ class TestEveryDeclaredParallelNameIsStatable(_IsolatedOverrides): class TestReadsWithoutAPublishedConfig(_IsolatedOverrides): - """The namespace has to answer in a process that publishes nothing. + """Overrides support shared SRT layers without published SRT configuration. - `multimodal_gen` lends its own TP group to shared `srt` layers from a - process with no `srt` config to publish against, and those layers ask for - `attn_tp_size` anyway -- through code `multimodal_gen` does not own, which - is why grepping that package for `get_parallel()` finds nothing while the - read plainly happens. + Multimodal generation supplies its own TP group and widths this way. """ def setUp(self): @@ -734,20 +616,12 @@ class TestReadsWithoutAPublishedConfig(_IsolatedOverrides): self.assertEqual(parallel.moe_tp_size, 2) def test_an_unstamped_width_still_names_the_cause(self): - """Without a stamp there is nothing to answer with, and the failure - has to say so rather than invent a width.""" with self.assertRaisesRegex(RuntimeError, r"not available"): get_parallel().attn_tp_size class TestPrivateAttributeProbing(_IsolatedOverrides): def test_probing_a_private_name_does_not_recurse(self): - """`copy` and `pickle` probe for hooks before `__init__` has run. - - `__getattr__` reaches for `self._config`, so if it did not refuse - underscore names outright, probing one on a half-built instance would - recurse until the stack ran out. - """ fresh = ParallelContext.__new__(ParallelContext) # slots unset for probe in ("_config", "_stamp", "_overrides", "__deepcopy__"): with self.assertRaises(AttributeError, msg=probe): @@ -760,14 +634,7 @@ class TestPrivateAttributeProbing(_IsolatedOverrides): class TestAWidthReadStaysTraceable(_IsolatedOverrides): - """A width read inside compiled model code must stay inside the graph. - - Shared layers read widths inside a compiled forward. A graph break there - is a performance regression and nothing else -- every suite stays green - through it -- so `fullgraph=True` is what turns it into a failure. This - pins the read path, whichever form it takes: the sibling leaf test - compiles names served by `__getattr__` and they trace too. - """ + """Parallel width reads must trace under ``torch.compile(fullgraph=True)``.""" def test_a_width_read_compiles_into_the_graph(self): import torch @@ -825,13 +692,7 @@ class TestParallelOverride(_IsolatedOverrides): class TestParallelDCP(_IsolatedOverrides): - """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. - """ + """DCP widths come from configuration; active DCP ranks come from publication.""" def _published(self, **fields): reset_context() @@ -860,27 +721,17 @@ class TestParallelDCP(_IsolatedOverrides): return get_parallel() def test_the_dcp_rank_is_where_the_tp_rank_falls_in_its_slice(self): - """A DCP group is a contiguous slice of the TP group, so the place in - one is the TP rank folded by the width.""" parallel = self._placed(5, tp_size=8, dcp_size=4) self.assertEqual(parallel.dcp_rank, 1) self.assertEqual(parallel.attn_dcp_rank, 1) def test_the_gated_off_rank_answers_without_a_spawn_bundle(self): - """Zero for every process when decode context parallelism is off, so a - reader on a path that never publishes a bundle -- a memory pool, an - attention backend built in a unit test -- still gets an answer. It - stood for a group that was never built before, and it has to keep - answering the same way.""" reset_context() self.addCleanup(reset_context) publish(ServerArgs(model_path="dummy", tp_size=8), role="test") self.assertEqual(get_parallel().attn_dcp_rank, 0) def test_the_dcp_rank_is_gated_on_a_width_the_configuration_carries(self): - """Zero where decode context parallelism is off, so a reader does not - have to ask whether it is on before asking where it sits -- and the - gated-off name is not answered at all, because no group holds it.""" parallel = self._placed(5, tp_size=8, dcp_size=1) self.assertFalse(parallel.dcp_enabled) self.assertEqual(parallel.attn_dcp_rank, 0) @@ -912,7 +763,7 @@ class _IsolatedServerArgs(CustomTestCase): class TestServerArgsOwnership(_IsolatedServerArgs): - """V2b: the context owns the slot; the legacy getters are identity shims.""" + """The context owns ServerArgs; supported legacy accessors share its storage.""" def test_legacy_setter_publishes_into_context(self): # Identity, not equality: the slot holds the very object published. @@ -2083,13 +1934,7 @@ class TestParallelLeafReads(_IsolatedServerArgs): class TestDerivedWidths(_IsolatedOverrides): - """The widths no flag sets are computed from the leaves and permanently - overridable. - - `attn_tp_size` and its siblings used to be read back off the group - coordinator that was built from them, which made the answer depend on - distributed init and, after an elastic scale, disagree with the leaves. - """ + """Derived widths use configuration unless explicitly overridden.""" def setUp(self): super().setUp() @@ -2129,10 +1974,6 @@ class TestDerivedWidths(_IsolatedOverrides): 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 caller states one, and naming - only some of them is refused: the caller owns the arithmetic, the - context only checks it.""" reset_context() self.addCleanup(reset_context) publish(ServerArgs(model_path="dummy", tp_size=8), role="test") @@ -2161,12 +2002,6 @@ class TestDerivedWidths(_IsolatedOverrides): self.assertIn("not available", str(caught.exception)) def test_a_permanent_override_and_a_live_group_both_win_over_the_leaves(self): - """Order is scoped override, then the stamp, then the published leaf. - - No group is consulted for a width -- `test_the_group_is_never_consulted` - in this class asserts that -- so a stamp is what an elastic scale-up - leaves behind, and the leaf answers only where there is none. - """ parallel = get_parallel() parallel.override_permanently(attn_tp_size=7) self.addCleanup(parallel.clear_stamp) @@ -2188,11 +2023,6 @@ class TestDerivedWidths(_IsolatedOverrides): self.assertEqual(widths["attn_dcp_size"], 1) def test_no_world_width_is_a_quotient_of_the_leaves(self): - """Deriving one would answer with the joining cohort's own width on a - scale joiner, which lays its groups out at `tp * pp` while WORLD spans - `ep_join_rank_offset + tp * pp`. The launch width comes off the group - that was actually built; the ceiling is not this function's to give - either, and `TestTheTwoWorldWidths` says where each comes from.""" widths = derive_parallel_widths( tp_size=4, attn_cp_size=1, @@ -2206,8 +2036,7 @@ class TestDerivedWidths(_IsolatedOverrides): {name for name in widths if "world" in name}, set(), ) - # Its own arithmetic spans the offset, which is the leaf the quotients - # above are not given and could not account for. + # WORLD includes the ranks below the joining cohort. from sglang.srt.runtime_context import launch_world_size_of self.assertEqual( @@ -2218,12 +2047,6 @@ class TestDerivedWidths(_IsolatedOverrides): ) def test_the_bare_name_is_gone(self): - """It answered two questions, so every reader had to remember which. - - Both spellings fail: reading it, and stating it -- the overridable set - is derived from the same declarations the read path is, so a name that - cannot be read cannot be stated either. - """ with self.assertRaisesRegex(AttributeError, r"has no 'world_size'"): get_parallel().world_size with self.assertRaisesRegex(ValueError, r"unknown parallel field"): @@ -2284,12 +2107,6 @@ class TestDerivedWidths(_IsolatedOverrides): self.assertEqual(parallel.attn_dp_size, 1) def test_reset_context_drops_the_permanent_override(self): - """The permanent override belongs to the lifecycle that made it. - - `_read` prefers it over the published leaf, so one that - outlived `reset_context()` would let the next test read the previous - topology. - """ parallel = get_parallel() parallel.override_permanently(attn_tp_size=4) self.assertEqual(parallel.attn_tp_size, 4) @@ -2320,27 +2137,6 @@ class TestDerivedWidths(_IsolatedOverrides): self.assertEqual(attn_dp_size, widths["attn_dp_size"]) def test_recomputing_from_published_leaves_matches_the_publish_bag(self): - """`initialize_model_parallel` no longer overrides anything -- see - `test_initialize_model_parallel_no_longer_touches_the_bag` below -- - so every real caller must forward leaves that already match its own - published config, because nothing corrects a mismatch anymore. - `scheduler.py`'s `ps.attn_dp_size`/`ps.moe_ep_size`/etc, and the - weight-cache daemon's own already-published config, both do -- this - pins that the formula they'd recompute from those leaves - (`derive_attention_widths`, `derive_parallel_widths`, the same ones - `publish` itself used) agrees with what's already in the bag, across - the widths `test_the_rank_helper_agrees_with_the_override` does not - vary -- moe_ep_size, moe_dp_size, and dcp_size -- using real - `publish()`. - - A caller that does NOT keep the two in sync is a bug in that caller, - not something this framework silently corrects: two real ones existed - (`test/registered/eplb/test_lplb_distributed.py` and - `test/manual/ep/test_flashinfer_dispatcher.py`, both publishing a - placeholder config and then building real groups at a width it never - reflected) and were fixed by publishing the actual width instead of - relying on a correction to paper over the mismatch. - """ shapes = ( dict(tp_size=8), dict(tp_size=8, dp_size=2, enable_dp_attention=True), @@ -2378,11 +2174,6 @@ class TestDerivedWidths(_IsolatedOverrides): self.assertEqual(published, recomputed) def test_initialize_model_parallel_builds_at_the_published_widths(self): - """The build takes every width from the context rather than from an - argument, so "published one width, built another" is no longer a state - a caller can reach -- there is nothing left to translate, and nothing - to correct afterwards either. - """ from unittest.mock import Mock from sglang.srt.distributed import parallel_state @@ -2423,7 +2214,6 @@ class TestDerivedWidths(_IsolatedOverrides): parallel_state.initialize_model_parallel() self.addCleanup(parallel_state.destroy_model_parallel) - # The first group built is TP, one group spanning the published width. self.assertEqual(built_at[0], [list(range(world_size))]) self.assertEqual(get_parallel().attn_tp_size, world_size) @@ -2454,9 +2244,6 @@ class TestTheDerivedHalfIsDeclared(CustomTestCase): ) def test_a_computed_name_names_the_function_that_computes_it(self): - """The declaration is not a second list to keep in step: a name that - is a function of the leaves points at a function called after it, and - that function existing is the whole of what publish needs.""" import importlib from sglang.srt.runtime_context import _derived_widths @@ -2469,9 +2256,6 @@ class TestTheDerivedHalfIsDeclared(CustomTestCase): self.assertTrue(callable(getattr(importlib.import_module(module), attr))) def test_the_arithmetic_produces_nothing_that_is_not_declared(self): - """The other side of it: a key the derivation returns and no - declaration names is a width the namespace never answers with, and the - group build re-states it into a name nobody can read.""" from sglang.srt.runtime_context import _derived_widths produced = set( @@ -2502,14 +2286,7 @@ class TestTheDerivedHalfIsDeclared(CustomTestCase): class TestAnEntryThatBuildsARunnerHandsOverItsPlacement(CustomTestCase): - """`ModelRunner.__init__` reads a recorded identity, so an entry that - publishes without a bundle and then builds one fails at construction. - - Every such entry is an `__main__`-reachable path, so nothing in the unit - suite exercises it; the benchmark entry was found this way rather than by - a test. This walks the sources instead: a module that publishes and builds - a runner has to pass `ranks=`. - """ + """Runner entry points must publish launcher placement.""" def test_every_publisher_that_builds_a_runner_passes_a_bundle(self): import ast as _ast @@ -2545,46 +2322,31 @@ class TestAnEntryThatBuildsARunnerHandsOverItsPlacement(CustomTestCase): class TestTheAccessorsHaveNoCallersOutsideTheirPackage(CustomTestCase): - """`parallel_state`'s getters are the definition, not a second spelling. + """SRT callers outside ``distributed`` use the runtime context. - Business code asks `get_parallel()`; a call that goes straight to the getter - is a read the context cannot redirect, which is what a scope needs it to be - able to do. The package that defines them is exempt -- a read there would - go through the context back into itself -- and so is `multimodal_gen`, which - has its own parallel state. + Multimodal generation has its own parallel state and is excluded. """ - #: May have callers. `get_self_pp_group` builds the single-rank group a - #: draft pipeline scope installs, so there is nothing for the context to - #: answer with until the scope has installed it; the other two are not - #: topology at all. + # Group constructors and non-topology helpers may have callers. ALLOWED = { "get_self_pp_group", "get_default_distributed_backend", "get_mooncake_transfer_engine", } - #: Zero callers required, but not deprecated either: the context has no - #: name that answers the same question. - #: - #: The three widths read a group the build does not check against the - #: configuration, so "the group's width" and "the configured width" are two - #: facts -- the MoE-DP group is the attention-CP group when the latter is - #: wider, and the other two are simply not pinned yet. Pinning them in - #: `_WIDTH_AND_GROUP` is what would let them move. + # Require zero callers, but do not deprecate: these getters have no + # equivalent context field. Group widths may differ from configured widths. NOT_ANSWERED_BY_THE_CONTEXT = { "get_moe_data_parallel_world_size", "get_moe_tensor_parallel_world_size", "get_dcp_world_size", - # Answers `None` where the context asserts, which is the whole point of - # the caller that wants it. + # Answers `None` where the context asserts. "get_dcp_group_no_assert", "get_torch_distributed_pg_options", } def _accessors(self): - """Derived from the source, not listed here: a guard whose subject set - is written by hand stops watching whatever gets added next.""" + """Find public getters defined in the parallel-state source.""" from sglang.srt.distributed import parallel_state as parallel_state_module source = _pathlib.Path(parallel_state_module.__file__).read_text().splitlines() @@ -2595,10 +2357,7 @@ class TestTheAccessorsHaveNoCallersOutsideTheirPackage(CustomTestCase): } def _callers(self, name): - """Every call in business code, including one hiding behind an import - alias -- `from ... import get_moe_dp_group as _g` then `_g()` is the - same reach past the context, and searching for the original spelling - alone reports zero while it is right there.""" + """Find getter calls outside the defining package, including import aliases.""" import re from sglang.srt.distributed import parallel_state as parallel_state_module @@ -2638,9 +2397,7 @@ class TestTheAccessorsHaveNoCallersOutsideTheirPackage(CustomTestCase): "context cannot answer them", ) - #: How many callers each exempt accessor has outside the defining package. - #: A ratchet, not a description: these may go down and never up, and a name - #: that reaches zero comes off the list. Anything not here must have none. + # Maximum caller counts; reduce these as callers migrate to the context. ALLOWED_CALLERS = { "get_self_pp_group": 1, "get_default_distributed_backend": 1, @@ -2648,10 +2405,6 @@ class TestTheAccessorsHaveNoCallersOutsideTheirPackage(CustomTestCase): } def test_the_exempt_accessors_do_not_grow_new_callers(self): - """The zero-caller rule above cannot cover the three that are not - topology, so they get a count instead. Ratchets only turn one way: a - number that has to go up means a new business-code reader of a name the - context should be answering.""" for name, allowed in sorted(self.ALLOWED_CALLERS.items()): callers = self._callers(name) self.assertLessEqual( @@ -2663,10 +2416,6 @@ class TestTheAccessorsHaveNoCallersOutsideTheirPackage(CustomTestCase): ) def test_every_getter_the_context_answers_is_deprecated(self): - """The other half of the ratchet: the deprecation set is derived from - the table that maps a context name to the getter behind it, so dropping - a getter out of that table would quietly take it off the list. This - fails if one of them stops being marked.""" from sglang.srt.distributed import parallel_state marked = set(parallel_state._CONTEXT_NAME_OF) @@ -2676,9 +2425,6 @@ class TestTheAccessorsHaveNoCallersOutsideTheirPackage(CustomTestCase): for name in sorted(unclassified): if name in marked: continue - # Not answered by the context and not exempt: a getter that is - # neither is a name with no home, which is what this module exists - # to prevent. self.assertIn( name, marked, @@ -2687,9 +2433,6 @@ class TestTheAccessorsHaveNoCallersOutsideTheirPackage(CustomTestCase): ) def test_calling_one_from_outside_the_package_is_deprecated(self): - """The getters stay -- they are the definition -- but a call that comes - from outside the package that defines them cannot be redirected by a - scope, so it says what to read instead.""" import warnings from sglang.srt.distributed import parallel_state @@ -2711,10 +2454,6 @@ class TestTheAccessorsHaveNoCallersOutsideTheirPackage(CustomTestCase): ) def test_nothing_the_context_answers_with_calls_back_into_the_package(self): - """The read path reaches the stored group, not the getter that used to - wrap it -- which is what lets the getters be deprecated without the - replacement tripping the warning meant for people who bypass it.""" - import inspect from sglang.srt.distributed import parallel_state @@ -2724,18 +2463,12 @@ class TestTheAccessorsHaveNoCallersOutsideTheirPackage(CustomTestCase): self.assertTrue(written, "no written-at-runtime names; this proves nothing") self.assertIn("tp_group", written) - # The read path is lookups only -- override, stamp, bag. Nothing in it - # can reach a getter, which is what lets them be deprecated without the - # replacement tripping the warning meant for people who bypass it. body = inspect.getsource(ParallelContext._read) self.assertNotIn("_ps()", body) self.assertNotIn("parallel_state", body) self.assertNotIn("sglang.srt.runtime_context", parallel_state._EXEMPT_CALLERS) def test_a_scope_reaches_callers_that_went_straight_to_the_getter(self): - """The getters read the context, so redirecting a group redirects them - too. PD multiplexing needs exactly this: the in-package readers have to - follow the prefill communicator, not just the ones asking the context.""" from sglang.srt.distributed import parallel_state stand_in = SimpleNamespace(world_size=1, rank_in_group=0) @@ -2744,8 +2477,6 @@ class TestTheAccessorsHaveNoCallersOutsideTheirPackage(CustomTestCase): self.assertIs(parallel_state.get_tp_group(), stand_in) def test_the_package_that_defines_them_is_not_warned_at(self): - """`srt/distributed/` keeps calling them: a read there would go through - the context back into itself.""" import warnings from sglang.srt.distributed import parallel_state @@ -2772,21 +2503,11 @@ class TestTheAccessorsHaveNoCallersOutsideTheirPackage(CustomTestCase): self.assertEqual([str(w.message) for w in seen], []) def test_the_guard_would_notice_a_caller(self): - """The subject set is derived, so this checks the search finds a real - call rather than that the list happens to be empty: `get_self_pp_group` - is exempt and does have one caller.""" self.assertTrue(self._callers("get_self_pp_group")) class TestTheTopologyIdentities(CustomTestCase): - """One set of identities, checked wherever the layout is written. - - Each one is injected in the direction that breaks it and in the direction - that keeps it: a guard that only ever fires is as uninformative as one that - never does. They hold unconditionally -- a caller that states one leaf owes - the quotients that follow from it, because a namespace describing no real - layout is what the guard exists to refuse. - """ + """Validate topology consistency at publication and override boundaries.""" def _publish_square(self): """tp=4 over two attention-DP replicas of two: every identity holds.""" @@ -2801,8 +2522,6 @@ class TestTheTopologyIdentities(CustomTestCase): ) def test_a_published_topology_is_consistent(self): - """The quiet direction, and the reason publish can check everything: - it is the one write that establishes the whole layout.""" self._publish_square() parallel = get_parallel() self.assertEqual(parallel.tp_size, 4) @@ -2841,9 +2560,6 @@ class TestTheTopologyIdentities(CustomTestCase): self.assertIn("4 != 1 * 1 * 3", message) def test_a_rank_the_attention_layout_cannot_produce_is_refused(self): - """`tp_rank` is not free of the attention ranks: the layout derives one - from the other, so a set that does not satisfy it places this process - in two different seats at once.""" self._publish_square() with self.assertRaises(ValueError) as caught: with get_parallel().override( @@ -2858,9 +2574,6 @@ class TestTheTopologyIdentities(CustomTestCase): self.assertIn("attn_tp_size + attn_tp_rank", str(caught.exception)) def test_the_published_ranks_satisfy_the_layout(self): - """The quiet direction for the same identity: publish derives the - attention ranks from `tp_rank` through that very equation, so a - published process always sits in one seat.""" self._publish_square() parallel = get_parallel() self.assertEqual( @@ -2871,9 +2584,6 @@ class TestTheTopologyIdentities(CustomTestCase): ) def test_a_refused_write_leaves_nothing_behind(self): - """The scope never opened, so the value it tried to state must not be - readable afterwards -- a half-applied override is the state this guard - exists to prevent.""" self._publish_square() with self.assertRaises(ValueError): with get_parallel().override( @@ -2883,11 +2593,6 @@ class TestTheTopologyIdentities(CustomTestCase): self.assertEqual(get_parallel().attn_tp_size, 2) def test_a_group_built_at_another_width_is_refused(self): - """The other end of the same identity: what the configuration says and - what the coordinators were actually built at. Stating a group is how - the build hands it over, so that write is where the disagreement - surfaces -- still attributable to the build, and before a collective - runs on the wrong peers.""" from sglang.srt.distributed.parallel_state import GroupCoordinator self._publish_square() @@ -2899,8 +2604,6 @@ class TestTheTopologyIdentities(CustomTestCase): message = str(caught.exception) self.assertIn("tp_group.world_size == tp_size", message) self.assertIn("built 8, configured 4", message) - # The refused write left nothing behind: the name is unwritten, not - # holding a group no identity accepts. with self.assertRaises(RuntimeError): get_parallel().tp_group @@ -2915,8 +2618,6 @@ class TestTheTopologyIdentities(CustomTestCase): self.assertIs(get_parallel().tp_group, right) def test_a_draft_scope_states_a_consistent_topology(self): - """The scope narrows four names at once, so the identity applies to it - -- and holds, which is what lets the guard stay on.""" from sglang.srt.distributed import parallel_state from sglang.srt.distributed.parallel_state import GroupCoordinator @@ -2929,12 +2630,7 @@ class TestTheTopologyIdentities(CustomTestCase): class TestTheParallelPhase(CustomTestCase): - """Publish says what the topology is; one phase builds it, once. - - Before this, whichever runner was constructed first brought the groups up - on its way past, so whether they existed depended on construction order -- - and a draft runner, which must not build them, went down the same path. - """ + """Entry points initialize parallel groups before constructing runners.""" def test_building_twice_is_refused(self): from sglang.srt.distributed import bootstrap @@ -2970,9 +2666,6 @@ class TestTheParallelPhase(CustomTestCase): self.assertIn("ran twice", str(caught.exception)) def test_every_publisher_that_builds_a_runner_runs_the_phase(self): - """The companion to the bundle census: an entry that publishes and then - builds a runner has to bring the parallel runtime up - itself, because the runner no longer does it on the way past.""" import ast as _ast offenders = [] @@ -3000,12 +2693,9 @@ class TestTheParallelPhase(CustomTestCase): class TestWhoAnswersDuringADraftScope(CustomTestCase): - """A draft worker runs in one process with the target, under a scope. + """Draft scopes expose a consistent topology. - Two things have to hold for that to be workable, and neither is visible - from a single read: inside the scope every source agrees on the draft's - shape, and a reader that runs *outside* it still gets the draft's answer - from whatever it carried out. + Objects constructed in a draft scope retain their placement after it exits. """ def _single_member_group(self): @@ -3027,9 +2717,6 @@ class TestWhoAnswersDuringADraftScope(CustomTestCase): ) def test_the_pipeline_swap_states_every_member_it_installs(self): - """`pp_size` is a configured leaf: unlike `pp_rank` it does not follow - the group being swapped underneath, so a scope that installs a group - without stating its width reports the target's.""" from sglang.srt.distributed import parallel_state group = self._single_member_group() @@ -3051,9 +2738,6 @@ class TestWhoAnswersDuringADraftScope(CustomTestCase): return group def test_the_tensor_swap_states_the_draft_has_no_attention_replica(self): - """The draft runs the whole model on the group being installed. Its - attention identity is therefore that group, with one replica -- while - the target this process also serves is attention-DP over four ranks.""" from sglang.srt.distributed import parallel_state reset_context() @@ -3078,12 +2762,8 @@ class TestWhoAnswersDuringADraftScope(CustomTestCase): self.assertEqual(parallel.attn_dp_rank, 0) self.assertEqual(parallel.attn_cp_size, 1) self.assertEqual(parallel.attn_cp_rank, 0) - # `dp_size` is the deployment's replica count, not a property of - # the group being installed, so the scope leaves it alone -- - # `require_mlp_tp_gather` asserts on it under dp attention. + # The scope leaves the deployment's replica count alone. self.assertEqual(parallel.dp_size, 2) - # The whole point of stating the rest: the identity the override - # path and the group build both check holds in here. self.assertEqual( parallel.tp_size, parallel.attn_tp_size * parallel.attn_dp_size * parallel.attn_cp_size, @@ -3092,12 +2772,6 @@ class TestWhoAnswersDuringADraftScope(CustomTestCase): self.assertEqual(get_parallel().dp_size, 2) def test_every_caller_says_whether_the_draft_owns_its_attention(self): - """The scope cannot work it out from the group it is handed: the same - call site passes an attention-TP slice for one draft and the target's - whole TP group for another, and the two want opposite answers. So the - worker states it, and a caller that forgets is the bug this catches -- - `owns_attention` has no default, but a missing one is a TypeError only - on the path that runs, and these paths need a GPU and a draft model.""" self.assertEqual( _scope_entries_that_say_nothing(_sources()), [], @@ -3105,10 +2779,6 @@ class TestWhoAnswersDuringADraftScope(CustomTestCase): ) def test_the_census_would_notice_one(self): - """Two ways for it to report zero and still be wrong: the matcher does - not recognise the call, or the walk never reaches the file. A draft - scope entered from `benchmark/` breaks the same way as one in the - package and no suite covers it, so the roots are part of the check.""" trees = { part for path in _sources() @@ -3130,13 +2800,6 @@ class TestWhoAnswersDuringADraftScope(CustomTestCase): self.assertEqual(_scope_entries_that_say_nothing([probe]), [f"{probe}:1"]) def test_a_full_width_swap_leaves_the_attention_layout_alone(self): - """The other caller. A draft built outside any scope carries the - target's whole TP group, and the graph capture installs *that* -- so - the process is still one of two attention-DP replicas and still gathers - with the other one. Narrowing here would claim a replica count it does - not have, and the reader that acts on it is a collective: the DP gather - takes its buffer size from the replica count and its communicator from - this group, so the two stop agreeing and the all-gather is refused.""" from sglang.srt.distributed import parallel_state reset_context() @@ -3160,11 +2823,6 @@ class TestWhoAnswersDuringADraftScope(CustomTestCase): self.assertEqual(parallel.dp_size, 2) def test_a_report_built_for_a_runner_follows_that_runner(self): - """A weight check is an on-demand request served from the scheduler - loop, so it runs outside the scope that describes a draft runner. Its - report has to name the runner it was built for, which is why it reads - the placement once, where it is built, instead of asking again when the - request arrives.""" from sglang.srt.distributed import parallel_state from sglang.srt.utils.weight_checker import WeightChecker @@ -3174,20 +2832,15 @@ class TestWhoAnswersDuringADraftScope(CustomTestCase): with parallel_state.patch_pipeline_parallel_group(group): checker = WeightChecker(get_model=lambda: None) - # The scope has closed and the context answers the target's shape again. self.assertEqual(get_parallel().pp_size, 2) info = checker._parallelism_info() self.assertEqual((info.pp_rank, info.pp_size), (0, 1)) class TestTheRecordIsNeverWrittenTo(CustomTestCase): - """`server_args` is the startup record; the bags are the truth afterwards. + """Runtime configuration changes use ``RuntimeContext.override``. - Writing a field onto it after `resolve_once()` has sealed it puts a second - answer where there is supposed to be one, and it is invisible to anything - reading the bag. The sanctioned writer is `RuntimeContext.override`, which - writes the bag and says so in its own contract. `arg_groups/` is exempt: it - is the resolution pipeline, so building the record is its job. + Only the resolution pipeline in ``arg_groups`` may write ``ServerArgs``. """ #: Assignments here are the record being built, not mutated behind a reader. @@ -3233,10 +2886,7 @@ class TestTheRecordIsNeverWrittenTo(CustomTestCase): class TestTheRetiredNamesAreGoneEverywhere(CustomTestCase): - """The package stopped re-exporting the getters and the build stopped - taking widths. Both are import-time or call-time failures in whatever tree - they survive in, and the trees beside the package have no suite to notice. - """ + """Reject retired getter imports and group-initialization width arguments.""" def _retired(self): from sglang.srt.distributed.parallel_state import _CONTEXT_NAME_OF @@ -3265,8 +2915,6 @@ class TestTheRetiredNamesAreGoneEverywhere(CustomTestCase): ) def test_nothing_passes_a_width_to_the_build(self): - """`multimodal_gen` is out: it has a function of this name that builds - its own parallelism from its own degrees.""" import ast as _ast import inspect @@ -3299,14 +2947,7 @@ class TestTheRetiredNamesAreGoneEverywhere(CustomTestCase): class TestNothingReadsThePlacementBeforeItIsFrozen(CustomTestCase): - """`ModelRunner.__init__` freezes its placement partway through. - - A method called before that point reads an attribute that does not exist - yet, and only on the configuration that reaches it -- a remote weight - transporter, a NUMA binding -- so the suites say nothing and a GPU job is - where it surfaces. The order is what makes it wrong, so the order is what - is checked. - """ + """Runner initialization must set placement attributes before reading them.""" def _model_runner(self): import ast as _ast @@ -3402,7 +3043,6 @@ class TestNothingReadsThePlacementBeforeItIsFrozen(CustomTestCase): ) def test_the_census_would_notice_one(self): - """The positive control: the walk has to find a read that is there.""" import ast as _ast import textwrap diff --git a/test/registered/unit/utils/test_weight_versions.py b/test/registered/unit/utils/test_weight_versions.py index 2e2b30b11..7c2508b4b 100644 --- a/test/registered/unit/utils/test_weight_versions.py +++ b/test/registered/unit/utils/test_weight_versions.py @@ -316,7 +316,6 @@ class _SchedulerStub: class TestSchedulerRecordWeightVersionChange(CustomTestCase): def _scheduler(self, *args, pp_size=1, **kwargs): - # The recording path asks the context for the pipeline width. enter_scope(self, published_topology(pp_size=pp_size)) scheduler = _SchedulerStub(*args, **kwargs) for name, value in (