diff --git a/src/hyperloom/agents/kernel/tools/_bypass_report.py b/src/hyperloom/agents/kernel/tools/_bypass_report.py index 18fd0d9cd2..56cadd7d13 100644 --- a/src/hyperloom/agents/kernel/tools/_bypass_report.py +++ b/src/hyperloom/agents/kernel/tools/_bypass_report.py @@ -536,11 +536,23 @@ def build_candidates( # it is not silently presented as ready to optimize. if kc.reusable and source_file and not cand["shape_dispatchable"]: cand["skip_reason"] = f"shape not dispatchable (provenance={shape_provenance}); need operand dims" - # Optimization ROI = GPU-time share x headroom (1 - efficiency); with no - # analytical efficiency, headroom=1 so it degrades to gpu_pct. - eff = cand.get("efficiency_percent") - eff = float(eff) if isinstance(eff, (int, float)) else 0.0 - headroom = 1.0 - min(max(eff, 0.0), 100.0) / 100.0 + # Optimization ROI = GPU-time share x headroom (1 - roofline attainment). + # Headroom must come from the BINDING side: ``efficiency_percent`` is + # compute-side and reads ~0 for a memory-bound kernel, so using it would + # award full headroom to a kernel already pinned at its bandwidth roof -- + # exactly the kernel with nothing to recover. ``roofline_attainment_pct`` + # already selects compute-vs-bandwidth util by ``bound_type``. With no + # analytical roofline, headroom=1 so this degrades to gpu_pct. + # A capped estimate is a modelling failure, not a saturated kernel: the + # FLOP/byte closed form overshot the roof and was clamped to 100%. + # Scoring it would read "no headroom" off a number that only means the + # formula does not fit this kernel -- e.g. a MoE grouped GEMM charged + # for all experts by the dense/elementwise form when topk of them run. + # Fall back to the raw share so the kernel keeps competing. + attainment = cand.get("roofline_attainment_pct") + if cand.get("roofline_estimate_capped") or not isinstance(attainment, (int, float)) or isinstance(attainment, bool): + attainment = 0.0 + headroom = 1.0 - min(max(float(attainment), 0.0), 100.0) / 100.0 cand["optimization_priority"] = round(float(cand.get("gpu_pct") or 0.0) * headroom, 4) # Deterministic per-kernel hint for the specialist prompt's action slot. suggestion = _build_suggestion(kc.category, str(cand.get("bound_type") or "")) diff --git a/src/hyperloom/inference_optimizer/breakdown/collectors/telemetry.py b/src/hyperloom/inference_optimizer/breakdown/collectors/telemetry.py index f43836104c..a6e5404705 100644 --- a/src/hyperloom/inference_optimizer/breakdown/collectors/telemetry.py +++ b/src/hyperloom/inference_optimizer/breakdown/collectors/telemetry.py @@ -233,6 +233,17 @@ def _max(key: str) -> float: vals = [v for v in vals if v is not None] return round(max(vals), 2) if vals else 0.0 + def _count(key: str) -> int: + """Number of samples carrying a numeric value for *key*. + + Args: + key (str): Sample field name. + + Returns: + int: The count of present values. + """ + return sum(1 for s in samples if _to_float(s.get(key)) is not None) + return { "samples": len(samples), "avg_power_w": _avg("power_w") or _avg("power"), @@ -240,6 +251,12 @@ def _max(key: str) -> float: "avg_temp_c": _avg("temperature_c") or _avg("temperature"), "max_temp_c": _max("temperature_c") or _max("temperature"), "avg_clock_mhz": _avg("clock_mhz") or _avg("sclk_mhz"), + "max_clock_mhz": _max("clock_mhz") or _max("sclk_mhz"), + "avg_mclk_mhz": _avg("mclk_mhz"), + # Distinct from ``samples``: only a subset of samples may carry clocks + # (older samplers omitted ``--showclocks``), and the effective-frequency + # roofline derate keys off this count. + "clock_samples": _count("clock_mhz") or _count("sclk_mhz"), } diff --git a/src/hyperloom/inference_optimizer/breakdown/schema.py b/src/hyperloom/inference_optimizer/breakdown/schema.py index 289ca71eba..f00e7381a0 100644 --- a/src/hyperloom/inference_optimizer/breakdown/schema.py +++ b/src/hyperloom/inference_optimizer/breakdown/schema.py @@ -1110,7 +1110,13 @@ class GpuMonitorAggregate(TypedDict, total=False): max_power_w (float): Peak power draw (watts). avg_temp_c (float): Average temperature (Celsius). max_temp_c (float): Peak temperature (Celsius). - avg_clock_mhz (float): Average clock frequency (MHz). + avg_clock_mhz (float): Average engine (sclk) frequency (MHz). + max_clock_mhz (float): Peak engine (sclk) frequency (MHz). + avg_mclk_mhz (float): Average memory (mclk) frequency (MHz). Recorded + for provenance; MI300-series parts expose a single mclk DPM state, + so this is expected to be constant. + clock_samples (int): Samples that carried an engine clock. Lower than + ``samples`` when the sampler predates ``--showclocks``. """ samples: int @@ -1119,6 +1125,9 @@ class GpuMonitorAggregate(TypedDict, total=False): avg_temp_c: float max_temp_c: float avg_clock_mhz: float + max_clock_mhz: float + avg_mclk_mhz: float + clock_samples: int class LaneTimelineEntry(TypedDict, total=False): diff --git a/src/hyperloom/inference_optimizer/multi_node/scripts/launch_infera_node.py b/src/hyperloom/inference_optimizer/multi_node/scripts/launch_infera_node.py index ae1505a0c9..fb8799b870 100644 --- a/src/hyperloom/inference_optimizer/multi_node/scripts/launch_infera_node.py +++ b/src/hyperloom/inference_optimizer/multi_node/scripts/launch_infera_node.py @@ -766,14 +766,18 @@ def _start_gpu_sampler(out_csv: Path, pid_file: Path, interval_s: int) -> None: q_csv = shlex.quote(str(out_csv)) q_rocm = shlex.quote(rocm) interval = max(1, int(interval_s)) + # ``--showclocks`` carries sclk/mclk, which the roofline uses to anchor the + # compute ceiling to the clock the workload actually sustained instead of + # the vendor boost clock. + query = "--showuse --showmemuse --showpower --showtemp --showclocks --csv" # header once, then loop: prepend epoch ts to each rocm-smi --csv data row. script = ( "set +e; " - f'H="ts,$({q_rocm} --showuse --showmemuse --showpower --showtemp --csv 2>/dev/null | head -1)"; ' + f'H="ts,$({q_rocm} {query} 2>/dev/null | head -1)"; ' f'[ -s {q_csv} ] || echo "$H" > {q_csv}; ' "while true; do " "TS=$(date +%s); " - f"{q_rocm} --showuse --showmemuse --showpower --showtemp --csv 2>/dev/null " + f"{q_rocm} {query} 2>/dev/null " f'| tail -n +2 | sed "s/^/$TS,/" >> {q_csv}; ' f"sleep {interval}; done" ) diff --git a/src/hyperloom/inference_optimizer/tests/conftest.py b/src/hyperloom/inference_optimizer/tests/conftest.py index 379166419d..9e07f27ea3 100644 --- a/src/hyperloom/inference_optimizer/tests/conftest.py +++ b/src/hyperloom/inference_optimizer/tests/conftest.py @@ -32,6 +32,24 @@ def _clear_kernel_request_handler_caches(): krh._default_kernel_batch_parallel.cache_clear() +@pytest.fixture(autouse=True) +def _disable_gpu_hw_probe(request, monkeypatch): + """Keep roofline assertions off real hardware measurements. + + The probe layer feeds measured compute and bandwidth roofs into every + ceiling. On a machine that has run a probe, a cached result would silently + replace the table values these tests assert against, so results would depend + on whether the developer happens to own a GPU. Tests that exercise the probe + itself opt out by name. + """ + from hyperloom.orchestrator.kernel import hw_probe + + if request.node.module.__name__.endswith("test_hw_probe"): + return + monkeypatch.setenv(hw_probe.DISABLE_ENV, "1") + hw_probe.clear_caches() + + def _bootstrap_kernel_agent_env() -> None: """Point HYPERLOOM_KERNEL_AGENT_ROOT at the in-repo kernel-agent checkout.""" if os.environ.get("HYPERLOOM_KERNEL_AGENT_ROOT"): diff --git a/src/hyperloom/inference_optimizer/tests/test_benchmark_result_branches_unit.py b/src/hyperloom/inference_optimizer/tests/test_benchmark_result_branches_unit.py index d574038f92..4c0840838f 100644 --- a/src/hyperloom/inference_optimizer/tests/test_benchmark_result_branches_unit.py +++ b/src/hyperloom/inference_optimizer/tests/test_benchmark_result_branches_unit.py @@ -338,6 +338,62 @@ def test_row_to_gpu_sample_empty_when_no_numeric(): assert br._row_to_gpu_sample(["timestamp", "note"], ["1000", "n/a"]) == {} +# Verbatim ``rocm-smi --showclocks --csv`` layout from an MI355X node: every +# clock arrives as a "(1412Mhz)" speed cell paired with a small-integer DPM +# level cell. +_MI355X_CLOCK_HEADER = [ + "ts", + "device", + "Temperature (Sensor junction) (C)", + "Temperature (Sensor memory) (C)", + "fclk clock speed:", + "fclk clock level:", + "mclk clock speed:", + "mclk clock level:", + "sclk clock speed:", + "sclk clock level:", + "socclk clock speed:", + "socclk clock level:", + "Current Socket Graphics Package Power (W)", + "GPU use (%)", + "GPU Memory Allocated (VRAM%)", +] +_MI355X_CLOCK_ROW = [ + "1700000000", + "card0", + "42.0", + "25.0", + "(1250Mhz)", + "0", + "(2000Mhz)", + "0", + "(1412Mhz)", + "1", + "(38Mhz)", + "S", + "254.0", + "0", + "92", +] + + +def test_row_to_gpu_sample_reads_showclocks_speeds_not_dpm_levels(): + s = br._row_to_gpu_sample(_MI355X_CLOCK_HEADER, _MI355X_CLOCK_ROW) + # The neighbouring "level" cell is a small integer that would otherwise be + # recorded as a frequency and silently derate the roofline to nothing. + assert s["clock_mhz"] == 1412 + assert s["mclk_mhz"] == 2000 + assert s["temperature_c"] == 42.0 + assert s["power_w"] == 254.0 + + +def test_row_to_gpu_sample_omits_mclk_when_not_sampled(): + header = ["timestamp", "Average Socket Power (W)", "sclk clock (MHz)"] + s = br._row_to_gpu_sample(header, ["1000", "310.5", "1400"]) + assert s["clock_mhz"] == 1400 + assert "mclk_mhz" not in s + + # ---- _aggregate_gpu_samples_by_role --------------------------------------- def test_aggregate_gpu_samples_by_role(): samples = [ diff --git a/src/hyperloom/inference_optimizer/tests/test_hw_probe.py b/src/hyperloom/inference_optimizer/tests/test_hw_probe.py new file mode 100644 index 0000000000..cfd6ab78c5 --- /dev/null +++ b/src/hyperloom/inference_optimizer/tests/test_hw_probe.py @@ -0,0 +1,478 @@ +# SPDX-FileCopyrightText: 2026 Advanced Micro Devices, Inc. +# SPDX-License-Identifier: MIT + +"""Tests for ``orchestrator.kernel.hw_probe``. + +The probe layer replaces hand-maintained per-SKU tables with on-node +measurement, so what matters here is that it either produces a *correct* number +or produces nothing at all. These tests cover the reduction of raw probe output +into per-precision rates, the guards that stop a bad measurement from inflating +a ceiling, the cache round-trip, and the promise that every unsupported or +broken path returns ``None`` so callers fall back through the existing chain. + +No GPU is required: the subprocess boundary is stubbed. A live counterpart runs +only when ``HYPERLOOM_GPU_PROBE_RUN_LIVE=1``. +""" + +from __future__ import annotations + +import json +import os +import subprocess + +import pytest + +from hyperloom.orchestrator.kernel import hw_probe +from hyperloom.orchestrator.kernel.hw_probe import ( + DISABLE_ENV, + DeviceInfo, + MfmaRate, + ProbeResult, + load_cached, + normalize_arch, + probe_compute_peak_tflops, + probe_hbm_bandwidth_gb_per_sec, + rocm_version, +) + +#: Verbatim stdout from the gfx950 matrix-core probe. Both bf16 variants and +#: both fp8 variants are present, which is the case the max-wins reduction +#: exists to handle. +_MFMA_STDOUT = ( + '{"kind":"device","arch":"gfx950:sramecc+:xnack-","cus":256,"boost_mhz":2400}\n' + '{"kind":"mfma","precision":"bf16","variant":"mfma_f32_16x16x32_bf16",' + '"flops_per_sec":2.337600e+15}\n' + '{"kind":"mfma","precision":"bf16","variant":"mfma_f32_16x16x16bf16_1k",' + '"flops_per_sec":1.212149e+15}\n' + '{"kind":"mfma","precision":"fp8","variant":"mfma_f32_16x16x32_fp8_fp8",' + '"flops_per_sec":2.423697e+15}\n' + '{"kind":"mfma","precision":"fp8","variant":"mfma_scale_f32_16x16x128_f8f6f4[e4m3]",' + '"flops_per_sec":5.018500e+15}\n' +) + + +@pytest.fixture(autouse=True) +def _clear_probe_env(monkeypatch): + """Keep probe env overrides and memoized lookups out of assertions.""" + monkeypatch.delenv(DISABLE_ENV, raising=False) + monkeypatch.delenv(hw_probe.TIMEOUT_ENV, raising=False) + hw_probe.clear_caches() + yield + hw_probe.clear_caches() + + +@pytest.fixture +def _isolated_cache(monkeypatch, tmp_path): + """Point the probe cache at a scratch directory.""" + monkeypatch.setenv("HYPERLOOM_CACHE_DIR", str(tmp_path)) + return tmp_path + + +def _result(**overrides) -> ProbeResult: + """Build a probe result with sensible gfx950 defaults. + + Args: + **overrides: Fields to replace. + + Returns: + A ``ProbeResult`` for use in assertions. + """ + base = { + "schema": hw_probe._SCHEMA_VERSION, + "device": DeviceInfo(arch="gfx950", cu_count=256, boost_sclk_mhz=2400.0), + "rocm_version": "7.2.4", + "mfma_rates": { + "bf16": MfmaRate("bf16", "mfma_f32_16x16x32_bf16", 3805.0), + "fp8": MfmaRate("fp8", "mfma_scale_f32_16x16x128_f8f6f4[e4m3]", 8168.0), + }, + "bandwidth_gb_per_sec": {1: 7133.2, 8: 7123.9}, + "probe_sclk_mhz": 2400.0, + "probed_at": 1.0, + } + base.update(overrides) + return ProbeResult(**base) + + +class TestArchNormalization: + """Target-feature suffixes must not fragment the cache key.""" + + @pytest.mark.parametrize( + ("raw", "expected"), + [ + ("gfx950:sramecc+:xnack-", "gfx950"), + ("gfx942", "gfx942"), + (" GFX950 ", "gfx950"), + ("", ""), + (None, ""), + ], + ) + def test_features_are_stripped(self, raw, expected) -> None: + assert normalize_arch(raw) == expected + + +class TestMfmaReduction: + """Reducing raw probe output to one rate per precision.""" + + @pytest.fixture + def _stub_probe(self, monkeypatch): + """Return the recorded gfx950 stdout and a 2400 MHz clock sample.""" + monkeypatch.setattr(hw_probe, "_run", lambda *a, **k: _MFMA_STDOUT) + monkeypatch.setattr(hw_probe, "_sample_sclk_once", lambda: 2400.0) + + def test_fastest_variant_wins_per_precision(self, _stub_probe) -> None: + """A precision offering several opcodes is scored by its fastest. + + On gfx950 fp8 is reachable through both a 16x16x32 opcode and the + ``f8f6f4`` path, and only the latter carries the doubled rate. Taking + the first match instead of the maximum would understate fp8 by 2x. + """ + device, rates, sclk = hw_probe._probe_mfma(hw_probe.Path("/probe")) + + assert device == DeviceInfo(arch="gfx950", cu_count=256, boost_sclk_mhz=2400.0) + assert sclk == 2400.0 + assert rates["bf16"].variant == "mfma_f32_16x16x32_bf16" + assert rates["fp8"].variant == "mfma_scale_f32_16x16x128_f8f6f4[e4m3]" + + def test_rates_land_near_the_architectural_rate(self, _stub_probe) -> None: + """The measurement must reproduce the documented ISA rate. + + This is the whole premise of retiring the tables, so it is asserted + against the published gfx950 figures rather than against itself. + """ + _, rates, _ = hw_probe._probe_mfma(hw_probe.Path("/probe")) + + assert rates["bf16"].flops_per_clk_per_cu == pytest.approx(4096.0, rel=0.10) + assert rates["fp8"].flops_per_clk_per_cu == pytest.approx(8192.0, rel=0.10) + + def test_empty_output_yields_no_rates(self, monkeypatch) -> None: + monkeypatch.setattr(hw_probe, "_run", lambda *a, **k: "") + assert hw_probe._probe_mfma(hw_probe.Path("/probe")) == (None, {}, 0.0) + + def test_device_line_without_cus_is_rejected(self, monkeypatch) -> None: + """A device report with no CU count cannot produce a per-CU rate.""" + monkeypatch.setattr( + hw_probe, + "_run", + lambda *a, **k: '{"kind":"device","arch":"gfx950","cus":0,"boost_mhz":2400}\n', + ) + assert hw_probe._probe_mfma(hw_probe.Path("/probe")) == (None, {}, 0.0) + + +class TestClockSampleGuard: + """Grossly implausible probe clocks are discarded in favour of boost. + + The sampled clock is the divisor, so reading it low scales every rate -- + and therefore the ceiling -- upward, which is the one direction this module + must not fail in. The guard only catches *gross* contamination such as idle + ticks; distinguishing the engine clock from the memory clock is the parser's + job (see :class:`TestSclkParsing`), because at 2000 MHz against a 2400 MHz + boost the memory clock is numerically indistinguishable from an ordinary + throttled engine clock. + """ + + def test_idle_level_sample_is_rejected_in_favour_of_boost(self, monkeypatch) -> None: + """An idle 95 MHz tick cannot describe a compute-saturating probe.""" + monkeypatch.setattr(hw_probe, "_run", lambda *a, **k: _MFMA_STDOUT) + monkeypatch.setattr(hw_probe, "_sample_sclk_once", lambda: 95.0) + + _, rates, sclk = hw_probe._probe_mfma(hw_probe.Path("/probe")) + + assert sclk == 2400.0 + assert rates["bf16"].flops_per_clk_per_cu < 4096.0 + + def test_missing_samples_fall_back_to_boost(self, monkeypatch) -> None: + monkeypatch.setattr(hw_probe, "_run", lambda *a, **k: _MFMA_STDOUT) + monkeypatch.setattr(hw_probe, "_sample_sclk_once", lambda: 0.0) + + _, _, sclk = hw_probe._probe_mfma(hw_probe.Path("/probe")) + + assert sclk == 2400.0 + + @pytest.mark.parametrize("sampled", [2200.0, 2000.0, 1300.0]) + def test_a_genuinely_throttled_clock_is_honoured(self, monkeypatch, sampled) -> None: + """A plausible sub-boost clock is real and must be used as-is. + + Substituting boost here would understate the rate and, once the roof is + rebuilt at the workload's own clock, understate the ceiling. + """ + monkeypatch.setattr(hw_probe, "_run", lambda *a, **k: _MFMA_STDOUT) + monkeypatch.setattr(hw_probe, "_sample_sclk_once", lambda: sampled) + + _, _, sclk = hw_probe._probe_mfma(hw_probe.Path("/probe")) + + assert sclk == sampled + + +class TestSclkParsing: + """Engine clock is read by column name, never by scanning the row.""" + + def _stub_csv(self, monkeypatch, stdout: str) -> None: + """Route ``rocm-smi`` through a canned CSV response.""" + monkeypatch.setattr(hw_probe.shutil, "which", lambda _: "/usr/bin/rocm-smi") + monkeypatch.setattr( + hw_probe.subprocess, + "run", + lambda *a, **k: subprocess.CompletedProcess(a[0], 0, stdout, ""), + ) + + def test_engine_clock_is_read_not_memory_clock(self, monkeypatch) -> None: + """mclk sits in the same numeric range and must not be picked up.""" + self._stub_csv( + monkeypatch, + "device,fclk clock speed:,mclk clock speed:,sclk clock speed:\n" + "card0,(1250Mhz),(2000Mhz),(1413Mhz)\n", + ) + assert hw_probe._sample_sclk_once() == 1413.0 + + def test_loaded_card_wins_over_idle_peers(self, monkeypatch) -> None: + """One GPU under probe among idle peers must report the probe's clock.""" + self._stub_csv( + monkeypatch, + "device,mclk clock speed:,sclk clock speed:\n" + "card0,(2000Mhz),(2394Mhz)\n" + "card1,(2000Mhz),(95Mhz)\n" + "card2,(2000Mhz),(95Mhz)\n", + ) + assert hw_probe._sample_sclk_once() == 2394.0 + + def test_missing_tool_reports_nothing(self, monkeypatch) -> None: + monkeypatch.setattr(hw_probe.shutil, "which", lambda _: None) + assert hw_probe._sample_sclk_once() == 0.0 + + +class TestCacheRoundTrip: + """Cached results must survive JSON and be keyed by arch and toolchain.""" + + def test_write_then_read_preserves_the_payload(self, _isolated_cache) -> None: + original = _result() + hw_probe._write_cache(original) + + restored = load_cached(arch="gfx950", rocm="7.2.4") + + assert restored is not None + assert restored.device == original.device + assert restored.mfma_rates["fp8"].variant == original.mfma_rates["fp8"].variant + # JSON object keys are strings; the active-GPU keys must come back as ints. + assert restored.bandwidth_gb_per_sec == {1: 7133.2, 8: 7123.9} + + def test_a_different_toolchain_is_a_cache_miss(self, _isolated_cache) -> None: + """A ROCm upgrade can move the numbers, so it must not reuse the old ones.""" + hw_probe._write_cache(_result()) + assert load_cached(arch="gfx950", rocm="6.4.0") is None + + def test_a_stale_schema_is_ignored(self, _isolated_cache) -> None: + hw_probe._write_cache(_result()) + path = next(_isolated_cache.glob("gpu_probes/*.json")) + payload = json.loads(path.read_text(encoding="utf-8")) + payload["schema"] = hw_probe._SCHEMA_VERSION + 1 + path.write_text(json.dumps(payload), encoding="utf-8") + + assert load_cached(arch="gfx950", rocm="7.2.4") is None + + def test_corrupt_cache_does_not_raise(self, _isolated_cache) -> None: + path = _isolated_cache / "gpu_probes" / "probe-gfx950-rocm7.2.4.json" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text("{not json", encoding="utf-8") + + assert load_cached(arch="gfx950", rocm="7.2.4") is None + + def test_missing_cache_is_a_miss_not_an_error(self, _isolated_cache) -> None: + assert load_cached(arch="gfx950", rocm="7.2.4") is None + + +class TestComputeRoof: + """Turning a measured rate into a ceiling.""" + + def test_roof_is_rate_times_cus_times_clock(self) -> None: + result = _result() + + roof = probe_compute_peak_tflops("bf16", sustained_sclk_mhz=2400.0, result=result) + + assert roof == pytest.approx(3805.0 * 256 * 2400e6 / 1e12) + + def test_roof_stays_below_the_vendor_peak(self) -> None: + """A measured roof above the vendor dense peak would be nonsense. + + The vendor peak is the architectural rate at boost, so a probe result + that exceeded it would mean the measurement, not the hardware, is wrong. + """ + roof = probe_compute_peak_tflops("bf16", sustained_sclk_mhz=2400.0, result=_result()) + + assert roof < 2516.6 + + def test_sustained_clock_scales_the_roof(self) -> None: + """Compute throughput is linear in engine clock.""" + at_boost = probe_compute_peak_tflops("bf16", sustained_sclk_mhz=2400.0, result=_result()) + throttled = probe_compute_peak_tflops("bf16", sustained_sclk_mhz=1200.0, result=_result()) + + assert throttled == pytest.approx(at_boost / 2.0) + + def test_absent_clock_falls_back_to_boost(self) -> None: + assert probe_compute_peak_tflops("bf16", result=_result()) == pytest.approx( + probe_compute_peak_tflops("bf16", sustained_sclk_mhz=2400.0, result=_result()) + ) + + def test_unprobed_precision_returns_none(self) -> None: + """fp4 was never probed here, so the caller must fall back.""" + assert probe_compute_peak_tflops("fp4", result=_result()) is None + + def test_unknown_precision_returns_none(self) -> None: + assert probe_compute_peak_tflops("int3", result=_result()) is None + + +class TestBandwidthLookup: + """Selecting a bandwidth measurement for a given load.""" + + def test_closest_active_gpu_count_is_used(self) -> None: + result = _result() + assert probe_hbm_bandwidth_gb_per_sec(active_gpus=1, result=result) == 7133.2 + assert probe_hbm_bandwidth_gb_per_sec(active_gpus=8, result=result) == 7123.9 + assert probe_hbm_bandwidth_gb_per_sec(active_gpus=7, result=result) == 7123.9 + + def test_per_gpu_bandwidth_is_flat_across_load_on_gfx950(self) -> None: + """Per-GPU HBM is private, so peers streaming must not change it. + + Locks in a measurement that contradicted an earlier unsynchronized + attempt: with all eight cards at 100% utilization the per-GPU figure is + unchanged, so any future regression toward a load-dependent number is a + measurement bug rather than a discovery. + """ + result = _result() + solo = probe_hbm_bandwidth_gb_per_sec(active_gpus=1, result=result) + loaded = probe_hbm_bandwidth_gb_per_sec(active_gpus=8, result=result) + + assert loaded == pytest.approx(solo, rel=0.02) + + def test_no_bandwidth_measurement_returns_none(self) -> None: + assert probe_hbm_bandwidth_gb_per_sec(result=_result(bandwidth_gb_per_sec={})) is None + + +class TestProbeAppliesOnlyToTheProbedPart: + """A probe describes the local device and nothing else. + + Rooflines are routinely computed for a GPU other than the one running the + code -- comparing parts, or replaying a recorded session from a different + node. Caught in real-workload replay: an MI300X session evaluated on a + gfx950 host picked up the local 7133 GB/s measurement in place of MI300X's + 5300 GB/s vendor peak and lifted that ceiling by 35%. Substituting one + part's hardware for another's is worse than not measuring, because it moves + the ceiling instead of falling back. + """ + + @pytest.fixture + def _gfx950_cache(self, _isolated_cache, monkeypatch): + """A cached gfx950 probe, as a gfx950 host would have.""" + hw_probe._write_cache(_result()) + monkeypatch.setattr(hw_probe, "detect_arch", lambda: "gfx950") + return _isolated_cache + + def test_foreign_part_does_not_borrow_the_local_probe(self, _gfx950_cache) -> None: + assert probe_hbm_bandwidth_gb_per_sec(gpu_type="mi300x") is None + assert probe_compute_peak_tflops("bf16", gpu_type="mi300x") is None + + def test_matching_part_uses_the_probe(self, _gfx950_cache) -> None: + assert probe_hbm_bandwidth_gb_per_sec(gpu_type="mi355x") == 7133.2 + assert probe_compute_peak_tflops("bf16", gpu_type="mi355x") is not None + + @pytest.mark.parametrize("gpu_type", ["mi300x", "mi308x", "mi325x"]) + def test_every_gfx942_part_is_excluded(self, _gfx950_cache, gpu_type) -> None: + """MI300X, MI308X and MI325X are all gfx942, none of them gfx950.""" + assert probe_hbm_bandwidth_gb_per_sec(gpu_type=gpu_type) is None + + def test_unknown_gpu_type_does_not_block_the_probe(self, _gfx950_cache) -> None: + """An unmappable key cannot contradict the cache, so it is not a veto.""" + assert probe_hbm_bandwidth_gb_per_sec(gpu_type="some-future-part") == 7133.2 + + @pytest.mark.parametrize( + ("gpu_type", "expected"), + [("mi300x", "gfx942"), ("mi355x", "gfx950"), ("nonsense", ""), (None, "")], + ) + def test_gpu_type_maps_to_architecture(self, gpu_type, expected) -> None: + assert hw_probe.expected_arch_for(gpu_type) == expected + + +class TestFailsSoft: + """Every unsupported path returns ``None`` so callers keep their fallback.""" + + def test_disable_flag_suppresses_reads(self, monkeypatch, _isolated_cache) -> None: + hw_probe._write_cache(_result()) + monkeypatch.setenv(DISABLE_ENV, "1") + + assert load_cached(arch="gfx950", rocm="7.2.4") is None + assert hw_probe.probe_and_cache() is None + + def test_missing_hipcc_skips_compilation(self, monkeypatch, tmp_path) -> None: + monkeypatch.setattr(hw_probe.shutil, "which", lambda _: None) + + assert hw_probe._compile("int main(){}", "probe", "gfx950", tmp_path) is None + + def test_compile_failure_returns_none(self, monkeypatch, tmp_path) -> None: + monkeypatch.setattr(hw_probe.shutil, "which", lambda _: "/usr/bin/hipcc") + monkeypatch.setattr( + hw_probe.subprocess, + "run", + lambda *a, **k: subprocess.CompletedProcess(a[0], 1, "", "error: no such target"), + ) + + assert hw_probe._compile("int main(){}", "probe", "gfx950", tmp_path) is None + + def test_undetectable_arch_skips_probing(self, monkeypatch) -> None: + monkeypatch.setattr(hw_probe, "detect_arch", lambda: "") + assert hw_probe.probe_and_cache() is None + + def test_run_timeout_is_swallowed(self, monkeypatch, tmp_path) -> None: + def _timeout(*a, **k): + raise subprocess.TimeoutExpired(cmd="probe", timeout=1.0) + + monkeypatch.setattr(hw_probe.subprocess, "run", _timeout) + + assert hw_probe._run(tmp_path / "probe", []) == "" + + def test_readers_tolerate_a_totally_absent_cache(self, monkeypatch) -> None: + monkeypatch.setattr(hw_probe, "load_cached", lambda **k: None) + assert probe_compute_peak_tflops("bf16") is None + assert probe_hbm_bandwidth_gb_per_sec() is None + + +class TestEnvironment: + """Environment knobs.""" + + def test_timeout_override_is_applied(self, monkeypatch) -> None: + monkeypatch.setenv(hw_probe.TIMEOUT_ENV, "42") + assert hw_probe._timeout_sec() == 42.0 + + @pytest.mark.parametrize("raw", ["", "abc", "0", "-5"]) + def test_bad_timeout_falls_back_to_default(self, monkeypatch, raw) -> None: + monkeypatch.setenv(hw_probe.TIMEOUT_ENV, raw) + assert hw_probe._timeout_sec() == hw_probe._DEFAULT_TIMEOUT_SEC + + +@pytest.mark.skipif( + os.environ.get("HYPERLOOM_GPU_PROBE_RUN_LIVE") != "1", + reason="live GPU probe disabled; set HYPERLOOM_GPU_PROBE_RUN_LIVE=1 to enable", +) +class TestLiveProbe: + """Runs the real probes against real hardware.""" + + def test_probe_recovers_documented_rates(self, _isolated_cache) -> None: + """Measured rates must land near the published architectural figures.""" + documented = {"bf16": 4096.0, "fp16": 4096.0, "fp8": 8192.0, "fp4": 16384.0} + + result = hw_probe.probe_and_cache(force=True) + + assert result is not None, "probe failed on a node that should support it" + assert result.device.cu_count > 0 + for precision, rate in result.mfma_rates.items(): + expected = documented.get(precision) + if expected is None: + continue + assert rate.flops_per_clk_per_cu == pytest.approx(expected, rel=0.15) + + def test_probe_result_is_cached_and_reused(self, _isolated_cache) -> None: + first = hw_probe.probe_and_cache(force=True) + assert first is not None + + reused = load_cached() + + assert reused is not None + assert reused.mfma_rates.keys() == first.mfma_rates.keys() diff --git a/src/hyperloom/inference_optimizer/tests/test_kernel_roi_ranking_unit.py b/src/hyperloom/inference_optimizer/tests/test_kernel_roi_ranking_unit.py new file mode 100644 index 0000000000..da1f486fb9 --- /dev/null +++ b/src/hyperloom/inference_optimizer/tests/test_kernel_roi_ranking_unit.py @@ -0,0 +1,226 @@ +# SPDX-FileCopyrightText: 2026 Advanced Micro Devices, Inc. +# SPDX-License-Identifier: MIT + +"""Tests for headroom-weighted ranking of kernel-opt candidates. + +Ranking on ``gpu_pct`` alone spends the attempt budget on whatever owns the +most trace time, including kernels already at their roofline with nothing left +to give. These tests pin the ROI ordering (share x headroom), the floor that +must stay on the raw share, and the degrade-to-``gpu_pct`` path for traces that +carry no roofline at all. + +Headroom is read from ``roofline_attainment_pct`` (the binding side), never +from the compute-side ``efficiency_percent`` alone: a memory-bound kernel +pinned at its bandwidth roof reports ~0 there, and scoring it on that axis +would rank the one kernel with nothing to recover at the very top. +""" + +from __future__ import annotations + +from types import SimpleNamespace + +import pytest + +from hyperloom.orchestrator.kernel import _kernel_decisions as kd + + +def _state(hot): + return SimpleNamespace( + last_trace_analyze={"hot_kernels_top15": hot, "task_groups": []}, + optimization_stack=[], + rejected_kernel_ids=[], + kernel_opt_attempts={}, + kernel_opt_task_attempts=None, + ) + + +def _hot(kid, *, name, gpu_pct, attain=None, eff=None, bound=None, roi=None, src="model.py"): + row = { + "kernel_id": kid, + "name": name, + "source_file": src, + "gpu_pct": gpu_pct, + "reusable_native_kernel": True, + } + if attain is not None: + row["roofline_attainment_pct"] = attain + if eff is not None: + row["efficiency_percent"] = eff + if bound is not None: + row["bound_type"] = bound + if roi is not None: + row["optimization_priority"] = roi + return row + + +class TestOptimizationRoi: + """The ROI helper itself.""" + + def test_attainment_scales_the_share(self): + row = {"roofline_attainment_pct": 25.0} + assert kd._kernel_optimization_roi(row, 40.0) == 30.0 + + def test_precomputed_priority_wins_over_recomputation(self): + # The bypass report already published an ROI; recomputing it here would + # fork the definition. + row = {"roofline_attainment_pct": 25.0, "optimization_priority": 3.5} + assert kd._kernel_optimization_roi(row, 40.0) == 3.5 + + def test_missing_roofline_degrades_to_raw_share(self): + # Rows without a roofline must rank exactly as before. + assert kd._kernel_optimization_roi({}, 12.5) == 12.5 + + def test_non_numeric_attainment_degrades_to_raw_share(self): + assert kd._kernel_optimization_roi({"roofline_attainment_pct": "n/a"}, 12.5) == 12.5 + + def test_attainment_is_clamped(self): + # A kernel reported at or above roofline has no headroom, not negative. + assert kd._kernel_optimization_roi({"roofline_attainment_pct": 140.0}, 10.0) == 0.0 + assert kd._kernel_optimization_roi({"roofline_attainment_pct": -20.0}, 10.0) == 10.0 + + def test_bool_is_not_treated_as_a_number(self): + assert kd._kernel_optimization_roi({"roofline_attainment_pct": True}, 9.0) == 9.0 + assert kd._kernel_optimization_roi({"optimization_priority": True}, 9.0) == 9.0 + + +class TestHeadroomIsReadOnTheBindingSide: + """``efficiency_percent`` is compute-side and must not be trusted alone.""" + + def test_saturated_memory_bound_kernel_is_not_given_headroom(self): + # Regression: aiter::add_rmsnorm sits at 100% of its bandwidth roof but + # reports efficiency_percent=0.279 because that number is compute-side. + # Scoring on it would hand a fully saturated kernel ~all its share. + row = { + "efficiency_percent": 0.279, + "roofline_attainment_pct": 100.0, + "bound_type": "memory_bound", + } + assert kd._kernel_optimization_roi(row, 1.7465) == 0.0 + + def test_memory_bound_without_attainment_is_unknown_not_full_headroom(self): + # The TraceLens route carries no attainment. A memory-bound row there + # cannot be scored on the compute axis, so it degrades to the raw share + # rather than being scored on the wrong one. + row = {"efficiency_percent": 0.3, "bound_type": "memory_bound"} + assert kd._kernel_optimization_roi(row, 4.0) == 4.0 + + def test_compute_bound_efficiency_is_the_binding_side(self): + # For a compute-bound kernel the compute-side number IS the attainment, + # so the TraceLens route can still rank it. + row = {"efficiency_percent": 75.0, "bound_type": "compute_bound"} + assert kd._kernel_optimization_roi(row, 8.0) == 2.0 + + def test_attainment_wins_when_both_are_present(self): + row = { + "efficiency_percent": 10.0, + "roofline_attainment_pct": 90.0, + "bound_type": "memory_bound", + } + assert kd._kernel_optimization_roi(row, 10.0) == pytest.approx(1.0) + + def test_efficiency_alone_without_bound_type_is_not_scored(self): + # Without knowing which side binds, the compute-side number could be + # either meaningful or ~0 by construction; refuse to guess. + assert kd._kernel_optimization_roi({"efficiency_percent": 90.0}, 5.0) == 5.0 + + +class TestCappedEstimatesAreRefused: + """A clamped roofline means the model missed, not that the kernel is full.""" + + def test_capped_attainment_falls_back_to_raw_share(self): + # Regression: aiter::moe_cktile2stages_gemm2_ck is a grouped GEMM billed + # for all 128 experts by the elementwise form when only topk=4 run. The + # estimate implied ~68 TB/s, overshot the roof, and was clamped to 100% + # -- which would zero the ROI of a kernel worth 14.6% of GPU time. + row = { + "roofline_attainment_pct": 100.0, + "bound_type": "memory_bound", + "roofline_estimate_capped": True, + } + assert kd._kernel_optimization_roi(row, 14.6232) == 14.6232 + + def test_capped_compute_side_estimate_is_also_refused(self): + row = { + "efficiency_percent": 100.0, + "bound_type": "compute_bound", + "roofline_estimate_capped": True, + } + assert kd._kernel_optimization_roi(row, 9.0) == 9.0 + + def test_uncapped_attainment_is_still_trusted(self): + row = { + "roofline_attainment_pct": 100.0, + "bound_type": "memory_bound", + "roofline_estimate_capped": False, + } + assert kd._kernel_optimization_roi(row, 14.0) == 0.0 + + +class TestRankingPrefersHeadroom: + """Selection order under the top_n cap.""" + + def test_smaller_kernel_with_headroom_outranks_a_saturated_bigger_one(self): + # 12% at 90% attainment has ROI 1.2; 11% at 30% has ROI 7.7. Ranking on + # gpu_pct alone would burn the single attempt on the kernel that cannot + # move. + hot = [ + _hot("k_big", name="gemm_saturated", gpu_pct=12.0, attain=90.0), + _hot("k_headroom", name="gemm_slack", gpu_pct=11.0, attain=30.0), + ] + untried = kd.untried_hot_reusable_kernels(_state(hot), min_gpu_pct=1.0, top_n=1) + assert untried == ["k_headroom"] + + def test_saturated_memory_bound_kernel_loses_to_a_smaller_one(self): + # The bug in trace form: the saturated norm owns more GPU time, but all + # of it is already at the bandwidth roof. + hot = [ + _hot("k_norm", name="add_rmsnorm", gpu_pct=1.75, attain=100.0, eff=0.279, bound="memory_bound"), + _hot("k_rope", name="rotary", gpu_pct=0.87, attain=22.8, eff=0.035, bound="memory_bound"), + ] + untried = kd.untried_hot_reusable_kernels(_state(hot), min_gpu_pct=0.5, top_n=1) + assert untried == ["k_rope"] + + def test_without_roofline_order_is_unchanged(self): + # No roofline anywhere => pure gpu_pct ordering, as before the change. + hot = [ + _hot("k_small", name="a", gpu_pct=11.0), + _hot("k_big", name="b", gpu_pct=12.0), + ] + untried = kd.untried_hot_reusable_kernels(_state(hot), min_gpu_pct=1.0, top_n=1) + assert untried == ["k_big"] + + def test_precomputed_priority_drives_selection(self): + hot = [ + _hot("k_a", name="a", gpu_pct=30.0, roi=0.5), + _hot("k_b", name="b", gpu_pct=5.0, roi=4.0), + ] + untried = kd.untried_hot_reusable_kernels(_state(hot), min_gpu_pct=1.0, top_n=1) + assert untried == ["k_b"] + + def test_mixed_rows_rank_together(self): + # A row with no roofline ranks on its raw share alongside ROI rows. + hot = [ + _hot("k_eff", name="a", gpu_pct=20.0, attain=95.0), # ROI 1.0 + _hot("k_raw", name="b", gpu_pct=6.0), # ROI 6.0 + _hot("k_mid", name="c", gpu_pct=10.0, attain=70.0), # ROI 3.0 + ] + untried = kd.untried_hot_reusable_kernels(_state(hot), min_gpu_pct=1.0, top_n=3) + assert untried == ["k_raw", "k_mid", "k_eff"] + + +class TestFloorStaysOnRawShare: + """``min_gpu_pct`` asks whether a kernel is big enough to bother with.""" + + def test_large_saturated_kernel_still_clears_the_floor(self): + # ROI is 0.5, well under the 1.0 floor, but the floor is not an ROI + # test: the kernel owns half the trace and stays a candidate. + hot = [_hot("k_big", name="a", gpu_pct=50.0, attain=99.0)] + untried = kd.untried_hot_reusable_kernels(_state(hot), min_gpu_pct=1.0, top_n=5) + assert untried == ["k_big"] + + def test_tiny_kernel_with_full_headroom_is_still_excluded(self): + # Headroom must not be able to promote a kernel that is too small to + # matter past the floor. + hot = [_hot("k_tiny", name="a", gpu_pct=0.4, attain=0.0)] + untried = kd.untried_hot_reusable_kernels(_state(hot), min_gpu_pct=1.0, top_n=5) + assert untried == [] diff --git a/src/hyperloom/inference_optimizer/tests/test_roofline_effective.py b/src/hyperloom/inference_optimizer/tests/test_roofline_effective.py new file mode 100644 index 0000000000..3adf53e3e3 --- /dev/null +++ b/src/hyperloom/inference_optimizer/tests/test_roofline_effective.py @@ -0,0 +1,395 @@ +# SPDX-FileCopyrightText: 2026 Advanced Micro Devices, Inc. +# SPDX-License-Identifier: MIT + +"""Tests for ``orchestrator.kernel.roofline_effective``. + +Covers the effective-frequency compute derate, the achievable-bandwidth memory +derate, extraction of sustained clocks from telemetry, and the guarantee that +every unmeasured / malformed path degrades to the historical boost-anchored +ceiling rather than inventing one. +""" + +from __future__ import annotations + +from types import SimpleNamespace + +import pytest + +from hyperloom.inference_optimizer.gpu_types import _AMD_GPU_DISPATCH_IDENTITIES +from hyperloom.orchestrator.kernel.roofline_ceiling import ( + HW_SPECS, + ModelMeta, + compute_compute_bound_ceiling_tok_per_sec, + compute_roofline_from_perfmodel, + compute_theoretical_peak_output_tok_per_sec, +) +from hyperloom.orchestrator.kernel.roofline_effective import ( + _BW_EFFICIENCY_ENV, + EffectiveClocks, + GpuFreqSpec, + effective_clock_provenance, + effective_clocks_from_entry, + effective_clocks_from_report, + effective_clocks_from_samples, + hbm_bw_efficiency, + resolve_effective_clocks_from_state, + resolve_freq_spec, + sclk_derate_factor, +) + + +@pytest.fixture(autouse=True) +def _clear_bw_env(monkeypatch): + """Keep the bandwidth-efficiency override out of unrelated assertions.""" + monkeypatch.delenv(_BW_EFFICIENCY_ENV, raising=False) + + +class TestBoostClocksAnchorTheVendorTable: + """The vendor ``peak_tflops`` table is ``CUs * FLOPs_per_clk * f_boost``. + + Inverting the published TFLOPS against CU count and the recorded boost clock + must land exactly on the architectural MFMA rate. This is the evidence that + the vendor peaks are boost-anchored, and it guards the boost constants the + derate divides by. + """ + + @pytest.mark.parametrize( + ("gpu_type", "precision", "expected_flops_per_clk_per_cu"), + [ + ("mi300x", "bf16", 2048.0), + ("mi300x", "fp8", 4096.0), + ("mi325x", "bf16", 2048.0), + ("mi355x", "bf16", 4096.0), + ("mi355x", "fp8", 8192.0), + ("mi355x", "mxfp4", 16384.0), + ], + ) + def test_vendor_peak_inverts_to_architectural_rate(self, gpu_type, precision, expected_flops_per_clk_per_cu): + peak_tflops = HW_SPECS[gpu_type]["peak_tflops"][precision] + cus = _AMD_GPU_DISPATCH_IDENTITIES[gpu_type][1] + boost_hz = resolve_freq_spec(gpu_type).boost_sclk_mhz * 1e6 + flops_per_clk_per_cu = peak_tflops * 1e12 / (cus * boost_hz) + assert flops_per_clk_per_cu == pytest.approx(expected_flops_per_clk_per_cu, rel=1e-3) + + def test_every_hw_spec_gpu_has_a_frequency_spec(self): + # A GPU with a compute peak but no boost clock would silently skip the + # derate, so the tables must stay in step. + assert set(HW_SPECS) == set(_AMD_GPU_DISPATCH_IDENTITIES) + for gpu_type in HW_SPECS: + assert resolve_freq_spec(gpu_type) is not None + + +class TestSclkDerateFactor: + """Compute throughput is linear in engine clock; the factor is that ratio.""" + + def test_at_boost_is_exactly_one(self): + clocks = EffectiveClocks(sclk_mhz=2400.0, samples=10) + assert sclk_derate_factor("mi355x", clocks) == 1.0 + + def test_below_boost_scales_linearly(self): + clocks = EffectiveClocks(sclk_mhz=1800.0, samples=10) + assert sclk_derate_factor("mi355x", clocks) == pytest.approx(0.75) + + def test_above_reference_clamps_to_one(self): + # A measurement above boost means the reference is wrong; inflating a + # ceiling on bad telemetry is worse than leaving it alone. + clocks = EffectiveClocks(sclk_mhz=3000.0, samples=10) + assert sclk_derate_factor("mi355x", clocks) == 1.0 + + def test_implausibly_low_clock_degrades_to_no_op(self): + clocks = EffectiveClocks(sclk_mhz=50.0, samples=10) + assert sclk_derate_factor("mi355x", clocks) == 1.0 + + @pytest.mark.parametrize( + "clocks", + [ + None, + EffectiveClocks(), + EffectiveClocks(sclk_mhz=1800.0, samples=0), + EffectiveClocks(sclk_mhz=0.0, samples=10), + ], + ids=["none", "empty", "no-samples", "no-clock"], + ) + def test_unmeasured_degrades_to_no_op(self, clocks): + assert sclk_derate_factor("mi355x", clocks) == 1.0 + + def test_unknown_gpu_degrades_to_no_op(self): + clocks = EffectiveClocks(sclk_mhz=1200.0, samples=10) + assert sclk_derate_factor("mi999x", clocks) == 1.0 + + def test_vendor_convention_uses_boost_reference(self): + clocks = EffectiveClocks(sclk_mhz=1050.0, samples=10) + assert sclk_derate_factor("mi300x", clocks, convention="vendor") == pytest.approx(0.5) + + def test_achievable_reference_overrides_boost_when_recorded(self): + # ref_sclk_mhz exists so a sub-boost achievable measurement is not + # double-counted once its true clock is known. + spec = GpuFreqSpec(boost_sclk_mhz=2400.0, ref_sclk_mhz=2000.0) + assert spec.reference_sclk("achievable") == 2000.0 + assert spec.reference_sclk("vendor") == 2400.0 + + def test_unset_reference_falls_back_to_boost(self): + spec = GpuFreqSpec(boost_sclk_mhz=2400.0) + assert spec.reference_sclk("achievable") == 2400.0 + + +class TestHbmBandwidthEfficiency: + """Memory is derated by access efficiency, never by clock (mclk is pinned).""" + + def test_measured_part_uses_its_calibrated_figure(self): + # Measured on an 8-GPU gfx950 node; see _GPU_FREQ_SPECS for methodology. + assert hbm_bw_efficiency("mi355x") == pytest.approx(0.89) + + def test_unmeasured_part_stays_at_vendor_peak(self): + # An unmeasured part must keep its historical ceiling rather than + # inherit another part's efficiency. + assert hbm_bw_efficiency("mi300x") == 1.0 + + def test_unknown_gpu_is_no_op(self): + assert hbm_bw_efficiency("mi999x") == 1.0 + + def test_env_override_applies(self, monkeypatch): + monkeypatch.setenv(_BW_EFFICIENCY_ENV, "0.72") + assert hbm_bw_efficiency("mi355x") == pytest.approx(0.72) + + @pytest.mark.parametrize("raw", ["", "abc", "0", "-0.5", "1.5"]) + def test_invalid_or_out_of_range_override_ignored(self, monkeypatch, raw): + monkeypatch.setenv(_BW_EFFICIENCY_ENV, raw) + assert hbm_bw_efficiency("mi355x") == pytest.approx(0.89) + + def test_table_efficiency_clamped_to_valid_range(self): + assert GpuFreqSpec(boost_sclk_mhz=2400.0, hbm_bw_efficiency=0.8).hbm_bw_efficiency == 0.8 + + +class TestEffectiveClocksFromSamples: + """Sustained clock is the mean over *active* samples only.""" + + def test_idle_samples_are_excluded(self): + # Idle ticks sit at a low DPM state; averaging them in would understate + # the sustained clock and over-derate the ceiling. + samples = [ + {"clock_mhz": 1413.0, "gpu_util_pct": 0.0}, + {"clock_mhz": 2000.0, "gpu_util_pct": 99.0}, + {"clock_mhz": 2100.0, "gpu_util_pct": 98.0}, + ] + clocks = effective_clocks_from_samples(samples) + assert clocks.sclk_mhz == pytest.approx(2050.0) + assert clocks.samples == 2 + assert clocks.measured + + def test_single_busy_gpu_is_not_swamped_by_idle_peers(self): + # Reproduces a real 8-GPU MI355X capture: one loaded card at ~2350 MHz + # while seven idle peers sat at ~95 MHz. Averaging the whole node gave + # 299.7 MHz, which would have derated the ceiling roughly 8x the wrong + # way and inflated within% accordingly. + samples = [{"clock_mhz": 2350.0, "gpu_util_pct": 100.0}] + samples += [{"clock_mhz": 95.0, "gpu_util_pct": 0.0} for _ in range(7 * 10)] + clocks = effective_clocks_from_samples(samples) + assert clocks.sclk_mhz == pytest.approx(2350.0) + assert sclk_derate_factor("mi355x", clocks) == pytest.approx(0.9792, abs=1e-3) + + def test_all_samples_kept_when_utilization_absent(self): + samples = [{"clock_mhz": 1800.0}, {"clock_mhz": 2000.0}] + clocks = effective_clocks_from_samples(samples) + assert clocks.sclk_mhz == pytest.approx(1900.0) + assert clocks.samples == 2 + + def test_all_idle_yields_unmeasured(self): + samples = [{"clock_mhz": 1413.0, "gpu_util_pct": 0.0}] + assert not effective_clocks_from_samples(samples).measured + + def test_alternate_sclk_key_accepted(self): + clocks = effective_clocks_from_samples([{"sclk_mhz": 1900.0}]) + assert clocks.sclk_mhz == pytest.approx(1900.0) + + def test_mclk_averaged_for_provenance(self): + samples = [ + {"clock_mhz": 2000.0, "mclk_mhz": 2000.0, "gpu_util_pct": 90.0}, + {"clock_mhz": 2000.0, "mclk_mhz": 2000.0, "gpu_util_pct": 90.0}, + ] + assert effective_clocks_from_samples(samples).mclk_mhz == pytest.approx(2000.0) + + @pytest.mark.parametrize( + "samples", + [None, [], "not-a-list", [None, 3], [{}], [{"clock_mhz": 0.0}], [{"clock_mhz": "x"}]], + ids=["none", "empty", "string", "junk", "no-keys", "zero", "unparseable"], + ) + def test_malformed_input_yields_unmeasured(self, samples): + assert not effective_clocks_from_samples(samples).measured + + def test_from_report_accepts_list_and_dict_shapes(self): + as_list = {"gpu_monitor": [{"clock_mhz": 1900.0}]} + as_dict = {"gpu_monitor": {"clock_mhz": 1900.0}} + assert effective_clocks_from_report(as_list).sclk_mhz == pytest.approx(1900.0) + assert effective_clocks_from_report(as_dict).sclk_mhz == pytest.approx(1900.0) + + @pytest.mark.parametrize("report", [None, {}, "nope", {"gpu_monitor": None}]) + def test_from_report_degrades_on_missing_telemetry(self, report): + assert not effective_clocks_from_report(report).measured + + +class TestCeilingsHonourTheDerates: + """End-to-end: the derates must move the ceilings, and only when measured.""" + + _CMP_KWARGS = dict( + gpu_type="mi355x", + num_gpus=8, + precision_tag="fp8", + active_weight_bytes=70_000_000_000, + weight_bytes=140_000_000_000, + weight_dtype_bytes=1.0, + ) + + def test_compute_ceiling_unchanged_without_telemetry(self): + base = compute_compute_bound_ceiling_tok_per_sec(**self._CMP_KWARGS) + at_boost = compute_compute_bound_ceiling_tok_per_sec( + **self._CMP_KWARGS, + clocks=EffectiveClocks(sclk_mhz=2400.0, samples=5), + ) + assert at_boost == pytest.approx(base) + + def test_compute_ceiling_scales_with_sustained_clock(self): + base = compute_compute_bound_ceiling_tok_per_sec(**self._CMP_KWARGS) + derated = compute_compute_bound_ceiling_tok_per_sec( + **self._CMP_KWARGS, + clocks=EffectiveClocks(sclk_mhz=1800.0, samples=5), + ) + assert derated == pytest.approx(base * 0.75) + + def test_memory_ceiling_scales_with_bandwidth_efficiency(self, monkeypatch): + kwargs = dict( + gpu_type="mi355x", + num_gpus=8, + weight_bytes=140_000_000_000, + num_layers=80, + num_kv_heads=8, + head_dim=128, + kv_dtype_bytes=2.0, + isl=1024, + osl=1024, + concurrency=64, + ) + # Baseline against an unmeasured part so the table efficiency does not + # confound the ratio, then confirm the override scales it. + monkeypatch.setenv(_BW_EFFICIENCY_ENV, "1.0") + base = compute_theoretical_peak_output_tok_per_sec(**kwargs) + monkeypatch.setenv(_BW_EFFICIENCY_ENV, "0.75") + assert compute_theoretical_peak_output_tok_per_sec(**kwargs) == pytest.approx(base * 0.75) + + def test_measured_part_ceiling_reflects_calibration(self, monkeypatch): + kwargs = dict( + gpu_type="mi355x", + num_gpus=8, + weight_bytes=140_000_000_000, + num_layers=80, + num_kv_heads=8, + head_dim=128, + kv_dtype_bytes=2.0, + isl=1024, + osl=1024, + concurrency=64, + ) + monkeypatch.setenv(_BW_EFFICIENCY_ENV, "1.0") + uncalibrated = compute_theoretical_peak_output_tok_per_sec(**kwargs) + monkeypatch.delenv(_BW_EFFICIENCY_ENV, raising=False) + assert compute_theoretical_peak_output_tok_per_sec(**kwargs) == pytest.approx(uncalibrated * 0.89) + + def test_perfmodel_path_scales_with_sustained_clock(self): + meta = ModelMeta( + weight_bytes=140_000_000_000, + num_layers=80, + num_kv_heads=8, + head_dim=128, + weight_dtype_bytes=2.0, + hidden_size=8192, + intermediate_size=28672, + vocab_size=128256, + num_attention_heads=64, + ) + kwargs = dict(meta=meta, gpu_type="mi355x", concurrency=32, isl=1024, osl=1024, num_gpus=8) + base = compute_roofline_from_perfmodel(**kwargs) + derated = compute_roofline_from_perfmodel(**kwargs, clocks=EffectiveClocks(sclk_mhz=1200.0, samples=5)) + assert base is not None and derated is not None + # Halving the clock halves the compute roof; the memory roof is + # untouched, so the blended decode figure must fall without vanishing. + assert derated.peak_achievable_tflops == pytest.approx(base.peak_achievable_tflops * 0.5) + assert derated.decode_mem_tok_per_s == pytest.approx(base.decode_mem_tok_per_s) + assert 0 < derated.decode_tok_per_s <= base.decode_tok_per_s + + +class TestResolvingClocksFromState: + """The ceiling reads clocks off the measurement, not off disk.""" + + def test_entry_round_trips_recorded_fields(self): + clocks = effective_clocks_from_entry( + {"effective_sclk_mhz": 1850.0, "effective_mclk_mhz": 2000.0, "effective_clock_samples": 12} + ) + assert clocks.sclk_mhz == pytest.approx(1850.0) + assert clocks.mclk_mhz == pytest.approx(2000.0) + assert clocks.samples == 12 + + def test_entry_without_sample_count_still_counts_as_measured(self): + clocks = effective_clocks_from_entry({"effective_sclk_mhz": 1850.0}) + assert clocks.measured + assert clocks.samples == 1 + + @pytest.mark.parametrize( + "entry", + [None, {}, "nope", {"effective_sclk_mhz": 0}, {"effective_sclk_mhz": "x"}], + ) + def test_entry_degrades_when_absent(self, entry): + assert not effective_clocks_from_entry(entry).measured + + def test_state_prefers_current_best_then_baseline(self): + state = SimpleNamespace( + last_baseline={"effective_sclk_mhz": 1500.0, "effective_clock_samples": 5}, + current_best={"effective_sclk_mhz": 1900.0, "effective_clock_samples": 7}, + ) + assert resolve_effective_clocks_from_state(state).sclk_mhz == pytest.approx(1900.0) + + def test_state_falls_back_to_baseline_when_optimized_arm_has_none(self): + state = SimpleNamespace( + last_baseline={"effective_sclk_mhz": 1500.0, "effective_clock_samples": 5}, + current_best={}, + ) + assert resolve_effective_clocks_from_state(state).sclk_mhz == pytest.approx(1500.0) + + def test_baseline_arm_is_pinned_and_ignores_current_best(self): + state = SimpleNamespace( + last_baseline={"effective_sclk_mhz": 1500.0, "effective_clock_samples": 5}, + current_best={"effective_sclk_mhz": 1900.0, "effective_clock_samples": 7}, + ) + clocks = resolve_effective_clocks_from_state(state, arm="baseline") + assert clocks.sclk_mhz == pytest.approx(1500.0) + + def test_state_without_arms_is_unmeasured(self): + assert not resolve_effective_clocks_from_state(SimpleNamespace()).measured + + +class TestProvenance: + """Provenance keeps a derated ``within%`` interpretable next to a raw one.""" + + def test_measured_provenance_reports_applied_derate(self): + prov = effective_clock_provenance( + "mi355x", + EffectiveClocks(sclk_mhz=1800.0, mclk_mhz=2000.0, samples=42), + ) + assert prov["effective_sclk_mhz"] == pytest.approx(1800.0) + assert prov["effective_mclk_mhz"] == pytest.approx(2000.0) + assert prov["effective_clock_samples"] == 42 + assert prov["reference_sclk_mhz"] == 2400.0 + assert prov["sclk_derate_factor"] == pytest.approx(0.75) + assert prov["hbm_bw_efficiency"] == pytest.approx(0.89) + assert prov["effective_derate_source"] == "measured_telemetry" + + def test_unmeasured_provenance_marks_no_derate(self): + prov = effective_clock_provenance("mi355x", None) + assert prov["effective_sclk_mhz"] is None + assert prov["effective_clock_samples"] == 0 + assert prov["sclk_derate_factor"] == 1.0 + assert prov["effective_derate_source"] == "unmeasured_no_derate" + + def test_unknown_gpu_reports_no_reference(self): + prov = effective_clock_provenance("mi999x", EffectiveClocks(sclk_mhz=1800.0, samples=5)) + assert prov["reference_sclk_mhz"] is None + assert prov["sclk_derate_factor"] == 1.0 diff --git a/src/hyperloom/inference_optimizer/tests/test_telemetry_clock_aggregate_unit.py b/src/hyperloom/inference_optimizer/tests/test_telemetry_clock_aggregate_unit.py new file mode 100644 index 0000000000..b7c6b8c2c6 --- /dev/null +++ b/src/hyperloom/inference_optimizer/tests/test_telemetry_clock_aggregate_unit.py @@ -0,0 +1,76 @@ +# SPDX-FileCopyrightText: 2026 Advanced Micro Devices, Inc. +# SPDX-License-Identifier: MIT + +"""Tests for the engine/memory clock fields of the GPU-monitor aggregate. + +These feed the effective-frequency roofline derate, so the aggregate has to +distinguish "no clocks sampled" from "clocks sampled and low". +""" + +from __future__ import annotations + +import json +from pathlib import Path + +from hyperloom.inference_optimizer.breakdown.collectors.telemetry import ( + _aggregate_gpu_monitor, +) + + +def _write_report(tmp_path: Path, samples: list[dict], name: str = "benchmark_report.json") -> Path: + """Write a benchmark report carrying *samples* under ``gpu_monitor``.""" + path = tmp_path / name + path.write_text(json.dumps({"gpu_monitor": samples}), encoding="utf-8") + return path + + +def test_aggregate_reports_clock_stats(tmp_path): + report = _write_report( + tmp_path, + [ + {"power_w": 900.0, "temperature_c": 70.0, "clock_mhz": 1800.0, "mclk_mhz": 2000.0}, + {"power_w": 1100.0, "temperature_c": 80.0, "clock_mhz": 2000.0, "mclk_mhz": 2000.0}, + ], + ) + agg = _aggregate_gpu_monitor([report], []) + assert agg["avg_clock_mhz"] == 1900.0 + assert agg["max_clock_mhz"] == 2000.0 + assert agg["avg_mclk_mhz"] == 2000.0 + assert agg["clock_samples"] == 2 + assert agg["samples"] == 2 + + +def test_clock_samples_distinguishes_unsampled_clocks(tmp_path): + # A sampler predating ``--showclocks`` still yields power/temp rows; the + # derate must be able to tell that apart from a genuinely low clock. + report = _write_report( + tmp_path, + [ + {"power_w": 900.0, "temperature_c": 70.0}, + {"power_w": 950.0, "temperature_c": 72.0, "clock_mhz": 1900.0}, + ], + ) + agg = _aggregate_gpu_monitor([report], []) + assert agg["samples"] == 2 + assert agg["clock_samples"] == 1 + assert agg["avg_clock_mhz"] == 1900.0 + + +def test_aggregate_accepts_alternate_sclk_key(tmp_path): + report = _write_report(tmp_path, [{"sclk_mhz": 1750.0}]) + agg = _aggregate_gpu_monitor([report], []) + assert agg["avg_clock_mhz"] == 1750.0 + assert agg["clock_samples"] == 1 + + +def test_aggregate_without_clocks_reports_zero(tmp_path): + report = _write_report(tmp_path, [{"power_w": 800.0}]) + agg = _aggregate_gpu_monitor([report], []) + assert agg["avg_clock_mhz"] == 0.0 + assert agg["max_clock_mhz"] == 0.0 + assert agg["avg_mclk_mhz"] == 0.0 + assert agg["clock_samples"] == 0 + + +def test_aggregate_empty_when_no_samples(tmp_path): + assert _aggregate_gpu_monitor([_write_report(tmp_path, [])], []) == {} diff --git a/src/hyperloom/orchestrator/actions/executors/benchmark_result.py b/src/hyperloom/orchestrator/actions/executors/benchmark_result.py index 9d48567f85..69015e255b 100644 --- a/src/hyperloom/orchestrator/actions/executors/benchmark_result.py +++ b/src/hyperloom/orchestrator/actions/executors/benchmark_result.py @@ -432,9 +432,21 @@ def _pick(*preds: Any) -> float | None: ) if power is not None: sample["power_w"] = power - clock = _pick(lambda c: "sclk" in c) + # ``--showclocks`` emits a "clk clock speed:" column alongside a + # "clk clock level:" DPM index. Excluding "level" matters: the level cell + # is a small integer that would otherwise be recorded as a frequency. + clock = _pick( + lambda c: "sclk" in c and "level" not in c, + lambda c: "sclk" in c, + ) if clock is not None: sample["clock_mhz"] = clock + mclk = _pick( + lambda c: "mclk" in c and "level" not in c, + lambda c: "mclk" in c, + ) + if mclk is not None: + sample["mclk_mhz"] = mclk util = _pick(lambda c: "gpu use" in c or "gpu_use" in c or c == "gpu%") if util is not None: sample["gpu_util_pct"] = util @@ -774,6 +786,7 @@ def extract_benchmark_measurement( warnings.append("raw_inferencex_result_used") _derive_tpot_if_missing(measurement, report) + _attach_effective_clocks(measurement, report) measurement["valid_measurement"] = is_valid_measurement(measurement) # Second-chance salvage from Magpie leak destinations when the @@ -875,6 +888,34 @@ def _is_scriptable_measurement(result: dict[str, Any]) -> bool: return framework_registry.is_scriptable(result.get("framework")) +def _attach_effective_clocks(measurement: dict[str, Any], report: dict[str, Any] | None) -> None: + """Record the engine clock the benchmark actually sustained, in place. + + Carried on the measurement so the roofline can anchor its compute ceiling to + the sustained clock instead of the vendor boost clock, without having to + rediscover the report from state. Absent clock telemetry leaves the keys + unset, which the ceiling reads as "no derate". + + Args: + measurement (dict[str, Any]): Normalized measurement, mutated in place. + report (dict[str, Any] | None): Parsed ``benchmark_report.json``. + """ + try: + from hyperloom.orchestrator.kernel.roofline_effective import ( + effective_clocks_from_report, + ) + + clocks = effective_clocks_from_report(report) + except Exception: # noqa: BLE001 — telemetry enrichment is best-effort + return + if not clocks.measured: + return + measurement["effective_sclk_mhz"] = round(clocks.sclk_mhz, 1) + measurement["effective_clock_samples"] = clocks.samples + if clocks.mclk_mhz > 0: + measurement["effective_mclk_mhz"] = round(clocks.mclk_mhz, 1) + + def is_valid_measurement(result: dict[str, Any] | None) -> bool: """Return whether a measurement reflects a usable benchmark result. diff --git a/src/hyperloom/orchestrator/kernel/_hw_probe_src.py b/src/hyperloom/orchestrator/kernel/_hw_probe_src.py new file mode 100644 index 0000000000..979bb1ddfb --- /dev/null +++ b/src/hyperloom/orchestrator/kernel/_hw_probe_src.py @@ -0,0 +1,410 @@ +# SPDX-FileCopyrightText: 2026 Advanced Micro Devices, Inc. +# SPDX-License-Identifier: MIT + +"""HIP sources for the on-node hardware probes. + +Embedded as strings rather than shipped as data files so the probes travel with +the wheel and need no ``package_data`` wiring. + +Two probes, answering the two terms of the roofline: + +``MFMA_PROBE_SRC`` + Matrix-core issue rate in FLOPs/clock/CU, per precision. Back-to-back MFMA + with enough independent accumulator chains to cover instruction latency puts + the matrix core at its issue limit, so the achieved rate *is* the + architectural rate. No library kernel is involved, so kernel quality cannot + leak into the number -- which is what keeps the resulting compute roof a + true upper bound rather than something a tuned kernel can beat. + +``BANDWIDTH_PROBE_SRC`` + Absolute achievable streaming-read bandwidth in GB/s. Reported as an + absolute figure rather than a fraction of a theoretical peak, because the + theoretical peak is not reliably derivable at runtime: ``hipDeviceProp_t`` + reports MI355X as 8192-bit at 2000 MHz, which yields 4096 GB/s under the + usual double-data-rate formula against an actual 8000 GB/s, since HBM3E + clocks its pins at four times the reported rate and that multiplier moves + with the HBM generation. + +Both are self-describing on stdout as JSON lines so the caller never has to +guess which variants a given architecture supports. +""" + +from __future__ import annotations + +#: Independent accumulator chains per thread. MFMA has multi-cycle latency, so +#: a single dependent chain measures latency rather than issue rate; eight +#: chains is comfortably past the point where the pipeline stays full. +MFMA_ACCUMULATORS = 8 + +#: Matrix-core issue-rate probe. +#: +#: Every candidate instruction is wrapped in ``#if __has_builtin``, which is only +#: meaningful in the *device* compilation pass -- the AMDGCN builtins are +#: invisible to the host pass, where the same guard always reports false. So the +#: kernel *signatures* are unconditional (the host pass needs the symbols in +#: order to launch them) while only their *bodies* are guarded, and a small +#: availability kernel reports back at runtime which bodies are real. One source +#: therefore compiles unchanged on any architecture and self-selects the +#: variants that target actually has, so a new part needs no table entry -- only +#: an added guarded block if it introduces a new opcode. +#: +#: On gfx950 the fp8 and fp4 rates do NOT come from separate opcodes: both use +#: ``mfma_scale_f32_16x16x128_f8f6f4`` and differ only in the cbsz/blgp format +#: selector (0 = e4m3 fp8, 4 = e2m1 fp4). Probing fp8 through the 16x16x32 +#: opcode instead reports the bf16 rate, because that variant carries identical +#: FLOPs per instruction. +MFMA_PROBE_SRC = r""" +#include +#include +#include +#include +#include + +typedef __bf16 bf16x4 __attribute__((ext_vector_type(4))); +typedef __bf16 bf16x8 __attribute__((ext_vector_type(8))); +typedef _Float16 f16x4 __attribute__((ext_vector_type(4))); +typedef _Float16 f16x8 __attribute__((ext_vector_type(8))); +typedef int i32x8 __attribute__((ext_vector_type(8))); +typedef float f32x4 __attribute__((ext_vector_type(4))); + +#define NACC 8 + +// Bit index per candidate, shared between the device-side availability report +// and the host-side registration. +#define BIT_BF16_16X16X32 0 +#define BIT_BF16_16X16X16 1 +#define BIT_F16_16X16X16 2 +#define BIT_FP8_16X16X32 3 +#define BIT_F8F6F4 4 +#define BIT_F16_16X16X32 5 + +// Body shared by every candidate: fill NACC independent chains, hammer the +// matrix core, then consume the result through a branch that never fires so +// the optimizer cannot delete the loop. +#define MFMA_BODY(SETUP, EXPR) \ + SETUP; \ + f32x4 acc[NACC]; \ + _Pragma("unroll") for (int i = 0; i < NACC; ++i) acc[i] = f32x4{}; \ + for (int it = 0; it < iters; ++it) { \ + _Pragma("unroll") for (int i = 0; i < NACC; ++i) acc[i] = (EXPR); \ + } \ + float s = 0; \ + _Pragma("unroll") for (int i = 0; i < NACC; ++i) s += acc[i][0]; \ + if (s == -1.0f) out[0] = s; + +#define BF16_SETUP \ + bf16x8 a, b; \ + for (int i = 0; i < 8; ++i) { a[i] = (__bf16)1.0f; b[i] = (__bf16)1.0f; } +#define BF16X4_SETUP \ + bf16x4 a, b; \ + for (int i = 0; i < 4; ++i) { a[i] = (__bf16)1.0f; b[i] = (__bf16)1.0f; } +#define F16X4_SETUP \ + f16x4 a, b; \ + for (int i = 0; i < 4; ++i) { a[i] = (_Float16)1.0f; b[i] = (_Float16)1.0f; } +#define F16_SETUP \ + f16x8 a, b; \ + for (int i = 0; i < 8; ++i) { a[i] = (_Float16)1.0f; b[i] = (_Float16)1.0f; } +#define I64_SETUP long a = 0x0101010101010101L, b = 0x0101010101010101L; + +// Reports which candidate bodies the device pass actually compiled. The host +// pass cannot answer this itself, so it asks the device at runtime. +__global__ void availability(unsigned* out) { + unsigned m = 0; +#if __has_builtin(__builtin_amdgcn_mfma_f32_16x16x32_bf16) + m |= 1u << BIT_BF16_16X16X32; +#endif +#if __has_builtin(__builtin_amdgcn_mfma_f32_16x16x16bf16_1k) + m |= 1u << BIT_BF16_16X16X16; +#endif +#if __has_builtin(__builtin_amdgcn_mfma_f32_16x16x16f16) + m |= 1u << BIT_F16_16X16X16; +#endif +#if __has_builtin(__builtin_amdgcn_mfma_f32_16x16x32_fp8_fp8) + m |= 1u << BIT_FP8_16X16X32; +#endif +#if __has_builtin(__builtin_amdgcn_mfma_scale_f32_16x16x128_f8f6f4) + m |= 1u << BIT_F8F6F4; +#endif +#if __has_builtin(__builtin_amdgcn_mfma_f32_16x16x32_f16) + m |= 1u << BIT_F16_16X16X32; +#endif + out[0] = m; +} + +// Signatures are unconditional so the host pass can take their addresses; only +// the bodies are guarded. An unavailable variant compiles to an empty kernel +// that is never registered, because its availability bit stays clear. +__global__ void mfma_bf16_16x16x32(float* out, int iters) { +#if __has_builtin(__builtin_amdgcn_mfma_f32_16x16x32_bf16) + MFMA_BODY(BF16_SETUP, __builtin_amdgcn_mfma_f32_16x16x32_bf16(a, b, acc[i], 0, 0, 0)) +#else + (void)out; + (void)iters; +#endif +} + +__global__ void mfma_bf16_16x16x16(float* out, int iters) { +#if __has_builtin(__builtin_amdgcn_mfma_f32_16x16x16bf16_1k) + MFMA_BODY(BF16X4_SETUP, __builtin_amdgcn_mfma_f32_16x16x16bf16_1k(a, b, acc[i], 0, 0, 0)) +#else + (void)out; + (void)iters; +#endif +} + +__global__ void mfma_f16_16x16x16(float* out, int iters) { +#if __has_builtin(__builtin_amdgcn_mfma_f32_16x16x16f16) + MFMA_BODY(F16X4_SETUP, __builtin_amdgcn_mfma_f32_16x16x16f16(a, b, acc[i], 0, 0, 0)) +#else + (void)out; + (void)iters; +#endif +} + +__global__ void mfma_f16_16x16x32(float* out, int iters) { +#if __has_builtin(__builtin_amdgcn_mfma_f32_16x16x32_f16) + MFMA_BODY(F16_SETUP, __builtin_amdgcn_mfma_f32_16x16x32_f16(a, b, acc[i], 0, 0, 0)) +#else + (void)out; + (void)iters; +#endif +} + +__global__ void mfma_fp8_16x16x32(float* out, int iters) { +#if __has_builtin(__builtin_amdgcn_mfma_f32_16x16x32_fp8_fp8) + MFMA_BODY(I64_SETUP, __builtin_amdgcn_mfma_f32_16x16x32_fp8_fp8(a, b, acc[i], 0, 0, 0)) +#else + (void)out; + (void)iters; +#endif +} + +// The gfx950 double/quad-rate path. cbsz/blgp must be compile-time immediates, +// so the format selector is a template parameter rather than an argument. +template +__global__ void mfma_f8f6f4(float* out, int iters) { +#if __has_builtin(__builtin_amdgcn_mfma_scale_f32_16x16x128_f8f6f4) + MFMA_BODY(i32x8 a{}; i32x8 b{}, + __builtin_amdgcn_mfma_scale_f32_16x16x128_f8f6f4(a, b, acc[i], FMT, FMT, 0, 0, 0, 0)) +#else + (void)out; + (void)iters; +#endif +} + +struct Candidate { + const char* precision; + const char* variant; + void (*kernel)(float*, int); + double flops_per_inst; // 2*M*N*K for one wavefront instruction + int bit; +}; + +int main(int argc, char** argv) { + int device = (argc > 1) ? atoi(argv[1]) : 0; + int iters = (argc > 2) ? atoi(argv[2]) : 20000; + if (hipSetDevice(device) != hipSuccess) { + fprintf(stderr, "hipSetDevice(%d) failed\n", device); + return 1; + } + hipDeviceProp_t prop; + if (hipGetDeviceProperties(&prop, device) != hipSuccess) { + fprintf(stderr, "hipGetDeviceProperties failed\n"); + return 1; + } + int cus = prop.multiProcessorCount; + + unsigned* dmask = nullptr; + if (hipMalloc(&dmask, sizeof(unsigned)) != hipSuccess) { + fprintf(stderr, "hipMalloc failed\n"); + return 1; + } + hipLaunchKernelGGL(availability, dim3(1), dim3(1), 0, 0, dmask); + unsigned mask = 0; + if (hipMemcpy(&mask, dmask, sizeof(unsigned), hipMemcpyDeviceToHost) != hipSuccess) { + fprintf(stderr, "availability probe failed\n"); + return 1; + } + (void)hipFree(dmask); + + const Candidate all[] = { + {"bf16", "mfma_f32_16x16x32_bf16", mfma_bf16_16x16x32, 2.0 * 16 * 16 * 32, + BIT_BF16_16X16X32}, + {"bf16", "mfma_f32_16x16x16bf16_1k", mfma_bf16_16x16x16, 2.0 * 16 * 16 * 16, + BIT_BF16_16X16X16}, + {"fp16", "mfma_f32_16x16x16f16", mfma_f16_16x16x16, 2.0 * 16 * 16 * 16, BIT_F16_16X16X16}, + {"fp16", "mfma_f32_16x16x32_f16", mfma_f16_16x16x32, 2.0 * 16 * 16 * 32, BIT_F16_16X16X32}, + {"fp8", "mfma_f32_16x16x32_fp8_fp8", mfma_fp8_16x16x32, 2.0 * 16 * 16 * 32, + BIT_FP8_16X16X32}, + {"fp8", "mfma_scale_f32_16x16x128_f8f6f4[e4m3]", mfma_f8f6f4<0>, 2.0 * 16 * 16 * 128, + BIT_F8F6F4}, + {"fp4", "mfma_scale_f32_16x16x128_f8f6f4[e2m1]", mfma_f8f6f4<4>, 2.0 * 16 * 16 * 128, + BIT_F8F6F4}, + }; + std::vector cands; + for (const auto& c : all) { + if (mask & (1u << c.bit)) cands.push_back(c); + } + + float* out = nullptr; + if (hipMalloc(&out, sizeof(float)) != hipSuccess) { + fprintf(stderr, "hipMalloc failed\n"); + return 1; + } + + const int threads = 256; + const int waves = threads / 64; + const int blocks = cus * 2; + + printf("{\"kind\":\"device\",\"arch\":\"%s\",\"cus\":%d,\"boost_mhz\":%.0f}\n", prop.gcnArchName, + cus, prop.clockRate / 1000.0); + for (const auto& c : cands) { + hipLaunchKernelGGL(c.kernel, dim3(blocks), dim3(threads), 0, 0, out, 100); + if (hipDeviceSynchronize() != hipSuccess) continue; + + std::vector runs; + for (int rep = 0; rep < 5; ++rep) { + hipEvent_t t0, t1; + if (hipEventCreate(&t0) != hipSuccess) break; + if (hipEventCreate(&t1) != hipSuccess) break; + (void)hipEventRecord(t0); + hipLaunchKernelGGL(c.kernel, dim3(blocks), dim3(threads), 0, 0, out, iters); + (void)hipEventRecord(t1); + if (hipEventSynchronize(t1) != hipSuccess) break; + float ms = 0; + if (hipEventElapsedTime(&ms, t0, t1) != hipSuccess || ms <= 0) break; + double insts = (double)blocks * waves * iters * NACC; + runs.push_back(insts * c.flops_per_inst / (ms * 1e-3)); + (void)hipEventDestroy(t0); + (void)hipEventDestroy(t1); + } + if (runs.empty()) continue; + std::sort(runs.begin(), runs.end()); + printf("{\"kind\":\"mfma\",\"precision\":\"%s\",\"variant\":\"%s\",\"flops_per_sec\":%.6e}\n", + c.precision, c.variant, runs.back()); + fflush(stdout); + } + (void)hipFree(out); + return 0; +} +""" + +#: Streaming-read bandwidth probe. +#: +#: Non-temporal (cache-bypassing) fully coalesced loads, so the figure is the +#: hardware streaming limit rather than any particular kernel's efficiency. The +#: decode roofline counts read traffic (weights plus KV cache), so a read probe +#: is the right shape; copy and triad land far lower (~61% of peak on MI355X) +#: and would model write-heavy traffic this roofline does not have. +#: +#: ``float4`` is spelled as a native ``ext_vector_type`` rather than HIP's +#: ``float4``, which is a class type the non-temporal builtins reject. +BANDWIDTH_PROBE_SRC = r""" +#include +#include +#include +#include +#include +#include + +typedef float vec_t __attribute__((ext_vector_type(4))); + +static double now_epoch() { + struct timespec ts; + clock_gettime(CLOCK_REALTIME, &ts); + return ts.tv_sec + ts.tv_nsec * 1e-9; +} + +__global__ void stream_read(const vec_t* __restrict__ src, size_t n, float* out) { + size_t stride = (size_t)gridDim.x * blockDim.x; + vec_t acc = {0.f, 0.f, 0.f, 0.f}; + for (size_t i = (size_t)blockIdx.x * blockDim.x + threadIdx.x; i < n; i += stride) { + vec_t v = __builtin_nontemporal_load(&src[i]); + acc += v; + } + float s = acc[0] + acc[1] + acc[2] + acc[3]; + if (s == -1.0f) out[0] = s; +} + +int main(int argc, char** argv) { + int device = (argc > 1) ? atoi(argv[1]) : 0; + double target_gib = (argc > 2) ? atof(argv[2]) : 16.0; + // Wall-clock instant at which to begin the timed loop. Concurrent instances + // are given a common value so their measured windows genuinely overlap: + // without it each process finishes allocation and warm-up at a different + // moment, and an "all GPUs loaded" run reports the same figure as a solo one. + double start_at = (argc > 3) ? atof(argv[3]) : 0.0; + double measure_sec = (argc > 4) ? atof(argv[4]) : 2.0; + if (hipSetDevice(device) != hipSuccess) { + fprintf(stderr, "hipSetDevice(%d) failed\n", device); + return 1; + } + hipDeviceProp_t prop; + if (hipGetDeviceProperties(&prop, device) != hipSuccess) return 1; + + size_t free_b = 0, total_b = 0; + if (hipMemGetInfo(&free_b, &total_b) != hipSuccess) return 1; + // Stay well clear of whatever else is resident; the probe must never be the + // reason a serving process hits an allocation failure. + size_t want = (size_t)(target_gib * (1ull << 30)); + size_t cap = (size_t)(free_b * 0.5); + size_t bytes = want < cap ? want : cap; + bytes &= ~(size_t)(sizeof(vec_t) - 1); + if (bytes < (1ull << 28)) { + fprintf(stderr, "insufficient free VRAM for bandwidth probe\n"); + return 1; + } + + vec_t* buf = nullptr; + if (hipMalloc(&buf, bytes) != hipSuccess) { + fprintf(stderr, "hipMalloc(%zu) failed\n", bytes); + return 1; + } + (void)hipMemset(buf, 1, bytes); + float* out = nullptr; + if (hipMalloc(&out, sizeof(float)) != hipSuccess) return 1; + + size_t n = bytes / sizeof(vec_t); + int threads = 256; + int blocks = prop.multiProcessorCount * 8; + + hipLaunchKernelGGL(stream_read, dim3(blocks), dim3(threads), 0, 0, buf, n, out); + if (hipDeviceSynchronize() != hipSuccess) return 1; + + // Spin to the shared start instant. Sleeping would risk waking late and + // missing the window the other instances are measuring in. + while (start_at > 0 && now_epoch() < start_at) { + } + + // Duration-based rather than a fixed iteration count: one pass over a 16 GiB + // buffer takes only ~2.4 ms at these rates, so a handful of iterations would + // finish well inside the process-start skew between concurrent instances and + // measure nothing about contention. + std::vector runs; + double deadline = now_epoch() + measure_sec; + while (now_epoch() < deadline) { + hipEvent_t t0, t1; + if (hipEventCreate(&t0) != hipSuccess) break; + if (hipEventCreate(&t1) != hipSuccess) break; + (void)hipEventRecord(t0); + hipLaunchKernelGGL(stream_read, dim3(blocks), dim3(threads), 0, 0, buf, n, out); + (void)hipEventRecord(t1); + if (hipEventSynchronize(t1) != hipSuccess) break; + float ms = 0; + if (hipEventElapsedTime(&ms, t0, t1) != hipSuccess || ms <= 0) break; + runs.push_back((double)bytes / (ms * 1e-3) / 1e9); + (void)hipEventDestroy(t0); + (void)hipEventDestroy(t1); + } + if (runs.empty()) { + fprintf(stderr, "no successful bandwidth iterations\n"); + return 1; + } + std::sort(runs.begin(), runs.end()); + printf("{\"kind\":\"bandwidth\",\"arch\":\"%s\",\"buffer_bytes\":%zu,\"gb_per_sec\":%.3f}\n", + prop.gcnArchName, bytes, runs.back()); + (void)hipFree(buf); + (void)hipFree(out); + return 0; +} +""" diff --git a/src/hyperloom/orchestrator/kernel/_kernel_decisions.py b/src/hyperloom/orchestrator/kernel/_kernel_decisions.py index ecafdd1f71..73f53f9991 100644 --- a/src/hyperloom/orchestrator/kernel/_kernel_decisions.py +++ b/src/hyperloom/orchestrator/kernel/_kernel_decisions.py @@ -1411,18 +1411,91 @@ def kernel_opt_attempts_count(state) -> int: return len(state.kernel_opt_task_attempts or {}) +def _kernel_optimization_roi(row: dict[str, Any], gpu_pct: float) -> float: + """GPU-time share weighted by how far the kernel sits below its roofline. + + Ranking candidates on ``gpu_pct`` alone spends the attempt budget on + whatever owns the most trace time, including kernels already running at + their roofline that have nothing left to give. Weighting by + ``1 - attainment`` puts attempts where measured headroom actually is: a + kernel at 11% of GPU time and 30% attainment outranks one at 12% and 90%. + + The bypass report already computes this as ``optimization_priority``, so + prefer that value and keep one definition of the ROI. Recompute only for + the TraceLens path, which carries roofline fields but no ROI, and degrade + to the raw share when attainment is unknown -- that reproduces the previous + ordering rather than inventing headroom. + + Args: + row: A ``hot_kernels`` entry. + gpu_pct: The kernel's already-parsed GPU-time share. + + Returns: + The ROI to rank by; equals ``gpu_pct`` when no efficiency is known. + """ + precomputed = row.get("optimization_priority") + if isinstance(precomputed, (int, float)) and not isinstance(precomputed, bool): + return float(precomputed) + attainment = _roofline_attainment_pct(row) + if attainment is None: + return gpu_pct + headroom = 1.0 - min(max(attainment, 0.0), 100.0) / 100.0 + return gpu_pct * headroom + + +def _roofline_attainment_pct(row: dict[str, Any]) -> float | None: + """How much of its roofline a kernel already attains, or ``None`` if unknown. + + Attainment must be read on the BINDING side. ``efficiency_percent`` is + compute-side, so a memory-bound kernel pinned at its bandwidth roof still + reports ~0 there; treating that as headroom would rank the one kernel with + nothing to recover at the top. The bypass route publishes the binding-side + value directly. The TraceLens route does not, so its compute-side number is + trusted only when the kernel is compute-bound, and a memory-bound row + without attainment stays unknown rather than being scored on the wrong axis. + + A clamped estimate is refused outright. ``roofline_estimate_capped`` marks + a closed form that overshot the roof and was truncated to 100%, which says + the model does not fit the kernel -- not that the kernel is saturated. A + MoE grouped GEMM billed for all experts by a dense form is the live case; + trusting it would zero the ROI of a kernel nobody has actually measured. + + Args: + row: A ``hot_kernels`` entry. + + Returns: + Attainment in percent, or ``None`` when the row cannot supply one. + """ + if row.get("roofline_estimate_capped"): + return None + attainment = row.get("roofline_attainment_pct") + if isinstance(attainment, (int, float)) and not isinstance(attainment, bool): + return float(attainment) + eff = row.get("efficiency_percent") + if not isinstance(eff, (int, float)) or isinstance(eff, bool): + return None + if str(row.get("bound_type") or "").strip().lower() != "compute_bound": + return None + return float(eff) + + def untried_hot_reusable_kernels( state, *, min_gpu_pct: float | None = None, top_n: int | None = None, ) -> list[str]: - """Hot kernels still owing a ``kernel_opt`` attempt (reusable, gpu_pct >= min_gpu_pct, untouched); capped to top_n by gpu_pct, one kernel_id per task_group. + """Hot kernels still owing a ``kernel_opt`` attempt (reusable, gpu_pct >= min_gpu_pct, untouched); capped to top_n by optimization ROI, one kernel_id per task_group. + + ROI is the GPU-time share weighted by roofline headroom (see + :func:`_kernel_optimization_roi`), so the cap keeps kernels that can still + move rather than the largest ones. Kernels whose roofline attainment is + unknown rank on their raw share, as before. Args: min_gpu_pct (float | None): Minimum GPU-share threshold; when ``None`` it is read from ``HYPERLOOM_KERNEL_OPT_MIN_GPU_PCT``. - top_n (int | None): Cap on enforced kernels by gpu_pct; when + top_n (int | None): Cap on enforced kernels by ROI; when ``None`` it is read from ``HYPERLOOM_KERNEL_OPT_GATE_TOP_N``. Returns: @@ -1486,8 +1559,11 @@ def untried_hot_reusable_kernels( _ensure_kernel_task_state(state) attempts = state.kernel_opt_task_attempts or {} - # Sort by gpu_pct desc so dedup picks the strongest member of each - # task_group. + # Sort by optimization ROI desc so dedup picks the strongest member of each + # task_group, and so the top_n cut keeps the kernels with headroom rather + # than merely the largest ones. The min_gpu_pct floor below stays on the + # raw share: it answers "is this kernel big enough to bother with", which + # headroom must not be able to talk it out of. rows: list[tuple[float, str, str, list[str], str, tuple[str, str, float]]] = [] for k in hot: if not isinstance(k, dict): @@ -1516,7 +1592,9 @@ def untried_hot_reusable_kernels( # Identity of the underlying kernel, independent of the synthetic # per-row kernel_id. Used only as a dedup fallback (see below). identity = (src, str(k.get("name") or k.get("operation") or ""), gpu_pct) - rows.append((gpu_pct, kid, src, members, group_key, identity)) + rows.append( + (_kernel_optimization_roi(k, gpu_pct), kid, src, members, group_key, identity) + ) rows.sort(key=lambda x: x[0], reverse=True) ranked: list[tuple[float, str, str, list[str], str, tuple[str, str, float]]] = [] @@ -1600,7 +1678,7 @@ def _matches_current_task(member_id: str, group_key: str, source: str) -> bool: recorded_source = str(attempt.get("last_source_file") or "") return not source or not recorded_source or source == recorded_source - for _pct, kid, src, members, group_key, _identity in ranked: + for _roi, kid, src, members, group_key, _identity in ranked: if members and all( _member_is_rejected(member) and _matches_current_task(member, group_key, src) diff --git a/src/hyperloom/orchestrator/kernel/hw_probe.py b/src/hyperloom/orchestrator/kernel/hw_probe.py new file mode 100644 index 0000000000..a73255bd29 --- /dev/null +++ b/src/hyperloom/orchestrator/kernel/hw_probe.py @@ -0,0 +1,884 @@ +# SPDX-FileCopyrightText: 2026 Advanced Micro Devices, Inc. +# SPDX-License-Identifier: MIT + +"""On-node hardware probes backing the roofline ceiling. + +Both terms of the roofline are hardware capabilities, and both are currently +read from hand-maintained per-SKU tables. That is avoidable: the hardware can be +asked directly, and asking is more accurate than the tables. + +For compute, the vendor peak is exactly ``ISA_rate x CUs x boost_clock`` -- every +published entry inverts onto the architectural MFMA rate with no residual. Two of +those three come straight from ``hipDeviceProp_t``, and the third is measurable: +a back-to-back MFMA kernel with enough independent accumulator chains to cover +instruction latency recovers the documented rate to within a few percent on every +precision (measured on gfx950: bf16 93%, fp16 96%, fp8 99%, fp4 97%). Because the +probe is raw MFMA rather than a library GEMM, nothing about kernel quality enters +the number, so the resulting roof stays a genuine upper bound that a tuned kernel +cannot beat -- unlike a microbenchmark-derived "achievable" figure, which bakes +one kernel's inefficiency into the ceiling and hides the very headroom the kernel +agent exists to recover. + +For memory the probe reports absolute achievable GB/s rather than a fraction of +a theoretical peak, because the theoretical peak is *not* derivable at runtime: +``hipDeviceProp_t`` describes MI355X as 8192-bit at 2000 MHz, which the usual +double-data-rate formula turns into 4096 GB/s against an actual 8000 GB/s, since +HBM3E clocks its pins at four times the reported rate and that multiplier moves +with the HBM generation. + +Everything here is best-effort and fails soft. Probing needs ``hipcc`` and a +visible GPU; when either is missing, or a compile or run fails, every entry point +returns ``None`` and the caller falls back through the existing chain +(max-achievable table, then vendor peak) exactly as before. + +Probing and reading are deliberately separate calls. :func:`probe_and_cache` +compiles and runs, and is meant to be invoked at a known point such as +environment setup; the resolver-facing readers only ever touch the cache, so a +roofline computation can never trigger a surprise compile in the middle of a +measurement window. +""" + +from __future__ import annotations + +import functools +import json +import logging +import os +import shutil +import subprocess +import threading +import time +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import Any + +from hyperloom.inference_optimizer.session.paths import deps_cache_root + +from ._hw_probe_src import BANDWIDTH_PROBE_SRC, MFMA_PROBE_SRC + +log = logging.getLogger(__name__) + +#: Set truthy to skip probing entirely; every reader then returns ``None`` and +#: callers keep their table-derived values. +DISABLE_ENV = "HYPERLOOM_GPU_PROBE_DISABLE" + +#: Overrides the per-subprocess timeout (seconds) for compiles and probe runs. +TIMEOUT_ENV = "HYPERLOOM_GPU_PROBE_TIMEOUT_SEC" + +#: Compiling the probe dominates; the runs themselves are ~1-3 s. +_DEFAULT_TIMEOUT_SEC = 180.0 + +#: Bumped whenever the cached payload's meaning changes, so a stale cache from +#: an older build is ignored rather than misread. +_SCHEMA_VERSION = 1 + +#: Floor, as a fraction of boost, on a sampled probe clock we are willing to +#: normalize by. The sampled clock is a *divisor*, so under-reading it inflates +#: the derived rate and therefore the ceiling -- the one direction this module +#: must never fail in. Below this the sample is treated as unusable and boost is +#: used instead, which can only understate the rate. +_MIN_PROBE_SCLK_FRACTION = 0.5 + +#: Lead time before the synchronized bandwidth window opens, covering +#: allocation, buffer fill, and warm-up on every concurrent instance. +_BANDWIDTH_START_LEAD_SEC = 8.0 + +#: Length of the synchronized bandwidth window. +_BANDWIDTH_MEASURE_SEC = 2.0 + + +@dataclass(frozen=True) +class DeviceInfo: + """Runtime-discovered device identity behind a probe result. + + Attributes: + arch (str): Normalized architecture, e.g. ``"gfx950"`` (feature suffixes + such as ``:sramecc+`` are stripped). + cu_count (int): Compute units reported by the runtime. This is what + makes a cut-down part correct without a table entry. + boost_sclk_mhz (float): Peak engine clock reported by the runtime. + """ + + arch: str + cu_count: int + boost_sclk_mhz: float + + +@dataclass(frozen=True) +class MfmaRate: + """Measured matrix-core issue rate for one precision. + + Attributes: + precision (str): Precision tag (``bf16``, ``fp16``, ``fp8``, ``fp4``). + variant (str): Winning instruction variant, recorded so a surprising + rate can be traced to the opcode that produced it. + flops_per_clk_per_cu (float): Measured architectural rate. + """ + + precision: str + variant: str + flops_per_clk_per_cu: float + + +@dataclass(frozen=True) +class ProbeResult: + """Everything one node's probe run established. + + Attributes: + schema (int): Payload schema version. + device (DeviceInfo): Runtime-discovered device identity. + rocm_version (str): ROCm version the probe was built against; part of + the cache key, since a toolchain change can move the numbers. + mfma_rates (dict[str, MfmaRate]): Winning rate per precision. + bandwidth_gb_per_sec (dict[int, float]): Per-GPU achievable streaming + bandwidth, keyed by how many GPUs were loaded concurrently. + probe_sclk_mhz (float): Engine clock sampled during the MFMA probe, used + to convert its FLOP/s into a per-clock rate. + probed_at (float): Unix timestamp of the run. + """ + + schema: int + device: DeviceInfo + rocm_version: str + mfma_rates: dict[str, MfmaRate] + bandwidth_gb_per_sec: dict[int, float] + probe_sclk_mhz: float + probed_at: float + + +def _disabled() -> bool: + """Whether probing has been switched off by the environment. + + Returns: + ``True`` when the disable flag is set to a truthy token. + """ + return os.environ.get(DISABLE_ENV, "").strip().lower() in {"1", "true", "yes", "on"} + + +def _timeout_sec() -> float: + """Per-subprocess timeout for compiles and probe runs. + + Returns: + The configured timeout, or the default when unset or unparseable. + """ + raw = os.environ.get(TIMEOUT_ENV, "") + try: + value = float(raw) + except (TypeError, ValueError): + return _DEFAULT_TIMEOUT_SEC + return value if value > 0 else _DEFAULT_TIMEOUT_SEC + + +def normalize_arch(arch: str | None) -> str: + """Strip target-feature suffixes from a gfx architecture string. + + ``hipDeviceProp_t`` reports ``gfx950:sramecc+:xnack-``; the features do not + change the matrix-core rate and would fragment the cache key. + + Args: + arch: Raw architecture string. + + Returns: + The bare architecture (``"gfx950"``), or ``""`` when unavailable. + """ + return (arch or "").split(":", 1)[0].strip().lower() + + +def rocm_version() -> str: + """Installed ROCm version, for the cache key. + + Returns: + The version string, or ``"unknown"`` when it cannot be read. + """ + for path in (Path("/opt/rocm/.info/version"), Path("/opt/rocm/.info/version-dev")): + try: + text = path.read_text(encoding="utf-8").strip() + except OSError: + continue + if text: + return text.split("-", 1)[0] + return "unknown" + + +@functools.lru_cache(maxsize=1) +def detect_arch() -> str: + """Architecture of the local GPU, without running a probe. + + Needed to find the cache entry. Tries torch first (cheapest where it is + installed), then the ``GFX Version`` column of ``rocm-smi + --showproductname``. Note that ``--showhw`` refuses CSV output, so it is not + a usable source here. + + Returns: + The normalized architecture, or ``""`` when undetectable. + """ + try: + import torch # noqa: PLC0415 (optional, and import cost is real) + + if torch.cuda.is_available(): + return normalize_arch(torch.cuda.get_device_properties(0).gcnArchName) + except Exception: # noqa: BLE001 (torch absent, no driver, no device) + pass + if not shutil.which("rocm-smi"): + return "" + try: + out = subprocess.run( + ["rocm-smi", "--showproductname", "--csv"], + capture_output=True, + text=True, + timeout=10.0, + check=False, + ).stdout + except (OSError, subprocess.SubprocessError): + return "" + for token in out.replace(",", " ").split(): + if token.lower().startswith("gfx") and any(ch.isdigit() for ch in token): + return normalize_arch(token) + return "" + + +def probe_cache_dir() -> Path: + """Directory holding compiled probes and their cached results. + + Returns: + The probe cache directory (not created). + """ + return deps_cache_root() / "gpu_probes" + + +def _cache_path(arch: str, rocm: str) -> Path: + """Path of the cached probe payload for an architecture and toolchain. + + Args: + arch: Normalized architecture. + rocm: ROCm version string. + + Returns: + The JSON cache path. + """ + return probe_cache_dir() / f"probe-{arch}-rocm{rocm}.json" + + +def _sample_sclk_once() -> float: + """Highest engine clock currently reported across visible GPUs. + + Reads the ``sclk clock speed:`` column by name. Scanning the row for any + plausible number instead would silently return the memory clock, which sits + in the same range and never droops -- and an under-read clock *inflates* the + derived per-clock rate, so this has to be exact rather than approximate. + + The maximum rather than the mean: the probe loads one GPU while its peers + idle near 95 MHz, and averaging those in would understate the clock the + probe actually ran at. + + Returns: + The clock in MHz, or ``0.0`` when unavailable. + """ + if not shutil.which("rocm-smi"): + return 0.0 + try: + out = subprocess.run( + ["rocm-smi", "--showclocks", "--csv"], + capture_output=True, + text=True, + timeout=10.0, + check=False, + ).stdout + except (OSError, subprocess.SubprocessError): + return 0.0 + import csv # noqa: PLC0415 (only needed on this path) + + best = 0.0 + try: + rows = list(csv.DictReader(out.splitlines())) + except csv.Error: + return 0.0 + for row in rows: + for key, value in row.items(): + if not key or "sclk" not in key.lower() or "speed" not in key.lower(): + continue + digits = "".join(ch for ch in str(value or "") if ch.isdigit()) + if digits: + best = max(best, float(digits)) + return best + + +class _SclkSampler: + """Samples engine clock in the background for the duration of a probe.""" + + def __init__(self, interval_sec: float = 0.25) -> None: + """Initialize the sampler. + + Args: + interval_sec: Delay between samples. + """ + self._interval = interval_sec + self._stop = threading.Event() + self._samples: list[float] = [] + self._thread: threading.Thread | None = None + + def _loop(self) -> None: + """Poll until stopped, keeping positive samples.""" + while not self._stop.is_set(): + value = _sample_sclk_once() + if value > 0: + self._samples.append(value) + self._stop.wait(self._interval) + + def __enter__(self) -> _SclkSampler: + """Start sampling. + + Returns: + This sampler. + """ + self._thread = threading.Thread(target=self._loop, daemon=True) + self._thread.start() + return self + + def __exit__(self, *exc: Any) -> None: + """Stop sampling and join the thread.""" + self._stop.set() + if self._thread is not None: + self._thread.join(timeout=5.0) + + @property + def mean_mhz(self) -> float: + """Mean of the collected samples. + + Returns: + The mean engine clock, or ``0.0`` when nothing was sampled. + """ + return sum(self._samples) / len(self._samples) if self._samples else 0.0 + + +def _compile(src: str, stem: str, arch: str, cache_dir: Path) -> Path | None: + """Compile one embedded probe source, reusing a current binary. + + Args: + src: HIP source text. + stem: Base name for the source and binary. + arch: Normalized offload architecture. + cache_dir: Directory to build in. + + Returns: + Path to the executable, or ``None`` when the toolchain is missing or the + compile fails. + """ + hipcc = shutil.which("hipcc") + if not hipcc: + log.debug("hw_probe: hipcc not found; skipping %s", stem) + return None + try: + cache_dir.mkdir(parents=True, exist_ok=True) + src_path = cache_dir / f"{stem}.hip" + binary = cache_dir / f"{stem}-{arch}" + if not src_path.exists() or src_path.read_text(encoding="utf-8") != src: + src_path.write_text(src, encoding="utf-8") + elif binary.exists() and binary.stat().st_mtime >= src_path.stat().st_mtime: + return binary + proc = subprocess.run( + [hipcc, "-O3", f"--offload-arch={arch}", str(src_path), "-o", str(binary)], + capture_output=True, + text=True, + timeout=_timeout_sec(), + check=False, + ) + except (OSError, subprocess.SubprocessError) as exc: + log.debug("hw_probe: compiling %s failed: %r", stem, exc) + return None + if proc.returncode != 0 or not binary.exists(): + log.debug("hw_probe: compiling %s failed: %s", stem, (proc.stderr or "")[:400]) + return None + return binary + + +def _run(binary: Path, args: list[str], *, env: dict[str, str] | None = None) -> str: + """Run a compiled probe and capture stdout. + + Args: + binary: Probe executable. + args: Command-line arguments. + env: Optional environment overlay (used to mask visible devices). + + Returns: + Captured stdout, or ``""`` on any failure. + """ + merged = {**os.environ, **(env or {})} + try: + proc = subprocess.run( + [str(binary), *args], + capture_output=True, + text=True, + timeout=_timeout_sec(), + check=False, + env=merged, + ) + except (OSError, subprocess.SubprocessError) as exc: + log.debug("hw_probe: running %s failed: %r", binary.name, exc) + return "" + if proc.returncode != 0: + log.debug("hw_probe: %s exited %d: %s", binary.name, proc.returncode, (proc.stderr or "")[:400]) + return "" + return proc.stdout or "" + + +def _parse_json_lines(text: str) -> list[dict[str, Any]]: + """Parse the probe's JSON-lines stdout, skipping unparseable lines. + + Args: + text: Raw stdout. + + Returns: + The decoded objects. + """ + out: list[dict[str, Any]] = [] + for line in text.splitlines(): + line = line.strip() + if not line.startswith("{"): + continue + try: + obj = json.loads(line) + except json.JSONDecodeError: + continue + if isinstance(obj, dict): + out.append(obj) + return out + + +def detect_gpu_count() -> int: + """Number of GPUs visible to this process. + + Returns: + The device count, or ``0`` when it cannot be determined. + """ + try: + import torch # noqa: PLC0415 + + if torch.cuda.is_available(): + return int(torch.cuda.device_count()) + except Exception: # noqa: BLE001 + pass + if not shutil.which("rocm-smi"): + return 0 + try: + out = subprocess.run( + ["rocm-smi", "--showid", "--csv"], + capture_output=True, + text=True, + timeout=10.0, + check=False, + ).stdout + except (OSError, subprocess.SubprocessError): + return 0 + return sum(1 for line in out.splitlines() if line.strip().lower().startswith("card")) + + +def _probe_mfma(binary: Path) -> tuple[DeviceInfo | None, dict[str, MfmaRate], float]: + """Run the matrix-core probe and reduce it to one rate per precision. + + The probe emits every instruction variant the device supports; the fastest + per precision wins. That matters because the wide and narrow forms of the + same precision differ by 2x, and on gfx950 the fp8 and fp4 rates come from a + shared ``f8f6f4`` opcode that a naive per-precision opcode guess misses + entirely. + + Rates are normalized by the clock sampled *during* the run rather than by + boost, so a probe that ran below boost is not misreported as a slower part. + + Args: + binary: Compiled MFMA probe. + + Returns: + ``(device, rates_by_precision, probe_sclk_mhz)``; the device is ``None`` + and the mapping empty when the probe produced nothing usable. + """ + with _SclkSampler() as sampler: + text = _run(binary, ["0", "20000"]) + records = _parse_json_lines(text) + if not records: + return None, {}, 0.0 + + device: DeviceInfo | None = None + for rec in records: + if rec.get("kind") == "device": + device = DeviceInfo( + arch=normalize_arch(str(rec.get("arch", ""))), + cu_count=int(rec.get("cus", 0) or 0), + boost_sclk_mhz=float(rec.get("boost_mhz", 0.0) or 0.0), + ) + break + if device is None or device.cu_count <= 0: + return None, {}, 0.0 + + # Reject a missing or implausibly low sample and use boost instead. A probe + # this compute-dense sits at or near boost, so the substitution costs little + # accuracy, and it errs toward understating the rate rather than inflating + # the ceiling. + sclk = sampler.mean_mhz + if sclk < device.boost_sclk_mhz * _MIN_PROBE_SCLK_FRACTION: + log.debug( + "hw_probe: sampled sclk %.1f MHz implausible against %.1f MHz boost; using boost", + sclk, + device.boost_sclk_mhz, + ) + sclk = device.boost_sclk_mhz + if sclk <= 0: + return device, {}, 0.0 + + rates: dict[str, MfmaRate] = {} + for rec in records: + if rec.get("kind") != "mfma": + continue + precision = str(rec.get("precision", "")).strip().lower() + flops_per_sec = float(rec.get("flops_per_sec", 0.0) or 0.0) + if not precision or flops_per_sec <= 0: + continue + rate = flops_per_sec / (sclk * 1e6) / device.cu_count + current = rates.get(precision) + if current is None or rate > current.flops_per_clk_per_cu: + rates[precision] = MfmaRate( + precision=precision, + variant=str(rec.get("variant", "")), + flops_per_clk_per_cu=rate, + ) + return device, rates, sclk + + +def _probe_bandwidth_at(binary: Path, devices: list[int]) -> float: + """Mean per-GPU streaming bandwidth with *devices* loaded concurrently. + + Every instance is handed the same wall-clock start instant so the + measurement windows genuinely overlap; letting them free-run makes an + all-GPU probe silently report the solo figure. + + Measuring under load rather than assuming: on MI355X the per-GPU figure + turns out to be flat at ~7130 GB/s whether one GPU streams or all eight do + (verified at 100% utilization on every card, ~990 W each), because each GPU + owns its HBM stacks and there is no shared path to contend for. Parts that + do share a memory path would show it here instead of going unnoticed. + + Args: + binary: Compiled bandwidth probe. + devices: Device indices to load simultaneously. + + Returns: + Mean per-GPU GB/s, or ``0.0`` when no device reported. + """ + results: dict[int, float] = {} + lock = threading.Lock() + # Enough lead time for every instance to allocate, fill, and warm up before + # the shared window opens. + start_at = time.time() + _BANDWIDTH_START_LEAD_SEC + + def _one(index: int) -> None: + """Run the probe on one device and record its bandwidth.""" + text = _run( + binary, + [str(index), "16", f"{start_at:.3f}", str(_BANDWIDTH_MEASURE_SEC)], + ) + for rec in _parse_json_lines(text): + if rec.get("kind") != "bandwidth": + continue + value = float(rec.get("gb_per_sec", 0.0) or 0.0) + if value > 0: + with lock: + results[index] = value + + threads = [threading.Thread(target=_one, args=(d,)) for d in devices] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=_timeout_sec() + 30.0) + return sum(results.values()) / len(results) if results else 0.0 + + +def probe_and_cache(*, force: bool = False) -> ProbeResult | None: + """Run both probes on this node and persist the result. + + Compiles and executes, so call it from setup rather than from anything on a + measurement path. Results are cached per architecture and ROCm version; an + existing cache short-circuits the whole thing unless *force* is set. + + Bandwidth is measured twice, with one GPU loaded and with all of them, since + the attainable per-GPU figure depends on how many peers are streaming. + + Args: + force: Re-probe and overwrite even when a cache entry exists. + + Returns: + The probe result, or ``None`` when probing is disabled, unsupported, or + failed at any step. + """ + if _disabled(): + log.debug("hw_probe: disabled via %s", DISABLE_ENV) + return None + arch = detect_arch() + if not arch: + log.debug("hw_probe: no AMD GPU architecture detected") + return None + rocm = rocm_version() + if not force: + cached = load_cached(arch=arch, rocm=rocm) + if cached is not None: + return cached + + cache_dir = probe_cache_dir() + mfma_bin = _compile(MFMA_PROBE_SRC, "mfma_probe", arch, cache_dir) + if mfma_bin is None: + return None + device, rates, sclk = _probe_mfma(mfma_bin) + if device is None or not rates: + log.debug("hw_probe: matrix-core probe produced no rates") + return None + + bandwidth: dict[int, float] = {} + bw_bin = _compile(BANDWIDTH_PROBE_SRC, "bw_probe", arch, cache_dir) + if bw_bin is not None: + count = detect_gpu_count() + single = _probe_bandwidth_at(bw_bin, [0]) + if single > 0: + bandwidth[1] = single + if count > 1: + allgpu = _probe_bandwidth_at(bw_bin, list(range(count))) + if allgpu > 0: + bandwidth[count] = allgpu + + result = ProbeResult( + schema=_SCHEMA_VERSION, + device=device, + rocm_version=rocm, + mfma_rates=rates, + bandwidth_gb_per_sec=bandwidth, + probe_sclk_mhz=sclk, + probed_at=time.time(), + ) + _write_cache(result) + return result + + +def _write_cache(result: ProbeResult) -> None: + """Persist a probe result, tolerating an unwritable cache directory. + + Args: + result: The result to store. + """ + path = _cache_path(result.device.arch, result.rocm_version) + payload = { + "schema": result.schema, + "device": asdict(result.device), + "rocm_version": result.rocm_version, + "mfma_rates": {k: asdict(v) for k, v in result.mfma_rates.items()}, + # JSON object keys are strings; readers coerce back to int. + "bandwidth_gb_per_sec": {str(k): v for k, v in result.bandwidth_gb_per_sec.items()}, + "probe_sclk_mhz": result.probe_sclk_mhz, + "probed_at": result.probed_at, + } + try: + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_suffix(".json.tmp") + tmp.write_text(json.dumps(payload, indent=2, sort_keys=True), encoding="utf-8") + tmp.replace(path) + except OSError as exc: + log.debug("hw_probe: could not write cache %s: %r", path, exc) + clear_caches() + + +def clear_caches() -> None: + """Drop memoized detection and cache reads. + + Needed after a fresh probe writes new results, and by tests that swap the + cache directory or stub the detection path underneath a warm memo. Tolerates + either function having been replaced by a plain callable, which is exactly + what a test that stubs detection does. + """ + for fn in (detect_arch, _load_cached_impl): + clear = getattr(fn, "cache_clear", None) + if callable(clear): + clear() + + +def load_cached(*, arch: str | None = None, rocm: str | None = None) -> ProbeResult | None: + """Read a cached probe result without touching the GPU. + + This is the resolver-facing entry point: it sits behind every ceiling + computation, so it must stay cheap. The disk read and the architecture + detection behind it are both memoized -- without that, resolving a ceiling + would shell out to ``rocm-smi`` and stat the cache every single call. + + Args: + arch: Architecture to look up; detected when omitted. + rocm: ROCm version to look up; detected when omitted. + + Returns: + The cached result, or ``None`` when probing is disabled or no usable + entry exists. + """ + if _disabled(): + return None + resolved_arch = normalize_arch(arch) or detect_arch() + if not resolved_arch: + return None + return _load_cached_impl(resolved_arch, rocm or rocm_version()) + + +@functools.lru_cache(maxsize=8) +def _load_cached_impl(arch: str, rocm: str) -> ProbeResult | None: + """Memoized cache read for one architecture and toolchain. + + Args: + arch: Normalized architecture. + rocm: ROCm version string. + + Returns: + The cached result, or ``None`` when absent or unusable. + """ + path = _cache_path(arch, rocm) + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return None + if not isinstance(payload, dict) or payload.get("schema") != _SCHEMA_VERSION: + return None + try: + device = DeviceInfo(**payload["device"]) + rates = {k: MfmaRate(**v) for k, v in (payload.get("mfma_rates") or {}).items()} + bandwidth = {int(k): float(v) for k, v in (payload.get("bandwidth_gb_per_sec") or {}).items()} + return ProbeResult( + schema=int(payload["schema"]), + device=device, + rocm_version=str(payload.get("rocm_version", "")), + mfma_rates=rates, + bandwidth_gb_per_sec=bandwidth, + probe_sclk_mhz=float(payload.get("probe_sclk_mhz", 0.0) or 0.0), + probed_at=float(payload.get("probed_at", 0.0) or 0.0), + ) + except (KeyError, TypeError, ValueError) as exc: + log.debug("hw_probe: malformed cache %s: %r", path, exc) + return None + + +def expected_arch_for(gpu_type: str | None) -> str: + """Architecture a GPU type key is expected to run on. + + Args: + gpu_type: GPU type key such as ``"mi300x"``. + + Returns: + The normalized architecture, or ``""`` when the key is unknown or + absent. + """ + if not gpu_type: + return "" + try: + from hyperloom.inference_optimizer.gpu_types import ( # noqa: PLC0415 + amd_gpu_dispatch_identity, + ) + + identity = amd_gpu_dispatch_identity(gpu_type) + except Exception: # noqa: BLE001 (unknown key must not be fatal) + return "" + return normalize_arch(identity[0]) if identity else "" + + +def _probe_for(gpu_type: str | None, result: ProbeResult | None) -> ProbeResult | None: + """Cached probe result, but only when it describes *gpu_type*. + + A roofline is frequently computed for a GPU that is not the one running the + code -- comparing against another part, or replaying a recorded session. The + cache only ever describes the local device, so handing it to a caller asking + about a different part silently substitutes one part's hardware for + another's. That is worse than having no measurement at all, since it moves + the ceiling in an unpredictable direction rather than falling back. + + Args: + gpu_type: GPU type the caller is asking about; ``None`` skips the check. + result: Caller-supplied result, returned as-is when present. + + Returns: + A probe result safe to use for *gpu_type*, or ``None``. + """ + if result is not None: + return result + probe = load_cached() + if probe is None: + return None + expected = expected_arch_for(gpu_type) + if expected and expected != probe.device.arch: + log.debug( + "hw_probe: ignoring %s probe for %s (expects %s)", + probe.device.arch, + gpu_type, + expected, + ) + return None + return probe + + +def probe_compute_peak_tflops( + precision: str | None, + *, + gpu_type: str | None = None, + sustained_sclk_mhz: float = 0.0, + result: ProbeResult | None = None, +) -> float | None: + """Measured compute roof in TFLOPS for one precision. + + Evaluates ``measured_MFMA_rate x CUs x clock``, where the clock is the + workload's sustained engine clock when telemetry supplied one and the + device's boost otherwise. Because the rate came from raw MFMA rather than a + library GEMM, this is a hardware limit that a tuned kernel cannot exceed. + + Args: + precision: Precision tag (``bf16``, ``fp16``, ``fp8``, ``fp4``). + gpu_type: GPU the ceiling is being built for. The probe is ignored + unless it describes this part. + sustained_sclk_mhz: Measured sustained engine clock; ``0`` falls back to + the device boost clock. + result: Pre-loaded probe result; read from cache when omitted. + + Returns: + The compute roof in TFLOPS, or ``None`` when the precision was never + probed, the cache describes a different part, or no cache exists. + """ + probe = _probe_for(gpu_type, result) + if probe is None: + return None + rate = probe.mfma_rates.get((precision or "").strip().lower()) + if rate is None or rate.flops_per_clk_per_cu <= 0: + return None + clock = sustained_sclk_mhz if sustained_sclk_mhz > 0 else probe.device.boost_sclk_mhz + if clock <= 0 or probe.device.cu_count <= 0: + return None + return rate.flops_per_clk_per_cu * probe.device.cu_count * clock * 1e6 / 1e12 + + +def probe_hbm_bandwidth_gb_per_sec( + *, + gpu_type: str | None = None, + active_gpus: int = 1, + result: ProbeResult | None = None, +) -> float | None: + """Measured per-GPU achievable streaming bandwidth. + + Picks the measurement taken with the closest number of GPUs loaded. On + MI355X the two measurements agree to within 0.2%, since per-GPU HBM is + private; the keying exists so a part that *does* share a memory path is + described correctly rather than assumed away. + + Args: + gpu_type: GPU the ceiling is being built for. The probe is ignored + unless it describes this part. + active_gpus: How many GPUs the workload actually loads. + result: Pre-loaded probe result; read from cache when omitted. + + Returns: + Per-GPU GB/s, or ``None`` when bandwidth was never probed or the cache + describes a different part. + """ + probe = _probe_for(gpu_type, result) + if probe is None or not probe.bandwidth_gb_per_sec: + return None + target = max(int(active_gpus), 1) + closest = min(probe.bandwidth_gb_per_sec, key=lambda k: (abs(k - target), k)) + value = probe.bandwidth_gb_per_sec[closest] + return value if value > 0 else None diff --git a/src/hyperloom/orchestrator/kernel/roofline_ceiling.py b/src/hyperloom/orchestrator/kernel/roofline_ceiling.py index 601fa9c178..9cb3fdf6a4 100644 --- a/src/hyperloom/orchestrator/kernel/roofline_ceiling.py +++ b/src/hyperloom/orchestrator/kernel/roofline_ceiling.py @@ -22,6 +22,14 @@ from hyperloom.inference_optimizer.model_config_utils import _merge_config_scopes +from .hw_probe import probe_compute_peak_tflops +from .roofline_effective import ( + EffectiveClocks, + effective_hbm_bw_gbps, + resolve_effective_clocks_from_state, + sclk_derate_factor, +) + #: GPU per-chip peak specs (keys match ``SharedState.gpu_type``, lowercase). #: ``hbm_bw_gbps`` is vendor peak; ``peak_tflops`` is DENSE peak (missing key @@ -618,6 +626,74 @@ def _resolve_peak_tflops(gpu_type: str | None, precision_tag: str | None) -> flo return _resolve_tflops(HW_SPECS, gpu_type, precision_tag) +def resolve_effective_compute_tflops( + gpu_type: str | None, + precision_tag: str | None, + clocks: "EffectiveClocks | None" = None, +) -> tuple[float, str]: + """Compute roof in TFLOPS, preferring measurement over tables. + + Precedence, each layer falling through to the next when it has nothing: + + 1. An on-node probe of the matrix-core issue rate, evaluated at the clock + the workload actually sustained. This is a hardware limit rather than a + library result, so it is a genuine upper bound. + 2. The max-achievable table, derated by the sustained/reference clock ratio. + 3. The vendor dense peak, derated the same way. + + The probe result is *already* clock-scaled, so it must not be passed through + the derate again -- the two mechanisms express the same correction and + applying both would square it. + + Args: + gpu_type: GPU type key for the table lookups. + precision_tag: Precision key. + clocks: Measured effective clocks; ``None`` leaves table peaks unscaled. + + Returns: + ``(tflops, source)`` where source is one of ``"probe"``, + ``"achievable_table"``, ``"vendor_table"`` or ``"unresolved"``; the + TFLOPS is ``0.0`` when nothing resolved. + """ + sustained = clocks.sclk_mhz if (clocks is not None and clocks.measured) else 0.0 + probed = probe_compute_peak_tflops( + precision_tag, gpu_type=gpu_type, sustained_sclk_mhz=sustained + ) + if probed is not None and probed > 0: + return probed, "probe" + + achievable = _resolve_achievable_tflops(gpu_type, precision_tag) + if achievable > 0: + peak, convention, source = achievable, "achievable", "achievable_table" + else: + peak, convention, source = ( + _resolve_peak_tflops(gpu_type, precision_tag), + "vendor", + "vendor_table", + ) + if peak <= 0: + return 0.0, "unresolved" + return peak * sclk_derate_factor(gpu_type, clocks, convention=convention), source + + +def _resolve_effective_compute_tflops( + gpu_type: str | None, + precision_tag: str | None, + clocks: "EffectiveClocks | None" = None, +) -> float: + """Compute roof in TFLOPS, discarding the provenance. + + Args: + gpu_type: GPU type key for the table lookups. + precision_tag: Precision key. + clocks: Measured effective clocks. + + Returns: + The effective compute peak in TFLOPS, or ``0.0`` on a lookup miss. + """ + return resolve_effective_compute_tflops(gpu_type, precision_tag, clocks)[0] + + @dataclass(frozen=True) class ModelMeta: """HF subset needed for the decode roofline ceiling. @@ -943,7 +1019,8 @@ def compute_theoretical_peak_output_tok_per_sec( spec = HW_SPECS.get((gpu_type or "").strip().lower()) if spec is None: return 0.0 - bw_total_bytes_per_sec = spec["hbm_bw_gbps"] * 1e9 * max(num_gpus, 1) + per_gpu_gbps, _ = effective_hbm_bw_gbps(gpu_type, spec["hbm_bw_gbps"], active_gpus=num_gpus) + bw_total_bytes_per_sec = per_gpu_gbps * 1e9 * max(num_gpus, 1) batch = max(concurrency, 1) kv_bytes = compute_kv_bytes_per_token( num_layers=num_layers, @@ -977,6 +1054,7 @@ def compute_compute_bound_ceiling_tok_per_sec( active_weight_bytes: int, weight_bytes: int, weight_dtype_bytes: float, + clocks: "EffectiveClocks | None" = None, ) -> float: """Decode-only compute-bound ceiling for ``output_throughput``. @@ -995,12 +1073,13 @@ def compute_compute_bound_ceiling_tok_per_sec( active_weight_bytes: Per-token active weight bytes at B=1. weight_bytes: Total weight bytes (fallback when active is missing). weight_dtype_bytes: Weight bytes-per-element. + clocks: Measured effective clocks; ``None`` leaves the peak unscaled. Returns: The compute-bound decode throughput ceiling, or ``0.0`` on missing input. """ - peak_tflops = _resolve_achievable_tflops(gpu_type, precision_tag) or _resolve_peak_tflops(gpu_type, precision_tag) + peak_tflops = _resolve_effective_compute_tflops(gpu_type, precision_tag, clocks) if peak_tflops <= 0 or weight_dtype_bytes <= 0: return 0.0 # B=1 per-token figure; fall back to dense weight_bytes when active is missing. @@ -1179,7 +1258,8 @@ def compute_diffusion_mem_img_per_sec(*, gpu_type: str, num_gpus: int, weight_by spec = HW_SPECS.get((gpu_type or "").strip().lower()) if spec is None: return 0.0 - bw = spec["hbm_bw_gbps"] * 1e9 * max(num_gpus, 1) + per_gpu_gbps, _ = effective_hbm_bw_gbps(gpu_type, spec["hbm_bw_gbps"], active_gpus=num_gpus) + bw = per_gpu_gbps * 1e9 * max(num_gpus, 1) if weight_bytes <= 0 or num_steps <= 0 or bw <= 0: return 0.0 per_step_s = weight_bytes / bw @@ -1294,6 +1374,7 @@ def compute_diffusion_compute_img_per_sec( num_layers: int, hidden_size: int, num_steps: int, + clocks: "EffectiveClocks | None" = None, ) -> float: """Compute-roofline ceiling for diffusion image throughput (images/sec). @@ -1314,11 +1395,12 @@ def compute_diffusion_compute_img_per_sec( num_layers: DiT transformer layers (for the attention-score term). hidden_size: DiT model dim (for the attention-score term). num_steps: Denoising steps per image. + clocks: Measured effective clocks; ``None`` leaves the peak unscaled. Returns: The compute-bound images/sec ceiling, or ``0.0`` on degenerate input. """ - peak_tflops = _resolve_achievable_tflops(gpu_type, precision_tag) or _resolve_peak_tflops(gpu_type, precision_tag) + peak_tflops = _resolve_effective_compute_tflops(gpu_type, precision_tag, clocks) if peak_tflops <= 0 or dit_params <= 0 or latent_tokens <= 0 or num_steps <= 0: return 0.0 linear = 2.0 * dit_params * latent_tokens @@ -1330,7 +1412,12 @@ def compute_diffusion_compute_img_per_sec( return peak_flops / flops_per_image -def _compute_diffusion_breakdown_from_state(state: Any, runtime: RuntimeWorkload) -> RooflineBreakdown: +def _compute_diffusion_breakdown_from_state( + state: Any, + runtime: RuntimeWorkload, + *, + clocks: "EffectiveClocks | None" = None, +) -> RooflineBreakdown: """Diffusion (xDiT) roofline breakdown in images/sec. Ceiling = ``min(memory, compute)`` (the binding side), like the LLM path. @@ -1344,6 +1431,7 @@ def _compute_diffusion_breakdown_from_state(state: Any, runtime: RuntimeWorkload Args: state: Shared run state (for the denoising step count). runtime: Resolved runtime workload (model_path / gpu_type / tp). + clocks: Measured effective clocks; ``None`` leaves the peak unscaled. Returns: The diffusion ``RooflineBreakdown``, or ``_EMPTY_BREAKDOWN`` when the @@ -1391,6 +1479,7 @@ def _compute_diffusion_breakdown_from_state(state: Any, runtime: RuntimeWorkload num_layers=num_layers, hidden_size=hidden, num_steps=num_steps, + clocks=clocks, ) mem_img_s = compute_diffusion_mem_img_per_sec( gpu_type=runtime.gpu_type, @@ -1408,6 +1497,7 @@ def compute_roofline_breakdown_from_state( state: Any, *, arm: str | None = None, + clocks: "EffectiveClocks | None" = None, ) -> RooflineBreakdown: """Primary decode ceiling + T_mem/T_cmp side projections. @@ -1419,15 +1509,19 @@ def compute_roofline_breakdown_from_state( Args: state: Shared run state to resolve the workload and dtype from. arm: Pins precision to a specific arm; ``None`` infers it. + clocks: Measured effective clocks for the run being modelled; ``None`` + keeps the boost-anchored ceiling. Returns: The decode ``RooflineBreakdown`` (``_EMPTY_BREAKDOWN`` on missing fields). """ runtime = resolve_runtime_workload(state, arm=arm) + if clocks is None: + clocks = resolve_effective_clocks_from_state(state, arm=arm) # Diffusion (xDiT) uses a distinct images/sec ceiling. if (runtime.framework or "").strip().lower() == "xdit": - return _compute_diffusion_breakdown_from_state(state, runtime) + return _compute_diffusion_breakdown_from_state(state, runtime, clocks=clocks) meta = load_model_meta( runtime.model_path, precision_hint=runtime.precision, @@ -1464,6 +1558,7 @@ def compute_roofline_breakdown_from_state( active_weight_bytes=meta.active_weight_bytes, weight_bytes=meta.weight_bytes, weight_dtype_bytes=meta.weight_dtype_bytes, + clocks=clocks, ) if mem <= 0 and cmp <= 0: return _EMPTY_BREAKDOWN @@ -1480,6 +1575,7 @@ def compute_roofline_breakdown_from_state( osl=runtime.osl, num_gpus=num_gpus, precision_tag=precision_tag, + clocks=clocks, ) if pm_bd is not None and pm_bd.decode_tok_per_s > 0: return RooflineBreakdown( @@ -1494,17 +1590,23 @@ def compute_roofline_breakdown_from_state( return legacy -def compute_peak_from_state(state: Any, *, arm: str | None = None) -> float: +def compute_peak_from_state( + state: Any, + *, + arm: str | None = None, + clocks: "EffectiveClocks | None" = None, +) -> float: """Convenience scalar wrapper for ``T_peak`` only (kept for backward compat; prefer ``compute_roofline_breakdown_from_state``). ``arm`` pins precision to a specific arm. Args: state: Shared run state to compute the ceiling from. arm: Pins precision to a specific arm; ``None`` infers it. + clocks: Measured effective clocks; ``None`` leaves the peak unscaled. Returns: The peak decode throughput (``peak_tok_per_sec``). """ - return compute_roofline_breakdown_from_state(state, arm=arm).peak_tok_per_sec + return compute_roofline_breakdown_from_state(state, arm=arm, clocks=clocks).peak_tok_per_sec def read_baseline_server_args(state: Any) -> str: @@ -1589,20 +1691,39 @@ def _resolve_achievable_tflops(gpu_type: str | None, precision_tag: str | None) return _resolve_tflops(HW_SPECS_ACHIEVABLE, gpu_type, precision_tag) -def resolve_compute_peak_provenance(gpu_type: str | None, precision_tag: str | None) -> dict[str, Any]: +def resolve_compute_peak_provenance( + gpu_type: str | None, + precision_tag: str | None, + clocks: "EffectiveClocks | None" = None, +) -> dict[str, Any]: """Provenance for the compute-peak TFLOPS used by every compute ceiling. - The unified convention is max-achievable (sustained) TFLOPS; the vendor - dense peak is only a coverage-gap fallback. Surfacing convention + value + - source keeps within%/gap interpretable. + Preferred convention is an on-node probe of the matrix-core issue rate, + which is a hardware limit rather than a library result. Failing that, the + max-achievable table; failing that, the vendor dense peak. Surfacing which + layer answered, alongside the value, keeps ``within%`` interpretable -- + a probe-derived roof and a table-derived one are not the same quantity, and + only the former is guaranteed unbeatable by a tuned kernel. Args: gpu_type: GPU type key for the peak lookup. precision_tag: Precision key for the peak lookup. + clocks: Measured effective clocks, so a probe-derived peak is reported + at the clock the workload actually sustained. Returns: ``{compute_peak_convention, compute_peak_tflops, compute_peak_source}``. """ + sustained = clocks.sclk_mhz if (clocks is not None and clocks.measured) else 0.0 + probed = probe_compute_peak_tflops( + precision_tag, gpu_type=gpu_type, sustained_sclk_mhz=sustained + ) + if probed is not None and probed > 0: + return { + "compute_peak_convention": "probe", + "compute_peak_tflops": probed, + "compute_peak_source": "on-node MFMA issue-rate probe (hardware roof)", + } ach = _resolve_achievable_tflops(gpu_type, precision_tag) if ach > 0: return { @@ -1858,6 +1979,7 @@ def compute_roofline_from_perfmodel( osl: int, num_gpus: int = 1, precision_tag: str = "bf16", + clocks: "EffectiveClocks | None" = None, ) -> "PerfModelBreakdown | None": """Bottom-up decode + prefill roofline using inlined GEMM/SDPA formulas. @@ -1877,6 +1999,7 @@ def compute_roofline_from_perfmodel( osl: Output sequence length. num_gpus: Number of GPUs (tensor-parallel degree). precision_tag: Precision key for the achievable TFLOPS lookup. + clocks: Measured effective clocks; ``None`` leaves the peak unscaled. Returns: The per-op ``PerfModelBreakdown``, or ``None`` when model metadata is @@ -1890,10 +2013,17 @@ def compute_roofline_from_perfmodel( if spec is None: return None - bw_gbps = spec["hbm_bw_gbps"] * max(num_gpus, 1) + per_gpu_gbps, _ = effective_hbm_bw_gbps(gpu_type, spec["hbm_bw_gbps"], active_gpus=num_gpus) + bw_gbps = per_gpu_gbps * max(num_gpus, 1) bw_bps = bw_gbps * 1e9 tag = (precision_tag or "bf16").strip().lower() - f_peak_tflops = _resolve_achievable_tflops(gpu_type, tag) * max(num_gpus, 1) + # Keep the achievable-table-only lookup (a miss must still degrade to the + # legacy path), but scale it to the clock the workload actually sustained. + f_peak_tflops = ( + _resolve_achievable_tflops(gpu_type, tag) + * max(num_gpus, 1) + * sclk_derate_factor(gpu_type, clocks, convention="achievable") + ) if f_peak_tflops <= 0: return None f_peak = f_peak_tflops * 1e12 diff --git a/src/hyperloom/orchestrator/kernel/roofline_effective.py b/src/hyperloom/orchestrator/kernel/roofline_effective.py new file mode 100644 index 0000000000..ca8d985c80 --- /dev/null +++ b/src/hyperloom/orchestrator/kernel/roofline_effective.py @@ -0,0 +1,490 @@ +# SPDX-FileCopyrightText: 2026 Advanced Micro Devices, Inc. +# SPDX-License-Identifier: MIT + +"""Effective-frequency and achievable-bandwidth derating for the roofline ceiling. + +The vendor compute peaks in ``roofline_ceiling.HW_SPECS`` are the architectural +product ``CUs x FLOPs_per_clk_per_CU x f_boost``. Both published tables invert +exactly at the peak *boost* engine clock:: + + MI355X 2516.6e12 / (256 CU * 2.40 GHz) = 4096 FLOPs/clk/CU + MI300X 1307.4e12 / (304 CU * 2.10 GHz) = 2048 FLOPs/clk/CU + +A serving workload does not sustain boost: engine clock settles at whatever the +power/CAC/thermal controller allows. Compute throughput is linear in engine +clock, so a ceiling anchored at boost overstates the reachable compute roof by +exactly the ratio of sustained to boost clock. + +Memory is a different mechanism and is deliberately NOT clock-scaled here. On +MI300-series parts mclk exposes a single DPM state (MI355X reports only +``2000Mhz``, and sampling confirms 2000 MHz with zero variance under load), so +memory clock does not droop under power constraints. The gap between vendor peak +HBM bandwidth and what a kernel actually attains is access efficiency -- row- +buffer locality, read/write turnaround, access pattern -- and is modelled as a +separate multiplicative efficiency. + +How large each correction actually is, measured on an 8-GPU MI355X node: + +* Engine clock holds near boost. Under 4 minutes of sustained 8-GPU bf16 GEMM + the mean sclk was 2379 MHz (99.1% of the 2400 MHz boost) at 1277 W against a + 1400 W cap, and 2372 MHz (98.8%) under sustained memory-bound load. So on this + part the compute derate is a small correction, not a large one. +* Bandwidth is the large error. Streaming reads reach ~89% of the vendor peak, + and the decode ceiling is memory-bound, so this is the term that actually + moves the reported roof. + +The clock derate is kept regardless: it costs nothing when clocks are healthy, +and it is the only thing that will catch a genuinely power- or thermally-limited +node, where a boost-anchored ceiling would silently overstate the roof. + +Both derates are no-ops until they have inputs: absent clock telemetry the +compute factor is ``1.0``, and the bandwidth efficiency defaults to ``1.0`` +(uncalibrated) so the ceiling matches its historical value until a measured +figure is supplied. +""" + +from __future__ import annotations + +import os +from dataclasses import dataclass +from typing import Any + +from .hw_probe import probe_hbm_bandwidth_gb_per_sec + +#: Environment override for the achievable-HBM-bandwidth efficiency, applied to +#: every GPU type. Accepts a fraction in ``(0, 1]``. +_BW_EFFICIENCY_ENV = "HYPERLOOM_ROOFLINE_HBM_BW_EFFICIENCY" + +#: Lower bound on a believable sustained/reference clock ratio. A measured +#: effective clock below this fraction of the reference is treated as bad +#: telemetry (e.g. a sampler that caught only idle ticks) rather than a real +#: operating point, and the derate degrades to a no-op. +_MIN_SCLK_RATIO = 0.1 + + +@dataclass(frozen=True) +class GpuFreqSpec: + """Per-GPU clock references and achievable-bandwidth efficiency. + + Attributes: + boost_sclk_mhz (float): Peak engine clock. This is the clock the vendor + dense ``peak_tflops`` table is derived from (verified by inverting + the published TFLOPS against CU count). + ref_sclk_mhz (float): Engine clock the max-achievable TFLOPS table was + measured at. Defaults to ``boost_sclk_mhz``: the measurement clock + for the TraceLens arch figures is not recorded upstream, and + assuming boost keeps the derate conservative (it can only reduce the + ceiling, never inflate it). Correct this once the real measurement + clock is known, otherwise a sub-boost measurement is double-counted. + hbm_bw_efficiency (float): Achievable / peak HBM bandwidth. ``1.0`` + means uncalibrated -- the ceiling keeps its historical vendor-peak + value. Override globally via ``HYPERLOOM_ROOFLINE_HBM_BW_EFFICIENCY``. + """ + + boost_sclk_mhz: float + ref_sclk_mhz: float = 0.0 + hbm_bw_efficiency: float = 1.0 + + def reference_sclk(self, convention: str) -> float: + """Reference clock the compute peak for *convention* is anchored at. + + Args: + convention: ``"achievable"`` for the sustained-TFLOPS table, + anything else for the vendor dense peak. + + Returns: + The reference engine clock in MHz. + """ + if convention == "achievable" and self.ref_sclk_mhz > 0: + return self.ref_sclk_mhz + return self.boost_sclk_mhz + + +#: Boost clocks are the published peak engine clocks, cross-checked by +#: inverting the vendor ``peak_tflops`` against the CU counts in +#: ``inference_optimizer.gpu_types`` (both land exactly on the architectural +#: MFMA rate, confirming the tables are boost-anchored). +#: +#: ``hbm_bw_efficiency`` is measured, not estimated, and only for parts we have +#: run on. Unmeasured parts stay at 1.0 so their ceiling keeps its historical +#: value rather than inheriting another part's number. +#: +#: MI355X, measured on an 8-GPU gfx950 node (ROCm 7.2.4) with a fully coalesced +#: non-temporal streaming-read kernel over a 16 GiB buffer: +#: +#: single GPU 7133 GB/s 89.2% of the 8000 GB/s vendor peak +#: all 8 GPUs loaded 7102-7156 GB/s 88.8-89.5% +#: +#: The per-GPU figure does not degrade under full load: each GPU owns its HBM +#: stacks, so there is no shared path to contend for. Establishing that needed a +#: synchronized start across the eight processes -- allowed to free-run, they +#: finish allocation at different moments and the aggregate reads low, which is +#: what an earlier unsynchronized attempt here reported. +#: +#: The decode roofline counts read traffic (weights + KV), so the streaming-read +#: figure is the right one; copy/triad (~61%) model write-heavy traffic this +#: model does not have. +#: +#: Cross-checked against a real run rather than only a microbenchmark: the +#: 095726Z MoE decode session (see ``test_roofline_ceiling``) measured 6244 tok/s +#: at TP=1, which back-solves to 6096 GB/s of actual traffic -- 76.2% of vendor +#: peak, comfortably under the 89% attainable ceiling. +#: +#: Prefer erring loose over tight: a ceiling a real workload can exceed is worse +#: than a slightly generous one, since ``within%`` above 100% is meaningless and +#: would discredit the metric. +_GPU_FREQ_SPECS: dict[str, GpuFreqSpec] = { + # CDNA3, 304 CU @ 2100 MHz -> 2048 FLOPs/clk/CU bf16. Bandwidth efficiency + # not yet measured on this part. + "mi300x": GpuFreqSpec(boost_sclk_mhz=2100.0), + "mi308x": GpuFreqSpec(boost_sclk_mhz=2100.0), + "mi325x": GpuFreqSpec(boost_sclk_mhz=2100.0), + # CDNA4, 256 CU @ 2400 MHz -> 4096 FLOPs/clk/CU bf16. + "mi355x": GpuFreqSpec(boost_sclk_mhz=2400.0, hbm_bw_efficiency=0.89), +} + + +@dataclass(frozen=True) +class EffectiveClocks: + """Engine/memory clocks a benchmark actually ran at. + + Attributes: + sclk_mhz (float): Mean sustained engine clock over the benchmark window; + ``0`` when unmeasured. + mclk_mhz (float): Mean memory clock; recorded for provenance only, since + the memory roof is not clock-scaled. + samples (int): Number of telemetry samples behind the means. + """ + + sclk_mhz: float = 0.0 + mclk_mhz: float = 0.0 + samples: int = 0 + + @property + def measured(self) -> bool: + """Whether a usable engine-clock measurement is present. + + Returns: + ``True`` when at least one sample yielded a positive engine clock. + """ + return self.sclk_mhz > 0 and self.samples > 0 + + +#: Minimum GPU utilization for a sample to count toward the effective clock. +#: The harvest window spans server start-up and inter-phase gaps, whose idle +#: ticks sit at a low DPM state (an idle MI355X reports ~95 MHz against a +#: 2400 MHz boost). Averaging those in would understate the sustained clock and derate +#: the ceiling too far, which inflates ``within%`` -- the opposite of the bug +#: this module exists to fix. +_MIN_ACTIVE_UTIL_PCT = 5.0 + + +def _to_float(value: Any) -> float | None: + """Coerce a telemetry field to ``float``. + + Args: + value: Raw sample value. + + Returns: + The float value, or ``None`` when absent or unparseable. + """ + if value is None or isinstance(value, bool): + return None + try: + return float(value) + except (TypeError, ValueError): + return None + + +def effective_clocks_from_samples( + samples: Any, + *, + min_util_pct: float = _MIN_ACTIVE_UTIL_PCT, +) -> EffectiveClocks: + """Mean sustained clocks over the *active* samples of a benchmark window. + + Samples below *min_util_pct* utilization are excluded so start-up and idle + ticks do not drag the mean down. When no sample carries utilization the + filter is skipped rather than dropping everything, so older telemetry still + yields a usable figure. + + Args: + samples: Iterable of flat ``gpu_monitor`` sample dicts. + min_util_pct: Utilization floor for a sample to count as active. + + Returns: + The measured ``EffectiveClocks``; unmeasured when no usable sample + carried an engine clock. + """ + if not isinstance(samples, (list, tuple)): + return EffectiveClocks() + rows = [s for s in samples if isinstance(s, dict)] + if not rows: + return EffectiveClocks() + + def _sclk(sample: dict[str, Any]) -> float | None: + """Engine clock of one sample, tolerating either key spelling.""" + return _to_float(sample.get("clock_mhz")) or _to_float(sample.get("sclk_mhz")) + + active = [s for s in rows if (_to_float(s.get("gpu_util_pct")) or 0.0) >= min_util_pct] + # No utilization recorded anywhere -> keep every sample rather than none. + if not active and not any(_to_float(s.get("gpu_util_pct")) is not None for s in rows): + active = rows + if not active: + return EffectiveClocks() + + sclks = [v for v in (_sclk(s) for s in active) if v is not None and v > 0] + if not sclks: + return EffectiveClocks() + mclks = [v for v in (_to_float(s.get("mclk_mhz")) for s in active) if v is not None and v > 0] + return EffectiveClocks( + sclk_mhz=sum(sclks) / len(sclks), + mclk_mhz=(sum(mclks) / len(mclks)) if mclks else 0.0, + samples=len(sclks), + ) + + +def effective_clocks_from_report(report: Any) -> EffectiveClocks: + """Effective clocks from a ``benchmark_report.json`` mapping. + + Reads the ``gpu_monitor`` block written by the sampler harvest, accepting + either the list-of-samples or single-sample shape. + + Args: + report: Parsed benchmark report. + + Returns: + The measured ``EffectiveClocks``, unmeasured when absent. + """ + if not isinstance(report, dict): + return EffectiveClocks() + gm = report.get("gpu_monitor") + if isinstance(gm, dict): + gm = [gm] + return effective_clocks_from_samples(gm) + + +def effective_clocks_from_entry(entry: Any) -> EffectiveClocks: + """Effective clocks recorded on a measurement / state arm entry. + + Reads the fields stamped by the benchmark-result normalizer, so the ceiling + does not have to rediscover ``benchmark_report.json`` from state. + + Args: + entry: A measurement or state arm mapping (``last_baseline``, + ``current_best``, ...). + + Returns: + The recorded ``EffectiveClocks``, unmeasured when absent. + """ + if not isinstance(entry, dict): + return EffectiveClocks() + sclk = _to_float(entry.get("effective_sclk_mhz")) + if sclk is None or sclk <= 0: + return EffectiveClocks() + samples = _to_float(entry.get("effective_clock_samples")) or 0.0 + return EffectiveClocks( + sclk_mhz=sclk, + mclk_mhz=_to_float(entry.get("effective_mclk_mhz")) or 0.0, + # A recorded clock with no sample count still describes a real run; + # floor at 1 so it is not discarded as unmeasured. + samples=max(int(samples), 1), + ) + + +def resolve_effective_clocks_from_state(state: Any, *, arm: str | None = None) -> EffectiveClocks: + """Effective clocks for the arm a roofline ceiling is being built for. + + Mirrors the ceiling's own arm selection: the optimized arm when it carries + clocks, otherwise baseline. Never raises; an unmeasured result leaves the + ceiling boost-anchored exactly as before. + + Args: + state: Shared run state carrying ``last_baseline`` / ``current_best``. + arm: Pins the source arm; ``None`` prefers ``current_best``. + + Returns: + The resolved ``EffectiveClocks``, unmeasured when no arm recorded any. + """ + baseline = getattr(state, "last_baseline", None) + if arm == "baseline": + return effective_clocks_from_entry(baseline) + current = effective_clocks_from_entry(getattr(state, "current_best", None)) + if current.measured: + return current + return effective_clocks_from_entry(baseline) + + +def resolve_freq_spec(gpu_type: str | None) -> GpuFreqSpec | None: + """Look up the frequency spec for *gpu_type* (case-insensitive). + + Args: + gpu_type: GPU type key. + + Returns: + The ``GpuFreqSpec``, or ``None`` for an unknown GPU. + """ + return _GPU_FREQ_SPECS.get((gpu_type or "").strip().lower()) + + +def _env_bw_efficiency() -> float: + """Read the bandwidth-efficiency override from the environment. + + Returns: + The override in ``(0, 1]``, or ``0.0`` when unset or unparseable. + """ + raw = os.environ.get(_BW_EFFICIENCY_ENV, "") + if not raw: + return 0.0 + try: + value = float(raw) + except (TypeError, ValueError): + return 0.0 + return value if 0.0 < value <= 1.0 else 0.0 + + +def hbm_bw_efficiency(gpu_type: str | None) -> float: + """Achievable / peak HBM bandwidth for *gpu_type*. + + The environment override wins over the per-GPU table so a calibrated figure + can be applied without a code change. Unknown GPUs and uncalibrated entries + return ``1.0``, leaving the memory roof at its vendor-peak value. + + Args: + gpu_type: GPU type key. + + Returns: + A derating fraction in ``(0, 1]``. + """ + override = _env_bw_efficiency() + if override > 0: + return override + spec = resolve_freq_spec(gpu_type) + if spec is None: + return 1.0 + eff = spec.hbm_bw_efficiency + return eff if 0.0 < eff <= 1.0 else 1.0 + + +def effective_hbm_bw_gbps( + gpu_type: str | None, + peak_gbps: float, + *, + active_gpus: int = 1, +) -> tuple[float, str]: + """Per-GPU achievable HBM bandwidth, preferring measurement over the table. + + Precedence: an on-node streaming-read probe, then the vendor peak scaled by + the calibrated efficiency (itself ``1.0``, i.e. the raw vendor peak, on + parts nobody has measured). + + The probe is preferred because it reports an absolute GB/s rather than a + fraction of a theoretical peak, and the theoretical peak is the shakier + input: it cannot be derived at runtime, since the DDR multiplier implied by + ``hipDeviceProp_t`` is wrong for HBM3E by a factor of two. + + Args: + gpu_type: GPU type key. + peak_gbps: Vendor peak per-GPU bandwidth from the spec table. + active_gpus: How many GPUs the workload loads concurrently. + + Returns: + ``(gb_per_sec, source)`` where source is ``"probe"`` or + ``"vendor_peak_x_efficiency"``. + """ + probed = probe_hbm_bandwidth_gb_per_sec(gpu_type=gpu_type, active_gpus=active_gpus) + if probed is not None and probed > 0: + return probed, "probe" + return peak_gbps * hbm_bw_efficiency(gpu_type), "vendor_peak_x_efficiency" + + +def sclk_derate_factor( + gpu_type: str | None, + clocks: EffectiveClocks | None, + *, + convention: str = "achievable", +) -> float: + """Ratio of sustained to reference engine clock, for scaling a compute peak. + + Compute throughput is linear in engine clock, so the reachable compute roof + scales by ``f_effective / f_reference``. The result is clamped to ``1.0``: + the reference is a boost (or boost-assumed) clock, so a measurement above it + means the reference is wrong, and inflating a ceiling on bad telemetry is + worse than leaving it alone. + + Args: + gpu_type: GPU type key, for the reference clock. + clocks: Measured clocks; ``None`` or unmeasured yields ``1.0``. + convention: Which compute-peak table the factor will scale -- + ``"achievable"`` or ``"vendor"``. + + Returns: + A factor in ``(0, 1]``; exactly ``1.0`` when the derate cannot be + applied, so callers degrade to their historical ceiling. + """ + if clocks is None or not clocks.measured: + return 1.0 + spec = resolve_freq_spec(gpu_type) + if spec is None: + return 1.0 + reference = spec.reference_sclk(convention) + if reference <= 0: + return 1.0 + ratio = clocks.sclk_mhz / reference + if ratio < _MIN_SCLK_RATIO: + # Implausibly low: treat as unusable telemetry rather than a real + # operating point. + return 1.0 + return min(ratio, 1.0) + + +def effective_clock_provenance( + gpu_type: str | None, + clocks: EffectiveClocks | None, + *, + convention: str = "achievable", + peak_gbps: float = 0.0, + active_gpus: int = 1, +) -> dict[str, Any]: + """Provenance describing how (and whether) the effective derates applied. + + Surfacing the measured clock, its reference, and both factors keeps a + derated ``within%`` interpretable next to an underated one. When a peak + bandwidth is supplied, the resolved memory roof and the layer that produced + it are reported too, since a probed roof and a table-derived one are not + interchangeable. + + Args: + gpu_type: GPU type key. + clocks: Measured clocks, if any. + convention: Compute-peak convention the factor is anchored to. + peak_gbps: Vendor peak per-GPU bandwidth; ``0`` omits the memory fields. + active_gpus: GPUs the workload loads, for selecting a probe measurement. + + Returns: + A provenance mapping describing the applied derates. + """ + spec = resolve_freq_spec(gpu_type) + factor = sclk_derate_factor(gpu_type, clocks, convention=convention) + bw_eff = hbm_bw_efficiency(gpu_type) + measured = clocks is not None and clocks.measured + memory: dict[str, Any] = {} + if peak_gbps > 0: + resolved_gbps, bw_source = effective_hbm_bw_gbps( + gpu_type, peak_gbps, active_gpus=active_gpus + ) + memory = { + "hbm_bw_gbps_effective": round(resolved_gbps, 1), + "hbm_bw_source": bw_source, + "hbm_bw_active_gpus": active_gpus, + } + return { + **memory, + "effective_sclk_mhz": round(clocks.sclk_mhz, 1) if measured else None, + "effective_mclk_mhz": (round(clocks.mclk_mhz, 1) if (clocks is not None and clocks.mclk_mhz > 0) else None), + "effective_clock_samples": clocks.samples if clocks is not None else 0, + "reference_sclk_mhz": (spec.reference_sclk(convention) if spec is not None else None), + "sclk_derate_factor": round(factor, 4), + "hbm_bw_efficiency": round(bw_eff, 4), + "effective_derate_source": ("measured_telemetry" if measured else "unmeasured_no_derate"), + } diff --git a/src/hyperloom/orchestrator/kernel/roofline_snapshot.py b/src/hyperloom/orchestrator/kernel/roofline_snapshot.py index 6689b75a96..443ee0eb58 100644 --- a/src/hyperloom/orchestrator/kernel/roofline_snapshot.py +++ b/src/hyperloom/orchestrator/kernel/roofline_snapshot.py @@ -222,13 +222,23 @@ def _compute_within_and_gap( return within, round(100.0 - within, 2) -def attach_perfmodel_breakdown(snapshot: dict[str, Any], state: Any, *, arm: str) -> None: +def attach_perfmodel_breakdown( + snapshot: dict[str, Any], + state: Any, + *, + arm: str, + clocks: Any = None, +) -> None: """Add ``roofline_provenance`` (+ ``perfmodel_breakdown`` when the PerfModel succeeds) for *arm*. - Best-effort and in place: any failure leaves *snapshot* untouched. + Best-effort and in place: any failure leaves *snapshot* untouched. *clocks* + is an optional ``EffectiveClocks`` for the run being modelled; when given, + the ceiling is anchored to the sustained engine clock instead of boost and + the provenance records the applied derates. """ try: from .roofline_ceiling import ( + HW_SPECS, apply_runtime_dtype, compute_roofline_from_perfmodel, load_model_meta, @@ -236,8 +246,14 @@ def attach_perfmodel_breakdown(snapshot: dict[str, Any], state: Any, *, arm: str resolve_runtime_dtype, resolve_runtime_workload, ) + from .roofline_effective import ( + effective_clock_provenance, + resolve_effective_clocks_from_state, + ) runtime = resolve_runtime_workload(state, arm=arm) + if clocks is None: + clocks = resolve_effective_clocks_from_state(state, arm=arm) meta = load_model_meta(runtime.model_path, precision_hint=runtime.precision) if meta is None: return @@ -252,10 +268,22 @@ def attach_perfmodel_breakdown(snapshot: dict[str, Any], state: Any, *, arm: str osl=runtime.osl, num_gpus=runtime.tp, precision_tag=compute_precision_tag, + clocks=clocks, ) + peak_provenance = resolve_compute_peak_provenance( + runtime.gpu_type, compute_precision_tag, clocks + ) + gpu_spec = HW_SPECS.get((runtime.gpu_type or "").strip().lower()) or {} snapshot["roofline_provenance"] = { "formula": "perfmodel" if pm_bd is not None else "legacy", - **resolve_compute_peak_provenance(runtime.gpu_type, compute_precision_tag), + **peak_provenance, + **effective_clock_provenance( + runtime.gpu_type, + clocks, + convention=str(peak_provenance.get("compute_peak_convention") or "achievable"), + peak_gbps=float(gpu_spec.get("hbm_bw_gbps") or 0.0), + active_gpus=runtime.tp, + ), "runtime_weight_dtype": rt.weight_dtype_tag, "runtime_weight_dtype_bytes": rt.weight_dtype_bytes, "runtime_activation_dtype_bytes": rt.activation_dtype_bytes,