Skip to content

WIP Ichristo/schema organizing #380

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
cijohnson wants to merge 1 commit into
base: main
Choose a base branch
from
Open

WIP Ichristo/schema organizing #380

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion AGENTS.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,7 +43,7 @@ make ut # sdist -> .test_venv -> run_all_unittests.py
### Test Structure
- Test functions use `test_` prefix (PyTest requirement)
- Customer test suites live under `cvs/tests/`; library code under `cvs/lib/`
- Configuration files in `cvs/input/config_file/` (JSON-structured, explicit parameters). Sample configs mark mandatory user fields with `<changeme>` in value strings — users must replace every `<changeme>` before running tests; unresolved placeholders hard-exit at startup via `_resolve_placeholders_in_dict` (`cvs/lib/utils_lib.py`), reached by `resolve_test_config_placeholders`, which nearly every test module calls. The `<changeme>` validators in `cvs/parsers/schemas.py` cover only the aorta and pytorch-xdit configs.
- Configuration files in `cvs/input/config_file/` (JSON-structured, explicit parameters). Sample configs mark mandatory user fields with `<changeme>` in value strings — users must replace every `<changeme>` before running tests; unresolved placeholders hard-exit at startup via `_resolve_placeholders_in_dict` (`cvs/lib/utils_lib.py`), reached by `resolve_test_config_placeholders`, which nearly every test module calls. The `<changeme>` validators in `cvs/schema/config_file/` (aorta and pytorch_xdit) cover only those config types.
- Tests require `--cluster_file` and `--config_file` CLI args (wired via `cvs/conftest.py`; the `orch` fixture is in `cvs/tests/conftest.py`)

### Orchestrator Patterns (Recommended)
Expand Down
9 changes: 6 additions & 3 deletions cvs/lib/inference/ADDING_A_SUITE.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,8 +65,10 @@ Create `cvs/lib/<your_framework>/utils/<your_framework>_config_loader.py`.
from pydantic import model_validator
from typing_extensions import Literal

from cvs.lib.utils.config_loader import BaseVariantConfig, _Forbid, substitute_config
from cvs.lib.inference.utils.inferencing_config_loader import (
from cvs.schema.base import _Forbid
from cvs.schema.common.base import BaseVariantConfig
from cvs.lib.utils.config_loader import substitute_config
from cvs.schema.config_file.inference.common.sweep import (
GoodputSlo, Roles, Run, Sweep, SeqCombo, validate_sweep_selector,
)
from cvs.lib.<your_framework>.utils.<your_framework>_parsing import GATED_METRICS
Expand Down Expand Up @@ -100,7 +102,8 @@ class VariantConfig(BaseVariantConfig):

@model_validator(mode="after")
def _check_thresholds_cover_sweep(self):
# Copy the two-axis check from inferencing_config_loader.py:
# Copy the two-axis check from cvs/schema/config_file/inference/common/sweep.py
# (`validate_thresholds_cover_sweep`):
# Axis 1: every sweep cell has a threshold entry; no key names a phantom cell.
# Axis 2: every present cell has a spec for every GATED_METRICS member.
# When enforce_thresholds=False: warn instead of raise.
Expand Down
315 changes: 46 additions & 269 deletions cvs/lib/inference/atom/atom_config_loader.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,277 +2,50 @@
Copyright 2025 Advanced Micro Devices, Inc.
All rights reserved.

ATOM suite config schema (``atom``).
ATOM suite config loader (``atom``).

Generic paths/model/container/threshold plumbing lives in
:mod:`cvs.lib.utils.config_loader`. Sweep selector types are shared with
:mod:`cvs.lib.inference.utils.inferencing_config_loader`.
Pydantic models live in ``cvs.schema.config_file.inference.atom.variant``.
'''

from __future__ import annotations

import re
from typing import Any

from pydantic import Field, field_validator, model_validator
from typing_extensions import Literal

from cvs.lib import globals
from cvs.lib.inference.atom.atom_parsing import GATED_METRICS
from cvs.lib.inference.utils.accuracy_config import AccuracyConfig
from cvs.lib.inference.utils.functional_config import FunctionalConfig
from cvs.lib.inference.utils.inferencing_config_loader import (
RoleServer,
Sweep,
validate_sweep_selector,
validate_thresholds_cover_sweep,
from cvs.lib.utils.config_loader import substitute_config
from cvs.schema.config_file.inference.atom.variant import (
ATOM_DRIVERS,
ATOM_PP_DRIVERS,
AtomParams,
AtomRoleServer,
AtomRoles,
AtomRunCard,
AtomVariantConfig,
MtpQualityConfig,
QuantParityConfig,
merge_mxfp4_triton_env,
)
from cvs.lib.inference.utils.long_context_accuracy_config import LongContextAccuracyConfig
from cvs.lib.inference.utils.platform_config import PlatformConfig
from cvs.lib.utils.config_loader import BaseVariantConfig, _Forbid, substitute_config

ATOM_DRIVERS = ("atom", "vllm", "vllm_atom", "sglang")
ATOM_PP_DRIVERS = ("vllm", "vllm_atom", "sglang")
# MI300X MXFP4 MoE + A4W4 GEMM require Triton; aiter A4W4 is unsupported on gfx942.
# atom.utils.envs treats only "1" as true — "true" is ignored.
_MXFP4_TRITON_ENV = {
"ATOM_USE_TRITON_MOE": "1",
"ATOM_USE_TRITON_GEMM": "1",
}


def merge_mxfp4_triton_env(precision: str, env: dict[str, str]) -> dict[str, str]:
"""Return server env with MXFP4 Triton defaults applied when unset."""
merged = dict(env or {})
if (precision or "").lower() == "mxfp4":
for key, value in _MXFP4_TRITON_ENV.items():
merged.setdefault(key, value)
for key in _MXFP4_TRITON_ENV:
if str(merged.get(key, "")).lower() == "true":
merged[key] = "1"
return merged


log = globals.log

# Written by test_discover_topology / resolve_multinode_fabric — not user env.
_ORCH_MANAGED_NETWORK_ENV = frozenset({"NCCL_SOCKET_IFNAME", "GLOO_SOCKET_IFNAME", "TP_SOCKET_IFNAME", "NCCL_IB_HCA"})
_IB_HCA_NETDEV_RE = re.compile(r"^mlx5_\d+$", re.IGNORECASE)


class AtomRoleServer(RoleServer):
# Extra CLI tokens for ``python -m atom.entrypoints.openai_server`` after
# ``--model`` / ``--server-port`` (e.g. ``-tp``, ``--kv_cache_dtype``).
atom_args: list[str] = []
# Extra CLI tokens appended to ``python3 -m sglang.launch_server`` (driver=sglang).
sglang_args: list[str] = []
# IB HCA devices for NCCL_IB_HCA (multinode only).
# absent or "auto" -> use whatever ibv_devinfo -l reports (test_discover_topology).
# explicit list -> validated at preflight against ibv_devinfo output.
ib_hca_devices: Literal["auto"] | list[str] | None = None
# Linux netdev for NCCL_SOCKET_IFNAME / GLOO_SOCKET_IFNAME on multinode PP runs.
# absent or "auto" -> resolved at runtime by test_discover_topology from cluster IPs.
ib_netdev: Literal["auto"] | str | None = None

@field_validator("ib_netdev", mode="after")
@classmethod
def _normalize_ib_netdev(cls, v):
raw = (v or "").strip()
if raw and raw.lower() != "auto" and _IB_HCA_NETDEV_RE.match(raw):
log.warning(
"roles.server.ib_netdev=%r looks like an IB HCA name; coercing to 'auto' "
"(socket netdev is discovered from cluster IPs at runtime)",
raw,
)
return "auto"
return v

@model_validator(mode="after")
def _strip_orchestrator_managed_network_env(self):
if not self.env:
return self
dropped = sorted(k for k in self.env if k in _ORCH_MANAGED_NETWORK_ENV)
if not dropped:
return self
log.warning(
"roles.server.env drops orchestrator-managed keys %s "
"(set by test_discover_topology / build_server_cmd instead)",
dropped,
)
self.env = {k: v for k, v in self.env.items() if k not in _ORCH_MANAGED_NETWORK_ENV}
return self


class AtomRoles(_Forbid):
server: AtomRoleServer = AtomRoleServer()


class AtomParams(_Forbid):
# ``atom`` = standalone ATOM openai_server + benchmark_serving.
# ``vllm_atom`` = vLLM coordinator + ATOM local kernels (true multinode PP).
# ``vllm`` = interim ROCm vLLM uplift (vllm serve + vllm bench serve).
# ``sglang`` = SGLang coordinator (launch_server + bench_serving) for PP runs.
driver: Literal["atom", "vllm", "vllm_atom", "sglang"] = "vllm"
backend: str = "vllm"
base_url: str = "http://0.0.0.0"
port_no: str = "8000"
dataset_name: str = "random"
burstiness: str = "1.0"
seed: str = "0"
request_rate: str = "inf"
random_range_ratio: str = "0.8"
random_prefix_len: str = "0"
tensor_parallelism: str = "8"
tokenizer_mode: str = "auto"
percentile_metrics: str = "ttft,tpot,itl,e2el"
metric_percentiles: str = "95,99"
num_prompts: str = "1000"
max_model_length: str = "8192"
client_poll_count: str = "50"
client_poll_wait_time: str = "60"
client_initial_wait_s: str = "120"
server_precheck_wait_s: str = "30"
server_warmup_wait_s: str = "330"
server_poll_count: str = "60"
server_poll_wait_time: str = "60"
reuse_server_across_sweep: str = "false"
bench_max_failed_requests: str = "0"
bench_extra_args: str = ""
result_filename: str = "results"
# Multinode (M5): omit or set nnodes=1 for single-node runs. When nnodes>1,
# cluster node_dict must list the same number of hosts and test_setup_sshd runs.
nnodes: str = "1"
pipeline_parallel_size: str = "1"
master_addr: str = ""
master_port: str = "29501"
# Optional single-node reference output_throughput for scaling.efficiency_pct.
scaling_baseline_output_throughput: str = ""


class AtomRunCard(_Forbid):
upstream_run_url: str = ""
atom_image_pin: str = ""
notes: str = ""


class MtpQualityConfig(_Forbid):
enabled: bool = False
chat_template_prompt: str = "Say hello in one short sentence."
chat_template_expected_sha256: str = ""


class QuantParityConfig(_Forbid):
enabled: bool = False
probe_prompt: str = "The capital of France is"
reference_config_stem: str = ""


ATOM_FRAMEWORKS = ("atom",)


class AtomVariantConfig(BaseVariantConfig):
framework: Literal["atom"]

gpu_arch: str
run_card: AtomRunCard = AtomRunCard()
roles: AtomRoles = AtomRoles()
params: AtomParams
sweep: Sweep
accuracy: AccuracyConfig = Field(default_factory=AccuracyConfig)
mtp_quality: MtpQualityConfig = Field(default_factory=MtpQualityConfig)
quant_parity: QuantParityConfig = Field(default_factory=QuantParityConfig)
functional: FunctionalConfig = Field(default_factory=FunctionalConfig)
long_context_accuracy: LongContextAccuracyConfig = Field(default_factory=LongContextAccuracyConfig)
platform: PlatformConfig = Field(default_factory=PlatformConfig)

def cell_key(self, isl, osl, concurrency):
p = self.params
key = f"ISL={isl},OSL={osl},TP={p.tensor_parallelism}"
nnodes = int(p.nnodes)
pp = int(p.pipeline_parallel_size)
if p.driver == "atom":
if nnodes > 1:
key += f",DP={nnodes},NNODES={nnodes}"
elif p.driver in ATOM_PP_DRIVERS:
if pp > 1 or nnodes > 1:
key += f",PP={p.pipeline_parallel_size}"
if nnodes > 1:
key += f",NNODES={p.nnodes}"
return f"{key},CONC={concurrency}"

def expected_cells(self) -> list[str]:
by_name = {c.name: c for c in self.sweep.sequence_combinations}
return [self.cell_key(by_name[r.combo].isl, by_name[r.combo].osl, r.concurrency) for r in self.sweep.runs]

@model_validator(mode="after")
def _apply_mxfp4_triton_env_defaults(self):
self.roles.server.env = merge_mxfp4_triton_env(self.model.precision, self.roles.server.env)
return self

@model_validator(mode="after")
def _check_thresholds_cover_sweep(self):
validate_thresholds_cover_sweep(
expected_cells=self.expected_cells(),
thresholds=self.thresholds,
enforce_thresholds=self.enforce_thresholds,
gated_metrics=GATED_METRICS,
)
if int(self.params.nnodes) > 1 and (self.params.scaling_baseline_output_throughput or "").strip():
missing = []
for cell in self.expected_cells():
specs = self.thresholds.get(cell) or {}
if "scaling.efficiency_pct" not in specs:
missing.append(cell)
if missing:
msg = (
"multinode variant with scaling_baseline_output_throughput requires "
f"scaling.efficiency_pct in every cell; missing: {missing}"
)
if self.enforce_thresholds:
raise ValueError(msg)
import warnings

warnings.warn(f"{msg} (enforce_thresholds=false -> record-only)", stacklevel=2)
return self

@model_validator(mode="after")
def _atom_multinode_uses_dp_not_pp(self):
if self.params.driver == "atom" and int(self.params.nnodes) > 1:
if int(self.params.pipeline_parallel_size) > 1:
raise ValueError(
"params.driver='atom' with nnodes>1 uses ATOM SPMD data parallel (-dp); "
"standalone ATOM cannot execute pipeline parallel. For true PP>1 use "
"params.driver='vllm_atom' or 'sglang'."
)
return self

@model_validator(mode="after")
def _pp_driver_distributed_consistency(self):
driver = self.params.driver
if driver not in ATOM_PP_DRIVERS:
return self
nn = int(self.params.nnodes)
pp = int(self.params.pipeline_parallel_size)
is_ray = self.roles.server.serve_args.get("distributed-executor-backend") == "ray"
if nn > 1 and pp == 1 and not is_ray:
raise ValueError(
f"params.driver={driver!r} with nnodes={nn} requires pipeline_parallel_size>1 "
f"(got pp={pp}) for multinode pipeline parallel"
)
if pp > 1 and nn == 1:
raise ValueError(
f"pipeline_parallel_size={pp} > 1 requires nnodes > 1 (got nnodes={nn}) for params.driver={driver!r}"
)
return self

@model_validator(mode="after")
def _atom_driver_requires_inline_server_args(self):
if self.params.driver == "atom" and not self.roles.server.atom_args:
raise ValueError(
"params.driver='atom' requires roles.server.atom_args "
"(inline ATOM openai_server CLI tokens, vLLM-style)"
)
return self
from cvs.schema.config_file.inference.common.sweep import validate_sweep_selector

__all__ = [
"ATOM_DRIVERS",
"ATOM_PP_DRIVERS",
"AtomParams",
"AtomRoleServer",
"AtomRoles",
"AtomRunCard",
"AtomVariantConfig",
"MtpQualityConfig",
"QuantParityConfig",
"expand_sweep",
"expand_sweep_parametrize",
"load_variant",
"merge_mxfp4_triton_env",
"orchestrator_container_from_variant",
"placeholder_gated_threshold_cell",
"reuse_server_flag",
"server_session_key",
"validate_sweep_selector",
]


def expand_sweep(sweep):
Expand All @@ -296,13 +69,11 @@ def expand_sweep(sweep):


def reuse_server_flag(params) -> bool:
"""Return True when ``params.reuse_server_across_sweep`` is a truthy string."""
raw = str(getattr(params, "reuse_server_across_sweep", "false")).strip().lower()
return raw in ("true", "1", "yes")


def server_session_key(variant_config, isl, osl):
"""Stable key for server reuse across sweep cells with identical model/shape."""
p = variant_config.params
roles = variant_config.roles.server
if p.driver == "atom":
Expand All @@ -326,7 +97,6 @@ def server_session_key(variant_config, isl, osl):


def expand_sweep_parametrize(sweep, fixturenames):
"""Build pytest parametrize args for inference or metric-tier collection."""
from cvs.lib.inference.atom.atom_parsing import METRIC_TIER_ORDER

cases, ids = expand_sweep(sweep)
Expand Down Expand Up @@ -364,9 +134,11 @@ def placeholder_gated_threshold_cell(
failed_max: int = 1_000_000_000,
success_rate_min: float = 0,
) -> dict[str, Any]:
"""Return one sweep cell's ``client.*`` specs covering every ``GATED_METRICS`` member."""
"""Return one sweep cell's ``client.*`` specs covering every gated metric."""
from cvs.lib.inference.atom.atom_parsing import GATED_METRICS

loose_ms = {"kind": "max_ms", "value": 1_000_000}
return {
out = {
"client.total_token_throughput": {"kind": "min_tok_s", "value": total_token_throughput_min},
"client.output_throughput": {"kind": "min_tok_s", "value": output_throughput_min},
"client.per_gpu_throughput": {"kind": "min_tok_s", "value": per_gpu_throughput_min},
Expand All @@ -393,10 +165,15 @@ def placeholder_gated_threshold_cell(
"client.success_rate": {"kind": "min", "value": success_rate_min},
"client.failed": {"kind": "max", "value": failed_max},
}
for m in GATED_METRICS:
key = f"client.{m}"
if key not in out:
kind = "max_ms" if m.endswith("_ms") else "max" if m == "failed" else "min"
out[key] = {"kind": kind, "value": 0 if kind == "min" else 1_000_000}
return out


def orchestrator_container_from_variant(variant: AtomVariantConfig) -> dict[str, Any]:
"""``container`` block for :class:`OrchestratorConfig` (includes server env)."""
block = variant.container.model_dump()
server_env = variant.roles.server.env
if server_env:
Expand Down
Loading
Loading