From b2ddc5d33ff31edea4a460eda76e655046e9d27c Mon Sep 17 00:00:00 2001 From: Derek Feichtinger Date: Tue, 7 Jul 2026 17:09:53 +0200 Subject: [PATCH] first complete histogramming release (terminal output) --- slurm-eff-tool.py | 171 ++++++++++++++++++++++++++-------------------- 1 file changed, 97 insertions(+), 74 deletions(-) diff --git a/slurm-eff-tool.py b/slurm-eff-tool.py index 45f50f6..daaf163 100755 --- a/slurm-eff-tool.py +++ b/slurm-eff-tool.py @@ -44,7 +44,7 @@ from typing import Any, Iterable, Mapping, Union, List, Tuple, cast import msgpack import gzip -VERSION=0.23 +VERSION=0.3 # Default GB per CPUs for the cluster default_mempercpu_gb = 2.2 @@ -233,6 +233,16 @@ NUMERIC_COLUMNS = { "waste_total", } +HISTO_COLUMNS = { + "CPU_Eff", + "Elig_Qtime", + "Mem_Eff", + "Planned_Time", + "Time_Eff", + "UsesMem", + "Walltime", + } + state_mappings = { "BOOT_FAIL": 0, "CANCELLED": 1, @@ -916,7 +926,12 @@ def aggregate_records(records: list[JobRecord], args: argparse.Namespace) -> lis if args.dflt_mpcpu: default_mempercpu_gb = float(args.dflt_mpcpu) - storvecs=args.histo.lower().split(",") + storvecs = [] + if args.histo: + if not (args.aggr_regexp or args.aggr_user): + sys.stderr.write("ERROR: Cannot use histogramming mode without an aggregation mode\n") + sys.exit(1) + storvecs=args.histo.lower().split(",") if args.aggr_regexp: compiled = [(pat, re.compile(pat)) for pat in args.aggr_regexp] @@ -1010,14 +1025,14 @@ def make_single_row(rec: JobRecord, dflt_mpcpu: float, vectors[colname] = [] if colname == "walltime": vectors[colname] = [rec.elapsed_sec/3600] - elif colname == "maxrss" and rec.maxrss_bytes is not None: - vectors[colname] = [rec.maxrss_bytes / (1024**3)] elif colname == "used_mem" and rec.mem_used_tres is not None: vectors[colname] = [rec.mem_used_tres] elif colname == "elig_qtime": vectors[colname] = [rec.elig_qtime_sec/3600] elif colname == "planned_time": vectors[colname] = [rec.planned_sec/3600] + elif colname in ["cpu_eff", "mem_eff", "time_eff"]: + pass else: sys.stderr.write(f"ERROR: Cannot gather vector information for metric {colname}\n") sys.exit(1) @@ -1064,6 +1079,8 @@ def make_aggregate_row(records: list[JobRecord], username: str, partition: str, walltime=mean_or_none([r.elapsed_sec for r in records if r.elapsed_sec is not None]) if walltime is not None: walltime /= 3600 + walltime_max=max([r.elapsed_sec for r in records + if r.elapsed_sec is not None]) / 3600 planned_sec = mean_or_none([r.planned_sec for r in records if r.planned_sec is not None]) @@ -1083,6 +1100,8 @@ def make_aggregate_row(records: list[JobRecord], username: str, partition: str, # requested should match what the scheduler gave alloc_mem = mean_or_none([r.mem_alloc_tres for r in records if r.mem_alloc_tres is not None]) used_mem = mean_or_none([r.mem_used_tres for r in records if r.mem_used_tres is not None]) + maxrss_max = max([r.maxrss_bytes for r in records + if r.maxrss_bytes is not None]) / (1024**3) or None memory_efficiency=mean_or_none(mem_eff_vals) count=len(records) @@ -1115,9 +1134,6 @@ def make_aggregate_row(records: list[JobRecord], username: str, partition: str, if colname == "walltime": vectors[colname] = [r.elapsed_sec/3600 for r in records if r.elapsed_sec is not None] - elif colname == "maxrss": - vectors[colname] = [r.maxrss_bytes / (1024**3) for r in records - if r.maxrss_bytes is not None] elif colname == "used_mem": vectors[colname] = [r.mem_used_tres for r in records if r.mem_used_tres is not None] @@ -1127,13 +1143,11 @@ def make_aggregate_row(records: list[JobRecord], username: str, partition: str, elif colname == "planned_time": vectors[colname] = [r.planned_sec/3600 for r in records if r.planned_sec is not None] + elif colname in ["cpu_eff", "mem_eff", "time_eff"]: + pass else: sys.stderr.write(f"ERROR: Cannot gather vector information for metric {colname}\n") sys.exit(1) - #planned_time, elig_qtime, - # vectors[colname] = [getattr(r,colname) for r in records - # if getattr(r,colname) is not None] - return OutputRow( username=username, @@ -1155,9 +1169,9 @@ def make_aggregate_row(records: list[JobRecord], username: str, partition: str, Mem_Eff=memory_efficiency, Time_Eff=mean_or_none(time_eff_vals), jobname=jobname, - maxrss_max=max([r.maxrss_bytes for r in records if r.maxrss_bytes is not None]) / (1024**3) or None, + maxrss_max=maxrss_max, walltime=walltime, - walltime_max=max([r.elapsed_sec for r in records if r.elapsed_sec is not None]) / 3600, + walltime_max=walltime_max, waste_Mem=waste_mem, waste_CPU=waste_cpu, waste_total=waste_total, @@ -1224,6 +1238,8 @@ def columns_for_sdev(base_cols: list[str]) -> list[str]: def format_value(value: Any, column: str | None = None) -> str: if value is None: return "NA" + if column == "Nr.": + return f"{value})" if column in ["ReqMem", "UsedMem","MemPerCPU", "MaxRSS_max", "AllocMem"]: return f"{value:.2f}G" if column in ["ReqWalltime", "Walltime", "Walltime_max", @@ -1239,6 +1255,13 @@ def print_table(rows: list[OutputRow], columns: list[str], sdev: bool) -> None: dicts = [r.as_dict(sdev=sdev) for r in rows] columns = columns_for_sdev(columns) if sdev else columns + # Add row numbers + counter = 0 + for d in dicts: + d['Nr.'] = counter + counter += 1 + columns = ['Nr.'] + columns + widths = {col: len(col) for col in columns} for d in dicts: for col in columns: @@ -1418,6 +1441,61 @@ def histogram_table( ] return edges, counts + +def histo_graphs(out_rows: list[OutputRow], colnames: list[str]): + for metric in colnames: + tmpfile = tempfile.NamedTemporaryFile(mode='w+t', delete=False, + dir='.',prefix=f"tmp-seff-{metric}", + suffix=".data") + + # create bins + edges, counts = histogram_table([getattr(row,"vectors")[metric] for row in out_rows]) + # currently we name the aggregations with numbers + names = [str(x) for x in range (0, len(out_rows))] + + tmpfile.write(f"# bin_center {' '.join(names)}\n") + for i in range(len(edges) - 1): + left = edges[i] + right = edges[i + 1] + center = (left + right) / 2.0 + # TODO: for log-bins: center = (left * right) ** 0.5 + + tmpfile.write(f"{center:.6g}") + for colnum in range(0,len(names)): + tmpfile.write(f" {counts[colnum][i]}") + tmpfile.write("\n") + tmpfile.close() + + gnuplot_cmd = f"""set title "{metric} distribution" noenhanced +set xlabel "{metric}" noenhanced +set ylabel "Jobs" +set style fill transparent solid 0.35 noborder +set encoding utf8 +# set terminal dumb size 120, 30 +set terminal dumb enhanced size 120, 30 + +set colorsequence classic +set terminal dumb ansi size 120, 30 +plot_cmd = 'plot "{tmpfile.name}" \\ +""" + # gnuplot_cmd += f' using 1:2 with boxes title "{names[0]}" \\\n' + # for col in range(1,len(names)): + # gnuplot_cmd += \ + # f', "" using 1:{col+2} with boxes title "{names[col]}" \\\n' + + gnuplot_cmd += f' using 1:2 with steps lw 2 title "{names[0]}" \\\n' + for col in range(1,len(names)): + gnuplot_cmd += \ + f', "" using 1:{col+2} with steps lw 2 title "{names[col]}" \\\n' + + gnuplot_cmd += "'\neval plot_cmd\n" + # print(f"DEBUG: GNUPLOT_SCRIPT:\n{gnuplot_cmd}") + run(["gnuplot"], + input=gnuplot_cmd, + text=True, + check=True,) + os.remove(tmpfile.name) + # HISTOGRAMMING CODE END def parse_args(argv: list[str]) -> argparse.Namespace: @@ -1479,7 +1557,7 @@ def parse_args(argv: list[str]) -> argparse.Namespace: p.add_argument("-O", "--output-raw", help="write raw sacct output cache to this file (text format, large).") p.add_argument("-F", "--from-raw", help="read raw sacct output cache from this file.") p.add_argument("-i", "--info", help="show metadata information for the given binary cache file") - p.add_argument("-H", "--histo", help="print histogram of the given metric.", default=None) + p.add_argument("-H", "--histo", help=f"print histogram of the given metric (Supported columns: {', '.join(HISTO_COLUMNS)}).", default=None) p.add_argument("--dflt-mpcpu", help=f"Default memory/CPU ratio of cluster [{default_mempercpu_gb} GB/cpu]. Used in memory waste calculation.") p.add_argument("--sdev", action="store_true", help="after each efficiency average, add sdev, max, and min columns") @@ -1560,7 +1638,9 @@ def main(argv: list[str] | None = None) -> int: write_binary_cache(records, args.write_binary_cache) # Aggregate - out_rows = aggregate_records(records, args) # , dflt_mpcpu=default_mempercpu_gb + out_rows = aggregate_records(records, args) + + # filter rows if args.expr: expr = SafeExpression(args.expr) out_rows = [row for row in out_rows \ @@ -1574,66 +1654,9 @@ def main(argv: list[str] | None = None) -> int: else: print_table(out_rows, output_columns, args.sdev) + # Histogramming if args.histo: - # DEBUG - # for row in out_rows: - # rowdict = row.as_dict() - # for colname, vals in rowdict["vectors"].items(): - # print(f'DEBUG #### {colname}: {vals}') - # - - for metric in args.histo.lower().split(','): - tmpfile = tempfile.NamedTemporaryFile(mode='w+t', delete=False, - dir='.',prefix=f"tmp-seff-{metric}", - suffix=".data") - - # create bins - edges, counts = histogram_table([getattr(row,"vectors")[metric] for row in out_rows]) - # currently we name the aggregations with numbers - names = [str(x) for x in range (0, len(out_rows))] - - tmpfile.write(f"# bin_center {' '.join(names)}\n") - for i in range(len(edges) - 1): - left = edges[i] - right = edges[i + 1] - center = (left + right) / 2.0 - # TODO: for log-bins: center = (left * right) ** 0.5 - - tmpfile.write(f"{center:.6g}") - for colnum in range(0,len(names)): - tmpfile.write(f" {counts[colnum][i]}") - tmpfile.write("\n") - tmpfile.close() - - gnuplot_cmd = f"""set title "{metric} distribution" -set xlabel "{metric}" -set ylabel "Jobs" -set style fill transparent solid 0.35 noborder -set encoding utf8 -# set terminal dumb size 120, 30 -set terminal dumb enhanced size 120, 30 - -set colorsequence classic -set terminal dumb ansi size 120, 30 -plot_cmd = 'plot "{tmpfile.name}" \\ -""" - # gnuplot_cmd += f' using 1:2 with boxes title "{names[0]}" \\\n' - # for col in range(1,len(names)): - # gnuplot_cmd += \ - # f', "" using 1:{col+2} with boxes title "{names[col]}" \\\n' - - gnuplot_cmd += f' using 1:2 with steps lw 2 title "{names[0]}" \\\n' - for col in range(1,len(names)): - gnuplot_cmd += \ - f', "" using 1:{col+2} with steps lw 2 title "{names[col]}" \\\n' - - gnuplot_cmd += "'\neval plot_cmd\n" - # print(f"DEBUG: GNUPLOT_SCRIPT:\n{gnuplot_cmd}") - run(["gnuplot"], - input=gnuplot_cmd, - text=True, - check=True,) - os.remove(tmpfile.name) + histo_graphs(out_rows, args.histo.lower().split(',')) return 0