diff --git a/bench/baseline.json b/bench/baseline.json index 9aaf494..4828462 100644 --- a/bench/baseline.json +++ b/bench/baseline.json @@ -1,16 +1,16 @@ { "_config": { - "N": 20000, - "crash_reps": 3, - "msg_n": 200000, + "N": 2000, + "crash_reps": 1, + "msg_n": 20000, "note": "refresh only with a commit that says why; tolerances come from tolerance_for() in the driver", - "wal_n": 4000 + "wal_n": 800 }, "durable.s1.mixread.ops_sec": { "dir": "higher", - "floor": 2302, + "floor": 2240, "tolerance_pct": 50, - "value": 9211 + "value": 8961 }, "durable.s1.mixread.p50us": { "dir": "lower", @@ -22,31 +22,31 @@ "dir": "lower", "floor": 100, "tolerance_pct": 50, - "value": 12 + "value": 4 }, "durable.s1.mixwrite.ops_sec": { "dir": "higher", - "floor": 255, + "floor": 248, "tolerance_pct": 50, - "value": 1023 + "value": 995 }, "durable.s1.mixwrite.p50us": { "dir": "lower", - "floor": 1720, + "floor": 824, "tolerance_pct": 50, - "value": 430 + "value": 206 }, "durable.s1.mixwrite.p99us": { "dir": "lower", - "floor": 2656, + "floor": 904, "tolerance_pct": 50, - "value": 664 + "value": 226 }, "durable.s1.query.ops_sec": { "dir": "higher", - "floor": 308641, + "floor": 233644, "tolerance_pct": 50, - "value": 1234567 + "value": 934579 }, "durable.s1.query.p50us": { "dir": "lower", @@ -58,13 +58,13 @@ "dir": "lower", "floor": 100, "tolerance_pct": 50, - "value": 1 + "value": 3 }, "durable.s1.read.ops_sec": { "dir": "higher", - "floor": 319284, + "floor": 190114, "tolerance_pct": 50, - "value": 1277139 + "value": 760456 }, "durable.s1.read.p50us": { "dir": "lower", @@ -76,85 +76,121 @@ "dir": "lower", "floor": 100, "tolerance_pct": 50, - "value": 1 + "value": 2 }, "durable.s1.seed.ops_sec": { "dir": "higher", - "floor": 1115, + "floor": 1080, "tolerance_pct": 15, - "value": 4460 + "value": 4323 }, "durable.s1.seed.p50us": { "dir": "lower", - "floor": 836, + "floor": 848, "tolerance_pct": 15, - "value": 209 + "value": 212 }, "durable.s1.seed.p99us": { "dir": "lower", - "floor": 2352, + "floor": 2108, "tolerance_pct": 15, - "value": 588 + "value": 527 + }, + "durable.s1.wmix.mean_batch": { + "dir": "higher", + "floor": 0.0, + "tolerance_pct": 100, + "value": 1.0 + }, + "durable.s1.wmix.ops_sec": { + "dir": "higher", + "floor": 810, + "tolerance_pct": 15, + "value": 3241 + }, + "durable.s1.wmix.p50us": { + "dir": "lower", + "floor": 852, + "tolerance_pct": 15, + "value": 213 + }, + "durable.s1.wmix.p99us": { + "dir": "lower", + "floor": 2032, + "tolerance_pct": 15, + "value": 508 + }, + "durable.s1.wmix.peak_batch": { + "dir": "higher", + "floor": 0, + "tolerance_pct": 100, + "value": 1 + }, + "durable.s1.wmix.peak_staged": { + "dir": "lower", + "floor": 196, + "tolerance_pct": 100, + "value": 49 }, "durable.s1.write.ops_sec": { "dir": "higher", - "floor": 581, + "floor": 1103, "tolerance_pct": 15, - "value": 2324 + "value": 4415 }, "durable.s1.write.p50us": { "dir": "lower", - "floor": 1764, + "floor": 856, "tolerance_pct": 15, - "value": 441 + "value": 214 }, "durable.s1.write.p99us": { "dir": "lower", - "floor": 2544, + "floor": 1932, "tolerance_pct": 15, - "value": 636 + "value": 483 }, "durable.sN.mixread.ops_sec": { "dir": "higher", - "floor": 1081, + "floor": 1111, "tolerance_pct": 50, - "value": 4324 + "value": 4445 }, "durable.sN.mixread.p50us": { "dir": "lower", - "floor": 248, + "floor": 348, "tolerance_pct": 50, - "value": 62 + "value": 87 }, "durable.sN.mixread.p99us": { "dir": "lower", - "floor": 18896, + "floor": 17540, "tolerance_pct": 50, - "value": 4724 + "value": 4385 }, "durable.sN.mixwrite.ops_sec": { "dir": "higher", - "floor": 120, + "floor": 123, "tolerance_pct": 50, - "value": 480 + "value": 493 }, "durable.sN.mixwrite.p50us": { "dir": "lower", - "floor": 2152, + "floor": 1460, "tolerance_pct": 50, - "value": 538 + "value": 365 }, "durable.sN.mixwrite.p99us": { "dir": "lower", - "floor": 23552, + "floor": 3888, "tolerance_pct": 50, - "value": 5888 + "value": 972 }, "durable.sN.query.ops_sec": { "dir": "higher", - "floor": 262329, + "floor": 211864, "tolerance_pct": 50, - "value": 1049317 + "value": 847457 }, "durable.sN.query.p50us": { "dir": "lower", @@ -170,9 +206,9 @@ }, "durable.sN.read.ops_sec": { "dir": "higher", - "floor": 313558, + "floor": 204415, "tolerance_pct": 50, - "value": 1254233 + "value": 817661 }, "durable.sN.read.p50us": { "dir": "lower", @@ -184,49 +220,85 @@ "dir": "lower", "floor": 100, "tolerance_pct": 50, - "value": 1 + "value": 2 }, "durable.sN.seed.ops_sec": { "dir": "higher", - "floor": 1116, + "floor": 1111, "tolerance_pct": 50, - "value": 4466 + "value": 4446 }, "durable.sN.seed.p50us": { "dir": "lower", - "floor": 840, + "floor": 848, "tolerance_pct": 50, - "value": 210 + "value": 212 }, "durable.sN.seed.p99us": { "dir": "lower", - "floor": 2536, + "floor": 1996, "tolerance_pct": 50, - "value": 634 + "value": 499 + }, + "durable.sN.wmix.mean_batch": { + "dir": "higher", + "floor": 0.0, + "tolerance_pct": 100, + "value": 3.38 + }, + "durable.sN.wmix.ops_sec": { + "dir": "higher", + "floor": 1575, + "tolerance_pct": 50, + "value": 6302 + }, + "durable.sN.wmix.p50us": { + "dir": "lower", + "floor": 13776, + "tolerance_pct": 50, + "value": 3444 + }, + "durable.sN.wmix.p99us": { + "dir": "lower", + "floor": 51212, + "tolerance_pct": 50, + "value": 12803 + }, + "durable.sN.wmix.peak_batch": { + "dir": "higher", + "floor": 7, + "tolerance_pct": 100, + "value": 28 + }, + "durable.sN.wmix.peak_staged": { + "dir": "lower", + "floor": 5488, + "tolerance_pct": 100, + "value": 1372 }, "durable.sN.write.ops_sec": { "dir": "higher", - "floor": 576, + "floor": 1117, "tolerance_pct": 50, - "value": 2304 + "value": 4470 }, "durable.sN.write.p50us": { "dir": "lower", - "floor": 1772, + "floor": 852, "tolerance_pct": 50, - "value": 443 + "value": 213 }, "durable.sN.write.p99us": { "dir": "lower", - "floor": 2716, + "floor": 2020, "tolerance_pct": 50, - "value": 679 + "value": 505 }, "ram.s1.mixread.ops_sec": { "dir": "higher", - "floor": 22384, + "floor": 2235, "tolerance_pct": 50, - "value": 89538 + "value": 8940 }, "ram.s1.mixread.p50us": { "dir": "lower", @@ -238,13 +310,13 @@ "dir": "lower", "floor": 100, "tolerance_pct": 50, - "value": 1 + "value": 3 }, "ram.s1.mixwrite.ops_sec": { "dir": "higher", - "floor": 2487, + "floor": 248, "tolerance_pct": 50, - "value": 9948 + "value": 993 }, "ram.s1.mixwrite.p50us": { "dir": "lower", @@ -256,19 +328,19 @@ "dir": "lower", "floor": 100, "tolerance_pct": 50, - "value": 2 + "value": 1 }, "ram.s1.msgrate.msgs_sec": { "dir": "higher", - "floor": 2087508, + "floor": 384911, "tolerance_pct": 15, - "value": 16700066 + "value": 3079291 }, "ram.s1.query.ops_sec": { "dir": "higher", - "floor": 247402, + "floor": 274725, "tolerance_pct": 50, - "value": 989609 + "value": 1098901 }, "ram.s1.query.p50us": { "dir": "lower", @@ -284,9 +356,9 @@ }, "ram.s1.read.ops_sec": { "dir": "higher", - "floor": 274393, + "floor": 312500, "tolerance_pct": 50, - "value": 1097574 + "value": 1250000 }, "ram.s1.read.p50us": { "dir": "lower", @@ -302,87 +374,87 @@ }, "ram.s1.seed.ops_sec": { "dir": "higher", - "floor": 61297, + "floor": 335570, "tolerance_pct": 15, - "value": 245188 + "value": 1342281 }, "ram.s1.seed.p50us": { "dir": "lower", "floor": 100, "tolerance_pct": 15, - "value": 4 + "value": 1 }, "ram.s1.seed.p99us": { "dir": "lower", "floor": 100, "tolerance_pct": 15, - "value": 9 + "value": 2 }, "ram.s1.write.ops_sec": { "dir": "higher", - "floor": 48866, + "floor": 237416, "tolerance_pct": 15, - "value": 195465 + "value": 949667 }, "ram.s1.write.p50us": { "dir": "lower", "floor": 100, "tolerance_pct": 15, - "value": 7 + "value": 1 }, "ram.s1.write.p99us": { "dir": "lower", "floor": 100, "tolerance_pct": 15, - "value": 12 + "value": 2 }, "ram.sN.mixread.ops_sec": { "dir": "higher", - "floor": 11229, + "floor": 2243, "tolerance_pct": 50, - "value": 44918 + "value": 8975 }, "ram.sN.mixread.p50us": { "dir": "lower", - "floor": 236, + "floor": 264, "tolerance_pct": 50, - "value": 59 + "value": 66 }, "ram.sN.mixread.p99us": { "dir": "lower", - "floor": 432, + "floor": 2080, "tolerance_pct": 50, - "value": 108 + "value": 520 }, "ram.sN.mixwrite.ops_sec": { "dir": "higher", - "floor": 1247, + "floor": 249, "tolerance_pct": 50, - "value": 4990 + "value": 997 }, "ram.sN.mixwrite.p50us": { "dir": "lower", - "floor": 256, + "floor": 288, "tolerance_pct": 50, - "value": 64 + "value": 72 }, "ram.sN.mixwrite.p99us": { "dir": "lower", - "floor": 516, + "floor": 344, "tolerance_pct": 50, - "value": 129 + "value": 86 }, "ram.sN.msgrate.msgs_sec": { "dir": "higher", - "floor": 355876, + "floor": 176056, "tolerance_pct": 50, - "value": 2847015 + "value": 1408450 }, "ram.sN.query.ops_sec": { "dir": "higher", - "floor": 307125, + "floor": 308641, "tolerance_pct": 50, - "value": 1228501 + "value": 1234567 }, "ram.sN.query.p50us": { "dir": "lower", @@ -398,9 +470,9 @@ }, "ram.sN.read.ops_sec": { "dir": "higher", - "floor": 340692, + "floor": 310945, "tolerance_pct": 50, - "value": 1362769 + "value": 1243781 }, "ram.sN.read.p50us": { "dir": "lower", @@ -416,38 +488,38 @@ }, "ram.sN.seed.ops_sec": { "dir": "higher", - "floor": 72890, + "floor": 363372, "tolerance_pct": 50, - "value": 291562 + "value": 1453488 }, "ram.sN.seed.p50us": { "dir": "lower", "floor": 100, "tolerance_pct": 50, - "value": 3 + "value": 1 }, "ram.sN.seed.p99us": { "dir": "lower", "floor": 100, "tolerance_pct": 50, - "value": 7 + "value": 2 }, "ram.sN.write.ops_sec": { "dir": "higher", - "floor": 60518, + "floor": 202922, "tolerance_pct": 50, - "value": 242072 + "value": 811688 }, "ram.sN.write.p50us": { "dir": "lower", "floor": 100, "tolerance_pct": 50, - "value": 6 + "value": 1 }, "ram.sN.write.p99us": { "dir": "lower", "floor": 100, "tolerance_pct": 50, - "value": 9 + "value": 4 } } \ No newline at end of file diff --git a/database/src/wal.c b/database/src/wal.c index 600f2b2..caf682e 100644 --- a/database/src/wal.c +++ b/database/src/wal.c @@ -393,7 +393,8 @@ int wo_wal_append_remove(wo_wal *w, uint32_t class_id, uint64_t id) { } int wo_wal_commit(wo_wal *w) { - if (!w->len) return 0; + if (!w->len) return 0; /* empty commits are not batches; do not count them */ + if (w->len > w->stat_peak_staged) w->stat_peak_staged = w->len; size_t at = 0; while (at < w->len) { ssize_t n = pwrite(w->fd, w->buf + at, w->len - at, (off_t)(w->off + at)); @@ -425,8 +426,16 @@ static void wal_die(const wo_wal *w, const char *op, uint32_t nrec) { void wo_wal_stage_fatal(const wo_wal *w) { wal_die(w, "staging a record", 1); } void wo_wal_commit_fatal(wo_wal *w, uint32_t nrec) { + int staged = w->len != 0; int rc = wo_wal_commit(w); - if (rc == 0) return; + if (rc == 0) { + if (staged) { /* count the barrier that actually happened */ + w->stat_batches++; + w->stat_records += nrec; + if (nrec > w->stat_peak_batch) w->stat_peak_batch = nrec; + } + return; + } wal_die(w, rc == WO_WAL_ERR_SYNC ? "fdatasync" : "pwrite", nrec); } diff --git a/database/src/wal.h b/database/src/wal.h index aeedc53..54b8120 100644 --- a/database/src/wal.h +++ b/database/src/wal.h @@ -55,6 +55,15 @@ typedef struct wo_wal { /* staged batch: appended by wal_append_*, flushed by wal_commit */ uint8_t *buf; size_t len, cap; + /* databasev2 4: group-commit diagnostics. Batching is worthless if + * batches are always one, and a throughput change would then have come + * from somewhere else — so the mechanism is measured, not assumed. + * peak_staged also settles whether the batch needs a cap with a number + * instead of a guess. Reported at exit under WO_WAL_STATS. */ + uint64_t stat_batches; /* non-empty commits */ + uint64_t stat_records; /* records those commits carried */ + uint64_t stat_peak_batch; /* most records in one barrier */ + uint64_t stat_peak_staged; /* most bytes staged behind one barrier */ } wo_wal; /* Open (create if missing) and preallocate [prealloc] bytes (best-effort; diff --git a/docs/examples/db-bench/main.wo b/docs/examples/db-bench/main.wo index bf8a6fe..87467b4 100644 --- a/docs/examples/db-bench/main.wo +++ b/docs/examples/db-bench/main.wo @@ -337,6 +337,91 @@ class Mixer { } } +-- databasev2 4 part A: every op a durable write, C at once. +-- +-- Why this leg exists. `mix` writes on one op in ten with C=4, so at most a +-- handful of writes are ever in flight and group commit has almost nothing to +-- batch: measured mean batch 1.01 over 3112 barriers, peak 3. That is a +-- property of the WORKLOAD, not of the mechanism, and without a write- +-- concurrent leg the iteration's payoff cannot be evaluated either way. +-- +-- Updates rather than inserts: comparable to what `mixwrite` measures, and the +-- row count stays flat so a long run does not turn into a growth test. +-- Histogram kind 2, because a replayed store still holds the seeding run's +-- kind-0/1 Hist rows and merging those would report someone else's latencies. +class WJob { + ops: Int + seed: Int + kmod: Int +} + +class WMixer { + id: Int + fn receive(msg: WJob) { + let hw: map = {}; + let s = msg.seed; + let i = 0; + while i < msg.ops { + s = lcg(s); + let key = s % msg.kmod; + let o0 = time.ticks(); + for r in from x in Item where x.k == key take 1 select x { + r.v = r.v + 1; + } + hist_add(hw, time.ticks() - o0); + i = i + 1; + } + hist_dump(hw, 2); + insert Meta { tag: "wmixdone${self.id}", val: msg.ops }; + } +} + +fn wmix_mode(total: Int, c: Int) -> Int { + let kmod = meta_val("kmod"); + if kmod < 1 { + print_err("wmix: seed first"); + return 1; + } + let per = total / c; + if per < 1 { + per = 1; + } + let wall0 = time.ticks(); + let i = 0; + while i < c { + let a: actor WJob = spawn WMixer { id: i }; + send(a, WJob { ops: per, seed: 4242 + i * 7919, kmod: kmod }); + i = i + 1; + } + let done = 0; + while done < c { + time.sleep(20); + done = 0; + i = 0; + while i < c { + if meta_val("wmixdone${i}") >= 0 { + done = done + 1; + } + i = i + 1; + } + } + let wall = time.ticks() - wall0; + let hw: map = {}; + let nw = 0; + for x in from x in Hist select x { + if x.kind == 2 { + if has(hw, x.b) { + set(hw, x.b, get(hw, x.b) + x.c); + } else { + set(hw, x.b, x.c); + } + nw = nw + x.c; + } + } + report("wmix", nw, wall, hw); + return 0; +} + fn mix_mode(total: Int, c: Int) -> Int { let kmod = meta_val("kmod"); if kmod < 1 { @@ -465,7 +550,7 @@ fn all_mode(n: Int) -> Int { fn usage() -> Int { print_err("usage: db-bench "); print_err(" all N | seed N | read N | query N | write N | wal N"); - print_err(" mix N C | msgrate N | verify | verify-acked M"); + print_err(" mix N C | wmix N C | msgrate N | verify | verify-acked M"); return 2; } @@ -508,6 +593,17 @@ fn main(args: multi Text) -> Int { if args[0] == "msgrate" { return msgrate_mode(n); } + if args[0] == "wmix" { + if len(args) < 3 { + return usage(); + } + let wc = parse_int(args[2]); + if wc == nil or wc < 1 { + print_err("db-bench: must be a positive number"); + return 2; + } + return wmix_mode(n, wc); + } if args[0] == "mix" { if len(args) < 3 { return usage(); diff --git a/runtime/src/main.c b/runtime/src/main.c index 01dd4e5..f8f6191 100644 --- a/runtime/src/main.c +++ b/runtime/src/main.c @@ -125,6 +125,16 @@ static void gc_pump(wo_vm *vm) { } } +/* databasev2 4: one diagnostic line about group commit, opt-in via + * WO_WAL_STATS. Off by default because it would otherwise pollute the output + * of every durable program; a gate that wants the numbers asks for them. */ +static void wal_stats_report(const wo_wal *w) { + if (!w || !getenv("WO_WAL_STATS")) return; + fprintf(stderr, "walstats batches=%llu records=%llu peak_batch=%llu peak_staged=%llu\n", + (unsigned long long)w->stat_batches, (unsigned long long)w->stat_records, + (unsigned long long)w->stat_peak_batch, (unsigned long long)w->stat_peak_staged); +} + int main(int argc, char **argv) { wo_module mod; char err[256]; @@ -245,7 +255,7 @@ int main(int argc, char **argv) { if (wo_engine_start(&mod, heap_mb << 20, nshards) != 0) { fprintf(stderr, "wovm: cannot start %u shards\n", nshards); wo_engine_stop(); - if (VM.rt.wal) wo_wal_close(&WAL); + if (VM.rt.wal) { wal_stats_report(&WAL); wo_wal_close(&WAL); } wo_db_destroy(&DB); wo_vm_destroy(&VM); wo_module_free(&mod); @@ -304,7 +314,7 @@ int main(int argc, char **argv) { * unwind. */ if (argv_val) wo_drop_kind(&VM.rt, WO_K_MULTI, argv_val); wo_engine_stop(); /* join + destroy the worker shards before the primary */ - if (VM.rt.wal) wo_wal_close(&WAL); + if (VM.rt.wal) { wal_stats_report(&WAL); wo_wal_close(&WAL); } wo_db_destroy(&DB); gc_pump(&VM); wo_vm_destroy(&VM); diff --git a/scripts/db-bench.py b/scripts/db-bench.py index 9d814cc..1c56dd5 100755 --- a/scripts/db-bench.py +++ b/scripts/db-bench.py @@ -26,6 +26,13 @@ QUICK = "--quick" in sys.argv WRITE_BASELINE = "--write-baseline" in sys.argv N = 2000 if QUICK else 20000 +# databasev2 4: the write-concurrent leg. `mix` writes on one op in ten with +# C=4, so group commit had almost nothing to batch there (measured mean batch +# 1.01, peak 3) — a property of that workload, not of the mechanism. C is high +# on purpose: batching is a function of how many writes are in flight, and +# measured mean batch rose 1.13 -> 1.76 -> 5.35 at C = 4 -> 16 -> 64. +WMIX_N = 4000 if QUICK else 20000 +WMIX_C = 32 if QUICK else 64 MSG_N = 20000 if QUICK else 200000 WAL_N = 800 if QUICK else 4000 CRASH_REPS = 1 if QUICK else 3 @@ -95,6 +102,55 @@ def parse_metrics(lines, into, prefix): if m: into[f"{prefix}.msgrate.msgs_sec"] = int(m.group(2)) +def wmix_leg(metrics, tag, env, data): + """Every op a durable write, WMIX_C at once — the leg that actually + exercises group commit. + + It reuses the store the `all` run just seeded (a fresh process replays it, + so `kmod` is there) and asks the runtime for its group-commit counters via + WO_WAL_STATS. The counters matter as much as the throughput: if batches are + always one the mechanism is inert and any throughput change came from + somewhere else, so a payoff would be attributed to the wrong cause.""" + e = dict(env); e["WO_WAL_STATS"] = "1" + rc, lines, _, _ = run(["wmix", str(WMIX_N), str(WMIX_C)], e, 1800) + if rc != 0: + bad(f"{tag}.wmix", f"rc={rc} tail={lines[-2:]}") + return + ops = p50 = p99 = None + batches = records = peak_batch = peak_staged = None + for l in lines: + f = l.split() + if f and f[0] == "wmix" and len(f) == 5: + ops, p50, p99 = int(f[2]), int(f[3]), int(f[4]) + elif f and f[0] == "walstats": + kv = dict(x.split("=", 1) for x in f[1:] if "=" in x) + batches = int(kv.get("batches", 0)); records = int(kv.get("records", 0)) + peak_batch = int(kv.get("peak_batch", 0)); peak_staged = int(kv.get("peak_staged", 0)) + if ops is None or batches is None: + bad(f"{tag}.wmix", "no report or no walstats line") + return + metrics[f"{tag}.wmix.ops_sec"] = ops + metrics[f"{tag}.wmix.p50us"] = p50 + metrics[f"{tag}.wmix.p99us"] = p99 + metrics[f"{tag}.wmix.peak_batch"] = peak_batch + metrics[f"{tag}.wmix.peak_staged"] = peak_staged + mean = round(records / batches, 2) if batches else 0 + metrics[f"{tag}.wmix.mean_batch"] = mean + ok(f"{tag}.wmix: {ops} ops/sec, p50 {p50}us p99 {p99}us; " + f"{records} records over {batches} barriers (mean {mean}, peak {peak_batch}), " + f"peak staged {peak_staged}B") + # The gate that matters. Only the MULTI-shard leg can batch: a worker's + # statements marshal to shard 0 and queue, while shard-0 statements run + # inline and commit one at a time by design (see db.c). + if tag.endswith(".sN"): + if mean > 1.0: + ok(f"{tag}.wmix batches form (mean {mean} > 1)") + else: + bad(f"{tag}.wmix-inert", + f"mean batch {mean} — group commit is not engaging, so a " + f"throughput change would not be attributable to it") + + def campaign(): metrics = {} ncores = os.cpu_count() or 1 @@ -123,6 +179,8 @@ def campaign(): bad(f"{tag}.mix.fds", f"grew {fdg}") else: ok(f"{tag}.mix.fds flat") + if flavor == "durable" and data: + wmix_leg(metrics, tag, env, data) if data: shutil.rmtree(data, ignore_errors=True) # msgrate once per shard count, RAM only (no store dependency) for shards in (1, ncores): @@ -225,6 +283,13 @@ def tolerance_for(key): mix*: scheduling-dependent small counts. read/query + all .sN.*: machine jitter, and at post-index-µs scale a 1µs histogram step on a 7µs p50 is already 14%.""" + # databasev2 4: batch SHAPE follows arrival timing, so gating it tightly + # would gate the scheduler — what must hold is that the mean exceeds one + # under contention, which wmix_leg asserts directly against the live run. + # wmix's throughput and latency are NOT waived: they are the payoff, and a + # blanket waiver here would have left the whole leg ungated. + if key.endswith((".wmix.mean_batch", ".wmix.peak_batch", ".wmix.peak_staged")): + return 100 if ".mixread." in key or ".mixwrite." in key: return 50 if ".sN." in key: return 50 if ".read." in key or ".query." in key: return 50 @@ -236,7 +301,7 @@ def write_baseline(metrics): "tolerances come from tolerance_for() in the driver"}} for k, v in sorted(metrics.items()): if k.endswith(("rss_growth_kb", "fd_growth")): continue - higher = k.endswith(("ops_sec", "msgs_sec")) + higher = k.endswith(("ops_sec", "msgs_sec", "mean_batch", "peak_batch")) floor_div = 8 if k.endswith("msgs_sec") else 4 # latency floors never sit below 100µs: at post-index µs scale a # 4×1µs "catastrophe line" is noise; the tripwire means "µs became