[CI] optional --assert-baseline and update perf JSON baselines

tests/dfx/conftest.py

@@ -279,6 +279,10 @@ def create_benchmark_indices(
     return indices
 
 
+# Omni perf/stability: ``vllm bench serve --omni`` result template (used if bench exits without writing JSON).
+OMNI_RESULT_TEMPLATE_PATH = Path(__file__).resolve().parent / "perf" / "scripts" / "result_omni_template.json"
+
+
 def _safe_filename_token(value: Any | None, *, default: str = "na") -> str:
     """Make a single path segment safe for result filenames on common filesystems."""
     if value is None:
@@ -289,7 +293,7 @@ def _safe_filename_token(value: Any | None, *, default: str = "na") -> str:
     return s if s else default
 
 
-def _resolve_baseline_value(
+def resolve_baseline_value(
     baseline_raw: Any,
     *,
     sweep_index: int | None,
@@ -330,7 +334,7 @@ def _baseline_thresholds_for_step(
 ) -> dict[str, Any]:
     """Resolve baseline config to one threshold per metric for this iteration."""
     return {
-        metric_name: _resolve_baseline_value(
+        metric_name: resolve_baseline_value(
             baseline_raw,
             sweep_index=sweep_index,
             max_concurrency=max_concurrency,
@@ -354,7 +358,12 @@ def run_benchmark(
     random_input_len: Any | None = None,
     random_output_len: Any | None = None,
 ) -> dict[str, Any]:
-    """Run one ``vllm bench serve --omni`` iteration and return parsed metrics."""
+    """Run one ``vllm bench serve --omni`` iteration and return parsed metrics.
+
+    After ``vllm bench`` writes the JSON, ``result["baseline"]`` holds the resolved per-metric thresholds
+    (when ``baseline_config`` is provided). If the benchmark exits without writing a result file,
+    ``result_omni_template.json`` is used as a fallback.
+    """
     current_dt = datetime.now().strftime("%Y%m%d-%H%M%S")
     ri = _safe_filename_token(random_input_len)
     ro = _safe_filename_token(random_output_len)
@@ -403,8 +412,17 @@ def _forward_stream(stream) -> None:
         result_dir = "./"
 
     result_path = os.path.join(result_dir, result_filename)
-    with open(result_path, encoding="utf-8") as f:
-        result = json.load(f)
+    if not os.path.exists(result_path):
+        with open(OMNI_RESULT_TEMPLATE_PATH, encoding="utf-8") as f:
+            template_result: dict[str, Any] = json.load(f)
+        Path(result_path).parent.mkdir(parents=True, exist_ok=True)
+        with open(result_path, "w", encoding="utf-8") as f:
+            json.dump(template_result, f, ensure_ascii=False, indent=2)
+        print(f"Benchmark result file not generated, fallback to template: {result_path}")
+        result = template_result
+    else:
+        with open(result_path, encoding="utf-8") as f:
+            result = json.load(f)
 
     if baseline_config:
         result["baseline"] = _baseline_thresholds_for_step(
@@ -432,3 +450,12 @@ def pytest_addoption(parser: pytest.Parser) -> None:
         default=None,
         help=("Path to benchmark config JSON. Example: --test-config-file tests/dfx/perf/tests/test_tts.json"),
     )
+    parser.addoption(
+        "--assert-baseline",
+        action="store_true",
+        default=False,
+        help=(
+            "When set, omni/diffusion perf runners compare metrics against the baseline block in the JSON config "
+            "(default: off)."
+        ),
+    )

tests/dfx/perf/scripts/run_benchmark.py

@@ -12,12 +12,17 @@
     create_unique_server_params,
     get_benchmark_params_for_server,
     load_configs,
+    resolve_baseline_value,
     run_benchmark,
 )
 from tests.helpers.runtime import OmniServer
 
 pytestmark = [pytest.mark.full_model]
 
+# Compare metrics to each test JSON ``baseline`` block only when pytest is run with ``--assert-baseline``
+# (registered in ``tests/dfx/conftest.py``; default: off). ``run_benchmark`` and ``_resolve_baseline_value`` are
+# defined in the same module.
+
 
 os.environ["VLLM_WORKER_MULTIPROC_METHOD"] = "spawn"
 
@@ -46,7 +51,6 @@ def _get_config_file_from_argv() -> str | None:
     CONFIG_FILE_PATH = _DEFAULT_CONFIG_FILE
 
 BENCHMARK_CONFIGS = load_configs(CONFIG_FILE_PATH)
-OMNI_RESULT_TEMPLATE_PATH = Path(__file__).parent / "result_omni_template.json"
 
 
 DEPLOY_CONFIGS_DIR = Path(__file__).parent.parent / "deploy"
@@ -87,16 +91,6 @@ def omni_server(request):
         print("OmniServer stopped")
 
 
-def _safe_filename_token(value: Any | None, *, default: str = "na") -> str:
-    """Make a single path segment safe for result filenames on common filesystems."""
-    if value is None:
-        return default
-    s = str(value).strip()
-    for bad in ("/", "\\", ":", "*", "?", '"', "<", ">", "|"):
-        s = s.replace(bad, "_")
-    return s if s else default
-
-
 benchmark_indices = create_benchmark_indices(BENCHMARK_CONFIGS, server_to_benchmark_mapping)
 
 
@@ -126,76 +120,23 @@ def benchmark_params(request, omni_server):
     }
 
 
-def _resolve_baseline_value(
-    baseline_raw: Any,
-    *,
-    sweep_index: int | None,
-    max_concurrency: Any = None,
-    request_rate: Any = None,
-) -> Any:
-    """Pick the baseline threshold for this sweep step.
-
-    Supported shapes per metric:
-    - **Scalar** — same threshold for every concurrency / QPS.
-    - **List** — aligned with ``max_concurrency`` / ``request_rate`` sweep order; use ``sweep_index``.
-    - **Dict** — keyed by concurrency or rate, e.g. ``{"1": 500, "4": 800}`` (keys are strings in JSON).
-
-    For dict lookup, ``max_concurrency`` is preferred when both are set (concurrency sweep).
-    """
-    if baseline_raw is None:
-        # If no baseline is set, the maximum value will be used.
-        return 100000
-    if isinstance(baseline_raw, dict):
-        if max_concurrency is not None:
-            for key in (max_concurrency, str(max_concurrency)):
-                if key in baseline_raw:
-                    return baseline_raw[key]
-        if request_rate is not None:
-            for key in (request_rate, str(request_rate)):
-                if key in baseline_raw:
-                    return baseline_raw[key]
-        raise KeyError(
-            f"baseline dict has no key for max_concurrency={max_concurrency!r} "
-            f"or request_rate={request_rate!r}; keys={list(baseline_raw.keys())!r}"
-        )
-    if isinstance(baseline_raw, (list, tuple)):
-        return baseline_raw[sweep_index]
-    return baseline_raw
-
-
-def _baseline_thresholds_for_step(
-    baseline_data: dict[str, Any],
-    *,
-    sweep_index: int | None = None,
-    max_concurrency: Any = None,
-    request_rate: Any = None,
-) -> dict[str, Any]:
-    """Resolve ``test.json`` ``baseline`` block to one threshold per metric (same as ``assert_result``)."""
-    return {
-        metric_name: _resolve_baseline_value(
-            baseline_raw,
-            sweep_index=sweep_index,
-            max_concurrency=max_concurrency,
-            request_rate=request_rate,
-        )
-        for metric_name, baseline_raw in baseline_data.items()
-    }
-
-
 def assert_result(
     result,
     params,
     num_prompt,
     *,
+    assert_baseline: bool,
     sweep_index: int | None = None,
-    max_concurrency: Any = None,
-    request_rate: Any = None,
+    max_concurrency: Any | None = None,
+    request_rate: Any | None = None,
 ) -> None:
     assert result["completed"] == num_prompt, "Request failures exist"
+    if not assert_baseline:
+        return
     baseline_data = params.get("baseline", {})
     for metric_name, baseline_raw in baseline_data.items():
         current_value = result[metric_name]
-        baseline_value = _resolve_baseline_value(
+        baseline_value = resolve_baseline_value(
             baseline_raw,
             sweep_index=sweep_index,
             max_concurrency=max_concurrency,
@@ -214,7 +155,7 @@ def assert_result(
 @pytest.mark.benchmark
 @pytest.mark.parametrize("omni_server", test_params, indirect=True)
 @pytest.mark.parametrize("benchmark_params", benchmark_indices, indirect=True)
-def test_performance_benchmark(omni_server, benchmark_params):
+def test_performance_benchmark(omni_server, benchmark_params, request):
     test_name = benchmark_params["test_name"]
     params = benchmark_params["params"]
     dataset_name = params.get("dataset_name", "")
@@ -226,6 +167,8 @@ def test_performance_benchmark(omni_server, benchmark_params):
     print(f"Running benchmark for model: {model}")
     print(f"Benchmark parameters: {benchmark_params}")
 
+    assert_baseline = request.config.getoption("--assert-baseline", default=False)
+
     def to_list(value, default=None):
         if value is None:
             return [] if default is None else [default]
@@ -264,7 +207,7 @@ def to_list(value, default=None):
         elif not isinstance(value, bool):
             args.extend([arg_name, str(value)])
 
-    # QPS test (sweep_index aligns with qps_list / num_prompt_list for this loop)
+    # QPS / request-rate sweep
     for i, (qps, num_prompt) in enumerate(zip(qps_list, num_prompt_list)):
         args = args + ["--request-rate", str(qps), "--num-prompts", str(num_prompt)]
         result = run_benchmark(
@@ -283,12 +226,13 @@ def to_list(value, default=None):
         assert_result(
             result,
             params,
-            num_prompt=num_prompt,
+            num_prompt,
+            assert_baseline=assert_baseline,
             sweep_index=i,
             request_rate=qps,
         )
 
-    # concurrency test (sweep_index aligns with max_concurrency_list for separate thresholds per concurrency)
+    # concurrency test
     for i, (concurrency, num_prompt) in enumerate(zip(max_concurrency_list, num_prompt_list)):
         args = args + ["--max-concurrency", str(concurrency), "--num-prompts", str(num_prompt), "--request-rate", "inf"]
         result = run_benchmark(
@@ -307,7 +251,8 @@ def to_list(value, default=None):
         assert_result(
             result,
             params,
-            num_prompt=num_prompt,
+            num_prompt,
+            assert_baseline=assert_baseline,
             sweep_index=i,
             max_concurrency=concurrency,
         )

tests/dfx/perf/scripts/run_diffusion_benchmark.py

@@ -11,6 +11,8 @@
 A config JSON file is REQUIRED via --test-config-file:
   pytest run_diffusion_benchmark.py --test-config-file tests/dfx/perf/tests/test_qwen_image_vllm_omni.json
 
+Optional: ``--assert-baseline`` compares metrics to the ``baseline`` block in each benchmark entry (default: off).
+
 All benchmark results for a session are consolidated into a single JSON file under
 BENCHMARK_RESULT_DIR (override via the DIFFUSION_BENCHMARK_DIR environment variable).
 Each entry in the file contains the test metadata (test_name, endpoint, benchmark_params,
@@ -652,12 +654,14 @@ def run_benchmark(
 # ---------------------------------------------------------------------------
 
 
-def assert_result(result: dict[str, Any], params: dict[str, Any]) -> None:
-    """Assert that benchmark metrics satisfy the configured baselines."""
+def assert_result(result: dict[str, Any], params: dict[str, Any], *, assert_baseline: bool) -> None:
+    """Assert request completion; optionally assert metrics against ``baseline`` in *params*."""
     num_prompts = params.get("num-prompts", 10)
     completed = result.get("completed_requests", result.get("completed", 0))
     assert completed == num_prompts, f"Expected {num_prompts} completed requests, got {completed}"
 
+    if not assert_baseline:
+        return
     if params.get("skip-performance-assertion", False):
         print("Skipping performance assertions.")
         return
@@ -699,16 +703,14 @@ def _resolve_benchmark_endpoint(server_cfg: dict[str, Any], params: dict[str, An
     indirect=True,
 )
 @pytest.mark.parametrize("benchmark_params", benchmark_indices, indirect=True)
-def test_diffusion_performance_benchmark(diffusion_server, benchmark_params):
-    """Run the diffusion performance benchmark and assert against baselines.
+def test_diffusion_performance_benchmark(diffusion_server, benchmark_params, request):
+    """Run the diffusion performance benchmark and verify request completion.
 
     One server is started per unique parallel configuration (module scope).
     For each server, all benchmark parameter sets defined in the config JSON
-    are executed sequentially; results are asserted against the baselines.
+    are executed sequentially; metrics are recorded to the aggregated result file.
 
-    Tracked metrics:
-        - throughput_qps          (higher is better)
-        - latency_p50, latency_p99 (lower is better)
+    Pass ``--assert-baseline`` to compare ``throughput_qps``, ``latency_mean``, etc. to the JSON ``baseline`` block.
     """
     test_name = benchmark_params["test_name"]
     params = benchmark_params["params"]
@@ -744,4 +746,5 @@ def test_diffusion_performance_benchmark(diffusion_server, benchmark_params):
     print(f"\n  Aggregated results: {AGGREGATED_RESULT_FILE}")
     print("=" * 60)
 
-    assert_result(result, params)
+    assert_baseline = request.config.getoption("--assert-baseline", default=False)
+    assert_result(result, params, assert_baseline=assert_baseline)