From 48d378fb024b002f55c3390f009279081d077f54 Mon Sep 17 00:00:00 2001 From: William Barnhart Date: Tue, 21 Jul 2026 23:35:01 -0400 Subject: [PATCH 1/3] Add CI performance benchmark runner --- extra/tools/ci_benchmark.py | 453 ++++++++++++++++++++++++++++++++++++ 1 file changed, 453 insertions(+) create mode 100644 extra/tools/ci_benchmark.py diff --git a/extra/tools/ci_benchmark.py b/extra/tools/ci_benchmark.py new file mode 100644 index 000000000..28904a037 --- /dev/null +++ b/extra/tools/ci_benchmark.py @@ -0,0 +1,453 @@ +#!/usr/bin/env python3 +"""Run Faust CI performance benchmarks against current checkout and releases. + +The script intentionally keeps the benchmark workload self-contained and free of +Kafka/network I/O so it can run reliably on shared GitHub Actions runners. +""" +from __future__ import annotations + +import argparse +import json +import os +import pathlib +import shutil +import subprocess +import sys +import tempfile +import urllib.error +import urllib.request +from dataclasses import dataclass +from typing import Any, Iterable + +DEFAULT_REPO = "faust-streaming/faust" +DEFAULT_RELEASE_COUNT = 2 +DEFAULT_MAX_REGRESSION = 1.25 + +BENCHMARK_CODE = r''' +from __future__ import annotations + +import gc +import json +import platform +import statistics +import time +from importlib import metadata + +from faust.types.tuples import Message, TP, tp_set_to_map + + +def _version() -> str: + for distribution_name in ("faust-streaming", "faust"): + try: + return metadata.version(distribution_name) + except metadata.PackageNotFoundError: + continue + return "unknown" + + +def _time_per_iteration(fn, iterations: int, *, rounds: int = 9, warmups: int = 3): + for _ in range(warmups): + fn(iterations) + measurements = [] + for _ in range(rounds): + gc.collect() + started = time.perf_counter_ns() + fn(iterations) + elapsed = time.perf_counter_ns() - started + measurements.append(elapsed / iterations) + return { + "value": statistics.median(measurements), + "range": statistics.pstdev(measurements) if len(measurements) > 1 else 0.0, + } + + +def _message_create(iterations: int) -> None: + for i in range(iterations): + Message( + "orders", + i % 32, + i, + timestamp=1_700_000_000.0, + timestamp_type=1, + headers=None, + key=b"key", + value=b"value", + checksum=None, + ) + + +def _message_refcount(iterations: int) -> None: + message = Message( + "orders", + 0, + 0, + timestamp=1_700_000_000.0, + timestamp_type=1, + headers=None, + key=b"key", + value=b"value", + checksum=None, + ) + for _ in range(iterations): + message.incref() + message.decref() + + +_TPS = {TP(f"topic-{i % 64}", i % 48) for i in range(4096)} + + +def _tp_set_to_map(iterations: int) -> None: + for _ in range(iterations): + tp_set_to_map(_TPS) + + +benchmarks = [ + ("message_create", "ns/op", _message_create, 80_000), + ("message_refcount", "ns/op", _message_refcount, 300_000), + ("tp_set_to_map", "ns/call", _tp_set_to_map, 4_000), +] + +results = [] +for name, unit, fn, iterations in benchmarks: + measurement = _time_per_iteration(fn, iterations) + results.append( + { + "name": name, + "unit": unit, + "value": measurement["value"], + "range": measurement["range"], + } + ) + +print( + json.dumps( + { + "python": platform.python_version(), + "implementation": platform.python_implementation(), + "faust_version": _version(), + "results": results, + }, + sort_keys=True, + ) +) +''' + + +@dataclass(frozen=True) +class Target: + label: str + install_spec: str + commit: str + kind: str + + +def _run( + cmd: list[str], + *, + cwd: pathlib.Path | None = None, + env: dict[str, str] | None = None, +) -> str: + completed = subprocess.run( + cmd, + cwd=cwd, + env=env, + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + check=True, + ) + return completed.stdout.strip() + + +def _git(*args: str, cwd: pathlib.Path | None = None) -> str: + return _run(["git", *args], cwd=cwd) + + +def _release_tags_from_github(repo: str, count: int, token: str | None) -> list[str]: + request = urllib.request.Request( + f"https://api.github.com/repos/{repo}/releases?per_page={max(count * 2, count)}", + headers={ + "Accept": "application/vnd.github+json", + "User-Agent": "faust-ci-benchmark", + }, + ) + if token: + request.add_header("Authorization", f"Bearer {token}") + try: + with urllib.request.urlopen(request, timeout=20) as response: + releases = json.loads(response.read().decode()) + except (urllib.error.URLError, TimeoutError, json.JSONDecodeError) as exc: + print(f"::warning::Could not read GitHub releases for {repo}: {exc}", file=sys.stderr) + return [] + tags: list[str] = [] + for release in releases: + if release.get("draft") or release.get("prerelease"): + continue + tag = release.get("tag_name") + if isinstance(tag, str) and tag: + tags.append(tag) + if len(tags) >= count: + break + return tags + + +def _release_tags_from_git(count: int, cwd: pathlib.Path) -> list[str]: + try: + tags = _git("tag", "--sort=-creatordate", cwd=cwd).splitlines() + except subprocess.CalledProcessError as exc: + print(f"::warning::Could not read local git tags: {exc.stdout}", file=sys.stderr) + return [] + return [tag for tag in tags if tag][:count] + + +def _discover_release_tags(repo: str, count: int, cwd: pathlib.Path) -> list[str]: + token = os.environ.get("GITHUB_TOKEN") or os.environ.get("GH_TOKEN") + tags = _release_tags_from_github(repo, count, token) + if len(tags) < count: + for tag in _release_tags_from_git(count, cwd): + if tag not in tags: + tags.append(tag) + if len(tags) >= count: + break + return tags[:count] + + +def _commit_for_ref(ref: str, cwd: pathlib.Path) -> str: + try: + return _git("rev-list", "-n", "1", ref, cwd=cwd) + except subprocess.CalledProcessError: + return ref + + +def _safe_dir_name(label: str) -> str: + return "".join(ch if ch.isalnum() or ch in "._-" else "-" for ch in label) + + +def _create_venv(path: pathlib.Path) -> pathlib.Path: + subprocess.check_call([sys.executable, "-m", "venv", str(path)]) + if os.name == "nt": + return path / "Scripts" / "python.exe" + return path / "bin" / "python" + + +def _install_target(python: pathlib.Path, install_spec: str, cwd: pathlib.Path) -> None: + base_env = os.environ.copy() + base_env.update( + { + "NO_CYTHON": "1", + "PIP_DISABLE_PIP_VERSION_CHECK": "1", + "PYTHONUNBUFFERED": "1", + } + ) + subprocess.check_call( + [ + str(python), + "-m", + "pip", + "install", + "--upgrade", + "pip", + "setuptools", + "wheel", + "setuptools_scm[toml]", + "Cython>=3.0.0", + ], + cwd=cwd, + env=base_env, + ) + subprocess.check_call( + [str(python), "-m", "pip", "install", "--no-deps", "--no-build-isolation", install_spec], + cwd=cwd, + env=base_env, + ) + + +def _run_target(target: Target, workdir: pathlib.Path, source_root: pathlib.Path) -> dict[str, Any]: + venv_dir = workdir / _safe_dir_name(target.label) + python = _create_venv(venv_dir) + _install_target(python, target.install_spec, source_root) + env = os.environ.copy() + env.update({"PYTHONHASHSEED": "0", "PYTHONUNBUFFERED": "1"}) + output = _run([str(python), "-c", BENCHMARK_CODE], env=env) + payload = json.loads(output.splitlines()[-1]) + payload.update({"target": target.label, "commit": target.commit, "kind": target.kind}) + return payload + + +def _round(value: float) -> float: + return round(float(value), 3) + + +def _benchmark_action_rows( + payloads: Iterable[dict[str, Any]], + *, + current_only: bool, +) -> list[dict[str, Any]]: + rows: list[dict[str, Any]] = [] + for payload in payloads: + if current_only and payload["target"] != "current": + continue + for result in payload["results"]: + metric_name = result["name"] if current_only else f"{payload['target']}/{result['name']}" + rows.append( + { + "name": metric_name, + "unit": result["unit"], + "value": _round(result["value"]), + "range": _round(result.get("range", 0.0)), + "extra": "\n".join( + [ + f"target: {payload['target']}", + f"kind: {payload['kind']}", + f"commit: {payload['commit']}", + f"faust: {payload['faust_version']}", + f"python: {payload['implementation']} {payload['python']}", + ] + ), + } + ) + return rows + + +def _result_map(payload: dict[str, Any]) -> dict[str, dict[str, Any]]: + return {result["name"]: result for result in payload["results"]} + + +def _format_value(result: dict[str, Any]) -> str: + return f"{_round(result['value'])} {result['unit']}" + + +def _write_summary( + path: pathlib.Path, + payloads: list[dict[str, Any]], + release_tags: list[str], + failures: list[str], + max_regression: float, +) -> None: + current = next(payload for payload in payloads if payload["target"] == "current") + current_results = _result_map(current) + lines = [ + "# Faust performance benchmark", + "", + f"Current commit: `{current['commit']}`", + f"Release baselines: {', '.join(f'`{tag}`' for tag in release_tags) or '_none found_'}", + f"Regression threshold: `{max_regression:.2f}x` slower than a release baseline", + "", + "| Benchmark | Current | Baseline | Delta | Status |", + "|---|---:|---:|---:|---|", + ] + for payload in payloads: + if payload["target"] == "current": + continue + baseline_results = _result_map(payload) + for name, current_result in current_results.items(): + baseline_result = baseline_results[name] + ratio = current_result["value"] / baseline_result["value"] + delta = (ratio - 1.0) * 100.0 + status = "❌ slower" if ratio > max_regression else "✅ ok" + lines.append( + "| {name} | {current_value} | {baseline_value} (`{target}`) | {delta:+.1f}% | {status} |".format( + name=name, + current_value=_format_value(current_result), + baseline_value=_format_value(baseline_result), + target=payload["target"], + delta=delta, + status=status, + ) + ) + lines.extend(["", "## Raw current results", "", "```json"]) + lines.append(json.dumps(_benchmark_action_rows([current], current_only=True), indent=2, sort_keys=True)) + lines.append("```") + if failures: + lines.extend(["", "## Regressions", ""]) + lines.extend(f"- {failure}" for failure in failures) + path.write_text("\n".join(lines) + "\n") + + +def _find_regressions(payloads: list[dict[str, Any]], max_regression: float) -> list[str]: + current = next(payload for payload in payloads if payload["target"] == "current") + current_results = _result_map(current) + failures: list[str] = [] + for payload in payloads: + if payload["target"] == "current": + continue + baseline_results = _result_map(payload) + for name, current_result in current_results.items(): + baseline_result = baseline_results[name] + ratio = current_result["value"] / baseline_result["value"] + if ratio > max_regression: + failures.append( + f"{name} is {ratio:.2f}x slower than {payload['target']} " + f"({_format_value(current_result)} vs {_format_value(baseline_result)})" + ) + return failures + + +def _float_env(name: str, default: float) -> float: + value = os.environ.get(name) + return float(value) if value else default + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--repo", default=os.environ.get("GITHUB_REPOSITORY", DEFAULT_REPO)) + parser.add_argument("--release-count", type=int, default=DEFAULT_RELEASE_COUNT) + parser.add_argument("--max-regression", type=float, default=_float_env("PERF_MAX_REGRESSION", DEFAULT_MAX_REGRESSION)) + parser.add_argument("--output-json", type=pathlib.Path, default=pathlib.Path("benchmark-results.json")) + parser.add_argument("--current-output-json", type=pathlib.Path, default=pathlib.Path("benchmark-current.json")) + parser.add_argument("--summary", type=pathlib.Path, default=pathlib.Path("benchmark-summary.md")) + parser.add_argument("--keep-workdir", action="store_true") + return parser.parse_args() + + +def main() -> int: + args = parse_args() + source_root = pathlib.Path.cwd() + release_tags = _discover_release_tags(args.repo, args.release_count, source_root) + current_commit = _commit_for_ref("HEAD", source_root) + targets = [Target("current", ".", current_commit, "checkout")] + targets.extend( + Target( + tag, + f"git+https://github.com/{args.repo}.git@{tag}", + _commit_for_ref(tag, source_root), + "release", + ) + for tag in release_tags + ) + + tempdir = pathlib.Path(tempfile.mkdtemp(prefix="faust-benchmark-")) + payloads: list[dict[str, Any]] = [] + try: + for target in targets: + print(f"::group::Benchmark {target.label}") + payload = _run_target(target, tempdir, source_root) + payloads.append(payload) + for result in payload["results"]: + print(f"{target.label}/{result['name']}: {_format_value(result)}") + print("::endgroup::") + finally: + if args.keep_workdir: + print(f"Keeping benchmark workdir: {tempdir}") + else: + shutil.rmtree(tempdir, ignore_errors=True) + + failures = _find_regressions(payloads, args.max_regression) if release_tags else [] + args.output_json.write_text( + json.dumps(_benchmark_action_rows(payloads, current_only=False), indent=2, sort_keys=True) + "\n" + ) + args.current_output_json.write_text( + json.dumps(_benchmark_action_rows(payloads, current_only=True), indent=2, sort_keys=True) + "\n" + ) + _write_summary(args.summary, payloads, release_tags, failures, args.max_regression) + + if failures: + print("::error::Performance regression threshold exceeded") + for failure in failures: + print(f"::error::{failure}") + return 1 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) From 7e6d35a837058f4f0eba40ce81c785dc569a9495 Mon Sep 17 00:00:00 2001 From: William Barnhart Date: Tue, 21 Jul 2026 23:35:11 -0400 Subject: [PATCH 2/3] Add performance benchmark GitHub Action --- .github/workflows/performance-benchmark.yml | 91 +++++++++++++++++++++ 1 file changed, 91 insertions(+) create mode 100644 .github/workflows/performance-benchmark.yml diff --git a/.github/workflows/performance-benchmark.yml b/.github/workflows/performance-benchmark.yml new file mode 100644 index 000000000..061b1789f --- /dev/null +++ b/.github/workflows/performance-benchmark.yml @@ -0,0 +1,91 @@ +name: Performance Benchmarks + +on: + pull_request: + branches: [master] + push: + branches: [master] + workflow_dispatch: + inputs: + max_regression: + description: Maximum allowed slowdown versus a release baseline (for example, 1.25 = 25%) + required: false + default: "1.25" + schedule: + - cron: "23 4 * * 1" + +concurrency: + group: performance-benchmark-${{ github.ref }} + cancel-in-progress: true + +permissions: + contents: write + pull-requests: write + +jobs: + benchmark: + name: Compare current branch to recent releases + runs-on: ubuntu-latest + timeout-minutes: 45 + + steps: + - name: Check out repository + uses: actions/checkout@v4 + with: + fetch-depth: 0 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: "3.11" + + - name: Run release comparison benchmarks + env: + GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} + GITHUB_REPOSITORY: ${{ github.repository }} + PERF_MAX_REGRESSION: ${{ inputs.max_regression }} + run: | + python -m pip install --upgrade pip + MAX_REGRESSION="${PERF_MAX_REGRESSION:-1.25}" + python extra/tools/ci_benchmark.py \ + --repo "$GITHUB_REPOSITORY" \ + --release-count 2 \ + --max-regression "$MAX_REGRESSION" \ + --output-json benchmark-results.json \ + --current-output-json benchmark-current.json \ + --summary benchmark-summary.md + + - name: Add benchmark summary + if: always() + run: | + if [ -f benchmark-summary.md ]; then + cat benchmark-summary.md >> "$GITHUB_STEP_SUMMARY" + else + echo "Benchmark did not produce a summary." >> "$GITHUB_STEP_SUMMARY" + fi + + - name: Upload benchmark artifacts + if: always() + uses: actions/upload-artifact@v4 + with: + name: performance-benchmark-results + path: | + benchmark-results.json + benchmark-current.json + benchmark-summary.md + if-no-files-found: ignore + + - name: Store benchmark history + if: github.event_name == 'push' && github.ref == 'refs/heads/master' + uses: benchmark-action/github-action-benchmark@v1 + with: + name: Faust core benchmarks + tool: customSmallerIsBetter + output-file-path: benchmark-current.json + github-token: ${{ secrets.GITHUB_TOKEN }} + auto-push: true + gh-pages-branch: gh-pages + benchmark-data-dir-path: dev/bench + alert-threshold: "125%" + comment-on-alert: true + fail-on-alert: false From 9232f25a6bb4bbc270ef3e25b5c3ca662aa37063 Mon Sep 17 00:00:00 2001 From: William Barnhart Date: Tue, 21 Jul 2026 23:48:53 -0400 Subject: [PATCH 3/3] Fix benchmark environment installation --- extra/tools/ci_benchmark.py | 385 +++++++++++++----------------------- 1 file changed, 135 insertions(+), 250 deletions(-) diff --git a/extra/tools/ci_benchmark.py b/extra/tools/ci_benchmark.py index 28904a037..fa19cf594 100644 --- a/extra/tools/ci_benchmark.py +++ b/extra/tools/ci_benchmark.py @@ -1,9 +1,5 @@ #!/usr/bin/env python3 -"""Run Faust CI performance benchmarks against current checkout and releases. - -The script intentionally keeps the benchmark workload self-contained and free of -Kafka/network I/O so it can run reliably on shared GitHub Actions runners. -""" +"""Benchmark the current Faust checkout against recent stable releases.""" from __future__ import annotations import argparse @@ -11,6 +7,7 @@ import os import pathlib import shutil +import statistics import subprocess import sys import tempfile @@ -36,37 +33,31 @@ from faust.types.tuples import Message, TP, tp_set_to_map -def _version() -> str: - for distribution_name in ("faust-streaming", "faust"): +def version(): + for name in ("faust-streaming", "faust"): try: - return metadata.version(distribution_name) + return metadata.version(name) except metadata.PackageNotFoundError: - continue + pass return "unknown" -def _time_per_iteration(fn, iterations: int, *, rounds: int = 9, warmups: int = 3): +def measure(fn, iterations, rounds=9, warmups=3): for _ in range(warmups): fn(iterations) - measurements = [] + samples = [] for _ in range(rounds): gc.collect() started = time.perf_counter_ns() fn(iterations) - elapsed = time.perf_counter_ns() - started - measurements.append(elapsed / iterations) - return { - "value": statistics.median(measurements), - "range": statistics.pstdev(measurements) if len(measurements) > 1 else 0.0, - } + samples.append((time.perf_counter_ns() - started) / iterations) + return statistics.median(samples), statistics.pstdev(samples) -def _message_create(iterations: int) -> None: +def message_create(iterations): for i in range(iterations): Message( - "orders", - i % 32, - i, + "orders", i % 32, i, timestamp=1_700_000_000.0, timestamp_type=1, headers=None, @@ -76,11 +67,9 @@ def _message_create(iterations: int) -> None: ) -def _message_refcount(iterations: int) -> None: +def message_refcount(iterations): message = Message( - "orders", - 0, - 0, + "orders", 0, 0, timestamp=1_700_000_000.0, timestamp_type=1, headers=None, @@ -93,43 +82,30 @@ def _message_refcount(iterations: int) -> None: message.decref() -_TPS = {TP(f"topic-{i % 64}", i % 48) for i in range(4096)} +TPS = {TP(f"topic-{i % 64}", i % 48) for i in range(4096)} -def _tp_set_to_map(iterations: int) -> None: +def group_topic_partitions(iterations): for _ in range(iterations): - tp_set_to_map(_TPS) + tp_set_to_map(TPS) benchmarks = [ - ("message_create", "ns/op", _message_create, 80_000), - ("message_refcount", "ns/op", _message_refcount, 300_000), - ("tp_set_to_map", "ns/call", _tp_set_to_map, 4_000), + ("message_create", "ns/op", message_create, 80_000), + ("message_refcount", "ns/op", message_refcount, 300_000), + ("tp_set_to_map", "ns/call", group_topic_partitions, 4_000), ] - results = [] -for name, unit, fn, iterations in benchmarks: - measurement = _time_per_iteration(fn, iterations) - results.append( - { - "name": name, - "unit": unit, - "value": measurement["value"], - "range": measurement["range"], - } - ) - -print( - json.dumps( - { - "python": platform.python_version(), - "implementation": platform.python_implementation(), - "faust_version": _version(), - "results": results, - }, - sort_keys=True, - ) -) +for name, unit, function, iterations in benchmarks: + value, spread = measure(function, iterations) + results.append({"name": name, "unit": unit, "value": value, "range": spread}) + +print(json.dumps({ + "python": platform.python_version(), + "implementation": platform.python_implementation(), + "faust_version": version(), + "results": results, +}, sort_keys=True)) ''' @@ -141,12 +117,7 @@ class Target: kind: str -def _run( - cmd: list[str], - *, - cwd: pathlib.Path | None = None, - env: dict[str, str] | None = None, -) -> str: +def run(cmd: list[str], *, cwd: pathlib.Path | None = None, env: dict[str, str] | None = None) -> str: completed = subprocess.run( cmd, cwd=cwd, @@ -154,245 +125,176 @@ def _run( text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, - check=True, ) + if completed.returncode: + print(completed.stdout, file=sys.stderr) + raise subprocess.CalledProcessError(completed.returncode, cmd, output=completed.stdout) return completed.stdout.strip() -def _git(*args: str, cwd: pathlib.Path | None = None) -> str: - return _run(["git", *args], cwd=cwd) +def git(*args: str, cwd: pathlib.Path) -> str: + return run(["git", *args], cwd=cwd) -def _release_tags_from_github(repo: str, count: int, token: str | None) -> list[str]: +def github_release_tags(repo: str, count: int) -> list[str]: request = urllib.request.Request( - f"https://api.github.com/repos/{repo}/releases?per_page={max(count * 2, count)}", - headers={ - "Accept": "application/vnd.github+json", - "User-Agent": "faust-ci-benchmark", - }, + f"https://api.github.com/repos/{repo}/releases?per_page={max(count * 3, count)}", + headers={"Accept": "application/vnd.github+json", "User-Agent": "faust-ci-benchmark"}, ) + token = os.environ.get("GITHUB_TOKEN") or os.environ.get("GH_TOKEN") if token: request.add_header("Authorization", f"Bearer {token}") try: with urllib.request.urlopen(request, timeout=20) as response: releases = json.loads(response.read().decode()) except (urllib.error.URLError, TimeoutError, json.JSONDecodeError) as exc: - print(f"::warning::Could not read GitHub releases for {repo}: {exc}", file=sys.stderr) + print(f"::warning::Could not query GitHub releases: {exc}", file=sys.stderr) return [] - tags: list[str] = [] + tags = [] for release in releases: if release.get("draft") or release.get("prerelease"): continue tag = release.get("tag_name") - if isinstance(tag, str) and tag: + if tag: tags.append(tag) - if len(tags) >= count: + if len(tags) == count: break return tags -def _release_tags_from_git(count: int, cwd: pathlib.Path) -> list[str]: - try: - tags = _git("tag", "--sort=-creatordate", cwd=cwd).splitlines() - except subprocess.CalledProcessError as exc: - print(f"::warning::Could not read local git tags: {exc.stdout}", file=sys.stderr) - return [] - return [tag for tag in tags if tag][:count] - - -def _discover_release_tags(repo: str, count: int, cwd: pathlib.Path) -> list[str]: - token = os.environ.get("GITHUB_TOKEN") or os.environ.get("GH_TOKEN") - tags = _release_tags_from_github(repo, count, token) +def discover_release_tags(repo: str, count: int, cwd: pathlib.Path) -> list[str]: + tags = github_release_tags(repo, count) if len(tags) < count: - for tag in _release_tags_from_git(count, cwd): - if tag not in tags: + try: + local_tags = git("tag", "--sort=-creatordate", cwd=cwd).splitlines() + except subprocess.CalledProcessError: + local_tags = [] + for tag in local_tags: + if tag and tag not in tags: tags.append(tag) - if len(tags) >= count: + if len(tags) == count: break return tags[:count] -def _commit_for_ref(ref: str, cwd: pathlib.Path) -> str: +def commit_for_ref(ref: str, cwd: pathlib.Path) -> str: try: - return _git("rev-list", "-n", "1", ref, cwd=cwd) + return git("rev-list", "-n", "1", ref, cwd=cwd) except subprocess.CalledProcessError: return ref -def _safe_dir_name(label: str) -> str: - return "".join(ch if ch.isalnum() or ch in "._-" else "-" for ch in label) +def safe_name(value: str) -> str: + return "".join(ch if ch.isalnum() or ch in "._-" else "-" for ch in value) -def _create_venv(path: pathlib.Path) -> pathlib.Path: +def create_venv(path: pathlib.Path) -> pathlib.Path: subprocess.check_call([sys.executable, "-m", "venv", str(path)]) - if os.name == "nt": - return path / "Scripts" / "python.exe" - return path / "bin" / "python" - - -def _install_target(python: pathlib.Path, install_spec: str, cwd: pathlib.Path) -> None: - base_env = os.environ.copy() - base_env.update( - { - "NO_CYTHON": "1", - "PIP_DISABLE_PIP_VERSION_CHECK": "1", - "PYTHONUNBUFFERED": "1", - } - ) - subprocess.check_call( - [ - str(python), - "-m", - "pip", - "install", - "--upgrade", - "pip", - "setuptools", - "wheel", - "setuptools_scm[toml]", - "Cython>=3.0.0", - ], + return path / ("Scripts/python.exe" if os.name == "nt" else "bin/python") + + +def install_target(python: pathlib.Path, install_spec: str, cwd: pathlib.Path) -> None: + env = os.environ.copy() + env.update({"NO_CYTHON": "1", "PIP_DISABLE_PIP_VERSION_CHECK": "1", "PYTHONUNBUFFERED": "1"}) + run( + [str(python), "-m", "pip", "install", "--upgrade", "pip", "setuptools", "wheel", "setuptools_scm[toml]"], cwd=cwd, - env=base_env, + env=env, ) - subprocess.check_call( - [str(python), "-m", "pip", "install", "--no-deps", "--no-build-isolation", install_spec], + # Install runtime dependencies as well. The benchmark imports Faust normally, + # and an isolated --no-deps install fails before measurements can start. + run( + [str(python), "-m", "pip", "install", "--no-build-isolation", install_spec], cwd=cwd, - env=base_env, + env=env, ) -def _run_target(target: Target, workdir: pathlib.Path, source_root: pathlib.Path) -> dict[str, Any]: - venv_dir = workdir / _safe_dir_name(target.label) - python = _create_venv(venv_dir) - _install_target(python, target.install_spec, source_root) +def run_target(target: Target, workdir: pathlib.Path, source_root: pathlib.Path) -> dict[str, Any]: + python = create_venv(workdir / safe_name(target.label)) + install_target(python, target.install_spec, source_root) env = os.environ.copy() env.update({"PYTHONHASHSEED": "0", "PYTHONUNBUFFERED": "1"}) - output = _run([str(python), "-c", BENCHMARK_CODE], env=env) + output = run([str(python), "-c", BENCHMARK_CODE], env=env) payload = json.loads(output.splitlines()[-1]) payload.update({"target": target.label, "commit": target.commit, "kind": target.kind}) return payload -def _round(value: float) -> float: +def rounded(value: float) -> float: return round(float(value), 3) -def _benchmark_action_rows( - payloads: Iterable[dict[str, Any]], - *, - current_only: bool, -) -> list[dict[str, Any]]: - rows: list[dict[str, Any]] = [] +def action_rows(payloads: Iterable[dict[str, Any]], *, current_only: bool) -> list[dict[str, Any]]: + rows = [] for payload in payloads: if current_only and payload["target"] != "current": continue for result in payload["results"]: - metric_name = result["name"] if current_only else f"{payload['target']}/{result['name']}" - rows.append( - { - "name": metric_name, - "unit": result["unit"], - "value": _round(result["value"]), - "range": _round(result.get("range", 0.0)), - "extra": "\n".join( - [ - f"target: {payload['target']}", - f"kind: {payload['kind']}", - f"commit: {payload['commit']}", - f"faust: {payload['faust_version']}", - f"python: {payload['implementation']} {payload['python']}", - ] - ), - } - ) + name = result["name"] if current_only else f"{payload['target']}/{result['name']}" + rows.append({ + "name": name, + "unit": result["unit"], + "value": rounded(result["value"]), + "range": rounded(result.get("range", 0.0)), + "extra": "\n".join([ + f"target: {payload['target']}", + f"kind: {payload['kind']}", + f"commit: {payload['commit']}", + f"faust: {payload['faust_version']}", + f"python: {payload['implementation']} {payload['python']}", + ]), + }) return rows -def _result_map(payload: dict[str, Any]) -> dict[str, dict[str, Any]]: +def result_map(payload: dict[str, Any]) -> dict[str, dict[str, Any]]: return {result["name"]: result for result in payload["results"]} -def _format_value(result: dict[str, Any]) -> str: - return f"{_round(result['value'])} {result['unit']}" +def regressions(payloads: list[dict[str, Any]], threshold: float) -> list[str]: + current = result_map(payloads[0]) + failures = [] + for baseline in payloads[1:]: + baseline_results = result_map(baseline) + for name, current_result in current.items(): + ratio = current_result["value"] / baseline_results[name]["value"] + if ratio > threshold: + failures.append(f"{name} is {ratio:.2f}x slower than {baseline['target']}") + return failures -def _write_summary( - path: pathlib.Path, - payloads: list[dict[str, Any]], - release_tags: list[str], - failures: list[str], - max_regression: float, -) -> None: - current = next(payload for payload in payloads if payload["target"] == "current") - current_results = _result_map(current) +def write_summary(path: pathlib.Path, payloads: list[dict[str, Any]], tags: list[str], failures: list[str], threshold: float) -> None: + current = result_map(payloads[0]) lines = [ - "# Faust performance benchmark", - "", - f"Current commit: `{current['commit']}`", - f"Release baselines: {', '.join(f'`{tag}`' for tag in release_tags) or '_none found_'}", - f"Regression threshold: `{max_regression:.2f}x` slower than a release baseline", - "", + "# Faust performance benchmark", "", + f"Current commit: `{payloads[0]['commit']}`", + f"Release baselines: {', '.join(f'`{tag}`' for tag in tags) or '_none found_'}", + f"Regression threshold: `{threshold:.2f}x`", "", "| Benchmark | Current | Baseline | Delta | Status |", "|---|---:|---:|---:|---|", ] - for payload in payloads: - if payload["target"] == "current": - continue - baseline_results = _result_map(payload) - for name, current_result in current_results.items(): - baseline_result = baseline_results[name] - ratio = current_result["value"] / baseline_result["value"] - delta = (ratio - 1.0) * 100.0 - status = "❌ slower" if ratio > max_regression else "✅ ok" + for baseline in payloads[1:]: + baseline_results = result_map(baseline) + for name, current_result in current.items(): + base = baseline_results[name] + ratio = current_result["value"] / base["value"] + status = "❌ slower" if ratio > threshold else "✅ ok" lines.append( - "| {name} | {current_value} | {baseline_value} (`{target}`) | {delta:+.1f}% | {status} |".format( - name=name, - current_value=_format_value(current_result), - baseline_value=_format_value(baseline_result), - target=payload["target"], - delta=delta, - status=status, - ) + f"| {name} | {rounded(current_result['value'])} {current_result['unit']} | " + f"{rounded(base['value'])} {base['unit']} (`{baseline['target']}`) | " + f"{(ratio - 1) * 100:+.1f}% | {status} |" ) - lines.extend(["", "## Raw current results", "", "```json"]) - lines.append(json.dumps(_benchmark_action_rows([current], current_only=True), indent=2, sort_keys=True)) - lines.append("```") if failures: - lines.extend(["", "## Regressions", ""]) - lines.extend(f"- {failure}" for failure in failures) + lines.extend(["", "## Regressions", *[f"- {failure}" for failure in failures]]) path.write_text("\n".join(lines) + "\n") -def _find_regressions(payloads: list[dict[str, Any]], max_regression: float) -> list[str]: - current = next(payload for payload in payloads if payload["target"] == "current") - current_results = _result_map(current) - failures: list[str] = [] - for payload in payloads: - if payload["target"] == "current": - continue - baseline_results = _result_map(payload) - for name, current_result in current_results.items(): - baseline_result = baseline_results[name] - ratio = current_result["value"] / baseline_result["value"] - if ratio > max_regression: - failures.append( - f"{name} is {ratio:.2f}x slower than {payload['target']} " - f"({_format_value(current_result)} vs {_format_value(baseline_result)})" - ) - return failures - - -def _float_env(name: str, default: float) -> float: - value = os.environ.get(name) - return float(value) if value else default - - def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--repo", default=os.environ.get("GITHUB_REPOSITORY", DEFAULT_REPO)) parser.add_argument("--release-count", type=int, default=DEFAULT_RELEASE_COUNT) - parser.add_argument("--max-regression", type=float, default=_float_env("PERF_MAX_REGRESSION", DEFAULT_MAX_REGRESSION)) + parser.add_argument("--max-regression", type=float, default=float(os.environ.get("PERF_MAX_REGRESSION") or DEFAULT_MAX_REGRESSION)) parser.add_argument("--output-json", type=pathlib.Path, default=pathlib.Path("benchmark-results.json")) parser.add_argument("--current-output-json", type=pathlib.Path, default=pathlib.Path("benchmark-current.json")) parser.add_argument("--summary", type=pathlib.Path, default=pathlib.Path("benchmark-summary.md")) @@ -402,29 +304,20 @@ def parse_args() -> argparse.Namespace: def main() -> int: args = parse_args() - source_root = pathlib.Path.cwd() - release_tags = _discover_release_tags(args.repo, args.release_count, source_root) - current_commit = _commit_for_ref("HEAD", source_root) - targets = [Target("current", ".", current_commit, "checkout")] - targets.extend( - Target( - tag, - f"git+https://github.com/{args.repo}.git@{tag}", - _commit_for_ref(tag, source_root), - "release", - ) - for tag in release_tags - ) + root = pathlib.Path.cwd() + tags = discover_release_tags(args.repo, args.release_count, root) + targets = [Target("current", ".", commit_for_ref("HEAD", root), "checkout")] + targets.extend(Target(tag, f"git+https://github.com/{args.repo}.git@{tag}", commit_for_ref(tag, root), "release") for tag in tags) tempdir = pathlib.Path(tempfile.mkdtemp(prefix="faust-benchmark-")) - payloads: list[dict[str, Any]] = [] + payloads = [] try: for target in targets: print(f"::group::Benchmark {target.label}") - payload = _run_target(target, tempdir, source_root) + payload = run_target(target, tempdir, root) payloads.append(payload) for result in payload["results"]: - print(f"{target.label}/{result['name']}: {_format_value(result)}") + print(f"{target.label}/{result['name']}: {rounded(result['value'])} {result['unit']}") print("::endgroup::") finally: if args.keep_workdir: @@ -432,21 +325,13 @@ def main() -> int: else: shutil.rmtree(tempdir, ignore_errors=True) - failures = _find_regressions(payloads, args.max_regression) if release_tags else [] - args.output_json.write_text( - json.dumps(_benchmark_action_rows(payloads, current_only=False), indent=2, sort_keys=True) + "\n" - ) - args.current_output_json.write_text( - json.dumps(_benchmark_action_rows(payloads, current_only=True), indent=2, sort_keys=True) + "\n" - ) - _write_summary(args.summary, payloads, release_tags, failures, args.max_regression) - - if failures: - print("::error::Performance regression threshold exceeded") - for failure in failures: - print(f"::error::{failure}") - return 1 - return 0 + failures = regressions(payloads, args.max_regression) if tags else [] + args.output_json.write_text(json.dumps(action_rows(payloads, current_only=False), indent=2, sort_keys=True) + "\n") + args.current_output_json.write_text(json.dumps(action_rows(payloads, current_only=True), indent=2, sort_keys=True) + "\n") + write_summary(args.summary, payloads, tags, failures, args.max_regression) + for failure in failures: + print(f"::error::{failure}") + return 1 if failures else 0 if __name__ == "__main__":