import time
import psutil
import os
[docs]
class conesProfiler:
"""
A ligthweight profiler tracking wall-clock time spent in the Solver, MPI_Transfer
and DA phases of a DA cycle, and the peak resident memory
usage, dumped to a csv report.
"""
def __init__(self, n_ensemble, n_decomp, run_id=None):
"""
:param n_ensemble: Number of ensemble members
:type n_ensemble: int
:param n_decomp: Number of domain decomposition ranks
:type n_decomp: int
:param run_id: identifier shared with par.log as returned by
cones_run_id(): keeps the timings joinable with the estimated
parameters and keeps successive runs apart in the csv
:type run_id: str
"""
self.n_ensemble = n_ensemble
self.n_decomp = n_decomp
self.run_id = run_id if run_id else "unset"
self.timestamps = {}
self.results = {
"run_id": self.run_id,
"cycle": 0,
"N": n_ensemble,
"P": n_decomp,
"Time_Solver": 0,
"Time_MPI_Transfer": 0,
"Time_DA": 0,
"Peak_RAM_GB": 0
}
[docs]
def reset_cycle(self):
"""
Zero the per-phase timers for the next cycle.
Peak_RAM_GB is intentionally not reset (run-lvel running maximum.
"""
self.results["Time_Solver"] = 0
self.results["Time_MPI_Transfer"] = 0
self.results["Time_DA"] = 0
[docs]
def start(self, label):
"""
Start timing a phase (e.g. "MPI_Transfer"
:param label: Name of the phase being timed
:type label: str
:returns: None
"""
self.timestamps[label] = time.perf_counter()
[docs]
def stop(self, label):
"""
Stop timing a phase, accmulate its duration and update the peak RAM usage
:param label: Name of the phase started with :func:'start'
:type label: str
:returns: None
"""
if label in self.timestamps:
duration = time.perf_counter() - self.timestamps[label]
key = f"Time_{label}"
if key in self.results:
self.results[key] += duration
# RAM monitoring (in GB)
process = psutil.Process(os.getpid())
current_mem = process.memory_info().rss / (1024 ** 3)
if current_mem > self.results["Peak_RAM_GB"]:
self.results["Peak_RAM_GB"] = current_mem
[docs]
def save_to_csv(self, filename=None):
"""
Append the accumulated results as one row to a CSV file,
writing the header first if the file does not exist yet
:param filename: output CSV file path. When omitted, it is derived
from the run id, so each run gets its own file.
:type filename: str
:returns: None
"""
if filename is None:
filename = "performance_%s.csv" % self.run_id if self.run_id != "unset" else "performance_results.csv"
file_exists = os.path.isfile(filename)
with open(filename, 'a') as f:
if not file_exists:
f.write(",".join(self.results.keys()) + "\n")
values = [v if isinstance(v, str) else str(round(v, 4))
for v in self.results.values()]
f.write(",".join(values) + "\n")
print(f"[Profiler] Data saved to {filename}")