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 ...` checks documents against the schema; a problem prints the file, the row and the rule, and the exit is 1. `bench/lib/send.sh ...` sends protocol lines to a host by hand. +`bench/table.sh ...` 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 `, `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 ` | 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+mem`: 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. diff --git a/bench/lib/bench.py b/bench/lib/bench.py index 16ea017..0949fe7 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 @@ -26,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): @@ -222,6 +224,71 @@ 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 +# 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 @@ -256,6 +323,122 @@ 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 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: + 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 @@ -506,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 = {} @@ -524,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) @@ -553,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 : 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 @@ -632,6 +882,10 @@ 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.learned = {} + self.jack_log = None + self.jack_log_end = 0 self.port = 5700 # -- governor @@ -668,8 +922,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 @@ -696,13 +956,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): @@ -713,10 +977,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 @@ -737,8 +1003,22 @@ def row(self, s, hkey, mode, rate, frames): 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" @@ -763,6 +1043,13 @@ def row(self, s, hkey, mode, rate, frames): 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) @@ -782,12 +1069,19 @@ 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) 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 +1105,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"): @@ -828,6 +1123,44 @@ def row(self, s, hkey, mode, rate, frames): 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 = [] + 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 @@ -921,11 +1254,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.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["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: @@ -966,12 +1299,18 @@ 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("--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/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 and grp""" + 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 ...", 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/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/scenarios.json b/bench/scenarios.json index 025f2a6..c0dd76e 100644 --- a/bench/scenarios.json +++ b/bench/scenarios.json @@ -17,9 +17,15 @@ { "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" }, + { "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/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/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 ...: 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 872ed7d..6ca2f00 100755 --- a/bench/tests/sabotage.sh +++ b/bench/tests/sabotage.sh @@ -64,6 +64,58 @@ 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