From 5952694d446dcaea19b9077f50fdfe7e0a76599b Mon Sep 17 00:00:00 2001
From: Pau Aliagas <linuxnow@gmail.com>
Date: Tue, 27 Oct 2026 06:37:21 +0100
Subject: [PATCH 1/7] Bench schema: xruns, latency_frames, stress, repeats and
 spread; modes plan and pool (#188)

Fields only, so the document stays omx-bench/1. A spread has to hold its
median and name a field the row has, and a stressed row has to report its
xruns. The same scenario with and without stress is two rows.
---
 bench/lib/validate.py                 |  22 ++++-
 bench/schema/omx-bench.v1.schema.json |  16 +++-
 bench/tests/sabotage.sh               |  22 +++++
 bench/tests/test_bench.py             | 127 ++++++++++++++++++++++++++
 4 files changed, 182 insertions(+), 5 deletions(-)
 create mode 100644 bench/tests/test_bench.py

diff --git a/bench/lib/validate.py b/bench/lib/validate.py
index 73689ec..842f1b4 100644
--- a/bench/lib/validate.py
+++ b/bench/lib/validate.py
@@ -96,10 +96,10 @@ def cross_rules(doc, errors):
     for n, s in enumerate(doc.get("scenarios", [])):
         if not isinstance(s, dict):
             continue
-        key = (s.get("id"), s.get("host"), s.get("mode"), s.get("rate"), s.get("frames"))
+        key = (s.get("id"), s.get("host"), s.get("mode"), s.get("rate"), s.get("frames"), s.get("stress", "none"))
         if key in seen:
-            errors.append((["scenarios", n], "rule:unique-row", "repeats row #%d (same id, host, mode, rate, frames)"
-                           % seen[key]))
+            errors.append((["scenarios", n], "rule:unique-row",
+                           "repeats row #%d (same id, host, mode, rate, frames, stress)" % seen[key]))
         seen[key] = n
         avg, mx = s.get("dsp_load_avg"), s.get("dsp_load_max")
         if isinstance(avg, (int, float)) and isinstance(mx, (int, float)) and mx < avg:
@@ -109,6 +109,22 @@ def cross_rules(doc, errors):
             errors.append((["scenarios", n], "rule:recover-only-ok", "has recover_ms on a row that is not ok"))
         if s.get("status") == "ok" and s.get("plugins") and s.get("add_ms") is None:
             errors.append((["scenarios", n], "rule:add-measured", "is ok with plugins and no add_ms"))
+        spread = s.get("spread")
+        if isinstance(spread, dict):
+            if s.get("repeats", 1) < 2:
+                errors.append((["scenarios", n], "rule:spread-needs-repeats", "has a spread and fewer than two repeats"))
+            for field, span in spread.items():
+                if field not in s:
+                    errors.append((["scenarios", n], "rule:spread-names-a-field", "spread of %s, which the row lacks"
+                                   % field))
+                elif len(span) != 2 or span[0] > span[1]:
+                    errors.append((["scenarios", n], "rule:spread-min-max", "spread of %s is %s, not [min, max]"
+                                   % (field, json.dumps(span))))
+                elif isinstance(s[field], (int, float)) and not span[0] <= s[field] <= span[1]:
+                    errors.append((["scenarios", n], "rule:spread-holds-median", "%s %s is outside its spread %s"
+                                   % (field, s[field], json.dumps(span))))
+        if s.get("stress", "none") != "none" and s.get("status") == "ok" and "xruns" not in s:
+            errors.append((["scenarios", n], "rule:stress-reports-xruns", "ran under stress and reports no xruns"))
         if "rss_mb_total" in s and "processes" not in s:
             errors.append((["scenarios", n], "rule:totals-name-processes", "has totals and no processes count"))
     m = re.match(r"^([a-z0-9]+)-", str(doc.get("id", "")))
diff --git a/bench/schema/omx-bench.v1.schema.json b/bench/schema/omx-bench.v1.schema.json
index 4a980b5..cefd615 100644
--- a/bench/schema/omx-bench.v1.schema.json
+++ b/bench/schema/omx-bench.v1.schema.json
@@ -58,13 +58,25 @@
           "recover_ms": { "type": ["number", "null"], "minimum": 0 },
           "status": { "enum": ["ok", "unsupported", "failed"] },
           "note": { "type": "string" },
-          "mode": { "enum": ["individual", "chain", "shared-process"] },
+          "mode": { "enum": ["individual", "chain", "shared-process", "plan", "pool"] },
           "processes": { "type": "integer", "minimum": 0 },
           "rss_mb_total": { "type": "number", "minimum": 0 },
           "pss_mb_total": { "type": "number", "minimum": 0 },
           "cpu_pct": { "type": "number", "minimum": 0 },
           "ctx_switches": { "type": "integer", "minimum": 0 },
-          "ctx_involuntary": { "type": "integer", "minimum": 0 }
+          "ctx_involuntary": { "type": "integer", "minimum": 0 },
+          "xruns": { "type": "integer", "minimum": 0 },
+          "latency_frames": { "type": ["integer", "null"], "minimum": 0 },
+          "stress": { "type": "string", "pattern": "^(none|busy[0-9]+\\+mem[0-9]+)$" },
+          "repeats": { "type": "integer", "minimum": 1 },
+          "spread": {
+            "type": "object",
+            "additionalProperties": {
+              "type": "array",
+              "minItems": 2,
+              "items": { "type": "number", "minimum": 0 }
+            }
+          }
         },
         "allOf": [
           {
diff --git a/bench/tests/sabotage.sh b/bench/tests/sabotage.sh
index 872ed7d..4cc1ff2 100755
--- a/bench/tests/sabotage.sh
+++ b/bench/tests/sabotage.sh
@@ -64,6 +64,28 @@ break_doc extra "${first_ok}[\"dsp\"] = 1"
 red "validator: a field the schema does not declare" "extra.json: scenario .*: scenarios/[0-9]+:additionalProperties" \
     "$here/validate.sh" "$work/extra.json"
 
+# --- the rules the unit tests guard: a copy of bench/ with one line changed must turn the named test red
+break_unit() {  # break_unit <label> <file under bench/> <old> <new> <test name>
+    local copy="$work/unit-$1"
+    cp -r "$here" "$copy"
+    python3 -I - "$copy/$2" "$3" "$4" <<'PY'
+import sys
+p, old, new = sys.argv[1:]
+s = open(p).read()
+assert s.count(old) == 1, "the sabotage did not land: " + old
+open(p, "w").write(s.replace(old, new))
+PY
+    red "unit: $1" "FAIL: ${5##*.}" python3 -I "$copy/tests/test_bench.py" "$5"
+}
+break_unit spread-median lib/validate.py 'elif isinstance(s[field], (int, float)) and not span[0] <= s[field] <= span[1]:' \
+    'elif False:' Schema.test_a_spread_that_does_not_hold_the_median_is_red
+break_unit stress-key lib/validate.py 's.get("rate"), s.get("frames"), s.get("stress", "none"))' \
+    's.get("rate"), s.get("frames"))' Schema.test_the_same_row_with_and_without_stress_are_two_rows
+break_unit stress-xruns lib/validate.py 'and s.get("status") == "ok" and "xruns" not in s:' 'and False:' \
+    Schema.test_a_stressed_row_without_xruns_is_red
+break_unit spread-repeats lib/validate.py 'if s.get("repeats", 1) < 2:' 'if False:' Schema.test_a_spread_of_one_run_is_red
+green "unit tests as they are" python3 -I "$here/tests/test_bench.py"
+
 # --- the teardown
 if [ "${1:-}" = "--teardown" ]; then
     sabotage() {  # sabotage <name> <old> <new>: a copy of bench/ with one line of the runner changed
diff --git a/bench/tests/test_bench.py b/bench/tests/test_bench.py
new file mode 100644
index 0000000..bd2e453
--- /dev/null
+++ b/bench/tests/test_bench.py
@@ -0,0 +1,127 @@
+"""Unit tests of the harness: run with `python3 -I bench/tests/test_bench.py`. They need no JACK server and no host:
+the schema and validator rules, and (below) the pieces of the runner that decide a number."""
+
+import copy
+import importlib.util
+import json
+import os
+import subprocess
+import sys
+import tempfile
+import unittest
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+BENCH = os.path.dirname(HERE)
+
+
+def load(name, path):
+    spec = importlib.util.spec_from_file_location(name, path)
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    return mod
+
+
+def example():
+    path = sorted(os.listdir(os.path.join(BENCH, "examples")))[0]
+    with open(os.path.join(BENCH, "examples", path)) as f:
+        return json.load(f)
+
+
+def validate(doc):
+    """(exit code, output) of validate.sh on doc"""
+    with tempfile.TemporaryDirectory() as d:
+        path = os.path.join(d, "doc.json")
+        with open(path, "w") as f:
+            json.dump(doc, f)
+        r = subprocess.run([os.path.join(BENCH, "validate.sh"), path], stdout=subprocess.PIPE,
+                           stderr=subprocess.STDOUT, text=True)
+        return r.returncode, r.stdout
+
+
+def first_ok(doc):
+    return [s for s in doc["scenarios"] if s["status"] == "ok"][0]
+
+
+class Schema(unittest.TestCase):
+    def test_the_shipped_example_is_valid(self):
+        code, out = validate(example())
+        self.assertEqual(code, 0, out)
+
+    def test_the_new_fields_are_valid_together(self):
+        doc = example()
+        s = first_ok(doc)
+        s.update({"mode": "plan", "xruns": 2, "latency_frames": 256, "stress": "busy24+mem256", "repeats": 3,
+                  "spread": {"xruns": [0, 5], "cpu_pct": [0, s["cpu_pct"] + 1]}})
+        code, out = validate(doc)
+        self.assertEqual(code, 0, out)
+
+    def test_a_negative_xrun_count_is_red(self):
+        doc = example()
+        first_ok(doc)["xruns"] = -1
+        code, out = validate(doc)
+        self.assertEqual(code, 1)
+        self.assertRegex(out, r"scenarios/\d+/xruns:minimum")
+
+    def test_a_stress_no_one_declared_is_red(self):
+        doc = example()
+        first_ok(doc)["stress"] = "heavy"
+        code, out = validate(doc)
+        self.assertEqual(code, 1)
+        self.assertRegex(out, r"scenarios/\d+/stress:pattern")
+
+    def test_a_mode_no_one_declared_is_red(self):
+        doc = example()
+        first_ok(doc)["mode"] = "grouped"
+        code, out = validate(doc)
+        self.assertEqual(code, 1)
+        self.assertRegex(out, r"scenarios/\d+/mode:enum")
+
+    def test_a_spread_that_does_not_hold_the_median_is_red(self):
+        doc = example()
+        s = first_ok(doc)
+        s.update({"repeats": 3, "xruns": 4, "spread": {"xruns": [0, 2]}})
+        code, out = validate(doc)
+        self.assertEqual(code, 1)
+        self.assertIn("rule:spread-holds-median", out)
+
+    def test_a_spread_with_min_above_max_is_red(self):
+        doc = example()
+        s = first_ok(doc)
+        s.update({"repeats": 3, "xruns": 1, "spread": {"xruns": [3, 0]}})
+        code, out = validate(doc)
+        self.assertEqual(code, 1)
+        self.assertIn("rule:spread-min-max", out)
+
+    def test_a_spread_of_one_run_is_red(self):
+        doc = example()
+        s = first_ok(doc)
+        s.update({"repeats": 1, "xruns": 1, "spread": {"xruns": [0, 2]}})
+        code, out = validate(doc)
+        self.assertEqual(code, 1)
+        self.assertIn("rule:spread-needs-repeats", out)
+
+    def test_a_stressed_row_without_xruns_is_red(self):
+        doc = example()
+        first_ok(doc)["stress"] = "busy4+mem256"
+        code, out = validate(doc)
+        self.assertEqual(code, 1)
+        self.assertIn("rule:stress-reports-xruns", out)
+
+    def test_the_same_row_with_and_without_stress_are_two_rows(self):
+        doc = example()
+        a = first_ok(doc)
+        b = copy.deepcopy(a)
+        a.update({"stress": "none", "xruns": 0})
+        b.update({"stress": "busy4+mem256", "xruns": 3})
+        doc["scenarios"].append(b)
+        code, out = validate(doc)
+        self.assertEqual(code, 0, out)
+        c = copy.deepcopy(b)
+        doc["scenarios"].append(c)
+        code, out = validate(doc)
+        self.assertEqual(code, 1)
+        self.assertIn("rule:unique-row", out)
+
+
+if __name__ == "__main__":
+    unittest.main()

From a35f99cbbd0b63537ee1cef9fe0201f3f325971c Mon Sep 17 00:00:00 2001
From: Pau Aliagas <linuxnow@gmail.com>
Date: Tue, 27 Oct 2026 06:38:24 +0100
Subject: [PATCH 2/7] Bench: count xruns and read the latency JACK reports
 (#188)

A client on the run's private server counts the xrun callbacks over each
measurement window and reads the capture latency at the playback port. If the
server's own log names xruns the probe never heard, the row fails instead of
reading zero.
---
 bench/lib/bench.py        | 113 ++++++++++++++++++++++++++++++++++++++
 bench/tests/sabotage.sh   |   5 ++
 bench/tests/test_bench.py | 106 +++++++++++++++++++++++++++++++++++
 3 files changed, 224 insertions(+)

diff --git a/bench/lib/bench.py b/bench/lib/bench.py
index 16ea017..06efc76 100644
--- a/bench/lib/bench.py
+++ b/bench/lib/bench.py
@@ -256,6 +256,105 @@ def connected(env, a, b):
     return b in run(["jack_lsp", "-c", a], env).split()
 
 
+class JackProbe:
+    """a JACK client of the private server that counts its xruns and reads the latency JACK reports. It registers no
+    port and no process callback, so it adds nothing to the cycle it watches. `lib` is libjack, or a stand-in in the
+    unit tests."""
+
+    NO_START = 0x01  # JackNoStartServer
+    SERVER_NAME = 0x04  # JackServerName
+    CAPTURE = 0  # JackCaptureLatency
+
+    class Range(ctypes.Structure):
+        _fields_ = [("min", ctypes.c_uint32), ("max", ctypes.c_uint32)]
+
+    def __init__(self, server, lib=None, name="omx-bench-probe"):
+        if lib is None:
+            for so in ("libjack.so.0", "libjack.so"):
+                try:
+                    lib = ctypes.CDLL(so, use_errno=True)
+                    break
+                except OSError:
+                    pass
+            else:
+                raise SystemExit("bench: libjack is needed to count xruns (package jackd2)")
+        self.lib = lib
+        self.total = 0
+        self._bind()
+        status = ctypes.c_int(0)
+        self.client = lib.jack_client_open(name.encode(), self.NO_START | self.SERVER_NAME, ctypes.byref(status),
+                                           server.encode())
+        if not self.client:
+            raise SystemExit("bench: cannot open a JACK client on server %s (status %d)" % (server, status.value))
+        self._callback = ctypes.CFUNCTYPE(ctypes.c_int, ctypes.c_void_p)(self._on_xrun)
+        if lib.jack_set_xrun_callback(self.client, self._callback, None) != 0:
+            raise SystemExit("bench: cannot set the xrun callback on server %s" % server)
+        if lib.jack_activate(self.client) != 0:
+            raise SystemExit("bench: cannot activate the xrun probe on server %s" % server)
+
+    def _bind(self):
+        lib = self.lib
+        vp, ci = ctypes.c_void_p, ctypes.c_int
+        for name, res, args in (
+                ("jack_client_open", vp, [ctypes.c_char_p, ci, ctypes.POINTER(ci), ctypes.c_char_p]),
+                ("jack_set_xrun_callback", ci, [vp, vp, vp]),
+                ("jack_activate", ci, [vp]),
+                ("jack_deactivate", ci, [vp]),
+                ("jack_client_close", ci, [vp]),
+                ("jack_port_by_name", vp, [vp, ctypes.c_char_p]),
+                ("jack_recompute_total_latencies", ci, [vp]),
+                ("jack_port_get_latency_range", None, [vp, ci, ctypes.POINTER(self.Range)])):
+            fn = getattr(lib, name)
+            try:
+                fn.restype, fn.argtypes = res, args
+            except AttributeError:
+                pass  # a stand-in function, not a ctypes one
+
+    def _on_xrun(self, _arg):
+        self.total += 1
+        return 0
+
+    def count(self):
+        """xruns since the probe opened"""
+        return self.total
+
+    def latency_frames(self, port):
+        """the capture latency JACK reports at an input port, in frames, the longest of its range: the delay from
+        the sources to the port as the clients on the path declared it; None when the port is not there"""
+        p = self.lib.jack_port_by_name(self.client, port.encode())
+        if not p:
+            return None
+        self.lib.jack_recompute_total_latencies(self.client)
+        rng = self.Range()
+        self.lib.jack_port_get_latency_range(p, self.CAPTURE, ctypes.byref(rng))
+        return int(rng.max)
+
+    def close(self):
+        if self.client:
+            self.lib.jack_deactivate(self.client)
+            self.lib.jack_client_close(self.client)
+            self.client = None
+
+
+def jack_log_xruns(path, offset):
+    """lines of the JACK server's own log from offset on that name an xrun, and the new end of the file"""
+    try:
+        with open(path, "rb") as f:
+            f.seek(offset)
+            raw = f.read()
+    except OSError:
+        return 0, offset
+    return sum(1 for line in raw.decode(errors="replace").splitlines() if "xrun" in line.lower()), offset + len(raw)
+
+
+def window_xruns(counted, logged):
+    """the xruns of one window: what the probe counted, unless the server's own log names xruns the probe never
+    heard, which would mean the probe is deaf and every row would read 0"""
+    if counted == 0 and logged > 0:
+        raise Failed("jackd logged %d xrun line(s) in the window and the probe counted none" % logged)
+    return counted
+
+
 # ---------------------------------------------------------------- the protocol
 
 
@@ -632,6 +731,9 @@ def __init__(self, args, table):
             self.env["LV2_PATH"] = ":".join([args.pianoteq_dir] + [d for d in lv2_dirs() if os.path.isdir(d)])
         self.saved_governor = {}
         self.jackd = None
+        self.probe = None
+        self.jack_log = None
+        self.jack_log_end = 0
         self.port = 5700
 
     # -- governor
@@ -668,8 +770,14 @@ def jack_start(self, rate, frames):
                 raise SystemExit("bench: the private JACK server did not start at %d Hz / %d frames (see %s)"
                                  % (rate, frames, self.logdir))
             time.sleep(0.1)
+        self.probe = JackProbe(self.server)
+        self.jack_log = os.path.join(self.logdir, "jackd-%d-%d.log" % (rate, frames))
+        self.jack_log_end = jack_log_xruns(self.jack_log, 0)[1]
 
     def jack_stop(self):
+        if self.probe:
+            self.probe.close()
+            self.probe = None
         if self.jackd:
             self.procs.stop(self.jackd)
             self.jackd = None
@@ -787,7 +895,11 @@ def row(self, s, hkey, mode, rate, frames):
             pids = host.pids()
             before = samples(pids)
             t_start = time.monotonic()
+            x_before = self.probe.count()
+            self.jack_log_end = jack_log_xruns(self.jack_log, self.jack_log_end)[1]
             loads = self.dsp_window(self.args.window)
+            logged, self.jack_log_end = jack_log_xruns(self.jack_log, self.jack_log_end)
+            r["xruns"] = window_xruns(self.probe.count() - x_before, logged)
             after = samples(pids)
             span = time.monotonic() - t_start
             if not host.pids() or any(not alive(p) for p in pids):
@@ -811,6 +923,7 @@ def row(self, s, hkey, mode, rate, frames):
             invol = sum(after[p]["invol"] - before[p]["invol"] for p in common)
             r["ctx_switches"] = vol + invol
             r["ctx_involuntary"] = invol
+            r["latency_frames"] = self.probe.latency_frames("system:playback_1")
             r["status"] = "ok"
 
             if s.get("kill"):
diff --git a/bench/tests/sabotage.sh b/bench/tests/sabotage.sh
index 4cc1ff2..6f1e0d2 100755
--- a/bench/tests/sabotage.sh
+++ b/bench/tests/sabotage.sh
@@ -84,6 +84,11 @@ break_unit stress-key lib/validate.py 's.get("rate"), s.get("frames"), s.get("st
 break_unit stress-xruns lib/validate.py 'and s.get("status") == "ok" and "xruns" not in s:' 'and False:' \
     Schema.test_a_stressed_row_without_xruns_is_red
 break_unit spread-repeats lib/validate.py 'if s.get("repeats", 1) < 2:' 'if False:' Schema.test_a_spread_of_one_run_is_red
+break_unit probe-count lib/bench.py 'self.total += 1' 'self.total += 0' Xruns.test_the_probe_counts_what_jack_reports
+break_unit probe-server lib/bench.py 'self.NO_START | self.SERVER_NAME' 'self.SERVER_NAME' Xruns.test_the_probe_joins_the_private_server_only
+break_unit probe-deaf lib/bench.py 'if counted == 0 and logged > 0:' 'if False:' Xruns.test_a_deaf_probe_is_failed_not_zero
+break_unit probe-latency lib/bench.py 'return int(rng.max)' 'return int(rng.min)' Xruns.test_the_latency_is_the_longest_of_the_range
+break_unit log-offset lib/bench.py 'f.seek(offset)' 'f.seek(0)' Xruns.test_the_servers_log_counts_new_xrun_lines_only
 green "unit tests as they are" python3 -I "$here/tests/test_bench.py"
 
 # --- the teardown
diff --git a/bench/tests/test_bench.py b/bench/tests/test_bench.py
index bd2e453..ab24e31 100644
--- a/bench/tests/test_bench.py
+++ b/bench/tests/test_bench.py
@@ -123,5 +123,111 @@ def test_the_same_row_with_and_without_stress_are_two_rows(self):
         self.assertIn("rule:unique-row", out)
 
 
+bench = load("bench_runner", os.path.join(BENCH, "lib", "bench.py"))
+
+
+class FakeJack:
+    """libjack as far as the probe uses it: the calls are Python functions, the xrun callback is kept for the test
+    to fire as JACK's notification thread would"""
+
+    def __init__(self, ports=None):
+        self.callback = None
+        self.ports = ports or {}
+        self.calls = []
+        self.opened = {}
+
+    def jack_client_open(self, name, options, status, server):
+        self.opened = {"name": name, "options": options, "server": server}
+        return 1
+
+    def jack_set_xrun_callback(self, client, cb, arg):
+        self.callback = cb
+        return 0
+
+    def jack_activate(self, client):
+        self.calls.append("activate")
+        return 0
+
+    def jack_deactivate(self, client):
+        self.calls.append("deactivate")
+        return 0
+
+    def jack_client_close(self, client):
+        self.calls.append("close")
+        return 0
+
+    def jack_port_by_name(self, client, name):
+        return 7 if name.decode() in self.ports else 0
+
+    def jack_recompute_total_latencies(self, client):
+        self.calls.append("recompute")
+        return 0
+
+    def jack_port_get_latency_range(self, port, mode, rng):
+        assert mode == 0, "the capture latency is what reaches a playback port"
+        rng._obj.min, rng._obj.max = 128, [v for v in self.ports.values()][0]
+
+    def xrun(self, n=1):
+        for _ in range(n):
+            assert self.callback(None) == 0
+
+
+class Xruns(unittest.TestCase):
+    def test_the_probe_counts_what_jack_reports(self):
+        lib = FakeJack()
+        probe = bench.JackProbe("bench1", lib=lib)
+        self.assertEqual(probe.count(), 0)
+        lib.xrun(3)
+        self.assertEqual(probe.count(), 3)
+        lib.xrun(2)
+        self.assertEqual(probe.count(), 5)
+
+    def test_the_probe_joins_the_private_server_only(self):
+        lib = FakeJack()
+        bench.JackProbe("bench1", lib=lib)
+        self.assertEqual(lib.opened["server"], b"bench1")
+        self.assertEqual(lib.opened["options"], 0x04 | 0x01, "a named server, and never one started by the probe")
+
+    def test_a_window_counts_the_xruns_inside_it(self):
+        lib = FakeJack()
+        probe = bench.JackProbe("bench1", lib=lib)
+        lib.xrun(4)
+        before = probe.count()
+        lib.xrun(2)
+        self.assertEqual(bench.window_xruns(probe.count() - before, 0), 2)
+
+    def test_the_latency_is_the_longest_of_the_range(self):
+        lib = FakeJack(ports={"system:playback_1": 512})
+        probe = bench.JackProbe("bench1", lib=lib)
+        self.assertEqual(probe.latency_frames("system:playback_1"), 512)
+        self.assertIn("recompute", lib.calls)
+        self.assertIsNone(probe.latency_frames("nobody:out"))
+
+    def test_close_leaves_the_server(self):
+        lib = FakeJack()
+        bench.JackProbe("bench1", lib=lib).close()
+        self.assertEqual(lib.calls[-2:], ["deactivate", "close"])
+
+    def test_a_deaf_probe_is_failed_not_zero(self):
+        with self.assertRaisesRegex(bench.Failed, "logged 2 xrun line"):
+            bench.window_xruns(0, 2)
+        self.assertEqual(bench.window_xruns(0, 0), 0)
+        self.assertEqual(bench.window_xruns(3, 1), 3)
+
+    def test_the_servers_log_counts_new_xrun_lines_only(self):
+        with tempfile.TemporaryDirectory() as d:
+            path = os.path.join(d, "jackd.log")
+            with open(path, "w") as f:
+                f.write("JackEngine::XRun: client = a was not finished\n")
+            n, end = bench.jack_log_xruns(path, 0)
+            self.assertEqual(n, 1)
+            with open(path, "a") as f:
+                f.write("started\nJackEngine::XRun: client = b\nan XRUN again\n")
+            n2, end2 = bench.jack_log_xruns(path, end)
+            self.assertEqual(n2, 2)
+            self.assertEqual(bench.jack_log_xruns(path, end2), (0, end2))
+            self.assertEqual(bench.jack_log_xruns(os.path.join(d, "none"), 5), (0, 5))
+
+
 if __name__ == "__main__":
     unittest.main()

From 5de85854d2c76359cd41769c64b34ee4fc8392b2 Mon Sep 17 00:00:00 2001
From: Pau Aliagas <linuxnow@gmail.com>
Date: Tue, 27 Oct 2026 06:39:32 +0100
Subject: [PATCH 3/7] Bench: a fixed stress load, with and without it as two
 rows (#188)

One busy loop pinned to each core the host may use, plus a process that keeps
touching 256 MiB. The row carries the profile name, so the same name on the
same machine is the same load. Scenarios that declare stress run both ways
unless --stress says otherwise.
---
 bench/lib/bench.py        | 61 +++++++++++++++++++++++++++-------
 bench/scenarios.json      |  6 ++--
 bench/tests/sabotage.sh   |  4 +++
 bench/tests/test_bench.py | 70 +++++++++++++++++++++++++++++++++++++++
 4 files changed, 126 insertions(+), 15 deletions(-)

diff --git a/bench/lib/bench.py b/bench/lib/bench.py
index 06efc76..2e3c274 100644
--- a/bench/lib/bench.py
+++ b/bench/lib/bench.py
@@ -222,6 +222,32 @@ def samples(pids):
     return out
 
 
+# ---------------------------------------------------------------- the stress load
+
+STRESS_MEM_MB = 256
+# one busy loop per core, each pinned to its core; one process that keeps touching a buffer of a fixed size
+BUSY_CODE = "import os, sys\nos.sched_setaffinity(0, {int(sys.argv[1])})\nwhile True:\n    pass\n"
+MEM_CODE = ("import sys, time\nbuf = bytearray(int(sys.argv[1]) << 20)\nwhile True:\n"
+            "    for i in range(0, len(buf), 4096):\n        buf[i] = (buf[i] + 1) & 255\n    time.sleep(0.1)\n")
+
+
+def stress_cores():
+    """the cores the host may run on: the stress sits on every one of them"""
+    return sorted(os.sched_getaffinity(0))
+
+
+def stress_profile(cores):
+    """the name a row carries: the same name on the same machine means the same load"""
+    return "busy%d+mem%d" % (len(cores), STRESS_MEM_MB)
+
+
+def stress_argvs(cores, mem_mb=STRESS_MEM_MB):
+    """[(process name, argv)] of the load: a busy loop pinned to each core, and the memory touch"""
+    out = [("stress-busy-%d" % c, [sys.executable, "-I", "-c", BUSY_CODE, str(c)]) for c in cores]
+    out.append(("stress-mem", [sys.executable, "-I", "-c", MEM_CODE, str(mem_mb)]))
+    return out
+
+
 # ---------------------------------------------------------------- JACK
 
 
@@ -804,13 +830,17 @@ def plan(self):
         for s in self.table["scenarios"]:
             if self.args.scenarios and s["id"] not in self.args.scenarios:
                 continue
+            loads = {"none": [False], "on": [True], "both": [False, True]}[self.args.stress] if s.get("stress") \
+                else [False]
             if s.get("family"):
-                for m in self.table["modes"][s["family"]]:
-                    if m["host"] in hosts:
-                        rows.append((s, m["host"], m["mode"]))
+                for load in loads:
+                    for m in self.table["modes"][s["family"]]:
+                        if m["host"] in hosts:
+                            rows.append((s, m["host"], m["mode"], load))
             else:
-                for h in hosts:
-                    rows.append((s, h, None))
+                for load in loads:
+                    for h in hosts:
+                        rows.append((s, h, None, load))
         return rows
 
     def plugin_list(self, s):
@@ -821,10 +851,12 @@ def plugin_list(self, s):
             keys = keys * s["repeat"]
         return keys
 
-    def row(self, s, hkey, mode, rate, frames):
+    def row(self, s, hkey, mode, rate, frames, stress=False):
+        cores = stress_cores()
         r = {"id": s["id"], "title": s["title"], "host": hkey, "rate": rate, "frames": frames, "plugins": [],
              "dsp_load_avg": None, "dsp_load_max": None, "samples": 0, "rss_mb": None, "add_ms": None,
-             "recover_ms": None, "status": "failed", "note": ""}
+             "recover_ms": None, "status": "failed", "note": "",
+             "stress": stress_profile(cores) if stress else "none"}
         if mode:
             r["mode"] = mode
         host = None
@@ -890,7 +922,10 @@ def row(self, s, hkey, mode, rate, frames):
                 mode = mode or ("individual" if hkey == "H4" else "shared-process")
             if mode:
                 r["mode"] = mode
-            time.sleep(1.0)
+            if stress:
+                for name, argv in stress_argvs(cores):
+                    sources.append(self.procs.spawn(argv, name))
+            time.sleep(2.0 if stress else 1.0)
 
             pids = host.pids()
             before = samples(pids)
@@ -1034,11 +1069,11 @@ def run(self):
         for rate in self.args.rates:
             for frames in self.args.frames:
                 self.jack_start(rate, frames)
-                for s, h, mode in rows:
-                    r = self.row(s, h, mode, rate, frames)
-                    log("%-8s %s %-14s %6d/%-4d %-11s dsp %s%% rss %s MB add %s ms %s" % (
+                for s, h, mode, stress in rows:
+                    r = self.row(s, h, mode, rate, frames, stress)
+                    log("%-8s %s %-14s %6d/%-4d %-11s dsp %s%% rss %s MB add %s ms xruns %s %s%s" % (
                         r["id"], h, r.get("mode", "-"), rate, frames, r["status"], r["dsp_load_avg"], r["rss_mb"],
-                        r["add_ms"], r["note"]))
+                        r["add_ms"], r.get("xruns"), "" if r["stress"] == "none" else r["stress"] + " ", r["note"]))
                     doc["scenarios"].append(r)
                 self.jack_stop()
         if self.args.qualify:
@@ -1079,6 +1114,8 @@ def main():
     ap.add_argument("--frames", type=csv_ints, default=[128, 256])
     ap.add_argument("--scenarios", type=lambda t: [x for x in t.split(",") if x], default=None)
     ap.add_argument("--hosts", type=lambda t: [x for x in t.split(",") if x], default=list(HOSTS))
+    ap.add_argument("--stress", choices=["none", "on", "both"], default="both",
+                    help="the scenarios that declare a stress run without it, with it, or both")
     ap.add_argument("--window", type=int, default=10, help="seconds of DSP load and /proc sampling per row")
     ap.add_argument("--pianoteq-dir", default=None, help="the directory of Pianoteq's LV2 bundle")
     ap.add_argument("--qualify", action="store_true", help="also time plugin-qualify --all --isolate")
diff --git a/bench/scenarios.json b/bench/scenarios.json
index 025f2a6..be735b2 100644
--- a/bench/scenarios.json
+++ b/bench/scenarios.json
@@ -17,9 +17,9 @@
     { "id": "clap16",   "title": "16 instances of one CLAP plugin, sine in", "plugins": ["comp"], "repeat": 16 },
     { "id": "pianoteq", "title": "Pianoteq 9, five-note chords", "plugins": ["pianoteq"] },
     { "id": "kill1",    "title": "one plugin, its process killed", "plugins": ["comp"], "kill": true },
-    { "id": "cvp3",  "title": "chains vs processes, 3 plugins",  "plugins": ["comp", "calf", "delay", "drive"], "count": 3,  "family": "chains-vs-processes" },
-    { "id": "cvp8",  "title": "chains vs processes, 8 plugins",  "plugins": ["comp", "calf", "delay", "drive"], "count": 8,  "family": "chains-vs-processes" },
-    { "id": "cvp16", "title": "chains vs processes, 16 plugins", "plugins": ["comp", "calf", "delay", "drive"], "count": 16, "family": "chains-vs-processes" }
+    { "id": "cvp3",  "title": "chains vs processes, 3 plugins",  "plugins": ["comp", "calf", "delay", "drive"], "count": 3, "stress": true, "family": "chains-vs-processes" },
+    { "id": "cvp8",  "title": "chains vs processes, 8 plugins",  "plugins": ["comp", "calf", "delay", "drive"], "count": 8, "stress": true, "family": "chains-vs-processes" },
+    { "id": "cvp16", "title": "chains vs processes, 16 plugins", "plugins": ["comp", "calf", "delay", "drive"], "count": 16, "stress": true, "family": "chains-vs-processes" }
   ],
   "modes": {
     "chains-vs-processes": [
diff --git a/bench/tests/sabotage.sh b/bench/tests/sabotage.sh
index 6f1e0d2..9b07259 100755
--- a/bench/tests/sabotage.sh
+++ b/bench/tests/sabotage.sh
@@ -89,6 +89,10 @@ break_unit probe-server lib/bench.py 'self.NO_START | self.SERVER_NAME' 'self.SE
 break_unit probe-deaf lib/bench.py 'if counted == 0 and logged > 0:' 'if False:' Xruns.test_a_deaf_probe_is_failed_not_zero
 break_unit probe-latency lib/bench.py 'return int(rng.max)' 'return int(rng.min)' Xruns.test_the_latency_is_the_longest_of_the_range
 break_unit log-offset lib/bench.py 'f.seek(offset)' 'f.seek(0)' Xruns.test_the_servers_log_counts_new_xrun_lines_only
+break_unit stress-pin lib/bench.py 'os.sched_setaffinity(0, {int(sys.argv[1])})' 'pass' Stress.test_a_busy_loop_runs_on_its_core_only
+break_unit stress-mem lib/bench.py 'bytearray(int(sys.argv[1]) << 20)' 'bytearray(int(sys.argv[1]) << 10)' Stress.test_the_memory_touch_holds_its_size
+break_unit stress-cores lib/bench.py 'return "busy%d+mem%d" % (len(cores), STRESS_MEM_MB)' 'return "busy1+mem%d" % STRESS_MEM_MB' Stress.test_the_profile_names_the_cores_and_the_memory
+break_unit stress-both lib/bench.py '"both": [False, True]' '"both": [True]' Stress.test_stress_rows_are_the_same_scenario_with_and_without
 green "unit tests as they are" python3 -I "$here/tests/test_bench.py"
 
 # --- the teardown
diff --git a/bench/tests/test_bench.py b/bench/tests/test_bench.py
index ab24e31..ebba340 100644
--- a/bench/tests/test_bench.py
+++ b/bench/tests/test_bench.py
@@ -8,6 +8,7 @@
 import subprocess
 import sys
 import tempfile
+import time
 import unittest
 
 HERE = os.path.dirname(os.path.abspath(__file__))
@@ -229,5 +230,74 @@ def test_the_servers_log_counts_new_xrun_lines_only(self):
             self.assertEqual(bench.jack_log_xruns(os.path.join(d, "none"), 5), (0, 5))
 
 
+def proc_field(pid, key):
+    with open("/proc/%d/status" % pid) as f:
+        for line in f:
+            if line.startswith(key + ":"):
+                return line.split(":", 1)[1].strip()
+
+
+class Stress(unittest.TestCase):
+    def test_the_profile_names_the_cores_and_the_memory(self):
+        self.assertEqual(bench.stress_profile([0, 1, 2, 3]), "busy4+mem256")
+        self.assertEqual(bench.stress_profile(list(range(24))), "busy24+mem256")
+
+    def test_the_profile_is_one_the_schema_accepts(self):
+        doc = example()
+        s = first_ok(doc)
+        s.update({"stress": bench.stress_profile([0, 1]), "xruns": 0})
+        code, out = validate(doc)
+        self.assertEqual(code, 0, out)
+
+    def test_one_busy_loop_per_core_and_one_memory_touch(self):
+        argvs = bench.stress_argvs([2, 5, 7])
+        self.assertEqual([n for n, _ in argvs], ["stress-busy-2", "stress-busy-5", "stress-busy-7", "stress-mem"])
+        self.assertEqual([a[-1] for _, a in argvs], ["2", "5", "7", "256"])
+        for _, a in argvs:
+            self.assertEqual(a[1], "-I", "no code from the current directory")
+
+    def test_a_busy_loop_runs_on_its_core_only(self):
+        core = bench.stress_cores()[-1]
+        _, argv = bench.stress_argvs([core])[0]
+        p = subprocess.Popen(argv)
+        try:
+            deadline = time.monotonic() + 5
+            while proc_field(p.pid, "Cpus_allowed_list") != str(core):
+                self.assertLess(time.monotonic(), deadline, "never pinned to core %d" % core)
+                time.sleep(0.02)
+            self.assertEqual(proc_field(p.pid, "State")[0], "R", "it should be busy, not asleep")
+        finally:
+            p.kill()
+            p.wait()
+
+    def test_the_memory_touch_holds_its_size(self):
+        _, argv = bench.stress_argvs([0], mem_mb=64)[-1]
+        p = subprocess.Popen(argv)
+        try:
+            deadline = time.monotonic() + 10
+            kb = 0
+            while kb < 64 * 1024:
+                self.assertLess(time.monotonic(), deadline, "resident only %d kB" % kb)
+                kb = int(proc_field(p.pid, "VmRSS").split()[0])
+                time.sleep(0.05)
+        finally:
+            p.kill()
+            p.wait()
+
+    def test_stress_rows_are_the_same_scenario_with_and_without(self):
+        import types
+        with open(os.path.join(BENCH, "scenarios.json")) as f:
+            table = json.load(f)
+        for choice, want in (("none", [False]), ("on", [True]), ("both", [False, True])):
+            fake = types.SimpleNamespace(args=types.SimpleNamespace(hosts=["H2"], scenarios=["cvp8", "clap1"],
+                                                                    stress=choice), table=table)
+            rows = bench.Runner.plan(fake)
+            self.assertEqual(sorted({st for sc, _, _, st in rows if sc["id"] == "cvp8"}), want)
+            self.assertEqual({st for sc, _, _, st in rows if sc["id"] == "clap1"}, {False},
+                             "a scenario that declares no stress never runs under it")
+            self.assertEqual(sorted(m for sc, _, m, st in rows if sc["id"] == "cvp8" and st == want[0]),
+                             ["chain", "individual"])
+
+
 if __name__ == "__main__":
     unittest.main()

From 7805372a40a66961c9c4220c564d388a931eaaf3 Mon Sep 17 00:00:00 2001
From: Pau Aliagas <linuxnow@gmail.com>
Date: Tue, 27 Oct 2026 06:41:01 +0100
Subject: [PATCH 4/7] Bench: repeat each row and write the median with its
 spread (#188)

--repeats (3 by default) runs a row again and writes the median of every
measure, a count taking the higher middle run, with [min, max] under spread. A
failed repeat fails the row and names which one; an unsupported row is not
repeated.
---
 bench/lib/bench.py        | 55 ++++++++++++++++++++++++-
 bench/tests/sabotage.sh   |  8 +++-
 bench/tests/test_bench.py | 84 +++++++++++++++++++++++++++++++++++++++
 3 files changed, 145 insertions(+), 2 deletions(-)

diff --git a/bench/lib/bench.py b/bench/lib/bench.py
index 2e3c274..5b627a6 100644
--- a/bench/lib/bench.py
+++ b/bench/lib/bench.py
@@ -11,6 +11,7 @@
 import shutil
 import signal
 import socket
+import statistics
 import subprocess
 import sys
 import tempfile
@@ -222,6 +223,45 @@ def samples(pids):
     return out
 
 
+# ---------------------------------------------------------------- repeats
+
+MEASURED = ("dsp_load_avg", "dsp_load_max", "rss_mb", "add_ms", "recover_ms", "processes", "rss_mb_total",
+            "pss_mb_total", "cpu_pct", "ctx_switches", "ctx_involuntary", "xruns", "latency_frames")
+COUNTS = ("processes", "ctx_switches", "ctx_involuntary", "xruns", "latency_frames", "samples")
+
+
+def aggregate(runs, wanted=None):
+    """one row from the repeats of one row: the median of every measure and, for each, its [min, max] under
+    `spread`. A count stays a count: with an even number of runs it takes the higher middle one, so a number of xruns
+    is never understated. A run that is failed ends it: that run's row, naming the repeat of `wanted`, and no figures
+    from the runs before it; one that is unsupported ends it as it is."""
+    for n, run in enumerate(runs, 1):
+        if run["status"] != "ok":
+            row = dict(run)
+            if row["status"] == "failed" and (wanted or len(runs)) > 1:
+                row["note"] = "repeat %d of %d: %s" % (n, wanted or len(runs), row["note"])
+            return row
+    row = dict(runs[0])
+    row["repeats"] = len(runs)
+    if len(runs) == 1:
+        return row
+    spread = {}
+    for key in MEASURED + ("samples",):
+        vals = [r[key] for r in runs if r.get(key) is not None]
+        if len(vals) != len(runs):
+            continue
+        if key in COUNTS:
+            row[key] = int(statistics.median_high(vals))
+        else:
+            row[key] = round(statistics.median(vals), 2)
+        if key != "samples":
+            spread[key] = [min(vals), max(vals)]
+    row["spread"] = spread
+    notes = list(dict.fromkeys(r["note"] for r in runs if r["note"]))
+    row["note"] = "; ".join(notes)
+    return row
+
+
 # ---------------------------------------------------------------- the stress load
 
 STRESS_MEM_MB = 256
@@ -976,6 +1016,15 @@ def row(self, s, hkey, mode, rate, frames, stress=False):
             self.assert_only_jack("after %s %s %d/%d" % (s["id"], hkey, rate, frames))
         return r
 
+    def measure(self, s, hkey, mode, rate, frames, stress):
+        """the row, run args.repeats times; a row that cannot run, or breaks, ends the repeats"""
+        runs = []
+        for _ in range(self.args.repeats):
+            runs.append(self.row(s, hkey, mode, rate, frames, stress))
+            if runs[-1]["status"] != "ok":
+                break
+        return aggregate(runs, self.args.repeats)
+
     def chain_formed(self, host, insts):
         want = ",".join(map(str, insts))
         end = time.monotonic() + READY_LIMIT_S
@@ -1070,7 +1119,7 @@ def run(self):
             for frames in self.args.frames:
                 self.jack_start(rate, frames)
                 for s, h, mode, stress in rows:
-                    r = self.row(s, h, mode, rate, frames, stress)
+                    r = self.measure(s, h, mode, rate, frames, stress)
                     log("%-8s %s %-14s %6d/%-4d %-11s dsp %s%% rss %s MB add %s ms xruns %s %s%s" % (
                         r["id"], h, r.get("mode", "-"), rate, frames, r["status"], r["dsp_load_avg"], r["rss_mb"],
                         r["add_ms"], r.get("xruns"), "" if r["stress"] == "none" else r["stress"] + " ", r["note"]))
@@ -1116,12 +1165,16 @@ def main():
     ap.add_argument("--hosts", type=lambda t: [x for x in t.split(",") if x], default=list(HOSTS))
     ap.add_argument("--stress", choices=["none", "on", "both"], default="both",
                     help="the scenarios that declare a stress run without it, with it, or both")
+    ap.add_argument("--repeats", type=int, default=3, help="runs of each row; the document holds the median and the "
+                    "spread (1 for a quick look, 3 at least for a document that says anything)")
     ap.add_argument("--window", type=int, default=10, help="seconds of DSP load and /proc sampling per row")
     ap.add_argument("--pianoteq-dir", default=None, help="the directory of Pianoteq's LV2 bundle")
     ap.add_argument("--qualify", action="store_true", help="also time plugin-qualify --all --isolate")
     ap.add_argument("--keep-governor", action="store_true", help="leave the CPU governor as it is")
     ap.add_argument("--logs", default=None, help="keep the host logs in this directory")
     args = ap.parse_args()
+    if args.repeats < 1:
+        ap.error("--repeats takes 1 or more")
     if set(args.hosts) - set(HOSTS):
         ap.error("--hosts takes H1,H2,H3,H4")
     table = json.load(open(os.path.join(BENCH, "scenarios.json")))
diff --git a/bench/tests/sabotage.sh b/bench/tests/sabotage.sh
index 9b07259..5b5da2e 100755
--- a/bench/tests/sabotage.sh
+++ b/bench/tests/sabotage.sh
@@ -75,7 +75,7 @@ s = open(p).read()
 assert s.count(old) == 1, "the sabotage did not land: " + old
 open(p, "w").write(s.replace(old, new))
 PY
-    red "unit: $1" "FAIL: ${5##*.}" python3 -I "$copy/tests/test_bench.py" "$5"
+    red "unit: $1" "(FAIL|ERROR): ${5##*.}" python3 -I "$copy/tests/test_bench.py" "$5"
 }
 break_unit spread-median lib/validate.py 'elif isinstance(s[field], (int, float)) and not span[0] <= s[field] <= span[1]:' \
     'elif False:' Schema.test_a_spread_that_does_not_hold_the_median_is_red
@@ -93,6 +93,12 @@ break_unit stress-pin lib/bench.py 'os.sched_setaffinity(0, {int(sys.argv[1])})'
 break_unit stress-mem lib/bench.py 'bytearray(int(sys.argv[1]) << 20)' 'bytearray(int(sys.argv[1]) << 10)' Stress.test_the_memory_touch_holds_its_size
 break_unit stress-cores lib/bench.py 'return "busy%d+mem%d" % (len(cores), STRESS_MEM_MB)' 'return "busy1+mem%d" % STRESS_MEM_MB' Stress.test_the_profile_names_the_cores_and_the_memory
 break_unit stress-both lib/bench.py '"both": [False, True]' '"both": [True]' Stress.test_stress_rows_are_the_same_scenario_with_and_without
+break_unit rep-high lib/bench.py 'statistics.median_high(vals)' 'statistics.median_low(vals)' Repeats.test_a_count_stays_a_count_and_is_never_understated
+break_unit rep-mean lib/bench.py 'round(statistics.median(vals), 2)' 'round(statistics.mean(vals), 2)' Repeats.test_the_median_is_the_middle_run_not_the_mean
+break_unit rep-spread lib/bench.py 'spread[key] = [min(vals), max(vals)]' 'spread[key] = [min(vals), min(vals)]' Repeats.test_the_median_and_the_spread_of_three_runs
+break_unit rep-failed lib/bench.py 'if row["status"] == "failed" and (wanted or len(runs)) > 1:' 'if False:' Repeats.test_a_failed_repeat_fails_the_row_and_names_it
+break_unit rep-stop lib/bench.py 'if runs[-1]["status"] != "ok":' 'if False:' Repeats.test_the_runner_stops_repeating_at_the_first_row_that_is_not_ok
+break_unit rep-count lib/bench.py 'for _ in range(self.args.repeats):' 'for _ in range(1):' Repeats.test_the_runner_runs_the_row_as_many_times_as_asked
 green "unit tests as they are" python3 -I "$here/tests/test_bench.py"
 
 # --- the teardown
diff --git a/bench/tests/test_bench.py b/bench/tests/test_bench.py
index ebba340..c7f7a91 100644
--- a/bench/tests/test_bench.py
+++ b/bench/tests/test_bench.py
@@ -299,5 +299,89 @@ def test_stress_rows_are_the_same_scenario_with_and_without(self):
                              ["chain", "individual"])
 
 
+def run_row(**kw):
+    row = {"id": "cvp8", "title": "t", "host": "H2", "mode": "chain", "rate": 48000, "frames": 256, "plugins": ["a"],
+           "dsp_load_avg": 4.0, "dsp_load_max": 5.0, "samples": 10, "rss_mb": 100.0, "add_ms": 50.0,
+           "recover_ms": None, "status": "ok", "note": "", "processes": 2, "rss_mb_total": 100.0,
+           "pss_mb_total": 40.0, "cpu_pct": 7.0, "ctx_switches": 1000, "ctx_involuntary": 10, "xruns": 0,
+           "latency_frames": 256, "stress": "none"}
+    row.update(kw)
+    return row
+
+
+class Repeats(unittest.TestCase):
+    def test_the_median_and_the_spread_of_three_runs(self):
+        runs = [run_row(xruns=0, cpu_pct=7.0), run_row(xruns=5, cpu_pct=9.5), run_row(xruns=2, cpu_pct=8.0)]
+        row = bench.aggregate(runs, 3)
+        self.assertEqual(row["xruns"], 2)
+        self.assertEqual(row["spread"]["xruns"], [0, 5])
+        self.assertEqual(row["cpu_pct"], 8.0)
+        self.assertEqual(row["spread"]["cpu_pct"], [7.0, 9.5])
+        self.assertEqual(row["repeats"], 3)
+        self.assertEqual(row["status"], "ok")
+
+    def test_a_count_stays_a_count_and_is_never_understated(self):
+        row = bench.aggregate([run_row(xruns=1, ctx_switches=10), run_row(xruns=4, ctx_switches=20)], 2)
+        self.assertEqual(row["xruns"], 4)
+        self.assertIsInstance(row["xruns"], int)
+        self.assertEqual(row["ctx_switches"], 20)
+        self.assertEqual(row["spread"]["xruns"], [1, 4])
+
+    def test_the_median_is_the_middle_run_not_the_mean(self):
+        row = bench.aggregate([run_row(dsp_load_avg=1.0), run_row(dsp_load_avg=2.0), run_row(dsp_load_avg=90.0)], 3)
+        self.assertEqual(row["dsp_load_avg"], 2.0)
+
+    def test_the_aggregate_row_is_a_valid_row(self):
+        doc = example()
+        doc["scenarios"] = [bench.aggregate([run_row(xruns=x, recover_ms=r) for x, r in
+                                             ((0, 900.0), (3, 1100.0), (1, 1000.0))], 3)]
+        code, out = validate(doc)
+        self.assertEqual(code, 0, out)
+
+    def test_one_run_is_a_row_with_no_spread(self):
+        row = bench.aggregate([run_row(xruns=3)], 1)
+        self.assertEqual(row["xruns"], 3)
+        self.assertEqual(row["repeats"], 1)
+        self.assertNotIn("spread", row)
+
+    def test_a_measure_missing_from_a_run_has_no_median(self):
+        row = bench.aggregate([run_row(recover_ms=None), run_row(recover_ms=1000.0), run_row(recover_ms=None)], 3)
+        self.assertIsNone(row["recover_ms"])
+        self.assertNotIn("recover_ms", row["spread"])
+
+    def test_a_failed_repeat_fails_the_row_and_names_it(self):
+        runs = [run_row(xruns=0), run_row(status="failed", note="the chain did not form", xruns=7)]
+        row = bench.aggregate(runs, 3)
+        self.assertEqual(row["status"], "failed")
+        self.assertEqual(row["note"], "repeat 2 of 3: the chain did not form")
+        self.assertNotIn("spread", row)
+        self.assertNotIn("repeats", row)
+
+    def test_an_unsupported_row_is_not_a_failure_of_a_repeat(self):
+        row = bench.aggregate([run_row(status="unsupported", note="jalv is not installed")], 3)
+        self.assertEqual(row["status"], "unsupported")
+        self.assertEqual(row["note"], "jalv is not installed")
+
+    def test_the_runner_stops_repeating_at_the_first_row_that_is_not_ok(self):
+        import types
+        calls = []
+        answers = [run_row(), run_row(status="failed", note="broke")]
+        fake = types.SimpleNamespace(args=types.SimpleNamespace(repeats=3))
+        fake.row = lambda *a: calls.append(a) or answers[len(calls) - 1]
+        row = bench.Runner.measure(fake, {}, "H2", "chain", 48000, 256, False)
+        self.assertEqual(len(calls), 2)
+        self.assertEqual(row["status"], "failed")
+
+    def test_the_runner_runs_the_row_as_many_times_as_asked(self):
+        import types
+        calls = []
+        fake = types.SimpleNamespace(args=types.SimpleNamespace(repeats=3))
+        fake.row = lambda *a: calls.append(a) or run_row(xruns=len(calls))
+        row = bench.Runner.measure(fake, {}, "H2", "chain", 48000, 256, False)
+        self.assertEqual(len(calls), 3)
+        self.assertEqual(row["xruns"], 2)
+        self.assertEqual(row["spread"]["xruns"], [1, 3])
+
+
 if __name__ == "__main__":
     unittest.main()

From 291cb9fd02988c9db023edb0beb0f9d24fd13522 Mon Sep 17 00:00:00 2001
From: Pau Aliagas <linuxnow@gmail.com>
Date: Tue, 27 Oct 2026 06:46:50 +0100
Subject: [PATCH 5/7] Bench: plan and pool modes beside individual and chain
 (#188)

The grouping-vs-processes family (grp8, grp16) runs one strip four ways under
plugin-hostd: one process per plugin, one chain, plan, and plan with the
shareable members it left alone in a pool. The plan and pool rows run with pins
on, a verdict per plugin (isolated for the ones the strip names, in-process for
the rest) written into the daemon's own pin store, and grouping on. A plan of
one segment, a pool the daemon did not form, or a daemon without the grouping
setting is a row that says so, never a number.
---
 bench/lib/bench.py        | 144 +++++++++++++++++++++-
 bench/scenarios.json      |  14 ++-
 bench/tests/sabotage.sh   |  10 ++
 bench/tests/test_bench.py | 244 ++++++++++++++++++++++++++++++++++++++
 4 files changed, 407 insertions(+), 5 deletions(-)

diff --git a/bench/lib/bench.py b/bench/lib/bench.py
index 5b627a6..0949fe7 100644
--- a/bench/lib/bench.py
+++ b/bench/lib/bench.py
@@ -27,6 +27,7 @@
 TAG = "omx-bench-%d" % os.getpid()
 RECOVER_LIMIT_S = 30.0
 READY_LIMIT_S = 20.0
+POOL_MAX = 8  # plugin-hostd's pool_max: a full pool opens a second one, which these rows do not read
 
 
 def log(msg):
@@ -402,6 +403,23 @@ def close(self):
             self.client = None
 
 
+def log_size(logdir, name):
+    try:
+        return os.path.getsize(os.path.join(logdir, name + ".log"))
+    except OSError:
+        return 0
+
+
+def log_since(logdir, name, start):
+    """what a host wrote to its log after byte start"""
+    try:
+        with open(os.path.join(logdir, name + ".log"), "rb") as f:
+            f.seek(start)
+            return f.read().decode(errors="replace")
+    except OSError:
+        return ""
+
+
 def jack_log_xruns(path, offset):
     """lines of the JACK server's own log from offset on that name an xrun, and the new end of the file"""
     try:
@@ -671,14 +689,34 @@ def files(self):
         return list(dict.fromkeys(out))
 
 
+# ---------------------------------------------------------------- pins for the plan and pool modes
+
+
+def set_verdicts(store, verdicts):
+    """the daemon's pin store with a verdict on the line of each uri in verdicts ({uri: word}), replacing any it had;
+    the other lines as they were. Failed when a uri has no line: the daemon never learned it"""
+    out, seen = [], set()
+    for line in store.splitlines():
+        words = line.split()
+        if len(words) >= 4 and words[0] == "pin_set" and words[1] in verdicts:
+            seen.add(words[1])
+            line = " ".join([w for w in words if not w.startswith("verdict=")] + ["verdict=" + verdicts[words[1]]])
+        out.append(line)
+    missing = sorted(set(verdicts) - seen)
+    if missing:
+        raise Failed("the daemon recorded no pin for %s" % ", ".join(missing))
+    return "\n".join(out) + "\n"
+
+
 # ---------------------------------------------------------------- hosts
 
 
 class Host:
     """one host program started for one scenario row"""
 
-    def __init__(self, key, procs, env, workdir, port):
+    def __init__(self, key, procs, env, workdir, port, conf=("require_pins 0",)):
         self.key, self.procs, self.env, self.workdir, self.port = key, procs, env, workdir, port
+        self.conf = conf
         self.proc = None
         self.ctl = None
         self.jalvs = {}
@@ -689,7 +727,7 @@ def start(self):
         elif self.key == "H2":
             conf = os.path.join(self.workdir, "plugin-hostd.conf")
             with open(conf, "w") as f:
-                f.write("require_pins 0\n")
+                f.write("".join(line + "\n" for line in self.conf))
             self.proc = self.procs.spawn(["plugin-hostd", "-n", "-p", str(self.port), "-c", conf], "H2", self.env)
         elif self.key == "H3":
             self.proc = self.procs.spawn(["mod-host", "-n", "-p", str(self.port)], "H3", self.env)
@@ -718,6 +756,53 @@ def workers(self):
             out.append((f[0], int(f[1]), f[2], f[3], ":".join(f[4:-1]), insts))
         return out
 
+    def wait_workers(self, want, what):
+        """until worker_list shows `want` workers, all up; the workers, or Failed after READY_LIMIT_S"""
+        end = time.monotonic() + READY_LIMIT_S
+        while True:
+            ws = self.workers()
+            if len(ws) == want and all(w[3] == "up" for w in ws):
+                return ws
+            if time.monotonic() > end:
+                raise Failed("%s: %d workers up of %d wanted after %.0f s" % (
+                    what, sum(1 for w in ws if w[3] == "up"), want, READY_LIMIT_S))
+            time.sleep(0.01)
+
+    def declare_plan(self, insts):
+        """plan 1 <insts>: the segments the daemon made, [[inst, ...], ...], once their workers are up"""
+        reply = self.ctl.cmd("plan 1 " + " ".join(map(str, insts)))
+        words = reply.split()
+        if len(words) < 3 or words[:2] != ["resp", "0"]:
+            raise Failed("plan answered '%s'" % reply)
+        segments = [[int(i) for i in seg.split(",")] for seg in words[2].split(";")]
+        if sorted(i for seg in segments for i in seg) != sorted(insts):
+            raise Failed("plan answered segments '%s' for instances %s" % (words[2], insts))
+        if len(segments) < 2:
+            raise Failed("the plan is one segment: no member of the strip was left alone, so it is the chain row")
+        self.wait_workers(len(segments), "plan")
+        return segments
+
+    def declare_pool(self, insts, uris, alone, name="g1"):
+        """plan, then the members the plan left alone that the pins would let share (every segment of one that is not
+        an `alone` instance) leave their workers for the pool: remove and add again with the pool's place. Returns
+        (segments, pooled instances)"""
+        segments = self.declare_plan(insts)
+        pooled = [seg[0] for seg in segments if len(seg) == 1 and seg[0] not in alone]
+        if len(pooled) < 2:
+            raise Failed("the plan left %d shareable member(s) alone: a pool needs two" % len(pooled))
+        if len(pooled) > POOL_MAX:
+            raise Failed("%d members for one pool, which holds %d" % (len(pooled), POOL_MAX))
+        for i in pooled:
+            reply = self.ctl.cmd("remove %d" % i)
+            if reply != "resp 0":
+                raise Failed("remove %d answered '%s'" % (i, reply))
+            self.add(uris[i], i, "pool:" + name)
+        ws = self.wait_workers(len(segments) - len(pooled) + 1, "pool")
+        holder = [w for w in ws if w[4] == "pool:" + name]
+        if len(holder) != 1 or sorted(holder[0][5]) != sorted(pooled):
+            raise Failed("pool:%s holds %s, not %s" % (name, [w[5] for w in holder], pooled))
+        return segments, pooled
+
     def add(self, uri, inst, placement=None):
         """add one plugin; returns the client name"""
         client = "effect_%d" % inst
@@ -798,6 +883,7 @@ def __init__(self, args, table):
         self.saved_governor = {}
         self.jackd = None
         self.probe = None
+        self.learned = {}
         self.jack_log = None
         self.jack_log_end = 0
         self.port = 5700
@@ -917,8 +1003,22 @@ def row(self, s, hkey, mode, rate, frames, stress=False):
 
             self.port = free_port(self.port + 2)
             wd = tempfile.mkdtemp(dir=self.work)
-            host = Host(hkey, self.procs, self.env, wd, self.port)
-            host.start()
+            conf, alone, uris = ("require_pins 0",), set(), [u for u, _, _ in picks]
+            if mode in ("plan", "pool"):
+                if hkey != "H2":
+                    raise Refused("only plugin-hostd plans and pools")
+                alone = {i for i, k in enumerate(keys) if k in s.get("unshared", [])}
+                conf = self.plan_conf(wd, uris, {uris[i] for i in alone})
+                notes.append("pins: in-process, isolated for %s" % ", ".join(
+                    dict.fromkeys(self.cat.name(keys[i]) for i in sorted(alone))))
+            host = Host(hkey, self.procs, self.env, wd, self.port, conf)
+            log_end = log_size(self.logdir, hkey)
+            try:
+                host.start()
+            except Failed:
+                if mode in ("plan", "pool") and "bad setting 'grouping'" in log_since(self.logdir, hkey, log_end):
+                    raise Refused("this plugin-hostd has no grouping setting: it predates the planner (plugin-hostd#185)")
+                raise
             if instrument:
                 sources.append(self.procs.spawn(["jack_midiseq", "seq"] + self.cat.midi["chords"], "midiseq"))
                 want = "seq"
@@ -943,6 +1043,13 @@ def row(self, s, hkey, mode, rate, frames, stress=False):
                 if reply != "resp 0":
                     raise Failed("chain answered '%s'" % reply)
                 self.chain_formed(host, insts)
+            elif mode == "plan":
+                segments = host.declare_plan(insts)
+                notes.append("plan: " + ";".join(",".join(map(str, g)) for g in segments))
+            elif mode == "pool":
+                segments, pooled = host.declare_pool(insts, uris, alone)
+                notes.append("plan: " + ";".join(",".join(map(str, g)) for g in segments)
+                             + "; pool:g1 holds " + ",".join(map(str, pooled)))
             add_ms = (time.monotonic() - t0) * 1000 if insts else None
             if clients:
                 graph = ports(self.env)
@@ -1016,6 +1123,35 @@ def row(self, s, hkey, mode, rate, frames, stress=False):
             self.assert_only_jack("after %s %s %d/%d" % (s["id"], hkey, rate, frames))
         return r
 
+    def plan_conf(self, wd, uris, alone):
+        """the conf of a plan or pool row: pins on, the store holding a verdict for each plugin the strip uses
+        (in-process, and isolated for those in `alone`), grouping on, no package pins to say otherwise"""
+        store = os.path.join(wd, "pins")
+        learned = self.learn_pins(uris)
+        with open(store, "w") as f:
+            f.write(set_verdicts(learned, {u: "isolated" if u in alone else "in-process" for u in uris}))
+        return ("require_pins 1", "pin_mode first-use", "pin_store " + store, "pins_dir", "grouping on")
+
+    def learn_pins(self, uris):
+        """the pin store a daemon writes when it first sees each plugin (files hashed, layout learned): the pins a
+        plugin-qualify run would write, minus the verdicts. One daemon start for each set of plugins in the run"""
+        key = tuple(dict.fromkeys(uris))
+        if key not in self.learned:
+            wd = tempfile.mkdtemp(dir=self.work)
+            store = os.path.join(wd, "pins")
+            self.port = free_port(self.port + 2)
+            host = Host("H2", self.procs, self.env, wd, self.port,
+                        ("require_pins 1", "pin_mode first-use", "pin_store " + store, "pins_dir"))
+            try:
+                host.start()
+                for i, uri in enumerate(key):
+                    host.add(uri, i, "own")
+            finally:
+                host.stop()
+            with open(store) as f:
+                self.learned[key] = f.read()
+        return self.learned[key]
+
     def measure(self, s, hkey, mode, rate, frames, stress):
         """the row, run args.repeats times; a row that cannot run, or breaks, ends the repeats"""
         runs = []
diff --git a/bench/scenarios.json b/bench/scenarios.json
index be735b2..c0dd76e 100644
--- a/bench/scenarios.json
+++ b/bench/scenarios.json
@@ -19,7 +19,13 @@
     { "id": "kill1",    "title": "one plugin, its process killed", "plugins": ["comp"], "kill": true },
     { "id": "cvp3",  "title": "chains vs processes, 3 plugins",  "plugins": ["comp", "calf", "delay", "drive"], "count": 3, "stress": true, "family": "chains-vs-processes" },
     { "id": "cvp8",  "title": "chains vs processes, 8 plugins",  "plugins": ["comp", "calf", "delay", "drive"], "count": 8, "stress": true, "family": "chains-vs-processes" },
-    { "id": "cvp16", "title": "chains vs processes, 16 plugins", "plugins": ["comp", "calf", "delay", "drive"], "count": 16, "stress": true, "family": "chains-vs-processes" }
+    { "id": "cvp16", "title": "chains vs processes, 16 plugins", "plugins": ["comp", "calf", "delay", "drive"], "count": 16, "stress": true, "family": "chains-vs-processes" },
+    { "id": "grp8",  "title": "grouping vs processes, 8 plugins, some of them pinned isolated",
+      "plugins": ["comp", "drive", "delay", "drive", "comp"], "count": 8, "unshared": ["drive"], "stress": true,
+      "family": "grouping-vs-processes" },
+    { "id": "grp16", "title": "grouping vs processes, 16 plugins, some of them pinned isolated",
+      "plugins": ["comp", "drive", "delay", "drive", "comp"], "count": 16, "unshared": ["drive"], "stress": true,
+      "family": "grouping-vs-processes" }
   ],
   "modes": {
     "chains-vs-processes": [
@@ -28,6 +34,12 @@
       { "host": "H1", "mode": "shared-process" },
       { "host": "H3", "mode": "shared-process" },
       { "host": "H4", "mode": "individual" }
+    ],
+    "grouping-vs-processes": [
+      { "host": "H2", "mode": "individual" },
+      { "host": "H2", "mode": "chain" },
+      { "host": "H2", "mode": "plan" },
+      { "host": "H2", "mode": "pool" }
     ]
   }
 }
diff --git a/bench/tests/sabotage.sh b/bench/tests/sabotage.sh
index 5b5da2e..9aa0b92 100755
--- a/bench/tests/sabotage.sh
+++ b/bench/tests/sabotage.sh
@@ -99,6 +99,16 @@ break_unit rep-spread lib/bench.py 'spread[key] = [min(vals), max(vals)]' 'sprea
 break_unit rep-failed lib/bench.py 'if row["status"] == "failed" and (wanted or len(runs)) > 1:' 'if False:' Repeats.test_a_failed_repeat_fails_the_row_and_names_it
 break_unit rep-stop lib/bench.py 'if runs[-1]["status"] != "ok":' 'if False:' Repeats.test_the_runner_stops_repeating_at_the_first_row_that_is_not_ok
 break_unit rep-count lib/bench.py 'for _ in range(self.args.repeats):' 'for _ in range(1):' Repeats.test_the_runner_runs_the_row_as_many_times_as_asked
+break_unit plan-one lib/bench.py 'if len(segments) < 2:' 'if False:' Grouping.test_a_plan_of_one_segment_fails_the_row
+break_unit plan-reply lib/bench.py 'words[:2] != ["resp", "0"]' 'False' Grouping.test_a_refused_plan_is_failed_with_the_daemons_words
+break_unit pool-alone lib/bench.py 'and seg[0] not in alone]' 'and True]' Grouping.test_the_pool_takes_the_shareable_members_the_plan_left_alone
+break_unit pool-holder lib/bench.py 'if len(holder) != 1 or sorted(holder[0][5]) != sorted(pooled):' 'if False:' Grouping.test_a_pool_under_another_name_is_failed
+break_unit pool-min lib/bench.py 'if len(pooled) < 2:' 'if False:' Grouping.test_a_pool_of_one_is_not_a_pool
+break_unit pins-replace lib/bench.py '[w for w in words if not w.startswith("verdict=")]' 'list(words)' Pins.test_each_uri_gets_its_verdict_and_the_rest_is_kept
+break_unit pins-missing lib/bench.py 'if missing:' 'if False:' Pins.test_a_uri_the_daemon_never_learned_is_failed
+break_unit conf-grouping lib/bench.py '"pins_dir", "grouping on")' '"pins_dir")' Pins.test_the_conf_of_a_plan_row_turns_pins_and_grouping_on_and_package_pins_off
+break_unit conf-verdict lib/bench.py '"isolated" if u in alone else "in-process"' '"in-process" if u in alone else "isolated"' Pins.test_the_conf_of_a_plan_row_turns_pins_and_grouping_on_and_package_pins_off
+break_unit scenario-strip scenarios.json '"count": 8, "unshared": ["drive"]' '"count": 8, "unshared": ["comp", "delay", "drive"]' Scenarios.test_the_grouping_scenarios_have_all_four_modes_and_a_strip_the_pins_cut
 green "unit tests as they are" python3 -I "$here/tests/test_bench.py"
 
 # --- the teardown
diff --git a/bench/tests/test_bench.py b/bench/tests/test_bench.py
index c7f7a91..13dc861 100644
--- a/bench/tests/test_bench.py
+++ b/bench/tests/test_bench.py
@@ -383,5 +383,249 @@ def test_the_runner_runs_the_row_as_many_times_as_asked(self):
         self.assertEqual(row["spread"]["xruns"], [1, 3])
 
 
+class FakeDaemon:
+    """plugin-hostd as far as the plan and pool modes talk to it, on a socket, NUL-framed like the real one: add,
+    remove, plan, worker_list. Its planner is the simplest honest one: runs of consecutive members that are not
+    `isolated` share a worker. `pool_apart` makes it put each pooled member in a worker of its own, and `pool_as` puts them
+    in a pool of another name, as a daemon that ignored the place would"""
+
+    def __init__(self, isolated, pool_apart=False, one_segment=False, pool_as=None):
+        import socketserver
+        import threading
+        self.isolated, self.pool_apart, self.one_segment, self.pool_as = set(isolated), pool_apart, one_segment, pool_as
+        self.inst = {}  # instance -> place
+        self.log = []
+        daemon = self
+
+        class Handler(socketserver.BaseRequestHandler):
+            def handle(self):
+                buf = b""
+                while True:
+                    chunk = self.request.recv(4096)
+                    if not chunk:
+                        return
+                    buf += chunk
+                    while b"\0" in buf:
+                        line, buf = buf.split(b"\0", 1)
+                        self.request.sendall(daemon.reply(line.decode()).encode() + b"\0")
+
+        self.server = socketserver.ThreadingTCPServer(("127.0.0.1", 0), Handler)
+        self.server.daemon_threads = True
+        self.port = self.server.server_address[1]
+        threading.Thread(target=self.server.serve_forever, daemon=True).start()
+
+    def close(self):
+        self.server.shutdown()
+        self.server.server_close()
+
+    def reply(self, line):
+        self.log.append(line)
+        w = line.split()
+        if w[0] == "add":
+            place = w[3] if len(w) > 3 else "own"
+            self.inst[int(w[2])] = self.pool_as if self.pool_as and place.startswith("pool:") else place
+            return "resp %s" % w[2]
+        if w[0] == "remove":
+            self.inst.pop(int(w[1]))
+            return "resp 0"
+        if w[0] == "plan":
+            members = [int(x) for x in w[2:]]
+            segs, cur = [], []
+            for i in members:
+                if i in self.isolated:
+                    if cur:
+                        segs.append(cur)
+                    segs.append([i])
+                    cur = []
+                else:
+                    cur.append(i)
+            if cur:
+                segs.append(cur)
+            if self.one_segment:
+                segs = [members]
+            for k, seg in enumerate(segs):
+                for i in seg:
+                    self.inst[i] = "chain:1.%d" % k if len(seg) > 1 else "own"
+            return "resp 0 " + ";".join(",".join(map(str, g)) for g in segs)
+        if w[0] == "worker_list":
+            workers = {}
+            for i, place in sorted(self.inst.items()):
+                key = place if place.startswith(("chain:", "pool:")) and not (
+                    self.pool_apart and place.startswith("pool:")) else "own%d" % i
+                workers.setdefault(key, (place, []))[1].append(i)
+            recs = ["%d:%d:clap:up:%s:%s" % (n, 1000 + n, place, ",".join(map(str, ids)))
+                    for n, (place, ids) in enumerate(workers.values())]
+            return "resp 0 " + " ".join(recs)
+        return "resp -902"
+
+
+class Gone:
+    def poll(self):
+        return None
+
+
+def daemon_host(daemon, case):
+    host = bench.Host.__new__(bench.Host)
+    host.key = "H2"
+    host.ctl = bench.Control(daemon.port, Gone())
+    case.addCleanup(host.ctl.close)
+    return host
+
+
+class Grouping(unittest.TestCase):
+    def setUp(self):
+        keep = bench.READY_LIMIT_S
+        bench.READY_LIMIT_S = 0.5
+        self.addCleanup(setattr, bench, "READY_LIMIT_S", keep)
+
+    uris = ["clap:/c#comp", "clap:/d#delay", "clap:/x#drive", "clap:/c#comp", "clap:/x#drive", "clap:/c#comp"]
+
+    def test_the_plan_is_read_as_segments_and_waited_for(self):
+        d = FakeDaemon(isolated=[2, 4])
+        try:
+            host = daemon_host(d, self)
+            for i, u in enumerate(self.uris):
+                host.add(u, i)
+            segs = host.declare_plan(list(range(6)))
+            self.assertEqual(segs, [[0, 1], [2], [3], [4], [5]])
+            self.assertIn("plan 1 0 1 2 3 4 5", d.log)
+        finally:
+            d.close()
+
+    def test_a_plan_of_one_segment_fails_the_row(self):
+        d = FakeDaemon(isolated=[], one_segment=True)
+        try:
+            host = daemon_host(d, self)
+            for i, u in enumerate(self.uris):
+                host.add(u, i)
+            with self.assertRaisesRegex(bench.Failed, "one segment"):
+                host.declare_plan(list(range(6)))
+        finally:
+            d.close()
+
+    def test_a_refused_plan_is_failed_with_the_daemons_words(self):
+        d = FakeDaemon(isolated=[])
+        d.reply = lambda line: "resp -501 0,1"
+        try:
+            with self.assertRaisesRegex(bench.Failed, "plan answered 'resp -501 0,1'"):
+                daemon_host(d, self).declare_plan([0, 1])
+        finally:
+            d.close()
+
+    def test_the_pool_takes_the_shareable_members_the_plan_left_alone(self):
+        d = FakeDaemon(isolated=[2, 4])
+        try:
+            host = daemon_host(d, self)
+            for i, u in enumerate(self.uris):
+                host.add(u, i)
+            segs, pooled = host.declare_pool(list(range(6)), self.uris, alone={2, 4})
+            self.assertEqual(segs, [[0, 1], [2], [3], [4], [5]])
+            self.assertEqual(pooled, [3, 5], "the isolated members stay out of the pool")
+            self.assertEqual(d.inst[3], "pool:g1")
+            self.assertEqual(d.inst[5], "pool:g1")
+            self.assertEqual(d.inst[2], "own")
+            self.assertIn("add clap:/c#comp 3 pool:g1", d.log)
+        finally:
+            d.close()
+
+    def test_a_pool_the_daemon_split_is_failed(self):
+        d = FakeDaemon(isolated=[2, 4], pool_apart=True)
+        try:
+            host = daemon_host(d, self)
+            for i, u in enumerate(self.uris):
+                host.add(u, i)
+            with self.assertRaises(bench.Failed):
+                host.declare_pool(list(range(6)), self.uris, alone={2, 4})
+        finally:
+            d.close()
+
+    def test_a_pool_under_another_name_is_failed(self):
+        d = FakeDaemon(isolated=[2, 4], pool_as="pool:other")
+        try:
+            host = daemon_host(d, self)
+            for i, u in enumerate(self.uris):
+                host.add(u, i)
+            with self.assertRaisesRegex(bench.Failed, "pool:g1 holds"):
+                host.declare_pool(list(range(6)), self.uris, alone={2, 4})
+        finally:
+            d.close()
+
+    def test_a_pool_of_one_is_not_a_pool(self):
+        d = FakeDaemon(isolated=[2, 3, 4])
+        try:
+            host = daemon_host(d, self)
+            for i, u in enumerate(self.uris):
+                host.add(u, i)
+            with self.assertRaisesRegex(bench.Failed, "left 1 shareable"):
+                host.declare_pool(list(range(6)), self.uris, alone={2, 3, 4})
+        finally:
+            d.close()
+
+
+class Scenarios(unittest.TestCase):
+    def test_every_name_a_scenario_uses_is_declared(self):
+        with open(os.path.join(BENCH, "scenarios.json")) as f:
+            table = json.load(f)
+        for sc in table["scenarios"]:
+            for key in sc["plugins"] + sc.get("unshared", []):
+                self.assertIn(key, table["plugins"], sc["id"])
+            if sc.get("family"):
+                self.assertIn(sc["family"], table["modes"], sc["id"])
+            self.assertTrue(set(sc.get("unshared", [])) <= set(sc["plugins"]), sc["id"])
+
+    def test_the_grouping_scenarios_have_all_four_modes_and_a_strip_the_pins_cut(self):
+        with open(os.path.join(BENCH, "scenarios.json")) as f:
+            table = json.load(f)
+        modes = [m["mode"] for m in table["modes"]["grouping-vs-processes"]]
+        self.assertEqual(modes, ["individual", "chain", "plan", "pool"])
+        for sc in table["scenarios"]:
+            if sc.get("family") != "grouping-vs-processes":
+                continue
+            keys = [sc["plugins"][n % len(sc["plugins"])] for n in range(sc["count"])]
+            cut = [k in sc["unshared"] for k in keys]
+            self.assertTrue(any(cut), "%s has nothing for the pins to refuse" % sc["id"])
+            single = [i for i, c in enumerate(cut) if not c and (i == 0 or cut[i - 1])
+                      and (i == len(cut) - 1 or cut[i + 1])]
+            self.assertGreaterEqual(len(single), 2, "%s leaves fewer than two shareable members alone" % sc["id"])
+
+
+class Pins(unittest.TestCase):
+    store = ("pin_set clap:/c#comp /c.clap=aa omx-layout/1:11\n"
+             "pin_set clap:/x#drive /x.clap=bb omx-layout/1:22 verdict=in-process\n"
+             "# a comment\n")
+
+    def test_each_uri_gets_its_verdict_and_the_rest_is_kept(self):
+        out = bench.set_verdicts(self.store, {"clap:/c#comp": "in-process", "clap:/x#drive": "isolated"})
+        self.assertEqual(out.splitlines(), ["pin_set clap:/c#comp /c.clap=aa omx-layout/1:11 verdict=in-process",
+                                            "pin_set clap:/x#drive /x.clap=bb omx-layout/1:22 verdict=isolated",
+                                            "# a comment"])
+
+    def test_a_uri_the_daemon_never_learned_is_failed(self):
+        with self.assertRaisesRegex(bench.Failed, "no pin for clap:/nobody#x"):
+            bench.set_verdicts(self.store, {"clap:/nobody#x": "in-process"})
+
+    def test_the_conf_of_a_plan_row_turns_pins_and_grouping_on_and_package_pins_off(self):
+        import types
+        with tempfile.TemporaryDirectory() as wd:
+            fake = types.SimpleNamespace(learn_pins=lambda uris: self.store)
+            conf = bench.Runner.plan_conf(fake, wd, ["clap:/c#comp", "clap:/x#drive"], {"clap:/x#drive"})
+            self.assertEqual(conf, ("require_pins 1", "pin_mode first-use", "pin_store " + os.path.join(wd, "pins"),
+                                    "pins_dir", "grouping on"))
+            with open(os.path.join(wd, "pins")) as f:
+                lines = f.read().splitlines()
+            self.assertIn("pin_set clap:/x#drive /x.clap=bb omx-layout/1:22 verdict=isolated", lines)
+            self.assertIn("pin_set clap:/c#comp /c.clap=aa omx-layout/1:11 verdict=in-process", lines)
+
+    def test_a_host_log_is_read_from_where_the_row_began(self):
+        with tempfile.TemporaryDirectory() as d:
+            with open(os.path.join(d, "H2.log"), "w") as f:
+                f.write("old: bad setting 'grouping'\n")
+            end = bench.log_size(d, "H2")
+            with open(os.path.join(d, "H2.log"), "a") as f:
+                f.write("new line\n")
+            self.assertEqual(bench.log_since(d, "H2", end), "new line\n")
+            self.assertEqual(bench.log_since(d, "none", 0), "")
+
+
 if __name__ == "__main__":
     unittest.main()

From 8818d37d31208019d8e9884254d2a46b3c6aa34d Mon Sep 17 00:00:00 2001
From: Pau Aliagas <linuxnow@gmail.com>
Date: Tue, 27 Oct 2026 06:48:11 +0100
Subject: [PATCH 6/7] Bench: table.sh renders the chains and grouping rows as
 markdown (#188)

The figures with their spread, the ratios of one process per plugin to one
chain, and the xruns summed per mode with and without stress, so the
grouped-vs-processes table is made from the documents and not typed.
---
 bench/lib/table.py        | 115 ++++++++++++++++++++++++++++++++++++++
 bench/table.sh            |   8 +++
 bench/tests/sabotage.sh   |   5 ++
 bench/tests/test_bench.py |  46 +++++++++++++++
 4 files changed, 174 insertions(+)
 create mode 100644 bench/lib/table.py
 create mode 100755 bench/table.sh

diff --git a/bench/lib/table.py b/bench/lib/table.py
new file mode 100644
index 0000000..f9f2dd8
--- /dev/null
+++ b/bench/lib/table.py
@@ -0,0 +1,115 @@
+"""Markdown tables from omx-bench/1 documents: one row per scenario row of the families that compare ways of running
+the same plugins (cvp and grp scenarios), the ratios of one process per plugin to one chain, and the xruns summed
+over the rows of each mode. A figure with a spread shows its median and [min, max]; a row that did not run is shown,
+with its status and its note."""
+
+import json
+import re
+import sys
+
+
+def label(row):
+    """the mode as the table names it: plugin-hostd's modes bare, the other hosts' with the host"""
+    mode = row.get("mode", "-")
+    return mode if row["host"] == "H2" else "%s %s" % (row["host"], mode)
+
+
+def figure(row, key, digits=None):
+    if row.get(key) is None:
+        return "-"
+    v = row[key]
+    text = ("%.*f" % (digits, v)) if digits is not None else str(v)
+    if "spread" in row and key in row["spread"]:
+        lo, hi = row["spread"][key]
+        if lo != hi:
+            text += " [%s-%s]" % (("%.*f" % (digits, lo)) if digits is not None else lo,
+                                  ("%.*f" % (digits, hi)) if digits is not None else hi)
+    return text
+
+
+def family_rows(doc):
+    """the rows of the scenarios that compare ways of running the same strip: cvp<N> and grp<N>"""
+    return [r for r in doc["scenarios"] if re.match(r"^(cvp|grp)[0-9]+$", r["id"]) and "mode" in r]
+
+
+def order(row):
+    modes = {"individual": 0, "chain": 1, "plan": 2, "pool": 3}
+    return (row["id"], row["rate"], row["frames"], row.get("stress", "none") != "none", row["host"] != "H2",
+            modes.get(row["mode"], 4), row["host"])
+
+
+def table(doc):
+    out = ["| scenario | rate/frames | N | stress | mode | procs | RSS MB | PSS MB | cpu_pct | ctx | DSP avg/max % | xruns |"
+           " latency frames | repeats |",
+           "|---|---|---|---|---|---|---|---|---|---|---|---|---|---|"]
+    for r in sorted(family_rows(doc), key=order):
+        head = "| %s | %d/%d | %d | %s | %s |" % (r["id"], r["rate"], r["frames"], len(r["plugins"]),
+                                                 r.get("stress", "none"), label(r))
+        if r["status"] != "ok":
+            out.append("%s %s: %s |" % (head, r["status"], r["note"]))
+            continue
+        dsp = "%s / %s" % (figure(r, "dsp_load_avg", 2), figure(r, "dsp_load_max", 2))
+        out.append("%s %s | %s | %s | %s | %s | %s | %s | %s | %s |" % (
+            head, figure(r, "processes"), figure(r, "rss_mb_total", 1), figure(r, "pss_mb_total", 1),
+            figure(r, "cpu_pct", 2), figure(r, "ctx_switches"), dsp, figure(r, "xruns"),
+            figure(r, "latency_frames"), r.get("repeats", 1)))
+    return out
+
+
+def ratios(doc):
+    """one process per plugin over one chain, plugin-hostd, no stress: how many times more each measure is"""
+    rows = {(r["id"], r["rate"], r["frames"], r["mode"]): r for r in family_rows(doc)
+            if r["host"] == "H2" and r["status"] == "ok" and r.get("stress", "none") == "none"}
+    out = ["| scenario | rate/frames | N | processes | RSS | PSS | cpu_pct | ctx | DSP avg |",
+           "|---|---|---|---|---|---|---|---|---|"]
+    for (sid, rate, frames, mode), ind in sorted(rows.items()):
+        grp = rows.get((sid, rate, frames, "chain"))
+        if mode != "individual" or not grp:
+            continue
+
+        def x(key):
+            return "%.2fx" % (ind[key] / grp[key]) if grp.get(key) else "-"
+        out.append("| %s | %d/%d | %d | %d vs %d | %s | %s | %s | %s | %s |" % (
+            sid, rate, frames, len(ind["plugins"]), ind["processes"], grp["processes"], x("rss_mb_total"),
+            x("pss_mb_total"), x("cpu_pct"), x("ctx_switches"), x("dsp_load_avg")))
+    return out
+
+
+def xrun_sums(doc):
+    """xruns summed over the rows of each mode, with and without stress, and how many rows that was"""
+    sums = {}
+    for r in family_rows(doc):
+        if r["status"] == "ok" and r.get("xruns") is not None:
+            key = (label(r), r.get("stress", "none") != "none")
+            n, total = sums.get(key, (0, 0))
+            sums[key] = (n + 1, total + r["xruns"])
+    out = ["| mode | stress | rows | xruns summed |", "|---|---|---|---|"]
+    for (mode, stressed), (n, total) in sorted(sums.items()):
+        out.append("| %s | %s | %d | %d |" % (mode, "yes" if stressed else "no", n, total))
+    return out
+
+
+def render(doc):
+    m = doc["machine"]
+    out = ["## %s: %s, %s, %d cores, governor %s" % (doc["id"], m["model"], m["cpu"], m["cores"], m["governor"]),
+           "", "Medians; [min-max] where the repeats differ.", ""]
+    out += table(doc) + ["", "### One process per plugin over one chain, no stress", ""] + ratios(doc)
+    out += ["", "### Xruns summed over the rows of each mode", ""] + xrun_sums(doc)
+    return "\n".join(out) + "\n"
+
+
+def main():
+    if len(sys.argv) < 2:
+        print("usage: table.sh <omx-bench/1 document>...", file=sys.stderr)
+        sys.exit(2)
+    for name in sys.argv[1:]:
+        with open(name) as f:
+            doc = json.load(f)
+        if doc.get("schema") != "omx-bench/1":
+            print("%s: not an omx-bench/1 document" % name, file=sys.stderr)
+            sys.exit(1)
+        print(render(doc))
+
+
+if __name__ == "__main__":
+    main()
diff --git a/bench/table.sh b/bench/table.sh
new file mode 100755
index 0000000..13ca543
--- /dev/null
+++ b/bench/table.sh
@@ -0,0 +1,8 @@
+#!/bin/bash
+# table.sh <file>...: the chains-vs-processes and grouping-vs-processes rows of omx-bench/1 documents as markdown:
+# the figures with their spread, the ratios of one process per plugin to one chain, the xruns summed per mode.
+set -euo pipefail
+
+here=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+command -v python3 >/dev/null || { echo "table.sh: python3 is needed" >&2; exit 2; }
+exec python3 -I "$here/lib/table.py" "$@"
diff --git a/bench/tests/sabotage.sh b/bench/tests/sabotage.sh
index 9aa0b92..6ca2f00 100755
--- a/bench/tests/sabotage.sh
+++ b/bench/tests/sabotage.sh
@@ -109,6 +109,11 @@ break_unit pins-missing lib/bench.py 'if missing:' 'if False:' Pins.test_a_uri_t
 break_unit conf-grouping lib/bench.py '"pins_dir", "grouping on")' '"pins_dir")' Pins.test_the_conf_of_a_plan_row_turns_pins_and_grouping_on_and_package_pins_off
 break_unit conf-verdict lib/bench.py '"isolated" if u in alone else "in-process"' '"in-process" if u in alone else "isolated"' Pins.test_the_conf_of_a_plan_row_turns_pins_and_grouping_on_and_package_pins_off
 break_unit scenario-strip scenarios.json '"count": 8, "unshared": ["drive"]' '"count": 8, "unshared": ["comp", "delay", "drive"]' Scenarios.test_the_grouping_scenarios_have_all_four_modes_and_a_strip_the_pins_cut
+break_unit table-ratio lib/table.py 'ind[key] / grp[key]' 'grp[key] / ind[key]' Table.test_the_ratios_of_individual_over_chain
+break_unit table-stress lib/table.py 'key = (label(r), r.get("stress", "none") != "none")' 'key = (label(r), False)' Table.test_xruns_are_summed_per_mode_with_and_without_stress
+break_unit table-spread lib/table.py 'if lo != hi:' 'if False:' Table.test_a_spread_shows_beside_its_median
+break_unit table-failed lib/table.py 'if r["status"] != "ok":' 'if False:' Table.test_a_failed_row_is_shown_with_its_reason
+break_unit table-family lib/table.py '(cvp|grp)[0-9]+$' '.*$' Table.test_a_scenario_that_is_not_a_family_is_left_out
 green "unit tests as they are" python3 -I "$here/tests/test_bench.py"
 
 # --- the teardown
diff --git a/bench/tests/test_bench.py b/bench/tests/test_bench.py
index 13dc861..945043b 100644
--- a/bench/tests/test_bench.py
+++ b/bench/tests/test_bench.py
@@ -589,6 +589,52 @@ def test_the_grouping_scenarios_have_all_four_modes_and_a_strip_the_pins_cut(sel
             self.assertGreaterEqual(len(single), 2, "%s leaves fewer than two shareable members alone" % sc["id"])
 
 
+table = load("bench_table", os.path.join(BENCH, "lib", "table.py"))
+
+
+def family_doc():
+    def row(mode, stress, procs, rss, pss, cpu, ctx, dsp, xr, **kw):
+        r = run_row(id="cvp8", plugins=["a"] * 8, mode=mode, stress=stress, processes=procs, rss_mb_total=rss,
+                    pss_mb_total=pss, cpu_pct=cpu, ctx_switches=ctx, dsp_load_avg=dsp, dsp_load_max=dsp + 1,
+                    xruns=xr, rss_mb=rss)
+        r.update(kw)
+        return r
+    doc = example()
+    doc["scenarios"] = [
+        row("individual", "none", 9, 880.0, 92.0, 6.0, 20000, 4.8, 0),
+        row("chain", "none", 2, 118.0, 36.0, 4.0, 2800, 3.9, 0),
+        row("individual", "busy4+mem256", 9, 881.0, 92.5, 6.5, 19800, 4.9, 10, repeats=3, spread={"xruns": [4, 18]}),
+        row("chain", "busy4+mem256", 2, 118.0, 36.0, 4.1, 2810, 3.9, 3),
+        row("plan", "none", 0, 0.0, 0.0, 0.0, 0, 0.0, 0, status="failed", note="the plan is one segment"),
+        run_row(id="idle", plugins=[], mode="individual")]
+    return doc
+
+
+class Table(unittest.TestCase):
+    def test_the_ratios_of_individual_over_chain(self):
+        lines = table.ratios(family_doc())
+        self.assertEqual(lines[2], "| cvp8 | 48000/256 | 8 | 9 vs 2 | 7.46x | 2.56x | 1.50x | 7.14x | 1.23x |")
+
+    def test_xruns_are_summed_per_mode_with_and_without_stress(self):
+        lines = table.xrun_sums(family_doc())
+        self.assertIn("| chain | no | 1 | 0 |", lines)
+        self.assertIn("| chain | yes | 1 | 3 |", lines)
+        self.assertIn("| individual | yes | 1 | 10 |", lines)
+
+    def test_a_spread_shows_beside_its_median(self):
+        lines = table.table(family_doc())
+        row = [x for x in lines if "individual | 9 | 881.0" in x][0]
+        self.assertIn("| 10 [4-18] |", row)
+        self.assertTrue(row.endswith("| 3 |"))
+
+    def test_a_failed_row_is_shown_with_its_reason(self):
+        lines = table.table(family_doc())
+        self.assertIn("| cvp8 | 48000/256 | 8 | none | plan | failed: the plan is one segment |", lines)
+
+    def test_a_scenario_that_is_not_a_family_is_left_out(self):
+        self.assertFalse([x for x in table.table(family_doc()) if "idle" in x])
+
+
 class Pins(unittest.TestCase):
     store = ("pin_set clap:/c#comp /c.clap=aa omx-layout/1:11\n"
              "pin_set clap:/x#drive /x.clap=bb omx-layout/1:22 verdict=in-process\n"

From 0c55eb151a16cfa4c691f3a9869bc60093ba5588 Mon Sep 17 00:00:00 2001
From: Pau Aliagas <linuxnow@gmail.com>
Date: Tue, 27 Oct 2026 06:49:47 +0100
Subject: [PATCH 7/7] Bench: document the new modes and fields, run the unit
 tests in CI (#188)

---
 .github/workflows/bench.yml |  7 +++++--
 bench/README.md             | 37 +++++++++++++++++++++++++++++++++++++
 2 files changed, 42 insertions(+), 2 deletions(-)

diff --git a/.github/workflows/bench.yml b/.github/workflows/bench.yml
index d6b1981..bd97eed 100644
--- a/.github/workflows/bench.yml
+++ b/.github/workflows/bench.yml
@@ -49,7 +49,10 @@ jobs:
           done
 
       - name: shellcheck
-        run: shellcheck bench/run.sh bench/validate.sh bench/lib/send.sh bench/tests/sabotage.sh
+        run: shellcheck bench/run.sh bench/validate.sh bench/table.sh bench/lib/send.sh bench/tests/sabotage.sh
+
+      - name: Unit tests of the harness
+        run: python3 -I bench/tests/test_bench.py
 
       - name: Every guard of the harness goes red when broken
         run: bench/tests/sabotage.sh --teardown
@@ -59,7 +62,7 @@ jobs:
           set -euo pipefail
           export XDG_RUNTIME_DIR=/tmp/xdg
           mkdir -p "$XDG_RUNTIME_DIR"
-          bench/run.sh --class ci --out smoke --rates 48000 --frames 256 --window 2 --scenarios lv2-1 --keep-governor
+          bench/run.sh --class ci --out smoke --rates 48000 --frames 256 --window 2 --repeats 1 --scenarios lv2-1 --keep-governor
           bench/validate.sh smoke/*.json
           python3 - smoke/*.json <<'EOF'
           import json, shutil, sys
diff --git a/bench/README.md b/bench/README.md
index 098dd06..4c3bc69 100644
--- a/bench/README.md
+++ b/bench/README.md
@@ -28,6 +28,8 @@ It needs the OpenMixer channel's `plugin-hostd`, `omx-clap-host`, `omx-plugins-c
 | `--scenarios a,b` | only these scenarios |
 | `--hosts H1,H2` | only these hosts |
 | `--window S` | seconds of measurement per row, 10 by default |
+| `--repeats N` | runs of each row, 3 by default; the document holds the median and the spread. 1 is a quick look; a document that is to say anything needs 3 or more |
+| `--stress none\|on\|both` | scenarios that declare stress run without it, with it, or both (the default) |
 | `--pianoteq-dir DIR` | the directory of Pianoteq's LV2 bundle; the `pianoteq` scenario runs only with it |
 | `--qualify` | also time `plugin-qualify --all --isolate` at the first rate and frame count |
 | `--keep-governor` | leave the CPU governor alone |
@@ -41,6 +43,9 @@ started is left, whatever happened, and exits 3 if one was.
 
 `bench/validate.sh <file>...` checks documents against the schema; a problem prints the file, the row and the rule,
 and the exit is 1. `bench/lib/send.sh <port> <line>...` sends protocol lines to a host by hand.
+`bench/table.sh <file>...` writes the chains and grouping rows as markdown: the figures with their spread, the ratios of
+one process per plugin to one chain, and the xruns summed per mode, with and without stress. A table of grouped against
+one process per plugin is made by running the `cvp3,cvp8,cvp16` scenarios on the machine and handing the document to it.
 
 Scenarios
 ---------
@@ -58,6 +63,7 @@ for an instrument, wired through the plugins in series into the playback ports.
 | `pianoteq` | Pianoteq 9 playing chords |
 | `kill1` | omx-comp, then `kill -9` of the process that holds it |
 | `cvp3`, `cvp8`, `cvp16` | chains vs processes: N plugins cycling omx-comp, Calf Compressor, omx-delay, omx-drive |
+| `grp8`, `grp16` | grouping vs processes: N CLAP plugins cycling omx-comp, omx-drive, omx-delay, omx-drive, omx-comp, where omx-drive is pinned `isolated` |
 
 H1 and H2 take our plugins as CLAP and Calf through the LV2 adapter. H3 and H4 take the LV2 builds of our plugins,
 and the row's note says so.
@@ -72,6 +78,22 @@ Chains vs processes runs the same N plugins in each of these modes, one row per
 | `shared-process` | H3 | one mod-host |
 | `individual` | H4 | N jalv processes |
 
+Grouping vs processes (`grp8`, `grp16`) runs one strip four ways under H2 only:
+
+| Mode | How plugin-hostd runs the strip | Processes |
+|------|---------------------------------|-----------|
+| `individual` | no chain declared, `require_pins 0` | the daemon and N workers |
+| `chain` | `chain 1 <ids>`, `require_pins 0`: the case where every plugin may share | the daemon and one worker |
+| `plan` | `grouping on`, `require_pins 1`, a verdict per plugin; `plan 1 <ids>` | the daemon and one worker per segment |
+| `pool` | the same plan, then the members it left alone that the pins let share leave for `pool:g1` | the daemon, the chain workers, the isolated ones, one pool worker |
+
+A row of `plan` or `pool` needs a plugin-hostd with the planner (plugin-hostd#185). Without it the daemon refuses the
+`grouping` setting and the row says `unsupported`, with that as its note. The runner starts a daemon once to let it learn
+each plugin's pin, then writes a verdict on each line of the daemon's own pin store: `isolated` for the plugins the
+scenario names under `unshared`, `in-process` for the rest, and starts the row's daemon on that store with the package
+pins off. A plan of one segment is a `failed` row (nothing was left for the pins to cut, so it is the `chain` row), and so
+is a pool that the daemon did not form from exactly the members the runner moved into it. One strip is one chain, so the
+pool holds that strip's leftovers; sharing a pool across chains is the planner's own later step.
 Reading the numbers
 -------------------
 
@@ -79,6 +101,10 @@ Reading the numbers
 |-------|---------|
 | `dsp_load_avg`, `dsp_load_max` | JACK's DSP load in percent, one `jack_cpu_load` reading a second over the window, the first dropped |
 | `samples` | how many readings |
+| `xruns` | JACK's xrun callbacks over the window, counted by a client of the run's private server (`libjack` through ctypes, no port and no process callback, so it adds no work to the cycle). If the server's own log names xruns in the window and the probe counted none, the row is `failed`: a deaf probe reads 0 on every row |
+| `latency_frames` | the capture latency JACK reports at `system:playback_1`, the longest of its range: what the clients on the path declared, not a measurement of the audio; a host that declares none contributes none |
+| `stress` | `none`, or `busy<cores>+mem<MiB>`: one busy loop pinned to each core the host may use, and a process that keeps touching 256 MiB, running beside the host for the whole window. The same name on the same machine is the same load. The same scenario with and without stress is two rows |
+| `repeats`, `spread` | a row run `repeats` times holds the median of each measure, a count taking the higher middle run so it is never understated, and `spread`: `{measure: [min, max]}`. A repeat that fails fails the row and its note says which; a row that is `unsupported` is not repeated |
 | `rss_mb`, `rss_mb_total` | resident memory summed over the host's processes at the end of the window |
 | `pss_mb_total` | proportional memory (`Pss` of `smaps_rollup`) summed the same way: shared libraries counted once across the processes |
 | `processes` | the host's processes: itself and every descendant, the workers of plugin-hostd included |
@@ -90,8 +116,19 @@ Reading the numbers
 
 Under H1, H3 and H4 the killed process is the host itself, so `kill1` ends `failed` with the note `host process lost`.
 
+`xruns`, `latency_frames`, `stress`, `repeats` and `spread`, and the modes `plan` and `pool`, are additions to
+`omx-bench/1`: a document without them is valid, and a reader that ignores them reads the document as before.
+
 `qualification` is written with `--qualify`: the CLAP plugins on the machine, the seconds `plugin-qualify` took, and how
 many it rated suitable.
 
 The DSP load of class `ci` comes from a container on a shared machine and tells nothing about a device; it is there to
 show the format. Compare rows of one document, or of documents of one class.
+
+Tests
+-----
+
+`python3 -I bench/tests/test_bench.py` needs no JACK server and no host: the schema and validator rules, the xrun probe
+against a stand-in `libjack`, the stress processes (pinned, resident) on the machine it runs on, the median and spread,
+the plan and pool steps against a stand-in daemon on a socket, the pin verdicts, and the tables. `bench/tests/sabotage.sh`
+breaks one line behind each of those guards in a copy and requires its test to go red; `--teardown` also runs the hosts.