From 190fc769c421e3f74fde28d6db5765b1f2bc5d12 Mon Sep 17 00:00:00 2001 From: Yupeng Tang Date: Wed, 8 Jul 2026 23:34:56 -0700 Subject: [PATCH 1/2] Add generic stage-aware perf hook infrastructure (#683) Summary: Benchpress's perf hook spans an entire benchmark run today, so PMU + sysstat data ends up smeared across every sub-benchmark in one set of CSVs. For benchmarks that run many workloads back-to-back (WDL prod_set, SPEC2017 intrate, ...), distinguishing IPC, topdown breakdown, mpstat etc. per sub-benchmark requires teasing apart timestamps after the fact, which is brittle. This diff adds a generic, opt-in stage-aware mode to the perf hook. It is not tied to any specific benchmark -- consumers (WDL, SPEC, ...) land on top of this and just emit stage markers. The mechanism: 1. The perf hook accepts a new option `stage_aware: true`. When set, `before_job` does NOT start any monitors. Instead it creates a FIFO under benchmark_metrics_/perf_stage.fifo, advertises its path via the env var BENCHPRESS_PERF_STAGE_FIFO, and spawns a coordinator thread. 2. The coordinator reads commands from the FIFO. Each "START " allocates a fresh set of perf monitors with that stage name as a sub-folder; each "STOP" terminates the monitors and writes their CSVs. Multiple START/STOP cycles are supported. `after_job` writes a final __EXIT__ to drain the coordinator. 3. Every existing perf monitor (mpstat, memstat, netstat, perfstat, vmstat, cpufreq*, power, topdown -- including IntelPerfSpect/3, BasePerfUtil, AMDPerfUtil, ARMPerfUtil, NVPerfUtil, NeoVerseV3PerfUtil) gains a `subdir` constructor arg that the base `Monitor.gen_path` joins under benchmark_metrics_/. Existing callers that don't pass `subdir` keep their flat layout, so default mode is byte-for-byte unchanged. The resulting per-stage layout, e.g.: benchmark_metrics_/ / mpstat.csv mem-stat.csv perf-stat.csv topdown-... .csv Benchmark-specific wiring (WDL prod_set, SPEC2017 intrate) lands in later diffs in this stack. Reviewed By: YifanYuan3 Differential Revision: D108110315 --- benchpress/plugins/hooks/perf.py | 225 +++++++++++++++++- .../plugins/hooks/perf_monitors/__init__.py | 33 ++- .../hooks/perf_monitors/cpufreq_cpuinfo.py | 6 +- .../hooks/perf_monitors/cpufreq_scaling.py | 6 +- .../plugins/hooks/perf_monitors/memstat.py | 4 +- .../plugins/hooks/perf_monitors/mpstat.py | 4 +- .../plugins/hooks/perf_monitors/netstat.py | 4 +- .../plugins/hooks/perf_monitors/perfstat.py | 6 +- .../plugins/hooks/perf_monitors/power.py | 4 +- .../plugins/hooks/perf_monitors/topdown.py | 81 ++++--- .../plugins/hooks/perf_monitors/vmstat.py | 4 +- 11 files changed, 309 insertions(+), 68 deletions(-) diff --git a/benchpress/plugins/hooks/perf.py b/benchpress/plugins/hooks/perf.py index 091caf022..fd0d9fa62 100644 --- a/benchpress/plugins/hooks/perf.py +++ b/benchpress/plugins/hooks/perf.py @@ -8,6 +8,7 @@ import logging import os +import threading import traceback from benchpress.lib import open_source @@ -76,6 +77,27 @@ logger = logging.getLogger(__name__) +# Env var used to advertise the stage FIFO path to a child benchmark process. +# When set, a stage-aware benchmark (e.g. WDL prod_set's run_prod.sh) writes +# stage markers like ``START `` and ``STOP`` to that FIFO so +# this hook can rotate the perf monitor set per sub-benchmark and emit one +# folder of perf data per stage. +PERF_STAGE_FIFO_ENV = "BENCHPRESS_PERF_STAGE_FIFO" + + +def _resolve_subdir_option(opts): + """Return whether the user opted into stage-aware mode. + + Stage-aware mode is requested either by setting ``stage_aware: true`` in + the perf hook options, or by leaving the default (False). Today the only + benchmark that emits stage markers is WDL prod_set, but the wiring is + generic. + """ + if not isinstance(opts, dict): + return False + return bool(opts.get("stage_aware", False)) + + class Perf(Hook): def before_job(self, opts, job): self.opts = DEFAULT_OPTIONS @@ -91,8 +113,180 @@ def before_job(self, opts, job): if not os.path.isdir(self.benchmark_metrics_dir): os.mkdir(self.benchmark_metrics_dir) + self._stage_aware = _resolve_subdir_option(opts) + self._stage_thread = None + self._stage_fifo_path = None + self._stage_lock = threading.Lock() + self._current_stage = None + self._current_monitors = [] + + if self._stage_aware: + self._start_stage_coordinator(job) + else: + # Legacy / default path: start one set of monitors that span the + # entire benchmark. + self.monitors = self._build_and_run_monitors(job, subdir=None) + + def after_job(self, opts, job): + if self._stage_aware: + self._stop_stage_coordinator() + return + for monitor in self.monitors: + monitor.terminate() + for monitor in self.monitors: + monitor.write_csv() + + # ------------------------------------------------------------------ + # Stage-aware mode + # ------------------------------------------------------------------ + + def _start_stage_coordinator(self, job): + """Open a FIFO under the benchmark metrics dir and spawn a thread + that reads ``START `` / ``STOP`` commands. Each START gets + a fresh monitor set whose CSVs land in + ``benchmark_metrics_//``. + + The FIFO path is exposed via the ``BENCHPRESS_PERF_STAGE_FIFO`` + env var so the child benchmark process can find it. We deliberately + DO NOT start any monitors yet -- the coordinator launches them + when the first STAGE command arrives. + """ + self._stage_fifo_path = os.path.join( + self.benchmark_metrics_dir, "perf_stage.fifo" + ) + # Recreate the FIFO each run so a stale one from a previous run + # never confuses us. + if os.path.exists(self._stage_fifo_path): + os.unlink(self._stage_fifo_path) + os.mkfifo(self._stage_fifo_path, 0o600) + os.environ[PERF_STAGE_FIFO_ENV] = self._stage_fifo_path + self._stage_stop_event = threading.Event() + self._stage_thread = threading.Thread( + target=self._stage_loop, + args=(job,), + name="perf-stage-coordinator", + daemon=True, + ) + self._stage_thread.start() + logger.info( + f"Perf hook: stage-aware mode active, FIFO at {self._stage_fifo_path}" + ) + + def _stop_stage_coordinator(self): + # Tell the coordinator to exit and unblock the FIFO read by writing + # a final STOP. A standalone writer also lets us flush any lingering + # monitor set without depending on the benchmark script having sent + # its own STOP. + self._stage_stop_event.set() + try: + with open(self._stage_fifo_path, "w") as f: + f.write("STOP\n") + f.write("__EXIT__\n") + except Exception as e: + logger.warning(f"Perf hook: failed to nudge stage FIFO: {e}") + if self._stage_thread is not None: + self._stage_thread.join(timeout=30) + # Clean up the FIFO and the env var so a subsequent benchmark in the + # same process doesn't accidentally re-use stale state. + try: + os.unlink(self._stage_fifo_path) + except OSError: + pass + os.environ.pop(PERF_STAGE_FIFO_ENV, None) + + def _stage_loop(self, job): + """Read commands from the FIFO until __EXIT__. Each line is one of: + + START + STOP + __EXIT__ + + START allocates fresh monitors and runs them; STOP terminates them + and writes their CSVs. Multiple START/STOP cycles are supported. + + Implementation note: each writer that closes the FIFO causes EOF on + the reader, so we have to re-open the FIFO after every burst rather + than holding a single file handle. Each ``open(..., "r")`` blocks + until a writer is available again -- which is exactly the behavior + we want for a coordinator that's awaiting the next stage marker. + """ + try: + while not self._stage_stop_event.is_set(): + with open(self._stage_fifo_path, "r") as fifo: + for line in fifo: + line = line.strip() + if not line: + continue + if line == "__EXIT__": + self._end_current_stage(job) + return + if line.startswith("START "): + stage = line[len("START ") :].strip() + self._begin_stage(job, stage) + continue + if line == "STOP": + self._end_current_stage(job) + continue + logger.warning( + f"Perf hook: ignoring unknown stage command: {line!r}" + ) + except Exception as e: + logger.warning( + f"Perf hook: stage coordinator crashed: {type(e).__name__}: {e}" + ) + + def _begin_stage(self, job, stage_name): + with self._stage_lock: + if self._current_stage is not None: + # Implicit stop of the previous stage -- the script forgot to + # close it. Don't lose data; flush before starting the next. + logger.warning( + f"Perf hook: implicit STOP of stage " + f"{self._current_stage!r} before START {stage_name!r}" + ) + self._end_current_stage_locked(job) + # Sanitize the stage name into a filesystem-friendly subdir. + sanitized = _sanitize_subdir(stage_name) + logger.info(f"Perf hook: starting stage {sanitized!r}") + self._current_stage = sanitized + self._current_monitors = self._build_and_run_monitors(job, subdir=sanitized) + + def _end_current_stage(self, job): + with self._stage_lock: + self._end_current_stage_locked(job) + + def _end_current_stage_locked(self, job): + if self._current_stage is None: + return + logger.info(f"Perf hook: stopping stage {self._current_stage!r}") + for monitor in self._current_monitors: + try: + monitor.terminate() + except Exception as e: + logger.warning( + f"Perf hook: terminating monitor {monitor.name} failed: {e}" + ) + for monitor in self._current_monitors: + try: + monitor.write_csv() + except Exception as e: + logger.warning(f"Perf hook: write_csv on {monitor.name} failed: {e}") + self._current_stage = None + self._current_monitors = [] + + # ------------------------------------------------------------------ + # Shared monitor setup + # ------------------------------------------------------------------ + + def _build_and_run_monitors(self, job, subdir): + """Instantiate every enabled monitor (with the given subdir) and + start it. Returns the list of started monitors. + + Refactored out of the original ``before_job`` body so both default + mode and stage-aware mode share the same monitor-bring-up logic. + """ should_run_perf_stat = True - self.monitors = [] + monitors = [] for mon_name in AVAIL_MONITORS.keys(): # `enabled` is a perf-hook-level flag, not a monitor constructor # arg. Pop it out before passing the rest to the monitor class. @@ -102,7 +296,7 @@ def before_job(self, opts, job): continue try: MonitorClass = AVAIL_MONITORS[mon_name] - monitor = MonitorClass(job_uuid=job.uuid, **init_args) + monitor = MonitorClass(job_uuid=job.uuid, subdir=subdir, **init_args) # We should disable PerfStat (and not run anything that uses PMU) # if IntelPerfSpect3 is enabled. if isinstance(monitor, topdown.IntelPerfSpect3) and monitor.supported: @@ -110,26 +304,37 @@ def before_job(self, opts, job): "Disabling PerfStat to avoid conflict with IntelPerfSpect3" ) should_run_perf_stat = False - self.monitors.append(monitor) + monitors.append(monitor) except Exception as e: logger.warning( f"Failed to load the perf monitor {mon_name} due to the following exception:" ) logger.warning(traceback.print_exception(type(e), e, e.__traceback__)) - for monitor in self.monitors: + for monitor in monitors: try: if isinstance(monitor, perfstat.PerfStat) and not should_run_perf_stat: continue monitor.run() except Exception as e: logger.warning( - f"Could not run perf monitor {mon_name} due to the following exception:" + f"Could not run perf monitor {monitor.name} due to the following exception:" ) logger.warning(traceback.print_exception(type(e), e, e.__traceback__)) + return monitors - def after_job(self, opts, job): - for monitor in self.monitors: - monitor.terminate() - for monitor in self.monitors: - monitor.write_csv() + +def _sanitize_subdir(name): + """Make a stage name safe to use as a directory component. + + Strip path separators and trim. Keep alphanumerics, dashes, underscores; + replace anything else with underscore. + """ + out = [] + for ch in name.strip(): + if ch.isalnum() or ch in "-_.": + out.append(ch) + else: + out.append("_") + cleaned = "".join(out).strip("._") + return cleaned or "unnamed_stage" diff --git a/benchpress/plugins/hooks/perf_monitors/__init__.py b/benchpress/plugins/hooks/perf_monitors/__init__.py index f652e750e..c02bd910e 100644 --- a/benchpress/plugins/hooks/perf_monitors/__init__.py +++ b/benchpress/plugins/hooks/perf_monitors/__init__.py @@ -21,12 +21,30 @@ class Monitor: def gen_path(self, filename): - return os.path.join( - get_artifacts_dir(), f"benchmark_metrics_{self.job_uuid}", filename - ) + """Build a path under the job's benchmark_metrics directory. + + When ``self.subdir`` is set (per-stage perf collection in stage-aware + mode), the path is nested one level further so each stage gets its + own sub-folder. Without ``subdir`` the layout is unchanged from + legacy behavior: - def __init__(self, interval, name, job_uuid): - """Initialize some common parameters and storage variables""" + benchmark_metrics_/ # no subdir + benchmark_metrics_// # stage-aware mode + """ + base = os.path.join(get_artifacts_dir(), f"benchmark_metrics_{self.job_uuid}") + if getattr(self, "subdir", None): + base = os.path.join(base, self.subdir) + return os.path.join(base, filename) + + def __init__(self, interval, name, job_uuid, subdir=None): + """Initialize some common parameters and storage variables. + + Args: + subdir: Optional sub-folder under benchmark_metrics_/. Used + by the perf hook's stage-aware mode to give each WDL prod_set + sub-benchmark its own slice of perf data. None preserves the + original flat layout. + """ self.name = name self.interval = interval # Reserved for result processing @@ -34,8 +52,13 @@ def __init__(self, interval, name, job_uuid): # Reserved for original output of the monitoring process self.output = "" self.job_uuid = job_uuid + self.subdir = subdir self.logpath = self.gen_path(f"{name}.log") self.csvpath = self.gen_path(f"{name}.csv") + # Make sure the (possibly nested) directory exists before opening + # the log file. Stage-aware mode creates per-stage dirs lazily so we + # can't rely on the perf hook to have made them. + os.makedirs(os.path.dirname(self.logpath), exist_ok=True) self.logfile = open(self.logpath, "w", buffering=1) # noqa: P201 def __del__(self): diff --git a/benchpress/plugins/hooks/perf_monitors/cpufreq_cpuinfo.py b/benchpress/plugins/hooks/perf_monitors/cpufreq_cpuinfo.py index 1e06be095..955f5e802 100644 --- a/benchpress/plugins/hooks/perf_monitors/cpufreq_cpuinfo.py +++ b/benchpress/plugins/hooks/perf_monitors/cpufreq_cpuinfo.py @@ -14,8 +14,10 @@ class CPUFreq(Monitor): - def __init__(self, interval, job_uuid): - super(CPUFreq, self).__init__(interval, "cpufreq_cpuinfo", job_uuid) + def __init__(self, interval, job_uuid, subdir=None): + super(CPUFreq, self).__init__( + interval, "cpufreq_cpuinfo", job_uuid, subdir=subdir + ) self.run_freq_collector = False self.supported = os.path.exists( "/sys/devices/system/cpu/cpu0/cpufreq/cpuinfo_cur_freq" diff --git a/benchpress/plugins/hooks/perf_monitors/cpufreq_scaling.py b/benchpress/plugins/hooks/perf_monitors/cpufreq_scaling.py index b2b40371e..52b070a8b 100644 --- a/benchpress/plugins/hooks/perf_monitors/cpufreq_scaling.py +++ b/benchpress/plugins/hooks/perf_monitors/cpufreq_scaling.py @@ -14,8 +14,10 @@ class CPUFreq(Monitor): - def __init__(self, interval, job_uuid): - super(CPUFreq, self).__init__(interval, "cpufreq_scaling", job_uuid) + def __init__(self, interval, job_uuid, subdir=None): + super(CPUFreq, self).__init__( + interval, "cpufreq_scaling", job_uuid, subdir=subdir + ) self.run_freq_collector = False self.supported = os.path.exists( "/sys/devices/system/cpu/cpu0/cpufreq/scaling_cur_freq" diff --git a/benchpress/plugins/hooks/perf_monitors/memstat.py b/benchpress/plugins/hooks/perf_monitors/memstat.py index d3b1f84ff..286c2203f 100644 --- a/benchpress/plugins/hooks/perf_monitors/memstat.py +++ b/benchpress/plugins/hooks/perf_monitors/memstat.py @@ -22,8 +22,8 @@ class MemStat(Monitor): - def __init__(self, interval, job_uuid, additional_counters=()): - super(MemStat, self).__init__(interval, "mem-stat", job_uuid) + def __init__(self, interval, job_uuid, additional_counters=(), subdir=None): + super(MemStat, self).__init__(interval, "mem-stat", job_uuid, subdir=subdir) counters = {"MemTotal", "MemFree", "MemAvailable", "SwapTotal", "SwapFree"} self.counters = counters.union(set(additional_counters)) self.run_collector = False diff --git a/benchpress/plugins/hooks/perf_monitors/mpstat.py b/benchpress/plugins/hooks/perf_monitors/mpstat.py index 7ba145885..9c3dabc0e 100644 --- a/benchpress/plugins/hooks/perf_monitors/mpstat.py +++ b/benchpress/plugins/hooks/perf_monitors/mpstat.py @@ -12,8 +12,8 @@ class MPStat(Monitor): - def __init__(self, interval, job_uuid): - super(MPStat, self).__init__(interval, "mpstat", job_uuid) + def __init__(self, interval, job_uuid, subdir=None): + super(MPStat, self).__init__(interval, "mpstat", job_uuid, subdir=subdir) self.headers = [] def run(self): diff --git a/benchpress/plugins/hooks/perf_monitors/netstat.py b/benchpress/plugins/hooks/perf_monitors/netstat.py index 079e07215..a8c6aa7df 100644 --- a/benchpress/plugins/hooks/perf_monitors/netstat.py +++ b/benchpress/plugins/hooks/perf_monitors/netstat.py @@ -14,8 +14,8 @@ class NetStat(Monitor): - def __init__(self, interval, job_uuid, additional_counters=()): - super(NetStat, self).__init__(interval, "net-stat", job_uuid) + def __init__(self, interval, job_uuid, additional_counters=(), subdir=None): + super(NetStat, self).__init__(interval, "net-stat", job_uuid, subdir=subdir) counters = {"rx_bytes", "rx_packets", "tx_bytes", "tx_packets"} self.counters = counters.union(set(additional_counters)) self.run_collector = False diff --git a/benchpress/plugins/hooks/perf_monitors/perfstat.py b/benchpress/plugins/hooks/perf_monitors/perfstat.py index 9663478c9..4575bffdd 100644 --- a/benchpress/plugins/hooks/perf_monitors/perfstat.py +++ b/benchpress/plugins/hooks/perf_monitors/perfstat.py @@ -39,8 +39,10 @@ def unpack_perf_stat_line(line, delim=","): class PerfStat(Monitor): - def __init__(self, interval, job_uuid, additional_events=(), delim=","): - super(PerfStat, self).__init__(interval, "perf-stat", job_uuid) + def __init__( + self, interval, job_uuid, additional_events=(), delim=",", subdir=None + ): + super(PerfStat, self).__init__(interval, "perf-stat", job_uuid, subdir=subdir) self.events = ["instructions", "cycles"] + list(additional_events) self.delim = delim diff --git a/benchpress/plugins/hooks/perf_monitors/power.py b/benchpress/plugins/hooks/perf_monitors/power.py index f63cccd7d..0ab301a92 100644 --- a/benchpress/plugins/hooks/perf_monitors/power.py +++ b/benchpress/plugins/hooks/perf_monitors/power.py @@ -79,8 +79,8 @@ def get_sensor_avg_power(self, sensor: dict): except OSError as e: return f"" - def __init__(self, job_uuid, interval=1.0, sensor_interval_ms=None): - super(Power, self).__init__(interval, "power", job_uuid) + def __init__(self, job_uuid, interval=1.0, sensor_interval_ms=None, subdir=None): + super(Power, self).__init__(interval, "power", job_uuid, subdir=subdir) self.power_sensors = self.search_sensors() if sensor_interval_ms is None: sensor_interval_ms = 1000 * self.interval diff --git a/benchpress/plugins/hooks/perf_monitors/topdown.py b/benchpress/plugins/hooks/perf_monitors/topdown.py index fb90e5553..800780ce3 100644 --- a/benchpress/plugins/hooks/perf_monitors/topdown.py +++ b/benchpress/plugins/hooks/perf_monitors/topdown.py @@ -13,7 +13,7 @@ import numpy as np import pandas as pd -from benchpress.lib.util import BENCHPRESS_ROOT, get_artifacts_dir +from benchpress.lib.util import BENCHPRESS_ROOT from . import logger, Monitor @@ -142,27 +142,30 @@ def get_os_release(): class IntelPerfSpect(Monitor): - def __init__(self, interval, job_uuid, mux_interval_msecs=125, perfspect_path=None): + def __init__( + self, + interval, + job_uuid, + mux_interval_msecs=125, + perfspect_path=None, + subdir=None, + ): # NOTE: PerfSpect 1.x does not support configurable sampling intervals. # The 'interval' parameter is accepted here only for API compatibility # with the DEFAULT_OPTIONS in perf.py, but it is not actually used # by this implementation. - super(IntelPerfSpect, self).__init__(interval, "perfspect", job_uuid) + super(IntelPerfSpect, self).__init__( + interval, "perfspect", job_uuid, subdir=subdir + ) self.mux_interval_msecs = mux_interval_msecs if perfspect_path is None: self.perfspect_path = os.path.join(BENCHPRESS_ROOT, "perfspect") else: self.perfspect_path = perfspect_path - self.collect_output_path = os.path.join( - get_artifacts_dir(), - "benchmark_metrics_" + self.job_uuid, - "perf-collect.csv", - ) - self.postprocess_output_path = os.path.join( - get_artifacts_dir(), - "benchmark_metrics_" + self.job_uuid, - "topdown-intel.csv", - ) + # Route both artifacts through gen_path so stage-aware mode (subdir set) + # nests them per-stage automatically. + self.collect_output_path = self.gen_path("perf-collect.csv") + self.postprocess_output_path = self.gen_path("topdown-intel.csv") if os.path.exists( os.path.join(self.perfspect_path, "perf-collect") ) and os.path.exists(os.path.join(self.perfspect_path, "perf-postprocess")): @@ -237,21 +240,21 @@ def __init__( job_uuid, mux_interval_msecs=125, perfspect_path=None, + subdir=None, ): - super(IntelPerfSpect3, self).__init__(interval, "perfspect3", job_uuid) + super(IntelPerfSpect3, self).__init__( + interval, "perfspect3", job_uuid, subdir=subdir + ) self.mux_interval_msecs = mux_interval_msecs if perfspect_path is None: self.perfspect_path = os.path.join(BENCHPRESS_ROOT, "perfspect") else: self.perfspect_path = perfspect_path - self.collect_output_path = os.path.join( - get_artifacts_dir(), "benchmark_metrics_" + self.job_uuid - ) - self.postprocess_output_path = os.path.join( - get_artifacts_dir(), - "benchmark_metrics_" + self.job_uuid, - "topdown-intel.sys.csv", - ) + # Routed through gen_path so stage-aware mode (subdir set) nests + # per-stage automatically. The collect dir is the parent benchmark + # metrics dir (perfspect writes its own files inside it). + self.collect_output_path = os.path.dirname(self.gen_path("placeholder")) + self.postprocess_output_path = self.gen_path("topdown-intel.sys.csv") if os.path.exists(os.path.join(self.perfspect_path, "perfspect")): self.supported = True else: @@ -311,8 +314,9 @@ def __init__( perf_postproc_script_name, perfutils_path=None, perf_postproc_args=None, + subdir=None, ): - super(BasePerfUtil, self).__init__(interval, name, job_uuid) + super(BasePerfUtil, self).__init__(interval, name, job_uuid, subdir=subdir) if perfutils_path is None: self.perfutils_path = os.path.join(BENCHPRESS_ROOT, "perfutils") else: @@ -322,16 +326,11 @@ def __init__( self.perf_postproc_addl_args = ( list(perf_postproc_args) if perf_postproc_args else [] ) - self.postprocess_timeseries_output_path = os.path.join( - get_artifacts_dir(), - "benchmark_metrics_" + self.job_uuid, - f"{name}-timeseries.csv", - ) - self.postprocess_summary_output_path = os.path.join( - get_artifacts_dir(), - "benchmark_metrics_" + self.job_uuid, - f"{name}-summary.csv", + # Routed through gen_path so stage-aware mode nests per-stage. + self.postprocess_timeseries_output_path = self.gen_path( + f"{name}-timeseries.csv" ) + self.postprocess_summary_output_path = self.gen_path(f"{name}-summary.csv") def run(self): perf_collect_script = os.path.join( @@ -391,7 +390,7 @@ def write_csv(self): class AMDPerfUtil: - def __init__(self, interval, job_uuid, **kwargs): + def __init__(self, interval, job_uuid, subdir=None, **kwargs): self.cpuinfo = get_cpuinfo() self.cpu_vendor = get_cpu_vendor(self.cpuinfo) if self.cpu_vendor != "amd": @@ -403,6 +402,7 @@ def __init__(self, interval, job_uuid, **kwargs): "amd-perf-collector", perf_collect_script_name="collect_amd_perf_counters.sh", perf_postproc_script_name="generate_amd_perf_report.py", + subdir=subdir, ) if self.amd_gen == "zen4": self.perfutil_zen4 = BasePerfUtil( @@ -412,6 +412,7 @@ def __init__(self, interval, job_uuid, **kwargs): perf_collect_script_name="collect_amd_zen4_perf_counters.sh", perf_postproc_script_name="generate_amd_perf_report.py", perf_postproc_args=["--arch", "zen4"], + subdir=subdir, ) elif self.amd_gen == "zen5": self.perfutil = BasePerfUtil( @@ -421,6 +422,7 @@ def __init__(self, interval, job_uuid, **kwargs): perf_collect_script_name="collect_amd_zen5_perf_counters.sh", perf_postproc_script_name="generate_amd_perf_report.py", perf_postproc_args=["--arch", "zen5"], + subdir=subdir, ) elif self.amd_gen == "zen5es": self.perfutil = BasePerfUtil( @@ -430,6 +432,7 @@ def __init__(self, interval, job_uuid, **kwargs): perf_collect_script_name="collect_amd_zen5_perf_counters.sh", perf_postproc_script_name="generate_amd_perf_report.py", perf_postproc_args=["--arch", "zen5es"], + subdir=subdir, ) def run(self): @@ -456,8 +459,10 @@ class ARMPerfUtil(Monitor): "https://git.gitlab.arm.com/telemetry-solution/telemetry-solution.git" ) - def __init__(self, job_uuid, interval=5, **kwargs): - super(ARMPerfUtil, self).__init__(interval, "arm-perf-collector", job_uuid) + def __init__(self, job_uuid, interval=5, subdir=None, **kwargs): + super(ARMPerfUtil, self).__init__( + interval, "arm-perf-collector", job_uuid, subdir=subdir + ) self.avail = self.install_if_not_available() if not self.avail: logger.warning( @@ -634,13 +639,14 @@ def write_csv(self): class NVPerfUtil(BasePerfUtil): - def __init__(self, interval, job_uuid, **kwargs): + def __init__(self, interval, job_uuid, subdir=None, **kwargs): super(NVPerfUtil, self).__init__( interval, job_uuid, "nv-perf-collector", perf_collect_script_name="collect_nvda_neoversev2_perf_counters.sh", perf_postproc_script_name="generate_arm_perf_report.py", + subdir=subdir, ) def run(self): @@ -648,13 +654,14 @@ def run(self): class NeoVerseV3PerfUtil(BasePerfUtil): - def __init__(self, interval, job_uuid, **kwargs): + def __init__(self, interval, job_uuid, subdir=None, **kwargs): super(NeoVerseV3PerfUtil, self).__init__( interval, job_uuid, "nv3-perf-collector", perf_collect_script_name="collect_neoversev3_perf_counters.sh", perf_postproc_script_name="generate_arm_neoversev3_perf_report.py", + subdir=subdir, ) def run(self): diff --git a/benchpress/plugins/hooks/perf_monitors/vmstat.py b/benchpress/plugins/hooks/perf_monitors/vmstat.py index e4727b415..fd284f316 100644 --- a/benchpress/plugins/hooks/perf_monitors/vmstat.py +++ b/benchpress/plugins/hooks/perf_monitors/vmstat.py @@ -14,8 +14,8 @@ class VMStat(Monitor): - def __init__(self, interval, job_uuid): - super(VMStat, self).__init__(interval, "vmstat", job_uuid) + def __init__(self, interval, job_uuid, subdir=None): + super(VMStat, self).__init__(interval, "vmstat", job_uuid, subdir=subdir) self.run_collector = False def collect_vmstat_snapshot(self): From b63ba6b1e02c5e3353bf9d489d8244fc74f11fa3 Mon Sep 17 00:00:00 2001 From: Yupeng Tang Date: Wed, 8 Jul 2026 23:34:56 -0700 Subject: [PATCH 2/2] Emit per-stage overall metrics for Manifold (#684) Summary: The generic stage-aware perf hook (D108110315) collects PMU + sysstat data in per-stage directories under benchmark_metrics_//. The raw staged CSVs are preserved by perfpub's recursive Manifold upload, but users still need a single processed summary per stage, analogous to the normal `overall-metrics.csv` that perfpub emits for a single benchmark. This change adds stage-aware summary generation to perfpub. It is benchmark-agnostic: any benchmark that drives the stage-aware hook (WDL prod_set, SPEC2017 intrate, ...) produces stage subdirs, and perfpub summarizes each one the same way. benchmark_metrics_/ / # e.g. memcpy_benchmark, 500.perlbench_r overall-metrics.csv # NEW perf-stat.csv nv-perf-collector-summary.csv ... stage_overall_metrics.csv # NEW aggregate index stage_overall_metrics.json # NEW aggregate index stage_perf_summary.csv # generic numeric summary from prior patch stage_perf_summary.json Implementation details: - For each immediate stage subdir that contains CSVs, perfpub temporarily chdirs into that subdir and reuses the same reader functions as the top-level path (`read_mpstat`, `read_memstat`, `read_cpufreq_*`, `read_perfstat`, `read_nv_perf_collector`, `read_arm_perf_collector`, etc.). This gives each stage the same processed metric lines as a normal single benchmark's `overall-metrics.csv`. - Adds the stage's score at the top when the score exists in the parent benchmark metrics JSON. - Writes per-stage `overall-metrics.csv`, plus top-level CSV/JSON indices for easy discovery in Manifold. - No XDB/dashboard changes. The goal is Manifold artifact usability. - `sample_avg_from_csv()` now tolerates partial CSV schemas by selecting only requested metric columns that exist and warning about missing ones. This is useful for stage dirs where a monitor didn't emit the full standard set of columns. Reviewed By: YifanYuan3 Differential Revision: D108195608 --- perfpub/utils.py | 331 ++++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 328 insertions(+), 3 deletions(-) diff --git a/perfpub/utils.py b/perfpub/utils.py index bfe642916..2c19173cb 100644 --- a/perfpub/utils.py +++ b/perfpub/utils.py @@ -194,6 +194,320 @@ def find_closest_timestamp_index(metric_times, target_datetime): return closest_idx +STAGE_PERF_SUMMARY_CSV = "stage_perf_summary.csv" +STAGE_PERF_SUMMARY_JSON = "stage_perf_summary.json" + + +def _numeric_series(series): + """Coerce a pandas Series to numeric values, tolerating comma separators.""" + if series.dtype == object: + series = series.astype(str).str.replace(",", "", regex=False) + return pd.to_numeric(series, errors="coerce").dropna() + + +def _summarize_numeric_series(series): + values = _numeric_series(series) + if values.empty: + return None + return { + "count": int(values.count()), + "mean": float(values.mean()), + "min": float(values.min()), + "p50": float(values.quantile(0.50)), + "p95": float(values.quantile(0.95)), + "max": float(values.max()), + } + + +def generate_stage_perf_summary(): + """Generate compact summary artifacts for per-stage perf data. + + The benchpress perf hook's stage-aware mode writes monitor CSVs under + immediate subdirectories of ``benchmark_metrics_/``: + + benchmark_metrics_//perf-stat.csv + benchmark_metrics_//mpstat.csv + benchmark_metrics_//nv-perf-collector-summary.csv + ... + + PerfPub already uploads the whole directory tree to Manifold via + ``manifold putr``. This helper adds two easy-to-discover top-level + artifacts before that upload happens: + + stage_perf_summary.csv + stage_perf_summary.json + + No XDB schema change is required. The summary is intentionally generic: + for every immediate subdirectory, every CSV file, every numeric column, + compute count/mean/min/p50/p95/max. + + Returns: + List of generated summary filenames. Empty when no per-stage CSVs + are found. + """ + rows = [] + json_summary = {"stages": {}} + + # Do not recurse into nested timestamp dirs produced by some monitors; + # stage-aware perf writes stage dirs directly under cwd. + for stage in sorted( + d for d in os.listdir(".") if os.path.isdir(d) and not d.startswith(".") + ): + csv_files = sorted( + f + for f in os.listdir(stage) + if f.endswith(".csv") + and f not in {STAGE_PERF_SUMMARY_CSV, STAGE_PERF_SUMMARY_JSON} + ) + if not csv_files: + continue + + stage_json = {"files": {}} + for csv_file in csv_files: + path = os.path.join(stage, csv_file) + try: + df = pd.read_csv(path) + except Exception as e: + print(f"Warning: failed to read stage perf CSV {path}: {e}") + continue + + file_json = {"rows": int(len(df)), "metrics": {}} + for col in df.columns: + stats = _summarize_numeric_series(df[col]) + if stats is None: + continue + file_json["metrics"][col] = stats + rows.append( + { + "stage": stage, + "file": csv_file, + "metric": col, + **stats, + } + ) + if file_json["metrics"]: + stage_json["files"][csv_file] = file_json + + if stage_json["files"]: + json_summary["stages"][stage] = stage_json + + if not rows: + return [] + + with open(STAGE_PERF_SUMMARY_JSON, "w") as f: + json.dump(json_summary, f, indent=2, sort_keys=True) + + pd.DataFrame(rows).to_csv(STAGE_PERF_SUMMARY_CSV, index=False) + print( + f"Generated per-stage perf summaries: {STAGE_PERF_SUMMARY_CSV}, " + f"{STAGE_PERF_SUMMARY_JSON} ({len(json_summary['stages'])} stages, " + f"{len(rows)} numeric metrics)" + ) + return [STAGE_PERF_SUMMARY_CSV, STAGE_PERF_SUMMARY_JSON] + + +STAGE_OVERALL_METRICS_CSV = "stage_overall_metrics.csv" +STAGE_OVERALL_METRICS_JSON = "stage_overall_metrics.json" +STAGE_OVERALL_METRICS_FILENAME = "overall-metrics.csv" + + +def _build_stage_overall_metrics_text(stage, bm_metrics, interval): + """Build an overall-metrics.csv-style text blob for one stage directory. + + This mirrors the normal top-level overall-metrics.csv generated by + process_metrics(), but is intentionally limited to the perf/sysstat data + available inside one stage-aware benchmark stage directory. The + caller must chdir into the stage directory before calling this helper. + """ + res = "" + # Include the stage score at the top when the benchmark parser + # reported one in the top-level metrics JSON. + try: + metrics = bm_metrics.get("metrics", {}) if bm_metrics else {} + if stage in metrics: + res += f'score,"{metrics[stage]}"\n' + except Exception: + pass + + # Stage directories already represent the exact sub-benchmark window, so + # process the whole CSV (last_secs=0, skip_last_secs=0) instead of slicing + # by the parent benchmark's breakdown.csv window. + last_secs = 0 + skip_last_secs = 0 + bm_epoch = None + start_time = None + end_time = None + + res += read_mpstat( + interval, last_secs, skip_last_secs, start_time, end_time, bm_epoch + ) + res += read_memstat( + interval, last_secs, skip_last_secs, start_time, end_time, bm_epoch + ) + res += read_cpufreq_scaling( + interval, last_secs, skip_last_secs, start_time, end_time, bm_epoch + ) + res += read_cpufreq_cpuinfo( + interval, last_secs, skip_last_secs, start_time, end_time, bm_epoch + ) + res += read_netstat( + interval, last_secs, skip_last_secs, start_time, end_time, bm_epoch + ) + res += read_perfstat( + interval, last_secs, skip_last_secs, start_time, end_time, bm_epoch + ) + res += read_amd_perf_collector( + interval, last_secs, skip_last_secs, start_time, end_time, bm_epoch + ) + res += read_amd_zen4_perf_collector( + interval, last_secs, skip_last_secs, start_time, end_time, bm_epoch + ) + res += read_amd_zen5_perf_collector( + interval, last_secs, skip_last_secs, start_time, end_time, bm_epoch + ) + res += read_nv_perf_collector( + interval, last_secs, skip_last_secs, start_time, end_time, bm_epoch + ) + res += read_arm_perf_collector( + interval, last_secs, skip_last_secs, start_time, end_time, bm_epoch + ) + res += read_neoversev3_perf_collector( + interval, last_secs, skip_last_secs, start_time, end_time, bm_epoch + ) + res += read_intel_perfspect( + interval, last_secs, skip_last_secs, start_time, end_time, bm_epoch + ) + # Include collector-provided PMU summaries (cache MPKI, TopDown %, BW, + # latency, etc.) such as nv-perf-collector-summary.csv. + res += _read_stage_summary_csvs() + return res + + +def _read_stage_summary_csvs(): + """Read *-summary.csv files in the current stage directory and emit + overall-metrics-style key,value lines. + + Stage-aware WDL perf collection produces collector summaries such as + nv-perf-collector-summary.csv. Those files contain PMU-derived metrics like + cache MPKI, TopDown BackendBound %, memory bandwidth, etc. The existing + perfpub readers primarily consume timeseries files, so without this helper + the per-stage overall-metrics.csv misses the most useful PMU summary rows. + + For each row in each summary CSV, emit only: + metric, + + This matches the existing perfpub overall-metrics.csv convention (one + representative value per metric). The richer count/min/p50/p95/max + distribution remains available in the top-level stage_perf_summary + artifacts, not in each stage's overall-metrics.csv. + """ + res = "" + summary_files = sorted( + f + for f in os.listdir(".") + if f.endswith("-summary.csv") or f.endswith("_summary.csv") + ) + for summary_file in summary_files: + try: + df = pd.read_csv(summary_file) + except Exception as e: + print(f"Warning: failed to read stage summary CSV {summary_file}: {e}") + continue + if "metric" not in df.columns or "mean" not in df.columns: + continue + for _, row in df.iterrows(): + metric = str(row.get("metric", "")).strip() + if not metric: + continue + # Preserve the simple key for the mean because that's what users + # expect in overall-metrics.csv. + mean = row.get("mean") + if pd.notna(mean): + res += f'{metric},"{mean}"\n' + return res + + +def _parse_overall_metrics_text(text): + """Parse key,value lines from an overall-metrics.csv-style blob.""" + out = {} + for line in text.splitlines(): + if not line or "," not in line: + continue + key, value = line.split(",", 1) + key = key.strip() + value = value.strip().strip('"') + if not key: + continue + out[key] = value + return out + + +def generate_stage_overall_metrics(bm_metrics=None, interval=5): + """Generate per-stage overall-metrics artifacts for WDL prod_set. + + For every immediate subdirectory with perf/sysstat CSV files, write: + + /overall-metrics.csv + + and also create top-level aggregate indices: + + stage_overall_metrics.csv + stage_overall_metrics.json + + This gives Manifold users the same processed PMU/sysstat summary they get + from a normal single-benchmark perfpub run, but scoped to each WDL + sub-benchmark stage. + + Returns: + List of generated files (top-level aggregate files plus per-stage + overall-metrics.csv paths). Empty when no stages are found. + """ + generated = [] + summary_rows = [] + summary_json = {"stages": {}} + root = os.getcwd() + + for stage in sorted( + d for d in os.listdir(".") if os.path.isdir(d) and not d.startswith(".") + ): + csv_files = [f for f in os.listdir(stage) if f.endswith(".csv")] + if not csv_files: + continue + + try: + os.chdir(os.path.join(root, stage)) + text = _build_stage_overall_metrics_text(stage, bm_metrics or {}, interval) + finally: + os.chdir(root) + + if not text.strip(): + continue + + stage_overall_path = os.path.join(stage, STAGE_OVERALL_METRICS_FILENAME) + with open(stage_overall_path, "w") as f: + f.write(text) + generated.append(stage_overall_path) + + metrics = _parse_overall_metrics_text(text) + summary_json["stages"][stage] = metrics + for metric, value in metrics.items(): + summary_rows.append({"stage": stage, "metric": metric, "value": value}) + + if not summary_rows: + return generated + + with open(STAGE_OVERALL_METRICS_JSON, "w") as f: + json.dump(summary_json, f, indent=2, sort_keys=True) + pd.DataFrame(summary_rows).to_csv(STAGE_OVERALL_METRICS_CSV, index=False) + generated.extend([STAGE_OVERALL_METRICS_CSV, STAGE_OVERALL_METRICS_JSON]) + print( + f"Generated per-stage overall metrics: {STAGE_OVERALL_METRICS_CSV}, " + f"{STAGE_OVERALL_METRICS_JSON} ({len(summary_json['stages'])} stages, " + f"{len(summary_rows)} metrics)" + ) + return generated + + def read_benchmark_metrics(): metrics_jsons = glob.glob("*_metrics_*.json") if len(metrics_jsons) == 0: @@ -392,7 +706,15 @@ def sample_avg_from_csv( return "" samples.to_csv(filename.split(".", maxsplit=1)[0] + ".sampled.csv") if metrics: - samples = samples[metrics] + present_metrics = [metric for metric in metrics if metric in samples.columns] + missing_metrics = [ + metric for metric in metrics if metric not in samples.columns + ] + if missing_metrics: + print(f"Columns {missing_metrics} not found in {filename}") + if not present_metrics: + return "" + samples = samples[present_metrics] if exclude_columns: for excl in exclude_columns: if excl in samples: @@ -659,12 +981,15 @@ def process_metrics( if os.path.exists(breakdown_path): start_time, end_time = parse_breakdown_csv(breakdown_path) - columns = "(" - # values = "(" db_fields = {} bm_metrics = read_benchmark_metrics() if not bm_metrics: return "" + # Generate compact per-stage perf summary artifacts before internal + # processing uploads the whole benchmark_metrics directory to Manifold. + # This is a no-op for benchmarks without stage-aware perf subdirectories. + generate_stage_perf_summary() + generate_stage_overall_metrics(bm_metrics, args.interval) bm_name = bm_metrics["benchmark_name"] db_fields["benchmark_name"] = f'"{bm_name}"' bm_epoch = datetime.fromtimestamp(bm_metrics["timestamp"])