[PowerX] isolate AMD monitor windows and worker lifecycle / 隔离 AMD 监控窗口与工作节点生命周期

.github/workflows/test-process-result.yml

@@ -10,6 +10,9 @@ on:
       - '.github/workflows/e2e-tests.yml'
       - '.github/workflows/test-process-result.yml'
       - 'benchmarks/benchmark_lib.sh'
+      - 'benchmarks/multi_node/amd_utils/**'
+      - 'runners/test_amd_monitor_wiring.py'
+      - 'runners/test_amd_power_lifecycle.py'
       - 'benchmarks/native_power_collect.sh'
       - 'benchmarks/native_power_lifecycle.sh'
       - 'runners/test_native_collector_barriers.py'
@@ -67,7 +70,7 @@ jobs:
         run: |
           cd utils
           uv run --no-project --exclude-newer PT12H --python 3.12 --with pytest --with pyyaml \
-            python -m pytest test_aggregate_power.py test_aggregate_power_multinode.py agentic/aggregation/ test_gb300_power_official_contract.py test_inject_srt_power_concurrencies.py test_process_result.py test_native_multinode_power.py ../runners/test_native_collector_barriers.py ../runners/test_native_collector_receipts.py ../runners/test_tilert_power_lifecycle.py -v
+            python -m pytest test_aggregate_power.py test_aggregate_power_multinode.py agentic/aggregation/ test_gb300_power_official_contract.py test_inject_srt_power_concurrencies.py test_process_result.py test_native_multinode_power.py ../runners/test_native_collector_barriers.py ../runners/test_native_collector_receipts.py ../runners/test_amd_monitor_wiring.py ../runners/test_amd_power_lifecycle.py ../runners/test_tilert_power_lifecycle.py -v
 
       - name: Test serving client result persistence
         run: |

benchmarks/benchmark_lib.sh

@@ -153,8 +153,12 @@ unset _benchmark_caller
 # --------------------------------
 
 GPU_MONITOR_PID=""
+GPU_MONITOR_SOURCE_PID=""
+GPU_MONITOR_PIPE=""
 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
@@ -196,8 +200,24 @@ start_gpu_monitor() {
         # 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" &
+        # Track both processes: killing only awk can leave amd-smi alive until
+        # its next write. Keep the FIFO beside this run's raw CSV, never shared.
+        GPU_MONITOR_PIPE="${output}.pipe.$$"
+        if ! mkfifo "$GPU_MONITOR_PIPE"; then
+            echo '[GPU Monitor] Warning: AMD telemetry FIFO is unavailable' >&2
+            # A colliding path may belong to another stream; teardown must not remove it.
+            GPU_MONITOR_PIPE=""
+            GPU_MONITOR_VENDOR=""
+            case "${REQUIRE_POWER:-0}" in
+                1|true|TRUE|yes|YES) return 1 ;;
+            esac
+            return 0
+        fi
+        PYTHONUNBUFFERED=1 amd-smi metric -p -c -t -u -w "$interval" --csv \
+            > "$GPU_MONITOR_PIPE" 2>/dev/null &
+        GPU_MONITOR_SOURCE_PID=$!
+        awk '/^timestamp,/{if(!h){print;h=1};next} h{print;fflush()}' \
+            < "$GPU_MONITOR_PIPE" > "$output" &
         GPU_MONITOR_PID=$!
         # Hardware energy-accumulator + identity snapshots; the end-side twin in
         # stop_gpu_monitor lets auditors cross-check the integrated energy
@@ -215,19 +235,21 @@ 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
+        # The monitor may exit during the coverage wait; still finish cleanup.
+        kill "$GPU_MONITOR_PID" 2>/dev/null || true
         wait "$GPU_MONITOR_PID" 2>/dev/null || true
         case "$GPU_MONITOR_VENDOR" in
             nvidia)
@@ -249,6 +271,13 @@ stop_gpu_monitor() {
             echo "[GPU Monitor] Collected $lines rows -> $GPU_METRICS_CSV"
         fi
     fi
+    if [[ -n "$GPU_MONITOR_SOURCE_PID" ]]; then
+        kill "$GPU_MONITOR_SOURCE_PID" 2>/dev/null || true
+        wait "$GPU_MONITOR_SOURCE_PID" 2>/dev/null || true
+    fi
+    [[ -z "$GPU_MONITOR_PIPE" ]] || rm -f "$GPU_MONITOR_PIPE"
+    GPU_MONITOR_SOURCE_PID=""
+    GPU_MONITOR_PIPE=""
     GPU_MONITOR_PID=""
     GPU_MONITOR_VENDOR=""
 }
@@ -271,6 +300,105 @@ _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
+            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) || ts > newest[gpu])
+                newest[gpu] = ts
+        }
+        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 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 + timeout_s ))
+    while :; do
+        covered=$(_amd_monitor_min_covered_tick)
+        # The first usable row may arrive after stop begins. Keep the same
+        # deadline for empty or unsupported streams instead of stopping early.
+        if [[ -n "$covered" && "$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 ${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() {
@@ -3284,9 +3412,15 @@ 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
+            agentx_monitor_stopped=1
         fi
     }
 
@@ -3330,10 +3464,11 @@ 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 after stopping completes. If a signal interrupts
+        # the normal coverage wait, abort cleanup must still kill the monitor.
+        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"

benchmarks/multi_node/amd_utils/bench.sh

@@ -52,6 +52,16 @@ profile_folder="${log_path}/${ENGINE}_isl_${chosen_isl}_osl_${chosen_osl}"
 mkdir -p "$profile_folder"
 
 source "$(dirname "$0")/../../benchmark_lib.sh"
+source "$(dirname "$0")/power.sh"
+power_required=0
+case "${REQUIRE_POWER:-0}" in
+    1|true|TRUE|yes|YES) power_required=1 ;;
+esac
+if ! wait_amd_multinode_power ready; then
+    [[ "$power_required" == 0 ]] || exit 1
+    echo 'PowerX: continuing without ready optional telemetry' >&2
+fi
+benchmark_exit_code=0
 
 REPO_ROOT="$(cd "$(dirname "$0")/../../.." && pwd)"
 
@@ -101,6 +111,7 @@ for max_concurrency in "${chosen_concurrencies[@]}"; do
         fi
     fi
 
+    point_exit_code=0
     run_benchmark_serving \
         --bench-serving-dir "$REPO_ROOT" \
         --model "$BENCH_MODEL" \
@@ -113,7 +124,8 @@ for max_concurrency in "${chosen_concurrencies[@]}"; do
         --max-concurrency "$max_concurrency" \
         --result-filename "$export_file" \
         --result-dir /workspace/ \
-        $extra_flags
+        $extra_flags || point_exit_code=$?
+    if [[ "$point_exit_code" != 0 ]]; then benchmark_exit_code=$point_exit_code; break; fi
 
     echo "-----------------------------------------"
 
@@ -123,3 +135,10 @@ for max_concurrency in "${chosen_concurrencies[@]}"; do
         sleep 10
     fi
 done
+
+# Stop every node while all prefill/decode servers are still alive.
+if ! wait_amd_multinode_power done; then
+    echo 'PowerX: collector completion failed' >&2
+    if [[ "$power_required" == 1 && "$benchmark_exit_code" == 0 ]]; then benchmark_exit_code=1; fi
+fi
+exit "$benchmark_exit_code"

benchmarks/multi_node/amd_utils/job.slurm

@@ -237,8 +237,9 @@ fi
 # Node Selection
 # =============================================================================
 
-NUM_NODES=$((xP + yD))
-echo "NUM_NODES: $NUM_NODES (xP=$xP + yD=$yD)"
+# Workers can span multiple physical nodes. Preserve the submit-time count.
+NUM_NODES="${NUM_NODES:-$(( ((PREFILL_TP_SIZE + GPUS_PER_NODE - 1) / GPUS_PER_NODE) * xP + ((DECODE_TP_SIZE + GPUS_PER_NODE - 1) / GPUS_PER_NODE) * yD ))}"
+echo "NUM_NODES: $NUM_NODES (prefill workers=$xP, decode workers=$yD)"
 
 FULL_NODELIST=$(scontrol show hostnames "$SLURM_JOB_NODELIST")
 SELECTED_NODES=$(echo "$FULL_NODELIST" | head -n $NUM_NODES)
@@ -318,6 +319,10 @@ export DRY_RUN="${DRY_RUN:-0}"
 export BENCHMARK_LOGS_DIR="${BENCHMARK_LOGS_DIR:-$(pwd)/benchmark_logs}"
 export KEEP_CONTAINERS="${KEEP_CONTAINERS:-0}"
 export ENGINE=$ENGINE
+export POWERX_HOST_UID=$(id -u)
+export POWERX_HOST_GID=$(id -g)
+export POWERX_COLLECTOR_REVISION=$(git -C "$DI_REPO_DIR" rev-parse HEAD)
+mkdir -p "$BENCHMARK_LOGS_DIR/power-control-${SLURM_JOB_ID}"
 
 # Eval-related env vars (threaded from submit.sh)
 export RUN_EVAL="${RUN_EVAL:-false}"
@@ -377,6 +382,16 @@ else
     echo "[WARN] $RDMA_CHECK_SCRIPT not found; skipping RDMA QoS/DCQCN pre-flight check"
 fi
 
+stage_native_power() {
+    srun --overlap --nodelist="$SELECTED_NODELIST_SRUN" --ntasks="$NUM_NODES" bash -c '
+        src="/tmp/slurm_job-${SLURM_JOB_ID}/native_power/node-${SLURM_PROCID}"
+        if [[ -d "$src" ]]; then
+            mkdir -p "$BENCHMARK_LOGS_DIR/native_power"
+            cp -r "$src" "$BENCHMARK_LOGS_DIR/native_power/"
+        fi
+    '
+}
+
 cleanup() {
   echo "[${SLURM_JOB_ID}] termination received on $(hostname); cleaning up container + stale logs..."
   # Backstop: on scancel/timeout/step-hang the foreground `exec docker run`
@@ -386,11 +401,13 @@ cleanup() {
   # other users' containers. (Ported from InferenceY 51ebfa88.)
   srun --nodelist="$SELECTED_NODELIST_SRUN" \
     bash -c 'eval "$DOCKER_CMD_DETECT"; $DOCKER_CMD rm -f '"$DOCKER_CONT_NAME"' 2>/dev/null || true' 2>/dev/null || true
+  stage_native_power || true
   rm -rf ${SLURM_SUBMIT_DIR}/logs 2>/dev/null || true
   echo "[${SLURM_JOB_ID}] cleanup done."
 }
 
-trap cleanup INT TERM HUP
+trap 'cleanup; exit 130' INT
+trap 'cleanup; exit 143' TERM HUP
 
 # Force NFS cache refresh on all nodes
 echo "Refreshing NFS caches on all nodes..."
@@ -412,6 +429,12 @@ DOCKER_ENV_COMMON=(
     -e SLURM_JOB_ID=\$SLURM_JOB_ID
     -e SLURM_JOB_NODELIST=\$SLURM_JOB_NODELIST
     -e NNODES=\$NNODES
+    -e REQUIRE_POWER=\${REQUIRE_POWER:-0}
+    -e POWERX_HOST_UID=\$POWERX_HOST_UID
+    -e POWERX_HOST_GID=\$POWERX_HOST_GID
+    -e POWERX_COLLECTOR_REVISION=\$POWERX_COLLECTOR_REVISION
+    -e POWERX_NODE_NAME=\$POWERX_NODE_NAME
+    -e POWERX_CLOCK_SYNCHRONIZED=\$POWERX_CLOCK_SYNCHRONIZED
     -e NODE_RANK=\$SLURM_PROCID
     -e NODE0_ADDR=\$NODE0_ADDR
     -e MODEL_DIR=/models
@@ -601,6 +624,10 @@ set -euo pipefail
 
 echo \"Rank \$SLURM_PROCID on \$(hostname)\"
 
+# Capture the host's synchronization state, not the container's missing D-Bus.
+export POWERX_NODE_NAME=\$(hostname)
+export POWERX_CLOCK_SYNCHRONIZED=\$(timedatectl show -p NTPSynchronized --value 2>/dev/null || true)
+
 # Per-node docker privilege detection
 eval \"\$DOCKER_CMD_DETECT\"
 echo \"[docker-detect] rank \$SLURM_PROCID: DOCKER_CMD=\$DOCKER_CMD\"
@@ -784,6 +811,17 @@ echo \"[rank 0] Main container exited (rc=\$DOCKER_EXIT_CODE). Stopping vllm-rou
 exit \$DOCKER_EXIT_CODE
 "
 
+BENCHMARK_STEP_RC=$?
+# Each host copies its own node-local, root-created artifacts as the runner user.
+# No raw telemetry is written as root into the shared checkout.
+if ! stage_native_power; then
+    echo 'PowerX: failed to stage native telemetry' >&2
+    case "${REQUIRE_POWER:-0}" in
+        1|true|TRUE|yes|YES)
+            if [[ "$BENCHMARK_STEP_RC" == 0 ]]; then BENCHMARK_STEP_RC=1; fi ;;
+    esac
+fi
+
 if [[ "${KEEP_CONTAINERS}" != "1" ]]; then
     srun --nodelist="$SELECTED_NODELIST_SRUN" bash -c 'eval "$DOCKER_CMD_DETECT"; $DOCKER_CMD rm -f '"$DOCKER_CONT_NAME"' '"$CLIENT_CONT_NAME"' 2>/dev/null || true'
 
@@ -794,3 +832,5 @@ if [[ "${KEEP_CONTAINERS}" != "1" ]]; then
         '
     fi
 fi
+
+exit "$BENCHMARK_STEP_RC"