From b34e9859b148f6cf2b048179354c1f0328dc291e Mon Sep 17 00:00:00 2001 From: Jin Pan Date: Wed, 29 Jul 2026 23:01:48 +0000 Subject: [PATCH 1/3] fix(tests): exclude GPU drain from dispatch timing Move the final synchronization outside the measured window so the benchmark matches its documented host-dispatch semantics, and lock the ordering with a regression test. --- tests/unit/test_launch_overhead.py | 19 ++++++++++++++++++- 1 file changed, 18 insertions(+), 1 deletion(-) diff --git a/tests/unit/test_launch_overhead.py b/tests/unit/test_launch_overhead.py index a1a4c44cb..3f4e7deab 100644 --- a/tests/unit/test_launch_overhead.py +++ b/tests/unit/test_launch_overhead.py @@ -112,12 +112,29 @@ def bench_wallclock(fn, n_warmup=20, n_iters=1000): t0 = time.perf_counter() for _ in range(n_iters): fn() - torch.cuda.synchronize() t1 = time.perf_counter() + torch.cuda.synchronize() return (t1 - t0) / n_iters * 1e6 # µs +def test_bench_wallclock_excludes_final_gpu_drain(monkeypatch): + """The final drain must not be part of the host-dispatch window.""" + sync_count = 0 + + def fake_synchronize(): + nonlocal sync_count + sync_count += 1 + + monkeypatch.setattr(torch.cuda, "synchronize", fake_synchronize) + monkeypatch.setattr(time, "perf_counter", lambda: float(sync_count)) + + measured_us = bench_wallclock(lambda: None, n_warmup=0, n_iters=1) + + assert measured_us == 0.0 + assert sync_count == 2 + + def main(): SIZE = 1024 * 256 # 256K elements — small enough that GPU time is trivial BLOCK = 256 From 056cdeaae7e0b4cb413a41f6c56e5500b933632b Mon Sep 17 00:00:00 2001 From: Jin Pan Date: Thu, 30 Jul 2026 00:33:21 +0000 Subject: [PATCH 2/3] feat(bench): unify timing contracts and CI gates Consolidate equivalent event timers behind the shipped do_bench implementation while preserving distinct profiler, host, graph, and distributed semantics. Carry raw latency and measurement contracts into CI so calibrated regressions fail closed without breaking existing dashboard output. --- .github/benchmark_thresholds.json | 53 ++ .github/dashboard/ingest/ingest.py | 7 +- .github/dashboard/ingest/test_ingest.py | 40 ++ .github/workflows/flydsl.yaml | 34 +- docs/testing_benchmarking_guide.md | 71 ++- python/flydsl/__init__.py | 7 +- python/flydsl/autotune.py | 207 +++++++- scripts/benchmark_compare.py | 75 +++ scripts/benchmark_log_parser.py | 190 +++++++ scripts/benchmark_output_to_csv.py | 44 +- scripts/compare_benchmark.py | 153 +++++- scripts/run_benchmark.sh | 185 ++----- tests/kernels/benchmark_common.py | 87 +--- tests/kernels/compare_allreduce_benchmark.py | 17 +- tests/kernels/test_fused_rope_cache.py | 20 +- tests/kernels/test_layernorm.py | 9 + .../kernels/test_moe_a8w4_mxscale_gfx1250.py | 22 +- tests/kernels/test_moe_sorting.py | 193 ++++--- tests/kernels/test_rmsnorm.py | 9 + tests/kernels/test_softmax.py | 9 + tests/perf/bench_tdm_bandwidth_gfx1250.py | 44 +- tests/unit/test_benchmark_compare.py | 486 ++++++++++++++++++ tests/unit/test_benchmark_log_parser.py | 118 +++++ tests/unit/test_benchmark_timer.py | 202 ++++++++ tests/unit/test_launch_overhead.py | 15 +- tests/unit/test_tdm_mcast_add_gfx1250.py | 23 +- 26 files changed, 1918 insertions(+), 402 deletions(-) create mode 100644 .github/benchmark_thresholds.json create mode 100644 scripts/benchmark_compare.py create mode 100644 scripts/benchmark_log_parser.py create mode 100644 tests/unit/test_benchmark_compare.py create mode 100644 tests/unit/test_benchmark_log_parser.py create mode 100644 tests/unit/test_benchmark_timer.py diff --git a/.github/benchmark_thresholds.json b/.github/benchmark_thresholds.json new file mode 100644 index 000000000..86773168a --- /dev/null +++ b/.github/benchmark_thresholds.json @@ -0,0 +1,53 @@ +{ + "version": 1, + "calibration": "Initial allowlist uses a 20% relative threshold above the documented ~14% gfx950 clock-variation band, plus a 10 us absolute floor.", + "architectures": { + "gfx942": [ + { + "op": "softmax", + "shape": "32768x8192", + "dtype": "bf16", + "relative_pct": 20.0, + "absolute_us": 10.0 + }, + { + "op": "layernorm", + "shape": "32768x8192", + "dtype": "bf16", + "relative_pct": 20.0, + "absolute_us": 10.0 + }, + { + "op": "rmsnorm", + "shape": "32768x8192", + "dtype": "bf16", + "relative_pct": 20.0, + "absolute_us": 10.0 + } + ], + "gfx950": [ + { + "op": "softmax", + "shape": "32768x8192", + "dtype": "bf16", + "relative_pct": 20.0, + "absolute_us": 10.0 + }, + { + "op": "layernorm", + "shape": "32768x8192", + "dtype": "bf16", + "relative_pct": 20.0, + "absolute_us": 10.0 + }, + { + "op": "rmsnorm", + "shape": "32768x8192", + "dtype": "bf16", + "relative_pct": 20.0, + "absolute_us": 10.0 + } + ], + "gfx1201": [] + } +} diff --git a/.github/dashboard/ingest/ingest.py b/.github/dashboard/ingest/ingest.py index c7037ed06..8f9087e16 100755 --- a/.github/dashboard/ingest/ingest.py +++ b/.github/dashboard/ingest/ingest.py @@ -147,8 +147,11 @@ def ingest_run(repo: str, run: dict, regression_pct: float) -> tuple[list[dict], "url": job.get("html_url"), } job_status.append(js) - if job.get("status") != "completed" or job.get("conclusion") != "success": - continue # only completed-successful jobs have parseable benchmark output + if job.get("status") != "completed" or job.get("conclusion") not in {"success", "failure"}: + continue + # Performance-gate failures still contain the benchmark table and + # comparison block that explain the regression. Parse those logs so a + # red CI result also remains visible in the dashboard. try: text = gh_text(f"repos/{repo}/actions/jobs/{job['id']}/logs") except RuntimeError as e: diff --git a/.github/dashboard/ingest/test_ingest.py b/.github/dashboard/ingest/test_ingest.py index 2d6b48b4c..a99b30064 100644 --- a/.github/dashboard/ingest/test_ingest.py +++ b/.github/dashboard/ingest/test_ingest.py @@ -194,6 +194,46 @@ def test_runner_of_matches_known_box_and_rejects_unknown(): assert ingest.runner_of("") is None +def test_ingest_run_parses_completed_failure_logs(monkeypatch): + job = { + "id": 9, + "name": "test (linux-flydsl-mi355-1)", + "status": "completed", + "conclusion": "failure", + "started_at": "2026-01-01T00:00:00Z", + "completed_at": "2026-01-01T00:01:00Z", + "html_url": "https://example.test/job/9", + } + monkeypatch.setattr(ingest, "run_jobs", lambda repo, run_id: [job]) + monkeypatch.setattr(ingest, "resolve_pr", lambda repo, run: 123) + monkeypatch.setattr( + ingest, + "gh_text", + lambda path: ( + "op shape dtype TB/s TFLOPS\n" + "softmax 32768x8192 bf16 4.000 -\n" + ), + ) + run = { + "id": 1, + "head_sha": "abc", + "head_branch": "feature", + "event": "pull_request", + "display_title": "timing", + "status": "completed", + "conclusion": "failure", + "html_url": "https://example.test/run/1", + "created_at": "2026-01-01T00:00:00Z", + "updated_at": "2026-01-01T00:01:00Z", + "actor": {"login": "user"}, + } + + records, summary = ingest.ingest_run("ROCm/FlyDSL", run, regression_pct=-3.0) + + assert records and records[0]["op"] == "softmax" + assert summary["jobs"][0]["conclusion"] == "failure" + + # --------------------------------------------------------------------------- # # list_runs — default scans all branches (so PR runs are included) # --------------------------------------------------------------------------- # diff --git a/.github/workflows/flydsl.yaml b/.github/workflows/flydsl.yaml index 2d038cefb..2fd164709 100644 --- a/.github/workflows/flydsl.yaml +++ b/.github/workflows/flydsl.yaml @@ -233,6 +233,7 @@ jobs: GIT_CONFIG_NOSYSTEM: "1" # Temporary quarantine for linux-flydsl-navi-2 GPU 1; see #858. CI_HIP_VISIBLE_DEVICES: ${{ contains(matrix.runners, 'navi') && '3' || '' }} + BENCHMARK_ARCH: ${{ contains(matrix.runners, 'mi355') && 'gfx950' || (contains(matrix.runners, 'mi325') && 'gfx942' || 'gfx1201') }} strategy: matrix: runners: [ @@ -342,9 +343,16 @@ jobs: set -e -o pipefail export PYTHONPATH=/tmp/aiter:${PYTHONPATH:-} export AITER_REPO=/tmp/aiter + export FLYDSL_PERFTEST_USE_EVENTS=0 cd /flydsl-test - BENCH_LOG_DIR=/tmp/flydsl_bench_current bash scripts/run_benchmark.sh 2>&1 | tee /tmp/bench_current.out - python3 scripts/benchmark_output_to_csv.py /tmp/bench_current.out /tmp/bench_current.csv + python3 -m pytest --collect-only -q -m benchmark \ + tests/kernels/test_rmsnorm.py \ + tests/kernels/test_allreduce.py \ + tests/kernels/test_moe_a8w4_mxscale_gfx1250.py \ + tests/unit/test_tdm_mcast_add_gfx1250.py + BENCH_LOG_DIR=/tmp/flydsl_bench_current \ + BENCH_OUTPUT_CSV=/tmp/bench_current.csv \ + bash scripts/run_benchmark.sh 2>&1 | tee /tmp/bench_current.out BASH - name: Run benchmark baselines @@ -378,12 +386,15 @@ jobs: ( set -e -o pipefail cd "${worktree}" + cp /flydsl-test/scripts/run_benchmark.sh scripts/run_benchmark.sh + cp /flydsl-test/scripts/benchmark_log_parser.py scripts/benchmark_log_parser.py export MLIR_PATH=/llvm-project/mlir_install python3 -m pip install -e . --use-pep517 2>&1 | tail -5 export PYTHONPATH=/tmp/aiter:${PYTHONPATH:-} export AITER_REPO=/tmp/aiter - BENCH_LOG_DIR="${log_dir}" bash scripts/run_benchmark.sh 2>&1 | tee "${output}" - python3 /flydsl-test/scripts/benchmark_output_to_csv.py "${output}" "${csv}" + export FLYDSL_PERFTEST_USE_EVENTS=0 + BENCH_LOG_DIR="${log_dir}" BENCH_OUTPUT_CSV="${csv}" \ + bash scripts/run_benchmark.sh 2>&1 | tee "${output}" ) status=$? if [ "${status}" -eq 0 ] && [ -s "${csv}" ]; then @@ -418,8 +429,12 @@ jobs: python3 -m pip install --only-binary=:all: "flydsl==${package_version}" 2>&1 | tail -5 export PYTHONPATH=/tmp/aiter:${PYTHONPATH:-} export AITER_REPO=/tmp/aiter - BENCH_LOG_DIR="${log_dir}" bash scripts/run_benchmark.sh 2>&1 | tee "${output}" - python3 /flydsl-test/scripts/benchmark_output_to_csv.py "${output}" "${csv}" + export FLYDSL_PERFTEST_USE_EVENTS=0 + BENCH_LOG_DIR="${log_dir}" BENCH_OUTPUT_CSV="${csv}" \ + bash scripts/run_benchmark.sh 2>&1 | tee "${output}" + if [ ! -s "${csv}" ]; then + python3 /flydsl-test/scripts/benchmark_output_to_csv.py "${output}" "${csv}" + fi ) status=$? if [ "${status}" -eq 0 ] && [ -s "${csv}" ]; then @@ -451,11 +466,14 @@ jobs: cd /flydsl-test main_label=\$(cat /tmp/bench_main_label 2>/dev/null || echo main) python3 scripts/compare_benchmark.py /tmp/bench_main.csv /tmp/bench_current.csv \ - --baseline-label \"\${main_label}\" --current-label current + --baseline-label \"\${main_label}\" --current-label current \ + --arch '${{ env.BENCHMARK_ARCH }}' \ + --threshold-config .github/benchmark_thresholds.json \ + --fail-on-regression " - name: Check benchmark performance (current vs latest tag) - if: steps.bench-baselines.outcome != 'skipped' + if: always() && steps.bench-baselines.outcome != 'skipped' timeout-minutes: 5 run: | docker exec flydsl_test bash -c " diff --git a/docs/testing_benchmarking_guide.md b/docs/testing_benchmarking_guide.md index 4a2b282e0..4810fe5b9 100644 --- a/docs/testing_benchmarking_guide.md +++ b/docs/testing_benchmarking_guide.md @@ -207,7 +207,8 @@ def my_kernel_test(Input, Output): Features: - Device memory profiling to determine rotation count -- Torch CUDA event timing +- `torch.profiler` device-time attribution by default +- Pipelined CUDA/HIP event fallback with `FLYDSL_PERFTEST_USE_EVENTS=1` - HIPGraph capture mode (`testGraph=True`) - Cache-aware iteration calculation @@ -225,7 +226,8 @@ High-level validation wrapper around `checkAllclose`. ### 4.2 `tests/kernels/benchmark_common.py` -Shared benchmark harness for performance comparison. +Compatibility wrappers and performance-comparison formatting. Event timing +delegates to the canonical `flydsl.do_bench` implementation. **Key functions:** ```python @@ -233,6 +235,35 @@ Shared benchmark harness for performance comparison. gpu_us = bench_gpu_us_torch(fn, warmup=20, iters=200) ``` +### 4.3 Canonical event timer + +`flydsl.do_bench` is the single implementation for eager CUDA/HIP event +measurement. Its explicit schedule keeps physically different experiments from +being hidden behind ambiguous helper names: + +```python +from flydsl import do_bench + +result = do_bench( + fn, + warmup=20, + rep=200, + schedule="pipelined", # or "per_iter" / "isolated" + statistic="mean", + return_result=True, +) +print(result.value_us, result.samples_us) +``` + +`BenchResult` always stores microseconds and records the schedule, statistic, +cache policy, warmup, and iteration count. The historical +`flydsl.autotune.do_bench(fn, warmup, rep, quantiles)` scalar interface remains +available and returns milliseconds for backward compatibility. + +Do not replace `run_perftest` with this helper when profiler attribution is the +question: profiler dwell, event elapsed time, host wall clock, and graph replay +are separate instruments. + --- ## 5. Test Utilities (`tests/utils.py`) @@ -308,7 +339,7 @@ def test_my_kernel(): ### 6.3 Benchmark Test Pattern ```python -from tests.kernels.benchmark_common import bench_gpu_us_torch +from flydsl import do_bench def benchmark_my_kernel(): # Setup @@ -318,7 +349,15 @@ def benchmark_my_kernel(): launch_fn(input_tensor, output_tensor) # Measure - gpu_us = bench_gpu_us_torch(run, warmup=20, iters=200) + result = do_bench( + run, + warmup=20, + rep=200, + schedule="pipelined", + statistic="mean", + return_result=True, + ) + gpu_us = result.value_us # Compute metrics total_bytes = 2 * M * N * elem_size @@ -326,6 +365,23 @@ def benchmark_my_kernel(): print(f"Time: {gpu_us:.1f} us, Bandwidth: {bandwidth_tbs:.2f} TB/s") ``` +### 6.4 Benchmark CI records and gates + +`run_benchmark.sh --output_csv PATH` preserves the existing throughput columns +and appends normalized `avg_us`, sample metadata, measurement semantics, and +GPU architecture. Human-readable stdout intentionally keeps its original +five-column format for dashboard compatibility. + +The general comparator uses the same relative-AND-absolute rule as the +allreduce gate. Hard failures are limited to the stable per-architecture +allowlist in `.github/benchmark_thresholds.json`; uncalibrated rows remain +report-only. Missing baselines skip the comparison rather than failing a PR. + +Initial gfx942/gfx950 rows use a 20% relative threshold plus a 10 us absolute +floor. The relative threshold is intentionally wider than the documented +approximately 14% gfx950 clock-variation band and must be recalibrated from CI +history before adding more rows. + --- ## 7. GEMM Test CLI Arguments @@ -358,6 +414,7 @@ python tests/kernels/test_preshuffle_gemm.py \ | `FLYDSL_RUNTIME_CACHE_DIR` | Compiler | Cache directory (default: `~/.flydsl/cache`) | | `RUN_TESTS_FULL` | `run_tests.sh` | Set to `1` to run all parametrized cases | | `BENCH_LOG_DIR` | `run_benchmark.sh` | Benchmark log directory (default: `/tmp/flydsl_bench`) | +| `BENCH_OUTPUT_CSV` | `run_benchmark.sh` | Write enriched CSV with raw us and measurement metadata | --- @@ -385,11 +442,15 @@ bash scripts/dumpir.sh |---|---| | `scripts/run_tests.sh` | Full test runner (pytest + examples + FileCheck) | | `scripts/run_benchmark.sh` | Benchmark harness with configurable shapes | +| `scripts/benchmark_log_parser.py` | Normalize benchmark logs, including raw latency in us | +| `scripts/compare_benchmark.py` | Throughput report plus calibrated raw-us regression gate | +| `.github/benchmark_thresholds.json` | Per-architecture hard-gate allowlist and thresholds | | `scripts/dumpir.sh` | IR dump helper script | | `tests/conftest.py` | Pytest fixtures (MLIR context, module, insert point) | | `tests/test_common.py` | `perftest()`, `checkAllclose()`, `verify_output()` | | `tests/utils.py` | `pertoken_quant()`, `shuffle_weight()` | -| `tests/kernels/benchmark_common.py` | `bench_gpu_us_torch()`, benchmark harness | +| `python/flydsl/autotune.py` | Canonical `do_bench()` event timer and `BenchResult` | +| `tests/kernels/benchmark_common.py` | Compatibility wrappers and benchmark formatting | | `tests/mlir/{LayoutAlgebra,Conversion,Transforms}/` | MLIR lit tests (18 files) | | `tests/python/examples/` | Python AOT examples | | `tests/kernels/test_*.py` | GPU kernel tests (12 files) | diff --git a/python/flydsl/__init__.py b/python/flydsl/__init__.py index fd80758ca..a0d51ce60 100644 --- a/python/flydsl/__init__.py +++ b/python/flydsl/__init__.py @@ -4,4 +4,9 @@ __version__ = "0.3.0" -from .autotune import Config as Config, autotune as autotune # noqa: E402 +from .autotune import ( # noqa: E402 + BenchResult as BenchResult, + Config as Config, + autotune as autotune, + do_bench as do_bench, +) diff --git a/python/flydsl/autotune.py b/python/flydsl/autotune.py index bd5853f88..a0aa15a48 100644 --- a/python/flydsl/autotune.py +++ b/python/flydsl/autotune.py @@ -8,8 +8,10 @@ import json import os from contextlib import nullcontext +from dataclasses import dataclass from pathlib import Path -from typing import Callable, Dict, List +from statistics import mean, median +from typing import Callable, Dict, List, Literal from .utils import env, log from .utils.file import atomic_write @@ -22,6 +24,39 @@ _ARTIFACT_VERSION = 1 +BenchSchedule = Literal["isolated", "pipelined", "per_iter"] +BenchStatistic = Literal["median", "median_average", "mean", "min", "max"] + + +@dataclass(frozen=True) +class BenchResult: + """A self-describing CUDA/HIP event benchmark result. + + Samples and the selected value are always stored in microseconds. Legacy + ``do_bench`` callers can continue requesting scalar milliseconds. + """ + + value_us: float + samples_us: tuple[float, ...] + instrument: str + schedule: BenchSchedule + statistic: BenchStatistic + warmup: int + iterations: int + cache_policy: str + + @property + def min_us(self) -> float: + return min(self.samples_us) + + @property + def max_us(self) -> float: + return max(self.samples_us) + + @property + def sample_count(self) -> int: + return len(self.samples_us) + def _tuning_enabled() -> bool: """Whether to bypass cached/default configs and run a fresh search.""" @@ -174,24 +209,162 @@ def from_dict(cls, d): ) -def do_bench(fn, warmup=5, rep=25, quantiles=None): - """Benchmark a GPU kernel using CUDA/HIP events. Returns median ms.""" - for _ in range(warmup): - fn() - torch.cuda.synchronize() - times = [] - for _ in range(rep): - start = torch.cuda.Event(enable_timing=True) - end = torch.cuda.Event(enable_timing=True) - start.record() - fn() - end.record() +def _filter_iqr(samples): + ordered = sorted(samples) + if len(ordered) < 8: + return ordered + q1, q3 = ordered[len(ordered) // 4], ordered[3 * len(ordered) // 4] + iqr = q3 - q1 + lo, hi = q1 - 1.5 * iqr, q3 + 1.5 * iqr + filtered = [sample for sample in ordered if lo <= sample <= hi] + return filtered or ordered + + +def _reduce_samples(samples, statistic): + if statistic == "median": + ordered = sorted(samples) + return float(ordered[len(ordered) // 2]) + if statistic == "median_average": + return float(median(samples)) + if statistic == "mean": + return float(mean(samples)) + if statistic == "min": + return float(min(samples)) + if statistic == "max": + return float(max(samples)) + raise ValueError(f"unsupported statistic: {statistic!r}") + + +def do_bench( + fn, + warmup=5, + rep=25, + quantiles=None, + *, + schedule: BenchSchedule = "isolated", + statistic: BenchStatistic = "median", + prep_fn=None, + flush_bytes=0, + iqr=False, + stream=None, + return_result=False, + unit="ms", +): + """Benchmark a GPU callable using CUDA/HIP events. + + The legacy call shape remains unchanged: isolated iterations, a median + scalar (or ranked quantiles), and milliseconds. New callers should request + ``return_result=True``; :class:`BenchResult` always stores microseconds and + records the measurement contract. + + ``pipelined`` uses one event pair around all calls and therefore returns a + single average sample. ``per_iter`` records one pair per call and performs + one final synchronization. ``isolated`` preserves the autotuner's existing + synchronize-after-each-call behavior. + """ + if torch is None: + raise RuntimeError("do_bench requires torch with CUDA/HIP support") + if warmup < 0 or rep <= 0: + raise ValueError("warmup must be >= 0 and rep must be > 0") + if schedule not in ("isolated", "pipelined", "per_iter"): + raise ValueError(f"unsupported schedule: {schedule!r}") + if statistic not in ("median", "median_average", "mean", "min", "max"): + raise ValueError(f"unsupported statistic: {statistic!r}") + if unit not in ("ms", "us"): + raise ValueError("unit must be 'ms' or 'us'") + if quantiles is not None and return_result: + raise ValueError("quantiles and return_result cannot be combined") + flush_bytes = int(flush_bytes or 0) + if flush_bytes < 0: + raise ValueError("flush_bytes must be >= 0") + if schedule == "pipelined" and statistic != "mean": + raise ValueError("pipelined schedule only supports statistic='mean'") + if schedule == "pipelined" and (prep_fn is not None or flush_bytes): + raise ValueError("pipelined schedule cannot include prep_fn or cache flushing") + + flush_device = getattr(stream, "device", "cuda") + flush_buf = ( + torch.empty(flush_bytes, dtype=torch.uint8, device=flush_device) + if flush_bytes + else None + ) + stream_context = torch.cuda.stream(stream) if stream is not None else nullcontext() + + def prepare(): + if flush_buf is not None: + flush_buf.zero_() + if prep_fn is not None: + prep_fn() + + def record(event): + if stream is None: + event.record() + else: + event.record(stream) + + with stream_context: + for _ in range(warmup): + prepare() + fn() torch.cuda.synchronize() - times.append(start.elapsed_time(end)) - times.sort() + + samples_us = [] + if schedule == "pipelined": + start = torch.cuda.Event(enable_timing=True) + end = torch.cuda.Event(enable_timing=True) + record(start) + for _ in range(rep): + prepare() + fn() + record(end) + end.synchronize() + samples_us.append(start.elapsed_time(end) * 1e3 / rep) + elif schedule == "per_iter": + starts = [torch.cuda.Event(enable_timing=True) for _ in range(rep)] + ends = [torch.cuda.Event(enable_timing=True) for _ in range(rep)] + for index in range(rep): + prepare() + record(starts[index]) + fn() + record(ends[index]) + ends[-1].synchronize() + samples_us.extend(starts[index].elapsed_time(ends[index]) * 1e3 for index in range(rep)) + else: + for _ in range(rep): + prepare() + start = torch.cuda.Event(enable_timing=True) + end = torch.cuda.Event(enable_timing=True) + record(start) + fn() + record(end) + if stream is None: + # Preserve the legacy autotuner contract: fn may enqueue + # work on streams other than the current one. + torch.cuda.synchronize() + else: + end.synchronize() + samples_us.append(start.elapsed_time(end) * 1e3) + + reduced_samples = _filter_iqr(samples_us) if iqr else sorted(samples_us) + value_us = _reduce_samples(reduced_samples, statistic) + result = BenchResult( + value_us=value_us, + samples_us=tuple(samples_us), + instrument="cuda_event", + schedule=schedule, + statistic=statistic, + warmup=warmup, + iterations=rep, + cache_policy=f"flush:{flush_bytes}" if flush_bytes else "warm", + ) + if return_result: + return result + + scale = 1.0 if unit == "us" else 1e-3 if quantiles: - return [times[min(int(q * len(times)), len(times) - 1)] for q in quantiles] - return times[len(times) // 2] + ranked = sorted(samples_us) + return [ranked[min(int(q * len(ranked)), len(ranked) - 1)] * scale for q in quantiles] + return value_us * scale class Autotuner: diff --git a/scripts/benchmark_compare.py b/scripts/benchmark_compare.py new file mode 100644 index 000000000..5dfad311e --- /dev/null +++ b/scripts/benchmark_compare.py @@ -0,0 +1,75 @@ +#!/usr/bin/env python3 +# SPDX-License-Identifier: Apache-2.0 +# Copyright (c) 2026 FlyDSL Project Contributors + +"""Shared, dependency-free benchmark regression policy helpers.""" + +from __future__ import annotations + +import json +from dataclasses import dataclass +from fnmatch import fnmatch +from pathlib import Path + + +@dataclass(frozen=True) +class Threshold: + relative_pct: float + absolute: float + direction: str = "lower_better" + + +@dataclass(frozen=True) +class Regression: + delta: float + delta_pct: float + regressed: bool + + +def compare_values(baseline: float, current: float, threshold: Threshold) -> Regression: + """Compare values using the allreduce relative-AND-absolute rule.""" + if baseline <= 0: + raise ValueError("baseline must be positive") + if threshold.direction == "lower_better": + delta = current - baseline + elif threshold.direction == "higher_better": + delta = baseline - current + else: + raise ValueError(f"unsupported metric direction: {threshold.direction!r}") + delta_pct = delta / baseline * 100.0 + return Regression( + delta=delta, + delta_pct=delta_pct, + regressed=delta_pct > threshold.relative_pct and delta > threshold.absolute, + ) + + +class ThresholdConfig: + """Versioned per-architecture hard-gate allowlist.""" + + def __init__(self, raw: dict): + if raw.get("version") != 1: + raise ValueError("benchmark threshold config must have version 1") + self._architectures = raw.get("architectures", {}) + + @classmethod + def from_path(cls, path: Path) -> "ThresholdConfig": + return cls(json.loads(path.read_text())) + + def match(self, *, arch: str, op: str, shape: str, dtype: str) -> Threshold | None: + for entry in self._architectures.get(arch, []): + if not fnmatch(op, entry.get("op", "*")): + continue + if not fnmatch(shape, entry.get("shape", "*")): + continue + if not fnmatch(dtype, entry.get("dtype", "*")): + continue + return Threshold( + relative_pct=float(entry["relative_pct"]), + absolute=float(entry["absolute_us"]), + direction="lower_better", + ) + return None + + def supports_arch(self, arch: str) -> bool: + return arch in self._architectures diff --git a/scripts/benchmark_log_parser.py b/scripts/benchmark_log_parser.py new file mode 100644 index 000000000..2549e04d9 --- /dev/null +++ b/scripts/benchmark_log_parser.py @@ -0,0 +1,190 @@ +#!/usr/bin/env python3 +# SPDX-License-Identifier: Apache-2.0 +# Copyright (c) 2026 FlyDSL Project Contributors + +"""Extract normalized benchmark metrics from FlyDSL benchmark logs.""" + +from __future__ import annotations + +import argparse +import re +from dataclasses import dataclass +from pathlib import Path + + +@dataclass(frozen=True) +class ParsedMetrics: + tbps: float | None = None + tflops: float | None = None + avg_us: float | None = None + statistic: str = "reported" + warmup: str = "-" + iters: str = "-" + instrument: str = "reported" + schedule: str = "unknown" + cache_policy: str = "unknown" + + +def _last_match(pattern: str, text: str): + matches = list(re.finditer(pattern, text)) + return matches[-1] if matches else None + + +def parse_metrics(text: str) -> ParsedMetrics: + tbps = tflops = avg_us = None + statistic = "reported" + + match = _last_match( + r"Throughput:\s*([0-9.]+)\s*us.*?([0-9.]+)\s*TFLOPS.*?BW:\s*([0-9.]+)\s*TB/s", + text, + ) + if match: + avg_us, tflops, tbps = map(float, match.groups()) + statistic = "mean" + + if tbps is None or tflops is None: + match = _last_match( + r"FlyDSL MoE .*?:\s*([0-9.]+)\s*us,\s*([0-9.]+)\s*TFLOPS.*?([0-9.]+)\s*TB/s", + text, + ) + if match: + avg_us, tflops, tbps = map(float, match.groups()) + + if tflops is None: + match = _last_match( + r"\|\s+(?:PASS|FAIL|--)\s+\|\s+[0-9.eE+-]+\s+[0-9.]+\s+\|\s+([0-9.]+)\s+([0-9.]+)", + text, + ) + if match: + avg_us, tflops = map(float, match.groups()) + + if tbps is None or tflops is None: + match = _last_match(r"TFLOPS=([0-9.]+)\s+TB/s=([0-9.]+)", text) + if match: + tflops, tbps = map(float, match.groups()) + timing = _last_match(r"us_p50=([0-9.]+)", text) + if timing: + avg_us = float(timing.group(1)) + statistic = "median" + + if tbps is None: + bandwidth = next(re.finditer(r"Bandwidth:\s*([0-9.]+)\s*GB/s", text), None) + if bandwidth: + tbps = float(bandwidth.group(1)) / 1000.0 + timing = next(re.finditer(r"Kernel avg time:\s*([0-9.]+)\s*ms", text), None) + if timing: + avg_us = float(timing.group(1)) * 1000.0 + statistic = "mean" + + contract = _last_match( + r"Benchmark contract:\s+instrument=(\S+)\s+schedule=(\S+)\s+" + r"cache=(\S+)\s+statistic=(\S+)\s+warmup=(\d+)\s+iters=(\d+)", + text, + ) + if contract: + instrument, schedule, cache_policy, statistic, warmup, iters = contract.groups() + else: + warmup, iters, instrument, schedule, cache_policy = ("-", "-", "reported", "unknown", "unknown") + + return ParsedMetrics( + tbps=tbps, + tflops=tflops, + avg_us=avg_us, + statistic=statistic, + warmup=warmup, + iters=iters, + instrument=instrument, + schedule=schedule, + cache_policy=cache_policy, + ) + + +def parse_moe_stage2(text: str): + pattern = re.compile( + r"FlyDSL MoE stage2 \[[^]]+\]\s+(\S+)\s+(atomic|reduce)\b.*?" + r"([0-9.]+)\s*us,\s*([0-9.]+)\s*TFLOPS.*?([0-9.]+)\s*TB/s" + ) + found = {} + for match in pattern.finditer(text): + dtype, mode = match.group(1), match.group(2) + found[mode] = ( + dtype, + ParsedMetrics( + avg_us=float(match.group(3)), + tflops=float(match.group(4)), + tbps=float(match.group(5)), + statistic="reported", + ), + ) + return found + + +def _fmt(value): + return "-" if value is None else f"{value:.3f}" + + +def measurement_contract(op: str): + if op in {"softmax", "layernorm", "rmsnorm", "rmsnorm_mixed_weight"}: + return ("10", "100", "torch_profiler", "per_iter_sync", "warm") + if op in {"rmsnorm_mixed_w_bwd", "rmsnorm_add_mixed_bwd"}: + return ("10", "100", "device_event", "pipelined", "warm") + return ("-", "-", "reported", "unknown", "unknown") + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + subparsers = parser.add_subparsers(dest="command", required=True) + + row = subparsers.add_parser("row") + row.add_argument("op") + row.add_argument("shape") + row.add_argument("dtype") + row.add_argument("log", type=Path) + + moe = subparsers.add_parser("moe-stage2") + moe.add_argument("op_prefix") + moe.add_argument("shape") + moe.add_argument("log", type=Path) + + args = parser.parse_args() + text = args.log.read_text(errors="ignore") if args.log.exists() else "" + + if args.command == "row": + metrics = parse_metrics(text) + if metrics.instrument == "reported": + warmup, iters, instrument, schedule, cache_policy = measurement_contract(args.op) + else: + warmup = metrics.warmup + iters = metrics.iters + instrument = metrics.instrument + schedule = metrics.schedule + cache_policy = metrics.cache_policy + print( + f"{args.op}\t{args.shape}\t{args.dtype}\t{_fmt(metrics.tbps)}\t" + f"{_fmt(metrics.tflops)}\t{_fmt(metrics.avg_us)}\t{metrics.statistic}\t" + f"{warmup}\t{iters}\t{instrument}\t{schedule}\t{cache_policy}" + ) + return 0 + + found = parse_moe_stage2(text) + emitted = False + for mode in ("atomic", "reduce"): + if mode not in found: + continue + dtype, metrics = found[mode] + print( + f"{args.op_prefix}_{mode}\t{args.shape}\t{dtype}\t{_fmt(metrics.tbps)}\t" + f"{_fmt(metrics.tflops)}\t{_fmt(metrics.avg_us)}\t{metrics.statistic}\t" + "-\t-\treported\tunknown\tunknown" + ) + emitted = True + if not emitted: + print( + f"{args.op_prefix}_atomic\t{args.shape}\t-\t-\t-\t-\treported\t" + "-\t-\treported\tunknown\tunknown" + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/benchmark_output_to_csv.py b/scripts/benchmark_output_to_csv.py index 7713f10be..d63769ad0 100644 --- a/scripts/benchmark_output_to_csv.py +++ b/scripts/benchmark_output_to_csv.py @@ -44,12 +44,52 @@ def main() -> int: continue if not (_is_metric(tbps) and _is_metric(tflops)): continue - rows.append([op, shape, dtype, tbps, tflops, _status(tbps, tflops)]) + rows.append( + [ + op, + shape, + dtype, + tbps, + tflops, + _status(tbps, tflops), + "-", + "reported", + "-", + "-", + "-", + "-", + "-", + "reported", + "unknown", + "unknown", + "unknown", + ] + ) args.output.parent.mkdir(parents=True, exist_ok=True) with args.output.open("w", newline="") as f: writer = csv.writer(f) - writer.writerow(["op", "shape", "dtype", "tbps", "tflops", "status"]) + writer.writerow( + [ + "op", + "shape", + "dtype", + "tbps", + "tflops", + "status", + "avg_us", + "statistic", + "min_us", + "max_us", + "sample_count", + "warmup", + "iters", + "instrument", + "schedule", + "cache_policy", + "arch", + ] + ) writer.writerows(rows) print(f"Wrote {len(rows)} benchmark row(s) to {args.output}") diff --git a/scripts/compare_benchmark.py b/scripts/compare_benchmark.py index 64ffaa945..5fedeb968 100644 --- a/scripts/compare_benchmark.py +++ b/scripts/compare_benchmark.py @@ -1,7 +1,7 @@ #!/usr/bin/env python3 # SPDX-License-Identifier: Apache-2.0 # Copyright (c) 2025 FlyDSL Project Contributors -"""Report performance ratios between two run_benchmark.sh CSV outputs.""" +"""Report benchmark deltas and enforce calibrated raw-us regression gates.""" from __future__ import annotations @@ -11,6 +11,8 @@ from dataclasses import dataclass from pathlib import Path +from benchmark_compare import ThresholdConfig, compare_values + @dataclass(frozen=True) class BenchmarkRow: @@ -19,6 +21,14 @@ class BenchmarkRow: dtype: str metric_name: str metric_value: float | None + avg_us: float | None + statistic: str + instrument: str + schedule: str + cache_policy: str + warmup: str + iters: str + arch: str status: str @@ -58,6 +68,14 @@ def _read_csv(path: Path) -> dict[tuple[str, str, str], BenchmarkRow]: dtype=dtype, metric_name=metric_name, metric_value=metric_value, + avg_us=_parse_float(raw.get("avg_us", "")), + statistic=raw.get("statistic", "") or "", + instrument=raw.get("instrument", "") or "", + schedule=raw.get("schedule", "") or "", + cache_policy=raw.get("cache_policy", "") or "", + warmup=raw.get("warmup", "") or "", + iters=raw.get("iters", "") or "", + arch=raw.get("arch", "") or "", status=raw["status"], ) return rows @@ -74,43 +92,114 @@ def main() -> int: parser.add_argument("current_csv", type=Path) parser.add_argument("--baseline-label", default="baseline") parser.add_argument("--current-label", default="current") + parser.add_argument("--arch", default="") + parser.add_argument("--threshold-config", type=Path) + parser.add_argument("--fail-on-regression", action="store_true") args = parser.parse_args() baseline = _read_csv(args.baseline_csv) current = _read_csv(args.current_csv) + thresholds = ThresholdConfig.from_path(args.threshold_config) if args.threshold_config else None + baseline_arches = {row.arch for row in baseline.values() if row.arch} + current_arches = {row.arch for row in current.values() if row.arch} + current_arch = args.arch or (next(iter(current_arches)) if len(current_arches) == 1 else "") + if thresholds and args.fail_on_regression and args.arch and baseline_arches != {args.arch}: + print( + f"Cannot enforce benchmark gate: baseline CSV architectures {sorted(baseline_arches)!r} " + f"do not match trusted runner architecture {args.arch!r}.", + file=sys.stderr, + ) + return 2 + if thresholds and args.fail_on_regression and args.arch and current_arches != {args.arch}: + print( + f"Cannot enforce benchmark gate: CSV architectures {sorted(current_arches)!r} " + f"do not match trusted runner architecture {args.arch!r}.", + file=sys.stderr, + ) + return 2 + if thresholds and args.fail_on_regression and not thresholds.supports_arch(current_arch): + print( + f"Cannot enforce benchmark gate: architecture {current_arch or ''!r} " + "is not declared in the threshold config.", + file=sys.stderr, + ) + return 2 print(f"=== Benchmark: {args.current_label} vs {args.baseline_label} ===") compared = 0 + failures = 0 for key in sorted(current.keys() & baseline.keys()): base = baseline[key] curr = current[key] + arch = current_arch or curr.arch or base.arch + arch_mismatch = bool(base.arch and curr.arch and base.arch != curr.arch) + threshold = None + gate_line = None + if arch_mismatch: + gate_line = f"[GATE SKIP: arch mismatch {base.arch} vs {curr.arch}]" + elif thresholds: + threshold = thresholds.match(arch=arch, op=curr.op, shape=curr.shape, dtype=curr.dtype) + + if threshold is not None: + if base.avg_us is None: + gate_line = "[GATE SKIP: baseline has no raw us]" + elif curr.avg_us is None: + failures += 1 + gate_line = "[BROKEN: current raw us unavailable]" + elif ( + base.statistic, + base.instrument, + base.schedule, + base.cache_policy, + base.warmup, + base.iters, + ) != ( + curr.statistic, + curr.instrument, + curr.schedule, + curr.cache_policy, + curr.warmup, + curr.iters, + ): + failures += 1 + gate_line = "[BROKEN: measurement contract mismatch]" + else: + regression = compare_values(base.avg_us, curr.avg_us, threshold) + tag = "REGRESSION" if regression.regressed else "OK" + if regression.regressed: + failures += 1 + gate_line = ( + f"latency {base.avg_us:.2f} -> {curr.avg_us:.2f} us " + f"delta={regression.delta:+.2f} us ({regression.delta_pct:+.1f}%) [{tag}]" + ) + if base.metric_value is None: + if gate_line: + print(f" {_format_key(key)} {gate_line}") continue - if curr.metric_value is None: - print(f" {_format_key(key)} {args.current_label}=missing [SKIP]") - continue - if curr.metric_name != base.metric_name: + print(f" {_format_key(key)} {args.current_label}=missing throughput [SKIP]") + elif curr.metric_name != base.metric_name: print( f" {_format_key(key)} metric mismatch: " f"{args.baseline_label}={base.metric_name}, " f"{args.current_label}={curr.metric_name} [SKIP]" ) - continue - - compared += 1 - delta = curr.metric_value - base.metric_value - delta_pct = (delta / base.metric_value) * 100.0 if base.metric_value else 0.0 - ratio = curr.metric_value / base.metric_value if base.metric_value else 0.0 - - print( - f" {_format_key(key)} " - f"{args.baseline_label}={base.metric_value:9.3f} {base.metric_name:<6s} " - f"{args.current_label}={curr.metric_value:9.3f} {curr.metric_name:<6s} " - f"ratio={ratio:6.3f}x delta={delta:+9.3f} ({delta_pct:+6.1f}%)" - ) + else: + compared += 1 + delta = curr.metric_value - base.metric_value + delta_pct = (delta / base.metric_value) * 100.0 if base.metric_value else 0.0 + ratio = curr.metric_value / base.metric_value if base.metric_value else 0.0 + print( + f" {_format_key(key)} " + f"{args.baseline_label}={base.metric_value:9.3f} {base.metric_name:<6s} " + f"{args.current_label}={curr.metric_value:9.3f} {curr.metric_name:<6s} " + f"ratio={ratio:6.3f}x delta={delta:+9.3f} ({delta_pct:+6.1f}%)" + ) + if gate_line: + print(" " * 57 + gate_line) skipped_new = len(set(current) - set(baseline)) if skipped_new: @@ -119,16 +208,38 @@ def main() -> int: skipped_missing = 0 for key in sorted(set(baseline) - set(current)): base = baseline[key] - if base.metric_value is None: + if current_arch and base.arch and current_arch != base.arch: + skipped_missing += 1 + print( + f" {_format_key(key)} {args.current_label}=missing row " + f"[GATE SKIP: arch mismatch {base.arch} vs {current_arch}]" + ) continue - skipped_missing += 1 - print(f" {_format_key(key)} {args.current_label}=missing row [SKIP]") + threshold = ( + thresholds.match( + arch=current_arch or base.arch, + op=base.op, + shape=base.shape, + dtype=base.dtype, + ) + if thresholds + else None + ) + if threshold is not None and base.avg_us is not None: + failures += 1 + print(f" {_format_key(key)} {args.current_label}=missing row [BROKEN]") + else: + skipped_missing += 1 + print(f" {_format_key(key)} {args.current_label}=missing row [SKIP]") if skipped_missing: print(f"\nSkipped {skipped_missing} baseline-only benchmark row(s).") if compared == 0: print("No comparable benchmark rows found.") + if failures: + print(f"\nBenchmark comparison found {failures} gated regression(s).") + return 1 if args.fail_on_regression else 0 print("\nBenchmark comparison report completed.") return 0 diff --git a/scripts/run_benchmark.sh b/scripts/run_benchmark.sh index 5b967a9e1..4f95bfc39 100755 --- a/scripts/run_benchmark.sh +++ b/scripts/run_benchmark.sh @@ -272,15 +272,16 @@ print_bound_info() { # Print one-line perf row (like run_tests.sh style). _fmt_table_header() { - # Use fixed widths and truncate long strings to keep columns aligned. - # op column is wide enough to host "moe__s2_atomic" / "_reduce" suffixes. - printf "\n%-22.22s %-34.34s %-10.10s %10s %10s\n" "op" "shape" "dtype" "TB/s" "TFLOPS" - printf "%-22.22s %-34.34s %-10.10s %10s %10s\n" "----------------------" "----------------------------------" "----------" "----------" "----------" + # Widths are minimums, not truncation limits: full keys must survive the + # human-readable table because tag baselines and the dashboard parse it. + printf "\n%-22s %-34s %-10s %10s %10s\n" "op" "shape" "dtype" "TB/s" "TFLOPS" + printf "%-22s %-34s %-10s %10s %10s\n" "----------------------" "----------------------------------" "----------" "----------" "----------" } _emit_row() { - op="$1"; shape="$2"; dtype="$3"; tbps="$4"; tflops="$5" - printf "%-22.22s %-34.34s %-10.10s %10s %10s\n" "${op}" "${shape}" "${dtype}" "${tbps}" "${tflops}" + op="$1"; shape="$2"; dtype="$3"; tbps="$4"; tflops="$5"; avg_us="${6:--}" + statistic="${7:-reported}" + printf "%-22s %-34s %-10s %10s %10s\n" "${op}" "${shape}" "${dtype}" "${tbps}" "${tflops}" if [ -n "${BENCH_OUTPUT_CSV:-}" ]; then status="ok" if [ "${tbps}" = "skip" ] || [ "${tflops}" = "skip" ]; then @@ -288,7 +289,12 @@ _emit_row() { elif [ "${tbps}" = "-" ] && [ "${tflops}" = "-" ]; then status="missing" fi - python3 - "${BENCH_OUTPUT_CSV}" "${op}" "${shape}" "${dtype}" "${tbps}" "${tflops}" "${status}" <<'PY' + min_us="${8:--}"; max_us="${9:--}"; sample_count="${10:--}" + warmup="${11:--}"; iters="${12:--}"; instrument="${13:-reported}" + schedule="${14:-unknown}"; cache_policy="${15:-unknown}" + python3 - "${BENCH_OUTPUT_CSV}" "${op}" "${shape}" "${dtype}" "${tbps}" "${tflops}" "${status}" \ + "${avg_us}" "${statistic}" "${min_us}" "${max_us}" "${sample_count}" "${warmup}" "${iters}" \ + "${instrument}" "${schedule}" "${cache_policy}" "${GPU_ARCH}" <<'PY' import csv import sys @@ -428,81 +434,12 @@ fi if [ -n "${BENCH_OUTPUT_CSV}" ]; then mkdir -p "$(dirname "${BENCH_OUTPUT_CSV}")" - printf "op,shape,dtype,tbps,tflops,status\n" >"${BENCH_OUTPUT_CSV}" + printf "op,shape,dtype,tbps,tflops,status,avg_us,statistic,min_us,max_us,sample_count,warmup,iters,instrument,schedule,cache_policy,arch\n" >"${BENCH_OUTPUT_CSV}" fi _py_parse_and_emit() { - # Args: op shape dtype log_path [M N] - python3 - "$@" <<'PY' -import re, sys - -op = sys.argv[1] -shape = sys.argv[2] -dtype = sys.argv[3] -path = sys.argv[4] -MN = sys.argv[5:] # deprecated (kept for backward-compat) - -tbps = None -tflops = None - -txt = "" -try: - with open(path, "r", errors="ignore") as f: - txt = f.read() -except Exception: - txt = "" - -# GEMM-style: "Throughput: ..., XX.XX TFLOPS, BW: Y.YYY TB/s" -m = None -for m in re.finditer(r"Throughput:.*?([0-9.]+)\s*TFLOPS.*?BW:\s*([0-9.]+)\s*TB/s", txt): - pass -if m: - tflops = float(m.group(1)) - tbps = float(m.group(2)) - -# MoE-style: "FlyDSL MoE stageX[dt]: ... XX.XX TFLOPS ... Y.YYY TB/s" -if tbps is None or tflops is None: - m = None - for m in re.finditer(r"FlyDSL MoE .*?\:\s*[0-9.]+\s*us,\s*([0-9.]+)\s*TFLOPS.*?([0-9.]+)\s*TB/s", txt): - pass - if m: - tflops = float(m.group(1)) - tbps = float(m.group(2)) - -# FlashAttention table: "| PASS | maxerr mincos | time_us tflops". -if tflops is None: - m = None - for m in re.finditer(r"\|\s+(?:PASS|FAIL|--)\s+\|\s+[0-9.eE+-]+\s+[0-9.]+\s+\|\s+([0-9.]+)\s+([0-9.]+)", txt): - pass - if m: - tflops = float(m.group(2)) - -# MLA decode: "TFLOPS=... TB/s=..." -if tbps is None or tflops is None: - m = None - for m in re.finditer(r"TFLOPS=([0-9.]+)\s+TB/s=([0-9.]+)", txt): - pass - if m: - tflops = float(m.group(1)) - tbps = float(m.group(2)) - -# Softmax/Norm-style: "Kernel avg time: X ms" + "Bandwidth: Y GB/s". -# Use the FIRST match: the base op (softmax/layernorm/rmsnorm) is benchmarked -# first, so any later "Bandwidth:" lines come from fused/quant variants printed -# by the same test (e.g. test_layernorm.py also runs fused_add/dynamicquant/ -# smoothquant). Taking the last match reported the slow scalar smoothquant path -# as "layernorm" (~1.69 vs the real ~5.6 TB/s base). -if tbps is None: - m_bw = next(re.finditer(r"Bandwidth:\s*([0-9.]+)\s*GB/s", txt), None) - if m_bw: - tbps = float(m_bw.group(1)) / 1000.0 - - -def fmt(x): - return "-" if x is None else f"{x:.3f}" - -print(f"{op}\t{shape}\t{dtype}\t{fmt(tbps)}\t{fmt(tflops)}") -PY + # Args: op shape dtype log_path + python3 "${SCRIPT_DIR}/benchmark_log_parser.py" row "$1" "$2" "$3" "$4" } _emit_moe_s2_rows() { @@ -511,43 +448,7 @@ _emit_moe_s2_rows() { # FlyDSL MoE stage2 [moe_gemm2] fp4 atomic | 7168x2048, ... | 1163.2 us, 1654.24 TFLOPS, 0.377 TB/s # Emit two table rows (op_prefix_atomic, op_prefix_reduce). Falls back to single row # tagged "mixed" if the log only has one mode (e.g., --gemm2_mode was overridden). - op_prefix="$1"; shape="$2"; log_path="$3" - python3 - "$op_prefix" "$shape" "$log_path" <<'PY' -import re, sys - -op_prefix, shape, path = sys.argv[1], sys.argv[2], sys.argv[3] -try: - with open(path, "r", errors="ignore") as f: - txt = f.read() -except Exception: - txt = "" - -pat = re.compile( - r"FlyDSL MoE stage2 \[[^]]+\]\s+(\S+)\s+(atomic|reduce)\b.*?" - r"([0-9.]+)\s*TFLOPS.*?([0-9.]+)\s*TB/s" -) -# keep last occurrence per mode -found = {} -for m in pat.finditer(txt): - dtype, mode = m.group(1), m.group(2) - found[mode] = (dtype, float(m.group(3)), float(m.group(4))) - -def fmt(x): - return "-" if x is None else f"{x:.3f}" - -# Always emit atomic row first (if any), then reduce row. -emitted = False -for mode in ("atomic", "reduce"): - if mode not in found: - continue - dtype, tflops, tbps = found[mode] - print(f"{op_prefix}_{mode}\t{shape}\t{dtype}\t{fmt(tbps)}\t{fmt(tflops)}") - emitted = True - -if not emitted: - # Nothing parsed — emit empty row so caller knows. - print(f"{op_prefix}_atomic\t{shape}\t-\t-\t-") -PY + python3 "${SCRIPT_DIR}/benchmark_log_parser.py" moe-stage2 "$1" "$2" "$3" } # ============================================================================ @@ -578,7 +479,7 @@ if [ "${RUN_SOFTMAX}" -eq 1 ]; then row="$(_py_parse_and_emit softmax "${M}x${N}" "${dtype}" "${log}")" # row is tab-separated; default IFS includes tabs. set -- $row - _emit_row "$1" "$2" "$3" "$4" "$5" + _emit_row "$1" "$2" "$3" "$4" "$5" "$6" "$7" "-" "-" "-" "$8" "$9" "${10}" "${11}" "${12}" done fi @@ -600,7 +501,7 @@ if [ "${RUN_LAYERNORM}" -eq 1 ]; then fi row="$(_py_parse_and_emit layernorm "${M}x${N}" "${dtype}" "${log}")" set -- $row - _emit_row "$1" "$2" "$3" "$4" "$5" + _emit_row "$1" "$2" "$3" "$4" "$5" "$6" "$7" "-" "-" "-" "$8" "$9" "${10}" "${11}" "${12}" done fi @@ -622,7 +523,7 @@ if [ "${RUN_RMSNORM}" -eq 1 ]; then fi row="$(_py_parse_and_emit rmsnorm "${M}x${N}" "${dtype}" "${log}")" set -- $row - _emit_row "$1" "$2" "$3" "$4" "$5" + _emit_row "$1" "$2" "$3" "$4" "$5" "$6" "$7" "-" "-" "-" "$8" "$9" "${10}" "${11}" "${12}" done # Training contract: FP16/BF16 activations with FP32 weights. These focused @@ -647,7 +548,7 @@ if [ "${RUN_RMSNORM}" -eq 1 ]; then fi row="$(_py_parse_and_emit rmsnorm_mixed_weight "${M}x${N}" "${activation_dtype}+f32w" "${log}")" set -- $row - _emit_row "$1" "$2" "$3" "$4" "$5" + _emit_row "$1" "$2" "$3" "$4" "$5" "$6" "$7" "-" "-" "-" "$8" "$9" "${10}" "${11}" "${12}" for mode in plain fused_add; do export ROCDSL_RMSNORM_MIXED_WEIGHT_BWD_MODE="$mode" @@ -665,7 +566,7 @@ if [ "${RUN_RMSNORM}" -eq 1 ]; then fi row="$(_py_parse_and_emit "$op" "${M}x${N}" "${activation_dtype}+f32w" "${log}")" set -- $row - _emit_row "$1" "$2" "$3" "$4" "$5" + _emit_row "$1" "$2" "$3" "$4" "$5" "$6" "$7" "-" "-" "-" "$8" "$9" "${10}" "${11}" "${12}" done done unset ROCDSL_RMSNORM_MIXED_WEIGHT_BENCH_SHAPE ROCDSL_RMSNORM_MIXED_WEIGHT_BWD_MODE @@ -717,7 +618,7 @@ if [ "${RUN_FLASH_ATTN}" -eq 1 ] && [ "${IS_CDNA}" = "true" ]; then shape_tag="B${batch}S${seq_len}H${heads}Hkv${kv_heads}D${head_dim}_${causal_tag}" row="$(_py_parse_and_emit flash_attn "${shape_tag}" "${dtype}" "${log}")" set -- $row - _emit_row "$1" "$2" "$3" "$4" "$5" + _emit_row "$1" "$2" "$3" "$4" "$5" "$6" "$7" "-" "-" "-" "$8" "$9" "${10}" "${11}" "${12}" done fi @@ -742,7 +643,7 @@ if [ "${RUN_MLA}" -eq 1 ] && [ "${IS_CDNA}" = "true" ]; then fi row="$(_py_parse_and_emit mla "${shape_tag}" "fp8" "${log}")" set -- $row - _emit_row "$1" "$2" "$3" "$4" "$5" + _emit_row "$1" "$2" "$3" "$4" "$5" "$6" "$7" "-" "-" "-" "$8" "$9" "${10}" "${11}" "${12}" done fi @@ -773,7 +674,7 @@ if [ "${RUN_PRESHUFFLE_GEMM}" -eq 1 ] && [ "${IS_CDNA}" = "true" ]; then gemm_shape_tag="${M}x${N}x${K}_tile${tile_m}x${tile_n}x${tile_k}" row="$(_py_parse_and_emit gemm "${gemm_shape_tag}" "${dtype}" "${log}")" set -- $row - _emit_row "$1" "$2" "$3" "$4" "$5" + _emit_row "$1" "$2" "$3" "$4" "$5" "$6" "$7" "-" "-" "-" "$8" "$9" "${10}" "${11}" "${12}" done GEMM_USE_ASYNC_COPY="${GEMM_USE_ASYNC_COPY:-1}" @@ -816,7 +717,7 @@ if [ "${RUN_PRESHUFFLE_GEMM}" -eq 1 ] && [ "${IS_CDNA}" = "true" ]; then shape_tag="${M}x${N}x${K}_tile${tile_m}x${tile_n}x${tile_k}_${waves_per_eu}tg" row="$(_py_parse_and_emit gemm_async "${shape_tag}" "${dtype}" "${log}")" set -- $row - _emit_row "$1" "$2" "$3" "$4" "$5" + _emit_row "$1" "$2" "$3" "$4" "$5" "$6" "$7" "-" "-" "-" "$8" "$9" "${10}" "${11}" "${12}" done if [ -n "${HGEMM_SHAPES:-}" ]; then @@ -860,7 +761,7 @@ if [ "${RUN_PRESHUFFLE_GEMM}" -eq 1 ] && [ "${IS_CDNA}" = "true" ]; then shape_tag="${M}x${N}x${K}_tile${tile_m}x${tile_n}x${tile_k}_sk${split_k}" row="$(_py_parse_and_emit hgemm "${shape_tag}" "${dtype}" "${log}")" set -- $row - _emit_row "$1" "$2" "$3" "$4" "$5" + _emit_row "$1" "$2" "$3" "$4" "$5" "$6" "$7" "-" "-" "-" "$8" "$9" "${10}" "${11}" "${12}" fi else _fail_or_skip "${log}" "hgemm" @@ -902,7 +803,7 @@ if [ "${RUN_PRESHUFFLE_GEMM}" -eq 1 ] && [ "${IS_CDNA}" = "true" ]; then shape_tag="${M}x${N}x${K}_tile${tile_m}x${tile_n}_${preshuffle_tag}" row="$(_py_parse_and_emit fp8_8wave_rowscale "${shape_tag}" "${dtype}" "${log}")" set -- $row - _emit_row "$1" "$2" "$3" "$4" "$5" + _emit_row "$1" "$2" "$3" "$4" "$5" "$6" "$7" "-" "-" "-" "$8" "$9" "${10}" "${11}" "${12}" fi else if grep -q "requires CDNA4\|Skipped:" "${log}" 2>/dev/null; then @@ -945,7 +846,7 @@ if [ "${RUN_PRESHUFFLE_GEMM}" -eq 1 ] && [ "${IS_CDNA}" = "true" ]; then SUCCESS_COUNT=$((SUCCESS_COUNT + 1)) row="$(_py_parse_and_emit gemm_fp4 "${gemm_shape_tag}" "${dtype}" "${log}")" set -- $row - _emit_row "$1" "$2" "$3" "$4" "$5" + _emit_row "$1" "$2" "$3" "$4" "$5" "$6" "$7" "-" "-" "-" "$8" "$9" "${10}" "${11}" "${12}" fi else # Skip gracefully on unsupported architectures or missing features @@ -995,14 +896,15 @@ if [ "${RUN_MOE}" -eq 1 ] && [ "${IS_CDNA}" = "true" ]; then shape_moe="t${tokens}-d${model_dim}x${inter_dim}-e${experts}k${topk}" dt_s1="$(grep -Eo 'FlyDSL MoE stage1\[[^]]+\]:' "${log}" | tail -1 | cut -d'[' -f2 | cut -d']' -f1 || true)" + us_s1="$(grep -Eo 'FlyDSL MoE stage1\[[^]]+\]:[[:space:]]*[0-9.]+[[:space:]]*us' "${log}" | tail -1 | awk '{print $(NF-1)}' || true)" tf_s1="$(grep -Eo 'FlyDSL MoE stage1\[[^]]+\]:.* ([0-9.]+) TFLOPS' "${log}" | tail -1 | awk '{print $(NF-1)}' || true)" tb_s1="$(grep -Eo 'FlyDSL MoE stage1\[[^]]+\]:.* ([0-9.]+) TB/s' "${log}" | tail -1 | awk '{print $(NF-1)}' || true)" if [ -n "${dt_s1}" ] && [ -n "${tf_s1}" ] && [ -n "${tb_s1}" ]; then - _emit_row "moe_gemm1" "${shape_moe}" "${dt_s1}" "${tb_s1}" "${tf_s1}" + _emit_row "moe_gemm1" "${shape_moe}" "${dt_s1}" "${tb_s1}" "${tf_s1}" "${us_s1:--}" "reported" fi - _emit_moe_s2_rows "moe_gemm2" "${shape_moe}" "${log}" | while IFS="$(printf '\t')" read -r _op _sh _dt _tb _tf; do - _emit_row "${_op}" "${_sh}" "${_dt}" "${_tb}" "${_tf}" + _emit_moe_s2_rows "moe_gemm2" "${shape_moe}" "${log}" | while IFS="$(printf '\t')" read -r _op _sh _dt _tb _tf _us _stat _warm _iters _instrument _schedule _cache; do + _emit_row "${_op}" "${_sh}" "${_dt}" "${_tb}" "${_tf}" "${_us:--}" "${_stat:-reported}" "-" "-" "-" "${_warm:--}" "${_iters:--}" "${_instrument:-reported}" "${_schedule:-unknown}" "${_cache:-unknown}" done done @@ -1041,14 +943,15 @@ if [ "${RUN_MOE}" -eq 1 ] && [ "${IS_CDNA}" = "true" ]; then shape_moe="t${tokens}-d${model_dim}x${inter_dim}-e${experts}k${topk}" dt_s1="$(grep -Eo 'FlyDSL MoE stage1\[[^]]+\]:' "${log}" | tail -1 | cut -d'[' -f2 | cut -d']' -f1 || true)" + us_s1="$(grep -Eo 'FlyDSL MoE stage1\[[^]]+\]:[[:space:]]*[0-9.]+[[:space:]]*us' "${log}" | tail -1 | awk '{print $(NF-1)}' || true)" tf_s1="$(grep -Eo 'FlyDSL MoE stage1\[[^]]+\]:.* ([0-9.]+) TFLOPS' "${log}" | tail -1 | awk '{print $(NF-1)}' || true)" tb_s1="$(grep -Eo 'FlyDSL MoE stage1\[[^]]+\]:.* ([0-9.]+) TB/s' "${log}" | tail -1 | awk '{print $(NF-1)}' || true)" if [ -n "${dt_s1}" ] && [ -n "${tf_s1}" ] && [ -n "${tb_s1}" ]; then - _emit_row "moe_fp4_s1" "${shape_moe}" "${dt_s1}" "${tb_s1}" "${tf_s1}" + _emit_row "moe_fp4_s1" "${shape_moe}" "${dt_s1}" "${tb_s1}" "${tf_s1}" "${us_s1:--}" "reported" fi - _emit_moe_s2_rows "moe_fp4_s2" "${shape_moe}" "${log}" | while IFS="$(printf '\t')" read -r _op _sh _dt _tb _tf; do - _emit_row "${_op}" "${_sh}" "${_dt}" "${_tb}" "${_tf}" + _emit_moe_s2_rows "moe_fp4_s2" "${shape_moe}" "${log}" | while IFS="$(printf '\t')" read -r _op _sh _dt _tb _tf _us _stat _warm _iters _instrument _schedule _cache; do + _emit_row "${_op}" "${_sh}" "${_dt}" "${_tb}" "${_tf}" "${_us:--}" "${_stat:-reported}" "-" "-" "-" "${_warm:--}" "${_iters:--}" "${_instrument:-reported}" "${_schedule:-unknown}" "${_cache:-unknown}" done fi else @@ -1096,14 +999,15 @@ if [ "${RUN_MOE}" -eq 1 ] && [ "${IS_CDNA}" = "true" ]; then shape_moe="t${tokens}-d${model_dim}x${inter_dim}-e${experts}k${topk}" dt_s1="$(grep -Eo 'FlyDSL MoE stage1\[[^]]+\]:' "${log}" | tail -1 | cut -d'[' -f2 | cut -d']' -f1 || true)" + us_s1="$(grep -Eo 'FlyDSL MoE stage1\[[^]]+\]:[[:space:]]*[0-9.]+[[:space:]]*us' "${log}" | tail -1 | awk '{print $(NF-1)}' || true)" tf_s1="$(grep -Eo 'FlyDSL MoE stage1\[[^]]+\]:.* ([0-9.]+) TFLOPS' "${log}" | tail -1 | awk '{print $(NF-1)}' || true)" tb_s1="$(grep -Eo 'FlyDSL MoE stage1\[[^]]+\]:.* ([0-9.]+) TB/s' "${log}" | tail -1 | awk '{print $(NF-1)}' || true)" if [ -n "${dt_s1}" ] && [ -n "${tf_s1}" ] && [ -n "${tb_s1}" ]; then - _emit_row "moe_w4a16_s1" "${shape_moe}" "${dt_s1}" "${tb_s1}" "${tf_s1}" + _emit_row "moe_w4a16_s1" "${shape_moe}" "${dt_s1}" "${tb_s1}" "${tf_s1}" "${us_s1:--}" "reported" fi - _emit_moe_s2_rows "moe_w4a16_s2" "${shape_moe}" "${log}" | while IFS="$(printf '\t')" read -r _op _sh _dt _tb _tf; do - _emit_row "${_op}" "${_sh}" "${_dt}" "${_tb}" "${_tf}" + _emit_moe_s2_rows "moe_w4a16_s2" "${shape_moe}" "${log}" | while IFS="$(printf '\t')" read -r _op _sh _dt _tb _tf _us _stat _warm _iters _instrument _schedule _cache; do + _emit_row "${_op}" "${_sh}" "${_dt}" "${_tb}" "${_tf}" "${_us:--}" "${_stat:-reported}" "-" "-" "-" "${_warm:--}" "${_iters:--}" "${_instrument:-reported}" "${_schedule:-unknown}" "${_cache:-unknown}" done done @@ -1142,14 +1046,15 @@ if [ "${RUN_MOE}" -eq 1 ] && [ "${IS_CDNA}" = "true" ]; then SUCCESS_COUNT=$((SUCCESS_COUNT + 1)) dt_s1="$(grep -Eo 'FlyDSL MoE stage1\[[^]]+\]:' "${log}" | tail -1 | cut -d'[' -f2 | cut -d']' -f1 || true)" + us_s1="$(grep -Eo 'FlyDSL MoE stage1\[[^]]+\]:[[:space:]]*[0-9.]+[[:space:]]*us' "${log}" | tail -1 | awk '{print $(NF-1)}' || true)" tf_s1="$(grep -Eo 'FlyDSL MoE stage1\[[^]]+\]:.* ([0-9.]+) TFLOPS' "${log}" | tail -1 | awk '{print $(NF-1)}' || true)" tb_s1="$(grep -Eo 'FlyDSL MoE stage1\[[^]]+\]:.* ([0-9.]+) TB/s' "${log}" | tail -1 | awk '{print $(NF-1)}' || true)" if [ -n "${dt_s1}" ] && [ -n "${tf_s1}" ] && [ -n "${tb_s1}" ]; then - _emit_row "moe_a8w4_s1" "${shape_moe}" "${dt_s1}" "${tb_s1}" "${tf_s1}" + _emit_row "moe_a8w4_s1" "${shape_moe}" "${dt_s1}" "${tb_s1}" "${tf_s1}" "${us_s1:--}" "reported" fi - _emit_moe_s2_rows "moe_a8w4_s2" "${shape_moe}" "${log}" | while IFS="$(printf '\t')" read -r _op _sh _dt _tb _tf; do - _emit_row "${_op}" "${_sh}" "${_dt}" "${_tb}" "${_tf}" + _emit_moe_s2_rows "moe_a8w4_s2" "${shape_moe}" "${log}" | while IFS="$(printf '\t')" read -r _op _sh _dt _tb _tf _us _stat _warm _iters _instrument _schedule _cache; do + _emit_row "${_op}" "${_sh}" "${_dt}" "${_tb}" "${_tf}" "${_us:--}" "${_stat:-reported}" "-" "-" "-" "${_warm:--}" "${_iters:--}" "${_instrument:-reported}" "${_schedule:-unknown}" "${_cache:-unknown}" done fi else diff --git a/tests/kernels/benchmark_common.py b/tests/kernels/benchmark_common.py index 8770f3915..423d22c14 100644 --- a/tests/kernels/benchmark_common.py +++ b/tests/kernels/benchmark_common.py @@ -31,6 +31,8 @@ if os.path.isdir(_EMBEDDED_FLYDSL) and _EMBEDDED_FLYDSL not in sys.path: sys.path.insert(0, _EMBEDDED_FLYDSL) +from flydsl.autotune import do_bench # noqa: E402 + @dataclass(frozen=True) class PerfRow: @@ -67,19 +69,14 @@ def print_perf_table(rows: List[PerfRow]) -> None: def bench_gpu_us_torch(fn: Callable[[], None], *, warmup: int = 20, iters: int = 200) -> float: """Measure device time using torch CUDA events (works for torch-launched kernels, incl. Triton).""" - import torch - - start = torch.cuda.Event(enable_timing=True) - end = torch.cuda.Event(enable_timing=True) - for _ in range(warmup): - fn() - torch.cuda.synchronize() - start.record() - for _ in range(iters): - fn() - end.record() - torch.cuda.synchronize() - return start.elapsed_time(end) * 1e3 / iters + return do_bench( + fn, + warmup=warmup, + rep=iters, + schedule="pipelined", + statistic="mean", + unit="us", + ) def maybe_enable_aiter() -> bool: @@ -468,59 +465,25 @@ def bench_kernel_us(run_fn, warmup=10, iters=50, flush_l2=True, prep_fn=None): """Per-iteration CUDA events timer with optional L2 flush and median latency.""" import torch - flush_buf = None + flush_bytes = 0 if flush_l2: l2_bytes = getattr( torch.cuda.get_device_properties(torch.cuda.current_device()), "L2_cache_size", 4 * 1024 * 1024 ) - alloc_bytes = max(l2_bytes * 2, 8 * 1024 * 1024) - flush_buf = torch.empty(alloc_bytes, dtype=torch.uint8, device="cuda") - - for _ in range(warmup): - if flush_buf is not None: - flush_buf.zero_() - if prep_fn is not None: - prep_fn() - run_fn() - torch.cuda.synchronize() - - if flush_buf is None and prep_fn is None: - # Single event pair preserves back-to-back launch pipelining (returns mean latency). - start = torch.cuda.Event(enable_timing=True) - end = torch.cuda.Event(enable_timing=True) - start.record() - for _ in range(iters): - run_fn() - end.record() - torch.cuda.synchronize() - return start.elapsed_time(end) * 1e3 / iters - - start_ev = [torch.cuda.Event(enable_timing=True) for _ in range(iters)] - end_ev = [torch.cuda.Event(enable_timing=True) for _ in range(iters)] - - for i in range(iters): - if flush_buf is not None: - flush_buf.zero_() - if prep_fn is not None: - prep_fn() - start_ev[i].record() - run_fn() - end_ev[i].record() - - torch.cuda.synchronize() - latencies = sorted(start_ev[i].elapsed_time(end_ev[i]) * 1e3 for i in range(iters)) - - n = len(latencies) - if n >= 8: - q1, q3 = latencies[n // 4], latencies[3 * n // 4] - iqr = q3 - q1 - lo, hi = q1 - 1.5 * iqr, q3 + 1.5 * iqr - filtered = [x for x in latencies if lo <= x <= hi] - if filtered: - latencies = filtered - - del flush_buf - return latencies[len(latencies) // 2] + flush_bytes = max(l2_bytes * 2, 8 * 1024 * 1024) + + per_iter = bool(flush_bytes or prep_fn is not None) + return do_bench( + run_fn, + warmup=warmup, + rep=iters, + schedule="per_iter" if per_iter else "pipelined", + statistic="median" if per_iter else "mean", + prep_fn=prep_fn, + flush_bytes=flush_bytes, + iqr=per_iter, + unit="us", + ) def bench_best_tile(target, dim, align): diff --git a/tests/kernels/compare_allreduce_benchmark.py b/tests/kernels/compare_allreduce_benchmark.py index 843a8ebc0..e13fecb43 100644 --- a/tests/kernels/compare_allreduce_benchmark.py +++ b/tests/kernels/compare_allreduce_benchmark.py @@ -2,7 +2,7 @@ """Compare two allreduce benchmark CSVs (main vs PR) and flag regressions. Usage: - python3 compare_benchmark.py + python3 compare_allreduce_benchmark.py Exit code 1 if any case regresses more than BOTH thresholds: - relative increase > MAX_REGRESSION_PCT (default 15%) @@ -10,11 +10,20 @@ """ import sys +from pathlib import Path import pandas as pd +sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "scripts")) +from benchmark_compare import Threshold, compare_values # noqa: E402 + MAX_REGRESSION_PCT = 15.0 MIN_ABS_REGRESSION_US = 10.0 +THRESHOLD = Threshold( + relative_pct=MAX_REGRESSION_PCT, + absolute=MIN_ABS_REGRESSION_US, + direction="lower_better", +) def main(): @@ -55,7 +64,11 @@ def main(): print("=== Allreduce Benchmark: PR vs main ===") for (shape, dtype), row in merged.iterrows(): - regressed = row["delta_pct"] > MAX_REGRESSION_PCT and row["delta_us"] > MIN_ABS_REGRESSION_US + regressed = compare_values( + row["avg_time_us_main"], + row["avg_time_us_pr"], + THRESHOLD, + ).regressed tag = "REGRESSION" if regressed else "OK" if regressed: fail_count += 1 diff --git a/tests/kernels/test_fused_rope_cache.py b/tests/kernels/test_fused_rope_cache.py index f9c279e06..acdf3ed14 100644 --- a/tests/kernels/test_fused_rope_cache.py +++ b/tests/kernels/test_fused_rope_cache.py @@ -54,6 +54,7 @@ import pytest import torch +from flydsl.autotune import do_bench from flydsl.runtime.device import get_rocm_arch as _get_rocm_arch from kernels.attention.fused_rope_cache_kernel import build_fused_rope_cache_module @@ -82,17 +83,14 @@ def _bench_gpu_us(fn, warmup: int = 20, iters: int = 200) -> float: """Measure GPU kernel time via CUDA events (true device time, no Python-loop overhead).""" - start = torch.cuda.Event(enable_timing=True) - end = torch.cuda.Event(enable_timing=True) - for _ in range(warmup): - fn() - torch.cuda.synchronize() - start.record() - for _ in range(iters): - fn() - end.record() - torch.cuda.synchronize() - return start.elapsed_time(end) * 1e3 / iters # ms → µs + return do_bench( + fn, + warmup=warmup, + rep=iters, + schedule="pipelined", + statistic="mean", + unit="us", + ) # --------------------------------------------------------------------------- diff --git a/tests/kernels/test_layernorm.py b/tests/kernels/test_layernorm.py index fe7022cd9..2fb9a0e47 100644 --- a/tests/kernels/test_layernorm.py +++ b/tests/kernels/test_layernorm.py @@ -180,8 +180,17 @@ def kernel_launch(): elem_bytes = 4 if dtype == "f32" else 2 total_bytes = (2 * M * N + 2 * N) * elem_bytes # read input + write output + (gamma+beta) bandwidth_gbs = total_bytes / (avg_us / 1e6) / 1e9 + benchmark_instrument = ( + "device_event" + if int(os.environ.get("FLYDSL_PERFTEST_USE_EVENTS", 0)) + else "torch_profiler" + ) print(f"Kernel avg time: {avg_ms:.4f} ms via run_perftest (warmup={WARMUP_ITERS}, iters={BENCH_ITERS})") + print( + f"Benchmark contract: instrument={benchmark_instrument} schedule=per_iter_sync " + f"cache=warm statistic=mean warmup={WARMUP_ITERS} iters={BENCH_ITERS}" + ) print(f"Bandwidth: {bandwidth_gbs:.2f} GB/s") if flydsl_gpu_us is not None: print(f"[Perf] FlyDSL layernorm gpu: {flydsl_gpu_us:.1f} us") diff --git a/tests/kernels/test_moe_a8w4_mxscale_gfx1250.py b/tests/kernels/test_moe_a8w4_mxscale_gfx1250.py index 2e80bbc16..f03c0ce3f 100644 --- a/tests/kernels/test_moe_a8w4_mxscale_gfx1250.py +++ b/tests/kernels/test_moe_a8w4_mxscale_gfx1250.py @@ -8,7 +8,6 @@ from __future__ import annotations import os -import statistics import sys _REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "../..")) @@ -22,6 +21,7 @@ import flydsl.compiler as flyc # noqa: E402 import flydsl.expr as fx # noqa: E402 +from flydsl.autotune import do_bench # noqa: E402 from flydsl.runtime.device import get_rocm_arch # noqa: E402 from kernels.moe.moe_a8w4_mxscale_gfx1250 import launch_moe_gemm_a8w4 # noqa: E402 from tests.kernels.utils import gemm_common_utils as gcu # noqa: E402 @@ -400,17 +400,13 @@ def kernels_only(): stage1_act=0, ) - for _ in range(10): - kernels_only() - torch.cuda.synchronize() - iters = 50 - starts = [torch.cuda.Event(enable_timing=True) for _ in range(iters)] - ends = [torch.cuda.Event(enable_timing=True) for _ in range(iters)] - for s, e in zip(starts, ends): - s.record() - kernels_only() - e.record() - torch.cuda.synchronize() - us = statistics.median(sorted(s.elapsed_time(e) * 1e3 for s, e in zip(starts, ends))) + us = do_bench( + kernels_only, + warmup=10, + rep=50, + schedule="per_iter", + statistic="median_average", + unit="us", + ) print(f"\ngrouped MoE kernels (quant+gemm1+gemm2) E16 m768 i512 t256 topk4: {us:.2f} us") assert us > 0 diff --git a/tests/kernels/test_moe_sorting.py b/tests/kernels/test_moe_sorting.py index b3168942c..0249e38b2 100644 --- a/tests/kernels/test_moe_sorting.py +++ b/tests/kernels/test_moe_sorting.py @@ -28,6 +28,7 @@ if torch is None or not torch.cuda.is_available(): pytest.skip("CUDA/ROCm not available.", allow_module_level=True) +from flydsl.autotune import do_bench # noqa: E402 from flydsl.runtime.device import is_rdna_arch # noqa: E402 if is_rdna_arch(): @@ -385,18 +386,21 @@ def run_test(T, E, topk, unit_size=UNIT_SIZE, max_tokens=None): # --- Benchmark (opt-in via MOE_SORTING_BENCH=1) --- gpu_time_us = None if passed and RUN_BENCH: - for _ in range(WARMUP_ITERS): - _call_flydsl(topk_ids, topk_weights, E, model_dim=4096, topk=topk, unit_size=unit_size) - torch.cuda.synchronize() - - start = torch.cuda.Event(enable_timing=True) - end = torch.cuda.Event(enable_timing=True) - start.record() - for _ in range(BENCH_ITERS): - _call_flydsl(topk_ids, topk_weights, E, model_dim=4096, topk=topk, unit_size=unit_size) - end.record() - torch.cuda.synchronize() - gpu_time_us = start.elapsed_time(end) * 1000.0 / BENCH_ITERS # ms → us + gpu_time_us = do_bench( + lambda: _call_flydsl( + topk_ids, + topk_weights, + E, + model_dim=4096, + topk=topk, + unit_size=unit_size, + ), + warmup=WARMUP_ITERS, + rep=BENCH_ITERS, + schedule="pipelined", + statistic="mean", + unit="us", + ) print(f" [perf] {gpu_time_us:.2f} us/call ({path})") status = "PASSED" if passed else "FAILED" @@ -883,40 +887,25 @@ def test_moe_softmax_sort_fallback(T, E, topk, dtype_str): # --------------------------------------------------------------------------- def bench_eager_us(fn, warmup=BENCH_WARMUP, iters=BENCH_MEASURE, flush_l2=True): """Per-iteration CUDA events timer with L2 flush and median latency.""" - flush_buf = None + flush_bytes = 0 if flush_l2: props = torch.cuda.get_device_properties(torch.cuda.current_device()) l2_bytes = getattr(props, "L2_cache_size", 4 * 1024 * 1024) - flush_buf = torch.empty(max(l2_bytes * 2, 8 * 1024 * 1024), dtype=torch.uint8, device="cuda") - - for _ in range(warmup): - if flush_buf is not None: - flush_buf.zero_() - fn() - torch.cuda.synchronize() - - starts = [torch.cuda.Event(enable_timing=True) for _ in range(iters)] - ends = [torch.cuda.Event(enable_timing=True) for _ in range(iters)] - for i in range(iters): - if flush_buf is not None: - flush_buf.zero_() - starts[i].record() - fn() - ends[i].record() - torch.cuda.synchronize() - - latencies = sorted(starts[i].elapsed_time(ends[i]) * 1e3 for i in range(iters)) - n = len(latencies) - if n >= 8: - q1, q3 = latencies[n // 4], latencies[3 * n // 4] - iqr = q3 - q1 - lo, hi = q1 - 1.5 * iqr, q3 + 1.5 * iqr - latencies = [x for x in latencies if lo <= x <= hi] or latencies - del flush_buf - return latencies[len(latencies) // 2] + flush_bytes = max(l2_bytes * 2, 8 * 1024 * 1024) + + return do_bench( + fn, + warmup=warmup, + rep=iters, + schedule="per_iter", + statistic="median", + flush_bytes=flush_bytes, + iqr=True, + unit="us", + ) -def bench_kernel_us(fn, warmup=BENCH_WARMUP, iters=BENCH_MEASURE): +def bench_profiled_device_us(fn, warmup=BENCH_WARMUP, iters=BENCH_MEASURE): """Pure on-device kernel time (per invocation, microseconds). Uses ``torch.profiler`` (CUPTI on CUDA, roctracer on ROCm) to capture @@ -928,7 +917,7 @@ def bench_kernel_us(fn, warmup=BENCH_WARMUP, iters=BENCH_MEASURE): - ``bench_graph_us`` measures end-to-end CUDA-graph replay latency, which still includes graph-replay overhead and any inter-kernel dispatch gaps on the GPU command processor. - - ``bench_kernel_us`` measures only the wall time the GPU is actually + - ``bench_profiled_device_us`` measures only the wall time the GPU is actually executing kernels — i.e. the floor on kernel runtime, with launch and dispatch effects removed. @@ -966,8 +955,48 @@ def bench_kernel_us(fn, warmup=BENCH_WARMUP, iters=BENCH_MEASURE): return total_us / iters -def bench_graph_us(fn, warmup=BENCH_WARMUP, iters=BENCH_MEASURE): - """CUDA graph benchmark — amortizes kernel launch overhead.""" +def _make_moe_graph_verifier(outputs_fn, unit_size, extra_outputs_fn=None): + """Verify every meaningful MoE sorting output after graph replay.""" + references = tuple(output.clone() for output in outputs_fn()) + extra_references = ( + tuple(output.clone() for output in extra_outputs_fn()) + if extra_outputs_fn is not None + else () + ) + num_padded = int(references[3][0].item()) + num_valid_blocks = num_padded // unit_size + + def verify(replay): + outputs = outputs_fn() + extra_outputs = extra_outputs_fn() if extra_outputs_fn is not None else () + for output in (*outputs, *extra_outputs): + output.fill_(1 if torch.is_floating_point(output) else -1) + torch.cuda.synchronize() + replay() + torch.cuda.synchronize() + ids, weights, expert_ids, nvalid, moe_buf = outputs + ref_ids, ref_weights, ref_expert_ids, ref_nvalid, ref_moe_buf = references + return ( + torch.equal(nvalid, ref_nvalid) + and torch.equal(moe_buf, ref_moe_buf) + and torch.equal(ids[:num_padded], ref_ids[:num_padded]) + and torch.equal(weights[:num_padded], ref_weights[:num_padded]) + and torch.equal(expert_ids[:num_valid_blocks], ref_expert_ids[:num_valid_blocks]) + and all( + torch.equal(output, reference) + for output, reference in zip(extra_outputs, extra_references) + ) + ) + + return verify + + +def bench_graph_us(fn, verify, warmup=BENCH_WARMUP, iters=BENCH_MEASURE, calls_per_replay=1): + """Measure verified CUDA graph replay latency in microseconds per call.""" + if verify is None: + raise ValueError("graph benchmarks require an output verifier") + if calls_per_replay <= 0: + raise ValueError("calls_per_replay must be positive") for _ in range(warmup): fn() torch.cuda.synchronize() @@ -979,26 +1008,26 @@ def bench_graph_us(fn, warmup=BENCH_WARMUP, iters=BENCH_MEASURE): fn() torch.cuda.current_stream().wait_stream(stream) torch.cuda.synchronize() - graph = torch.cuda.CUDAGraph() with torch.cuda.stream(stream): with torch.cuda.graph(graph, stream=stream): - fn() + for _ in range(calls_per_replay): + fn() torch.cuda.current_stream().wait_stream(stream) - for _ in range(warmup): - graph.replay() - torch.cuda.synchronize() except RuntimeError: return None # graph capture not supported - - start = torch.cuda.Event(enable_timing=True) - end = torch.cuda.Event(enable_timing=True) - start.record() - for _ in range(iters): - graph.replay() - end.record() - torch.cuda.synchronize() - return start.elapsed_time(end) * 1e3 / iters + if not verify(graph.replay): + raise AssertionError("graph replay did not reproduce the reference output") + + replay_us = do_bench( + graph.replay, + warmup=warmup, + rep=iters, + schedule="pipelined", + statistic="mean", + unit="us", + ) + return replay_us / calls_per_replay def run_bench_comparison(token_sweep=None): @@ -1064,22 +1093,42 @@ def fly_fn(): E, UNIT_SIZE, ) + return fly_nvalid + fly_fn() + torch.cuda.synchronize() + fly_verify = _make_moe_graph_verifier( + lambda: ( + fly_sorted_ids, + fly_sorted_w, + fly_sorted_eids, + fly_nvalid, + fly_moe_buf_2d, + ), + UNIT_SIZE, + ) fly_eager = bench_eager_us(fly_fn) - fly_graph = bench_graph_us(fly_fn) - fly_kernel = bench_kernel_us(fly_fn) + fly_graph = bench_graph_us(fly_fn, fly_verify) + fly_kernel = bench_profiled_device_us(fly_fn) ck_eager, ck_graph, ck_kernel = None, None, None if aiter_moe_sorting is not None: + ck_state = {} def ck_fn(): - aiter_moe_sorting( + outputs = aiter_moe_sorting( topk_ids, topk_weights, E, model_dim=model_dim, moebuf_dtype=torch.bfloat16, block_size=UNIT_SIZE ) + ck_state["outputs"] = outputs + return outputs + + ck_fn() + torch.cuda.synchronize() + ck_verify = _make_moe_graph_verifier(lambda: ck_state["outputs"], UNIT_SIZE) ck_eager = bench_eager_us(ck_fn) - ck_graph = bench_graph_us(ck_fn) - ck_kernel = bench_kernel_us(ck_fn) + ck_graph = bench_graph_us(ck_fn, ck_verify) + ck_kernel = bench_profiled_device_us(ck_fn) def fmt(v): return f"{v:8.1f}us" if v is not None else " N/A" @@ -1193,6 +1242,7 @@ def unfused_fn(): E, UNIT_SIZE, ) + return nvalid def fused_fn(): moe_softmax_sort_flydsl( @@ -1207,18 +1257,27 @@ def fused_fn(): dtype_str, unit_size=UNIT_SIZE, ) + return nvalid # Warm up both paths once before measurement (covers compile cache). unfused_fn() + torch.cuda.synchronize() + outputs_fn = lambda: (sorted_ids, sorted_w, sorted_eids, nvalid, moe_buf_2d) + unfused_verify = _make_moe_graph_verifier( + outputs_fn, + UNIT_SIZE, + extra_outputs_fn=lambda: (u_topk_w, u_topk_ids, u_tei), + ) fused_fn() torch.cuda.synchronize() + fused_verify = _make_moe_graph_verifier(outputs_fn, UNIT_SIZE) unfused_eager = bench_eager_us(unfused_fn) fused_eager = bench_eager_us(fused_fn) - unfused_graph = bench_graph_us(unfused_fn) - fused_graph = bench_graph_us(fused_fn) - unfused_kernel = bench_kernel_us(unfused_fn) - fused_kernel = bench_kernel_us(fused_fn) + unfused_graph = bench_graph_us(unfused_fn, unfused_verify) + fused_graph = bench_graph_us(fused_fn, fused_verify) + unfused_kernel = bench_profiled_device_us(unfused_fn) + fused_kernel = bench_profiled_device_us(fused_fn) path = "fused" if T <= 16 else "fallback" diff --git a/tests/kernels/test_rmsnorm.py b/tests/kernels/test_rmsnorm.py index c7175c2ca..eec5989f3 100644 --- a/tests/kernels/test_rmsnorm.py +++ b/tests/kernels/test_rmsnorm.py @@ -218,8 +218,17 @@ def kernel_launch(): weight_elem_bytes = 4 if weight_dtype == "f32" else 2 total_bytes = 2 * M * N * elem_bytes + N * weight_elem_bytes bandwidth_gbs = total_bytes / (avg_us / 1e6) / 1e9 + benchmark_instrument = ( + "device_event" + if int(os.environ.get("FLYDSL_PERFTEST_USE_EVENTS", 0)) + else "torch_profiler" + ) print(f"Kernel avg time: {avg_ms:.4f} ms via run_perftest (warmup={WARMUP_ITERS}, iters={BENCH_ITERS})") + print( + f"Benchmark contract: instrument={benchmark_instrument} schedule=per_iter_sync " + f"cache=warm statistic=mean warmup={WARMUP_ITERS} iters={BENCH_ITERS}" + ) print(f"Bandwidth: {bandwidth_gbs:.2f} GB/s") if flydsl_gpu_us is not None: print(f"[Perf] FlyDSL rmsnorm gpu: {flydsl_gpu_us:.1f} us") diff --git a/tests/kernels/test_softmax.py b/tests/kernels/test_softmax.py index a6a5d5dd0..0641fa7d7 100644 --- a/tests/kernels/test_softmax.py +++ b/tests/kernels/test_softmax.py @@ -95,7 +95,16 @@ def kernel_launch(): avg_ms = avg_us / 1000.0 total_bytes = 2 * M * N * (4 if dtype_str == "f32" else 2) # read input + write output bandwidth_gbs = total_bytes / (avg_us / 1e6) / 1e9 + benchmark_instrument = ( + "device_event" + if int(os.environ.get("FLYDSL_PERFTEST_USE_EVENTS", 0)) + else "torch_profiler" + ) print(f"Kernel avg time: {avg_ms:.4f} ms via run_perftest (warmup={WARMUP_ITERS}, iters={BENCH_ITERS})") + print( + f"Benchmark contract: instrument={benchmark_instrument} schedule=per_iter_sync " + f"cache=warm statistic=mean warmup={WARMUP_ITERS} iters={BENCH_ITERS}" + ) print(f"Bandwidth: {bandwidth_gbs:.2f} GB/s") if flydsl_gpu_us is not None: print(f"[Perf] FlyDSL softmax gpu: {flydsl_gpu_us:.1f} us") diff --git a/tests/perf/bench_tdm_bandwidth_gfx1250.py b/tests/perf/bench_tdm_bandwidth_gfx1250.py index 9ac37b509..584a97fd2 100644 --- a/tests/perf/bench_tdm_bandwidth_gfx1250.py +++ b/tests/perf/bench_tdm_bandwidth_gfx1250.py @@ -27,6 +27,7 @@ import flydsl.expr as fx from flydsl._mlir import ir from flydsl._mlir.dialects import vector +from flydsl.autotune import do_bench from flydsl.compiler.kernel_function import CompilationContext from flydsl.expr import arith, as_ir_value, const_expr, gpu, range_constexpr, tdm_ops from flydsl.expr.rocdl import cluster @@ -348,7 +349,7 @@ def launch(A: fx.Tensor, B: fx.Tensor, C: fx.Tensor, stream: fx.Stream = fx.Stre # --------------------------------------------------------------------------- -# Benchmark timing (adapted from benchmark_common.py:bench_kernel_us) +# Benchmark timing # --------------------------------------------------------------------------- @@ -359,37 +360,16 @@ def _bench_kernel_us(run_fn, warmup=10, iters=50, flush_mb=512): "L2_cache_size", 256 * 1024 * 1024, # fallback if L2_cache_size unavailable (cmodel reports 96 MB) ) - alloc_bytes = max(l2_bytes * 2, flush_mb * 1024 * 1024) - flush_buf = torch.empty(alloc_bytes, dtype=torch.uint8, device="cuda") - - for _ in range(warmup): - flush_buf.zero_() - run_fn() - torch.cuda.synchronize() - - start_ev = [torch.cuda.Event(enable_timing=True) for _ in range(iters)] - end_ev = [torch.cuda.Event(enable_timing=True) for _ in range(iters)] - - for i in range(iters): - flush_buf.zero_() - start_ev[i].record() - run_fn() - end_ev[i].record() - - torch.cuda.synchronize() - latencies = sorted(start_ev[i].elapsed_time(end_ev[i]) * 1e3 for i in range(iters)) - - n = len(latencies) - if n >= 8: - q1, q3 = latencies[n // 4], latencies[3 * n // 4] - iqr = q3 - q1 - lo, hi = q1 - 1.5 * iqr, q3 + 1.5 * iqr - filtered = [x for x in latencies if lo <= x <= hi] - if filtered: - latencies = filtered - - del flush_buf - return latencies[len(latencies) // 2] + return do_bench( + run_fn, + warmup=warmup, + rep=iters, + schedule="per_iter", + statistic="median", + flush_bytes=max(l2_bytes * 2, flush_mb * 1024 * 1024), + iqr=True, + unit="us", + ) # --------------------------------------------------------------------------- diff --git a/tests/unit/test_benchmark_compare.py b/tests/unit/test_benchmark_compare.py new file mode 100644 index 000000000..248a4724a --- /dev/null +++ b/tests/unit/test_benchmark_compare.py @@ -0,0 +1,486 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright (c) 2026 FlyDSL Project Contributors + +"""GPU-free tests for benchmark regression policy and CSV compatibility.""" + +import csv +import importlib +import json +import sys +from pathlib import Path + +import pytest + +SCRIPTS = Path(__file__).resolve().parents[2] / "scripts" +sys.path.insert(0, str(SCRIPTS)) + +from benchmark_compare import Threshold, ThresholdConfig, compare_values # noqa: E402 + +compare_benchmark = importlib.import_module("compare_benchmark") + + +@pytest.mark.parametrize( + "current,relative,absolute,expected", + [ + (121.0, 20.0, 10.0, True), + (119.0, 20.0, 10.0, False), + (121.0, 25.0, 10.0, False), + (109.0, 5.0, 10.0, False), + ], +) +def test_dual_threshold_requires_relative_and_absolute(current, relative, absolute, expected): + result = compare_values( + 100.0, + current, + Threshold(relative_pct=relative, absolute=absolute), + ) + assert result.regressed is expected + + +def test_higher_better_direction_flags_decrease(): + result = compare_values( + 100.0, + 75.0, + Threshold(relative_pct=20.0, absolute=10.0, direction="higher_better"), + ) + assert result.regressed + assert result.delta == 25.0 + + +def test_threshold_config_is_an_arch_allowlist(): + config = ThresholdConfig( + { + "version": 1, + "architectures": { + "gfx950": [ + { + "op": "rmsnorm", + "shape": "*", + "dtype": "bf16", + "relative_pct": 20, + "absolute_us": 10, + } + ] + }, + } + ) + + assert config.match(arch="gfx950", op="rmsnorm", shape="1x2", dtype="bf16") is not None + assert config.match(arch="gfx942", op="rmsnorm", shape="1x2", dtype="bf16") is None + + +def _write_csv( + path, + *, + avg_us, + status="ok", + arch="gfx950", + tbps="4.0", + tflops="-", + statistic="mean", +): + fields = [ + "op", + "shape", + "dtype", + "tbps", + "tflops", + "status", + "avg_us", + "statistic", + "instrument", + "schedule", + "cache_policy", + "warmup", + "iters", + "arch", + ] + with path.open("w", newline="") as handle: + writer = csv.DictWriter(handle, fieldnames=fields) + writer.writeheader() + writer.writerow( + { + "op": "softmax", + "shape": "32768x8192", + "dtype": "bf16", + "tbps": tbps, + "tflops": tflops, + "status": status, + "avg_us": avg_us, + "statistic": statistic, + "instrument": "reported", + "schedule": "unknown", + "cache_policy": "unknown", + "warmup": "10", + "iters": "100", + "arch": arch, + } + ) + + +def test_compare_benchmark_hard_fails_allowlisted_latency(tmp_path, monkeypatch): + baseline = tmp_path / "base.csv" + current = tmp_path / "current.csv" + thresholds = tmp_path / "thresholds.json" + _write_csv(baseline, avg_us="100") + _write_csv(current, avg_us="125") + thresholds.write_text( + json.dumps( + { + "version": 1, + "architectures": { + "gfx950": [ + { + "op": "softmax", + "shape": "*", + "dtype": "*", + "relative_pct": 20, + "absolute_us": 10, + } + ] + }, + } + ) + ) + monkeypatch.setattr( + sys, + "argv", + [ + "compare_benchmark.py", + str(baseline), + str(current), + "--threshold-config", + str(thresholds), + "--fail-on-regression", + ], + ) + + assert compare_benchmark.main() == 1 + + +def test_compare_benchmark_accepts_legacy_csv_without_raw_us(tmp_path, monkeypatch): + header = "op,shape,dtype,tbps,tflops,status\n" + row = "softmax,32768x8192,bf16,4.0,-,ok\n" + baseline = tmp_path / "base.csv" + current = tmp_path / "current.csv" + baseline.write_text(header + row) + current.write_text(header + row) + monkeypatch.setattr(sys, "argv", ["compare_benchmark.py", str(baseline), str(current)]) + + assert compare_benchmark.main() == 0 + + +def test_compare_benchmark_fails_when_allowlisted_row_disappears(tmp_path, monkeypatch): + baseline = tmp_path / "base.csv" + current = tmp_path / "current.csv" + thresholds = tmp_path / "thresholds.json" + _write_csv(baseline, avg_us="100") + current.write_text( + "op,shape,dtype,tbps,tflops,status,avg_us,arch\n" + "gemm,1x1x1,bf16,-,1.0,ok,10,gfx950\n" + ) + thresholds.write_text( + json.dumps( + { + "version": 1, + "architectures": { + "gfx950": [ + { + "op": "softmax", + "shape": "*", + "dtype": "*", + "relative_pct": 20, + "absolute_us": 10, + } + ] + }, + } + ) + ) + monkeypatch.setattr( + sys, + "argv", + [ + "compare_benchmark.py", + str(baseline), + str(current), + "--threshold-config", + str(thresholds), + "--fail-on-regression", + ], + ) + + assert compare_benchmark.main() == 1 + + +def test_raw_us_gate_is_independent_of_throughput_metric(tmp_path, monkeypatch): + baseline = tmp_path / "base.csv" + current = tmp_path / "current.csv" + thresholds = tmp_path / "thresholds.json" + _write_csv(baseline, avg_us="100", tbps="4.0", tflops="-") + _write_csv(current, avg_us="125", tbps="-", tflops="10.0") + thresholds.write_text( + json.dumps( + { + "version": 1, + "architectures": { + "gfx950": [ + { + "op": "softmax", + "shape": "*", + "dtype": "*", + "relative_pct": 20, + "absolute_us": 10, + } + ] + }, + } + ) + ) + monkeypatch.setattr( + sys, + "argv", + [ + "compare_benchmark.py", + str(baseline), + str(current), + "--threshold-config", + str(thresholds), + "--fail-on-regression", + ], + ) + assert compare_benchmark.main() == 1 + + +def test_arch_mismatch_never_hard_gates(tmp_path, monkeypatch): + baseline = tmp_path / "base.csv" + current = tmp_path / "current.csv" + thresholds = tmp_path / "thresholds.json" + _write_csv(baseline, avg_us="100", arch="gfx942") + _write_csv(current, avg_us="150", arch="gfx950") + thresholds.write_text( + json.dumps( + { + "version": 1, + "architectures": { + "gfx950": [ + { + "op": "softmax", + "shape": "*", + "dtype": "*", + "relative_pct": 20, + "absolute_us": 10, + } + ] + }, + } + ) + ) + monkeypatch.setattr( + sys, + "argv", + [ + "compare_benchmark.py", + str(baseline), + str(current), + "--threshold-config", + str(thresholds), + "--fail-on-regression", + ], + ) + assert compare_benchmark.main() == 0 + + +def test_allowlisted_current_row_without_raw_us_is_broken(tmp_path, monkeypatch): + baseline = tmp_path / "base.csv" + current = tmp_path / "current.csv" + thresholds = tmp_path / "thresholds.json" + _write_csv(baseline, avg_us="100") + _write_csv(current, avg_us="-") + thresholds.write_text( + json.dumps( + { + "version": 1, + "architectures": { + "gfx950": [ + { + "op": "softmax", + "shape": "*", + "dtype": "*", + "relative_pct": 20, + "absolute_us": 10, + } + ] + }, + } + ) + ) + monkeypatch.setattr( + sys, + "argv", + [ + "compare_benchmark.py", + str(baseline), + str(current), + "--threshold-config", + str(thresholds), + "--fail-on-regression", + ], + ) + assert compare_benchmark.main() == 1 + + +def test_baseline_only_row_on_another_arch_is_not_gated(tmp_path, monkeypatch): + baseline = tmp_path / "base.csv" + current = tmp_path / "current.csv" + thresholds = tmp_path / "thresholds.json" + _write_csv(baseline, avg_us="100", arch="gfx942") + current.write_text( + "op,shape,dtype,tbps,tflops,status,avg_us,arch\n" + "gemm,1x1x1,bf16,-,1.0,ok,10,gfx950\n" + ) + thresholds.write_text( + json.dumps( + { + "version": 1, + "architectures": { + "gfx942": [ + { + "op": "softmax", + "shape": "*", + "dtype": "*", + "relative_pct": 20, + "absolute_us": 10, + } + ], + "gfx950": [], + }, + } + ) + ) + monkeypatch.setattr( + sys, + "argv", + [ + "compare_benchmark.py", + str(baseline), + str(current), + "--threshold-config", + str(thresholds), + "--fail-on-regression", + ], + ) + assert compare_benchmark.main() == 0 + + +def test_measurement_contract_mismatch_fails_closed(tmp_path, monkeypatch): + baseline = tmp_path / "base.csv" + current = tmp_path / "current.csv" + thresholds = tmp_path / "thresholds.json" + _write_csv(baseline, avg_us="100", statistic="mean") + _write_csv(current, avg_us="100", statistic="median") + thresholds.write_text( + json.dumps( + { + "version": 1, + "architectures": { + "gfx950": [ + { + "op": "softmax", + "shape": "*", + "dtype": "*", + "relative_pct": 20, + "absolute_us": 10, + } + ] + }, + } + ) + ) + monkeypatch.setattr( + sys, + "argv", + [ + "compare_benchmark.py", + str(baseline), + str(current), + "--threshold-config", + str(thresholds), + "--fail-on-regression", + ], + ) + assert compare_benchmark.main() == 1 + + +def test_unknown_arch_cannot_disable_hard_gate(tmp_path, monkeypatch): + baseline = tmp_path / "base.csv" + current = tmp_path / "current.csv" + thresholds = tmp_path / "thresholds.json" + _write_csv(baseline, avg_us="100", arch="unknown") + _write_csv(current, avg_us="100", arch="unknown") + thresholds.write_text(json.dumps({"version": 1, "architectures": {"gfx950": []}})) + monkeypatch.setattr( + sys, + "argv", + [ + "compare_benchmark.py", + str(baseline), + str(current), + "--threshold-config", + str(thresholds), + "--fail-on-regression", + ], + ) + assert compare_benchmark.main() == 2 + + +def test_csv_arch_cannot_override_trusted_runner_arch(tmp_path, monkeypatch): + baseline = tmp_path / "base.csv" + current = tmp_path / "current.csv" + thresholds = tmp_path / "thresholds.json" + _write_csv(baseline, avg_us="100", arch="gfx950") + _write_csv(current, avg_us="100", arch="gfx1201") + thresholds.write_text( + json.dumps({"version": 1, "architectures": {"gfx950": [], "gfx1201": []}}) + ) + monkeypatch.setattr( + sys, + "argv", + [ + "compare_benchmark.py", + str(baseline), + str(current), + "--arch", + "gfx950", + "--threshold-config", + str(thresholds), + "--fail-on-regression", + ], + ) + assert compare_benchmark.main() == 2 + + +def test_baseline_arch_must_match_trusted_runner_arch(tmp_path, monkeypatch): + baseline = tmp_path / "base.csv" + current = tmp_path / "current.csv" + thresholds = tmp_path / "thresholds.json" + _write_csv(baseline, avg_us="100", arch="gfx1201") + _write_csv(current, avg_us="100", arch="gfx950") + thresholds.write_text( + json.dumps({"version": 1, "architectures": {"gfx950": [], "gfx1201": []}}) + ) + monkeypatch.setattr( + sys, + "argv", + [ + "compare_benchmark.py", + str(baseline), + str(current), + "--arch", + "gfx950", + "--threshold-config", + str(thresholds), + "--fail-on-regression", + ], + ) + assert compare_benchmark.main() == 2 diff --git a/tests/unit/test_benchmark_log_parser.py b/tests/unit/test_benchmark_log_parser.py new file mode 100644 index 000000000..cc47257bf --- /dev/null +++ b/tests/unit/test_benchmark_log_parser.py @@ -0,0 +1,118 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright (c) 2026 FlyDSL Project Contributors + +"""Golden log tests for raw-us benchmark extraction.""" + +import csv +import importlib +import sys +from pathlib import Path + +import pytest + +SCRIPTS = Path(__file__).resolve().parents[2] / "scripts" +sys.path.insert(0, str(SCRIPTS)) + +from benchmark_log_parser import parse_metrics, parse_moe_stage2 # noqa: E402 + +benchmark_output_to_csv = importlib.import_module("benchmark_output_to_csv") + + +@pytest.mark.parametrize( + "text,expected_us,expected_tbps,expected_tflops,expected_statistic", + [ + ( + "[flyc] Throughput: 123.4 us, 55.6 TFLOPS, BW: 1.234 TB/s", + 123.4, + 1.234, + 55.6, + "mean", + ), + ("| PASS | 1.0e-3 0.999 | 42.5 100.2", 42.5, None, 100.2, "reported"), + ( + "cos_diff=1e-3 TFLOPS=88.0 TB/s=2.5 err=0 us_p50=91.5 us_range=[90, 93]", + 91.5, + 2.5, + 88.0, + "median", + ), + ( + "Kernel avg time: 0.1250 ms\nBandwidth: 3500.0 GB/s", + 125.0, + 3.5, + None, + "mean", + ), + ], +) +def test_parse_metrics_normalizes_raw_microseconds( + text, + expected_us, + expected_tbps, + expected_tflops, + expected_statistic, +): + metrics = parse_metrics(text) + assert metrics.avg_us == expected_us + assert metrics.tbps == expected_tbps + assert metrics.tflops == expected_tflops + assert metrics.statistic == expected_statistic + + +def test_softmax_style_keeps_first_base_measurement(): + metrics = parse_metrics( + "Kernel avg time: 0.1000 ms\nBandwidth: 4000 GB/s\n" + "Kernel avg time: 0.5000 ms\nBandwidth: 1000 GB/s\n" + ) + assert metrics.avg_us == 100.0 + assert metrics.tbps == 4.0 + + +def test_machine_readable_contract_overrides_fallback_metadata(): + metrics = parse_metrics( + "Kernel avg time: 0.1000 ms\n" + "Benchmark contract: instrument=device_event schedule=pipelined " + "cache=cold statistic=median warmup=7 iters=50\n" + "Bandwidth: 4000 GB/s\n" + ) + assert metrics.instrument == "device_event" + assert metrics.schedule == "pipelined" + assert metrics.cache_policy == "cold" + assert metrics.statistic == "median" + assert metrics.warmup == "7" + assert metrics.iters == "50" + + +def test_parse_moe_stage2_preserves_modes_and_latency(): + found = parse_moe_stage2( + "FlyDSL MoE stage2 [moe_gemm2] fp4 atomic | x | " + "1163.2 us, 1654.24 TFLOPS, 0.377 TB/s\n" + "FlyDSL MoE stage2 [moe_gemm2] fp4 reduce | x | " + "1200.0 us, 1600.00 TFLOPS, 0.350 TB/s" + ) + assert found["atomic"][1].avg_us == 1163.2 + assert found["reduce"][1].tflops == 1600.0 + + +def test_legacy_five_column_output_converts_to_enriched_csv(tmp_path, monkeypatch): + source = tmp_path / "benchmark.out" + destination = tmp_path / "benchmark.csv" + long_shape = "5120x5120x8320_tile128x256x128_2tg" + source.write_text( + "op shape dtype TB/s TFLOPS\n" + f"gemm {long_shape} bf16 4.000 -\n" + ) + monkeypatch.setattr( + sys, + "argv", + ["benchmark_output_to_csv.py", str(source), str(destination)], + ) + + assert benchmark_output_to_csv.main() == 0 + with destination.open(newline="") as handle: + rows = list(csv.DictReader(handle)) + assert rows[0]["tbps"] == "4.000" + assert rows[0]["shape"] == long_shape + assert rows[0]["avg_us"] == "-" + assert rows[0]["statistic"] == "reported" + assert rows[0]["instrument"] == "reported" diff --git a/tests/unit/test_benchmark_timer.py b/tests/unit/test_benchmark_timer.py new file mode 100644 index 000000000..53cdf6776 --- /dev/null +++ b/tests/unit/test_benchmark_timer.py @@ -0,0 +1,202 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright (c) 2025 FlyDSL Project Contributors + +"""GPU-free contract tests for the canonical CUDA/HIP event timer.""" + +import importlib +from contextlib import nullcontext + +import pytest + +from flydsl import BenchResult, do_bench + +autotune = importlib.import_module("flydsl.autotune") + + +class _FakeEvent: + def __init__(self, cuda): + self._cuda = cuda + self._timestamp = None + self.recorded_stream = None + + def record(self, stream=None): + self._timestamp = self._cuda.clock_ms + self.recorded_stream = stream + + def synchronize(self): + self._cuda.event_synchronize_count += 1 + + def elapsed_time(self, other): + return other._timestamp - self._timestamp + + +class _FakeBuffer: + def __init__(self, device): + self.device = device + self.zero_count = 0 + + def zero_(self): + self.zero_count += 1 + + +class _FakeCuda: + def __init__(self): + self.clock_ms = 0.0 + self.global_synchronize_count = 0 + self.event_synchronize_count = 0 + self.events = [] + + def Event(self, enable_timing): + assert enable_timing + event = _FakeEvent(self) + self.events.append(event) + return event + + def synchronize(self): + self.global_synchronize_count += 1 + + def stream(self, stream): + return nullcontext(stream) + + +class _FakeTorch: + uint8 = object() + + def __init__(self): + self.cuda = _FakeCuda() + self.buffers = [] + + def empty(self, size, *, dtype, device): + assert size > 0 + assert dtype is self.uint8 + buffer = _FakeBuffer(device) + self.buffers.append(buffer) + return buffer + + +@pytest.fixture +def fake_torch(monkeypatch): + fake = _FakeTorch() + monkeypatch.setattr(autotune, "torch", fake) + return fake + + +def test_legacy_do_bench_keeps_scalar_milliseconds(fake_torch): + def fn(): + fake_torch.cuda.clock_ms += 0.002 + + value = do_bench(fn, warmup=1, rep=3) + quantiles = do_bench(fn, warmup=0, rep=3, quantiles=[0.5]) + + assert value == pytest.approx(0.002) + assert quantiles == pytest.approx([0.002]) + assert fake_torch.cuda.global_synchronize_count == 8 + assert fake_torch.cuda.event_synchronize_count == 0 + + +def test_structured_result_uses_microseconds(fake_torch): + def fn(): + fake_torch.cuda.clock_ms += 0.002 + + result = do_bench(fn, warmup=1, rep=3, return_result=True) + + assert isinstance(result, BenchResult) + assert result.value_us == pytest.approx(2.0) + assert result.samples_us == pytest.approx((2.0, 2.0, 2.0)) + assert result.schedule == "isolated" + assert result.statistic == "median" + assert result.sample_count == 3 + assert result.cache_policy == "warm" + + +def test_pipelined_schedule_returns_average_per_call(fake_torch): + def fn(): + fake_torch.cuda.clock_ms += 0.003 + + result = do_bench( + fn, + warmup=0, + rep=4, + schedule="pipelined", + statistic="mean", + return_result=True, + ) + + assert result.value_us == pytest.approx(3.0) + assert result.samples_us == pytest.approx((3.0,)) + + +def test_per_iter_schedule_preserves_prep_flush_and_stream(fake_torch): + prep_count = 0 + stream = object() + + def prep(): + nonlocal prep_count + prep_count += 1 + + def fn(): + fake_torch.cuda.clock_ms += 0.004 + + result = do_bench( + fn, + warmup=2, + rep=3, + schedule="per_iter", + prep_fn=prep, + flush_bytes=4096, + iqr=True, + stream=stream, + return_result=True, + ) + + assert result.value_us == pytest.approx(4.0) + assert prep_count == 5 + assert fake_torch.buffers[0].zero_count == 5 + assert result.cache_policy == "flush:4096" + assert all(event.recorded_stream is stream for event in fake_torch.cuda.events) + + +def test_pipelined_schedule_rejects_non_mean_statistic(fake_torch): + with pytest.raises(ValueError, match="only supports"): + do_bench(lambda: None, schedule="pipelined", statistic="median") + with pytest.raises(ValueError, match="cannot include"): + do_bench( + lambda: None, + schedule="pipelined", + statistic="mean", + prep_fn=lambda: None, + ) + + +def test_median_average_preserves_statistics_median_semantics(fake_torch): + durations = iter([0.001, 0.003]) + + def fn(): + fake_torch.cuda.clock_ms += next(durations) + + result = do_bench( + fn, + warmup=0, + rep=2, + schedule="per_iter", + statistic="median_average", + return_result=True, + ) + + assert result.value_us == pytest.approx(2.0) + + +def test_flush_buffer_uses_explicit_stream_device(fake_torch): + class Stream: + device = "cuda:7" + + do_bench( + lambda: None, + warmup=0, + rep=1, + schedule="per_iter", + flush_bytes=1024, + stream=Stream(), + ) + + assert fake_torch.buffers[0].device == "cuda:7" diff --git a/tests/unit/test_launch_overhead.py b/tests/unit/test_launch_overhead.py index 3f4e7deab..bda765df7 100644 --- a/tests/unit/test_launch_overhead.py +++ b/tests/unit/test_launch_overhead.py @@ -96,7 +96,7 @@ def triton_vec_add_kernel(a_ptr, b_ptr, c_ptr, n, BLOCK: tl.constexpr = 1024): HAS_TRITON = False -def bench_wallclock(fn, n_warmup=20, n_iters=1000): +def bench_host_dispatch_us(fn, n_warmup=20, n_iters=1000): """Measure wall-clock time per call (no GPU sync between calls). This measures CPU dispatch overhead: the time from Python calling the @@ -118,6 +118,9 @@ def bench_wallclock(fn, n_warmup=20, n_iters=1000): return (t1 - t0) / n_iters * 1e6 # µs +bench_wallclock = bench_host_dispatch_us + + def test_bench_wallclock_excludes_final_gpu_drain(monkeypatch): """The final drain must not be part of the host-dispatch window.""" sync_count = 0 @@ -129,7 +132,7 @@ def fake_synchronize(): monkeypatch.setattr(torch.cuda, "synchronize", fake_synchronize) monkeypatch.setattr(time, "perf_counter", lambda: float(sync_count)) - measured_us = bench_wallclock(lambda: None, n_warmup=0, n_iters=1) + measured_us = bench_host_dispatch_us(lambda: None, n_warmup=0, n_iters=1) assert measured_us == 0.0 assert sync_count == 2 @@ -165,19 +168,19 @@ def main(): assert err < 1e-5, f"FlyDSL correctness failed: max_err={err}" # ── Bench flyc.compile'd function ── - compiled_us = bench_wallclock( + compiled_us = bench_host_dispatch_us( lambda: compiled(a, b, c, SIZE, SIZE, BLOCK, VEC, stream), n_iters=N_ITERS, ) # ── Bench @flyc.jit (implicit path) ── - flydsl_us = bench_wallclock( + flydsl_us = bench_host_dispatch_us( lambda: vecAdd(a, b, c, SIZE, SIZE, BLOCK, VEC, stream), n_iters=N_ITERS, ) # ── Bench PyTorch ── - torch_us = bench_wallclock( + torch_us = bench_host_dispatch_us( lambda: torch.add(a, b, out=c), n_iters=N_ITERS, ) @@ -190,7 +193,7 @@ def main(): triton_vec_add_kernel[grid](a, b, c, SIZE) torch.cuda.synchronize() - triton_us = bench_wallclock( + triton_us = bench_host_dispatch_us( lambda: triton_vec_add_kernel[grid](a, b, c, SIZE), n_iters=N_ITERS, ) diff --git a/tests/unit/test_tdm_mcast_add_gfx1250.py b/tests/unit/test_tdm_mcast_add_gfx1250.py index f4fbd4359..7c99c2ebc 100644 --- a/tests/unit/test_tdm_mcast_add_gfx1250.py +++ b/tests/unit/test_tdm_mcast_add_gfx1250.py @@ -37,6 +37,7 @@ if torch is None or not torch.cuda.is_available(): pytest.skip("CUDA/ROCm not available.", allow_module_level=True) +from flydsl.autotune import do_bench # noqa: E402 from flydsl.runtime.device import get_rocm_arch # noqa: E402 _arch = str(get_rocm_arch()) @@ -224,21 +225,17 @@ def _run_tdm_mcast_add(grid_m, grid_n, cluster_x, cluster_y, n_warmup=0, n_iters launch_fn = _compile_tdm_mcast_add(grid_m, grid_n, cluster_x, cluster_y) stream = torch.cuda.Stream() - # Warmup - for _ in range(n_warmup): - launch_fn(a_dev, b_dev, c_dev, stream=stream) - torch.cuda.synchronize() - avg_us = 0.0 if n_iters > 1 and n_warmup > 0: - start_event = torch.cuda.Event(enable_timing=True) - end_event = torch.cuda.Event(enable_timing=True) - start_event.record(stream) - for _ in range(n_iters): - launch_fn(a_dev, b_dev, c_dev, stream=stream) - end_event.record(stream) - torch.cuda.synchronize() - avg_us = start_event.elapsed_time(end_event) * 1000.0 / n_iters + avg_us = do_bench( + lambda: launch_fn(a_dev, b_dev, c_dev, stream=stream), + warmup=n_warmup, + rep=n_iters, + schedule="pipelined", + statistic="mean", + stream=stream, + unit="us", + ) else: launch_fn(a_dev, b_dev, c_dev, stream=stream) torch.cuda.synchronize() From 0d4aa3d9aff764be9328066258422da0e1620459 Mon Sep 17 00:00:00 2001 From: Jin Pan Date: Thu, 30 Jul 2026 00:42:27 +0000 Subject: [PATCH 3/3] style: format timing umbrella Python changes Apply the repository Black configuration so the focused style check accepts every Python file touched by the umbrella PR. --- .github/dashboard/ingest/test_ingest.py | 5 +---- python/flydsl/autotune.py | 6 +----- scripts/benchmark_log_parser.py | 5 +---- tests/kernels/test_layernorm.py | 6 +----- tests/kernels/test_moe_sorting.py | 11 ++--------- tests/kernels/test_rmsnorm.py | 6 +----- tests/kernels/test_softmax.py | 6 +----- tests/unit/test_benchmark_compare.py | 18 ++++-------------- tests/unit/test_benchmark_log_parser.py | 8 ++------ 9 files changed, 14 insertions(+), 57 deletions(-) diff --git a/.github/dashboard/ingest/test_ingest.py b/.github/dashboard/ingest/test_ingest.py index a99b30064..a974c9d7e 100644 --- a/.github/dashboard/ingest/test_ingest.py +++ b/.github/dashboard/ingest/test_ingest.py @@ -209,10 +209,7 @@ def test_ingest_run_parses_completed_failure_logs(monkeypatch): monkeypatch.setattr( ingest, "gh_text", - lambda path: ( - "op shape dtype TB/s TFLOPS\n" - "softmax 32768x8192 bf16 4.000 -\n" - ), + lambda path: ("op shape dtype TB/s TFLOPS\n" "softmax 32768x8192 bf16 4.000 -\n"), ) run = { "id": 1, diff --git a/python/flydsl/autotune.py b/python/flydsl/autotune.py index a0aa15a48..3176182cf 100644 --- a/python/flydsl/autotune.py +++ b/python/flydsl/autotune.py @@ -283,11 +283,7 @@ def do_bench( raise ValueError("pipelined schedule cannot include prep_fn or cache flushing") flush_device = getattr(stream, "device", "cuda") - flush_buf = ( - torch.empty(flush_bytes, dtype=torch.uint8, device=flush_device) - if flush_bytes - else None - ) + flush_buf = torch.empty(flush_bytes, dtype=torch.uint8, device=flush_device) if flush_bytes else None stream_context = torch.cuda.stream(stream) if stream is not None else nullcontext() def prepare(): diff --git a/scripts/benchmark_log_parser.py b/scripts/benchmark_log_parser.py index 2549e04d9..4331db607 100644 --- a/scripts/benchmark_log_parser.py +++ b/scripts/benchmark_log_parser.py @@ -179,10 +179,7 @@ def main() -> int: ) emitted = True if not emitted: - print( - f"{args.op_prefix}_atomic\t{args.shape}\t-\t-\t-\t-\treported\t" - "-\t-\treported\tunknown\tunknown" - ) + print(f"{args.op_prefix}_atomic\t{args.shape}\t-\t-\t-\t-\treported\t" "-\t-\treported\tunknown\tunknown") return 0 diff --git a/tests/kernels/test_layernorm.py b/tests/kernels/test_layernorm.py index 2fb9a0e47..ca2032994 100644 --- a/tests/kernels/test_layernorm.py +++ b/tests/kernels/test_layernorm.py @@ -180,11 +180,7 @@ def kernel_launch(): elem_bytes = 4 if dtype == "f32" else 2 total_bytes = (2 * M * N + 2 * N) * elem_bytes # read input + write output + (gamma+beta) bandwidth_gbs = total_bytes / (avg_us / 1e6) / 1e9 - benchmark_instrument = ( - "device_event" - if int(os.environ.get("FLYDSL_PERFTEST_USE_EVENTS", 0)) - else "torch_profiler" - ) + benchmark_instrument = "device_event" if int(os.environ.get("FLYDSL_PERFTEST_USE_EVENTS", 0)) else "torch_profiler" print(f"Kernel avg time: {avg_ms:.4f} ms via run_perftest (warmup={WARMUP_ITERS}, iters={BENCH_ITERS})") print( diff --git a/tests/kernels/test_moe_sorting.py b/tests/kernels/test_moe_sorting.py index 0249e38b2..1428d80b5 100644 --- a/tests/kernels/test_moe_sorting.py +++ b/tests/kernels/test_moe_sorting.py @@ -958,11 +958,7 @@ def bench_profiled_device_us(fn, warmup=BENCH_WARMUP, iters=BENCH_MEASURE): def _make_moe_graph_verifier(outputs_fn, unit_size, extra_outputs_fn=None): """Verify every meaningful MoE sorting output after graph replay.""" references = tuple(output.clone() for output in outputs_fn()) - extra_references = ( - tuple(output.clone() for output in extra_outputs_fn()) - if extra_outputs_fn is not None - else () - ) + extra_references = tuple(output.clone() for output in extra_outputs_fn()) if extra_outputs_fn is not None else () num_padded = int(references[3][0].item()) num_valid_blocks = num_padded // unit_size @@ -982,10 +978,7 @@ def verify(replay): and torch.equal(ids[:num_padded], ref_ids[:num_padded]) and torch.equal(weights[:num_padded], ref_weights[:num_padded]) and torch.equal(expert_ids[:num_valid_blocks], ref_expert_ids[:num_valid_blocks]) - and all( - torch.equal(output, reference) - for output, reference in zip(extra_outputs, extra_references) - ) + and all(torch.equal(output, reference) for output, reference in zip(extra_outputs, extra_references)) ) return verify diff --git a/tests/kernels/test_rmsnorm.py b/tests/kernels/test_rmsnorm.py index eec5989f3..2da851e06 100644 --- a/tests/kernels/test_rmsnorm.py +++ b/tests/kernels/test_rmsnorm.py @@ -218,11 +218,7 @@ def kernel_launch(): weight_elem_bytes = 4 if weight_dtype == "f32" else 2 total_bytes = 2 * M * N * elem_bytes + N * weight_elem_bytes bandwidth_gbs = total_bytes / (avg_us / 1e6) / 1e9 - benchmark_instrument = ( - "device_event" - if int(os.environ.get("FLYDSL_PERFTEST_USE_EVENTS", 0)) - else "torch_profiler" - ) + benchmark_instrument = "device_event" if int(os.environ.get("FLYDSL_PERFTEST_USE_EVENTS", 0)) else "torch_profiler" print(f"Kernel avg time: {avg_ms:.4f} ms via run_perftest (warmup={WARMUP_ITERS}, iters={BENCH_ITERS})") print( diff --git a/tests/kernels/test_softmax.py b/tests/kernels/test_softmax.py index 0641fa7d7..571464555 100644 --- a/tests/kernels/test_softmax.py +++ b/tests/kernels/test_softmax.py @@ -95,11 +95,7 @@ def kernel_launch(): avg_ms = avg_us / 1000.0 total_bytes = 2 * M * N * (4 if dtype_str == "f32" else 2) # read input + write output bandwidth_gbs = total_bytes / (avg_us / 1e6) / 1e9 - benchmark_instrument = ( - "device_event" - if int(os.environ.get("FLYDSL_PERFTEST_USE_EVENTS", 0)) - else "torch_profiler" - ) + benchmark_instrument = "device_event" if int(os.environ.get("FLYDSL_PERFTEST_USE_EVENTS", 0)) else "torch_profiler" print(f"Kernel avg time: {avg_ms:.4f} ms via run_perftest (warmup={WARMUP_ITERS}, iters={BENCH_ITERS})") print( f"Benchmark contract: instrument={benchmark_instrument} schedule=per_iter_sync " diff --git a/tests/unit/test_benchmark_compare.py b/tests/unit/test_benchmark_compare.py index 248a4724a..2496b9fb9 100644 --- a/tests/unit/test_benchmark_compare.py +++ b/tests/unit/test_benchmark_compare.py @@ -175,10 +175,7 @@ def test_compare_benchmark_fails_when_allowlisted_row_disappears(tmp_path, monke current = tmp_path / "current.csv" thresholds = tmp_path / "thresholds.json" _write_csv(baseline, avg_us="100") - current.write_text( - "op,shape,dtype,tbps,tflops,status,avg_us,arch\n" - "gemm,1x1x1,bf16,-,1.0,ok,10,gfx950\n" - ) + current.write_text("op,shape,dtype,tbps,tflops,status,avg_us,arch\n" "gemm,1x1x1,bf16,-,1.0,ok,10,gfx950\n") thresholds.write_text( json.dumps( { @@ -335,10 +332,7 @@ def test_baseline_only_row_on_another_arch_is_not_gated(tmp_path, monkeypatch): current = tmp_path / "current.csv" thresholds = tmp_path / "thresholds.json" _write_csv(baseline, avg_us="100", arch="gfx942") - current.write_text( - "op,shape,dtype,tbps,tflops,status,avg_us,arch\n" - "gemm,1x1x1,bf16,-,1.0,ok,10,gfx950\n" - ) + current.write_text("op,shape,dtype,tbps,tflops,status,avg_us,arch\n" "gemm,1x1x1,bf16,-,1.0,ok,10,gfx950\n") thresholds.write_text( json.dumps( { @@ -440,9 +434,7 @@ def test_csv_arch_cannot_override_trusted_runner_arch(tmp_path, monkeypatch): thresholds = tmp_path / "thresholds.json" _write_csv(baseline, avg_us="100", arch="gfx950") _write_csv(current, avg_us="100", arch="gfx1201") - thresholds.write_text( - json.dumps({"version": 1, "architectures": {"gfx950": [], "gfx1201": []}}) - ) + thresholds.write_text(json.dumps({"version": 1, "architectures": {"gfx950": [], "gfx1201": []}})) monkeypatch.setattr( sys, "argv", @@ -466,9 +458,7 @@ def test_baseline_arch_must_match_trusted_runner_arch(tmp_path, monkeypatch): thresholds = tmp_path / "thresholds.json" _write_csv(baseline, avg_us="100", arch="gfx1201") _write_csv(current, avg_us="100", arch="gfx950") - thresholds.write_text( - json.dumps({"version": 1, "architectures": {"gfx950": [], "gfx1201": []}}) - ) + thresholds.write_text(json.dumps({"version": 1, "architectures": {"gfx950": [], "gfx1201": []}})) monkeypatch.setattr( sys, "argv", diff --git a/tests/unit/test_benchmark_log_parser.py b/tests/unit/test_benchmark_log_parser.py index cc47257bf..026438a1f 100644 --- a/tests/unit/test_benchmark_log_parser.py +++ b/tests/unit/test_benchmark_log_parser.py @@ -61,8 +61,7 @@ def test_parse_metrics_normalizes_raw_microseconds( def test_softmax_style_keeps_first_base_measurement(): metrics = parse_metrics( - "Kernel avg time: 0.1000 ms\nBandwidth: 4000 GB/s\n" - "Kernel avg time: 0.5000 ms\nBandwidth: 1000 GB/s\n" + "Kernel avg time: 0.1000 ms\nBandwidth: 4000 GB/s\n" "Kernel avg time: 0.5000 ms\nBandwidth: 1000 GB/s\n" ) assert metrics.avg_us == 100.0 assert metrics.tbps == 4.0 @@ -98,10 +97,7 @@ def test_legacy_five_column_output_converts_to_enriched_csv(tmp_path, monkeypatc source = tmp_path / "benchmark.out" destination = tmp_path / "benchmark.csv" long_shape = "5120x5120x8320_tile128x256x128_2tg" - source.write_text( - "op shape dtype TB/s TFLOPS\n" - f"gemm {long_shape} bf16 4.000 -\n" - ) + source.write_text("op shape dtype TB/s TFLOPS\n" f"gemm {long_shape} bf16 4.000 -\n") monkeypatch.setattr( sys, "argv",