From a948565ebde8a956094fa4815e3731bfe6363920 Mon Sep 17 00:00:00 2001 From: Wenyao Gao Date: Wed, 5 Aug 2026 11:50:54 -0700 Subject: [PATCH] =?UTF-8?q?[4/4]=20[Power]=20feat:=20strict=20AMD=20gpu=20?= =?UTF-8?q?monitor=20lifecycle=20for=20MI355X=20/=20MI355X=20=E5=8D=95?= =?UTF-8?q?=E8=8A=82=E7=82=B9=E5=8A=9F=E7=8E=87=E7=9B=91=E6=8E=A7=E4=B8=A5?= =?UTF-8?q?=E6=A0=BC=E7=94=9F=E5=91=BD=E5=91=A8=E6=9C=9F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Rebuild the stacked-PR history while preserving the merged source tree. / 在保留合并后代码树的前提下重建堆叠 PR 历史。 --- .github/workflows/benchmark-tmpl.yml | 9 +- benchmarks/benchmark_lib.sh | 81 ++++++-- utils/aggregate_power.py | 138 ++++++++++++ utils/process_result.py | 1 + utils/test_aggregate_power.py | 300 +++++++++++++++++++++++++++ utils/test_process_result.py | 148 +++++++++++++ 6 files changed, 654 insertions(+), 23 deletions(-) diff --git a/.github/workflows/benchmark-tmpl.yml b/.github/workflows/benchmark-tmpl.yml index d49d1a6081..6c4fe50fe5 100644 --- a/.github/workflows/benchmark-tmpl.yml +++ b/.github/workflows/benchmark-tmpl.yml @@ -362,7 +362,11 @@ jobs: uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 with: name: ${{ inputs.eval-only && 'eval_gpu_metrics_' || 'gpu_metrics_' }}${{ env.RESULT_FILENAME }} - path: gpu_metrics.csv + path: | + gpu_metrics.csv + gpu_metrics_energy_start.csv + gpu_metrics_energy_end.csv + gpu_metrics_identity.json if-no-files-found: ignore - name: Upload power audit bundle @@ -374,6 +378,9 @@ jobs: ${{ env.RESULT_FILENAME }}.json agg_${{ env.RESULT_FILENAME }}.json gpu_metrics.csv + gpu_metrics_energy_start.csv + gpu_metrics_energy_end.csv + gpu_metrics_identity.json power_validation_${{ env.RESULT_FILENAME }}.json if-no-files-found: ignore diff --git a/benchmarks/benchmark_lib.sh b/benchmarks/benchmark_lib.sh index ba6d23b10d..6979ca102f 100644 --- a/benchmarks/benchmark_lib.sh +++ b/benchmarks/benchmark_lib.sh @@ -110,6 +110,7 @@ unset _benchmark_caller GPU_MONITOR_PID="" GPU_MONITOR_VENDOR="" +GPU_MONITOR_INTERVAL=1 GPU_METRICS_CSV="${GPU_METRICS_CSV:-gpu_metrics.csv}" NVIDIA_GPU_MONITOR_QUERY="timestamp,index,power.draw,temperature.gpu,clocks.current.sm,clocks.current.memory,utilization.gpu,utilization.memory" export GPU_METRICS_CSV @@ -130,6 +131,7 @@ start_gpu_monitor() { done GPU_METRICS_CSV="$output" + GPU_MONITOR_INTERVAL="$interval" export GPU_METRICS_CSV if command -v nvidia-smi &>/dev/null; then @@ -141,10 +143,18 @@ start_gpu_monitor() { elif command -v amd-smi &>/dev/null; then GPU_MONITOR_VENDOR="amd" # Use amd-smi native watch mode (-w) which includes timestamps automatically. - # Pipe through awk to: skip preamble lines, keep first CSV header, skip repeated headers. - amd-smi metric -p -c -t -u -w "$interval" --csv 2>/dev/null \ - | awk '/^timestamp,/{if(!h){print;h=1};next} h{print}' > "$output" & + # PYTHONUNBUFFERED defeats the tool's own stdout block buffering (amd-smi is + # Python; measured on MI355X: trailing ticks were lost at kill without it). + # Pipe through awk to: skip preamble lines, keep first CSV header, skip repeated + # headers, and flush every row so killing the pipe cannot discard buffered samples. + PYTHONUNBUFFERED=1 amd-smi metric -p -c -t -u -w "$interval" --csv 2>/dev/null \ + | awk '/^timestamp,/{if(!h){print;h=1};next} h{print;fflush()}' > "$output" & GPU_MONITOR_PID=$! + # Hardware energy-accumulator + identity snapshots; the end-side twin in + # stop_gpu_monitor lets auditors cross-check the integrated energy + # against the accumulator delta. + _write_amd_smi_sidecar "${output%.csv}_energy_start.csv" metric -E --csv + _write_amd_smi_sidecar "${output%.csv}_identity.json" static --json echo "[GPU Monitor] Started AMD (PID=$GPU_MONITOR_PID, interval=${interval}s, output=$output)" else GPU_MONITOR_VENDOR="" @@ -156,34 +166,32 @@ start_gpu_monitor() { # Stop the background GPU monitor and report file size. stop_gpu_monitor() { if [[ -n "$GPU_MONITOR_PID" ]] && kill -0 "$GPU_MONITOR_PID" 2>/dev/null; then + # benchmark_end_time_unix is recorded shortly before the benchmark + # process exits, so the stream must cover one more sample past it for + # deterministic boundary interpolation. NVIDIA appends a one-shot + # post-exit sample below; amd-smi one-shot CSV has no timestamp column, + # so the AMD path instead lets the watch stream emit final ticks before + # the kill. Two extra intervals: amd-smi stamps integer seconds, so a + # tick in the same second as the window end still fails bracketing — + # the stream needs a tick at the NEXT whole second (measured on MI355X: + # end=...153.325 vs last sample ...153.0). + if [[ "$GPU_MONITOR_VENDOR" == "amd" ]]; then + sleep $(( ${GPU_MONITOR_INTERVAL:-1} + 2 )) + fi kill "$GPU_MONITOR_PID" 2>/dev/null wait "$GPU_MONITOR_PID" 2>/dev/null || true - # benchmark_end_time_unix is recorded shortly before the benchmark - # process exits. For the NVIDIA PR1 canary, append one post-exit sample - # so boundary interpolation is deterministic even when the run ends - # between 1 Hz monitor ticks. AMD one-shot CSV schemas vary by amd-smi - # version, so strict AMD lifecycle validation remains follow-up work. case "$GPU_MONITOR_VENDOR" in nvidia) - local append_final_nvidia_sample=true - local repaired_metrics="${GPU_METRICS_CSV}.repair.$$" - if [[ -s "$GPU_METRICS_CSV" ]] && - ! tail -c 1 "$GPU_METRICS_CSV" | grep -q '^$'; then - if sed '$d' "$GPU_METRICS_CSV" > "$repaired_metrics" && - mv "$repaired_metrics" "$GPU_METRICS_CSV"; then - echo "[GPU Monitor] Dropped truncated trailing sample" - else - rm -f "$repaired_metrics" - append_final_nvidia_sample=false - echo "[GPU Monitor] Warning: could not repair truncated trailing sample" >&2 - fi - fi - if [[ "$append_final_nvidia_sample" == true ]]; then + if _repair_truncated_gpu_metrics_tail; then nvidia-smi --query-gpu="$NVIDIA_GPU_MONITOR_QUERY" \ --format=csv,noheader >> "$GPU_METRICS_CSV" 2>/dev/null || echo "[GPU Monitor] Warning: final NVIDIA sample failed" >&2 fi ;; + amd) + _repair_truncated_gpu_metrics_tail || true + _write_amd_smi_sidecar "${GPU_METRICS_CSV%.csv}_energy_end.csv" metric -E --csv + ;; esac echo "[GPU Monitor] Stopped (PID=$GPU_MONITOR_PID)" if [[ -f "$GPU_METRICS_CSV" ]]; then @@ -196,6 +204,35 @@ stop_gpu_monitor() { GPU_MONITOR_VENDOR="" } +# Drop a partial trailing row left behind when the monitor dies mid-write. +# Returns non-zero when a truncated row was detected but could not be removed. +_repair_truncated_gpu_metrics_tail() { + local repaired_metrics="${GPU_METRICS_CSV}.repair.$$" + if [[ -s "$GPU_METRICS_CSV" ]] && + ! tail -c 1 "$GPU_METRICS_CSV" | grep -q '^$'; then + if sed '$d' "$GPU_METRICS_CSV" > "$repaired_metrics" && + mv "$repaired_metrics" "$GPU_METRICS_CSV"; then + echo "[GPU Monitor] Dropped truncated trailing sample" + else + rm -f "$repaired_metrics" + echo "[GPU Monitor] Warning: could not repair truncated trailing sample" >&2 + return 1 + fi + fi + return 0 +} + +# Write one best-effort amd-smi snapshot; remove the file rather than keep a +# partial one when the invocation fails. +_write_amd_smi_sidecar() { + local out="$1" + shift + if ! amd-smi "$@" > "$out" 2>/dev/null; then + rm -f "$out" + echo "[GPU Monitor] Warning: amd-smi $1 sidecar failed" >&2 + fi +} + # Block until the GPUs have released a prior job's memory before starting a run. # Polls rocm-smi VRAM% every 10s for up to 15 minutes; succeeds once the busiest # GPU is at <=10% VRAM, otherwise returns 1 so the caller aborts rather than diff --git a/utils/aggregate_power.py b/utils/aggregate_power.py index a269fc5091..24db2cd829 100644 --- a/utils/aggregate_power.py +++ b/utils/aggregate_power.py @@ -35,6 +35,7 @@ _INTEGRATION_METHOD = "per_device_trapezoidal_with_linear_boundary_interpolation" _DEFAULT_MAX_SAMPLE_GAP_S = 3.0 +_ACCUMULATOR_TOLERANCE = 0.05 _POWER_METRIC_KEYS = { "avg_power_w", "avg_total_gpu_power_w", @@ -501,6 +502,135 @@ def integrate_power( ) +def _read_energy_snapshot(path: Path) -> dict[str, float] | None: + """Map GPU id to accumulator joules from one ``amd-smi metric -E`` snapshot.""" + with path.open("r", newline="", encoding="utf-8", errors="replace") as f: + reader = csv.DictReader(f, skipinitialspace=True) + header = [column.strip() for column in (reader.fieldnames or [])] + reader.fieldnames = header + if "gpu" not in header or "total_energy_consumption" not in header: + return None + energy_by_gpu: dict[str, float] = {} + for row in reader: + gpu_id = (row.get("gpu") or "").strip() + if not gpu_id: + continue + try: + energy_j = float((row.get("total_energy_consumption") or "").strip()) + except ValueError: + continue + if math.isfinite(energy_j): + energy_by_gpu[gpu_id] = energy_j + return energy_by_gpu or None + + +def _stream_samples_by_gpu(csv_path: Path) -> dict[str, list[tuple[float, float]]]: + """Group every parseable (timestamp, watt) sample per GPU, sorted in time.""" + samples: dict[str, list[tuple[float, float]]] = {} + with csv_path.open("r", newline="", encoding="utf-8", errors="replace") as f: + reader = csv.DictReader(f, skipinitialspace=True) + header = [column.strip() for column in (reader.fieldnames or [])] + reader.fieldnames = header + timestamp_col, power_col, gpu_col = _detect_columns(header) + if not timestamp_col or not power_col or not gpu_col: + return {} + for row in reader: + timestamp = _parse_timestamp((row.get(timestamp_col) or "").strip()) + power = _parse_power((row.get(power_col) or "").strip()) + gpu_id = (row.get(gpu_col) or "").strip() + if timestamp is None or power is None or not gpu_id: + continue + if not math.isfinite(timestamp) or not math.isfinite(power): + continue + samples.setdefault(gpu_id, []).append((timestamp, power)) + return {gpu_id: sorted(values) for gpu_id, values in samples.items()} + + +def _trapezoid(samples: list[tuple[float, float]]) -> float: + """Integrate consecutive (timestamp, watt) samples with no window clipping.""" + return float( + sum( + (right_time - left_time) * (left_power + right_power) / 2.0 + for (left_time, left_power), (right_time, right_power) in zip( + samples, samples[1:] + ) + ) + ) + + +def cross_check_accumulator(csv_path: Path) -> dict | None: + """Compare the hardware energy-accumulator delta with the full stream integral. + + The snapshots bracket the whole monitor lifetime rather than one benchmark + window, so the comparison integrates the entire stream span: it validates the + sampling instrument, not a window's metrics. Advisory only — the result never + affects ``power_valid`` or the validation reason codes. + """ + start_path = csv_path.with_name(f"{csv_path.stem}_energy_start.csv") + end_path = csv_path.with_name(f"{csv_path.stem}_energy_end.csv") + start_exists = start_path.is_file() + end_exists = end_path.is_file() + if not start_exists and not end_exists: + return None + if not start_exists: + return {"available": False, "reason": "missing_start_snapshot"} + if not end_exists: + return {"available": False, "reason": "missing_end_snapshot"} + + start_energy = _read_energy_snapshot(start_path) + if start_energy is None: + return {"available": False, "reason": "unparseable_start_snapshot"} + end_energy = _read_energy_snapshot(end_path) + if end_energy is None: + return {"available": False, "reason": "unparseable_end_snapshot"} + + matched_gpu_ids = sorted(set(start_energy) & set(end_energy), key=_gpu_sort_key) + per_gpu_delta_j = { + gpu_id: end_energy[gpu_id] - start_energy[gpu_id] for gpu_id in matched_gpu_ids + } + unmatched_gpus = sorted(set(start_energy) ^ set(end_energy), key=_gpu_sort_key) + negative_delta_gpus = [ + gpu_id for gpu_id, delta in per_gpu_delta_j.items() if delta < 0 + ] + + stream = _stream_samples_by_gpu(csv_path) + integrated_gpu_ids = sorted(set(stream) & set(per_gpu_delta_j), key=_gpu_sort_key) + integrated_stream_j = float( + sum(_trapezoid(stream[gpu_id]) for gpu_id in integrated_gpu_ids) + ) + boundaries = [ + timestamp + for gpu_id in integrated_gpu_ids + for timestamp in (stream[gpu_id][0][0], stream[gpu_id][-1][0]) + ] + stream_span_s = max(boundaries) - min(boundaries) if boundaries else 0.0 + + accumulator_delta_j = float(sum(per_gpu_delta_j.values())) + relative_error: float | None = None + within_tolerance = False + if accumulator_delta_j > 0: + relative_error = ( + abs(accumulator_delta_j - integrated_stream_j) / accumulator_delta_j + ) + within_tolerance = ( + not negative_delta_gpus and relative_error <= _ACCUMULATOR_TOLERANCE + ) + + return { + "available": True, + "accumulator_delta_j": accumulator_delta_j, + "integrated_stream_j": integrated_stream_j, + "relative_error": relative_error, + "within_tolerance": within_tolerance, + "tolerance": _ACCUMULATOR_TOLERANCE, + "per_gpu_delta_j": per_gpu_delta_j, + "integrated_gpu_ids": integrated_gpu_ids, + "unmatched_gpus": unmatched_gpus, + "negative_delta_gpus": negative_delta_gpus, + "stream_span_s": stream_span_s, + } + + def _load_benchmark_data( bench_result_path: Path, ) -> tuple[BenchmarkData | None, list[str]]: @@ -646,6 +776,7 @@ def _validation_payload( power_valid: bool, reasons: list[str], metrics: dict[str, float], + accumulator_check: dict | None, ) -> dict: """Build the complete auditable power-validation sidecar payload.""" benchmark_window = None @@ -674,6 +805,7 @@ def _validation_payload( "per_gpu_max_sample_gap_s": integration.per_gpu_max_sample_gap_s, "per_gpu_energy_j": integration.per_gpu_energy_j, "device_issues": integration.device_issues, + "accumulator_check": accumulator_check, "metrics": { key: round(value, 6) for key, value in metrics.items() @@ -735,6 +867,11 @@ def run( metrics = {} print(f"[aggregate_power] Failed to patch {agg_result}: {exc}", file=sys.stderr) + try: + accumulator_check = cross_check_accumulator(csv_path) + except (OSError, csv.Error): + accumulator_check = {"available": False, "reason": "cross_check_error"} + try: _write_json_atomic( validation_result, @@ -746,6 +883,7 @@ def run( power_valid=power_valid, reasons=reasons, metrics=metrics, + accumulator_check=accumulator_check, ), ) except OSError as exc: diff --git a/utils/process_result.py b/utils/process_result.py index 085d4c74ba..79e378cd09 100644 --- a/utils/process_result.py +++ b/utils/process_result.py @@ -90,6 +90,7 @@ def record_power_internal_error( power_valid=False, reasons=reasons, metrics={}, + accumulator_check=None, ) validation_data["internal_error"] = { "type": type(error).__name__, diff --git a/utils/test_aggregate_power.py b/utils/test_aggregate_power.py index 7f4affc2cd..89e42034e8 100644 --- a/utils/test_aggregate_power.py +++ b/utils/test_aggregate_power.py @@ -10,6 +10,7 @@ - Whole-deployment power/energy/J-query/J-token metric semantics - Best-effort invalid results and REQUIRE_POWER strict failures - Atomic aggregate and validation artifacts + - Advisory energy-accumulator cross-check against the AMD sidecar snapshots """ from __future__ import annotations @@ -27,6 +28,7 @@ _parse_power, _parse_timestamp, aggregate_power, + cross_check_accumulator, integrate_power, main, patch_agg_result, @@ -76,6 +78,23 @@ def test_detect_columns_amd(): assert gpu == "gpu" +def test_detect_columns_amd_watch_mode_real_header(): + # AMDSMI 26.2.0 `metric -p -c -t -u -w 1 --csv` header (order-faithful + # subset, measured on MI355X): socket_power must win even though + # power_management also matches the power pattern later in the row. + header = [ + "timestamp", "gpu", "gfx_activity", "umc_activity", "mm_activity", + "vcn_activity", "jpeg_activity", "gfx_busy_inst_xcp_0", + "jpeg_busy_xcp_0", "vcn_busy_xcp_0", "socket_power", "gfx_voltage", + "soc_voltage", "mem_voltage", "throttle_status", "power_management", + "gfx_0_clk", "mem_0_clk", "edge", "hotspot", "mem", + ] + ts, pw, gpu = _detect_columns(header) + assert ts == "timestamp" + assert pw == "socket_power" + assert gpu == "gpu" + + def test_detect_columns_excludes_power_limit(): # power.limit must NOT be picked as the power column. header = ["timestamp", "index", "power.limit [W]", "power.draw [W]"] @@ -759,6 +778,8 @@ def test_run_emits_complete_whole_deployment_metric_contract(tmp_path: Path): assert audit["integration_method"] == ( "per_device_trapezoidal_with_linear_boundary_interpolation" ) + # No vendor accumulator sidecars: the advisory check stays absent. + assert audit["accumulator_check"] is None def test_run_best_effort_marks_invalid_power_and_preserves_benchmark(tmp_path: Path): @@ -875,6 +896,285 @@ def test_run_invalid_benchmark_denominator_is_auditable(tmp_path: Path): ] +# --------------------------------------------------------------------------- # +# Advisory hardware energy-accumulator cross-check +# --------------------------------------------------------------------------- # + +_ENERGY_SNAPSHOT_HEADER = ( + "gpu,socket_power,gfx_voltage,soc_voltage,mem_voltage," + "throttle_status,power_management,total_energy_consumption" +) + + +def _write_energy_snapshot(path: Path, energy_by_gpu: dict[str, float]) -> None: + """`amd-smi metric -E --csv` shape measured on MI355X, N/A cells included.""" + lines = [_ENERGY_SNAPSHOT_HEADER] + for gpu_id, energy_j in energy_by_gpu.items(): + lines.append(f"{gpu_id},238,N/A,N/A,N/A,N/A,ENABLED,{energy_j}") + path.write_text("\n".join(lines) + "\n", encoding="utf-8") + + +def _write_snapshot_pair( + csv: Path, + *, + start: dict[str, float], + end: dict[str, float], +) -> None: + """Sidecars named exactly as start_gpu_monitor/stop_gpu_monitor write them.""" + _write_energy_snapshot(csv.parent / "gpu_metrics_energy_start.csv", start) + _write_energy_snapshot(csv.parent / "gpu_metrics_energy_end.csv", end) + + +def _write_flat_stream(csv: Path, *, base: float, watts_by_gpu: dict[int, float]) -> None: + """11 samples at 1 Hz, so each GPU's full-stream span is exactly 10s.""" + _write_amd_csv( + csv, + [ + (base + offset, gpu, watts) + for gpu, watts in watts_by_gpu.items() + for offset in range(11) + ], + ) + + +def test_cross_check_accumulator_matches_full_stream_integral(tmp_path: Path): + csv = tmp_path / "gpu_metrics.csv" + _write_flat_stream(csv, base=1_700_000_000.0, watts_by_gpu={0: 500.0}) + _write_snapshot_pair(csv, start={"0": 1_000_000.0}, end={"0": 1_005_000.0}) + + result = cross_check_accumulator(csv) + + assert result is not None + assert result["available"] is True + assert result["accumulator_delta_j"] == pytest.approx(5_000.0) + assert result["integrated_stream_j"] == pytest.approx(5_000.0) + assert result["relative_error"] < 0.05 + assert result["within_tolerance"] is True + assert result["tolerance"] == 0.05 + assert result["per_gpu_delta_j"] == pytest.approx({"0": 5_000.0}) + assert result["integrated_gpu_ids"] == ["0"] + assert result["unmatched_gpus"] == [] + assert result["negative_delta_gpus"] == [] + assert result["stream_span_s"] == pytest.approx(10.0) + + +def test_cross_check_accumulator_flags_disagreement_beyond_tolerance(tmp_path: Path): + csv = tmp_path / "gpu_metrics.csv" + # 400 W sampled against a 5_000 J accumulator delta: the stream is 20% low. + _write_flat_stream(csv, base=1_700_000_000.0, watts_by_gpu={0: 400.0}) + _write_snapshot_pair(csv, start={"0": 1_000_000.0}, end={"0": 1_005_000.0}) + + result = cross_check_accumulator(csv) + + assert result["available"] is True + assert result["integrated_stream_j"] == pytest.approx(4_000.0) + assert result["relative_error"] == pytest.approx(0.2) + assert result["within_tolerance"] is False + + +def test_cross_check_accumulator_without_snapshots_returns_none(tmp_path: Path): + csv = tmp_path / "gpu_metrics.csv" + _write_flat_stream(csv, base=1_700_000_000.0, watts_by_gpu={0: 500.0}) + + assert cross_check_accumulator(csv) is None + + +def test_cross_check_accumulator_reports_missing_end_snapshot(tmp_path: Path): + csv = tmp_path / "gpu_metrics.csv" + _write_flat_stream(csv, base=1_700_000_000.0, watts_by_gpu={0: 500.0}) + _write_energy_snapshot(tmp_path / "gpu_metrics_energy_start.csv", {"0": 1_000_000.0}) + + assert cross_check_accumulator(csv) == { + "available": False, + "reason": "missing_end_snapshot", + } + + +def test_cross_check_accumulator_reports_unparseable_snapshot(tmp_path: Path): + csv = tmp_path / "gpu_metrics.csv" + _write_flat_stream(csv, base=1_700_000_000.0, watts_by_gpu={0: 500.0}) + _write_energy_snapshot(tmp_path / "gpu_metrics_energy_start.csv", {"0": 1_000_000.0}) + (tmp_path / "gpu_metrics_energy_end.csv").write_text( + "gpu,socket_power\n0,238\n", encoding="utf-8" + ) + + assert cross_check_accumulator(csv) == { + "available": False, + "reason": "unparseable_end_snapshot", + } + + +def test_cross_check_accumulator_flags_negative_delta(tmp_path: Path): + csv = tmp_path / "gpu_metrics.csv" + _write_flat_stream(csv, base=1_700_000_000.0, watts_by_gpu={0: 400.0, 1: 0.0}) + # GPU 1's counter wrapped. The summed delta still matches the stream, so a + # False verdict here can only come from the wrap itself. + _write_snapshot_pair( + csv, + start={"0": 1_000_000.0, "1": 2_000_000.0}, + end={"0": 1_005_000.0, "1": 1_999_000.0}, + ) + + result = cross_check_accumulator(csv) + + assert result["negative_delta_gpus"] == ["1"] + assert result["accumulator_delta_j"] == pytest.approx(4_000.0) + assert result["integrated_stream_j"] == pytest.approx(4_000.0) + assert result["relative_error"] == pytest.approx(0.0) + assert result["within_tolerance"] is False + + +def test_cross_check_accumulator_excludes_unmatched_gpu(tmp_path: Path): + csv = tmp_path / "gpu_metrics.csv" + # GPU 1 draws 10x GPU 0 but is absent from the end snapshot: counting it + # would blow the integral past tolerance. + _write_flat_stream(csv, base=1_700_000_000.0, watts_by_gpu={0: 500.0, 1: 5_000.0}) + _write_energy_snapshot( + tmp_path / "gpu_metrics_energy_start.csv", + {"0": 1_000_000.0, "1": 3_000_000.0}, + ) + _write_energy_snapshot( + tmp_path / "gpu_metrics_energy_end.csv", + {"0": 1_005_000.0}, + ) + + result = cross_check_accumulator(csv) + + assert result["unmatched_gpus"] == ["1"] + assert list(result["per_gpu_delta_j"]) == ["0"] + assert result["integrated_gpu_ids"] == ["0"] + assert result["accumulator_delta_j"] == pytest.approx(5_000.0) + assert result["integrated_stream_j"] == pytest.approx(5_000.0) + assert result["within_tolerance"] is True + + +def test_cross_check_accumulator_guards_zero_accumulator_delta(tmp_path: Path): + csv = tmp_path / "gpu_metrics.csv" + _write_flat_stream(csv, base=1_700_000_000.0, watts_by_gpu={0: 500.0}) + _write_snapshot_pair(csv, start={"0": 1_000_000.0}, end={"0": 1_000_000.0}) + + result = cross_check_accumulator(csv) + + assert result["accumulator_delta_j"] == pytest.approx(0.0) + assert result["relative_error"] is None + assert result["within_tolerance"] is False + + +def test_cross_check_accumulator_parses_real_amd_smi_snapshot(tmp_path: Path): + """Verbatim AMDSMI 26.2.0 rows: N/A cells, and an N/A accumulator reading.""" + csv = tmp_path / "gpu_metrics.csv" + _write_flat_stream(csv, base=1_700_000_000.0, watts_by_gpu={0: 500.0}) + (tmp_path / "gpu_metrics_energy_start.csv").write_text( + f"{_ENERGY_SNAPSHOT_HEADER}\n" + "0,238,N/A,N/A,N/A,N/A,ENABLED,178319501.682\n" + "1,241,N/A,N/A,N/A,N/A,ENABLED,178400000.000\n", + encoding="utf-8", + ) + (tmp_path / "gpu_metrics_energy_end.csv").write_text( + f"{_ENERGY_SNAPSHOT_HEADER}\n" + "0,240,N/A,N/A,N/A,N/A,ENABLED,178324501.682\n" + "1,239,N/A,N/A,N/A,N/A,ENABLED,N/A\n", + encoding="utf-8", + ) + + result = cross_check_accumulator(csv) + + assert result["available"] is True + assert result["per_gpu_delta_j"] == pytest.approx({"0": 5_000.0}) + assert result["unmatched_gpus"] == ["1"] + assert result["within_tolerance"] is True + + +def test_run_records_advisory_accumulator_check(tmp_path: Path): + base = 1_700_000_000.0 + csv = tmp_path / "gpu_metrics.csv" + _write_constant_window_samples( + csv, + start=base, + end=base + 10, + watts_per_gpu=500.0, + num_gpus=2, + ) + # The monitor lifetime spans 12s (one sample either side of the formal + # window), so the accumulator delta legitimately exceeds window energy. + _write_snapshot_pair( + csv, + start={"0": 1_000_000.0, "1": 2_000_000.0}, + end={"0": 1_006_000.0, "1": 2_006_000.0}, + ) + bench = tmp_path / "bench.json" + agg = tmp_path / "agg.json" + validation = tmp_path / "power_validation.json" + _write_bench_result( + bench, + start=base, + end=base + 10, + duration=10.0, + completed=10, + total_input=10_000, + total_output=2_000, + ) + agg.write_text(json.dumps({"hw": "mi355x"}), encoding="utf-8") + + exit_code = run(csv, bench, agg, expected_num_gpus=2, validation_result=validation) + + assert exit_code == 0 + patched = json.loads(agg.read_text()) + assert patched["power_valid"] == 1 + assert patched["total_gpu_energy_j"] == pytest.approx(10_000.0) + + audit = json.loads(validation.read_text()) + assert audit["power_valid"] is True + assert audit["reasons"] == [] + check = audit["accumulator_check"] + assert check["available"] is True + assert check["within_tolerance"] is True + assert check["accumulator_delta_j"] == pytest.approx(12_000.0) + assert check["integrated_stream_j"] == pytest.approx(12_000.0) + assert check["stream_span_s"] == pytest.approx(12.0) + + +def test_run_accumulator_mismatch_leaves_power_validity_untouched(tmp_path: Path): + base = 1_700_000_000.0 + csv = tmp_path / "gpu_metrics.csv" + _write_constant_window_samples( + csv, + start=base, + end=base + 10, + watts_per_gpu=500.0, + num_gpus=2, + ) + _write_snapshot_pair( + csv, + start={"0": 1_000_000.0, "1": 2_000_000.0}, + end={"0": 1_000_100.0, "1": 2_000_100.0}, + ) + bench = tmp_path / "bench.json" + agg = tmp_path / "agg.json" + validation = tmp_path / "power_validation.json" + _write_bench_result( + bench, + start=base, + end=base + 10, + duration=10.0, + completed=10, + total_input=10_000, + total_output=2_000, + ) + agg.write_text(json.dumps({"hw": "mi355x"}), encoding="utf-8") + + exit_code = run(csv, bench, agg, expected_num_gpus=2, validation_result=validation) + + assert exit_code == 0 + patched = json.loads(agg.read_text()) + assert patched["power_valid"] == 1 + assert patched["avg_power_w"] == pytest.approx(500.0) + audit = json.loads(validation.read_text()) + assert audit["power_valid"] is True + assert audit["reasons"] == [] + assert audit["accumulator_check"]["within_tolerance"] is False + + def test_cli_requires_expected_gpu_count(monkeypatch: pytest.MonkeyPatch): monkeypatch.setattr( sys, diff --git a/utils/test_process_result.py b/utils/test_process_result.py index dcc6af9f22..1142a117f8 100644 --- a/utils/test_process_result.py +++ b/utils/test_process_result.py @@ -1029,6 +1029,154 @@ def test_stop_gpu_monitor_drops_truncated_row_before_final_sample(self, tmp_path final_sample, ] + def test_stop_gpu_monitor_amd_waits_one_tick_and_snapshots_energy(self, tmp_path): + """AMD stop lets the watch stream bracket the window, then snapshots energy.""" + fake_bin = tmp_path / "bin" + fake_bin.mkdir() + args_log = tmp_path / "amd_args.txt" + fake_amd_smi = fake_bin / "amd-smi" + fake_amd_smi.write_text( + "#!/usr/bin/env bash\n" + f"printf '%s\\n' \"$*\" >> {str(args_log)!r}\n" + "printf 'gpu,total_energy_consumption\\n0,178319501.7\\n'\n" + ) + fake_amd_smi.chmod(0o755) + sleep_log = tmp_path / "sleep_args.txt" + contents = "timestamp,gpu,socket_power\n1785881113,0,238\n" + metrics = tmp_path / "gpu_metrics.csv" + metrics.write_text(contents) + benchmark_lib = Path(__file__).parents[1] / "benchmarks/benchmark_lib.sh" + script = f""" +source {str(benchmark_lib)!r} +kill() {{ return 0; }} +wait() {{ return 0; }} +sleep() {{ printf '%s\\n' "$1" > {str(sleep_log)!r}; }} +GPU_MONITOR_PID=999 +GPU_MONITOR_VENDOR=amd +GPU_MONITOR_INTERVAL=3 +GPU_METRICS_CSV={str(metrics)!r} +stop_gpu_monitor +""" + env = { + "PATH": f"{fake_bin}:/usr/bin:/bin", + "PYTHONDONTWRITEBYTECODE": "1", + } + + result = subprocess.run( + ["bash", "-c", script], + env=env, + capture_output=True, + text=True, + check=False, + ) + + assert result.returncode == 0, result.stderr + assert sleep_log.read_text().strip() == "5" + assert metrics.read_text() == contents + assert "metric -E --csv" in args_log.read_text() + energy_end = tmp_path / "gpu_metrics_energy_end.csv" + assert energy_end.read_text().startswith("gpu,total_energy_consumption") + + def test_stop_gpu_monitor_amd_drops_truncated_row_without_append(self, tmp_path): + """The AMD path repairs a partial trailing row but appends no sample.""" + fake_bin = tmp_path / "bin" + fake_bin.mkdir() + fake_amd_smi = fake_bin / "amd-smi" + fake_amd_smi.write_text( + "#!/usr/bin/env bash\n" + "printf 'gpu,total_energy_consumption\\n0,178319501.7\\n'\n" + ) + fake_amd_smi.chmod(0o755) + header = "timestamp,gpu,socket_power" + complete_sample = "1785881113,0,238" + metrics = tmp_path / "gpu_metrics.csv" + metrics.write_text(f"{header}\n{complete_sample}\n1785881114,0,2") + benchmark_lib = Path(__file__).parents[1] / "benchmarks/benchmark_lib.sh" + script = f""" +source {str(benchmark_lib)!r} +kill() {{ return 0; }} +wait() {{ return 0; }} +sleep() {{ return 0; }} +GPU_MONITOR_PID=999 +GPU_MONITOR_VENDOR=amd +GPU_METRICS_CSV={str(metrics)!r} +stop_gpu_monitor +""" + env = { + "PATH": f"{fake_bin}:/usr/bin:/bin", + "PYTHONDONTWRITEBYTECODE": "1", + } + + result = subprocess.run( + ["bash", "-c", script], + env=env, + capture_output=True, + text=True, + check=False, + ) + + assert result.returncode == 0, result.stderr + assert metrics.read_text().splitlines() == [header, complete_sample] + assert (tmp_path / "gpu_metrics_energy_end.csv").exists() + + def test_start_stop_gpu_monitor_amd_lifecycle(self, tmp_path): + """Watch rows survive the kill and both boundary snapshots are written.""" + fake_bin = tmp_path / "bin" + fake_bin.mkdir() + fake_amd_smi = fake_bin / "amd-smi" + fake_amd_smi.write_text( + "#!/usr/bin/env bash\n" + 'if [[ "$*" == *" -w "* ]]; then\n' + " echo \"'CTRL' + 'C' to stop watching output:\"\n" + " echo 'timestamp,gpu,socket_power,power_management'\n" + " for _ in $(seq 1 30); do\n" + " echo \"$(date +%s),0,238,ENABLED\"\n" + " echo 'timestamp,gpu,socket_power,power_management'\n" + " sleep 0.2\n" + " done\n" + 'elif [[ "$*" == *"metric -E --csv"* ]]; then\n' + " printf 'gpu,total_energy_consumption\\n0,178319501.7\\n'\n" + 'elif [[ "$1" == "static" ]]; then\n' + " printf '{\"gpu_data\": []}\\n'\n" + "fi\n" + ) + fake_amd_smi.chmod(0o755) + metrics = tmp_path / "gpu_metrics.csv" + benchmark_lib = Path(__file__).parents[1] / "benchmarks/benchmark_lib.sh" + script = f""" +source {str(benchmark_lib)!r} +start_gpu_monitor --output {str(metrics)!r} --interval 1 +sleep 0.8 +stop_gpu_monitor +""" + env = { + "PATH": f"{fake_bin}:/usr/bin:/bin", + "PYTHONDONTWRITEBYTECODE": "1", + } + + result = subprocess.run( + ["bash", "-c", script], + env=env, + capture_output=True, + text=True, + check=False, + timeout=30, + ) + + assert result.returncode == 0, result.stderr + lines = metrics.read_text().splitlines() + assert lines[0] == "timestamp,gpu,socket_power,power_management" + assert sum(1 for line in lines if line.startswith("timestamp,")) == 1 + assert "CTRL" not in metrics.read_text() + data_rows = [line for line in lines[1:] if line] + assert len(data_rows) >= 2 + assert all(row.split(",")[2] == "238" for row in data_rows) + energy_start = tmp_path / "gpu_metrics_energy_start.csv" + assert energy_start.read_text().startswith("gpu,total_energy_consumption") + assert (tmp_path / "gpu_metrics_energy_end.csv").exists() + identity = json.loads((tmp_path / "gpu_metrics_identity.json").read_text()) + assert identity == {"gpu_data": []} + # ============================================================================= # Integration: multinode power aggregation patches the agg JSON