From 340391a297ac516bd48dda5dedb697cf6b32ec33 Mon Sep 17 00:00:00 2001 From: Cheng Wan <54331508+ch-wan@users.noreply.github.com> Date: Sun, 23 Aug 2026 01:20:20 -0700 Subject: [PATCH] config: publish before the launcher reads effective configuration (#35910) Co-authored-by: Claude Opus 5 --- python/sglang/srt/entrypoints/engine.py | 97 ++++-- .../srt/managers/data_parallel_controller.py | 37 ++- python/sglang/srt/runtime_context.py | 82 ++++- .../multimodal/test_gpu_feature_transport.py | 22 +- .../test_resolution_declarations.py | 107 +++++++ .../test_resolution_reads_no_bag.py | 291 ++++++++++++++++++ .../unit/test_global_config_read_ratchet.py | 17 + ...test_supplied_instance_exposure_ratchet.py | 12 - 8 files changed, 591 insertions(+), 74 deletions(-) create mode 100644 test/registered/unit/server_args/test_resolution_reads_no_bag.py diff --git a/python/sglang/srt/entrypoints/engine.py b/python/sglang/srt/entrypoints/engine.py index 4500bd3ca..5c7279b80 100644 --- a/python/sglang/srt/entrypoints/engine.py +++ b/python/sglang/srt/entrypoints/engine.py @@ -100,12 +100,16 @@ from sglang.srt.parser.template_detection import resolve_auto_parsers from sglang.srt.parser.template_manager import TemplateManager from sglang.srt.plugins import load_plugins from sglang.srt.runtime_context import ( + configured_attn_cp_size, + configured_moe_dp_size, configured_pp_size, get_exec, get_model, get_parallel, get_serving, publish, + restore_context, + snapshot_context, ) from sglang.srt.server_args import PortArgs, ServerArgs from sglang.srt.utils import ( @@ -1089,43 +1093,63 @@ class Engine(EngineScoreMixin, EngineBase): # Engine.__init__ or CLI entry). load_plugins() + # Not read-only: the LoRA checks normalize adapter paths through late + # resolution, which a published config refuses. Hence before publish -- + # and before the parser detection below, which consumes the "auto" + # sentinel: a record rejected here has to stay retryable. server_args.check_server_args() - # Allocate ports for inter-process communications - if port_args is None: - port_args = PortArgs.init_new(server_args) - logger.info(f"{server_args=}") - - # Start the engine info bootstrap server if per-rank info is needed. - engine_info_bootstrap_server = None - if ( - server_args.remote_instance_weight_loader_start_seed_via_transfer_engine - and server_args.node_rank == 0 - ): - bootstrap_port = server_args.engine_info_bootstrap_port - if not is_port_available(bootstrap_port): - raise RuntimeError( - f"engine_info_bootstrap_port {bootstrap_port} is already in use. " - f"When running multiple instances on the same node, each instance must use a " - f"different --engine-info-bootstrap-port." - ) - engine_info_bootstrap_server = EngineInfoBootstrapServer( - host=server_args.host, port=bootstrap_port - ) - + # Needs a tokenizer and a chat template, so it cannot live in the + # pipeline; after the plugins, which may register the parser detected. if ( server_args.reasoning_parser == "auto" or server_args.tool_call_parser == "auto" ): resolve_auto_parsers(server_args) + # This publish replaces whatever was published before it, so the + # rollback below restores that rather than clearing the process: a + # caller that catches the launch error still has the context it had. + context_before_publish = snapshot_context() publish(server_args, role="tokenizer") - # Launch daemons (daemon mode only). The handles travel back to the - # Engine that spawned them; shutdown() reaps from there. - weight_cache_daemon_procs: List = [] - if server_args.weight_cache_mode == "daemon": - weight_cache_daemon_procs = cls._launch_weight_cache_daemons(server_args) + # Nothing below has spawned yet, so a failure here leaves a record the + # caller can hand back -- but only once the publication goes with it: + # the validation stage writes through late resolution, which refuses a + # record that is already published. + try: + # Allocate ports for inter-process communications + if port_args is None: + port_args = PortArgs.init_new(server_args) + logger.info(f"{server_args=}") + + # Start the engine info bootstrap server if per-rank info is needed. + engine_info_bootstrap_server = None + if ( + get_model().remote_instance_weight_loader_start_seed_via_transfer_engine + and server_args.node_rank == 0 + ): + bootstrap_port = server_args.engine_info_bootstrap_port + if not is_port_available(bootstrap_port): + raise RuntimeError( + f"engine_info_bootstrap_port {bootstrap_port} is already in use. " + f"When running multiple instances on the same node, each instance must use a " + f"different --engine-info-bootstrap-port." + ) + engine_info_bootstrap_server = EngineInfoBootstrapServer( + host=server_args.host, port=bootstrap_port + ) + + # Launch daemons (daemon mode only). The handles travel back to the + # Engine that spawned them; shutdown() reaps from there. + weight_cache_daemon_procs: List = [] + if server_args.weight_cache_mode == "daemon": + weight_cache_daemon_procs = cls._launch_weight_cache_daemons( + server_args + ) + except BaseException: + restore_context(context_before_publish) + raise # Launch scheduler processes # Passed only when there is one: this hook is an override point, and a @@ -1858,18 +1882,25 @@ def _calculate_rank_ranges( def _compute_parallelism_ranks( server_args: ServerArgs, tp_rank: int ) -> Tuple[int, int, int]: - """Compute attention-CP, MoE-DP, and MoE-EP ranks for a TP rank.""" + """Compute attention-CP, MoE-DP, and MoE-EP ranks for a TP rank. + + Called while the launcher is deciding what to spawn, so the sizes are the + configured ones -- the groups this is laying out do not exist yet. + """ attn_dp_size = get_parallel().dp_size if get_parallel().enable_dp_attention else 1 + tp_size = server_args.tp_size + attn_cp_size = configured_attn_cp_size() + moe_dp_size = configured_moe_dp_size() # Parallelism hierarchy (outermost to innermost): # - Attention: Global(TP) -> DP -> ATTN_CP -> ATTN_TP (innermost) # - MoE: Global(TP) -> MOE_DP -> EP -> MOE_TP (innermost) - attn_tp_size = server_args.tp_size // attn_dp_size // server_args.attn_cp_size - attn_cp_rank = (tp_rank // attn_tp_size) % server_args.attn_cp_size - moe_dp_rank = tp_rank // (server_args.tp_size // server_args.moe_dp_size) + attn_tp_size = tp_size // attn_dp_size // attn_cp_size + attn_cp_rank = (tp_rank // attn_tp_size) % attn_cp_size + moe_dp_rank = tp_rank // (tp_size // moe_dp_size) moe_ep_rank = ( tp_rank - % (server_args.tp_size // server_args.moe_dp_size) - // (server_args.tp_size // server_args.moe_dp_size // get_parallel().ep_size) + % (tp_size // moe_dp_size) + // (tp_size // moe_dp_size // get_parallel().ep_size) ) return attn_cp_rank, moe_dp_rank, moe_ep_rank diff --git a/python/sglang/srt/managers/data_parallel_controller.py b/python/sglang/srt/managers/data_parallel_controller.py index 6a4766fa3..0b3c4c904 100644 --- a/python/sglang/srt/managers/data_parallel_controller.py +++ b/python/sglang/srt/managers/data_parallel_controller.py @@ -49,7 +49,16 @@ from sglang.srt.observability.cpu_monitor import start_cpu_monitor_thread from sglang.srt.observability.req_time_stats import DPControllerReqTimeStats from sglang.srt.observability.startup_time import aggregate_scheduler_startup_times from sglang.srt.observability.trace import process_tracing_init, trace_set_thread_info -from sglang.srt.runtime_context import get_exec, get_parallel, publish +from sglang.srt.runtime_context import ( + configured_attn_cp_size, + configured_moe_dp_size, + configured_pp_size, + get_device, + get_disagg, + get_exec, + get_parallel, + publish, +) from sglang.srt.server_args import ( DP_ATTENTION_HANDSHAKE_PORT_DELTA, PortArgs, @@ -142,7 +151,7 @@ class DataParallelController: self.server_args = server_args self.port_args = port_args self.load_balance_method = LoadBalanceMethod.from_str( - server_args.load_balance_method + get_parallel().load_balance_method ) self.run_scheduler_process_func = run_scheduler_process_func @@ -212,7 +221,7 @@ class DataParallelController: self.soft_watchdog = Watchdog.create( debug_name="DataParallelController", - watchdog_timeout=server_args.soft_watchdog_timeout, + watchdog_timeout=get_device().soft_watchdog_timeout, soft=True, test_stuck_time=envs.SGLANG_TEST_STUCK_DP_CONTROLLER.get(), ) @@ -386,7 +395,7 @@ class DataParallelController: ) threads.append(thread) base_gpu_id += ( - server_args.tp_size * server_args.pp_size * server_args.gpu_id_step + server_args.tp_size * configured_pp_size() * server_args.gpu_id_step ) if server_args.node_rank == 0: @@ -607,8 +616,8 @@ class DataParallelController: scheduler_pipe_readers = [] - pp_size_per_node = max(server_args.pp_size // server_args.nnodes, 1) - nnodes_per_pp_rank = max(server_args.nnodes // server_args.pp_size, 1) + pp_size_per_node = max(configured_pp_size() // server_args.nnodes, 1) + nnodes_per_pp_rank = max(server_args.nnodes // configured_pp_size(), 1) pp_rank_range = range( pp_size_per_node * (server_args.node_rank // nnodes_per_pp_rank), pp_size_per_node * (server_args.node_rank // nnodes_per_pp_rank + 1), @@ -639,7 +648,7 @@ class DataParallelController: tp_rank, server_args.tp_size, get_parallel().dp_size, - server_args.attn_cp_size, + configured_attn_cp_size(), ) # compute zmq ports for this dp rank rank_port_args = PortArgs.init_new( @@ -675,18 +684,18 @@ class DataParallelController: # - Attention: Global(TP) -> DP -> ATTN_CP -> ATTN_TP (innermost) # - MoE: Global(TP) -> MOE_DP -> EP -> MOE_TP (innermost) attn_tp_size = ( - server_args.tp_size // attn_dp_size // server_args.attn_cp_size + server_args.tp_size // attn_dp_size // configured_attn_cp_size() ) - attn_cp_rank = (tp_rank // attn_tp_size) % server_args.attn_cp_size + attn_cp_rank = (tp_rank // attn_tp_size) % configured_attn_cp_size() moe_dp_rank = tp_rank // ( - server_args.tp_size // server_args.moe_dp_size + server_args.tp_size // configured_moe_dp_size() ) moe_ep_rank = ( tp_rank - % (server_args.tp_size // server_args.moe_dp_size) + % (server_args.tp_size // configured_moe_dp_size()) // ( server_args.tp_size - // server_args.moe_dp_size + // configured_moe_dp_size() // get_parallel().ep_size ) ) @@ -829,9 +838,9 @@ def run_data_parallel_controller_process( trace_modules=server_args.trace_modules, ) thread_label = "DP Controller" - if server_args.disaggregation_mode == "prefill": + if get_disagg().disaggregation_mode == "prefill": thread_label = "Prefill DP Controller" - elif server_args.disaggregation_mode == "decode": + elif get_disagg().disaggregation_mode == "decode": thread_label = "Decode DP Controller" trace_set_thread_info(thread_label) diff --git a/python/sglang/srt/runtime_context.py b/python/sglang/srt/runtime_context.py index 93be0ef3c..64167bd71 100644 --- a/python/sglang/srt/runtime_context.py +++ b/python/sglang/srt/runtime_context.py @@ -1187,9 +1187,10 @@ ROLE_NAMESPACE_SETS: dict[str, frozenset[str] | None] = { # Reads (almost) everything by design — the model-executing process. "scheduler": None, "test": None, - # Audited (record-mode smokes, plain + DP-attention): the DP controller - # reads only the elastic-EP gate; its module's static read set agrees. - "dp_controller": frozenset({"exec"}), + # The DP controller's static read set, checked against the module: the + # elastic-EP gate, the load-balance method, the watchdog timeout, and the + # disaggregation mode. + "dp_controller": frozenset({"exec", "parallel", "device", "disagg"}), # Record-mode audit (2026-08-06, text model, /generate + /get_server_info + # /v1/models): reads exactly {"serving"} — the per-instance managers read # self.server_args by design. Still declared full, because that run did not @@ -1407,6 +1408,81 @@ def set_global_dwdp_manager(manager: Any) -> None: _GLOBAL_DWDP_MANAGER = manager +def _group_leaves(group: _FlagGroupBase) -> dict[str, Any]: + """The leaf values of a flag group, recursively.""" + leaves: dict[str, Any] = {} + for name in type(group).__dataclass_fields__: + value = getattr(group, name) + if isinstance(value, _FlagGroupBase): + leaves[name] = _group_leaves(value) + elif isinstance(value, (dict, list)): + leaves[name] = type(value)(value) + else: + leaves[name] = value + return leaves + + +def _restore_leaves(group: _FlagGroupBase, leaves: dict[str, Any]) -> None: + for name, value in leaves.items(): + current = getattr(group, name) + if isinstance(current, _FlagGroupBase): + _restore_leaves(current, value) + elif isinstance(current, dict): + current.clear() + current.update(value) + elif isinstance(current, list): + current[:] = value + else: + setattr(group, name, value) + + +def snapshot_context() -> dict[str, Any]: + """Everything a publish replaces, so a failed launch can put it back. + + Enumerated from ``__slots__`` rather than listed by hand: a hand-picked copy + of context state is one field behind the day a slot is added, and the copy + that silently drops one is worse than none. Flag groups are snapshotted by + leaf, not by reference: publish writes *into* the same ``Flags`` object + (``capture.enable_torch_compile``), so a reference held here would already + carry the failed launch's value by the time it is put back. + """ + state: dict[str, Any] = {} + for name in RuntimeContext.__slots__: + if name == "parallel": + continue + value = getattr(_CONTEXT, name) + if isinstance(value, _FlagGroupBase): + state[name] = (value, _group_leaves(value)) + elif isinstance(value, list): + state[name] = list(value) + else: + state[name] = value + state["__parallel__"] = { + name: getattr(_CONTEXT.parallel, name) + for name in type(_CONTEXT.parallel).__slots__ + } + state["__dwdp__"] = get_global_dwdp_manager() + return state + + +def restore_context(state: dict[str, Any]) -> None: + """Put back what ``snapshot_context`` captured.""" + for name in RuntimeContext.__slots__: + if name == "parallel": + continue + value = state[name] + if isinstance(value, tuple) and isinstance(value[0], _FlagGroupBase): + group, leaves = value + setattr(_CONTEXT, name, group) + _restore_leaves(group, leaves) + else: + setattr(_CONTEXT, name, value) + for name, value in state["__parallel__"].items(): + setattr(_CONTEXT.parallel, name, value) + _adaptive_draft_token_bound.cache_clear() + set_global_dwdp_manager(state["__dwdp__"]) + + def reset_context() -> None: """Clear the context-owned store (unit-test teardown): drop the published ``server_args`` and install fresh ``Flags`` and ``Resources``. diff --git a/test/registered/unit/multimodal/test_gpu_feature_transport.py b/test/registered/unit/multimodal/test_gpu_feature_transport.py index e31034138..14424291a 100644 --- a/test/registered/unit/multimodal/test_gpu_feature_transport.py +++ b/test/registered/unit/multimodal/test_gpu_feature_transport.py @@ -445,18 +445,16 @@ class TestCudaVmmFeatureTransport(unittest.TestCase): pool = MagicMock() transport.pool = pool manager.cuda_vmm_feature_transport = transport - server_args = SimpleNamespace( - remote_instance_weight_loader_start_seed_via_transfer_engine=False, - reasoning_parser=None, - tool_call_parser=None, - weight_cache_mode=None, - enable_elastic_expert_backup=False, - elastic_ep_backend=None, - node_rank=0, - tokenizer_worker_num=1, - check_server_args=MagicMock(), - resolve_once=MagicMock(), - ) + # A real record: the launcher publishes it partway through, and what it + # reads after that comes out of the bags, which only project from a + # dataclass. The validation is stubbed so the dummy path still launches. + from sglang.srt.server_args import ServerArgs + + server_args = ServerArgs(model_path="dummy", tokenizer_worker_num=1) + server_args.check_server_args = MagicMock() + from sglang.srt.runtime_context import reset_context + + self.addCleanup(reset_context) scheduler_init_result = SimpleNamespace( all_child_pids=[], scheduler_infos=[], diff --git a/test/registered/unit/server_args/test_resolution_declarations.py b/test/registered/unit/server_args/test_resolution_declarations.py index 4e8447dd2..6bc6eef12 100644 --- a/test/registered/unit/server_args/test_resolution_declarations.py +++ b/test/registered/unit/server_args/test_resolution_declarations.py @@ -125,6 +125,40 @@ _REACHED_BY_SHAPES = frozenset( ) +def _late_resolvers(): + """Callables that reach `declare_late_resolution`, derived per module.""" + found = set() + for relative in ("server_args.py", "parser/template_detection.py"): + tree = ast.parse((_SRT / relative).read_text(encoding="utf-8-sig")) + functions = { + node.name: node + for node in ast.walk(tree) + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) + } + + def reaches(name, seen=None): + seen = seen if seen is not None else set() + if name in seen or name not in functions: + return False + seen.add(name) + for node in ast.walk(functions[name]): + if not isinstance(node, ast.Call): + continue + called = ( + node.func.attr + if isinstance(node.func, ast.Attribute) + else getattr(node.func, "id", None) + ) + if called in ("declare_late_resolution", "_late_resolution"): + return True + if called and reaches(called, seen): + return True + return False + + found |= {name for name in functions if reaches(name)} + return found + + def _server_args_writers(tree, path): """Assignment targets that land on a ServerArgs instance. @@ -529,6 +563,79 @@ class TestResolutionDeclarations(CustomTestCase): self.assertEqual(get_serving().reasoning_parser, "qwen3") self.assertEqual(server_args.reasoning_parser, get_serving().reasoning_parser) + def test_validation_can_still_resolve_before_the_record_is_published(self): + """The LoRA checks normalize in place, so they must precede publish. + + `check_server_args` is not read-only: it infers `enable_lora`, parses + adapter paths and normalizes target modules through late resolution, + which a published record refuses. The launcher order is what keeps this + legal, and this is the assertion that notices if it moves. + """ + from sglang.srt.runtime_context import get_lora, publish, reset_context + + server_args = self._resolve( + { + "enable_lora": True, + "max_lora_rank": 16, + "lora_target_modules": ["q_proj"], + } + ) + self.addCleanup(reset_context) + server_args.check_server_args() + publish(server_args, role="tokenizer") + self.assertEqual(get_lora().enable_lora, server_args.enable_lora) + self.assertEqual( + get_lora().lora_target_modules, server_args.lora_target_modules + ) + + def test_the_launcher_finishes_resolving_before_it_publishes(self): + """Every late resolver runs above the publish, in the source. + + A published record refuses to be written, so a late resolver below the + publish raises at startup rather than at test time -- and only for the + configuration that reaches it, which is why the LoRA path can break + while every other launch stays green. Both sides are derived: which + callables reach `declare_late_resolution`, and where the launcher calls + them. + """ + launcher = _SRT / "entrypoints/engine.py" + late = {"check_server_args", "resolve_auto_parsers"} | _late_resolvers() + tree = ast.parse(launcher.read_text(encoding="utf-8-sig")) + function = next( + node + for node in ast.walk(tree) + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) + and node.name == "_launch_subprocesses" + ) + published_at = [ + node.lineno + for node in ast.walk(function) + if isinstance(node, ast.Call) + and isinstance(node.func, ast.Name) + and node.func.id == "publish" + ] + self.assertEqual(len(published_at), 1, "the launcher publishes once") + too_late = sorted( + f"{name}() at line {node.lineno}" + for node in ast.walk(function) + if isinstance(node, ast.Call) + for name in [ + ( + node.func.attr + if isinstance(node.func, ast.Attribute) + else getattr(node.func, "id", None) + ) + ] + if name in late and node.lineno > published_at[0] + ) + self.assertEqual( + too_late, + [], + f"these resolve after the launcher publishes at line " + f"{published_at[0]}, and a published record refuses to be " + f"written:\n " + "\n ".join(too_late), + ) + def test_the_stash_agrees_with_the_fields_it_declared(self): mismatches = [] for shape in _SHAPES: diff --git a/test/registered/unit/server_args/test_resolution_reads_no_bag.py b/test/registered/unit/server_args/test_resolution_reads_no_bag.py new file mode 100644 index 000000000..e9a97d946 --- /dev/null +++ b/test/registered/unit/server_args/test_resolution_reads_no_bag.py @@ -0,0 +1,291 @@ +"""Resolution does not read the config bags, because they do not exist yet. + +The bags are projected from what resolution decides, so anything the pipeline +calls has to read the resolving state instead — `resolved_view(server_args)`, +or the view a handler already holds. A bag read reached from resolution raises +`config namespace ... not published`, and only on the branch that reaches it: +the diffusion-LM page-size pass needed one model family, the Marlin LoRA +validation needed one MoE runner backend. Both were written, merged into a +branch, and stayed green for everything except the configuration that triggers +them. + +`test_publish_precedes_bag_reads.py` is the same worry from the other side, but +it walks the *process entries* — it cannot see a helper the pipeline calls, and +neither of the two above appeared in it. + +The walk starts from three places: the symbols the pipeline imports, the live +resolution registries (every pass and override provider, taken from the +registries themselves rather than from the decorator that put it there -- most +providers register through a helper call), and the passes named at a +`run_post_process_pass(sa, fn)` call site. From there it follows calls +in-module, one hop out, and matches an accessor whether it is spelled bare or +through an object. + +What this still cannot see: a bag read reached through a method rather than a +module-level function, one behind an import the walk does not follow, and one +in a callable that reaches the pipeline through a variable no call site names. +It is a ratchet, not a proof. +""" + +import ast +import inspect +import pathlib +import unittest + +import sglang +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") + +_SRT = pathlib.Path(sglang.__file__).resolve().parent / "srt" + + +def _accessor_names(): + """Every bag accessor `runtime_context` exports, read from the module. + + Listing them by hand is how this went stale once already: the list had + eighteen names while the module exported twenty-five, so a resolution-time + `get_flags().x` or `get_resources().y` would have walked straight past. + """ + tree = ast.parse((_SRT / "runtime_context.py").read_text(encoding="utf-8-sig")) + names = { + node.name + for node in tree.body + if isinstance(node, ast.FunctionDef) + and (node.name.startswith("get_") or node.name.startswith("configured_")) + } + # The context object itself is not a bag: it exists before anything is + # published, and `declare_late_resolution` calls it deliberately to find + # out whether the record it was handed has been published yet. + return frozenset(names - {"get_context"}) + + +_BAG_ACCESSORS = _accessor_names() + +# `get_device` also names the device-string utility and the platform method, +# so only the bare spelling is the accessor. +_ATTRIBUTE_SPELLED = _BAG_ACCESSORS - {"get_device"} + +# The pipeline itself and the mechanism it publishes through: `runtime_context` +# defines the accessors, and `arg_groups` is the pipeline's own code. +_OWN = ("server_args.py", "runtime_context.py") + + +def _module_of(name): + """`sglang.srt.a.b` -> the file, if it is one of ours.""" + if not name or not name.startswith("sglang.srt."): + return None + rel = name[len("sglang.srt.") :].replace(".", "/") + for candidate in (_SRT / f"{rel}.py", _SRT / rel / "__init__.py"): + if candidate.exists(): + return candidate + return None + + +def _imported_symbols(paths): + """{module file: {symbol names imported from it}} across the given sources.""" + out = {} + for path in paths: + for node in ast.walk(ast.parse(path.read_text(encoding="utf-8-sig"))): + if not isinstance(node, ast.ImportFrom): + continue + target = _module_of(node.module) + if target is None or target.name in _OWN: + continue + out.setdefault(target, set()).update(alias.name for alias in node.names) + return out + + +def _registry_functions(): + """Every callable the resolution registries will call, from the registries. + + Not from decorator syntax: most model-override providers register through + a `_register_for(...)` helper rather than a decorator, so a scan keyed on + the decorator name walked past all of them -- 39 entries found where the + registries hold 65. However a provider registers, it is in the registry + once its module is imported, and `inspect` says where it came from. + """ + from sglang.srt.arg_groups import overrides + + functions = list(overrides.POST_PROCESS_PASSES) + functions += [fn for fns in overrides._MODEL_OVERRIDE_FNS.values() for fn in fns] + functions += [fn for _predicate, fn in overrides._PREDICATE_OVERRIDE_FNS] + return functions + + +def _registered_entries(): + """Entries the import map cannot reach: passes and override providers. + + A pass arrives at the pipeline as a value, and the registry calls its + providers by dictionary lookup. Both run during resolution, so a bag read + inside one raises exactly like a bag read in a handler -- and neither is + named by an import the walk can follow. + """ + entries = set() + for fn in _registry_functions(): + target = inspect.unwrap(fn) + name = getattr(target, "__name__", "") + if not name or name == "": + continue + source = inspect.getsourcefile(target) + if source is None: + continue + path = pathlib.Path(source).resolve() + if _SRT in path.parents: + entries.add((path, name)) + # A pass handed over by value is in no registry, so its call sites are read + # from the source. The entry carries the *defining* file: `_reaches_a_bag` + # walks functions in the entry's file, so a call-site key walks nothing. + by_value = set() + trees = {} + for path in sorted(_SRT.rglob("*.py")): + try: + trees[path] = ast.parse(path.read_text(encoding="utf-8-sig")) + except SyntaxError: + continue + for path, tree in trees.items(): + for node in ast.walk(tree): + if ( + isinstance(node, ast.Call) + and isinstance(node.func, ast.Name) + and node.func.id == "run_post_process_pass" + and len(node.args) >= 2 + and isinstance(node.args[1], ast.Name) + ): + by_value.add(node.args[1].id) + for name in sorted(by_value): + defined_in = [ + path + for path, tree in trees.items() + if any( + isinstance(node, ast.FunctionDef) and node.name == name + for node in tree.body + ) + ] + if not defined_in: + raise AssertionError( + f"pass {name!r} is handed to run_post_process_pass by value " + "but defined in no scanned module; the walk cannot see it" + ) + for path in defined_in: + entries.add((path, name)) + return entries + + +def _reaches_a_bag(path, entry): + """Does `entry` in `path` reach a bag accessor, following calls in-module?""" + tree = ast.parse(path.read_text(encoding="utf-8-sig")) + functions = { + node.name: node + for node in ast.walk(tree) + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) + } + seen = set() + + def walk(name): + if name in seen or name not in functions: + return None + seen.add(name) + for node in ast.walk(functions[name]): + if not isinstance(node, ast.Call): + continue + if isinstance(node.func, ast.Attribute): + # `rc.get_exec()`, `self.get_schedule()`: the same accessor + # reached through a module alias or an object. + if node.func.attr in _ATTRIBUTE_SPELLED: + return node.lineno + continue + if not isinstance(node.func, ast.Name): + continue + if node.func.id in _BAG_ACCESSORS: + return node.lineno + found = walk(node.func.id) + if found is not None: + return found + return None + + return walk(entry) + + +class TestResolutionReadsNoBag(CustomTestCase): + def test_the_accessor_set_is_derived_and_whole(self): + """A shrunken accessor set would make every other check pass quietly.""" + self.assertGreaterEqual( + len(_BAG_ACCESSORS), + 20, + f"only {len(_BAG_ACCESSORS)} accessors were derived from " + "runtime_context; the derivation broke", + ) + # Spelled out so a rename that drops one fails here. + for name in ("get_exec", "get_flags", "get_parallel", "get_resources"): + self.assertIn(name, _BAG_ACCESSORS) + + def test_the_walk_finds_something_to_walk(self): + """A collapsed import map would make the pin vacuous.""" + imported = _imported_symbols( + [_SRT / "server_args.py", _SRT / "arg_groups" / "overrides.py"] + ) + self.assertGreater( + len(imported), + 20, + f"the pipeline only imports from {len(imported)} of our modules; " + "the scan broke", + ) + + def test_the_registered_entries_are_found(self): + """The passes and providers are the half the import map cannot see.""" + entries = _registered_entries() + self.assertGreater( + len(entries), + 60, + f"only {len(entries)} passes and providers were found; the scan broke", + ) + # Every registered callable that lives in our tree has to appear: the + # derivation reads the registries through `inspect`, so an interpreter + # that imported a *different* checkout would resolve them outside + # `_SRT` and quietly leave the walk with nothing to walk. + missing = sorted( + name + for name in ( + getattr(inspect.unwrap(fn), "__name__", "") + for fn in _registry_functions() + ) + if name + and name != "" + and name not in {entry for _path, entry in entries} + ) + self.assertEqual( + missing, + [], + "a registered pass or provider did not resolve to a file under " + f"{_SRT}; the entry set is narrower than the registries:\n " + + "\n ".join(missing), + ) + + def test_nothing_the_pipeline_calls_reads_a_bag(self): + imported = _imported_symbols( + [_SRT / "server_args.py", _SRT / "arg_groups" / "overrides.py"] + ) + reachable = { + (path, symbol) for path, symbols in imported.items() for symbol in symbols + } | _registered_entries() + found = [] + for path, symbol in sorted(reachable): + line = _reaches_a_bag(path, symbol) + if line is not None: + found.append( + f"{path.relative_to(_SRT)}:{line} reached from " + f"{symbol}(), which resolution calls" + ) + self.assertEqual( + found, + [], + "resolution reaches a config-bag read, which raises on whichever " + "branch gets there first; read the resolving state instead:\n " + + "\n ".join(found), + ) + + +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 index 8af3a01a1..6d39fe0c3 100644 --- a/test/registered/unit/test_global_config_read_ratchet.py +++ b/test/registered/unit/test_global_config_read_ratchet.py @@ -56,6 +56,13 @@ _CONFIGURED_SIZE_CALL_SITES = { "the launch path decides how many scheduler processes to spawn; it runs " "before any of them exists, so there is no group to ask" ), + ("srt/entrypoints/engine.py", "configured_attn_cp_size"): ( + "the launcher's per-TP-rank layout, computed while deciding what to " + "spawn -- the groups it is laying out do not exist yet" + ), + ("srt/entrypoints/engine.py", "configured_moe_dp_size"): ( + "the MoE factor of that same pre-spawn layout" + ), ("srt/ray/engine.py", "configured_pp_size"): ( "the Ray driver sizes the actor placement group; the actors it is about " "to create are the ones that will hold the process groups" @@ -137,6 +144,16 @@ _CONFIGURED_SIZE_CALL_SITES = { ("srt/speculative/frozen_kv_mtp_cuda_graph_runner.py", "configured_pp_size"): ( "the same draft window, frozen-KV MTP" ), + ("srt/managers/data_parallel_controller.py", "configured_pp_size"): ( + "the controller lays out its schedulers' ranks before spawning them, so " + "the groups it is sizing for do not exist yet" + ), + ("srt/managers/data_parallel_controller.py", "configured_attn_cp_size"): ( + "the same pre-spawn rank arithmetic" + ), + ("srt/managers/data_parallel_controller.py", "configured_moe_dp_size"): ( + "the same pre-spawn rank arithmetic" + ), ("srt/entrypoints/v1_loads.py", "configured_pp_size"): ( "the /v1/loads accelerator count is arithmetic over the launch shape, " "reported from the tokenizer process, which holds no model groups" diff --git a/test/registered/unit/test_supplied_instance_exposure_ratchet.py b/test/registered/unit/test_supplied_instance_exposure_ratchet.py index edb4ad87b..bc94d70ec 100644 --- a/test/registered/unit/test_supplied_instance_exposure_ratchet.py +++ b/test/registered/unit/test_supplied_instance_exposure_ratchet.py @@ -158,14 +158,8 @@ _EXPOSED = { ("configs/model_config.py", "quantization"), ("configs/model_config.py", "speculative_algorithm"), ("configs/model_config.py", "speculative_draft_model_quantization"), - ("entrypoints/engine.py", "attn_cp_size"), ("entrypoints/engine.py", "enable_symm_mem"), - ("entrypoints/engine.py", "moe_dp_size"), ("entrypoints/engine.py", "reasoning_parser"), - ( - "entrypoints/engine.py", - "remote_instance_weight_loader_start_seed_via_transfer_engine", - ), ("entrypoints/engine.py", "tool_call_parser"), ("eplb/eplb_manager.py", "ep_dispatch_algorithm"), ("eplb/eplb_manager.py", "expert_distribution_recorder_buffer_size"), @@ -177,12 +171,6 @@ _EXPOSED = { ("layers/flashinfer_comm_fusion.py", "flashinfer_allreduce_fusion_backend"), ("lora/lora_manager.py", "enable_lora_overlap_loading"), ("lora/marlin_lora_temp/policy.py", "lora_paths"), - ("managers/data_parallel_controller.py", "attn_cp_size"), - ("managers/data_parallel_controller.py", "disaggregation_mode"), - ("managers/data_parallel_controller.py", "load_balance_method"), - ("managers/data_parallel_controller.py", "moe_dp_size"), - ("managers/data_parallel_controller.py", "pp_size"), - ("managers/data_parallel_controller.py", "soft_watchdog_timeout"), ("parser/template_detection.py", "model_path"), ("speculative/adaptive_spec_params.py", "speculative_algorithm"), ("speculative/adaptive_spec_params.py", "speculative_eagle_topk"),