[Config] One writer for the declaration stash; no exception to the write seal (#38752)
This commit is contained in:
@@ -64,7 +64,11 @@ import numpy as np
|
||||
import torch
|
||||
import torch.distributed as dist
|
||||
|
||||
from sglang.srt.arg_groups.overrides import resolution_result, resolving_view
|
||||
from sglang.srt.arg_groups.overrides import (
|
||||
declare_resolution,
|
||||
resolution_result,
|
||||
resolving_view,
|
||||
)
|
||||
from sglang.srt.configs.model_config import ModelConfig
|
||||
from sglang.srt.distributed.parallel_state import (
|
||||
destroy_distributed_environment,
|
||||
@@ -1034,8 +1038,8 @@ def main(server_args, bench_args):
|
||||
decode = dict(graph_config.get(Phase.DECODE) or {})
|
||||
decode["max_bs"] = max(bench_args.batch_size)
|
||||
graph_config[Phase.DECODE] = decode
|
||||
server_args = server_args.replace_resolved(
|
||||
"benchmark.one_batch", cuda_graph_config=graph_config
|
||||
declare_resolution(
|
||||
server_args, "benchmark.one_batch", cuda_graph_config=graph_config
|
||||
)
|
||||
server_args.resolve_once()
|
||||
cfg = resolving_view(server_args)
|
||||
|
||||
@@ -7,7 +7,7 @@ import logging
|
||||
from typing import Any
|
||||
|
||||
from sglang.srt.arg_groups.overrides import (
|
||||
declare_late_resolution,
|
||||
declare_resolution,
|
||||
resolving_view,
|
||||
)
|
||||
from sglang.srt.environ import envs
|
||||
@@ -24,9 +24,7 @@ def check_lora_server_args(server_args: Any):
|
||||
# Enable LoRA if any LoRA paths are provided for backward compatibility.
|
||||
if cfg.lora_paths:
|
||||
if cfg.enable_lora is None:
|
||||
declare_late_resolution(
|
||||
server_args, "check_lora_server_args", enable_lora=True
|
||||
)
|
||||
declare_resolution(server_args, "check_lora_server_args", enable_lora=True)
|
||||
logger.warning(
|
||||
"--enable-lora is set to True because --lora-paths is provided."
|
||||
)
|
||||
@@ -37,7 +35,7 @@ def check_lora_server_args(server_args: Any):
|
||||
|
||||
if cfg.enable_lora:
|
||||
if cfg.enable_lora_overlap_loading is None:
|
||||
declare_late_resolution(
|
||||
declare_resolution(
|
||||
server_args, "check_lora_server_args", enable_lora_overlap_loading=False
|
||||
)
|
||||
|
||||
@@ -93,11 +91,11 @@ def check_lora_server_args(server_args: Any):
|
||||
"Expected a string or a dictionary."
|
||||
)
|
||||
parsed_lora_paths.append(lora_ref)
|
||||
declare_late_resolution(
|
||||
declare_resolution(
|
||||
server_args, "check_lora_server_args", lora_paths=parsed_lora_paths
|
||||
)
|
||||
elif isinstance(cfg.lora_paths, dict):
|
||||
declare_late_resolution(
|
||||
declare_resolution(
|
||||
server_args,
|
||||
"check_lora_server_args",
|
||||
lora_paths=[
|
||||
@@ -111,9 +109,7 @@ def check_lora_server_args(server_args: Any):
|
||||
],
|
||||
)
|
||||
elif cfg.lora_paths is None:
|
||||
declare_late_resolution(
|
||||
server_args, "check_lora_server_args", lora_paths=[]
|
||||
)
|
||||
declare_resolution(server_args, "check_lora_server_args", lora_paths=[])
|
||||
else:
|
||||
raise ValueError(
|
||||
f"Invalid type for --lora-paths: {type(cfg.lora_paths)}. "
|
||||
@@ -123,7 +119,7 @@ def check_lora_server_args(server_args: Any):
|
||||
# Normalize target modules to a set; keep {"all"} as a sentinel
|
||||
# that gets resolved model-awarely in lora_manager.init_lora_shapes().
|
||||
if cfg.lora_target_modules:
|
||||
declare_late_resolution(
|
||||
declare_resolution(
|
||||
server_args,
|
||||
"check_lora_server_args",
|
||||
lora_target_modules=set(cfg.lora_target_modules),
|
||||
|
||||
@@ -17,10 +17,9 @@ Model-identity adjustments to the server configuration are DECLARED here and
|
||||
appended to the record's declaration stash (gate order, last writer wins).
|
||||
Nothing here writes back onto ``ServerArgs``: the record holds the user's raw
|
||||
input, and a decision is read through ``resolution_result`` or the published
|
||||
config bags — model code never mutates ``ServerArgs`` fields imperatively. The
|
||||
one channel that still leaves a field changed is ``declare_direct_writes``,
|
||||
which does not perform the write: it captures one an out-of-tree plugin already
|
||||
made, and undoing it would surprise the plugin's own reads.
|
||||
config bags — model code never mutates ``ServerArgs`` fields imperatively. That
|
||||
holds without exception: a resolver this tree does not own assigns onto a
|
||||
stand-in (``record_foreign_defaults``), and what it set is declared.
|
||||
|
||||
Two declaration forms, keyed on ``hf_config.architectures[0]``:
|
||||
|
||||
@@ -33,7 +32,6 @@ Two declaration forms, keyed on ``hf_config.architectures[0]``:
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import copy
|
||||
import dataclasses
|
||||
import json
|
||||
import logging
|
||||
@@ -116,9 +114,9 @@ def run_post_process_pass(server_args: Any, fn: Callable[..., dict]) -> None:
|
||||
an empty dict is a validation, and it may run on the published instance --
|
||||
it has to, because ``Engine(server_args=sa)`` after ``Engine.shutdown()``
|
||||
re-runs ``check_server_args`` on the very instance the context still holds.
|
||||
A pass that returns a non-empty dict there is refused, as
|
||||
``declare_late_resolution`` is -- post-publish changes go to the bags through
|
||||
``get_context().override(...)``.
|
||||
A pass that returns a non-empty dict there is refused by the guard in
|
||||
``declare_resolution``, as a late declaration is -- post-publish changes go
|
||||
to the bags through ``get_context().override(...)``.
|
||||
"""
|
||||
|
||||
declared = fn(ResolvedView(server_args, overlay=_declaration_overlay(server_args)))
|
||||
@@ -133,29 +131,12 @@ def run_post_process_pass(server_args: Any, fn: Callable[..., dict]) -> None:
|
||||
# a rebuild: `Engine(server_args=sa)` after `Engine.shutdown()` hands
|
||||
# back the same instance while the context still holds it, and
|
||||
# refusing on identity alone would fail that launch.
|
||||
try:
|
||||
published = get_context().server_args
|
||||
except ValueError:
|
||||
published = None
|
||||
if published is server_args:
|
||||
raise ValueError(
|
||||
f"run_post_process_pass({fn.__qualname__!r}) declared "
|
||||
f"{sorted(declared)} on the published config; the stash is "
|
||||
"projected at publish and never again, so this would be a "
|
||||
"silent no-op -- post-publish changes go to the bags via "
|
||||
"get_context().override(...)"
|
||||
)
|
||||
entry = (fn.__qualname__, dict(declared))
|
||||
stash = getattr(server_args, "_resolved_overrides", None)
|
||||
if stash is None:
|
||||
# Handlers hosting pass slots may be invoked directly on fixtures
|
||||
# that never ran the monolith dispatch (which owns the stash);
|
||||
# create it lazily. Real publishes always pass through the
|
||||
# dispatch first — the dispatch ASSIGNS the stash, so pass slots
|
||||
# must sit at or after it in __post_init__ order.
|
||||
stash = server_args._resolved_overrides = []
|
||||
stash.append(entry)
|
||||
validate_declarations(server_args, [entry])
|
||||
# Only a non-empty return is a declaration. An empty one is a
|
||||
# validation and may run on the published instance -- see above -- so it
|
||||
# must not reach the guard in `declare_resolution`.
|
||||
if declared:
|
||||
declare_resolution(server_args, fn.__qualname__, **declared)
|
||||
validate_declarations(server_args, [(fn.__qualname__, dict(declared))])
|
||||
|
||||
|
||||
def declare_resolution(server_args: Any, source: str, **fields: Any) -> None:
|
||||
@@ -167,52 +148,32 @@ def declare_resolution(server_args: Any, source: str, **fields: Any) -> None:
|
||||
(or `resolved_view(server_args)`), which
|
||||
`test_resolution_reads_the_declarations` pins.
|
||||
|
||||
For resolvers inside ``__post_init__``; launcher-stage resolution goes
|
||||
through ``declare_late_resolution``. A name that is not a field is rejected
|
||||
here rather than becoming an attribute nothing reads.
|
||||
Every declaration goes through here, whenever it is made: inside
|
||||
``__post_init__``, at launcher stage (LoRA normalization, the auto-detected
|
||||
parsers -- they decide what the process will run with, so they belong to the
|
||||
pipeline even though they run after it), and on a copy about to cross a
|
||||
process boundary. A name that is not a field is rejected here rather than
|
||||
becoming an attribute nothing reads.
|
||||
|
||||
Refuses the published config. The stash is projected at publish and never
|
||||
again, so a declaration afterwards is a silent no-op; post-publish changes
|
||||
go to the bags through ``get_context().override(...)``.
|
||||
"""
|
||||
if dataclasses.is_dataclass(type(server_args)):
|
||||
unknown = sorted(set(fields) - field_names(type(server_args)))
|
||||
if unknown:
|
||||
raise AttributeError(f"{source}: {unknown} are not ServerArgs fields")
|
||||
stash = getattr(server_args, "_resolved_overrides", None)
|
||||
if stash is None:
|
||||
stash = []
|
||||
server_args._resolved_overrides = stash
|
||||
stash.append((source, dict(fields)))
|
||||
|
||||
|
||||
def declare_late_resolution(server_args: Any, source: str, **fields: Any) -> None:
|
||||
"""Resolve fields on a config that is **not published yet**.
|
||||
|
||||
A few resolution rules cannot run inside ``__post_init__``: LoRA
|
||||
normalization and the auto-parser detection need the launcher's validation
|
||||
stage (and, for the parsers, a tokenizer / chat-template load). They still
|
||||
belong to the resolution pipeline — they decide what the process will run
|
||||
with — so their decision goes to the stash like any other, and the record
|
||||
keeps what the caller passed. Every holder of that instance reads the
|
||||
decision the same way the rest of the pipeline does: the bags it publishes,
|
||||
or ``resolution_result``, both of which survive the pickle to a child.
|
||||
|
||||
Refuses to touch the published instance: after publish the bags exist and a
|
||||
field write would desync them, which is what ``get_context().override`` is
|
||||
for.
|
||||
"""
|
||||
|
||||
try:
|
||||
published = get_context().server_args
|
||||
except ValueError:
|
||||
published = None
|
||||
if published is server_args:
|
||||
raise ValueError(
|
||||
f"declare_late_resolution({source!r}) called on the published config; "
|
||||
"post-publish changes go to the bags via get_context().override(...)"
|
||||
f"{source}: declared on the published config; the stash is "
|
||||
"projected at publish and never again, so this would be a silent "
|
||||
"no-op -- post-publish changes go to the bags via "
|
||||
"get_context().override(...)"
|
||||
)
|
||||
log = getattr(server_args, "_runtime_mutations", None)
|
||||
if log is None:
|
||||
log = []
|
||||
server_args._runtime_mutations = log
|
||||
log.append((source, dict(fields)))
|
||||
stash = getattr(server_args, "_resolved_overrides", None)
|
||||
if stash is None:
|
||||
stash = []
|
||||
@@ -220,59 +181,65 @@ def declare_late_resolution(server_args: Any, source: str, **fields: Any) -> Non
|
||||
stash.append((source, dict(fields)))
|
||||
|
||||
|
||||
def declare_direct_writes(
|
||||
class _ForeignDefaults:
|
||||
"""The stand-in handed to a resolver this tree does not own.
|
||||
|
||||
Reads fall through to the resolving view, so a plugin sees what resolution
|
||||
has decided so far rather than the raw input -- better than what it used to
|
||||
get, which was the record's own fields. Writes are captured here and
|
||||
declared by the caller, so the record is never written and the write seal
|
||||
has no exception.
|
||||
"""
|
||||
|
||||
__slots__ = ("_cfg", "_written")
|
||||
|
||||
def __init__(self, server_args: Any):
|
||||
object.__setattr__(self, "_cfg", resolving_view(server_args))
|
||||
object.__setattr__(self, "_written", {})
|
||||
|
||||
def __getattr__(self, name: str) -> Any:
|
||||
written = object.__getattribute__(self, "_written")
|
||||
if name in written:
|
||||
return written[name]
|
||||
return getattr(object.__getattribute__(self, "_cfg"), name)
|
||||
|
||||
def __setattr__(self, name: str, value: Any) -> None:
|
||||
object.__getattribute__(self, "_written")[name] = value
|
||||
|
||||
|
||||
def record_foreign_defaults(
|
||||
server_args: Any, source: str, resolve: Callable[[Any], Any]
|
||||
) -> Any:
|
||||
"""Run a resolver that writes the fields directly, and declare what it moved.
|
||||
"""Run a resolver this tree does not own, and declare what it set.
|
||||
|
||||
Out-of-tree platform plugins and registered speculative algorithms are
|
||||
handed a configuration and assign fields on it. That interface is not ours
|
||||
to change, so the assignment stays the contract -- it just lands on a
|
||||
stand-in instead of the record, and what it set is declared like any other
|
||||
decision. Nothing writes the record, which is why there is no longer a
|
||||
named hole in the seal.
|
||||
|
||||
Returns whatever the resolver returned, so a provider with a return value
|
||||
can go through the same capture.
|
||||
goes through the same capture.
|
||||
|
||||
Out-of-tree platform plugins are handed the record and set fields on it.
|
||||
Their implementations live outside this tree, so they cannot be converted
|
||||
by editing the resolver; and the raw snapshot is taken before the pipeline
|
||||
starts, so a plugin's default is neither declared nor raw. The write itself
|
||||
stays: this captures it into the stash so the projection and the bags carry
|
||||
it, but reverting the field would break the plugin's own reads of what it
|
||||
just set. It is the only field a record still carries from resolution.
|
||||
|
||||
Rebinding is what the diff sees, and rebinding is all it needs to see: a
|
||||
plugin that mutates a value in place reaches the projection anyway, because
|
||||
the raw snapshot and the stash entries hold the same object it mutated.
|
||||
Non-field names are dropped: a plugin scribbling on an attribute that is
|
||||
not configuration is not a decision, and it was invisible to the previous
|
||||
diff for the same reason.
|
||||
|
||||
A stand-in record (tests drive the hooks with a plain namespace) has no
|
||||
fields to diff and no projection to feed, so the resolver runs uncaptured.
|
||||
view to read, so the resolver runs against it directly and uncaptured.
|
||||
"""
|
||||
if not dataclasses.is_dataclass(server_args):
|
||||
return resolve(server_args)
|
||||
before = {
|
||||
field.name: getattr(server_args, field.name)
|
||||
for field in dataclasses.fields(server_args)
|
||||
recorder = _ForeignDefaults(server_args)
|
||||
result = resolve(recorder)
|
||||
written = {
|
||||
name: value
|
||||
for name, value in object.__getattribute__(recorder, "_written").items()
|
||||
if name in field_names(type(server_args))
|
||||
}
|
||||
already = len(getattr(server_args, "_resolved_overrides", None) or ())
|
||||
# The one place the input seal comes off. The plugin writes the record;
|
||||
# the diff below captures what it moved into the stash so the projection
|
||||
# and the bags carry it.
|
||||
from sglang.srt.server_args import record_writable
|
||||
|
||||
with record_writable(server_args):
|
||||
result = resolve(server_args)
|
||||
stash = getattr(server_args, "_resolved_overrides", None)
|
||||
if stash is None:
|
||||
stash = []
|
||||
server_args._resolved_overrides = stash
|
||||
# A resolver reached this way can also declare properly -- the in-tree
|
||||
# implementations of these hooks do. Those fields are already explained, and
|
||||
# recording them again would attribute them to the wrapper and bury an
|
||||
# actual direct write among the echoes.
|
||||
declared = {name for _source, fields in stash[already:] for name in fields}
|
||||
changed = {
|
||||
name: getattr(server_args, name)
|
||||
for name, previous in before.items()
|
||||
if name not in declared and getattr(server_args, name) is not previous
|
||||
}
|
||||
if changed:
|
||||
stash.append((source, changed))
|
||||
if written:
|
||||
declare_resolution(server_args, source, **written)
|
||||
return result
|
||||
|
||||
|
||||
@@ -303,40 +270,6 @@ def resolution_result(server_args: Any, field: str, default: Any = None) -> Any:
|
||||
return with_fallback(type(server_args), field, 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 -- the operator's
|
||||
input, not what resolution decided. 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 pre_capture_activation_reserve_mb_of(cfg: Any, gpu_mem: Optional[float]) -> float:
|
||||
"""The activation working-set reserve held back before cuda-graph capture.
|
||||
|
||||
|
||||
@@ -15,11 +15,9 @@ from sglang.srt.arg_groups.overrides import (
|
||||
_page_size_default,
|
||||
_pipeline_parallel_overlap_disable,
|
||||
_sampling_backend_default,
|
||||
declare_direct_writes,
|
||||
resolving_view,
|
||||
run_post_process_pass,
|
||||
)
|
||||
from sglang.srt.platforms import current_platform
|
||||
from sglang.srt.utils.common import get_device_memory_capacity
|
||||
|
||||
|
||||
@@ -204,6 +202,7 @@ def run_resolution_pipeline(server_args: Any) -> None:
|
||||
handle_mps_backends,
|
||||
handle_nccl_pre_warm,
|
||||
handle_npu_backends,
|
||||
handle_platform_defaults,
|
||||
handle_symm_mem_device_support,
|
||||
handle_xpu_backends,
|
||||
)
|
||||
@@ -217,13 +216,7 @@ def run_resolution_pipeline(server_args: Any) -> None:
|
||||
# keys off enable_symm_mem.
|
||||
handle_symm_mem_device_support(server_args)
|
||||
|
||||
# OOT platform plugins set fields directly (an interface this tree
|
||||
# does not own); the diff records what they applied.
|
||||
declare_direct_writes(
|
||||
server_args,
|
||||
f"platform:{current_platform.device_name}",
|
||||
current_platform.apply_server_args_defaults,
|
||||
)
|
||||
handle_platform_defaults(server_args)
|
||||
|
||||
gpu_mem = get_device_memory_capacity(cfg.device)
|
||||
|
||||
|
||||
@@ -9,6 +9,7 @@ from typing import Any
|
||||
|
||||
from sglang.srt.arg_groups.overrides import (
|
||||
declare_resolution,
|
||||
record_foreign_defaults,
|
||||
resolving_view,
|
||||
)
|
||||
from sglang.srt.hardware_backend.mlx.runtime import use_mlx
|
||||
@@ -84,6 +85,29 @@ def handle_nccl_pre_warm(server_args: Any):
|
||||
declare_resolution(server_args, "_handle_nccl_pre_warm", pre_warm_nccl=False)
|
||||
|
||||
|
||||
def handle_platform_defaults(server_args: Any):
|
||||
"""An out-of-tree platform's defaults, declared like every rule beside it.
|
||||
|
||||
`Platform.apply_server_args_defaults` is a plugin interface: the platform is
|
||||
handed a configuration and assigns the fields it wants defaulted. In-tree
|
||||
platforms do not implement it -- the base is a no-op and nothing overrides
|
||||
it -- so this captures nothing here and exists for the platforms that live
|
||||
outside this tree.
|
||||
|
||||
Ordering: it must precede `handle_gpu_memory_settings`, whose symm-mem
|
||||
prealloc default keys off `enable_symm_mem`.
|
||||
"""
|
||||
# `current_platform` is the plugin object; `get_platform()` is the facts
|
||||
# view over it and carries neither the name nor the hook.
|
||||
from sglang.srt.platforms import current_platform
|
||||
|
||||
record_foreign_defaults(
|
||||
server_args,
|
||||
f"platform:{current_platform.device_name}",
|
||||
current_platform.apply_server_args_defaults,
|
||||
)
|
||||
|
||||
|
||||
def handle_symm_mem_device_support(server_args: Any):
|
||||
cfg = resolving_view(server_args)
|
||||
# The symm-mem allocator compiles a CUDA plugin and links -lnccl, so off
|
||||
|
||||
@@ -8,9 +8,9 @@ from typing import TYPE_CHECKING, Optional
|
||||
from sglang.srt.arg_groups.overrides import (
|
||||
_speculative_moe_runner_default,
|
||||
attention_backends_of,
|
||||
declare_direct_writes,
|
||||
declare_resolution,
|
||||
model_config_of,
|
||||
record_foreign_defaults,
|
||||
resolved_view,
|
||||
resolving_view,
|
||||
run_post_process_pass,
|
||||
@@ -157,7 +157,7 @@ def handle_speculative_decoding(server_args: ServerArgs) -> None:
|
||||
|
||||
# TODO: move the per-algorithm validation below into spec module hooks.
|
||||
if isinstance(algo, CustomSpecAlgo) and algo.validate_server_args is not None:
|
||||
declare_direct_writes(
|
||||
record_foreign_defaults(
|
||||
server_args,
|
||||
"handle_speculative_decoding.custom_validate",
|
||||
algo.validate_server_args,
|
||||
@@ -175,13 +175,24 @@ def handle_speculative_decoding(server_args: ServerArgs) -> None:
|
||||
_init_adaptive_speculative_params(server_args)
|
||||
|
||||
if algo is not None:
|
||||
# A registered algorithm's callback lives outside this tree and sets
|
||||
# fields on the record, so the writes are captured around the call.
|
||||
declare_direct_writes(
|
||||
server_args,
|
||||
"handle_speculative_decoding.custom_algo",
|
||||
algo.handle_server_args,
|
||||
)
|
||||
# Imported here and not above: the name is only bound inside the
|
||||
# `speculative_algorithm is not None` branch, and this runs either way.
|
||||
from sglang.srt.speculative.spec_registry import CustomSpecAlgo
|
||||
|
||||
if isinstance(algo, CustomSpecAlgo):
|
||||
# A registered algorithm's callback lives outside this tree and
|
||||
# assigns fields, so it gets the stand-in and its writes are
|
||||
# declared.
|
||||
record_foreign_defaults(
|
||||
server_args,
|
||||
"handle_speculative_decoding.custom_algo",
|
||||
algo.handle_server_args,
|
||||
)
|
||||
else:
|
||||
# The in-tree dispatcher, which declares. It needs the record
|
||||
# itself: handed the stand-in, its `declare_resolution` calls would
|
||||
# stash on that instead.
|
||||
algo.handle_server_args(server_args)
|
||||
|
||||
|
||||
def _handle_dflash(server_args: ServerArgs) -> None:
|
||||
|
||||
@@ -29,7 +29,7 @@ import jinja2.ext
|
||||
import jinja2.nodes
|
||||
import jinja2.sandbox
|
||||
|
||||
from sglang.srt.arg_groups.overrides import declare_late_resolution, resolving_view
|
||||
from sglang.srt.arg_groups.overrides import declare_resolution, resolving_view
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -829,4 +829,4 @@ def resolve_auto_parsers(server_args) -> None:
|
||||
detected[attr] = _detect_auto_parser(attr, ctx, rules, label)
|
||||
|
||||
if detected:
|
||||
declare_late_resolution(server_args, "template-detection", **detected)
|
||||
declare_resolution(server_args, "template-detection", **detected)
|
||||
|
||||
@@ -24,6 +24,7 @@ import ray
|
||||
from ray.util.placement_group import PlacementGroup
|
||||
from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
|
||||
|
||||
from sglang.srt.arg_groups.overrides import declare_resolution
|
||||
from sglang.srt.entrypoints.engine import (
|
||||
Engine,
|
||||
SchedulerInitResult,
|
||||
@@ -471,12 +472,16 @@ class RayEngine(Engine):
|
||||
f"enable_dp_attention={parallel.enable_dp_attention}"
|
||||
)
|
||||
|
||||
# Set dist_init_addr on server_args so PortArgs.init_new() can compute
|
||||
# TCP addresses correctly (required for DP attention path).
|
||||
dp_server_args = server_args.replace_resolved(
|
||||
# Declared on the record itself so `PortArgs.init_new()` can compute
|
||||
# TCP addresses (required for the DP attention path). No copy: this
|
||||
# process does not publish here, and every other reader of the field
|
||||
# goes through the bags its own process projects.
|
||||
declare_resolution(
|
||||
server_args,
|
||||
"ray.dp_controller",
|
||||
dist_init_addr=f"{rank0_node_ip}:{port_args.nccl_port}",
|
||||
)
|
||||
dp_server_args = server_args
|
||||
# Create the DP controller in-process. This blocks until all actors
|
||||
# are initialized and their event loops have started.
|
||||
controller = RayDataParallelController(
|
||||
|
||||
@@ -20,6 +20,7 @@ from typing import Any, Dict, Optional
|
||||
|
||||
import ray
|
||||
|
||||
from sglang.srt.arg_groups.overrides import declare_resolution
|
||||
from sglang.srt.runtime_context import publish
|
||||
from sglang.srt.server_args import PortArgs, ServerArgs
|
||||
|
||||
@@ -54,11 +55,13 @@ class SchedulerActor:
|
||||
numa_bind_to_node,
|
||||
)
|
||||
|
||||
# Override dist_init_addr if provided (for multi-node), through
|
||||
# `replace_resolved` so the copy keeps the parent's resolution.
|
||||
# Declared, not copied: Ray deserializes the argument per call, so this
|
||||
# record is the actor's own and nothing else in the process holds it.
|
||||
# The field stays the operator's input; `PortArgs.init_new` and the bags
|
||||
# this actor publishes read the decision.
|
||||
if dist_init_addr:
|
||||
server_args = server_args.replace_resolved(
|
||||
"ray.scheduler_actor", dist_init_addr=dist_init_addr
|
||||
declare_resolution(
|
||||
server_args, "ray.scheduler_actor", dist_init_addr=dist_init_addr
|
||||
)
|
||||
|
||||
# Get actual GPU IDs from Ray runtime context
|
||||
|
||||
@@ -1282,7 +1282,7 @@ class _ServerArgsOverride:
|
||||
self._prev_parallel_config = ctx.parallel._config
|
||||
self._prev_capture = ctx.flags.capture.enable_torch_compile
|
||||
from sglang.srt.arg_groups.overrides import (
|
||||
declare_late_resolution,
|
||||
declare_resolution,
|
||||
)
|
||||
|
||||
server_args = ServerArgs(model_path="dummy")
|
||||
@@ -1306,7 +1306,7 @@ class _ServerArgsOverride:
|
||||
fields = set(type(server_args).__dataclass_fields__)
|
||||
declared = {n: v for n, v in self._fields.items() if n in fields}
|
||||
if declared:
|
||||
declare_late_resolution(server_args, "override_server_args", **declared)
|
||||
declare_resolution(server_args, "override_server_args", **declared)
|
||||
# What is left seeds the record's own private caches (`_model_config`
|
||||
# and friends), which are not configuration and never were.
|
||||
seeds = {n: v for n, v in self._fields.items() if n not in fields}
|
||||
|
||||
@@ -40,7 +40,6 @@ import functools
|
||||
import logging
|
||||
import tempfile
|
||||
import uuid
|
||||
from contextlib import contextmanager
|
||||
from typing import Any, NoReturn
|
||||
|
||||
from sglang.kernels.ops.kv_canary.consts import RealKvHashMode
|
||||
@@ -53,7 +52,7 @@ from sglang.srt.arg_groups.argparse_actions import (
|
||||
from sglang.srt.arg_groups.model_override_base import ep_joiner_of, ep_scale_joiner_of
|
||||
from sglang.srt.arg_groups.overrides import (
|
||||
remote_instance_transfer_engine_of,
|
||||
resolution_projection,
|
||||
resolution_result,
|
||||
resolving_view,
|
||||
)
|
||||
from sglang.srt.environ import envs
|
||||
@@ -170,6 +169,24 @@ from sglang.srt.utils.common import ( # noqa: F401
|
||||
)
|
||||
|
||||
|
||||
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)
|
||||
|
||||
|
||||
class ServerArgs:
|
||||
"""Server-wide configuration for SGLang.
|
||||
|
||||
@@ -252,9 +269,8 @@ class ServerArgs:
|
||||
from sglang.srt.arg_groups.pipeline import run_resolution_pipeline
|
||||
|
||||
# Sealed for the duration, not just afterwards: everything below this
|
||||
# line reads the input and declares against it, and the one channel
|
||||
# that still writes the record (`declare_direct_writes`, for
|
||||
# out-of-tree platform plugins) asks for the seal to be lifted by name.
|
||||
# line reads the input and declares against it. No exceptions -- even a
|
||||
# resolver from outside this tree assigns onto a stand-in, not here.
|
||||
self._input_frozen = True
|
||||
try:
|
||||
run_resolution_pipeline(self)
|
||||
@@ -298,58 +314,10 @@ class ServerArgs:
|
||||
`model_config` memo are not fields and do not appear.
|
||||
"""
|
||||
|
||||
return resolution_projection(self)
|
||||
|
||||
def replace_resolved(self, source: str, **changes: Any) -> ServerArgs:
|
||||
"""A copy of this record that stays resolved, and says what it changed.
|
||||
|
||||
`dataclasses.replace` builds a new instance, so the copy carries none of
|
||||
what makes a record resolved: no raw snapshot, no declarations, no
|
||||
finished flag. The next publish therefore resolves it again, which
|
||||
drops every decision the stash held -- the late ones (the auto-detected
|
||||
parsers) and the direct ones alike -- and re-runs the device probes in
|
||||
whatever process opened the copy. The Ray paths replace
|
||||
`dist_init_addr` on a resolved record, which is how they reach this.
|
||||
|
||||
The change is appended to the stash rather than left on the field: the
|
||||
projection reads the raw snapshot plus the declarations, so a field the
|
||||
copy set on its own would publish the parent's raw value instead.
|
||||
|
||||
The carry is shallow. The containers are copied so the copy's own
|
||||
declaration does not travel back into the parent, but everything inside
|
||||
them -- the stash entries, the raw-input values, the memoized
|
||||
`ModelConfig` -- is shared. That is fine for what this is for: a copy
|
||||
that immediately crosses a process boundary (Ray actors, the gateway's
|
||||
workers), where pickling severs the sharing. A caller that mutates the
|
||||
copy's deep structure in-process mutates the parent's too.
|
||||
"""
|
||||
replacement = dataclasses.replace(self, **changes)
|
||||
# Provenance, not resolution state: a copy was still launched by
|
||||
# whatever launched its parent, resolved or not.
|
||||
object.__setattr__(replacement, "_launch_command", self.launch_command)
|
||||
if not getattr(self, "_resolution_finished", False):
|
||||
# Not resolved yet: the copy goes through the gate itself.
|
||||
return replacement
|
||||
|
||||
# Everything outside the fields, enumerated from the instance: the raw
|
||||
# snapshot, the stash, and what resolution memoized -- including the
|
||||
# model-configuration memo, which the copy carries over rather than
|
||||
# rebuild.
|
||||
field_names = {field.name for field in dataclasses.fields(self)}
|
||||
for name, value in vars(self).items():
|
||||
if name in field_names or name == "_resolution_finished":
|
||||
continue
|
||||
if isinstance(value, (dict, list, set)):
|
||||
value = copy.copy(value)
|
||||
object.__setattr__(replacement, name, value)
|
||||
stash = getattr(replacement, "_resolved_overrides", None)
|
||||
if stash is None:
|
||||
stash = []
|
||||
object.__setattr__(replacement, "_resolved_overrides", stash)
|
||||
if changes:
|
||||
stash.append((source, dict(changes)))
|
||||
object.__setattr__(replacement, "_resolution_finished", True)
|
||||
return replacement
|
||||
return {
|
||||
field.name: _plain(resolution_result(self, field.name))
|
||||
for field in dataclasses.fields(self)
|
||||
}
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# CUDA graph configuration resolution
|
||||
@@ -663,27 +631,6 @@ def get_global_server_args() -> NoReturn:
|
||||
)
|
||||
|
||||
|
||||
@contextmanager
|
||||
def record_writable(server_args: Any):
|
||||
"""Lift the input seal for a resolver that genuinely writes the record.
|
||||
|
||||
There is exactly one: `declare_direct_writes`, which hands the record to an
|
||||
out-of-tree platform plugin that sets fields on it. Those implementations
|
||||
live outside this tree and cannot be converted by editing a resolver here,
|
||||
so the write stays and is captured into the stash afterwards. Naming the
|
||||
exception is the point -- an in-tree resolver that reaches for this is
|
||||
doing something it should be declaring instead.
|
||||
"""
|
||||
frozen = getattr(server_args, "_input_frozen", False)
|
||||
if frozen:
|
||||
object.__setattr__(server_args, "_input_frozen", False)
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
if frozen:
|
||||
object.__setattr__(server_args, "_input_frozen", True)
|
||||
|
||||
|
||||
def prepare_server_args(argv: list[str]) -> ServerArgs:
|
||||
"""
|
||||
Prepare the server arguments from the command line arguments.
|
||||
|
||||
Reference in New Issue
Block a user