This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hudi-rs.git
The following commit(s) were added to refs/heads/main by this push:
new 54364277 fix(benchmark): gate on recent other-process load and report
only finished cases (#967)
54364277 is described below
commit 543642771a1440f8470053831463052e190b3923
Author: Y Ethan Guo <[email protected]>
AuthorDate: Tue Nov 3 00:46:44 2026 -0800
fix(benchmark): gate on recent other-process load and report only finished
cases (#967)
---
benchmark/upsert/README.md | 19 ++--
benchmark/upsert/bench.py | 165 ++++++++++++++++++++++++++++---
benchmark/upsert/summarize.py | 52 +++++++---
benchmark/upsert/tests/test_measure.py | 127 +++++++++++++++++++++++-
benchmark/upsert/tests/test_summarize.py | 95 ++++++++++++++++++
5 files changed, 418 insertions(+), 40 deletions(-)
diff --git a/benchmark/upsert/README.md b/benchmark/upsert/README.md
index b9125fbd..88ea1471 100644
--- a/benchmark/upsert/README.md
+++ b/benchmark/upsert/README.md
@@ -202,12 +202,19 @@ counts must match across repetitions and byte counts
within 1 percent,
otherwise `summary.writes_consistent` is false.
**Headroom gate.** Before each case and each timed call, the harness refuses to
-go on if the 1-minute load average is above `--max-load1` (2.0), available
-memory is below `--min-mem-gib` (16), or free disk is below 3x the base data.
-A high load average alone is first given `--settle-timeout` seconds (180) to
-come down. `--force` turns load and memory refusals into warnings and marks
-the results, which `summarize.py` then renders only with `--allow-loaded`; a
-disk shortfall always refuses.
+go on if other processes have kept more than `--max-load` cores (2.0) busy over
+the last 10 seconds, available memory is below `--min-mem-gib` (16), or free
+disk is below 3x the base data. The load leaves out the CPU time of the
+benchmark and its phase subprocesses, so one commit's work never holds up the
+next; the 1-minute load average is recorded alongside for reference. Each
+phase subprocess's first check waits until it has 10 seconds of samples (it
+starts sampling when it starts, so most of that overlaps reading its inputs).
+Memory is read while the phase process holds its data, so `--min-mem-gib` is
+headroom beyond the benchmark's own footprint. High load alone is first given
+`--settle-timeout` seconds (180) to come down. `--force` turns load and memory
+refusals into warnings, judged over a 1-second window, and marks the results,
+which `summarize.py` then renders only with `--allow-loaded`; a disk shortfall
+always refuses.
**Reclaim.** After each commit, outside the timed region, the harness deletes
data files the latest file slices no longer reference, so a copy-on-write run
diff --git a/benchmark/upsert/bench.py b/benchmark/upsert/bench.py
index 1d945c9a..91ad3c0b 100755
--- a/benchmark/upsert/bench.py
+++ b/benchmark/upsert/bench.py
@@ -44,6 +44,7 @@ import statistics
import subprocess
import sys
import tempfile
+import threading
import time
import traceback
import types
@@ -213,7 +214,7 @@ class HudiCase(Case):
"target_file_size": "library default",
"upsert": "HudiTable.upsert(arrow_table)",
"delete": "HudiTable.delete_keys(keys)",
- "read": "HudiTable.read()",
+ "read": "HudiTable.read(); every read projects the data columns
(get_schema)",
"incremental_read": (
"HudiTable.read(HudiReadOptions().with_query_type(Incremental)"
".with_start_timestamp(<previous completion time>)"
@@ -253,6 +254,9 @@ class HudiCase(Case):
def open(self, root, plan):
self.plan = plan
self.t = self._hudi().HudiTable(str(root))
+ # Reads project the data columns, as the other cases return them; the
+ # five _hoodie_* meta columns would otherwise be decoded on every read.
+ self.columns = self.t.get_schema().names
self.last = None
@staticmethod
@@ -286,6 +290,7 @@ class HudiCase(Case):
return {"num_deleted": r.num_deletes, "instant": r.requested_time}
def _read(self, opts=None):
+ opts = (opts or
self._hudi().HudiReadOptions()).with_projection(self.columns)
batches = self.t.read(opts)
if not batches:
return None
@@ -401,16 +406,144 @@ def reclaim(case, root, after):
return {"reclaimed_files": len(stale), "reclaimed_bytes": freed}
+# Window over which the gate averages the CPU other processes use.
+LOAD_WINDOW_S = 10.0
+
+
+class OtherLoad:
+ """Cores kept busy by other processes over the last LOAD_WINDOW_S seconds.
+
+ The one-minute load average lags and counts the benchmark's own work, so a
+ gate on it waits for each commit's own load to decay. This samples system
+ busy CPU time once a second and subtracts the CPU time of this process and
+ its descendants: live ones through psutil, reaped ones through
+ getrusage(RUSAGE_CHILDREN). psutil's children_* fields cannot stand in for
+ the latter; they read 0 on macOS even after the child is reaped."""
+
+ def __init__(self):
+ self.samples = []
+ self.lock = threading.Lock()
+ threading.Thread(target=self._run, daemon=True).start()
+
+ @staticmethod
+ def sample():
+ """(time, system busy s, own CPU s), or None when no consistent reading
+ of this process tree's CPU could be taken."""
+ own = own_cpu_s()
+ return None if own is None else (time.monotonic(), system_busy_s(),
own)
+
+ def _run(self):
+ while True:
+ self._record()
+ time.sleep(1.0)
+
+ def _record(self):
+ # A sampler that died would leave its last sample as the start of every
+ # later window, averaging recent load away, so a failed sample is
skipped.
+ try:
+ s = self.sample()
+ except Exception:
+ return
+ if s is None:
+ return
+ with self.lock:
+ self.samples.append(s)
+ self.samples = [x for x in self.samples if s[0] - x[0] <= 2 *
LOAD_WINDOW_S]
+
+ def read(self, window):
+ """(busy cores, seconds covered) over the latest ``window`` seconds;
+ blocks until that much history exists, which only a process's first
+ reading waits for."""
+ deadline = time.monotonic() + window + 60
+ while True:
+ now = self.sample()
+ if now is not None:
+ with self.lock:
+ old = [x for x in self.samples if now[0] - x[0] >= window]
+ if old:
+ return busy_cores(old[-1], now), now[0] - old[-1][0]
+ elif time.monotonic() > deadline:
+ raise RuntimeError(
+ "could not read the benchmark's own CPU time consistently;
"
+ "an exited descendant is not being reaped"
+ )
+ time.sleep(0.5)
+
+
+def system_busy_s():
+ """Busy CPU seconds across all cores, computed as psutil's cpu_percent
does:
+ iowait is idle time, and guest time is already inside user time."""
+ t = psutil.cpu_times()
+ busy = sum(t) - t.idle
+ for field in ("iowait", "guest", "guest_nice"):
+ busy -= getattr(t, field, 0.0)
+ return busy
+
+
+def reaped_cpu_s():
+ r = resource.getrusage(resource.RUSAGE_CHILDREN)
+ return r.ru_utime + r.ru_stime
+
+
+def own_cpu_s(attempts=20):
+ """CPU seconds used by this process and every descendant, live or reaped,
or
+ None when no consistent reading was taken. A descendant moves from the live
+ sum to the reaped one when it is waited for. One that exits but is not yet
+ reaped, or is reaped while the tree is walked, is in neither sum and would
+ hide a whole phase's CPU, so such a walk is taken again."""
+ for _ in range(attempts):
+ reaped = reaped_cpu_s()
+ live, consistent = 0.0, True
+ for child in psutil.Process().children(recursive=True):
+ try:
+ c = child.cpu_times()
+ except psutil.NoSuchProcess: # also a zombie, on macOS
+ consistent = False
+ break
+ except psutil.AccessDenied:
+ continue
+ live += c.user + c.system
+ if consistent and reaped_cpu_s() == reaped:
+ me = resource.getrusage(resource.RUSAGE_SELF)
+ return me.ru_utime + me.ru_stime + reaped + live
+ time.sleep(0.01)
+ return None
+
+
+def busy_cores(start, end):
+ """Cores other processes kept busy between two OtherLoad samples."""
+ (t0, busy0, own0), (t1, busy1, own1) = start, end
+ return max(0.0, ((busy1 - busy0) - (own1 - own0)) / (t1 - t0))
+
+
+_other_load = None
+
+
+def other_load(window=LOAD_WINDOW_S):
+ global _other_load
+ if _other_load is None:
+ _other_load = OtherLoad()
+ return _other_load.read(window)
+
+
def headroom(work_dir, base_bytes, limits):
+ # Under --force a load reading only warns, so it need not wait for a full
+ # window of history.
+ load, window = other_load(1.0 if limits.force else LOAD_WINDOW_S)
avail = psutil.virtual_memory().available
r = {
+ "other_load": load,
+ "other_load_window_s": round(window, 1),
"load1": os.getloadavg()[0],
"mem_available_bytes": avail,
"disk_free_bytes": shutil.disk_usage(work_dir).free,
}
v = {}
- if r["load1"] > limits.max_load1:
- v["load1"] = f"load1 {r['load1']:.2f} > {limits.max_load1}"
+ if r["other_load"] > limits.max_load:
+ v["load"] = (
+ f"other processes kept {r['other_load']:.2f} cores busy over the
last "
+ f"{r['other_load_window_s']:.0f}s > {limits.max_load}"
+ )
if avail < limits.min_mem_gib * 2**30:
v["memory"] = (
f"available memory {avail / 2**30:.1f} GiB < {limits.min_mem_gib}
GiB"
@@ -427,10 +560,10 @@ def headroom(work_dir, base_bytes, limits):
def gate(work_dir, base_bytes, limits, what):
"""Wait for the machine to be idle, then return the readings.
- load1 is a one-minute moving average that the benchmark itself raises, so
- when it is the only violation the gate polls for up to --settle-timeout,
- letting the previous commit's own load decay instead of counting it against
- the next; under --force, where a load violation only warns, it does not
+ Load is the CPU other processes used over the last LOAD_WINDOW_S seconds,
+ so the benchmark's own work never counts against it. When load is the only
+ violation the gate polls for up to --settle-timeout for the other processes
+ to quiet down; under --force, where a load violation only warns, it does
not
wait. Memory and disk shortfalls do not settle, so they are judged at
once. A disk shortfall refuses even under --force: a copy-on-write commit
can rewrite the whole table, and running out of disk mid-commit hurts more
@@ -440,7 +573,7 @@ def gate(work_dir, base_bytes, limits, what):
while True:
r = headroom(work_dir, base_bytes, limits)
v = r["violations"]
- settling = list(v) == ["load1"] and not limits.force
+ settling = list(v) == ["load"] and not limits.force
if not settling or time.monotonic() - start >= limits.settle_timeout:
break
time.sleep(2)
@@ -814,6 +947,8 @@ def compare_tables(table, exp, key_cols):
def phase_main(argv):
+ # Starts sampling now, so the gate's first window fills while inputs load.
+ other_load(0.0)
p = argparse.ArgumentParser(prog="bench.py _phase")
p.add_argument("--case", required=True)
p.add_argument("--phase", required=True, choices=PHASES)
@@ -1094,6 +1229,7 @@ def summarize_case(case):
if readings:
s["headroom"] = {
"readings": len(readings),
+ "max_other_load": max(r["other_load"] for r in readings),
"max_load1": max(r["load1"] for r in readings),
"min_mem_available_bytes": min(r["mem_available_bytes"] for r in
readings),
"min_disk_free_bytes": min(r["disk_free_bytes"] for r in readings),
@@ -1183,7 +1319,8 @@ def run_rep(name, case, r, a, plan, bd, wd, root,
log_dir):
return rep
h = rep["headroom_before"]
print(
- f"{tag} headroom: load1 {h['load1']:.2f}, available memory "
+ f"{tag} headroom: other load {h['other_load']:.2f} (load1
{h['load1']:.2f}), "
+ f"available memory "
f"{h['mem_available_bytes'] / 2**30:.1f} GiB, "
f"free disk {h['disk_free_bytes'] / 2**30:.1f} GiB",
flush=True,
@@ -1207,7 +1344,7 @@ def run_rep(name, case, r, a, plan, bd, wd, root,
log_dir):
"--base-dir", str(bd),
"--workload-dir", str(wd),
"--out", str(out),
- "--max-load1", str(a.max_load1),
+ "--max-load", str(a.max_load),
"--min-mem-gib", str(a.min_mem_gib),
"--settle-timeout", str(a.settle_timeout),
"--base-bytes", str(a.base_bytes),
@@ -1284,10 +1421,11 @@ def run_rep(name, case, r, a, plan, bd, wd, root,
log_dir):
def add_headroom_args(p):
p.add_argument(
- "--max-load1",
+ "--max-load",
type=float,
default=2.0,
- help="refuse to time anything while the 1-minute load average is above
this",
+ help="refuse to time anything while other processes have kept more
than this "
+ f"many cores busy over the last {LOAD_WINDOW_S:.0f} seconds",
)
p.add_argument(
"--min-mem-gib",
@@ -1476,7 +1614,8 @@ def run(a, sc, params, cases, work, slug, out):
"reclaim": a.reclaim,
"extra_cases": [Path(x).name for x in a.extra_cases],
"headroom_limits": {
- "max_load1": a.max_load1,
+ "max_other_load": a.max_load,
+ "other_load_window_s": 1.0 if a.force else LOAD_WINDOW_S,
"min_mem_available_gib": a.min_mem_gib,
"min_disk_free": f"{DISK_FACTOR} x base dataset
({a.base_bytes} bytes)",
"settle_timeout_s": a.settle_timeout,
diff --git a/benchmark/upsert/summarize.py b/benchmark/upsert/summarize.py
index 5ed7cda7..3ec0fc10 100755
--- a/benchmark/upsert/summarize.py
+++ b/benchmark/upsert/summarize.py
@@ -124,11 +124,13 @@ def violated(case):
def case_row(name, case, scenario, scale, stat):
s = case.get("summary", {})
- st = s.get("stats", {})
+ ok = case.get("status") == "ok"
+ # A refused, failed or timed-out case keeps the stats of the repetitions it
+ # finished; they are not a result.
+ st = s.get("stats", {}) if ok else {}
by_phase = s.get("peak_rss_bytes_by_phase", {})
write_rss = [by_phase[p] for p in ("upsert", "delete") if p in by_phase]
- files = s.get("upsert_data_files_created_per_commit")
- ok = case.get("status") == "ok"
+ files = s.get("upsert_data_files_created_per_commit") if ok else None
inc = timing(st.get("incremental_mean_s"), stat)
if inc != "-" and "incremental_rows_mean" in s:
inc += f" ({s['incremental_rows_mean']:,.0f})"
@@ -145,8 +147,8 @@ def case_row(name, case, scenario, scale, stat):
timing(st.get("readback_s"), stat),
inc,
timing(st.get("time_travel_mean_s"), stat),
- _mib(s.get("upsert_bytes_per_commit")),
- _mib(s.get("delete_bytes_per_commit")),
+ _mib(s.get("upsert_bytes_per_commit")) if ok else "-",
+ _mib(s.get("delete_bytes_per_commit")) if ok else "-",
"-" if files is None else f"{files:g}",
_mib(by_phase.get("load")) if ok else "-",
_mib(max(write_rss)) if write_rss and ok else "-",
@@ -157,11 +159,15 @@ def case_row(name, case, scenario, scale, stat):
def case_notes(scenario, name, case):
s = case.get("summary", {})
+ ok = case.get("status") == "ok"
out = []
noisy = [
f"{label} {st['cv']:.0%}"
for key, label in TIMING_COLUMNS.items()
- if (st := s.get("stats", {}).get(key)) and st.get("cv") and st["cv"] >
CV_NOTE
+ if ok
+ and (st := s.get("stats", {}).get(key))
+ and st.get("cv")
+ and st["cv"] > CV_NOTE
]
if noisy:
out.append(
@@ -178,6 +184,27 @@ def case_notes(scenario, name, case):
return out
+def headroom_note(scenario, name, case):
+ """The machine readings behind a case's timings, or None if it has none.
+ Results written before the gate measured other-process load have only
load1."""
+ h = case.get("summary", {}).get("headroom")
+ if not h:
+ return None
+ mem = h.get("min_mem_available_bytes")
+ load = (
+ f"max other-process load {h['max_other_load']:.2f} cores, "
+ if "max_other_load" in h
+ else ""
+ )
+ return (
+ f"{scenario} {name} headroom over {h['readings']} readings: {load}"
+ f"max load1 {h['max_load1']:.2f}, min available memory "
+ f"{'-' if mem is None else f'{mem / 1024**3:.1f} GiB'}, min free disk "
+ f"{h['min_disk_free_bytes'] / 1024**3:.1f} GiB"
+ + (" (HEADROOM VIOLATED)" if violated(case) else "")
+ )
+
+
def render(rows, markdown):
if markdown:
out = [
@@ -257,16 +284,9 @@ def main():
for name, case in res.get("cases", {}).items():
rows.append(case_row(name, case, scenario, scale, a.stat))
notes += case_notes(scenario, name, case)
- h = case.get("summary", {}).get("headroom")
- if h:
- mem = h.get("min_mem_available_bytes")
- notes.append(
- f"{scenario} {name} headroom over {h['readings']}
readings: "
- f"max load1 {h['max_load1']:.2f}, min available memory "
- f"{'-' if mem is None else f'{mem / 1024**3:.1f} GiB'},
min free disk "
- f"{h['min_disk_free_bytes'] / 1024**3:.1f} GiB"
- + (" (HEADROOM VIOLATED)" if violated(case) else "")
- )
+ note = headroom_note(scenario, name, case)
+ if note:
+ notes.append(note)
if case.get("error"):
errors.append(f"{scenario} {name} ({case['status']}):
{case['error']}")
diff --git a/benchmark/upsert/tests/test_measure.py
b/benchmark/upsert/tests/test_measure.py
index 76559532..6214bc1b 100644
--- a/benchmark/upsert/tests/test_measure.py
+++ b/benchmark/upsert/tests/test_measure.py
@@ -20,7 +20,11 @@ run lock: the parts that decide what a number means or
whether it is taken."""
import argparse
import os
+import subprocess
+import sys
+import threading
import time
+import types
from pathlib import Path
import bench
@@ -93,7 +97,7 @@ def
test_reclaim_refuses_when_the_live_listing_is_empty(tmp_path):
def limits(**kw):
base = {
- "max_load1": 2.0,
+ "max_load": 2.0,
"min_mem_gib": 16.0,
"settle_timeout": 30.0,
"force": False,
@@ -107,6 +111,7 @@ def fake_headroom(monkeypatch, *violation_sets):
def headroom(work_dir, base_bytes, lim):
v = seq.pop(0) if len(seq) > 1 else seq[0]
return {
+ "other_load": 9.0,
"load1": 9.0,
"mem_available_bytes": 1,
"disk_free_bytes": 1,
@@ -118,7 +123,7 @@ def fake_headroom(monkeypatch, *violation_sets):
def test_gate_does_not_wait_on_memory_shortfall(monkeypatch):
- fake_headroom(monkeypatch, {"memory": "low", "load1": "busy"})
+ fake_headroom(monkeypatch, {"memory": "low", "load": "busy"})
start = time.monotonic()
with pytest.raises(bench.HeadroomRefused, match="low"):
bench.gate(".", 1, limits(), "x")
@@ -126,14 +131,14 @@ def
test_gate_does_not_wait_on_memory_shortfall(monkeypatch):
def test_gate_waits_for_load_to_settle(monkeypatch):
- fake_headroom(monkeypatch, {"load1": "busy"}, {"load1": "busy"}, {})
+ fake_headroom(monkeypatch, {"load": "busy"}, {"load": "busy"}, {})
r = bench.gate(".", 1, limits(), "x")
assert not r["violations"]
assert "headroom_violated" not in r
def test_force_downgrades_load_but_not_disk(monkeypatch):
- fake_headroom(monkeypatch, {"load1": "busy"})
+ fake_headroom(monkeypatch, {"load": "busy"})
r = bench.gate(".", 1, limits(force=True, settle_timeout=0), "x")
assert r["headroom_violated"]
fake_headroom(monkeypatch, {"disk": "full"})
@@ -147,10 +152,11 @@ def
test_forced_gate_does_not_wait_for_load_to_settle(monkeypatch):
def headroom(work_dir, base_bytes, lim):
calls.append(1)
return {
+ "other_load": 9.0,
"load1": 9.0,
"mem_available_bytes": 1,
"disk_free_bytes": 1,
- "violations": {"load1": "busy"},
+ "violations": {"load": "busy"},
}
monkeypatch.setattr(bench, "headroom", headroom)
@@ -190,3 +196,114 @@ def test_run_lock_takes_over_a_stale_lock(tmp_path):
lock.write_text(str(pid))
with bench.RunLock(tmp_path, "s"):
assert lock.read_text() == str(os.getpid())
+
+
+def test_busy_cores_excludes_the_benchmarks_own_cpu():
+ # 10 s apart: the system was busy for 50 core-seconds, 30 of them this
process's.
+ assert bench.busy_cores((0.0, 100.0, 5.0), (10.0, 150.0, 35.0)) == 2.0
+ # Own CPU time is sampled separately from system time, so it can come out
+ # slightly ahead; the reading never goes negative.
+ assert bench.busy_cores((0.0, 100.0, 5.0), (10.0, 110.0, 16.0)) == 0.0
+
+
+def test_headroom_judges_other_process_load_not_load1(monkeypatch, tmp_path):
+ windows = []
+
+ def fake_other_load(window):
+ windows.append(window)
+ return load, window
+
+ monkeypatch.setattr(bench.os, "getloadavg", lambda: (50.0, 50.0, 50.0))
+ monkeypatch.setattr(bench, "other_load", fake_other_load)
+ load = 1.0
+ r = bench.headroom(tmp_path, 1, limits(min_mem_gib=0))
+ assert "load" not in r["violations"] and r["load1"] == 50.0
+ load = 3.0
+ assert "load" in bench.headroom(tmp_path, 1,
limits(min_mem_gib=0))["violations"]
+ # A forced run only warns on load, so it does not wait for a full window.
+ bench.headroom(tmp_path, 1, limits(min_mem_gib=0, force=True))
+ assert windows == [bench.LOAD_WINDOW_S, bench.LOAD_WINDOW_S, 1.0]
+
+
+def test_other_load_reads_the_newest_sample_a_full_window_old(monkeypatch):
+ load = bench.OtherLoad.__new__(bench.OtherLoad)
+ load.lock = threading.Lock()
+ # One sample a second; the system ran 3 cores busy, 1 of them this process.
+ load.samples = [(float(t), 3.0 * t, 1.0 * t) for t in range(0, 11)]
+ monkeypatch.setattr(
+ bench.OtherLoad, "sample", staticmethod(lambda: (10.5, 31.5, 10.5))
+ )
+ assert load.read(10.0) == (2.0, 10.5)
+ assert load.read(1.0) == (2.0, 1.5)
+
+
+def test_own_cpu_counts_a_reaped_child():
+ before = bench.own_cpu_s()
+ subprocess.run(
+ [
+ sys.executable,
+ "-c",
+ "import time\nt=time.process_time()\nwhile
time.process_time()-t<0.3: pass",
+ ],
+ check=True,
+ )
+ assert bench.own_cpu_s() - before >= 0.25
+
+
+class _Child:
+ def __init__(self, *outcomes):
+ self.outcomes = list(outcomes)
+
+ def cpu_times(self):
+ o = self.outcomes.pop(0)
+ if isinstance(o, Exception):
+ raise o
+ return types.SimpleNamespace(user=o, system=0.0)
+
+
+def fake_tree(monkeypatch, child, reaped):
+ monkeypatch.setattr(
+ bench.psutil,
+ "Process",
+ lambda: types.SimpleNamespace(children=lambda recursive: [child]),
+ )
+ monkeypatch.setattr(bench, "reaped_cpu_s", lambda: reaped.pop(0))
+ monkeypatch.setattr(bench.time, "sleep", lambda s: None)
+
+
+def test_own_cpu_walks_again_when_a_child_is_reaped_mid_walk(monkeypatch):
+ # The child's 5 s show up in neither sum on the first walk.
+ child = _Child(5.0, 7.0)
+ fake_tree(monkeypatch, child, [10.0, 15.0, 15.0, 15.0])
+ me = bench.resource.getrusage(bench.resource.RUSAGE_SELF)
+ got = bench.own_cpu_s() - (me.ru_utime + me.ru_stime)
+ assert got == pytest.approx(22.0, abs=0.5)
+
+
+def test_own_cpu_walks_again_past_an_unreaped_child(monkeypatch):
+ child = _Child(bench.psutil.ZombieProcess(1), 7.0)
+ fake_tree(monkeypatch, child, [10.0, 15.0, 15.0])
+ me = bench.resource.getrusage(bench.resource.RUSAGE_SELF)
+ assert bench.own_cpu_s() - (me.ru_utime + me.ru_stime) == pytest.approx(
+ 22.0, abs=0.5
+ )
+
+
+def test_own_cpu_gives_up_rather_than_undercount(monkeypatch):
+ child = _Child(*[bench.psutil.ZombieProcess(1)] * 3)
+ fake_tree(monkeypatch, child, [10.0] * 6)
+ assert bench.own_cpu_s(attempts=3) is None
+
+
+def test_sampler_skips_a_failed_sample(monkeypatch):
+ load = bench.OtherLoad.__new__(bench.OtherLoad)
+ load.lock = threading.Lock()
+ load.samples = [(1.0, 1.0, 1.0)]
+
+ def boom():
+ raise OSError("transient")
+
+ for sample in (boom, lambda: None):
+ monkeypatch.setattr(bench.OtherLoad, "sample", staticmethod(sample))
+ load._record()
+ assert load.samples == [(1.0, 1.0, 1.0)]
diff --git a/benchmark/upsert/tests/test_summarize.py
b/benchmark/upsert/tests/test_summarize.py
new file mode 100644
index 00000000..a09d9586
--- /dev/null
+++ b/benchmark/upsert/tests/test_summarize.py
@@ -0,0 +1,95 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""How summarize.py renders a case row."""
+
+import pytest
+import summarize
+
+
+def case(status, repetitions):
+ stat = {"n": repetitions, "median": 0.5, "min": 0.4, "max": 0.6, "cv": 0.1}
+ return {
+ "status": status,
+ "repeat": 3,
+ "repetitions": [{"phases": {"readback": {}}}] * repetitions,
+ "summary": {
+ "repetitions": repetitions,
+ "stats": {
+ k: stat
+ for k in ("load_s", "upsert_mean_s", "delete_mean_s",
"readback_s")
+ },
+ "upsert_bytes_per_commit": 8 * summarize.MIB,
+ "delete_bytes_per_commit": 4 * summarize.MIB,
+ "upsert_data_files_created_per_commit": 2.0,
+ "correct": status == "ok",
+ },
+ }
+
+
+def test_a_case_that_finished_shows_its_timings_and_bytes():
+ row = summarize.case_row("hudi-mor", case("ok", 3), "uuid", "1,000 rows",
"median")
+ assert row[3:14] == [
+ "ok",
+ "3/3",
+ "0.50 s (0.40 to 0.60)",
+ "0.50 s (0.40 to 0.60)",
+ "0.50 s (0.40 to 0.60)",
+ "0.50 s (0.40 to 0.60)",
+ "-",
+ "-",
+ "8",
+ "4",
+ "2",
+ ]
+
+
[email protected]("status", ["refused", "failed", "timeout"])
+def test_a_case_that_did_not_finish_shows_no_timings_or_bytes(status):
+ row = summarize.case_row(
+ "hudi-mor", case(status, 2), "uuid", "1,000 rows", "median"
+ )
+ assert row[3:4] == [status]
+ assert row[5:14] == ["-"] * 9
+
+
[email protected]("status, noted", [("ok", True), ("failed", False)])
+def test_only_a_finished_case_gets_a_variation_note(status, noted):
+ c = case(status, 2)
+ c["summary"]["stats"]["upsert_mean_s"]["cv"] = 0.5
+ notes = summarize.case_notes("uuid", "hudi-mor", c)
+ assert any("upsert" in n for n in notes) == noted
+
+
[email protected](
+ "extra, shown",
+ [({"max_other_load": 1.25}, "max other-process load 1.25 cores, "), ({},
"")],
+)
+def test_headroom_note_shows_other_load_when_recorded(extra, shown):
+ c = case("ok", 3)
+ c["summary"]["headroom"] = {
+ "readings": 4,
+ "max_load1": 3.5,
+ "min_mem_available_bytes": 20 * 1024**3,
+ "min_disk_free_bytes": 100 * 1024**3,
+ **extra,
+ }
+ assert summarize.headroom_note("uuid", "hudi-mor", c) == (
+ f"uuid hudi-mor headroom over 4 readings: {shown}max load1 3.50, "
+ "min available memory 20.0 GiB, min free disk 100.0 GiB"
+ )
+ assert summarize.headroom_note("uuid", "hudi-mor", case("ok", 3)) is None
