diff --git a/.github/workflows/perf-gate.yml b/.github/workflows/perf-gate.yml new file mode 100644 index 00000000..39c01191 --- /dev/null +++ b/.github/workflows/perf-gate.yml @@ -0,0 +1,55 @@ +name: Perf gate + +# Hot-path instruction counts, PR against its merge base +# (benchmarks/hotpath.py gate; docs/TESTING.md, "Performance gate"). +# The workflow always runs so the check can be required; the gate itself is +# skipped (and passes) unless concurrency/src, rt/src or macros/src changed. + +on: + pull_request: + branches: ["main"] + merge_group: + +concurrency: + group: perf-gate-${{ github.event.pull_request.number || github.ref }} + cancel-in-progress: true + +env: + CARGO_TERM_COLOR: always + CARGO_INCREMENTAL: "0" + # Maximum growth of instructions per operation, in percent. + HOTPATH_THRESHOLD: "3" + +jobs: + perf-gate: + runs-on: ubuntu-latest + timeout-minutes: 40 + steps: + - uses: actions/checkout@v4 + with: + fetch-depth: 0 + - name: Base revision and hot-path changes + id: scope + env: + PR_BASE: ${{ github.event.pull_request.base.sha }} + MQ_BASE: ${{ github.event.merge_group.base_sha }} + run: | + base=$(git merge-base HEAD "${PR_BASE:-$MQ_BASE}") + echo "base=$base" >> "$GITHUB_OUTPUT" + if git diff --name-only "$base" HEAD | grep -qE '^(concurrency/src|rt/src|macros/src)/|^benchmarks/hotpath'; then + echo "run=true" >> "$GITHUB_OUTPUT" + else + echo "run=false" >> "$GITHUB_OUTPUT" + echo "Perf gate skipped: no change under concurrency/src, rt/src, macros/src or benchmarks/hotpath." >> "$GITHUB_STEP_SUMMARY" + fi + - uses: dtolnay/rust-toolchain@1.93.0 + if: steps.scope.outputs.run == 'true' + - name: Install valgrind + if: steps.scope.outputs.run == 'true' + run: sudo apt-get update -q && sudo apt-get install -y -q valgrind + - name: Instruction counts, head vs merge base + if: steps.scope.outputs.run == 'true' + run: > + python3 benchmarks/hotpath.py gate + --base ${{ steps.scope.outputs.base }} --head HEAD + --work ${{ runner.temp }}/hotpath diff --git a/CHANGELOG.md b/CHANGELOG.md index 22f39dfc..da7bd7f2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,14 @@ ## Unreleased +- CI: a hot-path performance gate (`perf-gate`). `benchmarks/hotpath.py gate` + counts instructions per operation with callgrind for async burst, sequential + and concurrent `request()` and `send` (on a current-thread runtime, so the + counts are deterministic) and threads-mode `request()`, and fails + a PR that grows any of them more than 3% over its merge base (it measures + only when `concurrency/src`, `rt/src` or `macros/src` changed). + `hotpath.py abba` is the wall-clock A/B on fixed cores, now a release + checklist step (docs/TESTING.md, "Performance gate"). - Design (0.9, approved for implementation after round 3): a unified start API (`docs/design/unified-start.md`, rank 5h). OTP's `start`, `start_link` and `start_monitor` wait for `started()`, and `spawn` does not. All four take one diff --git a/benchmarks/README.md b/benchmarks/README.md index e7eaf83e..c92e49e7 100644 --- a/benchmarks/README.md +++ b/benchmarks/README.md @@ -39,6 +39,22 @@ Runner self-tests (descendant cleanup and diagnostic retention): python3 -m unittest discover -s benchmarks -p test_runner.py ``` +## Hot-path regression checks + +`hotpath.py` compares a revision against its base on one shared driver +(`hotpath/driver.rs`: async burst, `request()` sequential and concurrent, +`send` with one and two producers, and the threads-mode equivalents): + +```sh +python3 benchmarks/hotpath.py gate # instruction counts (callgrind), CI +python3 benchmarks/hotpath.py abba --cores 12-15 --reps 30 # wall clock, before a release +``` + +`gate` is the `perf-gate` CI check: it fails when instructions per operation +grow more than 3% over the merge base. `abba` alternates base and head on the +same cores and reports the median and spread. See +[performance gate](../docs/TESTING.md#performance-gate). + ## Reliability campaign `run_reliability.py` drives isolated, resource-budgeted Linux campaigns diff --git a/benchmarks/hotpath.py b/benchmarks/hotpath.py new file mode 100755 index 00000000..c03a19a5 --- /dev/null +++ b/benchmarks/hotpath.py @@ -0,0 +1,217 @@ +#!/usr/bin/env python3 +"""Hot-path regression checks: a revision against its base. + + hotpath.py gate [--base REV] [--head REV] [--threshold PCT] + hotpath.py abba [--base REV] [--head REV] [--reps N] [--cores LIST] + +Both build `benchmarks/hotpath/driver.rs` (the head's copy, in both trees, so +both sides run the same workload) as a release example of each revision. + +`gate` counts instructions with callgrind, which is deterministic enough for +noisy CI runners. Each benchmark runs at two sizes; instructions per operation +is the difference divided by the extra operations, so process start, runtime +and actor setup cancel. It fails when any benchmark grows by more than +`--threshold` percent (default 3). It is the CI check. + +`abba` measures wall-clock throughput, alternating base and head on the same +cores (ABBA order), and reports the median and spread. Run it on a quiet +machine before a release; it is never run in CI. +""" +import argparse +import os +import re +import shutil +import statistics +import subprocess +import sys +import tempfile +from pathlib import Path + +ROOT = Path(__file__).resolve().parent.parent +DRIVER = ROOT / "benchmarks" / "hotpath" / "driver.rs" + + +def sh(cmd, **kw): + return subprocess.run(cmd, check=True, text=True, **kw) + + +def git(*args, cwd=ROOT): + return sh(["git", *args], cwd=cwd, capture_output=True).stdout.strip() + + +def default_base(head): + for ref in ("origin/main", "main"): + try: + return git("merge-base", head, ref) + except subprocess.CalledProcessError: + continue + sys.exit("no --base given and no main/origin/main to take a merge base with") + + +def build(work, label, rev): + """Check out `rev` in a worktree, add the driver, build it; return the binary.""" + tree = work / label + sh(["git", "worktree", "add", "--detach", "--force", str(tree), rev], cwd=ROOT, + capture_output=True) + shutil.copy(DRIVER, tree / "concurrency" / "examples" / "hotpath.rs") + target = work / f"target-{label}" # one per side: identical crate hashes would collide + env = dict(os.environ, CARGO_TARGET_DIR=str(target), CARGO_INCREMENTAL="0") + sh(["cargo", "build", "--release", "--locked", "-q", "-p", "spawned-concurrency", + "--example", "hotpath"], cwd=tree, env=env) + binary = work / f"hotpath-{label}" + shutil.copy(target / "release" / "examples" / "hotpath", binary) + return binary + + +def benchmarks(binary, only, gate=False): + """The driver's benchmarks; the gate takes only the deterministic ones.""" + names = sh([str(binary), "list-gate" if gate else "list"], capture_output=True).stdout.split() + return [n for n in names if not only or n in only] + + +IR = re.compile(r"Collected\s*:\s*(\d+)") + + +def instructions(binary, name, ops): + out = subprocess.run( + ["valgrind", "--tool=callgrind", "--fair-sched=yes", + "--callgrind-out-file=/dev/null", str(binary), name, str(ops)], + text=True, capture_output=True) + if out.returncode != 0: + sys.exit(f"{binary.name} {name} {ops} failed:\n{out.stderr[-3000:]}") + return int(IR.search(out.stderr).group(1)) + + +def per_op(binary, name, small, large, reps): + samples = [ + (instructions(binary, name, large) - instructions(binary, name, small)) / (large - small) + for _ in range(reps) + ] + return statistics.median(samples), samples + + +def report(lines): + text = "\n".join(lines) + "\n" + print(text) + summary = os.environ.get("GITHUB_STEP_SUMMARY") + if summary: + with open(summary, "a") as f: + f.write(text) + + +def gate(args, base_bin, head_bin, base, head): + names = benchmarks(head_bin, args.only, gate=True) + rows, failed = [], [] + for name in names: + b, bs = per_op(base_bin, name, args.small, args.large, args.reps) + h, hs = per_op(head_bin, name, args.small, args.large, args.reps) + delta = 100 * (h / b - 1) + bad = delta > args.threshold + if bad: + failed.append(name) + spread = max(max(bs) - min(bs), max(hs) - min(hs)) + rows.append(f"| `{name}` | {b:,.0f} | {h:,.0f} | {delta:+.1f}% | ±{spread:,.0f} | " + f"{'FAIL' if bad else 'ok'} |") + print(f"{name}: base {b:.0f} head {h:.0f} ({delta:+.1f}%)", file=sys.stderr) + verdict = (f"**FAIL**: {', '.join(failed)} grew more than {args.threshold}%" + if failed else f"**PASS**: no benchmark grew more than {args.threshold}%") + report([ + "## Hot-path instruction counts", + "", + f"Base `{base[:10]}`, head `{head[:10]}`. Instructions per operation " + f"(callgrind, sizes {args.small} and {args.large}, median of {args.reps}); " + f"threshold +{args.threshold}%.", + "", + "| benchmark | base Ir/op | head Ir/op | change | max spread | result |", + "|---|---:|---:|---:|---:|---|", + *rows, + "", + verdict, + ]) + return 1 if failed else 0 + + +def wallclock(binary, name, ops, cores): + cmd = [str(binary), name, str(ops)] + if cores: + cmd = ["taskset", "-c", cores, *cmd] + out = sh(cmd, capture_output=True).stdout.strip().splitlines()[-1] + return ops / float(out.split(",")[2]) + + +def abba(args, base_bin, head_bin, base, head): + names = benchmarks(head_bin, args.only) + data = {n: {"base": [], "head": []} for n in names} + csv = open(args.csv, "w") if args.csv else None + if csv: + csv.write("benchmark,side,rep,ops_per_s\n") + for rep in range(args.reps): + order = [("base", base_bin), ("head", head_bin)] + if rep % 2: + order.reverse() + for side, binary in order: + for name in names: + v = wallclock(binary, name, args.ops, args.cores) + data[name][side].append(v) + if csv: + csv.write(f"{name},{side},{rep},{v:.1f}\n") + rows = [] + for name in names: + b, h = data[name]["base"], data[name]["head"] + paired = statistics.median(hv / bv - 1 for bv, hv in zip(b, h)) + rows.append( + f"| `{name}` | {statistics.median(b)/1e3:,.0f}k ({min(b)/1e3:,.0f}k-{max(b)/1e3:,.0f}k) " + f"| {statistics.median(h)/1e3:,.0f}k ({min(h)/1e3:,.0f}k-{max(h)/1e3:,.0f}k) " + f"| {100*paired:+.1f}% |") + report([ + "## Hot-path wall clock (ABBA)", + "", + f"Base `{base[:10]}`, head `{head[:10]}`; {args.reps} interleaved repetitions of " + f"{args.ops} operations{', cores ' + args.cores if args.cores else ''}. " + "Operations per second: median (min-max); change is the median paired ratio.", + "", + "| benchmark | base | head | change |", + "|---|---:|---:|---:|", + *rows, + ]) + return 0 + + +def main(): + p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + p.add_argument("mode", choices=["gate", "abba"]) + p.add_argument("--head", default="HEAD") + p.add_argument("--base", help="default: merge base of --head and main") + p.add_argument("--only", nargs="*", help="benchmark names (default: all)") + p.add_argument("--work", help="build directory (default: a temporary one, removed)") + p.add_argument("--threshold", type=float, + default=float(os.environ.get("HOTPATH_THRESHOLD", "3")), + help="gate: max growth in percent (default 3, or $HOTPATH_THRESHOLD)") + p.add_argument("--small", type=int, default=2000, help="gate: smaller size") + p.add_argument("--large", type=int, default=6000, help="gate: larger size") + p.add_argument("--reps", type=int, help="gate: runs per size (default 3); abba: repetitions (default 20)") + p.add_argument("--ops", type=int, default=500_000, help="abba: operations per run") + p.add_argument("--cores", help="abba: taskset core list, e.g. 12-15") + p.add_argument("--csv", help="abba: raw samples") + args = p.parse_args() + args.reps = args.reps or (3 if args.mode == "gate" else 20) + + head = git("rev-parse", args.head) + base = git("rev-parse", args.base) if args.base else default_base(head) + work = Path(args.work or tempfile.mkdtemp(prefix="hotpath-")).resolve() + work.mkdir(parents=True, exist_ok=True) + try: + base_bin = build(work, "base", base) + head_bin = build(work, "head", head) + run = gate if args.mode == "gate" else abba + sys.exit(run(args, base_bin, head_bin, base, head)) + finally: + for label in ("base", "head"): + subprocess.run(["git", "worktree", "remove", "--force", str(work / label)], + cwd=ROOT, capture_output=True) + if not args.work: + shutil.rmtree(work, ignore_errors=True) + + +if __name__ == "__main__": + main() diff --git a/benchmarks/hotpath/driver.rs b/benchmarks/hotpath/driver.rs new file mode 100644 index 00000000..d446b7eb --- /dev/null +++ b/benchmarks/hotpath/driver.rs @@ -0,0 +1,249 @@ +//! Hot-path driver for `benchmarks/hotpath.py` (the CI instruction-count gate +//! and the manual wall-clock A/B). The script copies this file into +//! `concurrency/examples/` of both trees, so it must only use APIs that are +//! stable across releases: `#[actor]` handlers, `request()` and `send()`. +//! +//! Usage: hotpath (or `list`) +//! Prints `benchmark,ops,elapsed_s`. +use spawned_concurrency::actor; +use spawned_concurrency::message::Message; +use std::time::{Duration, Instant}; + +pub struct Q(pub u64); +impl Message for Q { + type Result = u64; +} +pub struct Hit(pub u64); +impl Message for Hit { + type Result = (); +} + +mod a { + use super::*; + use spawned_concurrency::tasks::{Actor, Context, Handler}; + pub struct Sink(pub u64); + #[actor] + impl Sink { + #[request_handler] + async fn q(&mut self, msg: Q, _ctx: &Context) -> u64 { + self.0 = self.0.wrapping_add(msg.0); + self.0 + } + #[send_handler] + async fn hit(&mut self, msg: Hit, _ctx: &Context) { + self.0 = self.0.wrapping_add(msg.0); + } + } +} + +mod t { + use super::*; + use spawned_concurrency::threads::{Actor, Context, Handler}; + pub struct Sink(pub u64); + #[actor] + impl Sink { + #[request_handler] + fn q(&mut self, msg: Q, _ctx: &Context) -> u64 { + self.0 = self.0.wrapping_add(msg.0); + self.0 + } + #[send_handler] + fn hit(&mut self, msg: Hit, _ctx: &Context) { + self.0 = self.0.wrapping_add(msg.0); + } + } +} + +/// (name, deterministic). The instruction-count gate runs only the +/// deterministic ones: everything on one thread (`async_ct_*`, a Tokio +/// current-thread runtime, identical counts run to run) or a lockstep +/// request/reply pair of threads (within 0.5%). Multi-threaded runtimes and +/// free-running OS-thread producers change how often actors park and wake +/// from run to run (up to ±20% instructions under callgrind), so those are +/// wall-clock only (`hotpath.py abba`). +const BENCHMARKS: &[(&str, bool)] = &[ + ("async_ct_burst", true), + ("async_ct_request_seq", true), + ("async_ct_request_conc16", true), + ("async_ct_send_p1", true), + ("async_ct_send_p2", true), + ("threads_request_seq", true), + ("threads_burst4", false), + ("threads_send_p1", false), + ("async_burst", false), + ("async_request_seq", false), + ("async_request_conc16", false), + ("async_rpc_thread", false), + ("async_send_p1", false), + ("async_send_p2", false), + ("threads_send_p2", false), +]; + +const WARMUP: u64 = 200; + +fn runtime(current_thread: bool) -> tokio::runtime::Runtime { + let mut builder = if current_thread { + tokio::runtime::Builder::new_current_thread() + } else { + let mut b = tokio::runtime::Builder::new_multi_thread(); + b.worker_threads(2); + b + }; + builder.enable_all().build().unwrap() +} + +/// Async actor benchmarks. Each returns the measured time; setup and warm-up +/// are outside it (the gate cancels them by differencing two sizes). +fn run_async(name: &str, n: u64) -> Duration { + use spawned_concurrency::tasks::ActorStart; + let current_thread = name.starts_with("async_ct_"); + let rt = runtime(current_thread); + let name = name.replacen("async_ct_", "async_", 1); + let name = name.as_str(); + let actor = rt.block_on(async { a::Sink(0).start() }); + let fence = |actor: &spawned_concurrency::tasks::ActorRef| { + rt.block_on(actor.request(Q(0))).unwrap(); + }; + for i in 0..WARMUP { + rt.block_on(actor.request(Q(i))).unwrap(); + } + let start = Instant::now(); + match name { + // Every request is enqueued before the first reply is awaited. + "async_burst" => rt.block_on(async { + let all = (0..n).map(|i| actor.request(Q(i))); + for r in futures::future::join_all(all).await { + std::hint::black_box(r.unwrap()); + } + }), + "async_request_seq" | "async_request_conc16" => { + let tasks = if name == "async_request_seq" { 1 } else { 16 }; + rt.block_on(async { + let joins: Vec<_> = (0..tasks) + .map(|_| { + let actor = actor.clone(); + tokio::spawn(async move { + for i in 0..n / tasks { + std::hint::black_box(actor.request(Q(i)).await.unwrap()); + } + }) + }) + .collect(); + for j in joins { + j.await.unwrap(); + } + }) + } + // A foreign OS thread blocking on each reply. + "async_rpc_thread" => { + let h = rt.handle().clone(); + let actor = actor.clone(); + std::thread::spawn(move || { + for i in 0..n { + std::hint::black_box(h.block_on(actor.request(Q(i))).unwrap()); + } + }) + .join() + .unwrap(); + } + "async_send_p1" | "async_send_p2" => { + let producers = if name == "async_send_p1" { 1 } else { 2 }; + if current_thread { + // Two producer tasks on the one runtime thread, interleaved + // in chunks; the actor runs when they yield. + rt.block_on(async { + let joins: Vec<_> = (0..producers) + .map(|_| { + let actor = actor.clone(); + tokio::spawn(async move { + for i in 0..n / producers { + actor.send(Hit(i)).unwrap(); + if i % 64 == 63 { + tokio::task::yield_now().await; + } + } + }) + }) + .collect(); + for j in joins { + j.await.unwrap(); + } + }); + fence(&actor); + return start.elapsed(); + } + std::thread::scope(|s| { + for _ in 0..producers { + let actor = actor.clone(); + s.spawn(move || { + for i in 0..n / producers { + actor.send(Hit(i)).unwrap(); + } + }); + } + }); + // FIFO per sender: this reply follows every message sent above. + fence(&actor); + } + _ => panic!("unknown benchmark {name}"), + } + start.elapsed() +} + +fn run_threads(name: &str, n: u64) -> Duration { + use spawned_concurrency::threads::ActorStart; + let actor = t::Sink(0).start(); + for i in 0..WARMUP { + actor.request(Q(i)).unwrap(); + } + let start = Instant::now(); + let (workers, send) = match name { + "threads_burst4" => (4, false), + "threads_request_seq" => (1, false), + "threads_send_p1" => (1, true), + "threads_send_p2" => (2, true), + _ => panic!("unknown benchmark {name}"), + }; + std::thread::scope(|s| { + for _ in 0..workers { + let actor = actor.clone(); + s.spawn(move || { + for i in 0..n / workers { + if send { + actor.send(Hit(i)).unwrap(); + } else { + std::hint::black_box(actor.request(Q(i)).unwrap()); + } + } + }); + } + }); + if send { + actor.request(Q(0)).unwrap(); + } + start.elapsed() +} + +fn main() { + let args: Vec = std::env::args().collect(); + // `list`: every benchmark; `list-gate`: the deterministic ones. + if let Some(list @ ("list" | "list-gate")) = args.get(1).map(String::as_str) { + for (b, deterministic) in BENCHMARKS { + if list == "list" || *deterministic { + println!("{b}"); + } + } + return; + } + let name = args[1].as_str(); + let n: u64 = args[2].parse().unwrap(); + assert!(BENCHMARKS.iter().any(|(b, _)| *b == name), "unknown benchmark {name}"); + let elapsed = if name.starts_with("async_") { + run_async(name, n) + } else { + run_threads(name, n) + }; + println!("{name},{n},{:.6}", elapsed.as_secs_f64()); + // Skip actor teardown: it is not part of any measurement. + std::process::exit(0); +} diff --git a/docs/TESTING.md b/docs/TESTING.md index 1d7fd39c..db5171d2 100644 --- a/docs/TESTING.md +++ b/docs/TESTING.md @@ -1815,14 +1815,72 @@ nice -n 19 taskset -c 26-29 target/release/deps/soak- \ Runs are by hand only: no schedule. Run a 4-hour soak before each release ([release checklist](#release-checklist)). +## Performance gate + +Robustness features must add no cost on the default path, and a performance +regression on the hot path is a release blocker. Two checks share one driver, +`benchmarks/hotpath/driver.rs`: async burst (every `request()` enqueued +before the first reply is awaited), sequential and 16-task `request()`, a +foreign thread blocking on each `request()`, `send` from one and two +producers, and the threads-mode burst (four requesters), sequential requests +and sends. `benchmarks/hotpath.py` copies the driver into both trees, so the +base and the head run the same workload; it uses only APIs that are stable +across releases (`#[actor]` handlers, `request()`, `send()`). + +**CI: instruction counts** (`perf-gate` workflow, every PR and the merge +queue). Wall-clock numbers on shared runners are too noisy, so the gate counts +instructions with callgrind. Each benchmark runs at two sizes (2,000 and 6,000 +operations, three runs each); instructions per operation is the difference +divided by the extra operations, so process start, runtime and actor setup +cancel. Only deterministic benchmarks are gated: the async ones on a Tokio +current-thread runtime (`async_ct_*`: burst, sequential and 16-task +`request()`, `send` from one and two producer tasks; identical counts run to +run) and threads-mode sequential `request()` (a lockstep pair of threads, +within 0.5%). On a multi-threaded runtime, or with free-running OS-thread +producers (threads-mode burst and `send`), how often actors park and wake +changes from run to run (up to ±20% instructions), so those run only in the +wall-clock check. The gate fails when any benchmark's instructions per operation grow +more than 3% over the merge base (`--threshold`, or `HOTPATH_THRESHOLD` in the +workflow). The job summary has the per-benchmark table. The workflow always +reports, so it can be a required check, but it only builds and measures when +`concurrency/src`, `rt/src`, `macros/src` or the driver changed. + +```sh +python3 benchmarks/hotpath.py gate # HEAD vs merge base with main +python3 benchmarks/hotpath.py gate --base origin/main --head origin/my-branch --threshold 3 +``` + +Instruction counts miss costs that are not instructions: lock contention, +extra wake-ups and parking, cache-line transfers. A change that moves work +between threads can pass the gate and still be slower, so the wall-clock +check below runs before each release. + +**Manual: wall clock, ABBA** (before each release, on a quiet machine). +`hotpath.py abba` alternates base and head on the same cores (base, head, +head, base, ...) and reports the median, the range and the median paired +change of operations per second: + +```sh +python3 benchmarks/hotpath.py abba --base origin/main --head \ + --cores 12-15 --reps 30 --csv /tmp/hotpath-abba.csv +``` + +Check `burst` and `rpc` in `benchmarks/run.py` (the baseline driver) the same +way when a change touches the reply path. The 0.8 gate measured async burst +37% and the default `request()` 24% slower than main this way. + ## Release checklist Before tagging a release (`gh release create` triggers `release.yaml`, which publishes to crates.io): -1. CI green on the release PR, including the `client-compat` gate. +1. CI green on the release PR, including the `client-compat` and `perf-gate` + checks. 2. A 4-hour [soak](#soak-harness) on fede passes on the release commit; link its `summary.txt` in the release PR. +3. The wall-clock [performance](#performance-gate) A/B against the previous + release on a quiet machine: no hot-path benchmark more than 5% slower. + Link the table in the release PR. ## CI hardening