[4/4] [Power] feat: strict AMD gpu monitor lifecycle for MI355X / MI355X 单节点功率监控严格生命周期

.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
 

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

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:

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__,