- {{ point.group }}
{{ point.dimensions }} |
+ {{ point.group }}
{{ point.dimensions }} |
{{ point.status }} |
{{ format_number(point.x) }} | {{ format_number(point.y) }} |
View configuration
diff --git a/src/cloudai/workloads/common/llm_serving.py b/src/cloudai/workloads/common/llm_serving.py
index d8c27e7d6..56955855a 100644
--- a/src/cloudai/workloads/common/llm_serving.py
+++ b/src/cloudai/workloads/common/llm_serving.py
@@ -16,6 +16,7 @@
from __future__ import annotations
+import dataclasses
import math
import re
import shlex
@@ -29,7 +30,8 @@
from typing_extensions import Self
import cloudai.metrics
-from cloudai.core import METRIC_ERROR, DockerImage, HFModel, Installable, MetricValue, ReportGenerationStrategy
+import cloudai.report_generator.inference_report as inference_report
+from cloudai.core import METRIC_ERROR, DockerImage, HFModel, Installable, MetricValue, ReportGenerationStrategy, TestRun
from cloudai.models.workload import CmdArgs, TestDefinition
from cloudai.systems.slurm import SlurmCommandGenStrategy
@@ -109,6 +111,58 @@ def calculate_decode_gpu_ids(
return gpu_ids[mid:]
+def disaggregated_role_node_counts(
+ tdef: LLMServingTestDefinition[Any], num_nodes: int, workload_name: str
+) -> tuple[int, int]:
+ """Resolve the same serving-role node split for launch and reporting."""
+ if tdef.cmd_args.prefill is None:
+ return (0, 0)
+
+ prefill_nodes = tdef.cmd_args.prefill.num_nodes
+ decode_nodes = tdef.cmd_args.decode.num_nodes
+ for role, count in (("prefill", prefill_nodes), ("decode", decode_nodes)):
+ if isinstance(count, list):
+ raise ValueError(f"{role}.num_nodes must be a single integer for command generation.")
+
+ prefill_nodes = cast(int | None, prefill_nodes)
+ decode_nodes = cast(int | None, decode_nodes)
+ if prefill_nodes is None and decode_nodes is None:
+ if num_nodes in (1, 2):
+ return (1, 1)
+ raise ValueError(
+ f"Disaggregated {workload_name} over more than 2 nodes requires both "
+ "prefill.num_nodes and decode.num_nodes."
+ )
+ if prefill_nodes is None or decode_nodes is None:
+ raise ValueError("Both prefill.num_nodes and decode.num_nodes must be set or both must be omitted.")
+ if prefill_nodes <= 0 or decode_nodes <= 0:
+ raise ValueError("prefill.num_nodes and decode.num_nodes must be positive integers.")
+ if prefill_nodes + decode_nodes != num_nodes:
+ raise ValueError(
+ f"prefill.num_nodes + decode.num_nodes must equal allocated nodes ({num_nodes}), "
+ f"got {prefill_nodes + decode_nodes}."
+ )
+ return (prefill_nodes, decode_nodes)
+
+
+def inference_role_config(values: dict[str, Any]) -> inference_report.InferenceRoleConfig:
+ """Normalize serving-role argument names shared by inference workloads."""
+ values = {key.replace("-", "_"): value for key, value in values.items()}
+ gpu_ids = values.get("gpu_ids")
+ expert_parallel = values.get("enable_expert_parallel")
+ return inference_report.InferenceRoleConfig(
+ model=values.get("model", values.get("model_path")),
+ num_nodes=values.get("num_nodes"),
+ gpu_ids=parse_gpu_ids(gpu_ids) if gpu_ids else None,
+ tensor_parallel_size=values.get("tensor_parallel_size", values.get("tp_size", values.get("tp"))),
+ pipeline_parallel_size=values.get("pipeline_parallel_size", values.get("pp_size", values.get("pp"))),
+ data_parallel_size=values.get("data_parallel_size", values.get("dp_size", values.get("dp"))),
+ expert_parallel=expert_parallel in (True, "") if expert_parallel is not None else None,
+ precision=values.get("dtype"),
+ quantization=values.get("quantization"),
+ )
+
+
class LLMServingArgs(CmdArgs):
"""Shared serve-argument serialization for LLM serving workloads."""
@@ -201,6 +255,37 @@ def serving_gpu_count(self) -> int | None:
counts = [role.serving_gpu_count for role in roles]
return sum(count for count in counts if count is not None) if all(counts) else None
+ def inference_role_configs(
+ self, tr: TestRun
+ ) -> tuple[inference_report.InferenceRoleConfig | None, inference_report.InferenceRoleConfig]:
+ try:
+ prefill_nodes, decode_nodes = (
+ disaggregated_role_node_counts(self, tr.nnodes, self.test_template_name)
+ if self.cmd_args.prefill is not None
+ else (0, tr.nnodes)
+ )
+ except ValueError:
+ return (
+ inference_role_config(self.cmd_args.prefill.model_dump(exclude_none=True))
+ if self.cmd_args.prefill is not None
+ else None,
+ inference_role_config(self.cmd_args.decode.model_dump(exclude_none=True)),
+ )
+ roles = []
+ for args, nodes in ((self.cmd_args.prefill, prefill_nodes), (self.cmd_args.decode, decode_nodes)):
+ roles.append(
+ dataclasses.replace(
+ inference_role_config(args.model_dump(exclude_none=True)),
+ num_nodes=nodes,
+ num_instances=1,
+ nodes_per_instance=nodes,
+ instances_per_node=1 if nodes == 1 else None,
+ )
+ if args is not None
+ else None
+ )
+ return roles[0], cast(inference_report.InferenceRoleConfig, roles[1])
+
@property
def docker_image(self) -> DockerImage:
if not self._docker_image:
@@ -463,30 +548,10 @@ def aggregated_node_count(self) -> int:
return num_nodes
def disaggregated_role_node_counts(self) -> tuple[int, int]:
- if not self.is_disaggregated or self.tdef.cmd_args.prefill is None:
+ if not self.is_disaggregated:
return (0, 0)
-
num_nodes, _ = self.get_cached_nodes_spec()
- prefill_nodes = self._role_num_nodes(self.tdef.cmd_args.prefill.num_nodes, "prefill")
- decode_nodes = self._role_num_nodes(self.tdef.cmd_args.decode.num_nodes, "decode")
-
- if prefill_nodes is None and decode_nodes is None:
- if num_nodes in (1, 2):
- return (1, 1)
- raise ValueError(
- f"Disaggregated {self.workload_name} over more than 2 nodes requires both "
- "prefill.num_nodes and decode.num_nodes."
- )
- if prefill_nodes is None or decode_nodes is None:
- raise ValueError("Both prefill.num_nodes and decode.num_nodes must be set or both must be omitted.")
- if prefill_nodes <= 0 or decode_nodes <= 0:
- raise ValueError("prefill.num_nodes and decode.num_nodes must be positive integers.")
- if prefill_nodes + decode_nodes != num_nodes:
- raise ValueError(
- f"prefill.num_nodes + decode.num_nodes must equal allocated nodes ({num_nodes}), "
- f"got {prefill_nodes + decode_nodes}."
- )
- return (prefill_nodes, decode_nodes)
+ return disaggregated_role_node_counts(self.tdef, num_nodes, self.workload_name)
def role_node_count(self, role: str) -> int:
if role == "serve":
diff --git a/src/cloudai/workloads/sglang/report_generation_strategy.py b/src/cloudai/workloads/sglang/report_generation_strategy.py
index 00c58d83f..16a72216b 100644
--- a/src/cloudai/workloads/sglang/report_generation_strategy.py
+++ b/src/cloudai/workloads/sglang/report_generation_strategy.py
@@ -23,7 +23,6 @@
SGLANG_SEMANTIC_EVAL_LOG_FILE,
SGLangBenchReport,
SglangTestDefinition,
- parse_sglang_bench_output,
parse_sglang_semantic_accuracy,
)
@@ -40,7 +39,7 @@ def report_title(self) -> str:
return "SGLang Benchmark Results"
def parse_output(self, path: Path) -> SGLangBenchReport | None:
- return parse_sglang_bench_output(path)
+ return SglangTestDefinition.parse_bench_output(path)
def parse_semantic_accuracy(self) -> float | None:
return parse_sglang_semantic_accuracy(self.test_run.output_path / SGLANG_SEMANTIC_EVAL_LOG_FILE)
diff --git a/src/cloudai/workloads/sglang/sglang.py b/src/cloudai/workloads/sglang/sglang.py
index 3fe93e53b..4bfcba4b9 100644
--- a/src/cloudai/workloads/sglang/sglang.py
+++ b/src/cloudai/workloads/sglang/sglang.py
@@ -24,6 +24,7 @@
from pydantic import ConfigDict, Field, field_validator, model_validator
import cloudai.metrics
+import cloudai.report_generator.inference_report as inference_report
from cloudai.core import JobStatusResult, System, TestRun
from cloudai.models.workload import CmdArgs
from cloudai.workloads.common.llm_serving import (
@@ -100,7 +101,7 @@ class SglangSemanticEvalCmdArgs(CmdArgs):
cli: str = "--host {host} --port {port} --eval-name gsm8k --num-examples 200 --num-threads 128 --model {model}"
-class SglangTestDefinition(LLMServingTestDefinition[SglangCmdArgs]):
+class SglangTestDefinition(LLMServingTestDefinition[SglangCmdArgs], inference_report.InferenceReportProvider):
"""Test object for SGLang."""
bench_cmd_args: SglangBenchCmdArgs = SglangBenchCmdArgs()
@@ -113,7 +114,7 @@ def validate_custom_bash(cls, custom_bash: CustomBash | None) -> CustomBash | No
return validate_custom_bash_patterns(custom_bash)
def was_run_successful(self, tr: TestRun) -> JobStatusResult:
- res = parse_sglang_bench_output(tr.output_path / SGLANG_BENCH_JSONL_FILE)
+ res = self.parse_bench_output(tr.output_path / SGLANG_BENCH_JSONL_FILE)
if res and res.completed > 0:
if self.semantic_eval_cmd_args is not None:
accuracy = parse_sglang_semantic_accuracy(tr.output_path / SGLANG_SEMANTIC_EVAL_LOG_FILE)
@@ -129,20 +130,66 @@ def was_run_successful(self, tr: TestRun) -> JobStatusResult:
error_message=f"SGLang bench jsonl does not contain successful requests in {tr.output_path}.",
)
+ @staticmethod
+ @cache
+ def parse_bench_output(jsonl_file: Path) -> SGLangBenchReport | None:
+ """Parse SGLang benchmark output from JSONL file."""
+ if not jsonl_file.is_file():
+ return None
+
+ with jsonl_file.open(encoding="utf-8", errors="ignore") as f:
+ for line in f:
+ try:
+ parsed = SGLangBenchReport.model_validate_json(line)
+ if parsed.completed <= 0:
+ return None
+ return parsed
+ except Exception as e:
+ logging.debug(f"Skipping invalid JSONL record in SGLang benchmark output: {e}")
+ continue
+
+ return None
+
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_reports(self, system: System, tr: TestRun) -> list[inference_report.InferenceReport]:
gpus_per_node = getattr(system, "gpus_per_node", None)
- results = parse_sglang_bench_output(tr.output_path / SGLANG_BENCH_JSONL_FILE)
+ results = self.parse_bench_output(tr.output_path / SGLANG_BENCH_JSONL_FILE)
accuracy = (
parse_sglang_semantic_accuracy(tr.output_path / SGLANG_SEMANTIC_EVAL_LOG_FILE)
if self.semantic_eval_cmd_args is not None
else None
)
- return llm_serving_metric_observations(
+ observations = llm_serving_metric_observations(
results,
accuracy,
allocated_gpu_count=tr.nnodes * gpus_per_node if gpus_per_node else None,
serving_gpu_count=self.serving_gpu_count,
)
+ if not observations:
+ return []
+ prefill, decode = self.inference_role_configs(tr)
+ config = inference_report.InferenceConfig(
+ model=self.cmd_args.model,
+ backend="sglang",
+ benchmark="sglang",
+ num_nodes=tr.nnodes,
+ container_image=self.cmd_args.docker_image_url,
+ gpus_per_node=gpus_per_node,
+ serving_gpu_count=self.serving_gpu_count,
+ input_tokens=self.bench_cmd_args.random_input,
+ output_tokens=self.bench_cmd_args.random_output,
+ concurrency=self.bench_cmd_args.max_concurrency,
+ request_count=self.bench_cmd_args.num_prompts,
+ request_rate=getattr(self.bench_cmd_args, "request_rate", None),
+ prefill=prefill,
+ decode=decode,
+ shared_nodes=tr.nnodes == 1
+ if prefill is not None and prefill.num_nodes is not None and decode.num_nodes is not None
+ else None,
+ )
+ return [inference_report.InferenceReport(config=config, observations=observations)]
class SGLangBenchReport(LLMServingBenchReport):
@@ -166,26 +213,6 @@ def derive_num_prompts(cls, data):
return data
-@cache
-def parse_sglang_bench_output(jsonl_file: Path) -> SGLangBenchReport | None:
- """Parse SGLang benchmark output from JSONL file."""
- if not jsonl_file.is_file():
- return None
-
- with jsonl_file.open(encoding="utf-8", errors="ignore") as f:
- for line in f:
- try:
- parsed = SGLangBenchReport.model_validate_json(line)
- if parsed.completed <= 0:
- return None
- return parsed
- except Exception as e:
- logging.debug(f"Skipping invalid JSONL record in SGLang benchmark output: {e}")
- continue
-
- return None
-
-
@cache
def parse_sglang_semantic_accuracy(log_file: Path) -> float | None:
"""Parse SGLang semantic validation accuracy from run_eval or legacy GSM8K output."""
diff --git a/src/cloudai/workloads/vllm/report_generation_strategy.py b/src/cloudai/workloads/vllm/report_generation_strategy.py
index fe58e2dbc..c7850d20b 100644
--- a/src/cloudai/workloads/vllm/report_generation_strategy.py
+++ b/src/cloudai/workloads/vllm/report_generation_strategy.py
@@ -14,39 +14,15 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-import json
-import logging
-from functools import cache
from pathlib import Path
-from cloudai.workloads.common.llm_serving import LLMServingBenchReport, LLMServingReportGenerationStrategy, all_gpu_ids
-
-from .vllm import VLLM_BENCH_JSON_FILE, VllmTestDefinition, parse_vllm_semantic_accuracy
-
-
-class VLLMBenchReport(LLMServingBenchReport):
- """Report for vLLM benchmark results."""
-
- output_throughput: float
- request_throughput: float
-
- @property
- def throughput(self) -> float:
- return self.output_throughput
-
-
-@cache
-def parse_vllm_bench_output(res_file: Path) -> VLLMBenchReport | None:
- """Parse the vLLM benchmark output file and return a VLLMBenchReport object."""
- if not res_file.is_file():
- return None
-
- try:
- data = json.loads(res_file.read_text())
- return VLLMBenchReport.model_validate(data)
- except Exception as e:
- logging.debug(f"Error parsing vLLM benchmark output: {e}")
- return None
+from cloudai.workloads.common.llm_serving import LLMServingReportGenerationStrategy, all_gpu_ids
+from cloudai.workloads.vllm.vllm import (
+ VLLM_BENCH_JSON_FILE,
+ VLLMBenchReport,
+ VllmTestDefinition,
+ parse_vllm_semantic_accuracy,
+)
class VLLMBenchReportGenerationStrategy(LLMServingReportGenerationStrategy[VllmTestDefinition, VLLMBenchReport]):
@@ -61,7 +37,7 @@ def report_title(self) -> str:
return "vLLM Benchmark Results"
def parse_output(self, path: Path) -> VLLMBenchReport | None:
- return parse_vllm_bench_output(path)
+ return VllmTestDefinition.parse_bench_output(path)
def parse_semantic_accuracy(self) -> float | None:
return parse_vllm_semantic_accuracy(self.test_run.output_path)
diff --git a/src/cloudai/workloads/vllm/vllm.py b/src/cloudai/workloads/vllm/vllm.py
index 3a96f0f03..8410a924f 100644
--- a/src/cloudai/workloads/vllm/vllm.py
+++ b/src/cloudai/workloads/vllm/vllm.py
@@ -26,11 +26,13 @@
from pydantic import ConfigDict, Field, field_validator
import cloudai.metrics
+import cloudai.report_generator.inference_report as inference_report
from cloudai.core import GitRepo, Installable, JobStatusResult, System, TestRun
from cloudai.models.workload import CmdArgs
from cloudai.workloads.common.llm_serving import (
CustomBash,
LLMServingArgs,
+ LLMServingBenchReport,
LLMServingCmdArgs,
LLMServingTestDefinition,
all_gpu_ids,
@@ -123,7 +125,7 @@ class VllmSemanticEvalCmdArgs(CmdArgs):
cli: str = "--host {host} --port {port} --num-questions 200 --save-results {output_path}/vllm-gsm8k.json"
-class VllmTestDefinition(LLMServingTestDefinition[VllmCmdArgs]):
+class VllmTestDefinition(LLMServingTestDefinition[VllmCmdArgs], inference_report.InferenceReportProvider):
"""Test object for vLLM."""
bench_cmd_args: VllmBenchCmdArgs = VllmBenchCmdArgs()
@@ -266,18 +268,67 @@ def was_run_successful(self, tr: TestRun) -> JobStatusResult:
is_successful=False, error_message=f"vLLM bench log does not contain benchmark result in {tr.output_path}."
)
+ @staticmethod
+ @cache
+ def parse_bench_output(res_file: Path) -> VLLMBenchReport | None:
+ """Parse the vLLM benchmark output file and return a VLLMBenchReport object."""
+ if not res_file.is_file():
+ return None
+
+ try:
+ data = json.loads(res_file.read_text())
+ return VLLMBenchReport.model_validate(data)
+ except Exception as e:
+ logging.debug(f"Error parsing vLLM benchmark output: {e}")
+ return None
+
def metric_observations(self, system: System, tr: TestRun) -> list[cloudai.metrics.MetricObservation]:
- from .report_generation_strategy import parse_vllm_bench_output
+ return [obs for report in self.inference_reports(system, tr) for obs in report.observations]
+ def inference_reports(self, system: System, tr: TestRun) -> list[inference_report.InferenceReport]:
gpus_per_node = getattr(system, "gpus_per_node", None)
- results = parse_vllm_bench_output(tr.output_path / VLLM_BENCH_JSON_FILE)
+ results = self.parse_bench_output(tr.output_path / VLLM_BENCH_JSON_FILE)
accuracy = parse_vllm_semantic_accuracy(tr.output_path) if self.semantic_eval_cmd_args is not None else None
- return llm_serving_metric_observations(
+ observations = llm_serving_metric_observations(
results,
accuracy,
allocated_gpu_count=tr.nnodes * gpus_per_node if gpus_per_node else None,
serving_gpu_count=self.serving_gpu_count,
)
+ if not observations:
+ return []
+ prefill, decode = self.inference_role_configs(tr)
+ config = inference_report.InferenceConfig(
+ model=self.cmd_args.model,
+ backend="vllm",
+ benchmark="vllm",
+ num_nodes=tr.nnodes,
+ container_image=self.cmd_args.docker_image_url,
+ gpus_per_node=gpus_per_node,
+ serving_gpu_count=self.serving_gpu_count,
+ input_tokens=self.bench_cmd_args.random_input_len,
+ output_tokens=self.bench_cmd_args.random_output_len,
+ concurrency=self.bench_cmd_args.max_concurrency,
+ request_count=self.bench_cmd_args.num_prompts,
+ request_rate=getattr(self.bench_cmd_args, "request_rate", None),
+ prefill=prefill,
+ decode=decode,
+ shared_nodes=tr.nnodes == 1
+ if prefill is not None and prefill.num_nodes is not None and decode.num_nodes is not None
+ else None,
+ )
+ return [inference_report.InferenceReport(config=config, observations=observations)]
+
+
+class VLLMBenchReport(LLMServingBenchReport):
+ """Report for vLLM benchmark results."""
+
+ output_throughput: float
+ request_throughput: float
+
+ @property
+ def throughput(self) -> float:
+ return self.output_throughput
@cache
diff --git a/tests/report_generation_strategy/test_pareto_report.py b/tests/report_generation_strategy/test_pareto_report.py
index d79693cdb..b4baed113 100644
--- a/tests/report_generation_strategy/test_pareto_report.py
+++ b/tests/report_generation_strategy/test_pareto_report.py
@@ -19,66 +19,129 @@
import json
from pathlib import Path
+import pydantic
import pytest
import toml
import cloudai.core
+import cloudai.metrics
import cloudai.models.scenario
+import cloudai.models.workload
+import cloudai.report_generator.inference_report as inference_report
import cloudai.report_generator.pareto_report as pareto_report
import cloudai.systems.slurm
-import cloudai.workloads.sglang
-import cloudai.workloads.vllm
-@pytest.fixture(
- params=list(itertools.product(["vllm", "sglang"], ["minimize", "maximize"], ["minimize", "maximize"])),
- ids=lambda values: "-".join(values),
-)
+class DummyArgs(pydantic.BaseModel):
+ """Input configuration restored from a saved sweep step."""
+
+ concurrency: int = 16
+
+
+class DummyInferenceDefinition(cloudai.models.workload.TestDefinition, inference_report.InferenceReportProvider):
+ """Supply independent reports without a benchmark parser or serving backend."""
+
+ cmd_args: DummyArgs
+
+ def inference_reports(
+ self, system: cloudai.core.System, tr: cloudai.core.TestRun
+ ) -> list[inference_report.InferenceReport]:
+ gpus_per_node = getattr(system, "gpus_per_node", None)
+ assert gpus_per_node is not None
+ reports = []
+ for measurement in json.loads((tr.output_path / "measurements.json").read_text()):
+ dimensions = {"phase": measurement["phase"]} if "phase" in measurement else {}
+ observations = [
+ cloudai.metrics.MetricObservation(metric, value * factor, {**dimensions, "statistic": statistic})
+ for metric, value, factors in (
+ (cloudai.metrics.TTFT, measurement["x"], (0.5, 0.75, 1)),
+ (cloudai.metrics.TPOT, measurement["tpot"], (1, 2, 3)),
+ )
+ for statistic, factor in zip(("mean", "median", "p99"), factors, strict=True)
+ ]
+ observations.extend(
+ [
+ cloudai.metrics.MetricObservation(
+ cloudai.metrics.OUTPUT_TOKEN_THROUGHPUT, measurement["y"], dimensions
+ ),
+ cloudai.metrics.MetricObservation(
+ cloudai.metrics.OUTPUT_TOKEN_THROUGHPUT_PER_ALLOCATED_GPU,
+ measurement["y"] / (tr.nnodes * gpus_per_node),
+ dimensions,
+ ),
+ ]
+ )
+ reports.append(
+ inference_report.InferenceReport(
+ config=inference_report.InferenceConfig(
+ model="example/model",
+ num_nodes=tr.nnodes,
+ gpus_per_node=gpus_per_node,
+ concurrency=measurement.get("concurrency", self.cmd_args.concurrency),
+ backend="example-backend",
+ container_image="example/image:latest",
+ serving_gpu_count=4,
+ input_tokens=128,
+ output_tokens=64,
+ users=8,
+ request_count=32,
+ request_rate=0,
+ shared_nodes=False,
+ prefill=inference_report.InferenceRoleConfig(num_nodes=1) if dimensions else None,
+ decode=inference_report.InferenceRoleConfig(
+ model="example/role-model",
+ num_nodes=1,
+ num_instances=1,
+ nodes_per_instance=1,
+ instances_per_node=1,
+ gpu_ids=[0, 1],
+ tensor_parallel_size=2,
+ pipeline_parallel_size=1,
+ data_parallel_size=1,
+ expert_parallel=False,
+ precision="bfloat16",
+ quantization="fp8",
+ ),
+ ),
+ observations=observations,
+ )
+ )
+ return reports
+
+
+@pytest.fixture(params=list(itertools.product(["minimize", "maximize"], repeat=2)), ids=lambda values: "-".join(values))
def pareto_reporter(
- slurm_system: cloudai.systems.slurm.SlurmSystem, tmp_path: Path, request: pytest.FixtureRequest
+ slurm_system: cloudai.systems.slurm.SlurmSystem,
+ tmp_path: Path,
+ request: pytest.FixtureRequest,
+ monkeypatch: pytest.MonkeyPatch,
) -> pareto_report.ParetoReporter:
- workload, x_direction, y_direction = request.param
- cls = (
- cloudai.workloads.vllm.VllmTestDefinition
- if workload == "vllm"
- else cloudai.workloads.sglang.SglangTestDefinition
+ monkeypatch.setitem(cloudai.core.Registry().test_definitions_map, "dummy", DummyInferenceDefinition)
+ monkeypatch.setitem(
+ cloudai.metrics.MetricCatalog._dimensions,
+ "phase",
+ cloudai.metrics.DimensionDefinition("phase", "Phase", str),
)
- definition = cls.model_validate(
- {
- "name": "test",
- "description": "test",
- "test_template_name": workload,
- "cmd_args": {"docker_image_url": "example/image:latest", "model": "example/model"},
- }
+ x_direction, y_direction = request.param
+ definition = DummyInferenceDefinition(
+ name="test", description="test", test_template_name="dummy", cmd_args=DummyArgs()
)
executed = copy.deepcopy(definition)
- executed.bench_cmd_args.max_concurrency = 8
- setattr(executed.cmd_args.decode, "tensor_parallel_size" if workload == "vllm" else "tp_size", 4)
+ executed.cmd_args.concurrency = 8
for relative, x, y in [("ordinary/0", 2, 20), ("ordinary/1", 1, 10), ("ordinary/2", 2, 20), ("sweep/0/1", 3, 15)]:
folder = tmp_path / relative
folder.mkdir(parents=True)
- data = {
- "num_prompts": 10,
- "completed": 10,
- "max_concurrency": 4,
- "mean_ttft_ms": x / 2,
- "median_ttft_ms": x * 0.75,
- "p99_ttft_ms": x,
- "mean_tpot_ms": 1,
- "median_tpot_ms": 2,
- "p99_tpot_ms": 3,
- "request_throughput": 1,
- "output_throughput": y,
- }
- (folder / ("vllm-bench.json" if workload == "vllm" else "sglang-bench.jsonl")).write_text(json.dumps(data))
- (folder / "vllm-bench.log").write_text(
- "============ Serving Benchmark Result ============\nSuccessful requests: 10\n"
- )
+ measurements = [{"x": x, "y": y, "tpot": 1}]
if relative.startswith("sweep"):
+ measurements = [
+ {"x": x, "y": y, "tpot": 1, "phase": "low"},
+ {"x": 4, "y": 30, "tpot": 2, "phase": "high", "concurrency": 32},
+ ]
resolved = cloudai.core.TestRun("sweep", executed, 2, [], output_path=folder)
(folder / "test-run.toml").write_text(
toml.dumps(cloudai.models.scenario.TestRunDetails.from_test_run(resolved, "", "").model_dump())
)
+ (folder / "measurements.json").write_text(json.dumps(measurements))
for relative in ("sweep/0/2", "sweep/controller", "empty-sweep/0/controller"):
(tmp_path / relative).mkdir(parents=True)
plot = {
@@ -109,6 +172,14 @@ def pareto_reporter(
"frontiers": [{"name": "All", "cases": []}],
},
{**plot, "name": "Ambiguous latency", "x": {"metric": "ttft"}},
+ {
+ **plot,
+ "name": "Selected phase",
+ "frontiers": [
+ {"name": "High", "match": {"phase": "high"}},
+ {"name": "Missing", "match": {"phase": "missing"}},
+ ],
+ },
],
}
)
@@ -129,19 +200,41 @@ def test_build_sections(pareto_reporter: pareto_report.ParetoReporter) -> None:
mode="json", by_alias=True, exclude_none=True
)
sweep_definition = copy.deepcopy(definition)
- sweep_definition["bench_cmd_args"]["max_concurrency"] = 8
- tp_key = "tensor_parallel_size" if definition["test_template_name"] == "vllm" else "tp_size"
- sweep_definition["cmd_args"]["decode"][tp_key] = 4
+ sweep_definition["cmd_args"]["concurrency"] = 8
context = {
- "Workload": definition["test_template_name"],
+ "Workload": "dummy",
+ "Model": "example/model",
"Nodes": 1,
"Allocated GPUs": 8,
- "Serving GPUs": 1,
- "Model": "example/model",
- "Input tokens": 16,
- "Output tokens": 128,
"Concurrency": 16,
- "Requests": 30,
+ "Backend": "example-backend",
+ "Container image": "example/image:latest",
+ "Serving GPUs": 4,
+ "Input tokens": 128,
+ "Output tokens": 64,
+ "Users": 8,
+ "Requests": 32,
+ "Request rate": 0,
+ "Prefill and decode share nodes": False,
+ "Serving / Model": "example/role-model",
+ "Serving / Nodes": 1,
+ "Serving / Instances": 1,
+ "Serving / Nodes per instance": 1,
+ "Serving / Instances per node": 1,
+ "Serving / GPU IDs": [0, 1],
+ "Serving / TP": 2,
+ "Serving / PP": 1,
+ "Serving / DP": 1,
+ "Serving / Expert parallel": False,
+ "Serving / Precision": "bfloat16",
+ "Serving / Quantization": "fp8",
+ }
+ sweep_context = {
+ **{key.replace("Serving / ", "Decode / "): value for key, value in context.items()},
+ "Nodes": 2,
+ "Allocated GPUs": 16,
+ "Concurrency": 8,
+ "Prefill / Nodes": 1,
}
points = [
{
@@ -173,14 +266,7 @@ def test_build_sections(pareto_reporter: pareto_report.ParetoReporter) -> None:
"y": 15,
"run": "sweep · iteration=0 · step=1",
"group": "All",
- "context": {
- **context,
- "Nodes": 2,
- "Allocated GPUs": 16,
- "Serving GPUs": 4,
- "Concurrency": 8,
- "Decode / TP": 4,
- },
+ "context": sweep_context,
"dimensions": "x.statistic=p99",
"config": toml.dumps({"num_nodes": 2, "test": sweep_definition}),
"artifact": "sweep/0/1",
@@ -191,12 +277,24 @@ def test_build_sections(pareto_reporter: pareto_report.ParetoReporter) -> None:
points.insert(
2, {**points[0], "run": "ordinary · iteration=2 · step=0", "artifact": "ordinary/2", "id": "pareto-0-point-2"}
)
+ sweep_point = points.pop()
+ for index, (phase, concurrency, x, y) in enumerate([("low", 8, 3, 15), ("high", 32, 4, 30)], start=3):
+ point = {
+ **sweep_point,
+ "x": x,
+ "y": y,
+ "run": "sweep · iteration=0 · step=1",
+ "context": {**sweep_point["context"], "Concurrency": concurrency},
+ "dimensions": f"x.phase={phase}, x.statistic=p99, y.phase={phase}",
+ "id": f"pareto-0-point-{index}",
+ }
+ points.append(point)
plot = pareto_reporter.config.plots[0]
expected_frontier = {
- ("minimize", "maximize"): [True, True, True, False],
- ("maximize", "maximize"): [True, False, True, True],
- ("minimize", "minimize"): [False, True, False, False],
- ("maximize", "minimize"): [False, True, False, True],
+ ("minimize", "maximize"): [True, True, True, False, True],
+ ("maximize", "maximize"): [False, False, False, False, True],
+ ("minimize", "minimize"): [False, True, False, False, False],
+ ("maximize", "minimize"): [False, True, False, True, True],
}[plot.x.resolved_direction.value, plot.y.resolved_direction.value]
for point, on_frontier in zip(points, expected_frontier, strict=True):
point["status"] = "Frontier" if on_frontier else "Dominated"
@@ -214,7 +312,7 @@ def test_build_sections(pareto_reporter: pareto_report.ParetoReporter) -> None:
"x_axis_label": "P99 Time to first token (ms)",
"y_axis_label": "Output token throughput (tokens/s)",
}
- chart_points = [{k: v for k, v in points[i].items() if k != "config"} for i in (1, 0, 2, 3)]
+ chart_points = [{k: v for k, v in points[i].items() if k != "config"} for i in (1, 0, 2, 3, 4)]
normalized_points = [
{
**point,
@@ -228,19 +326,24 @@ def test_build_sections(pareto_reporter: pareto_report.ParetoReporter) -> None:
for i, (point, y, group, status) in enumerate(
zip(
points,
- [20 / 8, 10 / 8, 20 / 8, 15 / 16],
- ["Ordinary", "Ordinary", "Ordinary", "Sweep"],
- ["Frontier", "Dominated", "Frontier", "Frontier"],
+ [20 / 8, 10 / 8, 20 / 8, 15 / 16, 30 / 16],
+ ["Ordinary", "Ordinary", "Ordinary", "Sweep", "Sweep"],
+ ["Frontier", "Dominated", "Frontier", "Dominated", "Frontier"],
strict=True,
)
)
]
+ normalized_points[3]["dimensions"] = "x.phase=low, x.statistic=mean, y.phase=low"
+ normalized_points[4]["dimensions"] = "x.phase=high, x.statistic=mean, y.phase=high"
+ normalized_points[4]["x"] = 2
normalized_data = [{k: v for k, v in point.items() if k != "config"} for point in normalized_points]
expected = [
{
"pareto": True,
"title": "All configurations",
- "summaries": [{"name": "All", "candidates": 4, "frontier": sum(expected_frontier), "excluded": 3}],
+ "summaries": [
+ {"name": "All", "candidates": len(points), "frontier": sum(expected_frontier), "excluded": 3}
+ ],
"points": points,
"excluded": excluded,
"chart": {
@@ -268,7 +371,7 @@ def test_build_sections(pareto_reporter: pareto_report.ParetoReporter) -> None:
"title": "Per allocated GPU",
"summaries": [
{"name": "Ordinary", "candidates": 3, "frontier": 2, "excluded": 0},
- {"name": "Sweep", "candidates": 1, "frontier": 1, "excluded": 3},
+ {"name": "Sweep", "candidates": 2, "frontier": 1, "excluded": 3},
],
"points": normalized_points,
"excluded": [{**item, "group": "Sweep"} for item in excluded],
@@ -294,11 +397,18 @@ def test_build_sections(pareto_reporter: pareto_report.ParetoReporter) -> None:
},
{
"label": "Sweep",
- "data": [normalized_data[3]],
+ "data": [normalized_data[4]],
"source_color_index": 1,
"is_frontier": True,
"hidden": False,
},
+ {
+ "label": "Sweep · dominated",
+ "data": [normalized_data[3]],
+ "source_color_index": 1,
+ "is_frontier": False,
+ "hidden": True,
+ },
],
},
},
@@ -306,7 +416,7 @@ def test_build_sections(pareto_reporter: pareto_report.ParetoReporter) -> None:
{
"pareto": True,
"title": title,
- "summaries": [{"name": "All", "candidates": 0, "frontier": 0, "excluded": 7}],
+ "summaries": [{"name": "All", "candidates": 0, "frontier": 0, "excluded": len(points) + 3}],
"points": [],
"excluded": [{"group": "All", "run": point["run"], "reason": reason} for point in points] + excluded,
"chart": {
@@ -335,7 +445,60 @@ def test_build_sections(pareto_reporter: pareto_report.ParetoReporter) -> None:
]
],
]
- assert pareto_reporter.build_sections(pareto_reporter.collect_runs()) == expected
+ selected = {**points[-1], "group": "High", "status": "Frontier", "id": "pareto-4-point-0"}
+ expected.append(
+ {
+ "pareto": True,
+ "title": "Selected phase",
+ "summaries": [
+ {"name": "High", "candidates": 1, "frontier": 1, "excluded": 3},
+ {"name": "Missing", "candidates": 0, "frontier": 0, "excluded": 3},
+ ],
+ "points": [selected],
+ "excluded": [{**item, "group": group} for group in ("High", "Missing") for item in excluded],
+ "chart": {
+ **chart,
+ "id": "pareto-4",
+ "datasets": [
+ {
+ "label": "High",
+ "data": [{key: value for key, value in selected.items() if key != "config"}],
+ "source_color_index": 0,
+ "is_frontier": True,
+ "hidden": False,
+ }
+ ],
+ },
+ }
+ )
+ runs = pareto_reporter.collect_runs()
+ assert pareto_reporter.build_sections(runs) == expected
+ _assert_unmatched_frontiers_keep_errors(pareto_reporter, runs, expected)
+
+
+def _assert_unmatched_frontiers_keep_errors(pareto_reporter, runs, expected) -> None:
+ for plot in pareto_reporter.config.plots:
+ for frontier in plot.frontiers:
+ frontier.match = {"benchmark": "not-present"}
+ filtered_expected = copy.deepcopy(expected)
+ for section in filtered_expected:
+ section["points"] = []
+ section["chart"]["datasets"] = []
+ section["excluded"] = [
+ item
+ for item in section["excluded"]
+ if item["reason"]
+ in {
+ "Missing test-run.toml for sweep step",
+ "No execution results found",
+ "No sweep steps found in iteration 0",
+ }
+ ]
+ for summary in section["summaries"]:
+ summary["candidates"] = 0
+ summary["frontier"] = 0
+ summary["excluded"] = sum(item["group"] == summary["name"] for item in section["excluded"])
+ assert pareto_reporter.build_sections(runs) == filtered_expected
def test_generate_html(pareto_reporter: pareto_report.ParetoReporter) -> None:
diff --git a/tests/workloads/sglang/test_report_gen_strategy.py b/tests/workloads/sglang/test_report_gen_strategy.py
index 74668f9b2..09c34992d 100644
--- a/tests/workloads/sglang/test_report_gen_strategy.py
+++ b/tests/workloads/sglang/test_report_gen_strategy.py
@@ -20,6 +20,9 @@
import pytest
+import cloudai.metrics
+import cloudai.report_generator.inference_report as inference_report
+import cloudai.util
from cloudai import TestRun
from cloudai.core import METRIC_ERROR
from cloudai.systems.slurm import SlurmSystem
@@ -33,7 +36,6 @@
from cloudai.workloads.sglang.sglang import (
SGLANG_BENCH_JSONL_FILE,
SGLANG_SEMANTIC_EVAL_LOG_FILE,
- parse_sglang_bench_output,
parse_sglang_semantic_accuracy,
)
@@ -75,7 +77,7 @@ def test_sglang_can_handle_directory(slurm_system: SlurmSystem, sglang_tr: TestR
def test_parse_sglang_bench_output_missing_file(tmp_path: Path) -> None:
- assert parse_sglang_bench_output(tmp_path / SGLANG_BENCH_JSONL_FILE) is None
+ assert SglangTestDefinition.parse_bench_output(tmp_path / SGLANG_BENCH_JSONL_FILE) is None
def test_parse_sglang_bench_output_skips_invalid_lines(tmp_path: Path) -> None:
@@ -85,7 +87,7 @@ def test_parse_sglang_bench_output_skips_invalid_lines(tmp_path: Path) -> None:
encoding="utf-8",
)
- parsed = parse_sglang_bench_output(report_path)
+ parsed = SglangTestDefinition.parse_bench_output(report_path)
assert parsed == SGLangBenchReport.model_validate_json(report_path.read_text().splitlines()[1])
@@ -97,7 +99,7 @@ def test_parse_sglang_bench_output_returns_none_for_unsuccessful_record(tmp_path
encoding="utf-8",
)
- assert parse_sglang_bench_output(report_path) is None
+ assert SglangTestDefinition.parse_bench_output(report_path) is None
@pytest.mark.parametrize(
@@ -193,3 +195,134 @@ def test_sglang_parses_num_prompts_from_input_lens(slurm_system: SlurmSystem, tm
strategy = SGLangBenchReportGenerationStrategy(slurm_system, tr)
assert strategy.can_handle_directory() is True
assert strategy.get_metric("throughput") == 2400.0
+
+
+INFERENCE_DEFINITION = {
+ "name": "sglang",
+ "description": "Inference report",
+ "test_template_name": "sglang",
+ "cmd_args": {
+ "docker_image_url": "example/image:latest",
+ "model": "example/model",
+ "decode": {
+ "tp_size": 2,
+ "pp_size": 2,
+ "dp_size": 1,
+ "enable_expert_parallel": True,
+ "dtype": "float16",
+ "quantization": "fp8",
+ },
+ },
+ "bench_cmd_args": {
+ "num_prompts": 20,
+ "max_concurrency": 8,
+ "request_rate": 12.5,
+ "random_input": 128,
+ "random_output": 64,
+ },
+}
+INFERENCE_PREFILL = {
+ "tp_size": 2,
+ "pp_size": 1,
+ "dp_size": 1,
+ "enable_expert_parallel": True,
+ "dtype": "float16",
+ "quantization": "fp8",
+}
+
+
+@pytest.mark.parametrize(
+ "overrides,nodes,role_nodes,serving_gpus,shared_nodes",
+ [
+ pytest.param({}, 2, 2, 4, None, id="combined"),
+ pytest.param(
+ {"cmd_args": {"prefill": {**INFERENCE_PREFILL, "num_nodes": 2}, "decode": {"num_nodes": 2}}},
+ 4,
+ 2,
+ 6,
+ False,
+ id="separate-nodes",
+ ),
+ pytest.param({"cmd_args": {"prefill": INFERENCE_PREFILL}}, 1, 1, 6, True, id="shared-node"),
+ pytest.param({"cmd_args": {"prefill": INFERENCE_PREFILL}}, 4, None, 6, None, id="unresolved-topology"),
+ ],
+)
+def test_inference_reports(
+ slurm_system: SlurmSystem,
+ tmp_path: Path,
+ overrides: dict,
+ nodes: int,
+ role_nodes: int | None,
+ serving_gpus: int,
+ shared_nodes: bool | None,
+) -> None:
+ definition = cloudai.util.deep_merge(INFERENCE_DEFINITION, overrides)
+ test = SglangTestDefinition.model_validate(definition)
+ tr = TestRun("sglang", test, nodes, [], output_path=tmp_path)
+ (tmp_path / "sglang-bench.jsonl").write_text(
+ json.dumps(
+ {
+ "num_prompts": 20,
+ "completed": 20,
+ "max_concurrency": 8,
+ "request_throughput": 10,
+ "output_throughput": 1000,
+ "mean_ttft_ms": 10,
+ "median_ttft_ms": 11,
+ "p99_ttft_ms": 12,
+ "mean_tpot_ms": 1,
+ "median_tpot_ms": 2,
+ "p99_tpot_ms": 3,
+ }
+ )
+ )
+ role: dict = dict(
+ num_nodes=role_nodes,
+ num_instances=1 if role_nodes else None,
+ nodes_per_instance=role_nodes,
+ instances_per_node=1 if role_nodes == 1 else None,
+ tensor_parallel_size=2,
+ data_parallel_size=1,
+ expert_parallel=True,
+ precision="float16",
+ quantization="fp8",
+ )
+ metrics = [
+ (cloudai.metrics.REQUEST_THROUGHPUT, 10, {}),
+ (cloudai.metrics.OUTPUT_TOKEN_THROUGHPUT, 1000, {}),
+ (cloudai.metrics.OUTPUT_TOKEN_THROUGHPUT_PER_ALLOCATED_GPU, 1000 / (nodes * 8), {}),
+ (cloudai.metrics.OUTPUT_TOKEN_THROUGHPUT_PER_SERVING_GPU, 1000 / serving_gpus, {}),
+ ]
+ 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)
+ )
+ expected = [
+ inference_report.InferenceReport(
+ config=inference_report.InferenceConfig(
+ model="example/model",
+ num_nodes=nodes,
+ backend="sglang",
+ benchmark="sglang",
+ container_image="example/image:latest",
+ gpus_per_node=8,
+ serving_gpu_count=serving_gpus,
+ input_tokens=128,
+ output_tokens=64,
+ concurrency=8,
+ request_count=20,
+ request_rate=12.5,
+ prefill=inference_report.InferenceRoleConfig(**role, pipeline_parallel_size=1)
+ if "prefill" in definition["cmd_args"]
+ else None,
+ decode=inference_report.InferenceRoleConfig(**role, pipeline_parallel_size=2),
+ shared_nodes=shared_nodes,
+ ),
+ observations=[
+ cloudai.metrics.MetricObservation(metric, value, dimensions) for metric, value, dimensions in metrics
+ ],
+ )
+ ]
+ assert test.inference_reports(slurm_system, tr) == expected
+ assert test.metric_observations(slurm_system, tr) == expected[0].observations
diff --git a/tests/workloads/vllm/test_report_gen_strategy.py b/tests/workloads/vllm/test_report_gen_strategy.py
index baac6defb..1f79b3cd6 100644
--- a/tests/workloads/vllm/test_report_gen_strategy.py
+++ b/tests/workloads/vllm/test_report_gen_strategy.py
@@ -14,11 +14,15 @@
# See the License for the specific language governing permissions and
# limitations under the License.
+import json
from pathlib import Path
from typing import cast
import pytest
+import cloudai.metrics
+import cloudai.report_generator.inference_report as inference_report
+import cloudai.util
from cloudai import TestRun
from cloudai.core import METRIC_ERROR
from cloudai.systems.slurm import SlurmSystem
@@ -29,7 +33,6 @@
VllmSemanticEvalCmdArgs,
VllmTestDefinition,
)
-from cloudai.workloads.vllm.report_generation_strategy import parse_vllm_bench_output
from cloudai.workloads.vllm.vllm import (
VLLM_BENCH_JSON_FILE,
VLLM_GSM8K_JSON_FILE,
@@ -72,14 +75,14 @@ def test_vllm_can_handle_directory(slurm_system: SlurmSystem, vllm_tr: TestRun)
def test_parse_vllm_bench_output_missing_file(tmp_path: Path) -> None:
- assert parse_vllm_bench_output(tmp_path / VLLM_BENCH_JSON_FILE) is None
+ assert VllmTestDefinition.parse_bench_output(tmp_path / VLLM_BENCH_JSON_FILE) is None
def test_parse_vllm_bench_output_invalid_json(tmp_path: Path) -> None:
report_path = tmp_path / VLLM_BENCH_JSON_FILE
report_path.write_text("{invalid", encoding="utf-8")
- assert parse_vllm_bench_output(report_path) is None
+ assert VllmTestDefinition.parse_bench_output(report_path) is None
@pytest.mark.parametrize(
@@ -152,3 +155,134 @@ def test_parse_vllm_semantic_accuracy_missing_or_invalid(tmp_path: Path) -> None
(tmp_path / VLLM_SEMANTIC_EVAL_LOG_FILE).write_text("no accuracy here\n", encoding="utf-8")
assert parse_vllm_semantic_accuracy(tmp_path) is None
+
+
+INFERENCE_DEFINITION = {
+ "name": "vllm",
+ "description": "Inference report",
+ "test_template_name": "vllm",
+ "cmd_args": {
+ "docker_image_url": "example/image:latest",
+ "model": "example/model",
+ "decode": {
+ "tensor_parallel_size": 2,
+ "pipeline_parallel_size": 2,
+ "data_parallel_size": 1,
+ "enable_expert_parallel": True,
+ "dtype": "float16",
+ "quantization": "fp8",
+ },
+ },
+ "bench_cmd_args": {
+ "num_prompts": 20,
+ "max_concurrency": 8,
+ "request_rate": 12.5,
+ "random_input_len": 128,
+ "random_output_len": 64,
+ },
+}
+INFERENCE_PREFILL = {
+ "tensor_parallel_size": 2,
+ "pipeline_parallel_size": 1,
+ "data_parallel_size": 1,
+ "enable_expert_parallel": True,
+ "dtype": "float16",
+ "quantization": "fp8",
+}
+
+
+@pytest.mark.parametrize(
+ "overrides,nodes,role_nodes,serving_gpus,shared_nodes",
+ [
+ pytest.param({}, 2, 2, 4, None, id="combined"),
+ pytest.param(
+ {"cmd_args": {"prefill": {**INFERENCE_PREFILL, "num_nodes": 2}, "decode": {"num_nodes": 2}}},
+ 4,
+ 2,
+ 6,
+ False,
+ id="separate-nodes",
+ ),
+ pytest.param({"cmd_args": {"prefill": INFERENCE_PREFILL}}, 1, 1, 6, True, id="shared-node"),
+ pytest.param({"cmd_args": {"prefill": INFERENCE_PREFILL}}, 4, None, 6, None, id="unresolved-topology"),
+ ],
+)
+def test_inference_reports(
+ slurm_system: SlurmSystem,
+ tmp_path: Path,
+ overrides: dict,
+ nodes: int,
+ role_nodes: int | None,
+ serving_gpus: int,
+ shared_nodes: bool | None,
+) -> None:
+ definition = cloudai.util.deep_merge(INFERENCE_DEFINITION, overrides)
+ test = VllmTestDefinition.model_validate(definition)
+ tr = TestRun("vllm", test, nodes, [], output_path=tmp_path)
+ (tmp_path / "vllm-bench.json").write_text(
+ json.dumps(
+ {
+ "num_prompts": 20,
+ "completed": 20,
+ "max_concurrency": 8,
+ "request_throughput": 10,
+ "output_throughput": 1000,
+ "mean_ttft_ms": 10,
+ "median_ttft_ms": 11,
+ "p99_ttft_ms": 12,
+ "mean_tpot_ms": 1,
+ "median_tpot_ms": 2,
+ "p99_tpot_ms": 3,
+ }
+ )
+ )
+ role: dict = dict(
+ num_nodes=role_nodes,
+ num_instances=1 if role_nodes else None,
+ nodes_per_instance=role_nodes,
+ instances_per_node=1 if role_nodes == 1 else None,
+ tensor_parallel_size=2,
+ data_parallel_size=1,
+ expert_parallel=True,
+ precision="float16",
+ quantization="fp8",
+ )
+ metrics = [
+ (cloudai.metrics.REQUEST_THROUGHPUT, 10, {}),
+ (cloudai.metrics.OUTPUT_TOKEN_THROUGHPUT, 1000, {}),
+ (cloudai.metrics.OUTPUT_TOKEN_THROUGHPUT_PER_ALLOCATED_GPU, 1000 / (nodes * 8), {}),
+ (cloudai.metrics.OUTPUT_TOKEN_THROUGHPUT_PER_SERVING_GPU, 1000 / serving_gpus, {}),
+ ]
+ 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)
+ )
+ expected = [
+ inference_report.InferenceReport(
+ config=inference_report.InferenceConfig(
+ model="example/model",
+ num_nodes=nodes,
+ backend="vllm",
+ benchmark="vllm",
+ container_image="example/image:latest",
+ gpus_per_node=8,
+ serving_gpu_count=serving_gpus,
+ input_tokens=128,
+ output_tokens=64,
+ concurrency=8,
+ request_count=20,
+ request_rate=12.5,
+ prefill=inference_report.InferenceRoleConfig(**role, pipeline_parallel_size=1)
+ if "prefill" in definition["cmd_args"]
+ else None,
+ decode=inference_report.InferenceRoleConfig(**role, pipeline_parallel_size=2),
+ shared_nodes=shared_nodes,
+ ),
+ observations=[
+ cloudai.metrics.MetricObservation(metric, value, dimensions) for metric, value, dimensions in metrics
+ ],
+ )
+ ]
+ assert test.inference_reports(slurm_system, tr) == expected
+ assert test.metric_observations(slurm_system, tr) == expected[0].observations
|