config: publishing is the process entry's job (#36251)

This commit is contained in:
Cheng Wan
2026-08-26 04:58:41 -07:00
committed by GitHub
parent 702de26310
commit d7b144f64e
21 changed files with 209 additions and 108 deletions
+6 -1
View File
@@ -82,7 +82,7 @@ from sglang.srt.mem_cache.base_prefix_cache import EvictParams
from sglang.srt.model_executor.cuda_graph_config import Phase, cuda_graph_fully_disabled from sglang.srt.model_executor.cuda_graph_config import Phase, cuda_graph_fully_disabled
from sglang.srt.model_executor.forward_batch_info import ForwardBatch from sglang.srt.model_executor.forward_batch_info import ForwardBatch
from sglang.srt.model_executor.model_runner import ModelRunner from sglang.srt.model_executor.model_runner import ModelRunner
from sglang.srt.runtime_context import get_parallel, get_schedule from sglang.srt.runtime_context import get_parallel, get_schedule, publish
from sglang.srt.sampling.sampling_params import SamplingParams from sglang.srt.sampling.sampling_params import SamplingParams
from sglang.srt.server_args import PortArgs, ServerArgs from sglang.srt.server_args import PortArgs, ServerArgs
from sglang.srt.speculative.spec_info import SpeculativeAlgorithm from sglang.srt.speculative.spec_info import SpeculativeAlgorithm
@@ -681,6 +681,8 @@ def correctness_test(
gpu_id, gpu_id,
tp_rank, tp_rank,
): ):
publish(server_args, role="scheduler")
# Configure the logger # Configure the logger
configure_logger(server_args, prefix=f" TP{tp_rank}") configure_logger(server_args, prefix=f" TP{tp_rank}")
rank_print = print if tp_rank == 0 else lambda *args, **kwargs: None rank_print = print if tp_rank == 0 else lambda *args, **kwargs: None
@@ -881,6 +883,9 @@ def latency_test(
gpu_id, gpu_id,
tp_rank, tp_rank,
): ):
# `main` runs this inline for tp_size == 1 and spawns it per rank otherwise;
# a spawned child arrives with nothing published.
publish(server_args, role="scheduler")
initialize_moe_config(server_args) initialize_moe_config(server_args)
initialize_fp8_gemm_config(server_args) initialize_fp8_gemm_config(server_args)
initialize_fp4_gemm_config(server_args) initialize_fp4_gemm_config(server_args)
+36
View File
@@ -29,6 +29,7 @@ Two declaration forms, keyed on ``hf_config.architectures[0]``:
from __future__ import annotations from __future__ import annotations
import copy
import dataclasses import dataclasses
import inspect import inspect
import json import json
@@ -380,6 +381,41 @@ def resolution_result(server_args: Any, field: str, default: Any = None) -> Any:
return getattr(server_args, field, default) return getattr(server_args, field, default)
def resolution_projection(server_args: Any) -> Dict[str, Any]:
"""Every field's resolved value, nested dataclasses expanded.
The whole-object shape of ``resolution_result``, for the exits that hand out
the entire configuration (``/server_info``, the gRPC and engine readbacks).
They used ``dataclasses.asdict``, which reads the fields -- correct only for
as long as declarations materialize onto the record, and the point of
declaring is that they will not. Field values only: the private resolution
bookkeeping and the ``model_config`` memo that a ``vars()`` dump carried into
the readback are not configuration.
"""
return {
field.name: _plain(resolution_result(server_args, field.name))
for field in dataclasses.fields(server_args)
}
def _plain(value: Any) -> Any:
"""``dataclasses.asdict``'s conversion, applied to one value: dataclasses
become dicts, containers recurse, everything else is deep-copied (a caller
mutating the dump must not reach the live configuration)."""
if dataclasses.is_dataclass(value) and not isinstance(value, type):
return {
field.name: _plain(getattr(value, field.name))
for field in dataclasses.fields(value)
}
if isinstance(value, tuple) and hasattr(value, "_fields"): # namedtuple
return type(value)(*(_plain(item) for item in value))
if isinstance(value, (list, tuple)):
return type(value)(_plain(item) for item in value)
if isinstance(value, dict):
return type(value)((_plain(k), _plain(v)) for k, v in value.items())
return copy.deepcopy(value)
def resolved_view(server_args: Any) -> ResolvedView: def resolved_view(server_args: Any) -> ResolvedView:
"""Read-only view of the resolving configuration for mid-resolution code """Read-only view of the resolving configuration for mid-resolution code
that is not a pass (``__post_init__`` handlers and hooks). Internal to that is not a pass (``__post_init__`` handlers and hooks). Internal to
@@ -24,7 +24,7 @@ from smg_grpc_proto import sglang_encoder_pb2, sglang_encoder_pb2_grpc
from sglang.srt.disaggregation.encoder.server import MMEncoder, launch_encoder from sglang.srt.disaggregation.encoder.server import MMEncoder, launch_encoder
from sglang.srt.managers.io_struct import async_sock_send, wrap_as_pickle from sglang.srt.managers.io_struct import async_sock_send, wrap_as_pickle
from sglang.srt.managers.schedule_batch import Modality from sglang.srt.managers.schedule_batch import Modality
from sglang.srt.runtime_context import get_disagg from sglang.srt.runtime_context import get_disagg, publish
from sglang.srt.server_args import PortArgs, ServerArgs from sglang.srt.server_args import PortArgs, ServerArgs
from sglang.srt.utils import random_uuid from sglang.srt.utils import random_uuid
from sglang.srt.utils.network import NetworkAddress, get_zmq_socket from sglang.srt.utils.network import NetworkAddress, get_zmq_socket
@@ -201,6 +201,7 @@ class SGLangEncoderServer(SGLangEncoderServicer):
async def serve_grpc_encoder(server_args: ServerArgs): async def serve_grpc_encoder(server_args: ServerArgs):
publish(server_args, role="encoder")
ctx = mp.get_context("spawn") ctx = mp.get_context("spawn")
zmq_ctx = zmq.asyncio.Context(10) zmq_ctx = zmq.asyncio.Context(10)
ipc_path_prefix = random_uuid() ipc_path_prefix = random_uuid()
@@ -53,6 +53,7 @@ from sglang.srt.runtime_context import (
get_observability, get_observability,
get_parallel, get_parallel,
get_serving, get_serving,
publish,
) )
from sglang.srt.server_args import PortArgs, ServerArgs from sglang.srt.server_args import PortArgs, ServerArgs
from sglang.srt.utils import configure_logger, random_uuid, set_prometheus_multiproc_dir from sglang.srt.utils import configure_logger, random_uuid, set_prometheus_multiproc_dir
@@ -1473,6 +1474,7 @@ def launch_dp_worker(
dispatch_path: str, dispatch_path: str,
result_path: str, result_path: str,
): ):
publish(server_args, role="encoder")
try: try:
configure_logger(server_args, prefix=f" encode_dp_worker[{dp_rank}]") configure_logger(server_args, prefix=f" encode_dp_worker[{dp_rank}]")
asyncio.run( asyncio.run(
@@ -57,12 +57,13 @@ from sglang.srt.multimodal.encoder_preprocessing import (
) )
from sglang.srt.observability.metrics_collector import EncoderMetricsCollector from sglang.srt.observability.metrics_collector import EncoderMetricsCollector
from sglang.srt.runtime_context import ( from sglang.srt.runtime_context import (
ensure_published, assert_published,
get_device, get_device,
get_disagg, get_disagg,
get_exec, get_exec,
get_mm, get_mm,
get_model, get_model,
publish,
) )
from sglang.srt.server_args import ServerArgs from sglang.srt.server_args import ServerArgs
from sglang.srt.utils import configure_media_url_security from sglang.srt.utils import configure_media_url_security
@@ -448,7 +449,7 @@ class MMEncoder:
``base_gpu_id + rank`` — the DP launcher's per-worker placement. It is ``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 this instance's value, not a config change, so it travels as an
argument.""" argument."""
ensure_published(server_args, role="encoder") assert_published(server_args, role="encoder")
logger.info(f"init MMEncoder {rank}/{server_args.tp_size}") logger.info(f"init MMEncoder {rank}/{server_args.tp_size}")
self.server_args = server_args self.server_args = server_args
configure_media_url_security( configure_media_url_security(
@@ -2036,6 +2037,7 @@ async def _handle_encoder_worker_request(encoder: MMEncoder, request):
def launch_encoder(server_args, schedule_path, dist_init_method, rank): def launch_encoder(server_args, schedule_path, dist_init_method, rank):
publish(server_args, role="encoder")
try: try:
asyncio.run(run_encoder(server_args, schedule_path, dist_init_method, rank)) asyncio.run(run_encoder(server_args, schedule_path, dist_init_method, rank))
except KeyboardInterrupt: except KeyboardInterrupt:
+1 -1
View File
@@ -1342,7 +1342,7 @@ class Engine(EngineScoreMixin, EngineBase):
) )
return msgspec_to_builtins( return msgspec_to_builtins(
{ {
**dataclasses.asdict(self.tokenizer_manager.server_args), **self.tokenizer_manager.server_args.resolved_dict(),
**self._scheduler_init_result.scheduler_infos[0], **self._scheduler_init_result.scheduler_infos[0],
"startup_time": self.tokenizer_manager.startup_time, "startup_time": self.tokenizer_manager.startup_time,
"internal_states": internal_states, "internal_states": internal_states,
+1 -2
View File
@@ -7,7 +7,6 @@ TokenizerManager's event loop.
""" """
import asyncio import asyncio
import dataclasses
import json import json
import logging import logging
from types import SimpleNamespace from types import SimpleNamespace
@@ -423,7 +422,7 @@ class RuntimeHandle:
return json.dumps(result, default=str) return json.dumps(result, default=str)
def get_server_info(self) -> str: def get_server_info(self) -> str:
result: Dict[str, Any] = dataclasses.asdict(self.tokenizer_manager.server_args) result: Dict[str, Any] = self.tokenizer_manager.server_args.resolved_dict()
result.update(self.scheduler_info) result.update(self.scheduler_info)
result["kv_events"] = ( result["kv_events"] = (
self.tokenizer_manager.server_args.describe_kv_events_publisher() self.tokenizer_manager.server_args.describe_kv_events_publisher()
+4 -2
View File
@@ -232,6 +232,8 @@ async def init_multi_tokenizer() -> ServerArgs:
server_args.api_key is None server_args.api_key is None
), "API key is not supported in multi-tokenizer mode" ), "API key is not supported in multi-tokenizer mode"
publish(server_args, role="tokenizer")
# Create a new ipc name for the current process # Create a new ipc name for the current process
port_args.tokenizer_ipc_name = ( port_args.tokenizer_ipc_name = (
f"ipc://{tempfile.NamedTemporaryFile(delete=False).name}" f"ipc://{tempfile.NamedTemporaryFile(delete=False).name}"
@@ -488,6 +490,7 @@ from sglang.srt.runtime_context import (
get_model, get_model,
get_parallel, get_parallel,
get_serving, get_serving,
publish,
) )
elastic_ep_router.route_class = ORJSONRoute elastic_ep_router.route_class = ORJSONRoute
@@ -811,10 +814,9 @@ async def server_info():
server_args = _global_state.tokenizer_manager.server_args server_args = _global_state.tokenizer_manager.server_args
# server_args.model_config is not serializable but should be excluded by asdict.
return msgspec_to_builtins( return msgspec_to_builtins(
{ {
**dataclasses.asdict(server_args), **server_args.resolved_dict(),
**_global_state.scheduler_info, **_global_state.scheduler_info,
"startup_time": _global_state.tokenizer_manager.startup_time, "startup_time": _global_state.tokenizer_manager.startup_time,
"internal_states": internal_states, "internal_states": internal_states,
@@ -123,7 +123,7 @@ from sglang.srt.observability.request_metrics_exporter import (
) )
from sglang.srt.observability.trace import SpanAttributes, extract_trace_headers from sglang.srt.observability.trace import SpanAttributes, extract_trace_headers
from sglang.srt.runtime_context import ( from sglang.srt.runtime_context import (
ensure_published, assert_published,
get_context, get_context,
get_device, get_device,
get_disagg, get_disagg,
@@ -409,7 +409,7 @@ class TokenizerManager(TokenizerControlMixin, TokenizerManagerScoreMixin):
): ):
# Parse args # Parse args
self.server_args = server_args self.server_args = server_args
ensure_published(server_args, role="tokenizer") assert_published(server_args, role="tokenizer")
self.startup_time: Optional[Dict[str, Any]] = None self.startup_time: Optional[Dict[str, Any]] = None
self.elastic_worker_count = get_parallel().config.dp_size self.elastic_worker_count = get_parallel().config.dp_size
self.elastic_pending_ep_size = None self.elastic_pending_ep_size = None
@@ -2040,7 +2040,7 @@ class TokenizerManager(TokenizerControlMixin, TokenizerManagerScoreMixin):
part that cannot be reconstructed afterwards. part that cannot be reconstructed afterwards.
""" """
try: try:
return self.resolved_config_dict(dataclasses.asdict(self.server_args)) return self.resolved_config_dict(self.server_args.resolved_dict())
except Exception as e: except Exception as e:
logger.error(f"Failed to snapshot the resolved config for the dump: {e!r}") logger.error(f"Failed to snapshot the resolved config for the dump: {e!r}")
return None return None
@@ -168,7 +168,7 @@ from sglang.srt.model_executor.runner import (
) )
from sglang.srt.platforms import current_platform from sglang.srt.platforms import current_platform
from sglang.srt.runtime_context import ( from sglang.srt.runtime_context import (
ensure_published, assert_published,
get_context, get_context,
get_device, get_device,
get_exec, get_exec,
@@ -332,13 +332,10 @@ class ModelRunner:
self.dist_port = nccl_port self.dist_port = nccl_port
self.server_args = server_args self.server_args = server_args
self.is_draft_worker = is_draft_worker self.is_draft_worker = is_draft_worker
# Set the global server_args in the scheduler process (target worker # The process entry published; a draft runner is not one (it must not
# only, so a draft init cannot clobber target-derived global state). # clobber the target's config), so only the target checks.
# Before the constructor's bag reads (page_size below): a standalone
# construction (benchmark/one_batch, the manual runner tests) has no
# earlier publish.
if not is_draft_worker: if not is_draft_worker:
ensure_published(server_args, role="scheduler") assert_published(server_args, role="scheduler")
# Set by maybe_init_lora_manager; stays None when LoRA is off and on # Set by maybe_init_lora_manager; stays None when LoRA is off and on
# draft runners, which serve adapters' target model unadapted. # draft runners, which serve adapters' target model unadapted.
self.lora_manager: Optional[LoRAManager] = None self.lora_manager: Optional[LoRAManager] = None
+34 -24
View File
@@ -969,11 +969,12 @@ class RuntimeContext:
in a readback: HiCache attach/detach, the generated forward-pass-metrics in a readback: HiCache attach/detach, the generated forward-pass-metrics
endpoint, tunables set via ``/set_internal_state``. endpoint, tunables set via ``/set_internal_state``.
``base`` defaults to ``dict(vars(server_args))`` (matching the legacy ``base`` defaults to ``server_args.resolved_dict()`` -- the record's
``vars`` dump); pass ``dataclasses.asdict(server_args)`` when nested fields as resolution decided them, nested dataclasses expanded. (It used
dataclass fields must be expanded first. Override leaves are flat to be ``dict(vars(server_args))``, which carried the private resolution
``ServerArgs`` field names, so overlaying them onto the top level of bookkeeping and the ``model_config`` memo into the readback.) Override
either base is exact. leaves are flat ``ServerArgs`` field names, so overlaying them onto the
top level of the base is exact.
The log is per process: it carries what *this* process overrode. A The log is per process: it carries what *this* process overrode. A
weight reload records ``model_path`` and ``load_format`` from the weight reload records ``model_path`` and ``load_format`` from the
@@ -984,7 +985,7 @@ class RuntimeContext:
The top-level ``/server_info`` fields are the startup record, not this The top-level ``/server_info`` fields are the startup record, not this
dump. dump.
""" """
d = dict(vars(self.server_args)) if base is None else dict(base) d = self.server_args.resolved_dict() if base is None else dict(base)
for _source, fields in self._overrides_log: for _source, fields in self._overrides_log:
d.update(fields) d.update(fields)
return d return d
@@ -1374,29 +1375,38 @@ def publish(server_args, *, role: str, hf_config: Any = None) -> RuntimeContext:
return _CONTEXT return _CONTEXT
def ensure_published(server_args, *, role: str) -> RuntimeContext: def assert_published(server_args, *, role: str) -> RuntimeContext:
"""Publish unless this exact record is already published under this role. """This record, under this role, is already published -- or fail loud.
Three constructors publish defensively, because each can be built with Publishing is the process entry's job: `run_scheduler_process`,
nothing published before it -- `ModelRunner` (a benchmark harness, the `init_multi_tokenizer`, a spawned encoder worker, the benchmark work
manual runner tests), `TokenizerManager`, and `MMEncoder` (spawned encoder functions. A constructor arriving here unpublished means one of those
workers). Inside a process that already published the same record, entries is missing.
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 A `publish` at this point re-projects the bags over a live process,
rather than a live bug. It is worth removing anyway: the drop is silent, it discarding every `override()` taken since and the provenance log with it,
depends on where a constructor happens to sit relative to the overrides so this raises.
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: if _CONTEXT._server_args is server_args and _CONTEXT._publish_role == role:
return _CONTEXT return _CONTEXT
return publish(server_args, role=role) if _CONTEXT._server_args is None:
detail = "nothing is published in this process"
elif _CONTEXT._server_args is not server_args:
detail = (
"a different record is published "
f"(role={_CONTEXT._publish_role!r}); this constructor was handed "
"one the process never published"
)
else:
detail = (
f"this record is published under role "
f"{_CONTEXT._publish_role!r}, not {role!r}"
)
raise RuntimeError(
f"config not published for role {role!r}: {detail}. The process entry "
"publishes -- add publish(server_args, role=...) there rather than "
"publishing from a constructor."
)
def publish_role() -> str | None: def publish_role() -> str | None:
+15
View File
@@ -3729,6 +3729,21 @@ class ServerArgs:
# handlers ran, not how far they got. # handlers ran, not how far they got.
self._declarations_materialized = True self._declarations_materialized = True
def resolved_dict(self) -> Dict[str, Any]:
"""This configuration as a plain dict of resolved field values.
What the whole-object readbacks report (`/server_info` and its gRPC and
in-process twins). `dataclasses.asdict(self)` reads the fields, which
carry resolution's result only while declarations materialize onto the
record; this reads the declarations, so it keeps answering with what
resolution decided once they stop. Nested dataclass fields are expanded
the way `asdict` expands them; the private resolution bookkeeping and the
`model_config` memo are not fields and do not appear.
"""
from sglang.srt.arg_groups.overrides import resolution_projection
return resolution_projection(self)
def replace_resolved(self, source: str, **changes: Any) -> ServerArgs: def replace_resolved(self, source: str, **changes: Any) -> ServerArgs:
"""A copy of this record that stays resolved, and says what it changed. """A copy of this record that stays resolved, and says what it changed.
+2 -2
View File
@@ -1,8 +1,8 @@
"""Who installs the startup record into the runtime context, derived from code. """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 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, spellings by hand is how the constructors' old defensive publish once read as
which turned a correct module into a reported violation. A publisher is *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`` -- defined by what it does -- it reaches ``RuntimeContext.set_server_args`` --
and a *constructor* publisher is an ``__init__`` that calls one. and a *constructor* publisher is an ``__init__`` that calls one.
""" """
@@ -19,6 +19,7 @@ from sglang.srt.distributed.parallel_state_wrapper import ParallelState
from sglang.srt.managers.schedule_batch import Req, ScheduleBatch from sglang.srt.managers.schedule_batch import Req, ScheduleBatch
from sglang.srt.model_executor.forward_batch_info import ForwardBatch, ForwardMode from sglang.srt.model_executor.forward_batch_info import ForwardBatch, ForwardMode
from sglang.srt.model_executor.model_runner import ModelRunner from sglang.srt.model_executor.model_runner import ModelRunner
from sglang.srt.runtime_context import publish
from sglang.srt.sampling.sampling_params import SamplingParams from sglang.srt.sampling.sampling_params import SamplingParams
from sglang.srt.server_args import PortArgs, ServerArgs from sglang.srt.server_args import PortArgs, ServerArgs
from sglang.srt.speculative.spec_info import SpeculativeAlgorithm from sglang.srt.speculative.spec_info import SpeculativeAlgorithm
@@ -52,6 +53,8 @@ class TestForwardSplitPrefill(CustomTestCase):
cls.port_args = PortArgs.init_new(cls.server_args) cls.port_args = PortArgs.init_new(cls.server_args)
publish(cls.server_args, role="scheduler")
# Load model and tokenizer # Load model and tokenizer
cls.model_config = ModelConfig.from_server_args(cls.server_args) cls.model_config = ModelConfig.from_server_args(cls.server_args)
cls.model_runner = ModelRunner( cls.model_runner = ModelRunner(
@@ -15,6 +15,7 @@ from unittest.mock import Mock, patch
from sglang.srt.managers.io_struct import GenerateReqInput from sglang.srt.managers.io_struct import GenerateReqInput
from sglang.srt.managers.tokenizer_manager import TokenizerManager from sglang.srt.managers.tokenizer_manager import TokenizerManager
from sglang.srt.runtime_context import publish
from sglang.srt.server_args import PortArgs, ServerArgs from sglang.srt.server_args import PortArgs, ServerArgs
from sglang.test.test_utils import DEFAULT_SMALL_MODEL_NAME_FOR_TEST from sglang.test.test_utils import DEFAULT_SMALL_MODEL_NAME_FOR_TEST
@@ -39,6 +40,7 @@ class TestTokenizerBatchEncode(unittest.TestCase):
): ):
mock_tokenizer.return_value = Mock(vocab_size=32000) mock_tokenizer.return_value = Mock(vocab_size=32000)
publish(self.server_args, role="tokenizer")
self.tokenizer_manager = TokenizerManager(self.server_args, self.port_args) self.tokenizer_manager = TokenizerManager(self.server_args, self.port_args)
def test_batch_encode_enabled(self): def test_batch_encode_enabled(self):
+5
View File
@@ -24,6 +24,7 @@ from sglang.srt.managers.tokenizer_manager import (
TokenizerManager, TokenizerManager,
) )
from sglang.srt.observability.req_time_stats import APIServerReqTimeStats from sglang.srt.observability.req_time_stats import APIServerReqTimeStats
from sglang.srt.runtime_context import publish
from sglang.srt.server_args import PortArgs, ServerArgs from sglang.srt.server_args import PortArgs, ServerArgs
from sglang.test.test_utils import DEFAULT_SMALL_MODEL_NAME_FOR_TEST from sglang.test.test_utils import DEFAULT_SMALL_MODEL_NAME_FOR_TEST
@@ -45,6 +46,7 @@ class TestInputFormatDetection(unittest.TestCase):
) as mock_tokenizer, ) as mock_tokenizer,
): ):
mock_tokenizer.return_value = Mock(vocab_size=32000) mock_tokenizer.return_value = Mock(vocab_size=32000)
publish(self.server_args, role="tokenizer")
self.tokenizer_manager = TokenizerManager(self.server_args, self.port_args) self.tokenizer_manager = TokenizerManager(self.server_args, self.port_args)
def test_detect_single_string(self): def test_detect_single_string(self):
@@ -143,6 +145,7 @@ class TestTokenizerInputPreparation(unittest.TestCase):
) as mock_tokenizer, ) as mock_tokenizer,
): ):
mock_tokenizer.return_value = Mock(vocab_size=32000) mock_tokenizer.return_value = Mock(vocab_size=32000)
publish(self.server_args, role="tokenizer")
self.tokenizer_manager = TokenizerManager(self.server_args, self.port_args) self.tokenizer_manager = TokenizerManager(self.server_args, self.port_args)
def test_prepare_single_string_input(self): def test_prepare_single_string_input(self):
@@ -203,6 +206,7 @@ class TestTokenizerResultExtraction(unittest.TestCase):
) as mock_tokenizer, ) as mock_tokenizer,
): ):
mock_tokenizer.return_value = Mock(vocab_size=32000) mock_tokenizer.return_value = Mock(vocab_size=32000)
publish(self.server_args, role="tokenizer")
self.tokenizer_manager = TokenizerManager(self.server_args, self.port_args) self.tokenizer_manager = TokenizerManager(self.server_args, self.port_args)
def test_extract_single_string_results(self): def test_extract_single_string_results(self):
@@ -327,6 +331,7 @@ class TestTokenizerManagerIntegration(unittest.TestCase):
) as mock_tokenizer, ) as mock_tokenizer,
): ):
mock_tokenizer.return_value = Mock(vocab_size=32000) mock_tokenizer.return_value = Mock(vocab_size=32000)
publish(self.server_args, role="tokenizer")
self.tokenizer_manager = TokenizerManager(self.server_args, self.port_args) self.tokenizer_manager = TokenizerManager(self.server_args, self.port_args)
def test_full_workflow_single_string(self): def test_full_workflow_single_string(self):
+7 -4
View File
@@ -20,6 +20,7 @@ from sglang.srt.managers.schedule_batch import (
from sglang.srt.model_executor.model_runner import ModelRunner from sglang.srt.model_executor.model_runner import ModelRunner
from sglang.srt.multimodal.processors.base_processor import BaseMultimodalProcessor from sglang.srt.multimodal.processors.base_processor import BaseMultimodalProcessor
from sglang.srt.parser.conversation import generate_chat_conv from sglang.srt.parser.conversation import generate_chat_conv
from sglang.srt.runtime_context import publish
from sglang.srt.server_args import ServerArgs from sglang.srt.server_args import ServerArgs
from sglang.test.test_utils import download_image_with_retry from sglang.test.test_utils import download_image_with_retry
@@ -141,16 +142,18 @@ class VisionLLMLogitsBase(unittest.IsolatedAsyncioTestCase):
return inputs return inputs
def get_sglang_model(self): def get_sglang_model(self):
server_args = ServerArgs(
model_path=self.model_path,
disable_cuda_graph=True,
)
publish(server_args, role="scheduler")
self.model_runner = ModelRunner( self.model_runner = ModelRunner(
model_config=ModelConfig(self.model_path, model_override_args="{}"), model_config=ModelConfig(self.model_path, model_override_args="{}"),
mem_fraction_static=0.8, mem_fraction_static=0.8,
gpu_id=0, gpu_id=0,
ps=ParallelState.trivial(), ps=ParallelState.trivial(),
nccl_port=12435, nccl_port=12435,
server_args=ServerArgs( server_args=server_args,
model_path=self.model_path,
disable_cuda_graph=True,
),
) )
return self.model_runner.model return self.model_runner.model
@@ -292,7 +292,7 @@ class TestServerInfoControlPlaneUpdates(CustomTestCase):
tokenizer_manager.record_config_updates("test", weight_version="v2") tokenizer_manager.record_config_updates("test", weight_version="v2")
self.assertEqual(tokenizer_manager.config_value("weight_version"), "v2") self.assertEqual(tokenizer_manager.config_value("weight_version"), "v2")
overlaid = tokenizer_manager.resolved_config_dict( overlaid = tokenizer_manager.resolved_config_dict(
dataclasses.asdict(server_args) server_args.resolved_dict()
) )
self.assertEqual(overlaid["weight_version"], "v2") self.assertEqual(overlaid["weight_version"], "v2")
finally: finally:
@@ -313,9 +313,9 @@ class TestServerInfoExistingFieldsPreserved(CustomTestCase):
""" """
def test_every_server_args_field_appears_in_response(self): def test_every_server_args_field_appears_in_response(self):
# `dataclasses.asdict(server_args)` is spread into the response; # `server_args.resolved_dict()` is spread into the response; asserting
# asserting every dataclass field surfaces is the strongest # every dataclass field surfaces is the strongest backward-compat
# backward-compat guarantee that's still implementation-agnostic. # guarantee that's still implementation-agnostic.
args = ServerArgs(model_path="dummy") args = ServerArgs(model_path="dummy")
info = _call_server_info_with(args) info = _call_server_info_with(args)
@@ -376,6 +376,32 @@ class TestResolutionDeclarations(CustomTestCase):
+ "\n ".join(differences), + "\n ".join(differences),
) )
def test_the_whole_object_readback_carries_only_fields(self):
"""`/server_info` and its gRPC and in-process twins report
`ServerArgs.resolved_dict()`.
The dump is exactly the field names, carrying the resolution result
for each. It holds none of the resolution bookkeeping (`_raw_input`, the
declaration stash, the finished flag) and no `ModelConfig` memo: none of
that is configuration, and all of it would cross IPC with the
readback.
"""
server_args = self._resolve({"tp_size": 2})
dump = server_args.resolved_dict()
self.assertEqual(
sorted(dump),
sorted(field.name for field in dataclasses.fields(server_args)),
"the readback dump is no longer exactly the fields",
)
leaked = sorted(
name
for name in vars(server_args)
if name not in dump and not name.startswith("__")
)
self.assertNotEqual(
leaked, [], "nothing to leak any more -- this check is now vacuous"
)
def test_every_published_leaf_is_what_resolution_decided(self): def test_every_published_leaf_is_what_resolution_decided(self):
"""One hop further than the check above: the leaf a reader reads. """One hop further than the check above: the leaf a reader reads.
@@ -78,15 +78,22 @@ _KNOWN_ENTRIES = frozenset(
"run_data_parallel_controller_process", "run_data_parallel_controller_process",
), ),
("srt/ray/scheduler_actor.py", "__init__"), ("srt/ray/scheduler_actor.py", "__init__"),
("srt/disaggregation/encoder/server.py", "__init__"),
("srt/disaggregation/encoder/http_server.py", "launch_server"), ("srt/disaggregation/encoder/http_server.py", "launch_server"),
("srt/managers/tokenizer_manager.py", "__init__"),
("srt/entrypoints/engine.py", "_launch_subprocesses"), ("srt/entrypoints/engine.py", "_launch_subprocesses"),
( (
"srt/elastic_ep/expert_backup_manager.py", "srt/elastic_ep/expert_backup_manager.py",
"run_expert_backup_manager_process", "run_expert_backup_manager_process",
), ),
("srt/weight_cache/daemon.py", "load"), ("srt/weight_cache/daemon.py", "load"),
# The multi-tokenizer worker, the benchmark work functions (run
# inline or spawned per rank), and the encoder's gRPC / spawned-TP /
# spawned-DP entries.
("srt/entrypoints/http_server.py", "init_multi_tokenizer"),
("benchmark/one_batch.py", "latency_test"),
("benchmark/one_batch.py", "correctness_test"),
("srt/disaggregation/encoder/grpc_server.py", "serve_grpc_encoder"),
("srt/disaggregation/encoder/server.py", "launch_encoder"),
("srt/disaggregation/encoder/runtime.py", "launch_dp_worker"),
} }
) )
+39 -53
View File
@@ -19,7 +19,7 @@ from sglang.srt.runtime_context import (
ParallelContext, ParallelContext,
RuntimeContext, RuntimeContext,
_FlagGroupBase, _FlagGroupBase,
ensure_published, assert_published,
get_context, get_context,
get_exec, get_exec,
get_flags, get_flags,
@@ -255,35 +255,30 @@ class TestServerArgsOwnership(_IsolatedServerArgs):
get_server_args() get_server_args()
class TestEnsurePublished(_IsolatedServerArgs): class TestAssertPublished(_IsolatedServerArgs):
"""A defensive publish must not re-project over a live process. """Publishing is the process entry's job; the constructors only check.
Three constructors publish because each can be built with nothing published `ModelRunner`, `TokenizerManager` and `MMEncoder` assert. A publish inside
first: `ModelRunner`, `TokenizerManager`, `MMEncoder`. Inside a process that a process that has already published re-projects the bags, discarding every
already published the same record, publishing again re-projects the bags -- `override()` taken since and the provenance log with it, so a constructor
discarding every `override()` taken since, and the provenance log with it. that finds nothing published fails loud.
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): def _record(self, **fields):
return ServerArgs(model_path="dummy", **fields) return ServerArgs(model_path="dummy", **fields)
def test_a_second_publish_of_the_same_record_keeps_the_overrides(self): def test_the_check_leaves_a_live_process_alone(self):
record = self._record(grammar_backend="xgrammar") record = self._record(grammar_backend="xgrammar")
publish(record, role="scheduler") publish(record, role="scheduler")
get_context().override("grammar.import_fallback", grammar_backend="none") get_context().override("grammar.import_fallback", grammar_backend="none")
ensure_published(record, role="scheduler") assert_published(record, role="scheduler")
self.assertEqual( self.assertEqual(
get_exec().kernel.grammar_backend, get_exec().kernel.grammar_backend,
"none", "none",
"the constructor's publish re-projected the bags, so the import " "the check re-projected the bags, so the import fallback was "
"fallback was discarded and the process reports a backend it is " "discarded and the process reports a backend it is not using",
"not using",
) )
self.assertEqual( self.assertEqual(
len(get_context().overrides_log()), len(get_context().overrides_log()),
@@ -291,45 +286,49 @@ class TestEnsurePublished(_IsolatedServerArgs):
"the provenance of the override went with it", "the provenance of the override went with it",
) )
def test_a_different_record_is_published(self): def test_a_different_record_fails(self):
first = self._record(grammar_backend="xgrammar") first = self._record(grammar_backend="xgrammar")
publish(first, role="scheduler") publish(first, role="scheduler")
second = self._record(grammar_backend="llguidance") second = self._record(grammar_backend="llguidance")
ensure_published(second, role="scheduler") with self.assertRaisesRegex(RuntimeError, "a different record is published"):
assert_published(second, role="scheduler")
self.assertIs(get_server_args(), second) self.assertIs(
self.assertEqual(get_exec().kernel.grammar_backend, "llguidance") get_server_args(),
first,
"the failing check published anyway",
)
def test_an_empty_slot_is_published(self): def test_an_empty_slot_fails(self):
"""The standalone case the defensive publish exists for.""" """An empty slot fails."""
reset_context() reset_context()
record = self._record(grammar_backend="xgrammar") record = self._record(grammar_backend="xgrammar")
ensure_published(record, role="scheduler") with self.assertRaisesRegex(
RuntimeError, "nothing is published in this process"
):
assert_published(record, role="scheduler")
self.assertIs(get_server_args(), record) def test_the_same_record_under_a_different_role_fails(self):
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.""" """The role decides which namespaces this process may read."""
record = self._record() record = self._record()
publish(record, role="tokenizer") publish(record, role="tokenizer")
ensure_published(record, role="scheduler") with self.assertRaisesRegex(RuntimeError, "published under role 'tokenizer'"):
assert_published(record, role="scheduler")
self.assertEqual(publish_role(), "scheduler") self.assertEqual(publish_role(), "tokenizer")
def test_every_constructor_that_publishes_is_classified(self): def test_no_constructor_publishes_outside_the_two_entries(self):
"""A new constructor publish has to say which of the two it is. """Publishing from an `__init__` is an entry's job or a bug.
Publishing in a constructor is right when the constructor *is* the It is right when the constructor *is* the entry -- an `Engine` being
entry -- a spawned worker, the Ray actor that stands in for (re)built, the Ray actor that stands in for `run_scheduler_process`,
`run_scheduler_process`, an `Engine` being (re)built, where resetting where resetting the bags is the point. It is wrong anywhere else,
the bags is the point -- and wrong when the process is already live because the process is already live with a record and re-projecting
with the same record, where it silently drops overrides. The drops its overrides. The census is pinned, so a new constructor publish
difference is not visible in the syntax, so the census is pinned: fails here until it is one of the two.
adding one fails here until it is classified.
Both the publisher set and "which `__init__` reaches one" come from Both the publisher set and "which `__init__` reaches one" come from
`sglang.test.config_publishers`, which derives them from the code -- `sglang.test.config_publishers`, which derives them from the code --
@@ -347,24 +346,11 @@ class TestEnsurePublished(_IsolatedServerArgs):
self.assertEqual( self.assertEqual(
constructor_publishers(srt), 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"), ("entrypoints/engine.py", "Engine", "publish"),
("ray/scheduler_actor.py", "SchedulerActor", "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 " "a constructor publishes and it is not one of the two entries; "
"it is; an entry uses publish(), one that may run inside a live " "publish at the process entry and let the constructor assert",
"process with the same record uses ensure_published()",
) )