From df45bbff7fd52c7d171ebd25860e142a04f93421 Mon Sep 17 00:00:00 2001 From: Ivan Podkidyshev Date: Fri, 30 Oct 2026 21:17:55 +0100 Subject: [PATCH 1/3] Expose Dynamo phase inference reports and total-token metrics Signed-off-by: Ivan Podkidyshev --- doc/pareto-frontiers.rst | 59 +++- doc/reporting.rst | 5 +- doc/workloads/ai_dynamo.rst | 3 + src/cloudai/metrics.py | 25 ++ src/cloudai/workloads/ai_dynamo/ai_dynamo.py | 268 +++++++++++++- .../ai_dynamo/report_generation_strategy.py | 8 +- .../ai_dynamo/test_report_gen_strategy.py | 327 +++++++++++++++++- 7 files changed, 677 insertions(+), 18 deletions(-) diff --git a/doc/pareto-frontiers.rst b/doc/pareto-frontiers.rst index 7c4957f33..93539b0c9 100644 --- a/doc/pareto-frontiers.rst +++ b/doc/pareto-frontiers.rst @@ -3,7 +3,7 @@ Pareto Frontiers The ``pareto`` scenario report writes ``pareto.html`` in the scenario results directory. Each plot compares two metrics. Successful iterations and sweep steps supply the points: one test case can contribute several points, -and several cases can contribute to one frontier. +and several cases can contribute to one frontier. Each configured Dynamo AIPerf phase contributes separately. Point details show serving instances, nodes per instance, and TP/PP/DP within each instance. Disaggregated runs show separate prefill and decode layouts and whether they share nodes; combined serving uses the @@ -12,7 +12,8 @@ runs show separate prefill and decode layouts and whether they share nodes; comb Usage Example ------------- -For :doc:`vLLM ` or :doc:`SGLang `, add this block to your scenario TOML: +For :doc:`vLLM `, :doc:`SGLang `, or +:doc:`Dynamo `, add this block to your scenario TOML: .. code-block:: toml @@ -35,7 +36,7 @@ Available Metrics ----------------- Use the metric keys below in either ``x.metric`` or ``y.metric``. Each axis must select exactly one measurement -from each execution. ``match`` selects among measurements of the same metric, such as different statistics or +for each point. ``match`` selects among measurements of the same metric, such as different statistics or message sizes. Missing or ambiguous measurements are listed as exclusions in the report. Both ``vllm`` and ``sglang`` expose these metrics: @@ -74,6 +75,32 @@ For either metric, select ``mean``, ``median``, or ``p99`` with ``match = { stat metrics have no statistic selector. ``accuracy`` is available when semantic validation is configured and its result is present; see the workload's Semantic Validation section. +Dynamo exposes the same metrics except ``output_token_throughput_per_serving_gpu``. It reads each phase's +``profile_export_aiperf.json``, with CSV support for older AIPerf and GenAI-Perf results. Its ``tpot`` measurements +use Inter Token Latency. Missing statistics are omitted. + +AIPerf JSON exports also provide these Dynamo metrics, all maximized by default: + +.. list-table:: + :header-rows: 1 + :widths: 65 35 + + * - Metric + - Unit + * - ``total_token_throughput`` + - tokens/s + * - ``total_token_throughput_per_user`` + - tokens/s/user + * - ``total_token_throughput_per_serving_gpu`` + - tokens/s/GPU + +Total throughput includes input and output tokens, using AIPerf's ``total_token_throughput.avg``. Per-user +throughput divides it by the phase's configured ``args.users``, falling back to ``args.concurrency``. Per-serving-GPU +throughput divides it by the configured GPU count across prefill and decode, including worker replicas and +multi-node groups. SGLang attention DP shares TP devices. Omitted settings follow CloudAI's standard launcher +defaults; custom commands that change GPU placement are outside this calculation. Packing and hardware-derived +defaults use the system's ``gpus_per_node``, or ``CUDA_VISIBLE_DEVICES`` when configured for the test. + The two GPU-normalized metrics divide total output tokens/s by different counts: - ``output_token_throughput_per_allocated_gpu`` uses the execution's ``num_nodes`` times the system's @@ -94,6 +121,7 @@ For inference, start with one of these comparisons: - **Streaming latency vs serving efficiency:** ``tpot`` on X, ``output_token_throughput_per_serving_gpu`` on Y. - **Streaming latency vs allocated GPU efficiency:** ``tpot`` on X, ``output_token_throughput_per_allocated_gpu`` on Y. +- **Dynamo total-token efficiency:** ``ttft`` on X, ``total_token_throughput_per_serving_gpu`` on Y. Keep the model, input/output token lengths, precision, and traffic pattern comparable. Vary concurrency or serving parameters to measure the tradeoff. Use P99 for tail latency, or select mean or median when that is @@ -114,6 +142,31 @@ latency plot alongside the streaming-latency example above: y = { metric = "output_token_throughput" } frontiers = [{ name = "All configurations" }] +Dynamo Phases +~~~~~~~~~~~~~ + +Each AIPerf phase with recorded metrics contributes separately. For example, phases with concurrency ``2`` +and ``4`` produce two independent points per successful iteration or sweep step. Point details show all X and Y +metric dimensions alongside the phase's benchmark settings. Phases without observations contribute no points. +The final-phase copy in ``aiperf_report.csv`` is not an extra point. + +To compare phases separately, select their configured names with ``match.phase``: + +.. code-block:: toml + + [[reports.pareto.plots]] + name = "Throughput by phase" + x = { metric = "ttft", match = { statistic = "p99" } } + y = { metric = "output_token_throughput" } + frontiers = [ + { name = "Round 1", match = { phase = "round_1" } }, + { name = "Round 2", match = { phase = "round_2" } }, + ] + +When an execution runs both AIPerf and GenAI-Perf, their points also have a ``benchmark`` dimension +(``aiperf`` or ``genai_perf``). Use ``match.benchmark`` to select one. Optional accuracy results describe the +separate accuracy evaluation and are shared across that execution's phases. + Comparing Versions or Configurations ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ diff --git a/doc/reporting.rst b/doc/reporting.rst index d11a02c27..f79d0106b 100644 --- a/doc/reporting.rst +++ b/doc/reporting.rst @@ -188,13 +188,14 @@ settings, plus ``MetricObservation`` outputs with their units and dimensions. Mu same execution have separate reports, even when their metric names are identical. Unknown optional settings remain ``None``. -vLLM and SGLang return one report per execution. +vLLM and SGLang return one report per execution; Dynamo returns separate reports for its benchmark phases. The Pareto reporter treats each returned report as one candidate point and applies ``match`` to its metric observations. ``metric_observations()`` remains available and returns the combined observations from these reports. Each serving role records its instance count and assigned nodes per instance. An instance is one serving launch, which can span nodes; TP, PP, and DP describe parallelism within it. ``shared_nodes`` identifies overlapping -prefill/decode placement. A combined prefill/decode instance is displayed as ``Serving``. +prefill/decode placement. Dynamo resolves worker counts from its configured topology and standard launcher +defaults. A combined prefill/decode instance is displayed as ``Serving``. .. _reporting-configuration-implementation: diff --git a/doc/workloads/ai_dynamo.rst b/doc/workloads/ai_dynamo.rst index f3e474d23..fc9075eb1 100644 --- a/doc/workloads/ai_dynamo.rst +++ b/doc/workloads/ai_dynamo.rst @@ -213,6 +213,9 @@ LLM serving tables and charts for TTFT, TPOT, successful prompts, output-token t GPU. When ``aiperf_accuracy`` is configured and its result is available, the report also compares model accuracy. Both AIPerf and GenAI-Perf CSV output are supported. +The scenario-level :doc:`Pareto report <../pareto-frontiers>` uses each configured AIPerf phase as a separate +point, with that phase's concurrency and benchmark settings. It also supports GenAI-Perf results. + AIPerf args are rendered as normal CLI flags. Multi-value AIPerf options should be passed with AIPerf CLI syntax, such as ``server-metrics-formats = "csv,json,jsonl"`` or ``gpu-telemetry = "node1:9401,node2:9401"``. ``server-metrics = "auto"`` expands to the frontend metrics endpoint, Dynamo worker metrics endpoints, and any CloudAI-started DCGM diff --git a/src/cloudai/metrics.py b/src/cloudai/metrics.py index 059547bfc..7616a6ffb 100644 --- a/src/cloudai/metrics.py +++ b/src/cloudai/metrics.py @@ -106,6 +106,8 @@ def validate_dimensions(cls, values: Mapping[str, Any]) -> MetricDimensions: SOURCE_MEMORY = DimensionDefinition("source_memory", "Source memory", Annotated[str, Field(strict=True, min_length=1)]) TARGET_MEMORY = DimensionDefinition("target_memory", "Target memory", Annotated[str, Field(strict=True, min_length=1)]) STATISTIC = DimensionDefinition("statistic", "Statistic", Literal["mean", "median", "p99"]) +PHASE = DimensionDefinition("phase", "Phase", Annotated[str, Field(strict=True, min_length=1)]) +BENCHMARK = DimensionDefinition("benchmark", "Benchmark", Annotated[str, Field(strict=True, min_length=1)]) BANDWIDTH = MetricDefinition( key="bandwidth", @@ -144,6 +146,24 @@ def validate_dimensions(cls, values: Mapping[str, Any]) -> MetricDimensions: unit="tokens/s", direction=OptimizationDirection.MAXIMIZE, ) +TOTAL_TOKEN_THROUGHPUT = MetricDefinition( + key="total_token_throughput", + display_name="Total token throughput", + unit="tokens/s", + direction=OptimizationDirection.MAXIMIZE, +) +TOTAL_TOKEN_THROUGHPUT_PER_USER = MetricDefinition( + key="total_token_throughput_per_user", + display_name="Total throughput per configured user", + unit="tokens/s/user", + direction=OptimizationDirection.MAXIMIZE, +) +TOTAL_TOKEN_THROUGHPUT_PER_SERVING_GPU = MetricDefinition( + key="total_token_throughput_per_serving_gpu", + display_name="Total throughput per serving GPU", + unit="tokens/s/GPU", + direction=OptimizationDirection.MAXIMIZE, +) OUTPUT_TOKEN_THROUGHPUT_PER_ALLOCATED_GPU = MetricDefinition( key="output_token_throughput_per_allocated_gpu", display_name="Output throughput per allocated GPU", @@ -180,6 +200,9 @@ def validate_dimensions(cls, values: Mapping[str, Any]) -> MetricDimensions: LATENCY, REQUEST_THROUGHPUT, OUTPUT_TOKEN_THROUGHPUT, + TOTAL_TOKEN_THROUGHPUT, + TOTAL_TOKEN_THROUGHPUT_PER_USER, + TOTAL_TOKEN_THROUGHPUT_PER_SERVING_GPU, OUTPUT_TOKEN_THROUGHPUT_PER_ALLOCATED_GPU, OUTPUT_TOKEN_THROUGHPUT_PER_SERVING_GPU, TTFT, @@ -198,6 +221,8 @@ def validate_dimensions(cls, values: Mapping[str, Any]) -> MetricDimensions: SOURCE_MEMORY, TARGET_MEMORY, STATISTIC, + PHASE, + BENCHMARK, ) diff --git a/src/cloudai/workloads/ai_dynamo/ai_dynamo.py b/src/cloudai/workloads/ai_dynamo/ai_dynamo.py index c814ea387..3d52cc8ca 100644 --- a/src/cloudai/workloads/ai_dynamo/ai_dynamo.py +++ b/src/cloudai/workloads/ai_dynamo/ai_dynamo.py @@ -14,12 +14,16 @@ # See the License for the specific language governing permissions and # limitations under the License. +import ast +import copy import csv +import dataclasses +import json import logging import math import shlex from pathlib import Path -from typing import ClassVar, Literal, Mapping, Optional, cast +from typing import Any, ClassVar, Literal, Mapping, Optional, cast from pydantic import ( AliasChoices, @@ -30,6 +34,8 @@ model_validator, ) +import cloudai.metrics +import cloudai.report_generator.inference_report as inference_report import cloudai.util from cloudai.core import ( DockerImage, @@ -43,7 +49,7 @@ ) from cloudai.models.workload import CmdArgs, TestDefinition from cloudai.systems.slurm import SlurmSystem -from cloudai.workloads.common.llm_serving import LLMServingBenchReport +from cloudai.workloads.common import llm_serving AIPERF_ARTIFACTS_DIR = "aiperf_artifacts" AIPERF_ACCURACY_ARTIFACTS_DIR = "aiperf_accuracy_artifacts" @@ -187,6 +193,22 @@ def has_extra_arg(self, option: str) -> bool: token == option or token.startswith(f"{option}=") for value in values for token in shlex.split(value) ) + def inference_args(self) -> dict[str, Any]: + """Include numeric parallelism overrides appended to the backend command.""" + values = {key.replace("-", "_"): value for key, value in self.args.model_dump(exclude_none=True).items()} + extra = [self.extra_args] if isinstance(self.extra_args, str) else self.extra_args or [] + tokens = [token for value in extra for token in shlex.split(value)] + aliases = {"tp": "tensor_parallel_size", "pp": "pipeline_parallel_size", "dp": "data_parallel_size"} + values.update({full: values[f"{short}_size"] for short, full in aliases.items() if f"{short}_size" in values}) + for token, following in zip(tokens, [*tokens[1:], ""], strict=False): + name, separator, value = token.partition("=") + key = name.removeprefix("--").replace("-", "_") + if key in (*aliases, "tp_size", "pp_size", "dp_size", *aliases.values()): + value = value if separator else following + values[aliases.get(key.removesuffix("_size"), key)] = int(value) + values[key] = int(value) + return values + @model_validator(mode="after") def validate_worker_topology(self) -> "WorkerConfig": """Validate scalar worker topology while allowing DSE lists before unrolling.""" @@ -533,17 +555,19 @@ def installables(self) -> list[Installable]: ] -class AIDynamoBenchReport(LLMServingBenchReport): +class AIDynamoBenchReport(llm_serving.LLMServingBenchReport): """Normalized AI Dynamo benchmark results used by comparison reports.""" output_throughput: float + request_throughput: float = math.nan + total_token_throughput: float = math.nan @property def throughput(self) -> float: return self.output_throughput -class AIDynamoTestDefinition(TestDefinition): +class AIDynamoTestDefinition(TestDefinition, inference_report.InferenceReportProvider): """Test definition for AI Dynamo.""" model_config = ConfigDict(extra="forbid") @@ -564,6 +588,37 @@ class AIDynamoTestDefinition(TestDefinition): _THROUGHPUT_NAMES: ClassVar[tuple[str, ...]] = ("Output Token Throughput (tokens/sec)",) _REQUEST_COUNT_NAMES: ClassVar[tuple[str, ...]] = ("Request Count (count)", "Request Count", "Request count") + @staticmethod + def _json_metric(data: Mapping[str, Any], name: str, statistic: str = "avg") -> float: + try: + return float(data.get(name, {}).get(statistic, math.nan)) + except (TypeError, ValueError, AttributeError): + return math.nan + + @classmethod + def parse_aiperf_json(cls, path: Path) -> AIDynamoBenchReport | None: + """Read one phase's native export; absent statistics remain unavailable.""" + try: + data = json.loads(path.read_text()) + count = int(cls._json_metric(data, "request_count")) + latencies = { + f"{statistic}_{metric}_ms": cls._json_metric(data, source, field) + for metric, source in (("ttft", "time_to_first_token"), ("tpot", "inter_token_latency")) + for statistic, field in (("mean", "avg"), ("median", "p50"), ("p99", "p99")) + } + return AIDynamoBenchReport( + num_prompts=count, + completed=count, + max_concurrency=1, + request_throughput=cls._json_metric(data, "request_throughput"), + output_throughput=cls._json_metric(data, "output_token_throughput"), + total_token_throughput=cls._json_metric(data, "total_token_throughput"), + **latencies, + ) + except (OSError, ValueError, OverflowError): + logging.debug("Cannot parse AIPerf export at %s", path, exc_info=True) + return None + @staticmethod def _parse_metric_rows(csv_file: Path) -> dict[str, dict[str, str]]: """Parse both single-table GenAI-Perf and multi-table AIPerf CSV output.""" @@ -602,7 +657,7 @@ def _find_value( @classmethod def parse_benchmark_csv(cls, csv_file: Path, max_concurrency: int = 1) -> AIDynamoBenchReport | None: - """Normalize a benchmark CSV into the shared LLM serving result shape.""" + """Parse one benchmark or phase CSV without substituting another phase's output.""" try: metrics = cls._parse_metric_rows(csv_file) except (OSError, csv.Error) as e: @@ -612,10 +667,13 @@ def parse_benchmark_csv(cls, csv_file: Path, max_concurrency: int = 1) -> AIDyna ttft_mean = cls._find_value(metrics, cls._TTFT_NAMES, ("avg", "mean", "Value")) tpot_mean = cls._find_value(metrics, cls._TPOT_NAMES, ("avg", "mean", "Value")) throughput = cls._find_value(metrics, cls._THROUGHPUT_NAMES, ("avg", "Value")) - if ttft_mean is None or tpot_mean is None or throughput is None: + if ttft_mean is None and tpot_mean is None and throughput is None: return None request_count = cls._find_value(metrics, cls._REQUEST_COUNT_NAMES, ("avg", "Value")) + if request_count is not None and not math.isfinite(request_count): + return None + request_throughput = cls._find_value(metrics, ("Request Throughput (requests/sec)",), ("avg", "Value")) median_ttft = cls._find_value(metrics, cls._TTFT_NAMES, ("p50", "median")) p99_ttft = cls._find_value(metrics, cls._TTFT_NAMES, ("p99",)) median_tpot = cls._find_value(metrics, cls._TPOT_NAMES, ("p50", "median")) @@ -623,16 +681,208 @@ def parse_benchmark_csv(cls, csv_file: Path, max_concurrency: int = 1) -> AIDyna return AIDynamoBenchReport( num_prompts=int(request_count or 0), completed=int(request_count or 0), - mean_ttft_ms=ttft_mean, + mean_ttft_ms=ttft_mean if ttft_mean is not None else math.nan, median_ttft_ms=median_ttft if median_ttft is not None else math.nan, p99_ttft_ms=p99_ttft if p99_ttft is not None else math.nan, - mean_tpot_ms=tpot_mean, + mean_tpot_ms=tpot_mean if tpot_mean is not None else math.nan, median_tpot_ms=median_tpot if median_tpot is not None else math.nan, p99_tpot_ms=p99_tpot if p99_tpot is not None else math.nan, - output_throughput=throughput, + output_throughput=throughput if throughput is not None else math.nan, + request_throughput=request_throughput if request_throughput is not None else math.nan, max_concurrency=max_concurrency, ) + @model_validator(mode="before") + @classmethod + def restore_script_paths(cls, value: Any) -> Any: + """Unwrap script paths in saved TOMLs until the shared serialization fix (#5006455).""" + if not isinstance(value, dict): + return value + value = copy.deepcopy(value) + configs = [value] + cmd_args = value.get("cmd_args") + if isinstance(cmd_args, dict): + configs.extend(cmd_args.get(name) for name in ("genai_perf", "aiperf", "aiperf_accuracy")) + for config in configs: + script = config.get("script") if isinstance(config, dict) else None + if not isinstance(script, dict): + continue + for key in ("src", "_installed_path"): + path = script.get(key) + if isinstance(path, str) and path.startswith(("PosixPath(", "WindowsPath(")) and path.endswith(")"): + script[key] = ast.literal_eval(path.split("(", 1)[1][:-1]) + return value + + @model_validator(mode="before") + @classmethod + def restore_repository_version(cls, value: Any) -> Any: + """Read dumps from before the repository commit moved into cmd_args.""" + if isinstance(value, dict) and "repo" in value: + value = dict(value) + repo = value.pop("repo") + value["cmd_args"] = dict(value["cmd_args"]) + value["cmd_args"].setdefault("dynamo_version", repo["commit"]) + return value + + def metric_observations(self, system: System, tr: TestRun) -> list[cloudai.metrics.MetricObservation]: + return [obs for report in self.inference_reports(system, tr) for obs in report.observations] + + def inference_role_configs( + self, tr: TestRun, gpus_per_node: int | None + ) -> tuple[inference_report.InferenceRoleConfig | None, inference_report.InferenceRoleConfig | None, bool | None]: + prefill = self.cmd_args.dynamo.prefill_worker + decode = self.cmd_args.dynamo.decode_worker + shared_nodes = None + if prefill.num_nodes and decode.num_nodes: + shared_nodes = bool( + set((prefill.nodes or "").split(",")) & set((decode.nodes or "").split(",")) - {""} + ) or tr.nnodes < cast(int, prefill.num_nodes) + cast(int, decode.num_nodes) + distributed = self.cmd_args.dynamo.backend == "sglang_dsr1" + roles = [] + for worker in (prefill, decode): + if not worker.num_nodes: + roles.append(None) + continue + values = worker.inference_args() + nodes = cast(int, worker.num_nodes) + if distributed: + default_size = nodes * gpus_per_node if gpus_per_node else None + values["tensor_parallel_size"] = values.get("tp_size", default_size) + values["data_parallel_size"] = values.get("data_parallel_size", values.get("dp_size", default_size)) + values.setdefault("data_parallel_size", values.get("dp_size", 1)) + role = llm_serving.inference_role_config(values) + footprint = (role.tensor_parallel_size or 0) * (role.pipeline_parallel_size or 1) + count: int | None = 1 + if worker.multiple_workers_per_node and not shared_nodes and not distributed: + count = gpus_per_node // footprint if gpus_per_node and footprint else None + nodes_per_instance = nodes if distributed else cast(int, worker.nodes_per_worker or 1) + multinode = nodes_per_instance > 1 + roles.append( + dataclasses.replace( + role, + num_nodes=nodes, + nodes_per_instance=nodes_per_instance, + instances_per_node=None if multinode else count, + num_instances=nodes // nodes_per_instance + if multinode + else (nodes * count if count is not None else None), + ) + ) + return roles[0], roles[1], shared_nodes + + def serving_gpu_count(self, roles: tuple[inference_report.InferenceRoleConfig | None, ...]) -> int | None: + """Count configured engine devices, including replicas; attention DP shares TP GPUs.""" + total = 0 + for role, worker in zip( + roles, (self.cmd_args.dynamo.prefill_worker, self.cmd_args.dynamo.decode_worker), strict=True + ): + if role is None: + continue + if role.tensor_parallel_size is None or role.num_instances is None: + return None + dp_attention = ( + self.cmd_args.dynamo.backend == "sglang_dsr1" + or worker.has_extra_arg("--enable-dp-attention") + or worker.inference_args().get("enable_dp_attention", False) in (True, "") + ) + dp = 1 if self.cmd_args.dynamo.backend != "vllm" and dp_attention else role.data_parallel_size or 1 + total += role.tensor_parallel_size * (role.pipeline_parallel_size or 1) * dp * role.num_instances + return total or None + + def parse_benchmark(self, benchmark: Workload, output_path: Path) -> AIDynamoBenchReport | None: + csv_file = output_path / benchmark.report_name + if isinstance(benchmark, AIPerf): + artifacts = output_path / benchmark.artifact_dir_name + results = self.parse_aiperf_json(artifacts / "profile_export_aiperf.json") + if results is not None: + return results + if (artifacts / "profile_export_aiperf.csv").is_file(): + csv_file = artifacts / "profile_export_aiperf.csv" + return self.parse_benchmark_csv(csv_file) + + @staticmethod + def benchmark_observations( + results: AIDynamoBenchReport | None, + accuracy: float | None, + allocated_gpus: int | None, + serving_gpus: int | None, + users: Any, + ) -> list[cloudai.metrics.MetricObservation]: + observations = llm_serving.llm_serving_metric_observations( + results, accuracy, allocated_gpu_count=allocated_gpus + ) + if results is None or results.completed <= 0 or not math.isfinite(results.total_token_throughput): + return observations + for metric, divisor in ( + (cloudai.metrics.TOTAL_TOKEN_THROUGHPUT, 1), + (cloudai.metrics.TOTAL_TOKEN_THROUGHPUT_PER_USER, users), + (cloudai.metrics.TOTAL_TOKEN_THROUGHPUT_PER_SERVING_GPU, serving_gpus), + ): + try: + divisor = float(divisor) if divisor is not None else math.nan + except (TypeError, ValueError): + continue + if math.isfinite(divisor) and divisor > 0: + observations.append( + cloudai.metrics.MetricObservation(metric, results.total_token_throughput / divisor, {}) + ) + return observations + + def inference_reports(self, system: System, tr: TestRun) -> list[inference_report.InferenceReport]: + reports = [] + gpus_per_node = getattr(system, "gpus_per_node", None) + allocated_gpus = tr.nnodes * gpus_per_node if gpus_per_node else None + accuracy = parse_aiperf_accuracy(tr.output_path) if self.cmd_args.aiperf_accuracy else None + visible_gpus = self.extra_env_vars.get("CUDA_VISIBLE_DEVICES") + available_gpus = len(str(visible_gpus).split(",")) if visible_gpus is not None else gpus_per_node + prefill, decode, shared_nodes = self.inference_role_configs(tr, available_gpus) + serving_gpus = self.serving_gpu_count((prefill, decode)) + for script in self.cmd_args.workloads_list: + benchmark = self.get_workload_map()[script] + phases = self.cmd_args.aiperf_phases if benchmark is self.cmd_args.aiperf else None + for phase in phases or [None]: + resolved = self.cmd_args.resolve_aiperf_phase(phase) if phase else benchmark + dimensions = {} + if len(self.cmd_args.workloads_list) > 1: + dimensions[cloudai.metrics.BENCHMARK.key] = ( + "aiperf" if benchmark is self.cmd_args.aiperf else "genai_perf" + ) + if phase: + dimensions[cloudai.metrics.PHASE.key] = phase.name + results = self.parse_benchmark(resolved, tr.output_path) + args = { + key.replace("-", "_"): value for key, value in resolved.args.model_dump(exclude_none=True).items() + } + users = args.get("users", args.get("concurrency")) + measured = self.benchmark_observations(results, accuracy, allocated_gpus, serving_gpus, users) + observations = [ + dataclasses.replace(observation, dimensions={**dimensions, **observation.dimensions}) + for observation in measured + ] + if not observations: + continue + config = inference_report.InferenceConfig( + model=self.cmd_args.dynamo.model, + backend=self.cmd_args.dynamo.backend, + benchmark="aiperf" if isinstance(resolved, AIPerf) else "genai_perf", + phase=phase.name if phase else None, + num_nodes=tr.nnodes, + gpus_per_node=gpus_per_node, + serving_gpu_count=serving_gpus, + container_image=self.cmd_args.docker_image_url, + input_tokens=args.get("synthetic_input_tokens_mean", args.get("input_sequence_length")), + output_tokens=args.get("output_tokens_mean", args.get("output_sequence_length")), + concurrency=args.get("concurrency"), + users=args.get("users"), + request_count=args.get("request_count"), + request_rate=args.get("request_rate"), + prefill=prefill, + decode=decode, + shared_nodes=shared_nodes, + ) + reports.append(inference_report.InferenceReport(config=config, observations=observations)) + return reports + @model_validator(mode="after") def workload_scripts(self) -> "AIDynamoTestDefinition": """Populate prefill/decode args.""" diff --git a/src/cloudai/workloads/ai_dynamo/report_generation_strategy.py b/src/cloudai/workloads/ai_dynamo/report_generation_strategy.py index 11ddb6db7..9cc6b73ea 100644 --- a/src/cloudai/workloads/ai_dynamo/report_generation_strategy.py +++ b/src/cloudai/workloads/ai_dynamo/report_generation_strategy.py @@ -17,6 +17,7 @@ from __future__ import annotations import logging +import math from pathlib import Path from typing import ClassVar @@ -75,7 +76,12 @@ def parse_results(self) -> AIDynamoBenchReport | None: csv_file = self._benchmark_csv() if csv_file is None: return None - return AIDynamoTestDefinition.parse_benchmark_csv(csv_file, self._max_concurrency()) + results = AIDynamoTestDefinition.parse_benchmark_csv(csv_file, self._max_concurrency()) + if results is None or any( + math.isnan(value) for value in (results.mean_ttft_ms, results.mean_tpot_ms, results.output_throughput) + ): + return None + return results def used_gpus_count(self) -> int: """Match the GPU accounting used when AI Dynamo produces its result CSV.""" diff --git a/tests/workloads/ai_dynamo/test_report_gen_strategy.py b/tests/workloads/ai_dynamo/test_report_gen_strategy.py index 0402dd609..14fb58ef4 100644 --- a/tests/workloads/ai_dynamo/test_report_gen_strategy.py +++ b/tests/workloads/ai_dynamo/test_report_gen_strategy.py @@ -14,10 +14,16 @@ # See the License for the specific language governing permissions and # limitations under the License. +import json from pathlib import Path import pytest +import cloudai.core +import cloudai.metrics +import cloudai.report_generator.inference_report as inference_report +import cloudai.systems.slurm +import cloudai.util from cloudai import TestRun from cloudai.core import METRIC_ERROR from cloudai.systems.slurm.slurm_system import SlurmSystem @@ -45,6 +51,9 @@ def get_csv_content() -> str: "Inter Token Latency (ms),12.34,23.45,34.56,45.67,56.78,67.89,78.90,89.01,90.12\n" "Output Sequence Length (tokens),101.01,202.02,303.03,404.04,505.05,606.06,707.07,808.08,909.09\n" "Input Sequence Length (tokens),123.45,234.56,345.67,456.78,567.89,678.90,789.01,890.12,901.23\n" + "Output Token Throughput (tokens/sec),200\n" + "Request Throughput (requests/sec),4\n" + "Request Count (count),50\n" ) @@ -59,6 +68,7 @@ def get_aiperf_csv_content() -> str: "Output Token Throughput (tokens/sec),595.68\n" "Total Token Throughput (tokens/sec),954.47\n" "Request Count,50.00\n" + "Request Throughput (requests/sec),1.25\n" ) @@ -208,9 +218,36 @@ def test_ai_dynamo_get_metric_genai_perf(slurm_system: SlurmSystem, ai_dynamo_tr # Explicit prefix also works. assert strategy.get_metric("genai_perf:Time To First Token (ms):avg") == 111.12 assert strategy.get_metric("genai_perf:Inter Token Latency (ms):p50") == 89.01 - - -def test_ai_dynamo_get_metric_aiperf(slurm_system: SlurmSystem, ai_dynamo_aiperf_tr: TestRun) -> None: + observations = ai_dynamo_tr.test.metric_observations(slurm_system, ai_dynamo_tr) + expected = [ + ("request_throughput", 4, {}), + ("output_token_throughput", 200, {}), + ("output_token_throughput_per_allocated_gpu", 200 / 8, {}), + ("ttft", 111.12, {"statistic": "mean"}), + ("ttft", 888.88, {"statistic": "median"}), + ("ttft", 444.44, {"statistic": "p99"}), + ("tpot", 12.34, {"statistic": "mean"}), + ("tpot", 89.01, {"statistic": "median"}), + ("tpot", 45.67, {"statistic": "p99"}), + ] + assert [(obs.metric.key, obs.value, dict(obs.dimensions)) for obs in observations] == expected + ai_dynamo_tr.test.cmd_args.workloads = "genai_perf.sh,aiperf.sh" + mixed = ai_dynamo_tr.test.metric_observations(slurm_system, ai_dynamo_tr) + assert [(obs.metric.key, obs.value, dict(obs.dimensions)) for obs in mixed] == [ + (key, value, {"benchmark": "genai_perf", **dimensions}) for key, value, dimensions in expected + ] + [ + ("request_throughput", 1.25, {"benchmark": "aiperf"}), + ("output_token_throughput", 595.68, {"benchmark": "aiperf"}), + ("output_token_throughput_per_allocated_gpu", 595.68 / 8, {"benchmark": "aiperf"}), + ("ttft", 49.87, {"benchmark": "aiperf", "statistic": "mean"}), + ("tpot", 2.83, {"benchmark": "aiperf", "statistic": "mean"}), + ] + + +@pytest.mark.parametrize( + "result", ["valid", "missing", "malformed", "nonfinite", "empty", "no-gpu-count", "missing-latency"] +) +def test_ai_dynamo_get_metric_aiperf(slurm_system: SlurmSystem, ai_dynamo_aiperf_tr: TestRun, result: str) -> None: strategy = AIDynamoReportGenerationStrategy(slurm_system, ai_dynamo_aiperf_tr) # aiperf fixture uses workloads="aiperf.sh" — bare names resolve to aiperf_report.csv. @@ -222,6 +259,31 @@ def test_ai_dynamo_get_metric_aiperf(slurm_system: SlurmSystem, ai_dynamo_aiperf assert strategy.get_metric("aiperf:Time to First Token (ms):avg") == 49.87 assert strategy.get_metric("aiperf:Output Token Throughput (tokens/sec):avg") == 595.68 assert strategy.get_metric("aiperf:Total Token Throughput (tokens/sec):avg") == 954.47 + path = ai_dynamo_aiperf_tr.output_path / "aiperf_report.csv" + if result == "missing": + path.unlink() + elif result == "malformed": + path.write_text("Invalid CSV\n") + elif result == "nonfinite": + path.write_text(path.read_text().replace("595.68", "nan")) + elif result == "empty": + path.write_text(path.read_text().replace("Request Count,50.00", "Request Count,0")) + elif result == "no-gpu-count": + slurm_system.gpus_per_node = None + elif result == "missing-latency": + path.write_text(path.read_text().replace("Inter Token Latency (ms),2.83,2.78,2.91\n", "")) + assert strategy.parse_results() is None + expected = [] + if result not in {"missing", "malformed", "empty"}: + expected.append(("request_throughput", 1.25, {})) + if result != "nonfinite": + expected.append(("output_token_throughput", 595.68, {})) + if result != "no-gpu-count": + expected.append(("output_token_throughput_per_allocated_gpu", 595.68 / 8, {})) + expected.append(("ttft", 49.87, {"statistic": "mean"})) + expected.extend([("tpot", 2.83, {"statistic": "mean"})] if result != "missing-latency" else []) + observations = ai_dynamo_aiperf_tr.test.metric_observations(slurm_system, ai_dynamo_aiperf_tr) + assert [(obs.metric.key, obs.value, dict(obs.dimensions)) for obs in observations] == expected def test_ai_dynamo_get_metric_split_aiperf_accuracy( @@ -231,6 +293,11 @@ def test_ai_dynamo_get_metric_split_aiperf_accuracy( assert strategy.get_metric("accuracy") == 0.35 assert strategy.get_metric("Inter Token Latency (ms)") == 2.83 + assert cloudai.metrics.MetricObservation(cloudai.metrics.ACCURACY, 0.35, {}) in ( + ai_dynamo_aiperf_with_split_accuracy_tr.test.metric_observations( + slurm_system, ai_dynamo_aiperf_with_split_accuracy_tr + ) + ) def test_ai_dynamo_accuracy_metric_requires_aiperf_accuracy_config( @@ -330,3 +397,257 @@ def test_parse_aiperf_accuracy_missing_or_invalid(tmp_path: Path) -> None: (tmp_path / "accuracy_results.csv").write_text("Task,Correct,Total,Accuracy\nOVERALL,n/a,100,n/a\n") assert parse_aiperf_accuracy(tmp_path) is None + + +INFERENCE_WORKER = { + "num_nodes": 2, + "cmd": "python -m dynamo.vllm", + "worker_initialized_regex": "ready", + "args": { + "tensor_parallel_size": 2, + "pipeline_parallel_size": 1, + "data_parallel_size": 1, + "enable_expert_parallel": True, + "dtype": "float16", + "quantization": "fp8", + }, +} +INFERENCE_DEFINITION = { + "name": "inference", + "description": "Inference report", + "test_template_name": "AIDynamo", + "cmd_args": { + "docker_image_url": "example/image:latest", + "workloads": "aiperf.sh", + "dynamo": { + "model": "example/model", + "backend": "vllm", + "prefill_worker": INFERENCE_WORKER, + "decode_worker": cloudai.util.deep_merge(INFERENCE_WORKER, {"args": {"pipeline_parallel_size": 2}}), + }, + "aiperf": { + "args": { + "synthetic-input-tokens-mean": 128, + "output-tokens-mean": 64, + "request-count": 20, + "concurrency": 4, + "request-rate": 12.5, + } + }, + "aiperf_phases": [ + {"name": "low", "args": {"concurrency": 8}}, + {"name": "high", "args": {"concurrency": 32, "users": 16}}, + ], + }, +} +INFERENCE_ROLE = dict( + model="example/model", + num_nodes=2, + num_instances=2, + nodes_per_instance=1, + instances_per_node=1, + tensor_parallel_size=2, + pipeline_parallel_size=1, + data_parallel_size=1, + expert_parallel=True, + precision="float16", + quantization="fp8", +) + + +@pytest.mark.parametrize( + "worker_overrides,backend,nodes,hardware,environment,role_overrides,decode_overrides,serving_gpus,shared_nodes", + [ + pytest.param({}, "vllm", 4, 8, {}, {}, {}, 12, False, id="replicas"), + pytest.param( + {"multiple_workers_per_node": True}, + "vllm", + 4, + 8, + {}, + {"num_instances": 8, "instances_per_node": 4}, + {"num_instances": 4, "instances_per_node": 2}, + 32, + False, + id="packing", + ), + pytest.param({"multiple_workers_per_node": True}, "vllm", 2, 8, {}, {}, {}, 12, True, id="shared-nodes"), + pytest.param( + {"nodes_per_worker": 2}, + "vllm", + 4, + 8, + {}, + {"num_instances": 1, "nodes_per_instance": 2, "instances_per_node": None}, + {"num_instances": 1, "nodes_per_instance": 2, "instances_per_node": None}, + 6, + False, + id="multi-node-workers", + ), + pytest.param( + {"args": {"data_parallel_size": 2}, "extra_args": "--enable-dp-attention"}, + "sglang", + 4, + 8, + {}, + {"data_parallel_size": 2}, + {"data_parallel_size": 2}, + 12, + False, + id="attention-dp", + ), + pytest.param( + {"args": {"data_parallel_size": None, "pipeline_parallel_size": 1}}, + "sglang_dsr1", + 4, + 8, + {}, + { + "tensor_parallel_size": 16, + "data_parallel_size": 16, + "num_instances": 1, + "nodes_per_instance": 2, + "instances_per_node": None, + }, + { + "tensor_parallel_size": 16, + "pipeline_parallel_size": 1, + "data_parallel_size": 16, + "num_instances": 1, + "nodes_per_instance": 2, + "instances_per_node": None, + }, + 32, + False, + id="distributed-defaults", + ), + pytest.param( + {"multiple_workers_per_node": True}, + "vllm", + 4, + None, + {}, + {"num_instances": None, "instances_per_node": None}, + {"num_instances": None, "instances_per_node": None}, + None, + False, + id="unknown-hardware", + ), + pytest.param( + {"multiple_workers_per_node": True}, + "vllm", + 4, + 8, + {"CUDA_VISIBLE_DEVICES": "0,1,2,3"}, + {"num_instances": 4, "instances_per_node": 2}, + {}, + 16, + False, + id="visible-gpus", + ), + pytest.param( + {"extra_args": "--pipeline-parallel-size=1 --data-parallel-size 2"}, + "vllm", + 4, + 8, + {}, + {"data_parallel_size": 2}, + {"pipeline_parallel_size": 1, "data_parallel_size": 2}, + 16, + False, + id="argument-overrides", + ), + ], +) +def test_inference_reports( + slurm_system: SlurmSystem, + tmp_path: Path, + worker_overrides: dict, + backend: str, + nodes: int, + hardware: int | None, + environment: dict, + role_overrides: dict, + decode_overrides: dict, + serving_gpus: int | None, + shared_nodes: bool, +) -> None: + definition = cloudai.util.deep_merge( + INFERENCE_DEFINITION, + { + "cmd_args": { + "dynamo": {"backend": backend, "prefill_worker": worker_overrides, "decode_worker": worker_overrides} + }, + "extra_env_vars": environment, + }, + ) + test = AIDynamoTestDefinition.model_validate(definition) + tr = TestRun("inference", test, nodes, [], output_path=tmp_path) + slurm_system.gpus_per_node = hardware + expected = [] + for phase, concurrency, users, total in [("low", 8, None, 1500), ("high", 32, 16, 3000)]: + artifact = tmp_path / "aiperf_artifacts" / phase + artifact.mkdir(parents=True) + (artifact / "profile_export_aiperf.json").write_text( + json.dumps( + { + "request_count": {"avg": 20}, + "request_throughput": {"avg": 10}, + "output_token_throughput": {"avg": 1000}, + "total_token_throughput": {"avg": total}, + "time_to_first_token": {"avg": 10, "p50": 11, "p99": 12}, + "inter_token_latency": {"avg": 1, "p50": 2, "p99": 3}, + } + ) + ) + (tmp_path / f"aiperf_{phase}_report.csv").write_text(get_aiperf_csv_content()) + metrics: list[tuple[cloudai.metrics.MetricDefinition, float, dict[str, str]]] = [ + (cloudai.metrics.REQUEST_THROUGHPUT, 10, {}), + (cloudai.metrics.OUTPUT_TOKEN_THROUGHPUT, 1000, {}), + ] + if hardware is not None: + metrics.append((cloudai.metrics.OUTPUT_TOKEN_THROUGHPUT_PER_ALLOCATED_GPU, 1000 / (nodes * 8), {})) + metrics.extend( + (metric, value, {"statistic": statistic}) + for metric, values in ((cloudai.metrics.TTFT, (10, 11, 12)), (cloudai.metrics.TPOT, (1, 2, 3))) + for statistic, value in zip(("mean", "median", "p99"), values, strict=True) + ) + metrics.extend( + [ + (cloudai.metrics.TOTAL_TOKEN_THROUGHPUT, total, {}), + (cloudai.metrics.TOTAL_TOKEN_THROUGHPUT_PER_USER, total / (users or concurrency), {}), + ] + ) + if serving_gpus is not None: + metrics.append((cloudai.metrics.TOTAL_TOKEN_THROUGHPUT_PER_SERVING_GPU, total / serving_gpus, {})) + expected.append( + inference_report.InferenceReport( + config=inference_report.InferenceConfig( + model="example/model", + num_nodes=nodes, + backend=backend, + benchmark="aiperf", + phase=phase, + container_image="example/image:latest", + gpus_per_node=hardware, + serving_gpu_count=serving_gpus, + input_tokens=128, + output_tokens=64, + concurrency=concurrency, + users=users, + request_count=20, + request_rate=12.5, + prefill=inference_report.InferenceRoleConfig(**{**INFERENCE_ROLE, **role_overrides}), + decode=inference_report.InferenceRoleConfig( + **{**INFERENCE_ROLE, "pipeline_parallel_size": 2, **decode_overrides} + ), + shared_nodes=shared_nodes, + ), + observations=[ + cloudai.metrics.MetricObservation(metric, value, {"phase": phase, **dimensions}) + for metric, value, dimensions in metrics + ], + ) + ) + assert test.inference_reports(slurm_system, tr) == expected + assert test.metric_observations(slurm_system, tr) == [obs for report in expected for obs in report.observations] From db7afddf91c68844e11c3e79e9eeb5e0e557ae6a Mon Sep 17 00:00:00 2001 From: Ivan Podkidyshev Date: Fri, 30 Oct 2026 21:39:15 +0100 Subject: [PATCH 2/3] Preserve Dynamo metrics with unresolved parallelism arguments Signed-off-by: Ivan Podkidyshev --- src/cloudai/workloads/ai_dynamo/ai_dynamo.py | 20 +++++++++++++++---- .../ai_dynamo/test_report_gen_strategy.py | 14 ++++++++++++- 2 files changed, 29 insertions(+), 5 deletions(-) diff --git a/src/cloudai/workloads/ai_dynamo/ai_dynamo.py b/src/cloudai/workloads/ai_dynamo/ai_dynamo.py index 3d52cc8ca..278ec59a6 100644 --- a/src/cloudai/workloads/ai_dynamo/ai_dynamo.py +++ b/src/cloudai/workloads/ai_dynamo/ai_dynamo.py @@ -201,12 +201,19 @@ def inference_args(self) -> dict[str, Any]: aliases = {"tp": "tensor_parallel_size", "pp": "pipeline_parallel_size", "dp": "data_parallel_size"} values.update({full: values[f"{short}_size"] for short, full in aliases.items() if f"{short}_size" in values}) for token, following in zip(tokens, [*tokens[1:], ""], strict=False): + if not token.startswith("--"): + continue name, separator, value = token.partition("=") key = name.removeprefix("--").replace("-", "_") if key in (*aliases, "tp_size", "pp_size", "dp_size", *aliases.values()): value = value if separator else following - values[aliases.get(key.removesuffix("_size"), key)] = int(value) - values[key] = int(value) + canonical = aliases.get(key.removesuffix("_size"), key) + try: + number = int(value) + except ValueError: + number = None + short = next(short for short, full in aliases.items() if full == canonical) + values.update({canonical: number, key: number, f"{short}_size": number}) return values @model_validator(mode="after") @@ -751,7 +758,7 @@ def inference_role_configs( values["data_parallel_size"] = values.get("data_parallel_size", values.get("dp_size", default_size)) values.setdefault("data_parallel_size", values.get("dp_size", 1)) role = llm_serving.inference_role_config(values) - footprint = (role.tensor_parallel_size or 0) * (role.pipeline_parallel_size or 1) + footprint = (role.tensor_parallel_size or 0) * (role.pipeline_parallel_size or 0) count: int | None = 1 if worker.multiple_workers_per_node and not shared_nodes and not distributed: count = gpus_per_node // footprint if gpus_per_node and footprint else None @@ -778,7 +785,12 @@ def serving_gpu_count(self, roles: tuple[inference_report.InferenceRoleConfig | ): if role is None: continue - if role.tensor_parallel_size is None or role.num_instances is None: + if ( + role.tensor_parallel_size is None + or role.pipeline_parallel_size is None + or role.data_parallel_size is None + or role.num_instances is None + ): return None dp_attention = ( self.cmd_args.dynamo.backend == "sglang_dsr1" diff --git a/tests/workloads/ai_dynamo/test_report_gen_strategy.py b/tests/workloads/ai_dynamo/test_report_gen_strategy.py index 14fb58ef4..1ea6d0a3a 100644 --- a/tests/workloads/ai_dynamo/test_report_gen_strategy.py +++ b/tests/workloads/ai_dynamo/test_report_gen_strategy.py @@ -546,7 +546,7 @@ def test_parse_aiperf_accuracy_missing_or_invalid(tmp_path: Path) -> None: id="visible-gpus", ), pytest.param( - {"extra_args": "--pipeline-parallel-size=1 --data-parallel-size 2"}, + {"extra_args": "--served-model-name tp --pipeline-parallel-size=1 --data-parallel-size 2"}, "vllm", 4, 8, @@ -557,6 +557,18 @@ def test_parse_aiperf_accuracy_missing_or_invalid(tmp_path: Path) -> None: False, id="argument-overrides", ), + pytest.param( + {"extra_args": "--tensor-parallel-size ${TP} --pipeline-parallel-size ${PP} --data-parallel-size"}, + "vllm", + 4, + 8, + {}, + {"tensor_parallel_size": None, "pipeline_parallel_size": None, "data_parallel_size": None}, + {"tensor_parallel_size": None, "pipeline_parallel_size": None, "data_parallel_size": None}, + None, + False, + id="unresolved-parallelism-overrides", + ), ], ) def test_inference_reports( From 1784c713e396831a844e5548561f56fa46f938de Mon Sep 17 00:00:00 2001 From: Ivan Podkidyshev Date: Fri, 30 Oct 2026 21:43:51 +0100 Subject: [PATCH 3/3] Remove redundant module imports from Dynamo report tests Signed-off-by: Ivan Podkidyshev --- tests/workloads/ai_dynamo/test_report_gen_strategy.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/tests/workloads/ai_dynamo/test_report_gen_strategy.py b/tests/workloads/ai_dynamo/test_report_gen_strategy.py index 1ea6d0a3a..0f42b51aa 100644 --- a/tests/workloads/ai_dynamo/test_report_gen_strategy.py +++ b/tests/workloads/ai_dynamo/test_report_gen_strategy.py @@ -19,10 +19,8 @@ import pytest -import cloudai.core import cloudai.metrics import cloudai.report_generator.inference_report as inference_report -import cloudai.systems.slurm import cloudai.util from cloudai import TestRun from cloudai.core import METRIC_ERROR