config: one control-plane log for the process (#35028)
This commit is contained in:
@@ -54,6 +54,32 @@ resolved configuration lives in the namespace bags.**
|
|||||||
There is **no write-through** to the `ServerArgs` instance — it stays pristine.
|
There is **no write-through** to the `ServerArgs` instance — it stays pristine.
|
||||||
There is no in-place mutation entry on the instance at all: it is read-only after
|
There is no in-place mutation entry on the instance at all: it is read-only after
|
||||||
resolution.
|
resolution.
|
||||||
|
- **Reading a leaf when the caller holds the field *name*** (a readback endpoint,
|
||||||
|
a control-plane handler): `get_context().config_leaf(name)` — the read side of
|
||||||
|
`override`. It resolves the flat name through the same `NS` map the write side
|
||||||
|
uses and raises on a name that is not a config leaf. Code that knows its field
|
||||||
|
when it is written reads the bag leaf directly; `config_leaf` is for
|
||||||
|
name-driven code, not a way around the seed ratchet.
|
||||||
|
- **Post-startup control-plane changes** — a weight update, a HiCache mirror
|
||||||
|
attach, a parser resolved from the chat template — go through
|
||||||
|
`TokenizerManager.record_config_updates(source, **fields)`, a named wrapper
|
||||||
|
over `get_context().override`. One process keeps one log: the request dumps
|
||||||
|
ship `get_context().overrides_log()`, and `config_value(name)` /
|
||||||
|
`resolved_config_dict(base)` answer from the bags. The exposure ratchet
|
||||||
|
resolves the wrapper, so a field recorded through it joins the post-publish
|
||||||
|
override surface exactly like a direct `override` and needs the same ordering
|
||||||
|
judgment against any supplied-instance read of it
|
||||||
|
(`test_supplied_instance_exposure_ratchet.py`).
|
||||||
|
- **`model_path` and `served_model_name` are answered off the manager.** Both are
|
||||||
|
`NS` leaves and `override` accepts them, but the tokenizer-side weight reload
|
||||||
|
records only `load_format` and writes the two path fields as `TokenizerManager`
|
||||||
|
attributes (`_MANAGER_OWNED_FIELDS`); `config_value` and `resolved_config_dict`
|
||||||
|
overlay them on top of the bags. Bags do not cross a process boundary (above),
|
||||||
|
so recording those two in the tokenizer process would leave every other
|
||||||
|
process's bag on the old path while the log claimed a process-wide change. The
|
||||||
|
scheduler rewrites its own copy where the reload happens —
|
||||||
|
`ModelRunner.update_model_fields` overrides `model_path` / `load_format` for
|
||||||
|
the target runner.
|
||||||
- **Late launcher-stage resolution (pre-publish)**: a few rules cannot run inside
|
- **Late launcher-stage resolution (pre-publish)**: a few rules cannot run inside
|
||||||
`__post_init__` — LoRA normalization, and the auto-parser detection that needs a
|
`__post_init__` — LoRA normalization, and the auto-parser detection that needs a
|
||||||
tokenizer/chat-template load. They are resolution, not mutation, and they write
|
tokenizer/chat-template load. They are resolution, not mutation, and they write
|
||||||
@@ -143,6 +169,51 @@ bag to override at all.
|
|||||||
named `server_args`), and a factory whose contract is "build X from the record
|
named `server_args`), and a factory whose contract is "build X from the record
|
||||||
you are handed" (`create_kt_config_from_server_args`, `DllmConfig.from_server_args`).
|
you are handed" (`create_kt_config_from_server_args`, `DllmConfig.from_server_args`).
|
||||||
|
|
||||||
|
### Four ways a config sweep breaks something no test runs
|
||||||
|
|
||||||
|
Each of these shipped in a review round and cost a real defect; each now has a
|
||||||
|
guard, named here so the next sweep checks the same four things by hand first.
|
||||||
|
|
||||||
|
1. **The other implementations of an interface.** Dropping a parameter means
|
||||||
|
auditing implementers, not just callers: `CustomSpecAlgo` is the plugin
|
||||||
|
base for speculative algorithms, and the dispatch calls it with the
|
||||||
|
built-in's argument list. Nothing in the tree implements it, so only a
|
||||||
|
plugin user hits the `TypeError`.
|
||||||
|
Guard: `test_plugin_hook_signatures.py`.
|
||||||
|
2. **Publish order inside a process entry, not per file.** A file containing a
|
||||||
|
`publish` says nothing about whether a given read runs before it. Spawned
|
||||||
|
workers (`MMEncoder` for encoder DP/TP, the Ray scheduler actor) start with
|
||||||
|
an empty context, so a bag read above the publish raises only there.
|
||||||
|
Guard: `test_publish_precedes_bag_reads.py`.
|
||||||
|
3. **The role namespace a process publishes under.** `ROLE_NAMESPACE_SETS`
|
||||||
|
narrows what each role may read; the DP controller is audited for `exec`
|
||||||
|
alone. A helper that reaches for another namespace passes every default-mode
|
||||||
|
test and aborts startup under `SGLANG_ROLE_NAMESPACES=enforce`. Prefer
|
||||||
|
answering from the caller's own namespaces over widening the set.
|
||||||
|
4. **Sibling surfaces of a readback.** Changing what one entry point reports
|
||||||
|
means enumerating the others: HTTP, gRPC and in-process `Engine` each have
|
||||||
|
their own server-info and model-info, and each passes its own tests while
|
||||||
|
its users lose the field.
|
||||||
|
Guard: `test_effective_state_surfaces.py`.
|
||||||
|
|
||||||
|
A fifth, from the same rounds: the accessor **name itself**. Called as an
|
||||||
|
object member (`manager.get_disagg()`), or shadowed by a same-named import
|
||||||
|
(`from model_loader import get_model` next to the model bag, where the later
|
||||||
|
import silently wins and the loader call gets a zero-argument bag), it imports
|
||||||
|
fine and fails only when that path runs. The invariant is one line: the name
|
||||||
|
means the process-wide bag, takes no arguments, and is bound once per module.
|
||||||
|
`ruff --select F811` catches the import collision; the member-call shapes are an
|
||||||
|
`AttributeError` at call time only because `RuntimeContext` has no bag-named
|
||||||
|
member and no `__getattr__` -- a delegating `__getattr__` would make them
|
||||||
|
silent, and that is when this needs a guard again rather than a rule.
|
||||||
|
|
||||||
|
Write these guards over a **derived** set, never a hand-kept list: an entry
|
||||||
|
naming a function that no longer exists, or a field list missing the one field
|
||||||
|
nobody migrated, passes green forever. Both happened here -- a `_ENTRY_POINTS`
|
||||||
|
row for a method the Ray actor does not have, and an effective-field set
|
||||||
|
without `load_format` -- and both were invisible because the assertion had
|
||||||
|
slack (`>= len(...) - 1`) or compared key names instead of value sources.
|
||||||
|
|
||||||
### `get_parallel()`: config leaves vs live topology
|
### `get_parallel()`: config leaves vs live topology
|
||||||
|
|
||||||
Config leaves (`nccl_port`, `enable_dp_attention`, `dp_size`, `ep_size`,
|
Config leaves (`nccl_port`, `enable_dp_attention`, `dp_size`, `ep_size`,
|
||||||
@@ -182,6 +253,9 @@ this).
|
|||||||
parallel bag's own leaf — are what see post-publish overrides. Only the
|
parallel bag's own leaf — are what see post-publish overrides. Only the
|
||||||
instance-derived accessors (the ones with no leaf to read) answer from the
|
instance-derived accessors (the ones with no leaf to read) answer from the
|
||||||
startup record and therefore do not.
|
startup record and therefore do not.
|
||||||
|
- **a leaf the caller names at runtime** (a readback reporting a list of fields)
|
||||||
|
→ `get_context().config_leaf(name)`; it resolves the name through `NS` and
|
||||||
|
raises on a non-leaf. A call site that knows its field reads the bag leaf.
|
||||||
- **the live topology** → `get_parallel()`.
|
- **the live topology** → `get_parallel()`.
|
||||||
- **a value derived from published leaves** → an accessor in `runtime_context` that
|
- **a value derived from published leaves** → an accessor in `runtime_context` that
|
||||||
derives it *from the bags*: `mamba_extra_buffer_enabled()` /
|
derives it *from the bags*: `mamba_extra_buffer_enabled()` /
|
||||||
|
|||||||
@@ -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.observability.trace import SpanAttributes, extract_trace_headers
|
||||||
from sglang.srt.runtime_context import (
|
from sglang.srt.runtime_context import (
|
||||||
|
get_context,
|
||||||
get_device,
|
get_device,
|
||||||
get_disagg,
|
get_disagg,
|
||||||
get_exec,
|
get_exec,
|
||||||
@@ -379,8 +380,6 @@ class InputFormat(Enum):
|
|||||||
CROSS_ENCODER_PAIRS = 3 # Cross-encoder pairs like [["query", "document"]]
|
CROSS_ENCODER_PAIRS = 3 # Cross-encoder pairs like [["query", "document"]]
|
||||||
|
|
||||||
|
|
||||||
_SERVER_ARGS_FIELDS = frozenset(f.name for f in dataclasses.fields(ServerArgs))
|
|
||||||
|
|
||||||
_MANAGER_OWNED_FIELDS = ("model_path", "served_model_name")
|
_MANAGER_OWNED_FIELDS = ("model_path", "served_model_name")
|
||||||
|
|
||||||
|
|
||||||
@@ -410,7 +409,6 @@ class TokenizerManager(TokenizerControlMixin, TokenizerManagerScoreMixin):
|
|||||||
# the in-process path re-projects the object the launcher published.
|
# the in-process path re-projects the object the launcher published.
|
||||||
set_global_server_args_for_tokenizer(server_args)
|
set_global_server_args_for_tokenizer(server_args)
|
||||||
self.startup_time: Optional[Dict[str, Any]] = None
|
self.startup_time: Optional[Dict[str, Any]] = None
|
||||||
self._config_updates: List[Tuple[str, Dict[str, Any]]] = []
|
|
||||||
self.elastic_worker_count = get_parallel().dp_size
|
self.elastic_worker_count = get_parallel().dp_size
|
||||||
self.elastic_pending_ep_size = None
|
self.elastic_pending_ep_size = None
|
||||||
self.elastic_scale_phase = "idle"
|
self.elastic_scale_phase = "idle"
|
||||||
@@ -2065,31 +2063,19 @@ class TokenizerManager(TokenizerControlMixin, TokenizerManagerScoreMixin):
|
|||||||
return success, message, num_paused_requests
|
return success, message, num_paused_requests
|
||||||
|
|
||||||
def record_config_updates(self, source: str, **fields) -> None:
|
def record_config_updates(self, source: str, **fields) -> None:
|
||||||
"""Record a control-plane config change for this engine.
|
"""Record a control-plane config change: a weight update, a parser
|
||||||
|
resolved from the chat template, a HiCache mirror attach.
|
||||||
|
|
||||||
These are post-startup facts the config bags do not model (weight
|
These land in the config bags like every other post-publish change, so
|
||||||
version, model path, the tokenizer's HiCache mirror); the readback
|
one log carries the provenance for the whole process.
|
||||||
endpoints overlay them onto the startup config. The process-global
|
|
||||||
sibling is ``RuntimeContext.override`` / ``resolved_server_args_dict``,
|
|
||||||
which writes the config bags.
|
|
||||||
"""
|
"""
|
||||||
unknown = sorted(f for f in fields if f not in _SERVER_ARGS_FIELDS)
|
get_context().override(source, **fields)
|
||||||
if unknown:
|
|
||||||
raise ValueError(
|
|
||||||
f"{unknown} are not ServerArgs fields; the readback endpoints "
|
|
||||||
"overlay these onto a serialized ServerArgs, so an unknown key "
|
|
||||||
"would surface as a phantom config entry."
|
|
||||||
)
|
|
||||||
self._config_updates.append((source, dict(fields)))
|
|
||||||
|
|
||||||
def config_value(self, name: str):
|
def config_value(self, name: str):
|
||||||
"""The value in effect for one config field, control-plane updates first."""
|
"""The value in effect for one config field."""
|
||||||
if name in _MANAGER_OWNED_FIELDS:
|
if name in _MANAGER_OWNED_FIELDS:
|
||||||
return getattr(self, name)
|
return getattr(self, name)
|
||||||
for _source, fields in reversed(self._config_updates):
|
return get_context().config_leaf(name)
|
||||||
if name in fields:
|
|
||||||
return fields[name]
|
|
||||||
return getattr(self.server_args, name)
|
|
||||||
|
|
||||||
def _dump_config_snapshot(self) -> Optional[Dict[str, Any]]:
|
def _dump_config_snapshot(self) -> Optional[Dict[str, Any]]:
|
||||||
"""The config in effect, or None when it cannot be serialized.
|
"""The config in effect, or None when it cannot be serialized.
|
||||||
@@ -2104,18 +2090,18 @@ class TokenizerManager(TokenizerControlMixin, TokenizerManagerScoreMixin):
|
|||||||
return None
|
return None
|
||||||
|
|
||||||
def resolved_config_dict(self, base: Dict[str, Any]) -> Dict[str, Any]:
|
def resolved_config_dict(self, base: Dict[str, Any]) -> Dict[str, Any]:
|
||||||
"""``base`` (a serialized ``ServerArgs``) with the control-plane updates on top."""
|
"""``base`` (a serialized ``ServerArgs``) with the control-plane changes on top."""
|
||||||
resolved = dict(base)
|
resolved = get_context().resolved_server_args_dict(base)
|
||||||
for _source, fields in self._config_updates:
|
|
||||||
resolved.update(fields)
|
|
||||||
for name in _MANAGER_OWNED_FIELDS:
|
for name in _MANAGER_OWNED_FIELDS:
|
||||||
resolved[name] = getattr(self, name)
|
resolved[name] = getattr(self, name)
|
||||||
return resolved
|
return resolved
|
||||||
|
|
||||||
def _update_model_path_info(self, model_path: str, load_format: str):
|
def _update_model_path_info(self, model_path: str, load_format: str):
|
||||||
|
# These two stay on the manager: the readback reads them from here,
|
||||||
|
# and a bag write would not reach the other processes anyway.
|
||||||
self.served_model_name = model_path
|
self.served_model_name = model_path
|
||||||
self.record_config_updates("tokenizer.update_weights", load_format=load_format)
|
|
||||||
self.model_path = model_path
|
self.model_path = model_path
|
||||||
|
self.record_config_updates("tokenizer.update_weights", load_format=load_format)
|
||||||
|
|
||||||
async def _wait_for_model_update_from_disk(
|
async def _wait_for_model_update_from_disk(
|
||||||
self, obj: UpdateWeightFromDiskReqInput
|
self, obj: UpdateWeightFromDiskReqInput
|
||||||
@@ -2986,7 +2972,7 @@ class TokenizerManager(TokenizerControlMixin, TokenizerManagerScoreMixin):
|
|||||||
logger.info(log_message)
|
logger.info(log_message)
|
||||||
to_dump_with_server_args = {
|
to_dump_with_server_args = {
|
||||||
"server_args": self.server_args,
|
"server_args": self.server_args,
|
||||||
"config_updates": list(self._config_updates),
|
"config_updates": get_context().overrides_log(),
|
||||||
"resolved_config": self._dump_config_snapshot(),
|
"resolved_config": self._dump_config_snapshot(),
|
||||||
"requests": data_list.copy(),
|
"requests": data_list.copy(),
|
||||||
}
|
}
|
||||||
@@ -3071,7 +3057,7 @@ class TokenizerManager(TokenizerControlMixin, TokenizerManagerScoreMixin):
|
|||||||
# Write the data to the file
|
# Write the data to the file
|
||||||
data_to_dump_with_server_args = {
|
data_to_dump_with_server_args = {
|
||||||
"server_args": self.server_args,
|
"server_args": self.server_args,
|
||||||
"config_updates": list(self._config_updates),
|
"config_updates": get_context().overrides_log(),
|
||||||
"resolved_config": self._dump_config_snapshot(),
|
"resolved_config": self._dump_config_snapshot(),
|
||||||
"requests": data_to_dump,
|
"requests": data_to_dump,
|
||||||
"launch_command": " ".join(sys.argv),
|
"launch_command": " ".join(sys.argv),
|
||||||
|
|||||||
@@ -903,6 +903,29 @@ class RuntimeContext:
|
|||||||
bag._set(name, value)
|
bag._set(name, value)
|
||||||
self._overrides_log.append((source, dict(fields)))
|
self._overrides_log.append((source, dict(fields)))
|
||||||
|
|
||||||
|
def config_leaf(self, name: str):
|
||||||
|
"""One resolved config leaf by field name — the read side of ``override``.
|
||||||
|
|
||||||
|
Callers that hold a field name rather than a namespace (a readback
|
||||||
|
endpoint, a control-plane handler) would otherwise have to know which
|
||||||
|
bag it lives in.
|
||||||
|
"""
|
||||||
|
bags = self._config_bags
|
||||||
|
if bags is None:
|
||||||
|
raise ValueError("config not published; cannot read a config leaf")
|
||||||
|
from sglang.srt.arg_groups.arg_utils import namespace_of
|
||||||
|
|
||||||
|
path = namespace_of(type(self._server_args)).get(name)
|
||||||
|
if path is None:
|
||||||
|
raise ValueError(f"{name!r} is not a config leaf (no NS namespace)")
|
||||||
|
parts = path.split(".")
|
||||||
|
bag = self.config_bag(parts[0])
|
||||||
|
for seg in parts[1:]:
|
||||||
|
bag = object.__getattribute__(bag, "_subs").get(seg)
|
||||||
|
if bag is None:
|
||||||
|
raise ValueError(f"subgroup {seg!r} missing under {path!r}")
|
||||||
|
return getattr(bag, name)
|
||||||
|
|
||||||
def overrides_log(self) -> list:
|
def overrides_log(self) -> list:
|
||||||
"""Provenance of post-publish ``override`` calls: ``[(source, {field: value})]``.
|
"""Provenance of post-publish ``override`` calls: ``[(source, {field: value})]``.
|
||||||
|
|
||||||
@@ -925,12 +948,14 @@ class RuntimeContext:
|
|||||||
``ServerArgs`` field names, so overlaying them onto the top level of
|
``ServerArgs`` field names, so overlaying them onto the top level of
|
||||||
either base is exact.
|
either base is exact.
|
||||||
|
|
||||||
This covers the process-global bags only. Control-plane facts the bags
|
The log is per process: it carries what *this* process overrode. A
|
||||||
do not model (weight version, model path, the tokenizer's HiCache
|
weight reload records ``model_path`` and ``load_format`` from the
|
||||||
mirror) live on the tokenizer manager, and
|
scheduler process (``ModelRunner.update_model_fields``); the tokenizer
|
||||||
``TokenizerManager.resolved_config_dict`` overlays those for the
|
process records only ``load_format`` and keeps ``model_path`` /
|
||||||
top-level ``/server_info`` body. The two are separate logs, not one
|
``served_model_name`` as ``TokenizerManager`` attributes, which
|
||||||
merged dict.
|
``TokenizerManager.resolved_config_dict`` overlays on top of this dump.
|
||||||
|
The top-level ``/server_info`` fields are the startup record, not this
|
||||||
|
dump.
|
||||||
"""
|
"""
|
||||||
d = dict(vars(self.server_args)) if base is None else dict(base)
|
d = dict(vars(self.server_args)) if base is None else dict(base)
|
||||||
for _source, fields in self._overrides_log:
|
for _source, fields in self._overrides_log:
|
||||||
|
|||||||
@@ -75,7 +75,9 @@ class _MockTokenizerManager:
|
|||||||
self.model_path = self.server_args.model_path
|
self.model_path = self.server_args.model_path
|
||||||
# The manager tracks the served name itself; a weight update rewrites it.
|
# The manager tracks the served name itself; a weight update rewrites it.
|
||||||
self.served_model_name = "test-model"
|
self.served_model_name = "test-model"
|
||||||
self._config_updates = []
|
# Stands in for the context's resolved leaves: an override replaces the
|
||||||
|
# field's one live value, the seed stays on server_args.
|
||||||
|
self._config_overrides = {}
|
||||||
|
|
||||||
# Mock hf_config for _resolve_chat_encoding_spec check
|
# Mock hf_config for _resolve_chat_encoding_spec check
|
||||||
mock_hf_config = Mock()
|
mock_hf_config = Mock()
|
||||||
@@ -114,10 +116,9 @@ class _MockTokenizerManager:
|
|||||||
self.request_logger = Mock(log_requests=False, log_requests_level=0)
|
self.request_logger = Mock(log_requests=False, log_requests_level=0)
|
||||||
|
|
||||||
def config_value(self, name: str):
|
def config_value(self, name: str):
|
||||||
"""The manager's overlay accessor: no control-plane update recorded."""
|
"""The value in effect for one config field."""
|
||||||
for _source, fields in reversed(self._config_updates):
|
if name in self._config_overrides:
|
||||||
if name in fields:
|
return self._config_overrides[name]
|
||||||
return fields[name]
|
|
||||||
return getattr(self.server_args, name)
|
return getattr(self.server_args, name)
|
||||||
|
|
||||||
|
|
||||||
@@ -160,15 +161,12 @@ class ServingChatTestCase(unittest.TestCase):
|
|||||||
self.fastapi_request.headers = {}
|
self.fastapi_request.headers = {}
|
||||||
|
|
||||||
def test_parsers_follow_the_control_plane_overlay(self):
|
def test_parsers_follow_the_control_plane_overlay(self):
|
||||||
"""Template detection records the parsers on the manager, not on its
|
"""Template detection records the parsers through `override`, so they
|
||||||
ServerArgs — the instance keeps what the launcher passed."""
|
answer from the bags; `ServerArgs` keeps the launcher's seed."""
|
||||||
self.tm.server_args.tool_call_parser = "auto"
|
self.tm.server_args.tool_call_parser = "auto"
|
||||||
self.tm.server_args.reasoning_parser = "auto"
|
self.tm.server_args.reasoning_parser = "auto"
|
||||||
self.tm._config_updates.append(
|
self.tm._config_overrides.update(
|
||||||
(
|
{"tool_call_parser": "qwen25", "reasoning_parser": None}
|
||||||
"template-detection",
|
|
||||||
{"tool_call_parser": "qwen25", "reasoning_parser": None},
|
|
||||||
)
|
|
||||||
)
|
)
|
||||||
|
|
||||||
chat = OpenAIServingChat(self.tm, self.template_manager)
|
chat = OpenAIServingChat(self.tm, self.template_manager)
|
||||||
@@ -180,9 +178,7 @@ class ServingChatTestCase(unittest.TestCase):
|
|||||||
def test_the_xgrammar_gate_follows_the_overlay(self):
|
def test_the_xgrammar_gate_follows_the_overlay(self):
|
||||||
"""A detected `reasoning_parser` must gate xgrammar, not the seed's "auto"."""
|
"""A detected `reasoning_parser` must gate xgrammar, not the seed's "auto"."""
|
||||||
self.tm.server_args.reasoning_parser = "auto"
|
self.tm.server_args.reasoning_parser = "auto"
|
||||||
self.tm._config_updates.append(
|
self.tm._config_overrides["reasoning_parser"] = "qwen3"
|
||||||
("template-detection", {"reasoning_parser": "qwen3"})
|
|
||||||
)
|
|
||||||
chat = OpenAIServingChat(self.tm, self.template_manager)
|
chat = OpenAIServingChat(self.tm, self.template_manager)
|
||||||
self.assertEqual(chat.reasoning_parser, "qwen3")
|
self.assertEqual(chat.reasoning_parser, "qwen3")
|
||||||
# the gate reads the same value the parser was built from
|
# the gate reads the same value the parser was built from
|
||||||
|
|||||||
@@ -54,7 +54,9 @@ class MockTokenizerManager:
|
|||||||
tool_call_parser=None,
|
tool_call_parser=None,
|
||||||
incremental_streaming_output=False,
|
incremental_streaming_output=False,
|
||||||
)
|
)
|
||||||
self._config_updates = []
|
# Stands in for the context's resolved leaves: an override replaces the
|
||||||
|
# field's one live value, the seed stays on server_args.
|
||||||
|
self._config_overrides = {}
|
||||||
self.tokenizer = Mock()
|
self.tokenizer = Mock()
|
||||||
self.tokenizer.encode.return_value = [1, 2, 3]
|
self.tokenizer.encode.return_value = [1, 2, 3]
|
||||||
self.tokenizer.chat_template = None
|
self.tokenizer.chat_template = None
|
||||||
@@ -64,10 +66,9 @@ class MockTokenizerManager:
|
|||||||
self.create_abort_task = Mock()
|
self.create_abort_task = Mock()
|
||||||
|
|
||||||
def config_value(self, name: str):
|
def config_value(self, name: str):
|
||||||
"""The manager's overlay accessor: no control-plane update recorded."""
|
"""The value in effect for one config field."""
|
||||||
for _source, fields in reversed(self._config_updates):
|
if name in self._config_overrides:
|
||||||
if name in fields:
|
return self._config_overrides[name]
|
||||||
return fields[name]
|
|
||||||
return getattr(self.server_args, name)
|
return getattr(self.server_args, name)
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -43,15 +43,13 @@ def _stub_tokenizer_manager(
|
|||||||
"""A manager carrying the state `/server_info` and its writers read.
|
"""A manager carrying the state `/server_info` and its writers read.
|
||||||
|
|
||||||
`__new__` skips `__init__`, which would open the ZMQ sockets and start
|
`__new__` skips `__init__`, which would open the ZMQ sockets and start
|
||||||
the handle loop; `_config_updates` is the log `record_config_updates`
|
the handle loop.
|
||||||
appends to.
|
|
||||||
"""
|
"""
|
||||||
tokenizer_manager = TokenizerManager.__new__(TokenizerManager)
|
tokenizer_manager = TokenizerManager.__new__(TokenizerManager)
|
||||||
tokenizer_manager.server_args = server_args
|
tokenizer_manager.server_args = server_args
|
||||||
tokenizer_manager.model_path = server_args.model_path
|
tokenizer_manager.model_path = server_args.model_path
|
||||||
tokenizer_manager.served_model_name = server_args.served_model_name
|
tokenizer_manager.served_model_name = server_args.served_model_name
|
||||||
tokenizer_manager.startup_time = None
|
tokenizer_manager.startup_time = None
|
||||||
tokenizer_manager._config_updates = []
|
|
||||||
tokenizer_manager.get_internal_state = get_internal_state
|
tokenizer_manager.get_internal_state = get_internal_state
|
||||||
return tokenizer_manager
|
return tokenizer_manager
|
||||||
|
|
||||||
|
|||||||
@@ -1,9 +1,9 @@
|
|||||||
"""Control-plane config updates stay on the tokenizer manager.
|
"""Control-plane config updates go into the process log, not onto the record.
|
||||||
|
|
||||||
Regression: runtime updates (weight version, model path, HiCache attach) were
|
Regression: runtime updates (weight version, model path, HiCache attach) were
|
||||||
written onto the manager's ServerArgs instance so that the readback endpoints
|
written onto the manager's ServerArgs instance so that the readback endpoints
|
||||||
would show them. The record stays pristine; the updates live in a separate log
|
would show them. The record stays pristine; the update lands in the runtime
|
||||||
that the endpoints overlay on top of it.
|
context, which is where a reader of any field asks for the value in effect.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
import re
|
import re
|
||||||
@@ -12,6 +12,7 @@ from pathlib import Path
|
|||||||
|
|
||||||
import sglang
|
import sglang
|
||||||
from sglang.srt.managers.tokenizer_manager import TokenizerManager
|
from sglang.srt.managers.tokenizer_manager import TokenizerManager
|
||||||
|
from sglang.srt.runtime_context import get_context, publish, reset_context
|
||||||
from sglang.srt.server_args import ServerArgs
|
from sglang.srt.server_args import ServerArgs
|
||||||
from sglang.test.ci.ci_register import register_cpu_ci
|
from sglang.test.ci.ci_register import register_cpu_ci
|
||||||
from sglang.test.test_utils import CustomTestCase
|
from sglang.test.test_utils import CustomTestCase
|
||||||
@@ -19,28 +20,31 @@ from sglang.test.test_utils import CustomTestCase
|
|||||||
register_cpu_ci(est_time=5, suite="base-a-test-cpu")
|
register_cpu_ci(est_time=5, suite="base-a-test-cpu")
|
||||||
|
|
||||||
|
|
||||||
def _manager(**fields):
|
def _manager(case, **fields):
|
||||||
|
"""A manager over a published config: the updates it records go to the bags."""
|
||||||
|
server_args = ServerArgs(model_path="dummy", **fields)
|
||||||
|
publish(server_args, role="tokenizer")
|
||||||
|
case.addCleanup(reset_context)
|
||||||
manager = TokenizerManager.__new__(TokenizerManager)
|
manager = TokenizerManager.__new__(TokenizerManager)
|
||||||
manager.server_args = ServerArgs(model_path="dummy", **fields)
|
manager.server_args = server_args
|
||||||
manager._config_updates = []
|
|
||||||
return manager
|
return manager
|
||||||
|
|
||||||
|
|
||||||
class TestTokenizerConfigUpdates(CustomTestCase):
|
class TestTokenizerConfigUpdates(CustomTestCase):
|
||||||
def test_startup_config_shows_through_until_something_updates_it(self):
|
def test_startup_config_shows_through_until_something_updates_it(self):
|
||||||
manager = _manager(weight_version="v1")
|
manager = _manager(self, weight_version="v1")
|
||||||
self.assertEqual(manager.config_value("weight_version"), "v1")
|
self.assertEqual(manager.config_value("weight_version"), "v1")
|
||||||
|
|
||||||
manager.record_config_updates("test", weight_version="v2")
|
manager.record_config_updates("test", weight_version="v2")
|
||||||
self.assertEqual(manager.config_value("weight_version"), "v2")
|
self.assertEqual(manager.config_value("weight_version"), "v2")
|
||||||
|
|
||||||
def test_the_serverargs_instance_is_not_written(self):
|
def test_the_serverargs_instance_is_not_written(self):
|
||||||
manager = _manager(weight_version="v1")
|
manager = _manager(self, weight_version="v1")
|
||||||
manager.record_config_updates("test", weight_version="v2")
|
manager.record_config_updates("test", weight_version="v2")
|
||||||
self.assertEqual(manager.server_args.weight_version, "v1")
|
self.assertEqual(manager.server_args.weight_version, "v1")
|
||||||
|
|
||||||
def test_the_readback_dict_carries_the_updates(self):
|
def test_the_readback_dict_carries_the_updates(self):
|
||||||
manager = _manager(hicache_storage_backend=None)
|
manager = _manager(self, hicache_storage_backend=None)
|
||||||
manager.record_config_updates(
|
manager.record_config_updates(
|
||||||
"test", hicache_storage_backend="file", hicache_write_policy="write_through"
|
"test", hicache_storage_backend="file", hicache_write_policy="write_through"
|
||||||
)
|
)
|
||||||
@@ -54,7 +58,7 @@ class TestTokenizerConfigUpdates(CustomTestCase):
|
|||||||
self.assertEqual(resolved["model_path"], "dummy")
|
self.assertEqual(resolved["model_path"], "dummy")
|
||||||
|
|
||||||
def test_detach_reports_the_backend_as_gone(self):
|
def test_detach_reports_the_backend_as_gone(self):
|
||||||
manager = _manager(hicache_storage_backend="file")
|
manager = _manager(self, hicache_storage_backend="file")
|
||||||
manager.record_config_updates(
|
manager.record_config_updates(
|
||||||
"test",
|
"test",
|
||||||
hicache_storage_backend=None,
|
hicache_storage_backend=None,
|
||||||
@@ -63,22 +67,27 @@ class TestTokenizerConfigUpdates(CustomTestCase):
|
|||||||
self.assertIsNone(manager.config_value("hicache_storage_backend"))
|
self.assertIsNone(manager.config_value("hicache_storage_backend"))
|
||||||
|
|
||||||
def test_an_unknown_field_is_refused(self):
|
def test_an_unknown_field_is_refused(self):
|
||||||
manager = _manager()
|
manager = _manager(self)
|
||||||
with self.assertRaisesRegex(ValueError, "not ServerArgs fields"):
|
with self.assertRaisesRegex(ValueError, "not a resolved config leaf"):
|
||||||
manager.record_config_updates("test", waight_version="v2")
|
manager.record_config_updates("test", waight_version="v2")
|
||||||
|
|
||||||
|
def test_a_name_that_is_not_a_config_leaf_is_refused(self):
|
||||||
|
manager = _manager(self)
|
||||||
|
with self.assertRaisesRegex(ValueError, "not a config leaf"):
|
||||||
|
manager.config_value("waight_version")
|
||||||
|
|
||||||
def test_the_source_is_kept_for_provenance(self):
|
def test_the_source_is_kept_for_provenance(self):
|
||||||
manager = _manager(weight_version="v1")
|
manager = _manager(self, weight_version="v1")
|
||||||
manager.record_config_updates("http.update_weight_version", weight_version="v2")
|
manager.record_config_updates("http.update_weight_version", weight_version="v2")
|
||||||
self.assertEqual(
|
self.assertEqual(
|
||||||
manager._config_updates,
|
get_context().overrides_log(),
|
||||||
[("http.update_weight_version", {"weight_version": "v2"})],
|
[("http.update_weight_version", {"weight_version": "v2"})],
|
||||||
)
|
)
|
||||||
|
|
||||||
def test_the_dump_snapshot_identifies_the_running_checkpoint(self):
|
def test_the_dump_snapshot_identifies_the_running_checkpoint(self):
|
||||||
import dataclasses
|
import dataclasses
|
||||||
|
|
||||||
manager = _manager(load_format="auto")
|
manager = _manager(self, load_format="auto")
|
||||||
manager.model_path = "at-startup"
|
manager.model_path = "at-startup"
|
||||||
manager.served_model_name = "at-startup"
|
manager.served_model_name = "at-startup"
|
||||||
manager._update_model_path_info("after-reload", "dummy")
|
manager._update_model_path_info("after-reload", "dummy")
|
||||||
@@ -87,6 +96,10 @@ class TestTokenizerConfigUpdates(CustomTestCase):
|
|||||||
self.assertEqual(snapshot["model_path"], "after-reload")
|
self.assertEqual(snapshot["model_path"], "after-reload")
|
||||||
self.assertEqual(snapshot["served_model_name"], "after-reload")
|
self.assertEqual(snapshot["served_model_name"], "after-reload")
|
||||||
self.assertEqual(snapshot["load_format"], "dummy")
|
self.assertEqual(snapshot["load_format"], "dummy")
|
||||||
|
self.assertEqual(
|
||||||
|
get_context().overrides_log(),
|
||||||
|
[("tokenizer.update_weights", {"load_format": "dummy"})],
|
||||||
|
)
|
||||||
self.assertEqual(manager.server_args.model_path, "dummy")
|
self.assertEqual(manager.server_args.model_path, "dummy")
|
||||||
|
|
||||||
def test_an_unsnapshotable_config_does_not_lose_the_dump(self):
|
def test_an_unsnapshotable_config_does_not_lose_the_dump(self):
|
||||||
@@ -94,7 +107,7 @@ class TestTokenizerConfigUpdates(CustomTestCase):
|
|||||||
def __deepcopy__(self, memo):
|
def __deepcopy__(self, memo):
|
||||||
raise RuntimeError("refuses to be copied")
|
raise RuntimeError("refuses to be copied")
|
||||||
|
|
||||||
manager = _manager()
|
manager = _manager(self)
|
||||||
manager.model_path = "dummy"
|
manager.model_path = "dummy"
|
||||||
manager.served_model_name = "dummy"
|
manager.served_model_name = "dummy"
|
||||||
manager.server_args.custom_sigquit_handler = Hostile()
|
manager.server_args.custom_sigquit_handler = Hostile()
|
||||||
@@ -105,7 +118,7 @@ class TestTokenizerConfigUpdates(CustomTestCase):
|
|||||||
import dataclasses
|
import dataclasses
|
||||||
import pickle
|
import pickle
|
||||||
|
|
||||||
manager = _manager()
|
manager = _manager(self)
|
||||||
manager.model_path = "dummy"
|
manager.model_path = "dummy"
|
||||||
manager.served_model_name = "dummy"
|
manager.served_model_name = "dummy"
|
||||||
# What --custom-sigquit-handler leaves on a real ServerArgs.
|
# What --custom-sigquit-handler leaves on a real ServerArgs.
|
||||||
@@ -113,7 +126,7 @@ class TestTokenizerConfigUpdates(CustomTestCase):
|
|||||||
|
|
||||||
payload = {
|
payload = {
|
||||||
"server_args": manager.server_args,
|
"server_args": manager.server_args,
|
||||||
"config_updates": list(manager._config_updates),
|
"config_updates": get_context().overrides_log(),
|
||||||
"resolved_config": manager.resolved_config_dict(
|
"resolved_config": manager.resolved_config_dict(
|
||||||
dataclasses.asdict(manager.server_args)
|
dataclasses.asdict(manager.server_args)
|
||||||
),
|
),
|
||||||
@@ -128,7 +141,7 @@ class TestTokenizerConfigUpdates(CustomTestCase):
|
|||||||
self.assertTrue(pickle.dumps(payload))
|
self.assertTrue(pickle.dumps(payload))
|
||||||
|
|
||||||
def test_the_model_path_readback_follows_the_manager(self):
|
def test_the_model_path_readback_follows_the_manager(self):
|
||||||
manager = _manager()
|
manager = _manager(self)
|
||||||
manager.model_path = "after-update"
|
manager.model_path = "after-update"
|
||||||
manager.served_model_name = "after-update"
|
manager.served_model_name = "after-update"
|
||||||
resolved = manager.resolved_config_dict({"model_path": "at-startup"})
|
resolved = manager.resolved_config_dict({"model_path": "at-startup"})
|
||||||
@@ -200,8 +213,9 @@ class TestControlPlaneFieldsAreNotReadFromTheInstance(CustomTestCase):
|
|||||||
self.assertEqual(
|
self.assertEqual(
|
||||||
stale,
|
stale,
|
||||||
[],
|
[],
|
||||||
"control-plane fields change at runtime and the update lives on the "
|
"control-plane fields change at runtime and the update lands in "
|
||||||
"TokenizerManager; read them with config_value() / "
|
"the process bags (model_path / served_model_name stay manager "
|
||||||
|
"attributes); read them with config_value() / "
|
||||||
"resolved_config_dict() so the readback reflects the change:\n"
|
"resolved_config_dict() so the readback reflects the change:\n"
|
||||||
+ "\n".join(stale),
|
+ "\n".join(stale),
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -367,6 +367,8 @@ _EXPOSED_CUDA_ONLY: frozenset = frozenset()
|
|||||||
# some code overrides post-publish. Each needs an ordering judgment, not a blanket
|
# some code overrides post-publish. Each needs an ordering judgment, not a blanket
|
||||||
# conversion; the list exists so a new one is a decision made when it is written.
|
# conversion; the list exists so a new one is a decision made when it is written.
|
||||||
_OVERRIDDEN_AND_READ = {
|
_OVERRIDDEN_AND_READ = {
|
||||||
|
("entrypoints/engine.py", "reasoning_parser"),
|
||||||
|
("entrypoints/engine.py", "tool_call_parser"),
|
||||||
("configs/model_config.py", "dtype"),
|
("configs/model_config.py", "dtype"),
|
||||||
("configs/model_config.py", "model_path"),
|
("configs/model_config.py", "model_path"),
|
||||||
("disaggregation/decode_kvcache_offload_manager.py", "hicache_storage_backend"),
|
("disaggregation/decode_kvcache_offload_manager.py", "hicache_storage_backend"),
|
||||||
@@ -442,11 +444,53 @@ def _expanded_override_keys(rel, tree, call, kw) -> set:
|
|||||||
):
|
):
|
||||||
return set()
|
return set()
|
||||||
|
|
||||||
|
def loop_variable_values(name: str) -> set:
|
||||||
|
"""The values a `for name, ... in (<literal tuples>)` loop binds.
|
||||||
|
|
||||||
|
A handler that records one field per loop iteration spells the field
|
||||||
|
names in the loop's own literal, so they are still static.
|
||||||
|
"""
|
||||||
|
values = set()
|
||||||
|
for node in ast.walk(tree):
|
||||||
|
if not isinstance(node, ast.For):
|
||||||
|
continue
|
||||||
|
target = node.target
|
||||||
|
names = (
|
||||||
|
[target]
|
||||||
|
if isinstance(target, ast.Name)
|
||||||
|
else list(getattr(target, "elts", []))
|
||||||
|
)
|
||||||
|
if not names or not isinstance(names[0], ast.Name) or names[0].id != name:
|
||||||
|
continue
|
||||||
|
if not (node.lineno <= call.lineno <= (node.end_lineno or node.lineno)):
|
||||||
|
continue
|
||||||
|
for item in getattr(node.iter, "elts", []):
|
||||||
|
first = (
|
||||||
|
item.elts[0] if isinstance(item, ast.Tuple) and item.elts else item
|
||||||
|
)
|
||||||
|
if isinstance(first, ast.Constant) and isinstance(first.value, str):
|
||||||
|
values.add(first.value)
|
||||||
|
return values
|
||||||
|
|
||||||
def dict_keys(node) -> set:
|
def dict_keys(node) -> set:
|
||||||
assert isinstance(node, ast.Dict) and all(
|
assert isinstance(
|
||||||
isinstance(key, ast.Constant) for key in node.keys
|
node, ast.Dict
|
||||||
), f"non-literal dict in override expansion at {rel}:{call.lineno}"
|
), f"non-literal dict in override expansion at {rel}:{call.lineno}"
|
||||||
return {key.value for key in node.keys}
|
keys = set()
|
||||||
|
for key in node.keys:
|
||||||
|
if isinstance(key, ast.Constant):
|
||||||
|
keys.add(key.value)
|
||||||
|
continue
|
||||||
|
assert isinstance(
|
||||||
|
key, ast.Name
|
||||||
|
), f"non-literal dict key in override expansion at {rel}:{call.lineno}"
|
||||||
|
bound = loop_variable_values(key.id)
|
||||||
|
assert bound, (
|
||||||
|
f"dict key {key.id!r} at {rel}:{call.lineno} is not bound by a "
|
||||||
|
"literal loop; extend the resolver"
|
||||||
|
)
|
||||||
|
keys |= bound
|
||||||
|
return keys
|
||||||
|
|
||||||
if isinstance(kw.value, ast.Dict):
|
if isinstance(kw.value, ast.Dict):
|
||||||
return dict_keys(kw.value)
|
return dict_keys(kw.value)
|
||||||
@@ -941,17 +985,29 @@ class TestSuppliedInstanceExposure(CustomTestCase):
|
|||||||
raise AssertionError(f"unparsable module in the census: {rel}")
|
raise AssertionError(f"unparsable module in the census: {rel}")
|
||||||
for node in ast.walk(tree):
|
for node in ast.walk(tree):
|
||||||
if not (
|
if not (
|
||||||
isinstance(node, ast.Call)
|
isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute)
|
||||||
and isinstance(node.func, ast.Attribute)
|
|
||||||
and node.func.attr == "override"
|
|
||||||
):
|
):
|
||||||
continue
|
continue
|
||||||
base = node.func.value
|
base = node.func.value
|
||||||
if (
|
is_override = node.func.attr == "override" and (
|
||||||
isinstance(base, ast.Call)
|
isinstance(base, ast.Call)
|
||||||
and isinstance(base.func, ast.Name)
|
and isinstance(base.func, ast.Name)
|
||||||
and base.func.id == "get_context"
|
and base.func.id == "get_context"
|
||||||
):
|
)
|
||||||
|
# `record_config_updates` is a named wrapper over override, so
|
||||||
|
# its call sites are override sites. Its body forwards **kwargs
|
||||||
|
# and names no field, so skip the forwarding call itself.
|
||||||
|
is_wrapper = node.func.attr == "record_config_updates"
|
||||||
|
if not (is_override or is_wrapper):
|
||||||
|
continue
|
||||||
|
inside_wrapper = any(
|
||||||
|
isinstance(fn, (ast.FunctionDef, ast.AsyncFunctionDef))
|
||||||
|
and fn.name == "record_config_updates"
|
||||||
|
and fn.lineno <= node.lineno <= (fn.end_lineno or fn.lineno)
|
||||||
|
for fn in ast.walk(tree)
|
||||||
|
)
|
||||||
|
if inside_wrapper:
|
||||||
|
continue
|
||||||
for kw in node.keywords:
|
for kw in node.keywords:
|
||||||
if kw.arg == "source":
|
if kw.arg == "source":
|
||||||
# Override metadata, not a config field.
|
# Override metadata, not a config field.
|
||||||
|
|||||||
Reference in New Issue
Block a user