diff --git a/.claude/skills/sglang-runtime-context/SKILL.md b/.claude/skills/sglang-runtime-context/SKILL.md index 95f699a6af10..a5f68174aa29 100644 --- a/.claude/skills/sglang-runtime-context/SKILL.md +++ b/.claude/skills/sglang-runtime-context/SKILL.md @@ -69,10 +69,10 @@ resolved configuration lives in the namespace bags.** - **Per-instance boundaries** — the tokenizer-manager family, everything under `entrypoints/`, and the tokenizer-process multimodal processors read `self.server_args`: several `Engine`s can share one process, and the process-global - bags are last-publish-wins across engines. (The mm-processor boundary is not yet - airtight: `BaseMultimodalProcessor.process_mm_data` still reads `base_gpu_id` / - `rl_on_policy_target` through `get_server_args()` — a known last-publish-wins gap, - not a pattern to copy.) + bags are last-publish-wins across engines. `base_gpu_id` also differs per worker + (the encode-server DP workers each specialize their own copy), so no process-global + value can stand in for it — `BaseMultimodalProcessor._fast_image_processor_device` + is the shape to copy. - **Whole-object passes** (`f(server_args)` handing the instance along) keep the supplied-instance contract; don't rewrite the parameter reads to bag reads unless the field is runtime-mutated (see the elastic-EP `ep_size` case in diff --git a/python/sglang/kernels/ops/layernorm/mhc.py b/python/sglang/kernels/ops/layernorm/mhc.py index 114afac1aa70..1f310d82a58e 100644 --- a/python/sglang/kernels/ops/layernorm/mhc.py +++ b/python/sglang/kernels/ops/layernorm/mhc.py @@ -579,10 +579,10 @@ def prewarm_mhc_pre( the TileLang/DeepGEMM on-disk JIT cache, so this cost is paid only on a cold cache; later server runs hit the cache. Driven once per process from load_weights. """ - from sglang.srt.runtime_context import get_server_args + from sglang.srt.runtime_context import get_schedule hc_mult, hidden_size = residual.shape[-2], residual.shape[-1] - max_num_tokens = get_server_args().chunked_prefill_size + max_num_tokens = get_schedule().chunked_prefill_size buckets = get_mhc_pre_token_count_representatives( max_num_tokens, hc_mult * hidden_size ) diff --git a/python/sglang/srt/batch_overlap/two_batch_overlap.py b/python/sglang/srt/batch_overlap/two_batch_overlap.py index 08a05b3ae5fa..2b52ab8ca6ab 100644 --- a/python/sglang/srt/batch_overlap/two_batch_overlap.py +++ b/python/sglang/srt/batch_overlap/two_batch_overlap.py @@ -40,7 +40,7 @@ compute_position, ) from sglang.srt.model_executor.forward_context import get_attn_backend -from sglang.srt.runtime_context import get_device, get_parallel, get_server_args +from sglang.srt.runtime_context import get_device, get_exec, get_parallel from sglang.srt.speculative.spec_info import SpecInput from sglang.srt.utils import BumpAllocator, empty_context, get_bool_env_var, is_hip @@ -634,7 +634,7 @@ def derive_fields_related_to_seq_len_for_two_chunk( sum_field=None, ) _, child_b.extend_start_loc = compute_position( - get_server_args().attention_backend, + get_exec().kernel.attention_backend, child_b.extend_prefix_lens, child_b.extend_seq_lens, child_b.extend_num_tokens, diff --git a/python/sglang/srt/layers/attention/attention_registry.py b/python/sglang/srt/layers/attention/attention_registry.py index 6c5400dc230c..39e9aa51d3e6 100644 --- a/python/sglang/srt/layers/attention/attention_registry.py +++ b/python/sglang/srt/layers/attention/attention_registry.py @@ -376,7 +376,9 @@ def attn_backend_wrapper(runner: "ModelRunner", full_attn_backend: "AttentionBac "gdn_backend.sm100_flashinfer_default", linear_attn_prefill_backend=prefill_default, ) - initialize_linear_attn_config(runner.server_args, prefill_default) + initialize_linear_attn_config( + runner.server_args, prefill_default=prefill_default + ) hybrid_backend_cls = HybridLinearAttnBackend if hybrid_gdn_config(runner.model_config) is not None: if is_blackwell(): diff --git a/python/sglang/srt/layers/rotary_embedding/mrope.py b/python/sglang/srt/layers/rotary_embedding/mrope.py index de0de81a3365..a49246802566 100644 --- a/python/sglang/srt/layers/rotary_embedding/mrope.py +++ b/python/sglang/srt/layers/rotary_embedding/mrope.py @@ -18,7 +18,7 @@ yarn_get_mscale_simple, yarn_linear_ramp_mask, ) -from sglang.srt.runtime_context import get_exec, get_server_args +from sglang.srt.runtime_context import get_exec from sglang.srt.utils import ( cpu_has_amx_support, is_cuda, @@ -42,7 +42,6 @@ from sgl_kernel import multimodal_rotary_embedding from sglang.kernels.ops.attention.mrope import apply_interleaved_rope_triton -from sglang.srt.runtime_context import get_server_args def apply_interleaved_rope(x: torch.Tensor, mrope_section: list) -> torch.Tensor: @@ -144,7 +143,7 @@ def get_cos_sin_with_position(self, positions): last_dim = cos_sin.size()[-1] cos, sin = cos_sin.chunk(2, dim=-1) if self.mrope_interleaved: - if support_triton(get_server_args().attention_backend): + if support_triton(get_exec().kernel.attention_backend): cos = apply_interleaved_rope_triton(cos, self.mrope_section) sin = apply_interleaved_rope_triton(sin, self.mrope_section) else: diff --git a/python/sglang/srt/managers/mm_utils.py b/python/sglang/srt/managers/mm_utils.py index 3462154a3b33..97e96c06d693 100644 --- a/python/sglang/srt/managers/mm_utils.py +++ b/python/sglang/srt/managers/mm_utils.py @@ -38,6 +38,7 @@ get_parallel, get_schedule, get_server_args, + get_serving, ) from sglang.srt.utils import flatten_nested_list, is_hip, is_npu, print_warning_once from sglang.srt.utils.stale_shm_cleanup import make_shm_name @@ -554,7 +555,7 @@ def _acknowledge_deferred_cuda_ipc_cache_hits( return # The pool's recycler counts the whole TP group, so the acknowledgement must # match that count even when an attention subgroup is smaller. - consumer_count = max(get_server_args().tp_size, 1) + consumer_count = max(parallel.tp_size, 1) for item in items: item.acknowledge_deferred_cuda_ipc_feature(consumer_count) @@ -2024,7 +2025,7 @@ def wrap_shm_features(obj): """ Scan the object for multimodal tensors and wrap them in SHM pointers. """ - if _get_is_default_transport() or get_server_args().skip_tokenizer_init: + if _get_is_default_transport() or get_serving().skip_tokenizer_init: return obj if obj.mm_inputs: @@ -2085,7 +2086,7 @@ def unwrap_shm_features(obj): Restore ShmPointerMMData wrappers back into standard torch.Tensors. Handles both single requests and batch requests. """ - if _get_is_default_transport() or get_server_args().skip_tokenizer_init: + if _get_is_default_transport() or get_serving().skip_tokenizer_init: return obj # Handle batch requests if isinstance(obj, BaseBatchReq): diff --git a/python/sglang/srt/mem_cache/allocation.py b/python/sglang/srt/mem_cache/allocation.py index 45467f58f3e3..d87482278622 100644 --- a/python/sglang/srt/mem_cache/allocation.py +++ b/python/sglang/srt/mem_cache/allocation.py @@ -26,7 +26,7 @@ evict_from_tree_cache, ) from sglang.srt.mem_cache.memory_pool import HybridReqToTokenPool, ReqToTokenPool -from sglang.srt.runtime_context import get_server_args +from sglang.srt.runtime_context import get_exec, get_server_args from sglang.srt.utils import ( is_cpu, is_cuda, @@ -65,7 +65,7 @@ def write_cache_indices( prefix_tensors: list[torch.Tensor], req_to_token_pool: ReqToTokenPool, ): - if support_triton(get_server_args().attention_backend): + if support_triton(get_exec().kernel.attention_backend): prefix_pointers = torch.tensor( [t.data_ptr() for t in prefix_tensors], dtype=torch.uint64, @@ -106,7 +106,7 @@ def get_last_loc( req_pool_indices_tensor: torch.Tensor, prefix_lens_tensor: torch.Tensor, ) -> torch.Tensor: - attn_backend = get_server_args().attention_backend + attn_backend = get_exec().kernel.attention_backend uses_triton_dispatch = attn_backend not in ("ascend", "torch_native") if _is_hip and uses_triton_dispatch: diff --git a/python/sglang/srt/models/gpt_oss.py b/python/sglang/srt/models/gpt_oss.py index 4fa6c351e01a..1ec1e196252c 100644 --- a/python/sglang/srt/models/gpt_oss.py +++ b/python/sglang/srt/models/gpt_oss.py @@ -72,7 +72,6 @@ get_exec, get_forward, get_parallel, - get_server_args, ) from sglang.srt.utils import ( LazyValue, @@ -421,7 +420,7 @@ def __init__( # Choose dtype of sinks based on attention backend: trtllm_mha requires float32, # others can use bfloat16 - attn_backend = get_server_args().attention_backend + attn_backend = get_exec().kernel.attention_backend sinks_dtype = torch.float32 if attn_backend == "trtllm_mha" else torch.bfloat16 self.sinks = nn.Parameter( torch.empty(self.num_heads, dtype=sinks_dtype), requires_grad=False diff --git a/python/sglang/srt/models/inkling_common/dense_mlp.py b/python/sglang/srt/models/inkling_common/dense_mlp.py index 82954319dfc0..cafaafc60a59 100644 --- a/python/sglang/srt/models/inkling_common/dense_mlp.py +++ b/python/sglang/srt/models/inkling_common/dense_mlp.py @@ -18,7 +18,7 @@ lora_compatible_layout_enabled, ) from sglang.srt.models.llama import LlamaMLP -from sglang.srt.runtime_context import get_exec, get_server_args +from sglang.srt.runtime_context import get_exec, get_model logger = logging.getLogger(__name__) @@ -484,7 +484,7 @@ def _shared_scales( # All shared experts must share one global weight scale (reshard with # single_global_scale=True). ModelOpt's input_scale = amax / (6 * 448). flat2 = scale2.reshape(-1).float() - if get_server_args().load_format == "dummy" and not bool( + if get_model().load_format == "dummy" and not bool( torch.all(flat2 == flat2[0]) ): # Dummy loading uses per-element noise; replace it with a valid scale. diff --git a/python/sglang/srt/multimodal/processors/base_processor.py b/python/sglang/srt/multimodal/processors/base_processor.py index 91af949e25e2..2f91b40ddd7e 100644 --- a/python/sglang/srt/multimodal/processors/base_processor.py +++ b/python/sglang/srt/multimodal/processors/base_processor.py @@ -20,7 +20,6 @@ MultimodalProcessorOutput, ) from sglang.srt.multimodal.processors.executor import MultimodalProcessorExecutor -from sglang.srt.runtime_context import get_server_args from sglang.srt.utils import ( CLIENT_MEDIA_EXCEPTIONS, envs, @@ -485,6 +484,37 @@ def _resolve_processor(self, processor=None): return self._processor, self._tokenizer return processor, processor.tokenizer + def _fast_image_processor_device(self, processor) -> Optional[str]: + """The device for the fast image processor, or None to leave it unset. + + Resolved from this processor's own ``server_args``: engines sharing a + tokenizer process each carry their own ``base_gpu_id``. + """ + server_args = self.server_args + if _is_cpu or server_args.rl_on_policy_target is not None: + return "cpu" + if _is_xpu: + return "xpu" + if not _is_npu: + return f"cuda:{server_args.base_gpu_id}" + if processor.__class__.__name__ not in {"Glm4vProcessor", "Glm46VProcessor"}: + # For qwen-vl, the processor hits a reshape issue from the Ascend + # dims restriction. + from sglang.srt.hardware_backend.npu.modules.qwen_vl_processor import ( + npu_apply_qwen_image_preprocess_patch, + ) + + npu_apply_qwen_image_preprocess_patch() + return "npu" + if processor.__class__.__name__ == "Glm46VProcessor": + from sglang.srt.hardware_backend.npu.modules.glm46v_processor import ( + npu_apply_glm46v_image_preprocess_patch, + ) + + npu_apply_glm46v_image_preprocess_patch() + return "npu" + return None + def process_mm_data( self, input_text, @@ -537,31 +567,9 @@ def process_mm_data( and isinstance(processor.image_processor, BaseImageProcessor) and not self.disable_fast_image_processor ): - if _is_cpu or get_server_args().rl_on_policy_target is not None: - kwargs["device"] = "cpu" - elif _is_xpu: - kwargs["device"] = "xpu" - elif not _is_npu: - base_gpu_id = get_server_args().base_gpu_id - kwargs["device"] = f"cuda:{base_gpu_id}" - elif processor.__class__.__name__ not in { - "Glm4vProcessor", - "Glm46VProcessor", - }: - # Note: for qwen-vl, processor has some reshape issue because of dims restriction on Ascend. - from sglang.srt.hardware_backend.npu.modules.qwen_vl_processor import ( - npu_apply_qwen_image_preprocess_patch, - ) - - npu_apply_qwen_image_preprocess_patch() - kwargs["device"] = "npu" - elif processor.__class__.__name__ == "Glm46VProcessor": - from sglang.srt.hardware_backend.npu.modules.glm46v_processor import ( - npu_apply_glm46v_image_preprocess_patch, - ) - - npu_apply_glm46v_image_preprocess_patch() - kwargs["device"] = "npu" + device = self._fast_image_processor_device(processor) + if device is not None: + kwargs["device"] = device # Avoid double BOS when the chat template already wrote one. if self._tokenizer_auto_adds_specials and isinstance(input_text, str): diff --git a/test/registered/unit/mem_cache/test_dllm_fdfo_kv_reuse.py b/test/registered/unit/mem_cache/test_dllm_fdfo_kv_reuse.py index c48fcc0e9428..787ed1827f2f 100644 --- a/test/registered/unit/mem_cache/test_dllm_fdfo_kv_reuse.py +++ b/test/registered/unit/mem_cache/test_dllm_fdfo_kv_reuse.py @@ -7,9 +7,9 @@ import torch from sglang.srt.dllm.mixin.scheduler import DllmManager -from sglang.srt.mem_cache import allocation from sglang.srt.mem_cache.allocation import alloc_for_extend from sglang.srt.mem_cache.memory_pool import ReqToTokenPool +from sglang.srt.runtime_context import get_context from sglang.test.ci.ci_register import register_cpu_ci register_cpu_ci(est_time=5, suite="base-a-test-cpu") @@ -117,16 +117,11 @@ def setUp(self): self.pool = ReqToTokenPool( size=8, max_context_len=64, device="cpu", enable_memory_saver=False ) - self._old_support_triton = allocation.support_triton - self._old_get_server_args = allocation.get_server_args - allocation.support_triton = lambda _: False - allocation.get_server_args = lambda: SimpleNamespace( + override = get_context().override_server_args( attention_backend="torch_native", dcp_size=1 ) - - def tearDown(self): - allocation.support_triton = self._old_support_triton - allocation.get_server_args = self._old_get_server_args + override.install() + self.addCleanup(override.restore) def test_alloc_for_extend_mixed_reuse_allocates_only_fresh_and_writes_rows(self): allocator = _FakeAllocator(base=200) diff --git a/test/registered/unit/multimodal/test_processor_device_selection.py b/test/registered/unit/multimodal/test_processor_device_selection.py new file mode 100644 index 000000000000..57905f1dd876 --- /dev/null +++ b/test/registered/unit/multimodal/test_processor_device_selection.py @@ -0,0 +1,81 @@ +"""The fast-image-processor device comes from the processor's own ServerArgs. + +Regression: the device decision read the published global ServerArgs, which is +last-publish-wins. Two engines in one tokenizer process then shared whichever +config published last, so one engine's images were preprocessed on the other +engine's GPU. +""" + +import unittest +from unittest.mock import patch + +from sglang.srt.multimodal.processors.base_processor import BaseMultimodalProcessor +from sglang.srt.server_args import ServerArgs +from sglang.test.ci.ci_register import register_cpu_ci +from sglang.test.test_utils import CustomTestCase + +register_cpu_ci(est_time=5, suite="base-a-test-cpu") + +BASE = "sglang.srt.multimodal.processors.base_processor" + + +class _Processor: + pass + + +class _StubProcessor(BaseMultimodalProcessor): + async def process_mm_data_async(self, *args, **kwargs): + raise NotImplementedError + + +def _make(**fields): + processor = _StubProcessor.__new__(_StubProcessor) + processor.server_args = ServerArgs(model_path="dummy", **fields) + return processor + + +class TestFastImageProcessorDevice(CustomTestCase): + def _device(self, processor, **platform): + flags = {"_is_cpu": False, "_is_xpu": False, "_is_npu": False} + flags.update(platform) + with patch.multiple(BASE, **flags): + return processor._fast_image_processor_device(_Processor()) + + def test_device_follows_the_instance_base_gpu_id(self): + self.assertEqual(self._device(_make(base_gpu_id=3)), "cuda:3") + + def test_engines_in_one_process_keep_their_own_device(self): + first, second = _make(base_gpu_id=0), _make(base_gpu_id=5) + self.assertEqual(self._device(first), "cuda:0") + self.assertEqual(self._device(second), "cuda:5") + + def test_publishing_another_config_does_not_move_the_device(self): + from sglang.srt.runtime_context import get_context + + processor = _make(base_gpu_id=2) + override = get_context().override_server_args(base_gpu_id=7) + override.install() + self.addCleanup(override.restore) + self.assertEqual(self._device(processor), "cuda:2") + + def test_rl_on_policy_target_forces_cpu(self): + processor = _make(base_gpu_id=3, rl_on_policy_target="fsdp") + self.assertEqual(self._device(processor), "cpu") + + def test_cpu_and_xpu_platforms_win_over_base_gpu_id(self): + processor = _make(base_gpu_id=3) + self.assertEqual(self._device(processor, _is_cpu=True), "cpu") + self.assertEqual(self._device(processor, _is_xpu=True), "xpu") + + def test_npu_glm4v_leaves_the_device_unset(self): + class Glm4vProcessor: + pass + + processor = _make(base_gpu_id=3) + with patch.multiple(BASE, _is_cpu=False, _is_xpu=False, _is_npu=True): + device = processor._fast_image_processor_device(Glm4vProcessor()) + self.assertIsNone(device) + + +if __name__ == "__main__": + unittest.main() diff --git a/test/registered/unit/test_global_config_read_ratchet.py b/test/registered/unit/test_global_config_read_ratchet.py new file mode 100644 index 000000000000..b18579166692 --- /dev/null +++ b/test/registered/unit/test_global_config_read_ratchet.py @@ -0,0 +1,165 @@ +"""Ratchet guard: process-global config reads may only decrease. + +``get_server_args()`` returns the published ``ServerArgs`` — one process's +startup record. Config decisions read the namespace accessors instead +(``get_exec()`` / ``get_memory()`` / …), which carry the resolved value +including post-publish overrides, and per-runner values come from the runner +that owns them. + +Two shapes count as a read: the direct ``get_server_args().field``, and the +alias ``sa = get_server_args()`` followed by ``sa.field`` in the same function. +A whole-object pass (``def f(server_args)``) is not a global read and is not +counted — there the caller decided which instance to hand over. + +What legitimately remains: + +- **Derived APIs.** ``@property`` and method members of ``ServerArgs`` + (``mamba_cache_chunk_size``, ``get_model_config()``, + ``enable_mamba_extra_buffer()``, …) are computed from several fields plus the + HF config, so they are not namespace leaves and ``ServerArgs`` is their only + home. Exempt by name below. +- **Config-intent reads of live-shadowed sizes.** ``get_parallel()`` shadows + ``tp/pp/dcp/attn_cp/moe_dp_size`` with the live topology, so a config-intent + read of one has nowhere else to go. Each exempt site needs an answer the live + property cannot give: + + - ``dsa_indexer.pp_size`` gates ``pp_size > 1 and not get_pp_group()...``, and + the short circuit is the point: with PP off the group is never touched, which + is what lets the ``Indexer`` be constructed before distributed init. The live + property would demand the group either way. + - ``allocation.dcp_size`` asks whether DCP was *configured*; the live property + reads ``get_dcp_group()``, and that group is only installed when DCP is on. + - ``cuda_ipc_transport_utils.tp_size`` runs in the tokenizer process, which has + no groups at all (the call site already guards for "not published yet"). +- The alias-form baseline is not zero yet. Lowering it is the next slice; the + failure message lists the sites whenever the count moves. +""" + +from sglang.test.ci.ci_register import register_cpu_ci + +register_cpu_ci(est_time=5, suite="base-a-test-cpu") + +import ast +import unittest +from pathlib import Path + +import sglang +from sglang.test.test_utils import CustomTestCase + +# srt is the migrated surface; the rest of the package has no reads today and is +# scanned so a new one cannot appear there unnoticed. +_PACKAGE_ROOT = Path(next(iter(sglang.__path__))) + +_DERIVED_MEMBERS = frozenset( + { + "cutedsl_moe_max_num_tokens", + "enable_mamba_extra_buffer", + "enable_mamba_extra_buffer_lazy", + "get_attention_backends", + "get_model_config", + "mamba_cache_chunk_size", + "max_speculative_num_draft_tokens", + "model_config", + "use_mla_backend", + } +) + +_CONFIG_INTENT_SIZES = frozenset( + { + ("srt/layers/attention/dsa/dsa_indexer.py", "pp_size"), + ("srt/mem_cache/allocation.py", "dcp_size"), + ("srt/utils/cuda_ipc_transport_utils.py", "tp_size"), + } +) + +_DIRECT_BASELINE = 0 +_ALIAS_BASELINE = 12 + + +def _is_global_call(node) -> bool: + return ( + isinstance(node, ast.Call) + and isinstance(node.func, ast.Name) + and node.func.id == "get_server_args" + ) + + +def _collect(rel: str, tree: ast.AST): + """The (direct, alias) field reads in one module.""" + direct, alias = [], [] + + def counted(attr: str) -> bool: + return attr not in _DERIVED_MEMBERS and (rel, attr) not in _CONFIG_INTENT_SIZES + + for node in ast.walk(tree): + if ( + isinstance(node, ast.Attribute) + and _is_global_call(node.value) + and counted(node.attr) + ): + direct.append(f"{rel}:{node.lineno}: get_server_args().{node.attr}") + + if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): + continue + params = {a.arg for a in list(node.args.args) + list(node.args.kwonlyargs)} + bound = {} + for inner in ast.walk(node): + if isinstance(inner, ast.Assign) and _is_global_call(inner.value): + for target in inner.targets: + if isinstance(target, ast.Name) and target.id not in params: + bound.setdefault(target.id, inner.lineno) + if not bound: + continue + for inner in ast.walk(node): + if ( + isinstance(inner, ast.Attribute) + and isinstance(inner.value, ast.Name) + and inner.value.id in bound + and inner.lineno >= bound[inner.value.id] + and counted(inner.attr) + ): + alias.append( + f"{rel}:{inner.lineno}: {inner.value.id}.{inner.attr} " + f"(bound from get_server_args() at line {bound[inner.value.id]})" + ) + return direct, alias + + +def _field_reads(): + direct, alias = [], [] + for path in sorted(_PACKAGE_ROOT.rglob("*.py")): + rel = path.relative_to(_PACKAGE_ROOT).as_posix() + try: + tree = ast.parse(path.read_text()) + except SyntaxError: + continue + module_direct, module_alias = _collect(rel, tree) + direct += module_direct + alias += module_alias + return direct, alias + + +class TestGlobalConfigReadRatchet(CustomTestCase): + def _check(self, kind, reads, baseline): + if len(reads) > baseline: + self.fail( + f"{kind} process-global config field reads grew: {len(reads)} > " + f"baseline {baseline}. Read the namespace accessor for the " + "field's namespace, or the owning runner for a per-runner " + "field:\n" + "\n".join(reads) + ) + if len(reads) < baseline: + self.fail( + f"{kind} process-global config field reads shrank: {len(reads)} < " + f"baseline {baseline}. Lower the baseline in this file to lock " + "in the progress." + ) + + def test_global_field_reads_match_the_baseline(self): + direct, alias = _field_reads() + self._check("direct", direct, _DIRECT_BASELINE) + self._check("alias-form", alias, _ALIAS_BASELINE) + + +if __name__ == "__main__": + unittest.main()