HPCE/slurm-efficiency-tools
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

first complete histogramming release (terminal output)

Browse Source
This commit is contained in:
feichtinger
2026-07-08 13:36:46 +02:00
parent 8029058381
commit b2ddc5d33f
1 changed files with 97 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files
slurm-eff-tool.py
+97 -74
View File
@@ -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