config: the resolution pipeline's dispatcher leaves the record (#36896)
Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5
parent
46ccd7ce3e
commit
48b88e1256
@@ -311,6 +311,11 @@ def _hook_declarations(dispatch, source_module):
|
||||
return out
|
||||
|
||||
|
||||
# The dispatcher's own file: its imports are what map a bare-name call in it
|
||||
# to the family that defines the callable.
|
||||
_DISPATCH_MODULE = _SRT / "arg_groups" / "pipeline.py"
|
||||
|
||||
|
||||
def _hook_functions():
|
||||
"""Module-level resolution functions under `arg_groups/`.
|
||||
|
||||
@@ -341,7 +346,7 @@ def _pipeline():
|
||||
# against `arg_groups/` alongside the record's own methods.
|
||||
hooks = _hook_functions()
|
||||
methods.update({name: node for name, node in hooks.items() if name not in methods})
|
||||
dispatch = methods["_run_resolution_pipeline"]
|
||||
dispatch = methods["run_resolution_pipeline"]
|
||||
# A step is either a record method (`self._x()`) or a bare-name hook call.
|
||||
steps = [
|
||||
name
|
||||
@@ -501,7 +506,7 @@ def _declaration_positions():
|
||||
build_index, build_step, build_method, build_line_in_body = site
|
||||
first_build = (build_index, build_step)
|
||||
|
||||
source_module = _SRT / "server_args.py"
|
||||
source_module = _DISPATCH_MODULE
|
||||
imported = {}
|
||||
for node in ast.walk(_parsed(source_module)):
|
||||
if isinstance(node, ast.ImportFrom) and node.module:
|
||||
@@ -551,9 +556,9 @@ def _declaration_positions():
|
||||
# the dispatcher*: a handler body sits further down the file than the
|
||||
# dispatcher that calls it, so a line number taken from one scope says
|
||||
# nothing about ordering against the other.
|
||||
dispatch = methods["_run_resolution_pipeline"]
|
||||
dispatch = methods["run_resolution_pipeline"]
|
||||
build_line = step_lines[first_build[1]]
|
||||
for field, line in _hook_declarations(dispatch, _SRT / "server_args.py").items():
|
||||
for field, line in _hook_declarations(dispatch, _DISPATCH_MODULE).items():
|
||||
if field in wanted and line > build_line:
|
||||
declared_at[field] = max(
|
||||
declared_at.get(field, (build_index, 1)), (10**6, 1)
|
||||
@@ -656,8 +661,8 @@ class TestModelConfigReadsResolvedInput(CustomTestCase):
|
||||
documents a hazard that no longer exists and hides the day one appears.
|
||||
"""
|
||||
steps, methods, reached, step_lines = _pipeline()
|
||||
dispatch = methods["_run_resolution_pipeline"]
|
||||
hooks = _hook_declarations(dispatch, _SRT / "server_args.py")
|
||||
dispatch = methods["run_resolution_pipeline"]
|
||||
hooks = _hook_declarations(dispatch, _DISPATCH_MODULE)
|
||||
build_line = min(
|
||||
step_lines[step]
|
||||
for step in steps
|
||||
@@ -695,8 +700,8 @@ class TestModelConfigReadsResolvedInput(CustomTestCase):
|
||||
then, rather than keeping a note about a hazard that is gone.
|
||||
"""
|
||||
steps, methods, reached, step_lines = _pipeline()
|
||||
dispatch = methods["_run_resolution_pipeline"]
|
||||
positions = _opaque_callback_positions(dispatch, _SRT / "server_args.py")
|
||||
dispatch = methods["run_resolution_pipeline"]
|
||||
positions = _opaque_callback_positions(dispatch, _DISPATCH_MODULE)
|
||||
self.assertEqual(
|
||||
sorted(positions),
|
||||
[
|
||||
|
||||
@@ -526,14 +526,16 @@ class TestResolutionDeclarations(CustomTestCase):
|
||||
reset_context()
|
||||
child = pickle.loads(blob)
|
||||
entered = []
|
||||
original = ServerArgs._run_resolution_pipeline
|
||||
from sglang.srt.arg_groups import pipeline as pipeline_module
|
||||
|
||||
def counted(self, _original=original):
|
||||
original = pipeline_module.run_resolution_pipeline
|
||||
|
||||
def counted(server_args, _original=original):
|
||||
entered.append(1)
|
||||
return _original(self)
|
||||
return _original(server_args)
|
||||
|
||||
with unittest.mock.patch.object(
|
||||
ServerArgs, "_run_resolution_pipeline", counted
|
||||
pipeline_module, "run_resolution_pipeline", counted
|
||||
):
|
||||
publish(child, role="scheduler")
|
||||
self.assertEqual(
|
||||
@@ -848,7 +850,7 @@ class TestResolutionDeclarations(CustomTestCase):
|
||||
"capturing like apply_server_args_defaults",
|
||||
)
|
||||
|
||||
pipeline = (_SRT / "server_args.py").read_text(encoding="utf-8-sig")
|
||||
pipeline = (_SRT / "arg_groups" / "pipeline.py").read_text(encoding="utf-8-sig")
|
||||
for hook in sorted(taking_the_record):
|
||||
self.assertIn(
|
||||
f"current_platform.{hook},",
|
||||
@@ -892,9 +894,11 @@ class TestResolutionDeclarations(CustomTestCase):
|
||||
server_args.attention_backend = "triton"
|
||||
server_args.schedule_conservativeness = 0.5
|
||||
|
||||
with unittest.mock.patch.object(
|
||||
server_args_module, "current_platform", _Plugin()
|
||||
):
|
||||
from sglang.srt.arg_groups import pipeline as pipeline_module
|
||||
|
||||
# The write capture runs in the dispatcher, so that is the namespace the
|
||||
# plugin has to be installed in.
|
||||
with unittest.mock.patch.object(pipeline_module, "current_platform", _Plugin()):
|
||||
server_args = self._resolve({})
|
||||
self.assertEqual(
|
||||
(
|
||||
|
||||
@@ -445,14 +445,16 @@ class TestResolutionIsReproducible(_RestoresProcessState, CustomTestCase):
|
||||
record.resolve_once()
|
||||
|
||||
entries = []
|
||||
original = ServerArgs._run_resolution_pipeline
|
||||
from sglang.srt.arg_groups import pipeline as pipeline_module
|
||||
|
||||
def counted(self):
|
||||
original = pipeline_module.run_resolution_pipeline
|
||||
|
||||
def counted(server_args):
|
||||
entries.append(1)
|
||||
return original(self)
|
||||
return original(server_args)
|
||||
|
||||
with unittest.mock.patch.object(
|
||||
ServerArgs, "_run_resolution_pipeline", counted
|
||||
pipeline_module, "run_resolution_pipeline", counted
|
||||
):
|
||||
record.resolve_once()
|
||||
self.assertEqual(
|
||||
@@ -964,7 +966,7 @@ class TestTheResolutionSeamHasOneCaller(CustomTestCase):
|
||||
for path in sorted(package_root.rglob("*.py")):
|
||||
try:
|
||||
source = path.read_text()
|
||||
if "_run_resolution_pipeline" not in source:
|
||||
if "run_resolution_pipeline" not in source:
|
||||
continue
|
||||
tree = ast.parse(source)
|
||||
except SyntaxError:
|
||||
@@ -984,8 +986,8 @@ class TestTheResolutionSeamHasOneCaller(CustomTestCase):
|
||||
for node in ast.walk(tree):
|
||||
if (
|
||||
isinstance(node, ast.Call)
|
||||
and isinstance(node.func, ast.Attribute)
|
||||
and node.func.attr == "_run_resolution_pipeline"
|
||||
and isinstance(node.func, ast.Name)
|
||||
and node.func.id == "run_resolution_pipeline"
|
||||
):
|
||||
rel = path.relative_to(package_root).as_posix()
|
||||
callers.append((rel, ".".join(scopes.get(id(node), ()))))
|
||||
@@ -1131,12 +1133,14 @@ class TestResolutionStaysLazy(CustomTestCase):
|
||||
import sglang
|
||||
|
||||
srt = pathlib.Path(next(iter(sglang.__path__))).resolve() / "srt"
|
||||
tree = ast.parse((srt / "server_args.py").read_text(encoding="utf-8-sig"))
|
||||
tree = ast.parse(
|
||||
(srt / "arg_groups" / "pipeline.py").read_text(encoding="utf-8-sig")
|
||||
)
|
||||
dispatch = next(
|
||||
node
|
||||
for node in ast.walk(tree)
|
||||
if isinstance(node, ast.FunctionDef)
|
||||
and node.name == "_run_resolution_pipeline"
|
||||
and node.name == "run_resolution_pipeline"
|
||||
)
|
||||
early_return = min(
|
||||
(
|
||||
|
||||
@@ -81,37 +81,6 @@ def _field_reads(fn, holders):
|
||||
yield node.lineno, node.attr
|
||||
|
||||
|
||||
def _resolution_handlers():
|
||||
"""The `ServerArgs` methods the dispatcher reaches, transitively."""
|
||||
tree = ast.parse((_SRT / "server_args.py").read_text(encoding="utf-8-sig"))
|
||||
cls = next(
|
||||
node
|
||||
for node in ast.walk(tree)
|
||||
if isinstance(node, ast.ClassDef) and node.name == "ServerArgs"
|
||||
)
|
||||
methods = {
|
||||
node.name: node
|
||||
for node in cls.body
|
||||
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
|
||||
}
|
||||
assert "_run_resolution_pipeline" in methods, "the dispatcher was renamed"
|
||||
seen, stack = set(), ["_run_resolution_pipeline"]
|
||||
while stack:
|
||||
name = stack.pop()
|
||||
if name in seen or name not in methods:
|
||||
continue
|
||||
seen.add(name)
|
||||
for node in ast.walk(methods[name]):
|
||||
if (
|
||||
isinstance(node, ast.Call)
|
||||
and isinstance(node.func, ast.Attribute)
|
||||
and isinstance(node.func.value, ast.Name)
|
||||
and node.func.value.id == "self"
|
||||
):
|
||||
stack.append(node.func.attr)
|
||||
return {name: methods[name] for name in seen}
|
||||
|
||||
|
||||
_DECLARERS = frozenset(
|
||||
{
|
||||
"_declare",
|
||||
@@ -469,35 +438,30 @@ class TestResolutionReadsTheDeclarations(CustomTestCase):
|
||||
+ "\n ".join(offenders),
|
||||
)
|
||||
|
||||
def test_no_handler_reads_a_field_off_self(self):
|
||||
handlers = _resolution_handlers()
|
||||
# What the dispatcher reaches inside the class is these read wrappers;
|
||||
# the package side is covered by
|
||||
# `test_no_hook_reads_a_field_off_the_record`. Pinned rather than
|
||||
# counted: a walk that collapsed to the wrappers would clear any floor
|
||||
# low enough to admit them.
|
||||
self.assertEqual(
|
||||
set(handlers),
|
||||
{
|
||||
"_run_resolution_pipeline",
|
||||
"_handle_hardware_runtime_validation",
|
||||
"_handle_page_size",
|
||||
"_handle_pipeline_parallelism",
|
||||
"_handle_sampling_backend",
|
||||
},
|
||||
f"the walk reached {sorted(handlers)}; if the dispatcher grew or "
|
||||
"lost a handler, add it here after checking it reads the view",
|
||||
def test_the_record_hosts_no_resolution_handler(self):
|
||||
"""The pipeline and every step it runs live under `arg_groups/`.
|
||||
|
||||
While a step was a method, it could read a raw field off `self` and
|
||||
`test_no_handler_reads_a_field_off_self` had to say it could not. There
|
||||
is no such method left, so the invariant is now the stronger one: the
|
||||
record hosts none of them. What the steps read is checked on the
|
||||
package side, by `test_no_hook_reads_a_field_off_the_record`.
|
||||
"""
|
||||
handlers = sorted(
|
||||
name
|
||||
for name, node in _record_members().items()
|
||||
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
|
||||
and (
|
||||
name.split(".")[-1].startswith(("_handle_", "_validate_"))
|
||||
or "resolution_pipeline" in name
|
||||
)
|
||||
)
|
||||
offenders = []
|
||||
for name, fn in sorted(handlers.items()):
|
||||
for lineno, field in _field_reads(fn, {"self"}):
|
||||
offenders.append(f"server_args.py:{lineno} {name} reads self.{field}")
|
||||
self.assertEqual(
|
||||
offenders,
|
||||
handlers,
|
||||
[],
|
||||
"a resolution handler reads its own field; the field holds the raw "
|
||||
"input. Bind `cfg = resolving_view(self)` and read that:\n "
|
||||
+ "\n ".join(offenders),
|
||||
"a resolution handler is back on the record; it belongs in an "
|
||||
"`arg_groups` family, where the package-side guards can see it:\n "
|
||||
+ "\n ".join(handlers),
|
||||
)
|
||||
|
||||
def test_no_member_recomputes_from_a_raw_field(self):
|
||||
|
||||
Reference in New Issue
Block a user