Enable determinism check in CI

.ci/determinism_report.py

@@ -0,0 +1,121 @@
+# Copyright Advanced Micro Devices, Inc.
+# SPDX-License-Identifier: MIT
+
+"""Schema and I/O for the xFuser determinism check report.
+
+Written by ``run.py`` after each experiment that ran with the determinism
+check enabled, and read by ``.github/actions/check-determinism-report`` to
+decide whether a benchmark run's architecture should be treated as failed.
+Both sides import this module so the report's shape has a single definition.
+This module deliberately holds only raw data and its I/O: how a report gets
+rendered (e.g. icons, summary tables) and what counts as a failure are
+policy decisions left to the action that consumes the report.
+"""
+
+import json
+import logging
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+REPORT_FILENAME = "determinism_report.json"
+
+STATUS_PASSED = "passed"
+STATUS_FAILED = "failed"
+STATUS_ERROR = "error"
+STATUS_UNAVAILABLE = "unavailable"
+
+logger = logging.getLogger(__name__)
+logger.setLevel(logging.INFO)
+if not logger.handlers:
+    _handler = logging.StreamHandler()
+    _handler.setFormatter(
+        logging.Formatter(
+            fmt="%(asctime)s - %(levelname)s: %(message)s",
+            datefmt="%Y-%m-%d %H:%M:%S",
+        )
+    )
+    logger.addHandler(_handler)
+    logger.propagate = False
+
+
+def make_entry(
+    name: str,
+    status: str = STATUS_UNAVAILABLE,
+    failed_checks: int = 0,
+    dump_bytes: int = 0,
+    output_directory: str = "",
+) -> Dict[str, Any]:
+    """Build one experiment's entry in the report's ``experiments`` list.
+
+    Args:
+        name: Name of the experiment the entry describes.
+        status: One of the ``STATUS_*`` constants.
+        failed_checks: Number of determinism checks that failed.
+        dump_bytes: Bytes occupied by dumps produced by failed checks.
+        output_directory: Path to the experiment's output directory.
+
+    Returns:
+        A dictionary with the entry's fields, ready to be placed in the
+        report's ``experiments`` list.
+    """
+    return {
+        "name": name,
+        "status": status,
+        "failed_checks": failed_checks,
+        "dump_bytes": dump_bytes,
+        "output_directory": output_directory,
+    }
+
+
+def write_report(results_directory: Path, entries: List[Dict[str, Any]]) -> Optional[Path]:
+    """Build and write the determinism report, logging the outcome.
+
+    Aggregates `entries` into a report and writes it to
+    ``<results_directory>/determinism_report.json``, creating the directory
+    if necessary. Failures are logged rather than raised, so this can be
+    called as a fire-and-forget step at the end of a benchmark run.
+
+    Args:
+        results_directory: Directory the report is written into.
+        entries: Per-experiment entries, as built by `make_entry`.
+
+    Returns:
+        The path the report was written to, or None if writing failed.
+    """
+    results_directory = Path(results_directory)
+    path = results_directory / REPORT_FILENAME
+    report = {
+        "experiments": entries,
+        "failed_experiments": sum(1 for e in entries if e["status"] == STATUS_FAILED),
+        "failed_checks": sum(e["failed_checks"] for e in entries),
+    }
+    try:
+        results_directory.mkdir(parents=True, exist_ok=True)
+        path.write_text(json.dumps(report, indent=2) + "\n")
+    except OSError as exc:
+        logger.error("Failed to write determinism report to %s: %s", path, exc)
+        return None
+
+    logger.info(
+        "Determinism report written to %s (%d of %d experiments failed).",
+        path,
+        report["failed_experiments"],
+        len(entries),
+    )
+    return path
+
+
+def load_report(results_directory: Path) -> Optional[Dict[str, Any]]:
+    """Load the determinism report from a results directory.
+
+    Args:
+        results_directory: Directory expected to hold
+            ``determinism_report.json``.
+
+    Returns:
+        The parsed report, or None if the file doesn't exist.
+    """
+    path = Path(results_directory) / REPORT_FILENAME
+    if not path.is_file():
+        return None
+    return json.loads(path.read_text())

.ci/run.py

@@ -23,6 +23,7 @@
 from huggingface_hub import snapshot_download, scan_cache_dir, DryRunFileInfo
 from memory_sampler import MemorySampler
 from miopen_driver_commands import MIOpenDriverCommandCollector, env_enabled
+import determinism_report
 
 
 logger = logging.getLogger(__name__)
@@ -324,12 +325,21 @@ def _delete_model_cache(model: str, revision: Optional[str] = None, dry_run: boo
 
 
 def _is_determinism_check_enabled(exp: Experiment) -> bool:
-    """Assuming that xFuser doesn't enable determinism check by default, and that the `exp.args`
-    has configuration arguments AND `--override-args-json` already merged, checks if the check
-    is enabled by the experiment config.
+    """Decide whether the determinism check should be reported for an experiment.
+
+    Assumes that xFuser doesn't enable the determinism check by default, and
+    that `exp.args` has configuration arguments AND `--override-args-json`
+    already merged.
+
+    Args:
+        exp: The experiment to check.
+
+    Returns:
+        True if `CI_RUN_PY_FORCE_DETERMINISM_CHECK` is set, or if the
+        experiment's own args enable `determinism_check`.
     """
     # global overrider
-    if os.environ.get("CI_RUN_PY_FORCE_DETERMINISM_REPORT", "0") == "1":
+    if os.environ.get("CI_RUN_PY_FORCE_DETERMINISM_CHECK", "0") == "1":
         return True
 
     ret = False
@@ -346,38 +356,62 @@ def _is_determinism_check_enabled(exp: Experiment) -> bool:
     return ret
 
 
-def _report_determinism_check_results(exp: Experiment, benchmark_output_directory: Path) -> None:
+def _report_determinism_check_results(exp: Experiment, benchmark_output_directory: Path) -> Dict[str, Any]:
+    """Report determinism check results for one experiment.
+
+    Args:
+        exp: The experiment the determinism check ran for.
+        benchmark_output_directory: The experiment's output directory, where
+            xFuser leaves whatever determinism check results it produced.
+
+    Returns:
+        A summary entry built by `determinism_report.make_entry`.
+    """
     global tried_import_xfuser_determinism_check_results, determinism_check_results, readable_bytes
     if not tried_import_xfuser_determinism_check_results:
         tried_import_xfuser_determinism_check_results = True
         determinism_check_results, readable_bytes = import_xfuser_determinism_check_results()
         if readable_bytes is None:
             readable_bytes = lambda x: f"{x} bytes"  # noqa: E731
 
-    if determinism_check_results is None:  # import must have failed
-        return
+    status = determinism_report.STATUS_UNAVAILABLE
+    failed_checks = 0
+    dump_bytes = 0
 
-    try:
-        det_check = determinism_check_results(benchmark_output_directory)
-        if not isinstance(det_check, dict):
-            logger.error(f"Expected a dictionary in results report, got {type(det_check)}")
-            return
-        if 0 == len(det_check):
-            logger.info(f"Determinism checks passed for {exp.name}")
-            return
-
-        if 1 != len(det_check):
-            logger.error("Expected exactly one top-level directory in results report")
-        if "" in det_check:
-            res = det_check[""]
-            logger.warning(
-                f"Determinism check failed for {exp.name}: {res[0]} checks failed, "
-                f"{readable_bytes(res[1])} is occupied by dumps"
-            )
-        else:
-            logger.error("Results for the top-level directory not found. Skipping report.")
-    except Exception:  # ruff: ignore[blind-except]
-        logger.error("Error getting determinism check results.", exc_info=True)
+    if determinism_check_results is not None:  # otherwise the import failed
+        try:
+            det_check = determinism_check_results(benchmark_output_directory)
+            if not isinstance(det_check, dict):
+                logger.error(f"Expected a dictionary in results report, got {type(det_check)}")
+                status = determinism_report.STATUS_ERROR
+            elif 0 == len(det_check):
+                logger.info(f"Determinism checks passed for {exp.name}")
+                status = determinism_report.STATUS_PASSED
+            elif "" in det_check:
+                if 1 != len(det_check):
+                    logger.error("Expected exactly one top-level directory in results report")
+                res = det_check[""]
+                logger.warning(
+                    f"Determinism check failed for {exp.name}: {res[0]} checks failed, "
+                    f"{readable_bytes(res[1])} is occupied by dumps"
+                )
+                status = determinism_report.STATUS_FAILED
+                failed_checks = int(res[0])
+                dump_bytes = int(res[1])
+            else:
+                logger.error("Results for the top-level directory not found. Skipping report.")
+                status = determinism_report.STATUS_ERROR
+        except Exception:  # ruff: ignore[blind-except]
+            logger.error("Error getting determinism check results.", exc_info=True)
+            status = determinism_report.STATUS_ERROR
+
+    return determinism_report.make_entry(
+        exp.name,
+        status=status,
+        failed_checks=failed_checks,
+        dump_bytes=dump_bytes,
+        output_directory=str(benchmark_output_directory),
+    )
 
 
 def _run_experiment(
@@ -423,9 +457,6 @@ def _run_experiment(
         logger.info(f"Experiment {exp.name} failed!")
         return False
 
-    if _is_determinism_check_enabled(exp):
-        _report_determinism_check_results(exp, benchmark_output_directory)
-
     logger.info(f"Experiment: {exp.name} completed successfully.")
     return True
 
@@ -671,6 +702,7 @@ def main():
     # Download models and run Experiments
     preserve_original_state = not args.clear_model_cache and not args.no_clear_model_cache
     timing: Dict[str, Any] = {"download_model": {}, "experiments": []}
+    determinism_entries: List[Dict[str, Any]] = []
 
     override_args = json.loads(args.override_args_json)
     # assumes `override_args` aren't mutated in the loop below
@@ -736,6 +768,9 @@ def main():
                 logger.error(msg)
                 continue
 
+            if not args.dry_run and _is_determinism_check_enabled(exp):
+                determinism_entries.append(_report_determinism_check_results(exp, benchmark_output_directory))
+
             if not args.dry_run:
                 latency_output_filepath = Path(benchmark_output_directory) / "timings.json" # benchmark scripts are expected to write latencies to "timings.json"
                 median_latency = _get_median_latency(latency_output_filepath)
@@ -765,6 +800,9 @@ def main():
             except Exception as e:
                 logger.error(e, stack_info=True, exc_info=True)
 
+    if determinism_entries:
+        determinism_report.write_report(Path(args.results_directory), determinism_entries)
+
     if args.print_timing_summary and (timing.get("download_model") or timing.get("experiments")):
         _print_timing_summary(timing)
 

.github/actions/check-determinism-report/action.yml

@@ -0,0 +1,26 @@
+# Copyright Advanced Micro Devices, Inc.
+# SPDX-License-Identifier: MIT
+
+name: Check Determinism Report
+description: >
+  Summarise the determinism report of one architecture and fail when any
+  benchmark was non-deterministic, so GitHub's own run-failure notification
+  reaches whoever dispatched the run.
+
+inputs:
+  results_dir:
+    description: 'Benchmark results directory holding determinism_report.json'
+    required: true
+  arch:
+    description: 'GPU tag from the matrix (e.g. gfx942, mi300, gfx950)'
+    required: true
+
+runs:
+  using: composite
+  steps:
+    - name: Check determinism report
+      shell: bash
+      env:
+        RESULTS_DIR: ${{ inputs.results_dir }}
+        ARCH: ${{ inputs.arch }}
+      run: python3 "$GITHUB_ACTION_PATH/scripts/check-determinism.py"