config: a defensive publish must not re-project over a live process (#35904)
This commit is contained in:
@@ -206,8 +206,6 @@ def launch_server(server_args: ServerArgs):
|
||||
global dp_dispatcher, encoder, encoder_scheduler, local_runtime, send_sockets
|
||||
|
||||
configure_logger(server_args, prefix=" encode_server")
|
||||
# Publish before the launch path reads configuration; each encoder built
|
||||
# below re-projects the same object in its process.
|
||||
publish(server_args, role="encoder")
|
||||
if get_parallel().dp_size > 1:
|
||||
dp_dispatcher = launch_dp_runtime(server_args)
|
||||
|
||||
@@ -57,12 +57,12 @@ from sglang.srt.multimodal.encoder_preprocessing import (
|
||||
)
|
||||
from sglang.srt.observability.metrics_collector import EncoderMetricsCollector
|
||||
from sglang.srt.runtime_context import (
|
||||
ensure_published,
|
||||
get_device,
|
||||
get_disagg,
|
||||
get_exec,
|
||||
get_mm,
|
||||
get_model,
|
||||
publish,
|
||||
)
|
||||
from sglang.srt.server_args import ServerArgs
|
||||
from sglang.srt.utils import configure_media_url_security
|
||||
@@ -448,9 +448,7 @@ class MMEncoder:
|
||||
``base_gpu_id + rank`` — the DP launcher's per-worker placement. It is
|
||||
this instance's value, not a config change, so it travels as an
|
||||
argument."""
|
||||
# The DP and TP encoder workers are spawned, so this constructor is
|
||||
# the first publish in those processes.
|
||||
publish(server_args, role="encoder")
|
||||
ensure_published(server_args, role="encoder")
|
||||
logger.info(f"init MMEncoder {rank}/{server_args.tp_size}")
|
||||
self.server_args = server_args
|
||||
configure_media_url_security(
|
||||
|
||||
@@ -1117,9 +1117,6 @@ class Engine(EngineScoreMixin, EngineBase):
|
||||
):
|
||||
resolve_auto_parsers(server_args)
|
||||
|
||||
# Resolution is complete here; this process goes on to host the
|
||||
# tokenizer manager or the multi-tokenizer router, whose own publish
|
||||
# re-projects the same object.
|
||||
publish(server_args, role="tokenizer")
|
||||
|
||||
# Launch daemons (daemon mode only). The handles travel back to the
|
||||
|
||||
@@ -122,6 +122,7 @@ from sglang.srt.observability.request_metrics_exporter import (
|
||||
)
|
||||
from sglang.srt.observability.trace import SpanAttributes, extract_trace_headers
|
||||
from sglang.srt.runtime_context import (
|
||||
ensure_published,
|
||||
get_context,
|
||||
get_device,
|
||||
get_disagg,
|
||||
@@ -138,7 +139,6 @@ from sglang.srt.sampling.sampling_params import SamplingParams
|
||||
from sglang.srt.server_args import (
|
||||
PortArgs,
|
||||
ServerArgs,
|
||||
set_global_server_args_for_tokenizer,
|
||||
)
|
||||
from sglang.srt.utils import (
|
||||
configure_gc_warning,
|
||||
@@ -405,9 +405,7 @@ class TokenizerManager(TokenizerControlMixin, TokenizerManagerScoreMixin):
|
||||
):
|
||||
# Parse args
|
||||
self.server_args = server_args
|
||||
# In a tokenizer-worker process this is the process's first publish;
|
||||
# the in-process path re-projects the object the launcher published.
|
||||
set_global_server_args_for_tokenizer(server_args)
|
||||
ensure_published(server_args, role="tokenizer")
|
||||
self.startup_time: Optional[Dict[str, Any]] = None
|
||||
self.elastic_worker_count = get_parallel().dp_size
|
||||
self.elastic_pending_ep_size = None
|
||||
|
||||
@@ -168,6 +168,7 @@ from sglang.srt.model_executor.runner import (
|
||||
)
|
||||
from sglang.srt.platforms import current_platform
|
||||
from sglang.srt.runtime_context import (
|
||||
ensure_published,
|
||||
get_context,
|
||||
get_exec,
|
||||
get_global_dwdp_manager,
|
||||
@@ -188,7 +189,6 @@ from sglang.srt.server_args import ( # noqa: F401 (re-export)
|
||||
ServerArgs,
|
||||
add_chunked_prefix_cache_attention_backend,
|
||||
get_global_server_args,
|
||||
set_global_server_args_for_scheduler,
|
||||
)
|
||||
from sglang.srt.speculative.spec_info import SpeculativeAlgorithm
|
||||
from sglang.srt.speculative.spec_utils import resolve_num_tokens_per_req
|
||||
@@ -336,7 +336,7 @@ class ModelRunner:
|
||||
# construction (benchmark/one_batch, the manual runner tests) has no
|
||||
# earlier publish.
|
||||
if not is_draft_worker:
|
||||
set_global_server_args_for_scheduler(server_args)
|
||||
ensure_published(server_args, role="scheduler")
|
||||
# Set by maybe_init_lora_manager; stays None when LoRA is off and on
|
||||
# draft runners, which serve adapters' target model unadapted.
|
||||
self.lora_manager: Optional[LoRAManager] = None
|
||||
|
||||
@@ -48,6 +48,7 @@ from __future__ import annotations
|
||||
|
||||
import dataclasses
|
||||
import functools
|
||||
import logging
|
||||
import math
|
||||
import os
|
||||
import sys
|
||||
@@ -57,6 +58,8 @@ from typing import TYPE_CHECKING, Any
|
||||
if TYPE_CHECKING:
|
||||
from sglang.srt.server_args import ServerArgs
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# Imported lazily so this module has no import-time dependencies: any module can
|
||||
# import get_parallel at module level without risking an import cycle.
|
||||
@@ -1304,7 +1307,18 @@ def publish(server_args, *, role: str, hf_config: Any = None) -> RuntimeContext:
|
||||
f"publish role {role!r} has no ROLE_NAMESPACE_SETS entry; declare "
|
||||
"its namespace set (None for the full tree)."
|
||||
)
|
||||
discarded = _CONTEXT.overrides_log()
|
||||
_CONTEXT.set_server_args(server_args)
|
||||
if discarded:
|
||||
logger.warning(
|
||||
"publish(role=%s) re-projected the config bags and dropped %d "
|
||||
"override(s) taken since the last publish: %s",
|
||||
role,
|
||||
len(discarded),
|
||||
", ".join(
|
||||
f"{source}({', '.join(sorted(fields))})" for source, fields in discarded
|
||||
),
|
||||
)
|
||||
_CONTEXT._publish_role = role
|
||||
if _ROLE_NS_MODE == "record":
|
||||
# The '-' marker distinguishes a zero-read role from a process where
|
||||
@@ -1320,6 +1334,31 @@ def publish(server_args, *, role: str, hf_config: Any = None) -> RuntimeContext:
|
||||
return _CONTEXT
|
||||
|
||||
|
||||
def ensure_published(server_args, *, role: str) -> RuntimeContext:
|
||||
"""Publish unless this exact record is already published under this role.
|
||||
|
||||
Three constructors publish defensively, because each can be built with
|
||||
nothing published before it -- `ModelRunner` (a benchmark harness, the
|
||||
manual runner tests), `TokenizerManager`, and `MMEncoder` (spawned encoder
|
||||
workers). Inside a process that already published the same record,
|
||||
publishing again re-projects the bags: every `override()` taken between the
|
||||
two calls is discarded, and the provenance log with it.
|
||||
|
||||
No override sits in one of those windows today, so this removes a hazard
|
||||
rather than a live bug. It is worth removing anyway: the drop is silent, it
|
||||
depends on where a constructor happens to sit relative to the overrides
|
||||
around it, and `publish` now says what a re-projection discarded so the
|
||||
next one is loud.
|
||||
|
||||
So these callers ask for the end state -- this record, this role, published
|
||||
-- and get a no-op when that already holds. An engine rebuild still calls
|
||||
`publish` directly, because there the reset is the point.
|
||||
"""
|
||||
if _CONTEXT._server_args is server_args and _CONTEXT._publish_role == role:
|
||||
return _CONTEXT
|
||||
return publish(server_args, role=role)
|
||||
|
||||
|
||||
def publish_role() -> str | None:
|
||||
"""The role recorded by the last ``publish`` (None for a legacy set)."""
|
||||
return _CONTEXT._publish_role
|
||||
|
||||
@@ -0,0 +1,117 @@
|
||||
"""Who installs the startup record into the runtime context, derived from code.
|
||||
|
||||
Two guards need this answer and neither should keep its own list: matching the
|
||||
spellings by hand is how `ensure_published` once read as *not* publishing,
|
||||
which turned a correct module into a reported violation. A publisher is
|
||||
defined by what it does -- it reaches ``RuntimeContext.set_server_args`` --
|
||||
and a *constructor* publisher is an ``__init__`` that calls one.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import ast
|
||||
import pathlib
|
||||
from typing import Dict, Set, Tuple
|
||||
|
||||
_DEFINING_MODULES = ("runtime_context.py", "server_args.py")
|
||||
|
||||
|
||||
def publisher_names(srt_root: pathlib.Path) -> frozenset:
|
||||
"""Module-level functions that transitively install the record."""
|
||||
reaches: Set[str] = set()
|
||||
graph: Dict[str, Set[str]] = {}
|
||||
for relative in _DEFINING_MODULES:
|
||||
tree = ast.parse((srt_root / relative).read_text(encoding="utf-8-sig"))
|
||||
for node in tree.body:
|
||||
if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
|
||||
continue
|
||||
called: Set[str] = set()
|
||||
installs = False
|
||||
for inner in ast.walk(node):
|
||||
if not isinstance(inner, ast.Call):
|
||||
continue
|
||||
if (
|
||||
isinstance(inner.func, ast.Attribute)
|
||||
and inner.func.attr == "set_server_args"
|
||||
):
|
||||
installs = True
|
||||
elif isinstance(inner.func, ast.Name):
|
||||
called.add(inner.func.id)
|
||||
graph[node.name] = called
|
||||
if installs:
|
||||
reaches.add(node.name)
|
||||
growing = True
|
||||
while growing:
|
||||
growing = False
|
||||
for name, called in graph.items():
|
||||
if name not in reaches and called & reaches:
|
||||
reaches.add(name)
|
||||
growing = True
|
||||
return frozenset(reaches)
|
||||
|
||||
|
||||
def constructor_publishers(srt_root: pathlib.Path) -> Set[Tuple[str, str, str]]:
|
||||
"""``{(module, class, publisher)}`` for every ``__init__`` that publishes.
|
||||
|
||||
Keyed by the owning class, not just the module: two constructors in one
|
||||
module would otherwise collapse into a single entry, and a *new* defensive
|
||||
publish added next to a listed one would leave the census unchanged --
|
||||
which is exactly the "adding one fails the pin" property it exists for.
|
||||
|
||||
Reached through a local name or through a module attribute
|
||||
(``runtime_context.publish(...)``), and through a helper defined in the
|
||||
same module -- a constructor that publishes one hop away is the same
|
||||
hazard as one that publishes directly.
|
||||
"""
|
||||
publishers = publisher_names(srt_root)
|
||||
found: Set[Tuple[str, str, str]] = set()
|
||||
for path in sorted(srt_root.rglob("*.py")):
|
||||
try:
|
||||
tree = ast.parse(path.read_text(encoding="utf-8-sig"))
|
||||
except SyntaxError:
|
||||
raise AssertionError(f"unparsable module in the census: {path}")
|
||||
local: Dict[str, ast.AST] = {
|
||||
node.name: node
|
||||
for node in ast.walk(tree)
|
||||
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
|
||||
}
|
||||
|
||||
def publishes(node, seen=None):
|
||||
"""The publisher this callable reaches, if any."""
|
||||
seen = seen if seen is not None else set()
|
||||
for inner in ast.walk(node):
|
||||
if not isinstance(inner, ast.Call):
|
||||
continue
|
||||
if isinstance(inner.func, ast.Name):
|
||||
name = inner.func.id
|
||||
elif isinstance(inner.func, ast.Attribute):
|
||||
name = inner.func.attr
|
||||
else:
|
||||
continue
|
||||
if name in publishers:
|
||||
return name
|
||||
if name in local and name not in seen:
|
||||
seen.add(name)
|
||||
reached = publishes(local[name], seen)
|
||||
if reached is not None:
|
||||
return reached
|
||||
return None
|
||||
|
||||
for owner in ast.walk(tree):
|
||||
if not isinstance(owner, ast.ClassDef):
|
||||
continue
|
||||
for node in owner.body:
|
||||
if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
|
||||
continue
|
||||
if node.name != "__init__":
|
||||
continue
|
||||
reached = publishes(node)
|
||||
if reached is not None:
|
||||
found.add(
|
||||
(
|
||||
path.relative_to(srt_root).as_posix(),
|
||||
owner.name,
|
||||
reached,
|
||||
)
|
||||
)
|
||||
return found
|
||||
@@ -31,7 +31,7 @@ _RATCHETS = [
|
||||
(
|
||||
"set_global_server_args_for_*",
|
||||
r"\bset_global_server_args_for_(?:scheduler|tokenizer)\s*\(",
|
||||
4,
|
||||
2,
|
||||
),
|
||||
]
|
||||
|
||||
|
||||
@@ -42,6 +42,7 @@ import unittest
|
||||
|
||||
import sglang
|
||||
from sglang.test.ci.ci_register import register_cpu_ci
|
||||
from sglang.test.config_publishers import publisher_names
|
||||
from sglang.test.test_utils import CustomTestCase
|
||||
|
||||
register_cpu_ci(est_time=40, suite="base-a-test-cpu")
|
||||
@@ -158,6 +159,9 @@ def _calls(fn):
|
||||
return out
|
||||
|
||||
|
||||
_PUBLISH_NAMES = publisher_names(_PACKAGE_ROOT / "srt")
|
||||
|
||||
|
||||
class _Module:
|
||||
"""One parsed module: what it calls the config API, and what it defines.
|
||||
|
||||
@@ -183,9 +187,7 @@ class _Module:
|
||||
if node.module in _CONFIG_MODULES:
|
||||
for alias in node.names:
|
||||
local = alias.asname or alias.name
|
||||
if alias.name == "publish" or alias.name.startswith(
|
||||
"set_global_server_args"
|
||||
):
|
||||
if alias.name in _PUBLISH_NAMES:
|
||||
self.publishers.add(local)
|
||||
elif alias.name in _ACCESSORS:
|
||||
self.accessors.add(local)
|
||||
|
||||
@@ -19,11 +19,15 @@ from sglang.srt.runtime_context import (
|
||||
ParallelContext,
|
||||
RuntimeContext,
|
||||
_FlagGroupBase,
|
||||
ensure_published,
|
||||
get_context,
|
||||
get_exec,
|
||||
get_flags,
|
||||
get_parallel,
|
||||
get_server_args,
|
||||
max_speculative_num_draft_tokens,
|
||||
publish,
|
||||
publish_role,
|
||||
reset_context,
|
||||
)
|
||||
from sglang.srt.server_args import ServerArgs
|
||||
@@ -248,6 +252,119 @@ class TestServerArgsOwnership(_IsolatedServerArgs):
|
||||
get_server_args()
|
||||
|
||||
|
||||
class TestEnsurePublished(_IsolatedServerArgs):
|
||||
"""A defensive publish must not re-project over a live process.
|
||||
|
||||
Three constructors publish because each can be built with nothing published
|
||||
first: `ModelRunner`, `TokenizerManager`, `MMEncoder`. Inside a process that
|
||||
already published the same record, publishing again re-projects the bags --
|
||||
discarding every `override()` taken since, and the provenance log with it.
|
||||
|
||||
No current override sits in one of those windows, so what these assertions
|
||||
protect is the mechanism, not a reproduction: the drop is silent and depends
|
||||
on where a constructor happens to sit relative to the overrides around it.
|
||||
"""
|
||||
|
||||
def _record(self, **fields):
|
||||
return ServerArgs(model_path="dummy", **fields)
|
||||
|
||||
def test_a_second_publish_of_the_same_record_keeps_the_overrides(self):
|
||||
record = self._record(grammar_backend="xgrammar")
|
||||
publish(record, role="scheduler")
|
||||
get_context().override("grammar.import_fallback", grammar_backend="none")
|
||||
|
||||
ensure_published(record, role="scheduler")
|
||||
|
||||
self.assertEqual(
|
||||
get_exec().kernel.grammar_backend,
|
||||
"none",
|
||||
"the constructor's publish re-projected the bags, so the import "
|
||||
"fallback was discarded and the process reports a backend it is "
|
||||
"not using",
|
||||
)
|
||||
self.assertEqual(
|
||||
len(get_context().overrides_log()),
|
||||
1,
|
||||
"the provenance of the override went with it",
|
||||
)
|
||||
|
||||
def test_a_different_record_is_published(self):
|
||||
first = self._record(grammar_backend="xgrammar")
|
||||
publish(first, role="scheduler")
|
||||
second = self._record(grammar_backend="llguidance")
|
||||
|
||||
ensure_published(second, role="scheduler")
|
||||
|
||||
self.assertIs(get_server_args(), second)
|
||||
self.assertEqual(get_exec().kernel.grammar_backend, "llguidance")
|
||||
|
||||
def test_an_empty_slot_is_published(self):
|
||||
"""The standalone case the defensive publish exists for."""
|
||||
reset_context()
|
||||
record = self._record(grammar_backend="xgrammar")
|
||||
|
||||
ensure_published(record, role="scheduler")
|
||||
|
||||
self.assertIs(get_server_args(), record)
|
||||
self.assertEqual(publish_role(), "scheduler")
|
||||
|
||||
def test_the_same_record_under_a_different_role_is_republished(self):
|
||||
"""The role decides which namespaces this process may read."""
|
||||
record = self._record()
|
||||
publish(record, role="tokenizer")
|
||||
|
||||
ensure_published(record, role="scheduler")
|
||||
|
||||
self.assertEqual(publish_role(), "scheduler")
|
||||
|
||||
def test_every_constructor_that_publishes_is_classified(self):
|
||||
"""A new constructor publish has to say which of the two it is.
|
||||
|
||||
Publishing in a constructor is right when the constructor *is* the
|
||||
entry -- a spawned worker, the Ray actor that stands in for
|
||||
`run_scheduler_process`, an `Engine` being (re)built, where resetting
|
||||
the bags is the point -- and wrong when the process is already live
|
||||
with the same record, where it silently drops overrides. The
|
||||
difference is not visible in the syntax, so the census is pinned:
|
||||
adding one fails here until it is classified.
|
||||
|
||||
Both the publisher set and "which `__init__` reaches one" come from
|
||||
`sglang.test.config_publishers`, which derives them from the code --
|
||||
a hand-written spelling list here missed a constructor that publishes
|
||||
one hop away through a helper. The derivation follows helpers defined
|
||||
in the same module; a constructor that publishes through a helper in
|
||||
*another* module is not seen, which is the one hole left here.
|
||||
"""
|
||||
import pathlib
|
||||
|
||||
import sglang
|
||||
from sglang.test.config_publishers import constructor_publishers
|
||||
|
||||
srt = pathlib.Path(sglang.__file__).resolve().parent / "srt"
|
||||
self.assertEqual(
|
||||
constructor_publishers(srt),
|
||||
{
|
||||
# Entries: nothing published yet, or a rebuild that must not
|
||||
# inherit the previous engine's runtime overrides.
|
||||
("entrypoints/engine.py", "Engine", "publish"),
|
||||
("ray/scheduler_actor.py", "SchedulerActor", "publish"),
|
||||
# Defensive: the process is usually already live with this
|
||||
# record, and `launch_server` publishes before building the
|
||||
# in-process encoder.
|
||||
("disaggregation/encoder/server.py", "MMEncoder", "ensure_published"),
|
||||
(
|
||||
"managers/tokenizer_manager.py",
|
||||
"TokenizerManager",
|
||||
"ensure_published",
|
||||
),
|
||||
("model_executor/model_runner.py", "ModelRunner", "ensure_published"),
|
||||
},
|
||||
"a constructor publishes and this census does not know which kind "
|
||||
"it is; an entry uses publish(), one that may run inside a live "
|
||||
"process with the same record uses ensure_published()",
|
||||
)
|
||||
|
||||
|
||||
class TestServerArgsScopedOverride(_IsolatedServerArgs):
|
||||
"""ctx.override_server_args: the config tier's scoped test override —
|
||||
tests force execution paths by overriding the context, not by
|
||||
|
||||
Reference in New Issue
Block a user