diff --git a/.github/workflows/bench.yml b/.github/workflows/bench.yml new file mode 100644 index 00000000..3904a6c7 --- /dev/null +++ b/.github/workflows/bench.yml @@ -0,0 +1,126 @@ +name: bench + +on: + pull_request: + branches: + - "**" + workflow_dispatch: + inputs: + compare-fail: + description: "Regression threshold (pytest-benchmark --benchmark-compare-fail), e.g. median:25%" + default: "median:25%" + +permissions: + contents: read + +# Only the most recent run per PR/branch matters; cancel superseded ones. +concurrency: + group: bench-${{ github.workflow }}-${{ github.ref }} + cancel-in-progress: true + +jobs: + bench: + name: bench (${{ matrix.os }}) + runs-on: ${{ matrix.os }} + timeout-minutes: 30 + strategy: + # Each runner compares base vs head against itself, so one runner + # failing its threshold should not stop the others from reporting. + fail-fast: false + matrix: + os: ["ubuntu-latest", "macos-latest"] + env: + # Compare base vs head on the SAME machine; storage lives outside the + # repo so switching git refs never touches it. + BENCH_STORAGE: ${{ runner.temp }}/benchmarks + BENCH_PATHS: benches/bench_protocol.py benches/bench_client.py + COMPARE_FAIL: ${{ inputs.compare-fail || 'median:25%' }} + defaults: + run: + working-directory: nats-core + steps: + - name: Check out repository + uses: actions/checkout@v5 + with: + # Full history so we can check out the PR base commit. + fetch-depth: 0 + + - name: Set up Go + uses: actions/setup-go@v6 + with: + go-version: "stable" + + - name: Install NATS Server + run: go install github.com/nats-io/nats-server/v2@latest + shell: bash + working-directory: . + + - name: Set up Python + uses: actions/setup-python@v6 + with: + python-version: "3.13" + + - name: Install uv + uses: astral-sh/setup-uv@v6 + + - name: Resolve refs + id: refs + shell: bash + run: | + base="${{ github.event.pull_request.base.sha }}" + head="${{ github.event.pull_request.head.sha || github.sha }}" + if [ -z "$base" ]; then + # Not a PR (e.g. workflow_dispatch): compare against parent commit. + base="$(git rev-parse "${head}^")" + fi + echo "base=$base" >>"$GITHUB_OUTPUT" + echo "head=$head" >>"$GITHUB_OUTPUT" + echo "Base: $base" + echo "Head: $head" + + # --- Baseline: the PR's merge base / target --------------------------- + - name: Check out base + env: + BASE_SHA: ${{ steps.refs.outputs.base }} + run: git checkout --force --detach "$BASE_SHA" + working-directory: . + + - name: Install dependencies (base) + run: uv sync --dev + + - name: Run benchmarks (base) + run: | + uv run pytest $BENCH_PATHS \ + --benchmark-only \ + --benchmark-save=base \ + --benchmark-storage="file://$BENCH_STORAGE" \ + --benchmark-json="$RUNNER_TEMP/base.json" + + # --- Candidate: the PR head, compared against the baseline ------------ + - name: Check out head + env: + HEAD_SHA: ${{ steps.refs.outputs.head }} + run: git checkout --force --detach "$HEAD_SHA" + working-directory: . + + - name: Install dependencies (head) + run: uv sync --dev + + - name: Run benchmarks (head) and compare + run: | + uv run pytest $BENCH_PATHS \ + --benchmark-only \ + --benchmark-storage="file://$BENCH_STORAGE" \ + --benchmark-compare=0001 \ + --benchmark-compare-fail="$COMPARE_FAIL" \ + --benchmark-json="$RUNNER_TEMP/head.json" + + - name: Upload benchmark results + if: always() + uses: actions/upload-artifact@v4 + with: + name: benchmarks-${{ matrix.os }} + path: | + ${{ runner.temp }}/base.json + ${{ runner.temp }}/head.json + if-no-files-found: warn diff --git a/nats-core/benches/bench_client.py b/nats-core/benches/bench_client.py index 2255aee1..472af795 100644 --- a/nats-core/benches/bench_client.py +++ b/nats-core/benches/bench_client.py @@ -63,3 +63,102 @@ def teardown(loop, server, client): result = benchmark.pedantic(execute, setup=setup, teardown=teardown, iterations=1, rounds=1) return result + + +REQUEST_SIZES = [1, 16, 128, 1024, 8192] + +# The latency bench measures a single round-trip and lets pytest-benchmark +# repeat it across rounds, so no count is needed. The throughput bench drives a +# fixed number of requests through a bounded in-flight window and measures how +# fast they all complete; an unbounded gather starves under its own backlog. +THROUGHPUT_COUNT = 10_000 +THROUGHPUT_INFLIGHT = 100 + + +def _start_responder(subject): + """Spin up a server, client, and an echo responder on ``subject``.""" + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + server = loop.run_until_complete(run(port=0)) + client = loop.run_until_complete(connect(server.client_url)) + + async def start(): + sub = await client.subscribe(subject) + + async def respond(): + async for msg in sub.messages: + if msg.reply: + await client.publish(msg.reply, msg.data) + + return sub, asyncio.ensure_future(respond()) + + sub, responder = loop.run_until_complete(start()) + return (loop, server, client, sub, responder), {} + + +def _stop_responder(loop, server, client, sub, responder): + """Tear down the responder, client, server, and event loop.""" + + async def stop(): + responder.cancel() + try: + await responder + except asyncio.CancelledError: + pass + + loop.run_until_complete(stop()) + loop.run_until_complete(client.close()) + loop.run_until_complete(server.shutdown()) + loop.close() + asyncio.set_event_loop(None) + + +@pytest.mark.parametrize("size", REQUEST_SIZES) +def test_bench_request_latency(benchmark, size): + """Benchmark a single request/reply round-trip (latency).""" + subject = "bench.request" + payload = b"x" * size + + # Set the server up once; pytest-benchmark repeats just the round-trip, + # giving a per-request latency distribution instead of one aggregate. + (loop, server, client, sub, responder), _ = _start_responder(subject) + + def one_request(): + loop.run_until_complete(client.request(subject, payload, timeout=5.0)) + + benchmark.extra_info["message_size"] = size + + try: + benchmark(one_request) + finally: + _stop_responder(loop, server, client, sub, responder) + + +@pytest.mark.parametrize("size", REQUEST_SIZES) +def test_bench_request_throughput(benchmark, size): + """Benchmark concurrent in-flight request/reply (throughput bound).""" + subject = "bench.request" + payload = b"x" * size + + def execute(loop, server, client, sub, responder): + async def request_n(): + sem = asyncio.Semaphore(THROUGHPUT_INFLIGHT) + + async def one(): + async with sem: + await client.request(subject, payload, timeout=5.0) + + await asyncio.gather(*(one() for _ in range(THROUGHPUT_COUNT))) + + loop.run_until_complete(request_n()) + + benchmark.extra_info["message_size"] = size + benchmark.extra_info["message_count"] = THROUGHPUT_COUNT + + return benchmark.pedantic( + execute, + setup=lambda: _start_responder(subject), + teardown=_stop_responder, + iterations=1, + rounds=1, + ) diff --git a/nats-core/benches/bench_protocol.py b/nats-core/benches/bench_protocol.py index 4a0add2c..13ecc3c4 100644 --- a/nats-core/benches/bench_protocol.py +++ b/nats-core/benches/bench_protocol.py @@ -22,7 +22,7 @@ def test_bench_encode_connect(benchmark): @pytest.mark.parametrize("size", [1, 2, 4, 8, 16, 32, 64, 128, 256, 512, 1024, 2048, 4096, 8192]) def test_bench_encode_pub_with_payload(benchmark, size): """Benchmark encoding PUB command with various payload sizes.""" - subject = "test.subject" + subject = b"test.subject" payload = b"x" * size benchmark(command.encode_pub, subject, payload) @@ -30,16 +30,16 @@ def test_bench_encode_pub_with_payload(benchmark, size): def test_bench_encode_pub_with_reply(benchmark): """Benchmark encoding PUB command with reply subject.""" - subject = "test.subject" + subject = b"test.subject" payload = b"hello world" - reply = "reply.subject" + reply = b"reply.subject" benchmark(command.encode_pub, subject, payload, reply=reply) def test_bench_encode_hpub_single_header(benchmark): """Benchmark encoding HPUB command with single header.""" - subject = "test.subject" + subject = b"test.subject" payload = b"hello world" header_data = command.encode_headers({"X-Custom": "value"}) @@ -48,7 +48,7 @@ def test_bench_encode_hpub_single_header(benchmark): def test_bench_encode_hpub_multiple_headers(benchmark): """Benchmark encoding HPUB command with multiple headers.""" - subject = "test.subject" + subject = b"test.subject" payload = b"hello world" header_data = command.encode_headers( { @@ -65,7 +65,7 @@ def test_bench_encode_hpub_multiple_headers(benchmark): def test_bench_encode_hpub_multivalue_headers(benchmark): """Benchmark encoding HPUB command with multi-value headers.""" - subject = "test.subject" + subject = b"test.subject" payload = b"hello world" header_data = command.encode_headers( { @@ -79,9 +79,9 @@ def test_bench_encode_hpub_multivalue_headers(benchmark): def test_bench_encode_hpub_with_reply(benchmark): """Benchmark encoding HPUB command with reply subject and headers.""" - subject = "test.subject" + subject = b"test.subject" payload = b"hello world" - reply = "reply.subject" + reply = b"reply.subject" header_data = command.encode_headers({"X-Custom": "value"}) benchmark(command.encode_hpub, subject, payload, reply=reply, header_data=header_data)