From 68cc018a4bb96b492a5fbad352bf24593e06589e Mon Sep 17 00:00:00 2001 From: Wenyao Gao Date: Thu, 27 Aug 2026 05:57:08 -0700 Subject: [PATCH 1/3] =?UTF-8?q?fix(power):=20poll=20AMD=20telemetry=20unti?= =?UTF-8?q?l=20it=20covers=20the=20stop=20request=20before=20killing=20the?= =?UTF-8?q?=20monitor=20/=20=E4=BF=AE=E5=A4=8D=EF=BC=9AAMD=20=E9=81=A5?= =?UTF-8?q?=E6=B5=8B=E8=A6=86=E7=9B=96=E5=81=9C=E6=AD=A2=E8=AF=B7=E6=B1=82?= =?UTF-8?q?=E5=90=8E=E5=86=8D=E7=BB=88=E6=AD=A2=E5=8A=9F=E8=80=97=E7=9B=91?= =?UTF-8?q?=E6=8E=A7=E8=BF=9B=E7=A8=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The AMD stop path used an open-loop fixed sleep (interval+2) before killing the awk pipeline consumer, so rows still in the OS pipe or not yet emitted by a slow amd-smi iteration were lost and the last on-file tick could trail the aiperf window end by several seconds (run 32433563482 conc1: last tick 1787277605 vs end ...609.157), failing power validation with benchmark_window_not_bracketed. Replace the fixed tail with a bounded poll of the output CSV: wait until every observed GPU has a usable tick (numeric epoch timestamp, power > 0) stamped at the first whole second past stop entry — amd-smi stamps integer seconds, and the window end never exceeds the stop-entry wall clock, so that tick strictly covers any fractional window end. Bounded by AMD_MONITOR_STOP_TIMEOUT_S (default 30 s, 0 skips); non-epoch timestamps keep the legacy fixed tail; on timeout or early monitor death it warns and proceeds so aggregation attributes the failure (fail-safe, never fail-silent). AgentX INT/TERM/EXIT traps stop in abort mode (timeout 0) to keep signal teardown fast; the explicit post-replay stop performs the full wait. The NVIDIA branch is unchanged. AMD 停止路径原先在杀掉 awk 管道消费者前只做固定时长的 sleep，导致管道缓冲中的 采样行丢失、文件中最后一个时间戳可能落后于基准窗口结束数秒，功耗校验因 benchmark_window_not_bracketed 失败。本补丁改为有界轮询输出 CSV：等到每个 GPU 都有一条时间戳到达停止时刻下一整秒的可用采样（数值时间戳、功率 > 0）再终止监控， 由 AMD_MONITOR_STOP_TIMEOUT_S 限定（默认 30 秒，0 表示跳过）；非 epoch 时间戳 回退到旧的固定等待；超时或监控提前退出时告警并继续，由聚合端归因。AgentX 的 INT/TERM/EXIT trap 以 abort 模式停止（跳过等待），正常收尾仍执行完整覆盖等待。 NVIDIA 分支不变。 Co-Authored-By: Claude Fable 5 --- benchmarks/benchmark_lib.sh | 129 +++++++++-- .../aggregation/test_power_lifecycle.py | 217 +++++++++++++++++- utils/test_process_result.py | 19 +- 3 files changed, 342 insertions(+), 23 deletions(-) diff --git a/benchmarks/benchmark_lib.sh b/benchmarks/benchmark_lib.sh index 4819f50599..c85180452e 100644 --- a/benchmarks/benchmark_lib.sh +++ b/benchmarks/benchmark_lib.sh @@ -111,6 +111,8 @@ unset _benchmark_caller GPU_MONITOR_PID="" GPU_MONITOR_VENDOR="" GPU_MONITOR_INTERVAL=1 +# Bounded wait for AMD telemetry to cover a stop request; 0 skips the wait. +AMD_MONITOR_STOP_TIMEOUT_S="${AMD_MONITOR_STOP_TIMEOUT_S:-30}" 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 @@ -166,17 +168,18 @@ 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). + # The aggregator requires, for every GPU, a usable sample stamped at or + # after the (fractional) benchmark window end, which is always <= the + # wall clock when this stop runs. NVIDIA appends a one-shot post-exit + # sample below; amd-smi one-shot CSV has no timestamp column, so the + # AMD path polls the output file until every GPU's watch stream shows + # a usable tick at the next whole second — amd-smi stamps integer + # seconds, so that tick strictly covers any fractional window end + # (measured on MI355X: end=...609.157 vs last sample ...605). Observing + # the file rather than sleeping also defeats pipe-buffer loss when the + # awk consumer is killed: covered rows are already on disk. if [[ "$GPU_MONITOR_VENDOR" == "amd" ]]; then - sleep $(( ${GPU_MONITOR_INTERVAL:-1} + 2 )) + _wait_for_amd_stop_coverage fi kill "$GPU_MONITOR_PID" 2>/dev/null wait "$GPU_MONITOR_PID" 2>/dev/null || true @@ -222,6 +225,96 @@ _repair_truncated_gpu_metrics_tail() { return 0 } +# Print the newest telemetry tick (whole epoch seconds) that EVERY observed +# GPU has covered with a usable sample (numeric epoch timestamp, numeric +# power > 0), or nothing when the stream holds no usable epoch-stamped row +# (e.g. an amd-smi build emitting ISO timestamps). Column detection mirrors +# _POWER_COL_RE/_POWER_EXCLUDE_RE/_GPU_INDEX_COL_RE in utils/aggregate_power.py. +# POSIX awk only: the ROCm container images ship mawk/busybox awk. +_amd_monitor_min_covered_tick() { + [[ -f "$GPU_METRICS_CSV" ]] || return 0 + awk -F, ' + NR == 1 { + for (i = 1; i <= NF; i++) { + name = tolower($i) + gsub(/^ +| +$/, "", name) + sub(/\r$/, "", name) + if (!power_col && name ~ /power/ && name !~ /limit|cap|max|min/) + power_col = i + if (!gpu_col && name ~ /^(index|gpu|gpu_id|gpu_index|card|device)$/) + gpu_col = i + } + next + } + !power_col || !gpu_col { next } + { + # amd-smi quotes list-valued cells that embed commas; neutralize + # them so the power cell keeps its header-relative position. + line = $0 + sub(/\r$/, "", line) + if (line ~ /"/) { + n = split(line, seg, /"/) + line = "" + for (i = 1; i <= n; i++) { + if (i % 2 == 0) gsub(/,/, ";", seg[i]) + line = line seg[i] + } + } + count = split(line, cell, /,/) + if (count < power_col || count < gpu_col) next + if (cell[1] !~ /^[0-9]+(\.[0-9]+)?$/) next + if (cell[power_col] !~ /^[0-9]+(\.[0-9]+)?$/) next + if (cell[power_col] + 0 <= 0) next + if (cell[gpu_col] == "") next + gpu = cell[gpu_col] + if (!(gpu in newest) || cell[1] + 0 > newest[gpu]) + newest[gpu] = cell[1] + 0 + } + END { + have = 0 + for (gpu in newest) + if (!have || newest[gpu] < min) { min = newest[gpu]; have = 1 } + if (have) printf "%d\n", min + } + ' "$GPU_METRICS_CSV" 2>/dev/null + return 0 +} + +# Block until every observed GPU has a usable tick at/after the first whole +# second past stop entry, so any window end preceding the stop request is +# bracketed on file. Bounded by AMD_MONITOR_STOP_TIMEOUT_S; always returns 0 — +# on timeout or early monitor death it warns and lets aggregation attribute +# the missing coverage (fail-safe, never fail-silent). +_wait_for_amd_stop_coverage() { + local target deadline covered + if [[ "${AMD_MONITOR_STOP_TIMEOUT_S:-30}" -le 0 ]]; then + return 0 + fi + target=$(( $(date +%s) + 1 )) + deadline=$(( target + ${AMD_MONITOR_STOP_TIMEOUT_S:-30} )) + while :; do + covered=$(_amd_monitor_min_covered_tick) + if [[ -z "$covered" ]]; then + # Non-epoch timestamps or an unusable stream: keep the legacy + # fixed tail so older amd-smi builds behave exactly as before. + sleep $(( ${GPU_MONITOR_INTERVAL:-1} + 2 )) + return 0 + fi + if [[ "$covered" -ge "$target" ]]; then + return 0 + fi + if ! _background_process_is_running "$GPU_MONITOR_PID"; then + echo "[GPU Monitor] Warning: AMD monitor exited before covering the stop request (covered=$covered target=$target)" >&2 + return 0 + fi + if [[ "$(date +%s)" -ge "$deadline" ]]; then + echo "[GPU Monitor] Warning: AMD telemetry never covered the stop request within ${AMD_MONITOR_STOP_TIMEOUT_S}s (covered=$covered target=$target)" >&2 + return 0 + fi + sleep 1 + done +} + # Write one best-effort amd-smi snapshot; remove the file rather than keep a # partial one when the invocation fails. _write_amd_smi_sidecar() { @@ -2206,8 +2299,14 @@ run_agentic_replay_and_write_outputs() ( esac _stop_agentx_power_monitor() { + local mode="${1:-}" if [ "$agentx_monitor_stopped" = "0" ]; then agentx_monitor_stopped=1 + if [ "$mode" = "abort" ]; then + # A cancelled run's power validity is moot; skip the AMD + # coverage wait so signal teardown stays fast. + AMD_MONITOR_STOP_TIMEOUT_S=0 + fi stop_gpu_monitor fi } @@ -2252,10 +2351,12 @@ run_agentic_replay_and_write_outputs() ( agentx_monitor_stopped=0 # This function runs in a subshell, so these handlers cannot replace # launcher-owned traps. The stopped flag keeps explicit and signal/EXIT - # cleanup idempotent. - trap '_stop_agentx_power_monitor' EXIT - trap '_stop_agentx_power_monitor; exit 130' INT - trap '_stop_agentx_power_monitor; exit 143' TERM + # cleanup idempotent. Abort mode only ever fires when the explicit + # post-replay stop did not run (abnormal exit), where the coverage + # wait would only slow teardown down. + trap '_stop_agentx_power_monitor abort' EXIT + trap '_stop_agentx_power_monitor abort; exit 130' INT + trap '_stop_agentx_power_monitor abort; exit 143' TERM fi echo "$REPLAY_CMD" > "$result_dir/benchmark_command.txt" diff --git a/utils/agentic/aggregation/test_power_lifecycle.py b/utils/agentic/aggregation/test_power_lifecycle.py index de0e32df36..80fbf2e882 100644 --- a/utils/agentic/aggregation/test_power_lifecycle.py +++ b/utils/agentic/aggregation/test_power_lifecycle.py @@ -2,6 +2,7 @@ from __future__ import annotations +import csv import os import re import signal @@ -198,8 +199,9 @@ def test_multinode_formal_window_is_left_running_when_replay_is_interrupted(tmp_ def test_shared_lifecycle_installs_idempotent_signal_cleanup(): benchmark_lib = BENCHMARK_LIB.read_text() - assert "trap '_stop_agentx_power_monitor; exit 130' INT" in benchmark_lib - assert "trap '_stop_agentx_power_monitor; exit 143' TERM" in benchmark_lib + assert "trap '_stop_agentx_power_monitor abort' EXIT" in benchmark_lib + assert "trap '_stop_agentx_power_monitor abort; exit 130' INT" in benchmark_lib + assert "trap '_stop_agentx_power_monitor abort; exit 143' TERM" in benchmark_lib assert 'if [ "$agentx_monitor_stopped" = "0" ]' in benchmark_lib @@ -230,7 +232,9 @@ def test_signal_stops_monitor_once_without_replacing_parent_trap( start_gpu_monitor() {{ printf 'monitor-pid:%s\n' "${{BASHPID:-$$}}" >> {str(event_log)!r} }} -stop_gpu_monitor() {{ printf 'monitor-stop\n' >> {str(event_log)!r}; }} +stop_gpu_monitor() {{ + printf 'monitor-stop:%s\n' "${{AMD_MONITOR_STOP_TIMEOUT_S:-unset}}" >> {str(event_log)!r} +}} fake_replay() {{ sleep 30; }} trap 'printf "parent-exit\\n" >> {str(event_log)!r}' EXIT trap 'printf "parent-int\\n" >> {str(event_log)!r}; exit 130' INT @@ -263,7 +267,212 @@ def test_signal_stops_monitor_once_without_replacing_parent_trap( assert proc.returncode == expected_rc, stderr events = _events(tmp_path) - assert events.count("monitor-stop") == 1 + stop_events = [event for event in events if event.startswith("monitor-stop")] + # Signal teardown must stop exactly once, in abort mode: the coverage wait + # is skipped by setting AMD_MONITOR_STOP_TIMEOUT_S=0 before stopping. + assert stop_events == ["monitor-stop:0"] expected_parent_event = "parent-int" if sent_signal == signal.SIGINT else "parent-term" assert expected_parent_event in events assert events[-1] == "parent-exit" + + +# --------------------------------------------------------------------------- # +# stop_gpu_monitor AMD coverage wait (runs the real helper, no stubs) +# --------------------------------------------------------------------------- # + +# AMDSMI 26.2.0 `metric -p -c -t -u -w 1 --csv` header (order-faithful subset, +# measured on MI355X; mirrors test_detect_columns_amd_watch_mode_real_header). +_MI355X_WATCH_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" +) + +# amd-smi quotes list-valued cells with embedded commas; the coverage helper +# must keep the power cell at its header-relative position through them. +_WATCH_ROW_FORMAT = ( + "%s,%s,0,0,N/A,\"['N/A', 'N/A']\",\"['N/A', 'N/A']\",\"[0, 0]\"," + "\"[0, 0]\",\"[0, 0]\",%s,N/A,N/A,N/A,N/A,ENABLED,1404,2000,N/A,40,25\\n" +) + + +def _bash_single_quote(text: str) -> str: + return "'" + text.replace("'", "'\\''") + "'" + + +def _run_amd_stop( + tmp_path: Path, + *, + producer_script: str, + timeout_s: int, + interval: int = 1, + setup_script: str = "", +) -> subprocess.CompletedProcess[str]: + """Run the real stop_gpu_monitor against a scripted AMD telemetry producer.""" + csv_path = tmp_path / "gpu_metrics.csv" + bin_dir = tmp_path / "bin" + bin_dir.mkdir() + amd_smi = bin_dir / "amd-smi" + # Quiet stand-in for the _energy_end sidecar snapshot taken at stop. + amd_smi.write_text("#!/bin/bash\nprintf 'gpu,total_energy_consumption\\n0,100.0\\n'\n") + amd_smi.chmod(0o755) + script = f""" +source {str(BENCHMARK_LIB)!r} +GPU_METRICS_CSV={str(csv_path)!r} +printf '%s\\n' {_bash_single_quote(_MI355X_WATCH_HEADER)} > "$GPU_METRICS_CSV" +emit_row() {{ + printf {_bash_single_quote(_WATCH_ROW_FORMAT)} "$1" "$2" "$3" >> "$GPU_METRICS_CSV" +}} +{setup_script} +( {producer_script} ) & +GPU_MONITOR_PID=$! +printf '%s\\n' "$GPU_MONITOR_PID" > {str(tmp_path / "producer.pid")!r} +GPU_MONITOR_VENDOR=amd +GPU_MONITOR_INTERVAL={interval} +AMD_MONITOR_STOP_TIMEOUT_S={timeout_s} +date +%s > {str(tmp_path / "pre.txt")!r} +stop_gpu_monitor +date +%s > {str(tmp_path / "post.txt")!r} +""" + return subprocess.run( + ["bash", "-c", script], + env={ + **os.environ, + "PATH": f"{bin_dir}:/usr/bin:/bin", + "PYTHONDONTWRITEBYTECODE": "1", + }, + capture_output=True, + text=True, + check=False, + timeout=60, + ) + + +def _min_covered_tick(csv_path: Path) -> int: + """Newest usable tick (numeric ts, power > 0) covered by every GPU.""" + newest: dict[str, float] = {} + with csv_path.open(newline="", encoding="utf-8") as f: + for row in csv.DictReader(f): + try: + timestamp = float((row.get("timestamp") or "").strip()) + power = float((row.get("socket_power") or "").strip()) + except ValueError: + continue + gpu = (row.get("gpu") or "").strip() + if not gpu or power <= 0: + continue + newest[gpu] = max(newest.get(gpu, 0.0), timestamp) + assert newest, "no usable telemetry rows" + return int(min(newest.values())) + + +def _stop_epochs(tmp_path: Path) -> tuple[int, int]: + pre = int((tmp_path / "pre.txt").read_text().strip()) + post = int((tmp_path / "post.txt").read_text().strip()) + return pre, post + + +def _assert_producer_dead(tmp_path: Path) -> None: + producer_pid = int((tmp_path / "producer.pid").read_text().strip()) + with pytest.raises(ProcessLookupError): + os.kill(producer_pid, 0) + + +def test_amd_stop_waits_until_every_gpu_covers_stop_request(tmp_path: Path): + producer = """ +while :; do + now=$(date +%s) + emit_row "$now" 0 500 + emit_row "$now" 1 505 + sleep 0.2 +done +""" + started = time.monotonic() + result = _run_amd_stop(tmp_path, producer_script=producer, timeout_s=30) + duration = time.monotonic() - started + + assert result.returncode == 0, result.stderr + assert "never covered the stop request" not in result.stdout + result.stderr + pre, _ = _stop_epochs(tmp_path) + # Every GPU has a usable tick at/after the first whole second past stop + # entry, so any fractional window end before the stop is bracketed. + assert _min_covered_tick(tmp_path / "gpu_metrics.csv") >= pre + 1 + _assert_producer_dead(tmp_path) + assert duration < 10 + + +def test_amd_stop_ignores_degenerate_rows_for_coverage(tmp_path: Path): + setup = """ +stale=$(( $(date +%s) - 30 )) +emit_row "$stale" 0 500 +emit_row "$stale" 1 505 +""" + producer = """ +while :; do + now=$(date +%s) + emit_row "$now" 0 N/A + emit_row "$now" 1 N/A + sleep 0.2 +done +""" + result = _run_amd_stop( + tmp_path, + producer_script=producer, + timeout_s=2, + setup_script=setup, + ) + + assert result.returncode == 0, result.stderr + assert "never covered the stop request" in result.stderr + pre, post = _stop_epochs(tmp_path) + assert post - pre >= 2 + _assert_producer_dead(tmp_path) + + +def test_amd_stop_requires_coverage_per_gpu(tmp_path: Path): + setup = """ +stale=$(( $(date +%s) - 30 )) +emit_row "$stale" 1 505 +""" + producer = """ +while :; do + emit_row "$(date +%s)" 0 500 + sleep 0.2 +done +""" + result = _run_amd_stop( + tmp_path, + producer_script=producer, + timeout_s=2, + setup_script=setup, + ) + + assert result.returncode == 0, result.stderr + # GPU 1 never covers the stop request, so min-over-GPUs coverage times out + # even though GPU 0 keeps producing fresh usable ticks. + assert "never covered the stop request" in result.stderr + pre, post = _stop_epochs(tmp_path) + assert post - pre >= 2 + _assert_producer_dead(tmp_path) + + +def test_amd_stop_falls_back_to_fixed_tail_for_iso_timestamps(tmp_path: Path): + producer = """ +while :; do + emit_row "$(date +%Y-%m-%dT%H:%M:%S)" 0 500 + sleep 0.2 +done +""" + started = time.monotonic() + result = _run_amd_stop(tmp_path, producer_script=producer, timeout_s=30, interval=0) + duration = time.monotonic() - started + + assert result.returncode == 0, result.stderr + assert "never covered the stop request" not in result.stdout + result.stderr + assert "exited before covering" not in result.stdout + result.stderr + pre, post = _stop_epochs(tmp_path) + # Non-epoch timestamps keep the legacy interval+2 fixed tail (one shot). + assert post - pre >= 2 + assert duration < 10 + _assert_producer_dead(tmp_path) diff --git a/utils/test_process_result.py b/utils/test_process_result.py index 360a5ae0b3..078429af0d 100644 --- a/utils/test_process_result.py +++ b/utils/test_process_result.py @@ -7,6 +7,7 @@ import json import subprocess import sys +import time from pathlib import Path import pytest @@ -1033,8 +1034,13 @@ 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.""" + def test_stop_gpu_monitor_amd_covers_stop_request_and_snapshots_energy(self, tmp_path): + """AMD stop returns once telemetry covers the stop entry, then snapshots energy. + + A usable tick stamped past the stop request satisfies the coverage + poll on its first pass: the legacy fixed tail sleep never runs, the + stream is not mutated, and the end-side accumulator snapshot is + written.""" fake_bin = tmp_path / "bin" fake_bin.mkdir() args_log = tmp_path / "amd_args.txt" @@ -1046,7 +1052,8 @@ def test_stop_gpu_monitor_amd_waits_one_tick_and_snapshots_energy(self, tmp_path ) fake_amd_smi.chmod(0o755) sleep_log = tmp_path / "sleep_args.txt" - contents = "timestamp,gpu,socket_power\n1785881113,0,238\n" + covered_tick = int(time.time()) + 30 + contents = f"timestamp,gpu,socket_power\n{covered_tick},0,238\n" metrics = tmp_path / "gpu_metrics.csv" metrics.write_text(contents) benchmark_lib = Path(__file__).parents[1] / "benchmarks/benchmark_lib.sh" @@ -1054,7 +1061,7 @@ def test_stop_gpu_monitor_amd_waits_one_tick_and_snapshots_energy(self, tmp_path source {str(benchmark_lib)!r} kill() {{ return 0; }} wait() {{ return 0; }} -sleep() {{ printf '%s\\n' "$1" > {str(sleep_log)!r}; }} +sleep() {{ printf '%s\\n' "$1" >> {str(sleep_log)!r}; }} GPU_MONITOR_PID=999 GPU_MONITOR_VENDOR=amd GPU_MONITOR_INTERVAL=3 @@ -1075,7 +1082,8 @@ def test_stop_gpu_monitor_amd_waits_one_tick_and_snapshots_energy(self, tmp_path ) assert result.returncode == 0, result.stderr - assert sleep_log.read_text().strip() == "5" + assert not sleep_log.exists() + assert "never covered the stop request" not in result.stderr assert metrics.read_text() == contents assert "metric -E --csv" in args_log.read_text() energy_end = tmp_path / "gpu_metrics_energy_end.csv" @@ -1104,6 +1112,7 @@ def test_stop_gpu_monitor_amd_drops_truncated_row_without_append(self, tmp_path) GPU_MONITOR_PID=999 GPU_MONITOR_VENDOR=amd GPU_METRICS_CSV={str(metrics)!r} +AMD_MONITOR_STOP_TIMEOUT_S=0 stop_gpu_monitor """ env = { From 41ea427792baaa0d68dd46fda7072bd8c02e7d8c Mon Sep 17 00:00:00 2001 From: Wenyao Gao Date: Thu, 27 Aug 2026 05:57:33 -0700 Subject: [PATCH 2/3] =?UTF-8?q?fix(power):=20count=20degenerate=20boundary?= =?UTF-8?q?=20rows=20instead=20of=20bracketing=20or=20poisoning=20with=20t?= =?UTF-8?q?hem=20/=20=E4=BF=AE=E5=A4=8D=EF=BC=9A=E7=BB=9F=E8=AE=A1?= =?UTF-8?q?=E7=AA=97=E5=8F=A3=E5=A4=96=E7=9A=84=E5=BC=82=E5=B8=B8=E5=8A=9F?= =?UTF-8?q?=E8=80=97=E8=A1=8C=EF=BC=8C=E4=B8=8D=E5=86=8D=E7=94=A8=E5=85=B6?= =?UTF-8?q?=E4=BC=AA=E9=80=A0=E7=AA=97=E5=8F=A3=E8=A6=86=E7=9B=96=E6=88=96?= =?UTF-8?q?=E6=B1=A1=E6=9F=93=E6=A0=A1=E9=AA=8C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit SMI teardown rows can carry N/A or 0 W power cells. Inside the +/-3 s ingest band but outside the formal window, an N/A row previously flipped validity via invalid_power_sample, and a 0 W row could silently satisfy end bracketing while corrupting the boundary interpolation. integrate_power now skips such rows (power missing, non-finite, or <= 0 outside [start, end]) and counts them per GPU in a new additive sidecar field boundary_degenerate_rows. In-window semantics are unchanged: in-window N/A still yields invalid_power_sample and in-window 0 W still integrates. No new reason codes; sidecar schema_version stays 1; aggregate rows are untouched. SMI 收尾阶段可能输出功率为 N/A 或 0 W 的行：在 ±3 秒摄取带内但位于正式窗口外时， N/A 行会误置 invalid_power_sample，0 W 行则可能伪造窗口末端覆盖并污染边界插值。 integrate_power 现在跳过此类行（窗口外且功率缺失、非有限或 <= 0），并按 GPU 计入 新增的 sidecar 字段 boundary_degenerate_rows。窗口内语义不变：窗口内 N/A 仍记 invalid_power_sample，窗口内 0 W 仍参与积分。不新增 reason 代码，sidecar schema_version 保持 1，聚合结果行结构不变。 Includes a regression fixture for run 32433563482 conc1 (MI355X integer-second ticks ending 4 s before the fractional aiperf window end) asserting the producer failure stays attributed as benchmark_window_not_bracketed. Co-Authored-By: Claude Fable 5 --- utils/aggregate_power.py | 24 ++++- utils/test_aggregate_power.py | 176 ++++++++++++++++++++++++++++++++++ 2 files changed, 198 insertions(+), 2 deletions(-) diff --git a/utils/aggregate_power.py b/utils/aggregate_power.py index b1368580f0..d231b796c4 100644 --- a/utils/aggregate_power.py +++ b/utils/aggregate_power.py @@ -9,7 +9,11 @@ aggregate and a validation sidecar, but does not fail the benchmark. Power studies can set ``REQUIRE_POWER=1`` to fail after those audit artifacts exist. The aggregate carries numeric ``power_valid`` (1/0) for metric ingestion; the -sidecar is the canonical source for boolean validity and reason codes. +sidecar is the canonical source for boolean validity and reason codes. Rows in +the ingest band but outside the formal window whose power is missing, +non-finite, or <= 0 are teardown noise: they are skipped and counted in the +sidecar's ``boundary_degenerate_rows`` instead of poisoning validity or faking +window bracketing. """ from __future__ import annotations @@ -22,7 +26,7 @@ import os import re import sys -from dataclasses import dataclass +from dataclasses import dataclass, field from datetime import datetime, timezone from pathlib import Path from statistics import mean @@ -65,6 +69,10 @@ class PowerIntegration: per_gpu_max_sample_gap_s: dict[str, float] per_gpu_energy_j: dict[str, float] device_issues: dict[str, list[str]] + # Rows in the ingest band but outside the formal window whose power was + # missing/N-A/non-finite/<=0, skipped and counted per GPU; "unknown" + # buckets rows without a GPU identity. + boundary_degenerate_rows: dict[str, int] = field(default_factory=dict) avg_power_w: float | None = None avg_total_gpu_power_w: float | None = None total_gpu_energy_j: float | None = None @@ -368,6 +376,7 @@ def integrate_power( # expose timestamps at lower resolution than their sampling cadence, so # duplicate-timestamp readings are averaged rather than treated as corrupt. raw_samples: dict[str, dict[float, list[float]]] = {} + boundary_degenerate: dict[str, int] = {} saw_missing_gpu_identity = False try: with csv_path.open("r", newline="", encoding="utf-8", errors="replace") as f: @@ -408,6 +417,15 @@ def integrate_power( power = _parse_power((row.get(power_col) or "").strip()) gpu_id = (row.get(gpu_col) or "").strip() + if (power is None or not math.isfinite(power) or power <= 0.0) and ( + timestamp < start_unix or timestamp > end_unix + ): + # SMI teardown rows can carry N/A or 0 W cells: outside the + # formal window they are counted, never used to satisfy + # bracketing or to poison in-window validity. + key = gpu_id or "unknown" + boundary_degenerate[key] = boundary_degenerate.get(key, 0) + 1 + continue if power is None: _append_reason(reasons, "invalid_power_sample") continue @@ -502,6 +520,7 @@ def integrate_power( per_gpu_max_sample_gap_s=per_gpu_max_sample_gap_s, per_gpu_energy_j=per_gpu_energy_j, device_issues=device_issues, + boundary_degenerate_rows=boundary_degenerate, avg_power_w=avg_power_w, avg_total_gpu_power_w=avg_total_gpu_power_w, total_gpu_energy_j=total_gpu_energy_j, @@ -812,6 +831,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, + "boundary_degenerate_rows": integration.boundary_degenerate_rows, "accumulator_check": accumulator_check, "metrics": { key: round(value, 6) diff --git a/utils/test_aggregate_power.py b/utils/test_aggregate_power.py index 1b7029a454..18687195b2 100644 --- a/utils/test_aggregate_power.py +++ b/utils/test_aggregate_power.py @@ -573,6 +573,182 @@ def test_run_rejects_malformed_telemetry_inside_window( assert audit["reasons"] == [expected_reason] +# --------------------------------------------------------------------------- # +# Boundary-degenerate teardown rows (outside the window, inside the band) +# --------------------------------------------------------------------------- # + +# AMDSMI 26.2.0 `metric -p -c -t -u -w 1 --csv` header (order-faithful subset, +# measured on MI355X; see test_detect_columns_amd_watch_mode_real_header). +_MI355X_WATCH_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" +) + + +def _mi355x_watch_row(timestamp: int, gpu: int, socket_power: str) -> str: + """One data row in the shape captured from run 32433563482 (conc1): + integer-second epoch, quoted list cells with embedded commas, N/A cells, + and a trailing carriage return.""" + return ( + f"{timestamp},{gpu},0,0,N/A,\"['N/A', 'N/A', 'N/A', 'N/A']\"," + "\"['N/A', 'N/A']\",\"[0, 0, 0, 0, 0, 0, 0, 0]\",\"[0, 0]\",\"[0, 0]\"," + f"{socket_power},N/A,N/A,N/A,N/A,ENABLED,1404,2000,N/A,40,25\r" + ) + + +def test_integrate_power_skips_na_power_rows_outside_window(tmp_path: Path): + """N/A-power teardown rows past the window end are counted, not poisonous.""" + csv = tmp_path / "gpu_metrics.csv" + base = 1_700_000_000.0 + lines = ["timestamp,gpu,socket_power,temperature"] + for offset in range(-2, 13): + for gpu in range(2): + lines.append(f"{base + offset},{gpu},500.0,65") + for gpu in range(2): + lines.append(f"{base + 11},{gpu},N/A,65") + csv.write_text("\n".join(lines) + "\n", encoding="utf-8") + + result = integrate_power( + csv, + start_unix=base, + end_unix=base + 10, + expected_num_gpus=2, + ) + + assert result.power_valid is True + assert result.invalid_reasons == () + assert result.boundary_degenerate_rows == {"0": 1, "1": 1} + assert result.total_gpu_energy_j == pytest.approx(10_000.0) + + +def test_integrate_power_does_not_bracket_with_zero_power_tail(tmp_path: Path): + """A 0 W teardown row past the window end must not fake end bracketing. + + Legacy behavior accepted the 0 W row as a valid boundary sample, corrupting + the end interpolation with a bogus value; this intentionally flips that + case to an explicit benchmark_window_not_bracketed failure.""" + csv = tmp_path / "gpu_metrics.csv" + base = 1_700_000_000.0 + lines = ["timestamp,gpu,socket_power,temperature"] + # Good rows stop at end-4; the only post-end sample per GPU has power 0. + for offset in range(-1, 7): + for gpu in range(2): + lines.append(f"{base + offset},{gpu},500.0,65") + for gpu in range(2): + lines.append(f"{base + 11},{gpu},0,65") + csv.write_text("\n".join(lines) + "\n", encoding="utf-8") + + result = integrate_power( + csv, + start_unix=base, + end_unix=base + 10, + expected_num_gpus=2, + ) + + assert result.power_valid is False + assert "benchmark_window_not_bracketed" in result.invalid_reasons + assert result.boundary_degenerate_rows == {"0": 1, "1": 1} + + +def test_integrate_power_keeps_zero_power_semantics_inside_window(tmp_path: Path): + """Frozen legacy behavior: an in-window 0 W sample still integrates.""" + csv = tmp_path / "gpu_metrics.csv" + base = 1_700_000_000.0 + lines = ["timestamp,gpu,socket_power,temperature"] + for offset in range(-1, 12): + watts = "0.0" if offset == 5 else "500.0" + lines.append(f"{base + offset},0,{watts},65") + csv.write_text("\n".join(lines) + "\n", encoding="utf-8") + + result = integrate_power( + csv, + start_unix=base, + end_unix=base + 10, + expected_num_gpus=1, + ) + + assert result.power_valid is True + assert result.invalid_reasons == () + assert result.boundary_degenerate_rows == {} + # Trapezoids dip to 0 at t=5: 8 x 500 + 2 x 250 = 4500 J. + assert result.total_gpu_energy_j == pytest.approx(4_500.0) + + +def test_run_writes_boundary_degenerate_rows_to_sidecar(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, + ) + 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, + total_output=2_000, + total_input=10_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 + audit = json.loads(validation.read_text()) + # Present-and-empty for clean streams: readers can rely on the key. + assert audit["boundary_degenerate_rows"] == {} + + +def test_integrate_power_regression_mi355x_integer_ticks_end_gap(tmp_path: Path): + """Run-32433563482 conc1 regression: amd-smi integer-second ticks stop 4 s + before the fractional aiperf window end (last tick 1787277605 vs end + ...609.157497), so bracketing fails and the producer-side telemetry loss is + attributed as benchmark_window_not_bracketed on every GPU. + + The retrieved artifact's trailing rows all carry valid socket_power + (254-264 W) with N/A activity/voltage cells; the N/A- and 0-power teardown + rows appended past the window end are the documented synthetic degenerate + shapes, asserting they are counted rather than used for bracketing.""" + csv = tmp_path / "gpu_metrics.csv" + start = 1_787_277_560.155891 + end = 1_787_277_609.157497 + last_tick = 1_787_277_605 + powers = [259, 255, 263, 264, 256, 254, 259, 259] + lines = [_MI355X_WATCH_HEADER] + for tick in range(1_787_277_555, last_tick + 1): + for gpu in range(8): + lines.append(_mi355x_watch_row(tick, gpu, str(powers[gpu]))) + # amd-smi watch mode emits a blank line between tick groups. + lines.append("") + for gpu in range(8): + lines.append(_mi355x_watch_row(1_787_277_610, gpu, "N/A")) + for gpu in range(8): + lines.append(_mi355x_watch_row(1_787_277_611, gpu, "0")) + csv.write_text("\n".join(lines) + "\n", encoding="utf-8") + + result = integrate_power( + csv, + start_unix=start, + end_unix=end, + expected_num_gpus=8, + ) + + assert result.power_valid is False + assert result.invalid_reasons == ("benchmark_window_not_bracketed",) + assert result.device_issues == { + str(gpu): ["benchmark_window_not_bracketed"] for gpu in range(8) + } + assert result.boundary_degenerate_rows == {str(gpu): 2 for gpu in range(8)} + + def test_run_patches_agg_with_power_and_joules(tmp_path: Path): base = 1_700_000_000.0 csv = tmp_path / "gpu_metrics.csv" From 3c088145271d35c0e6e988fd6de3e00c8dacf34d Mon Sep 17 00:00:00 2001 From: Wenyao Gao Date: Thu, 27 Aug 2026 06:40:01 -0700 Subject: [PATCH 3/3] =?UTF-8?q?fix(power):=20harden=20AMD=20stop-coverage?= =?UTF-8?q?=20poll=20against=20bad=20timeout=20and=20ms=20epochs=20/=20?= =?UTF-8?q?=E5=8A=A0=E5=9B=BA=20AMD=20=E5=81=9C=E6=AD=A2=E8=A6=86=E7=9B=96?= =?UTF-8?q?=E8=BD=AE=E8=AF=A2=EF=BC=9A=E5=AE=B9=E9=94=99=E9=9D=9E=E6=B3=95?= =?UTF-8?q?=E8=B6=85=E6=97=B6=E5=80=BC=E5=B9=B6=E5=BD=92=E4=B8=80=E5=8C=96?= =?UTF-8?q?=E6=AF=AB=E7=A7=92=E6=97=B6=E9=97=B4=E6=88=B3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Sanitize AMD_MONITOR_STOP_TIMEOUT_S: a non-integer value (e.g. "30s") previously aborted stop_gpu_monitor via a bash arithmetic error under set -e, leaking the monitor process and skipping tail repair and the energy sidecar; it now warns and falls back to the default 30. - Normalize millisecond epoch timestamps (>1e12) in the awk coverage helper, mirroring _parse_timestamp in utils/aggregate_power.py, so a ms-stamping amd-smi build cannot trivially satisfy the stop target. - Shell-contract tests pin both behaviors. Co-Authored-By: Claude Fable 5 --- benchmarks/benchmark_lib.sh | 25 ++++++--- .../aggregation/test_power_lifecycle.py | 54 ++++++++++++++++++- 2 files changed, 72 insertions(+), 7 deletions(-) diff --git a/benchmarks/benchmark_lib.sh b/benchmarks/benchmark_lib.sh index c85180452e..8baa29ebf5 100644 --- a/benchmarks/benchmark_lib.sh +++ b/benchmarks/benchmark_lib.sh @@ -266,9 +266,14 @@ _amd_monitor_min_covered_tick() { if (cell[power_col] !~ /^[0-9]+(\.[0-9]+)?$/) next if (cell[power_col] + 0 <= 0) next if (cell[gpu_col] == "") next + ts = cell[1] + 0 + # Mirror _parse_timestamp in utils/aggregate_power.py: normalize + # millisecond epochs so a ms-stamping amd-smi build cannot + # trivially satisfy any second-scale stop target. + if (ts > 1e12) ts /= 1000 gpu = cell[gpu_col] - if (!(gpu in newest) || cell[1] + 0 > newest[gpu]) - newest[gpu] = cell[1] + 0 + if (!(gpu in newest) || ts > newest[gpu]) + newest[gpu] = ts } END { have = 0 @@ -286,12 +291,20 @@ _amd_monitor_min_covered_tick() { # on timeout or early monitor death it warns and lets aggregation attribute # the missing coverage (fail-safe, never fail-silent). _wait_for_amd_stop_coverage() { - local target deadline covered - if [[ "${AMD_MONITOR_STOP_TIMEOUT_S:-30}" -le 0 ]]; then + local target deadline covered timeout_s + # A non-integer timeout (e.g. "30s") would abort the whole stop_gpu_monitor + # call under `set -e` at the arithmetic below, leaking the monitor process + # and skipping tail repair + the energy sidecar; fall back to the default. + timeout_s="${AMD_MONITOR_STOP_TIMEOUT_S:-30}" + if [[ ! "$timeout_s" =~ ^-?[0-9]+$ ]]; then + echo "[GPU Monitor] Warning: ignoring non-integer AMD_MONITOR_STOP_TIMEOUT_S='$timeout_s', using 30" >&2 + timeout_s=30 + fi + if [[ "$timeout_s" -le 0 ]]; then return 0 fi target=$(( $(date +%s) + 1 )) - deadline=$(( target + ${AMD_MONITOR_STOP_TIMEOUT_S:-30} )) + deadline=$(( target + timeout_s )) while :; do covered=$(_amd_monitor_min_covered_tick) if [[ -z "$covered" ]]; then @@ -308,7 +321,7 @@ _wait_for_amd_stop_coverage() { return 0 fi if [[ "$(date +%s)" -ge "$deadline" ]]; then - echo "[GPU Monitor] Warning: AMD telemetry never covered the stop request within ${AMD_MONITOR_STOP_TIMEOUT_S}s (covered=$covered target=$target)" >&2 + echo "[GPU Monitor] Warning: AMD telemetry never covered the stop request within ${timeout_s}s (covered=$covered target=$target)" >&2 return 0 fi sleep 1 diff --git a/utils/agentic/aggregation/test_power_lifecycle.py b/utils/agentic/aggregation/test_power_lifecycle.py index 80fbf2e882..b80c7ae48f 100644 --- a/utils/agentic/aggregation/test_power_lifecycle.py +++ b/utils/agentic/aggregation/test_power_lifecycle.py @@ -305,7 +305,7 @@ def _run_amd_stop( tmp_path: Path, *, producer_script: str, - timeout_s: int, + timeout_s: int | str, interval: int = 1, setup_script: str = "", ) -> subprocess.CompletedProcess[str]: @@ -359,6 +359,8 @@ def _min_covered_tick(csv_path: Path) -> int: power = float((row.get("socket_power") or "").strip()) except ValueError: continue + if timestamp > 1e12: # millisecond epoch, mirror _parse_timestamp + timestamp /= 1000.0 gpu = (row.get("gpu") or "").strip() if not gpu or power <= 0: continue @@ -457,6 +459,56 @@ def test_amd_stop_requires_coverage_per_gpu(tmp_path: Path): _assert_producer_dead(tmp_path) +def test_amd_stop_survives_non_integer_timeout(tmp_path: Path): + producer = """ +while :; do + now=$(date +%s) + emit_row "$now" 0 500 + emit_row "$now" 1 505 + sleep 0.2 +done +""" + started = time.monotonic() + result = _run_amd_stop(tmp_path, producer_script=producer, timeout_s="30s") + duration = time.monotonic() - started + + assert result.returncode == 0, result.stderr + # A non-integer timeout must not unwind stop_gpu_monitor via a bash + # arithmetic error (which would leak the monitor and skip tail repair + # and the energy sidecar): it warns, falls back to 30, and still waits. + assert "ignoring non-integer AMD_MONITOR_STOP_TIMEOUT_S='30s'" in result.stderr + assert "never covered the stop request" not in result.stdout + result.stderr + pre, _ = _stop_epochs(tmp_path) + assert _min_covered_tick(tmp_path / "gpu_metrics.csv") >= pre + 1 + _assert_producer_dead(tmp_path) + assert duration < 10 + + +def test_amd_stop_normalizes_millisecond_epoch_timestamps(tmp_path: Path): + producer = """ +while :; do + now=$(( $(date +%s) * 1000 + 123 )) + emit_row "$now" 0 500 + emit_row "$now" 1 505 + sleep 0.2 +done +""" + started = time.monotonic() + result = _run_amd_stop(tmp_path, producer_script=producer, timeout_s=30) + duration = time.monotonic() - started + + assert result.returncode == 0, result.stderr + # Raw millisecond epochs (~1.8e12) dwarf any second-scale target, so + # without normalization the poll would return + # instantly with zero tail coverage; mirrored _parse_timestamp + # normalization makes the poll wait for real coverage instead. + assert "never covered the stop request" not in result.stdout + result.stderr + pre, _ = _stop_epochs(tmp_path) + assert _min_covered_tick(tmp_path / "gpu_metrics.csv") >= pre + 1 + _assert_producer_dead(tmp_path) + assert duration < 10 + + def test_amd_stop_falls_back_to_fixed_tail_for_iso_timestamps(tmp_path: Path): producer = """ while :; do