diff --git a/.ai/PERFORMANCE.md b/.ai/PERFORMANCE.md index 47d91f8321..58d5d0ac1f 100644 --- a/.ai/PERFORMANCE.md +++ b/.ai/PERFORMANCE.md @@ -95,6 +95,54 @@ production bundle, runs the harness against `baseline.json`, and: Thresholds live per-scenario in `scenarios.py` (`warn_ms` / `fail_ms`, keyed by metric). Keep them generous: this is a smoke alarm, not a microbenchmark. +## Load tests and published numbers + +`benchmarks/streaming/` measures streaming-callback capacity per server setup +(see `benchmarks/README.md` for how to run it). Simulated browsers follow the +renderer's `StreamClient`: one downlink per browser, a long NDJSON response on +ASGI and polls with the same backoff on WSGI. Frames carry the server's +`time.time()`, so latency is measured end to end on one clock. + +Pitfalls the harness guards against: + +- One Python client process tops out at a few hundred browsers and then + reports its own lag as server latency. Clients run 100 browsers per process, + on CPUs separate from the server, and a point is flagged `client_bound` when + a client hits 85% CPU. +- Client output goes to temp files, not pipes: a full pipe blocks the client + forever because the runner only reads after the clients exit. +- Multi-worker setups need shared storage (Redis); with a per-process secret + the workers also reject each other's stream tokens, so the runner sets + a shared secret (`LOAD_SECRET`). The server imports dash from the checkout + under test (`PYTHONPATH`), not whatever is installed. + +`benchmarks/callbacks/` runs the same way for plain callbacks, HTTP vs +websocket per backend; both load tests share `benchmarks/loadkit.py`. + +`benchmarks/swarm/` drives real headless Chromium (Playwright) against a +deployed app, on as many machines as needed. Pitfalls it guards against: + +- Timings are taken in the page (`performance.now()` from the click to the DOM + update), so machines never need agreeing clocks; only the start is shared, + and an agent that got its browsers up after it is flagged `late`. +- A failed reload restarts the page's click counts; the expected output text + is derived from them, so a stale count would time out every later click. +- Steps stop when the swarm stops delivering its load (failed agents, users + that never loaded, agents over 85% CPU), not only when the server slows. +- `page.on("websocket")` does not see sockets opened by a SharedWorker; to + confirm a click went over the websocket, check that it made no HTTP POST. + +`benchmarks/serialization/` times `dash._utils.to_json` (plotly's +`to_json_plotly`) on one large figure, alone and next to a small value orjson +cannot serialize, plus the figure inside a layout. Each engine runs in its own +process; the stdlib one blocks the orjson import, so it is a real install +without orjson, not just a config switch. See finding 6. + +`benchmarks/publish.py` commits results to `benchmarks/results/` (read by the +docs) and the table in `README.md`. `.github/workflows/benchmarks-publish.yml` +runs weekly and opens a pull request only when the numbers moved +significantly (`--only-significant`). + ## Profiling a slow scenario The runner can capture a **Chrome DevTools CPU profile** of a scenario and print @@ -198,3 +246,48 @@ exists for growing containers; it is kept as a contrast with loose thresholds. limit) for a component tree nested deeper than ~254. The `deep_nesting` scenario is capped at 120 to stay clear. Worth remembering before recommending deeply-recursive layouts. + +### 6. One non-native value sends the whole response down plotly's slow path (#4065) + +With orjson, `to_json_plotly` tries `orjson.dumps` on the whole response. If +any value is not orjson-native (a component, a `go.Figure` object, a +`pd.Timestamp`), it discards the attempt, walks everything in Python with +`clean_to_json_compatible`, then dumps again. A `go.Figure` also pays a +`deepcopy` of its data in `to_dict()`. Layouts always take this path, since +they are made of components. The cost scales with the number of Python-level +values in the response: numpy arrays pass through the walk whole, lists of +floats or datetimes are visited one value at a time. 10 traces x 20k points, +Python 3.13, plotly 6.7: + +| figure data | alone | + html.Div | as go.Figure | in a layout | no orjson, alone | +|--|--:|--:|--:|--:|--:| +| numpy array | 3 ms | 3 ms | 5 ms | 5 ms | 10 ms | +| list of float | 6 ms | 40 ms | 75 ms | 75 ms | 52 ms | +| pd.DatetimeIndex | 14 ms | 50 ms | 42 ms | 42 ms | 75 ms | +| list of datetime | 8 ms | 171 ms | 375 ms | 373 ms | 354 ms | + +The fix belongs in plotly.py, not Dash: + +- pass `default=` to `orjson.dumps` so only the non-native values are + converted, keeping the old path as a fallback on `TypeError`; +- give figures and traces a `_to_json_dict()` that skips the `deepcopy` and + copies only the containers the base64 conversion changes; skip lists of + scalars with one `set(map(type, lst))` check instead of a Python loop; +- in `PlotlyJSONEncoder.default`, return `isoformat()` first for exact + `datetime`/`date` (it now tries five encoders that raise before that one). + +A local patch to plotly 6.7 doing this brought every orjson case within ~3 ms +of the figure alone (list of datetime as `go.Figure` or in a layout: 375 -> +11 ms), and the stdlib list of datetime from 354 to 109 ms. Its JSON was +identical except numpy `datetime64`, which then always uses orjson's short +form (`2024-01-01T00:00:00`), as the fast path already did. + +Without orjson, what is left is the stdlib encoder: 200k floats take ~50 ms, +and 200k datetimes ~70 ms just to make their `isoformat()` strings, before +encoding. Bulk conversion does not help: numpy, pandas and pyarrow all pay +per object to read a list of datetimes (77-149 ms), and pyarrow casts a +timestamp column to strings in 5 ms but has no JSON writer for a nested +response, so the strings come back out as Python objects at the same cost. +Native encoders write bytes without Python objects: orjson and msgspec do +200k floats in ~5 ms and 200k datetimes in 2-4 ms. Making orjson a Dash +dependency is the change that helps every case. diff --git a/.github/workflows/benchmarks-publish.yml b/.github/workflows/benchmarks-publish.yml new file mode 100644 index 0000000000..9ac2e3a63d --- /dev/null +++ b/.github/workflows/benchmarks-publish.yml @@ -0,0 +1,236 @@ +name: Publish Benchmarks + +# Measures the benchmarks on the dev branch weekly (and on demand). Only when +# the numbers moved enough to matter (a capacity step in the load tests, a +# renderer median by over 25% and 5 ms, or a serialization median by over 25% +# and 5 ms) does it open a pull request that +# commits them to benchmarks/results/ (read by the docs) and the README table. +# Per-PR renderer regressions are benchmarks.yml's job, not this one. +# +# Needs "Allow GitHub Actions to create and approve pull requests" in the +# repo settings. A pull request opened with GITHUB_TOKEN starts no other +# workflows: close and reopen it to run the checks before merging. + +on: + schedule: + - cron: '0 6 * * 1' + workflow_dispatch: + inputs: + streaming: + description: Run the streaming load sweep too + type: boolean + default: true + callbacks: + description: Run the HTTP vs websocket callback sweep too + type: boolean + default: true + +permissions: + contents: read + +concurrency: + group: benchmarks-publish + cancel-in-progress: false + +jobs: + renderer: + name: Renderer benchmarks + runs-on: ubuntu-latest + timeout-minutes: 30 + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-node@v4 + with: + node-version: '24' + cache: npm + - run: npm ci + - uses: actions/setup-python@v5 + with: + python-version: '3.12' + cache: pip + - name: Install Dash + run: | + python -m pip install --upgrade pip + python -m pip install "setuptools<80.0.0" + python -m pip install -e .[ci,dev,testing] + - run: npm run build + - uses: browser-actions/setup-chrome@c785b87e244131f27c9f19c1a33e2ead956ab7ce # v1 + with: + chrome-version: stable + - name: Set up virtual display + run: | + sudo apt-get update + sudo apt-get install -y xvfb + sudo Xvfb :99 -ac -screen 0 1400x1000x24 & + echo "DISPLAY=:99" >> $GITHUB_ENV + - run: python -m benchmarks.run --out benchmarks/results.json + - uses: actions/upload-artifact@v4 + with: + name: renderer-results + path: benchmarks/results.json + + streaming: + name: Streaming load sweep + if: github.event_name == 'schedule' || (github.event_name == 'workflow_dispatch' && inputs.streaming) + runs-on: ubuntu-latest + timeout-minutes: 120 + services: + redis: + image: redis:7-alpine + ports: + - 6379:6379 + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-node@v4 + with: + node-version: '24' + cache: npm + - run: npm ci + - uses: actions/setup-python@v5 + with: + python-version: '3.12' + cache: pip + - name: Install Dash + run: | + python -m pip install --upgrade pip + python -m pip install "setuptools<80.0.0" + python -m pip install -e .[ci,dev,testing,async,fastapi,quart,redis] + python -m pip install gunicorn uvicorn aiohttp psutil + - run: npm run build + - name: Run the sweep + env: + BENCH_RUNNER: GitHub Actions ubuntu-latest + run: | + ulimit -n 65536 + python -m benchmarks.streaming.load \ + --redis-url redis://127.0.0.1:6379/0 \ + --out benchmarks/streaming/results.json + - uses: actions/upload-artifact@v4 + with: + name: streaming-results + path: benchmarks/streaming/results.json + + callbacks: + name: Callback load sweep + if: github.event_name == 'schedule' || (github.event_name == 'workflow_dispatch' && inputs.callbacks) + runs-on: ubuntu-latest + timeout-minutes: 120 + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-node@v4 + with: + node-version: '24' + cache: npm + - run: npm ci + - uses: actions/setup-python@v5 + with: + python-version: '3.12' + cache: pip + - name: Install Dash + run: | + python -m pip install --upgrade pip + python -m pip install "setuptools<80.0.0" + python -m pip install -e .[ci,dev,testing,async,fastapi,quart] + python -m pip install gunicorn uvicorn aiohttp psutil + - run: npm run build + - name: Run the sweep + env: + BENCH_RUNNER: GitHub Actions ubuntu-latest + run: | + ulimit -n 65536 + python -m benchmarks.callbacks.load --out benchmarks/callbacks/results.json + - uses: actions/upload-artifact@v4 + with: + name: callbacks-results + path: benchmarks/callbacks/results.json + + serialization: + name: Serialization benchmarks + runs-on: ubuntu-latest + timeout-minutes: 30 + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-node@v4 + with: + node-version: '24' + cache: npm + - run: npm ci + - uses: actions/setup-python@v5 + with: + python-version: '3.12' + cache: pip + - name: Install Dash + run: | + python -m pip install --upgrade pip + python -m pip install "setuptools<80.0.0" + python -m pip install -e .[ci,dev,testing] + - run: npm run build + # Runs with orjson and again with its import blocked. + - run: python -m benchmarks.serialization.bench --out benchmarks/serialization/results.json + - uses: actions/upload-artifact@v4 + with: + name: serialization-results + path: benchmarks/serialization/results.json + + publish: + name: Open the results pull request + needs: [renderer, streaming, callbacks, serialization] + # Only dev's numbers are committed, even when a dispatch runs elsewhere. + if: >- + always() && github.ref == 'refs/heads/dev' && + (needs.renderer.result == 'success' || needs.streaming.result == 'success' || + needs.callbacks.result == 'success' || needs.serialization.result == 'success') + runs-on: ubuntu-latest + permissions: + contents: write + pull-requests: write + env: + BRANCH: benchmarks/update-results + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: '3.12' + - uses: actions/download-artifact@v4 + with: + path: results + - name: Update the results + env: + BENCH_RUNNER: GitHub Actions ubuntu-latest + run: | + args="" + [ -f results/renderer-results/results.json ] && args="$args --renderer results/renderer-results/results.json" + [ -f results/streaming-results/results.json ] && args="$args --streaming results/streaming-results/results.json" + [ -f results/callbacks-results/results.json ] && args="$args --callbacks results/callbacks-results/results.json" + [ -f results/serialization-results/results.json ] && args="$args --serialization results/serialization-results/results.json" + python -m pip install -e . + python -m benchmarks.publish --commit "${GITHUB_SHA::9}" \ + --only-significant --changes "$RUNNER_TEMP/changes.md" $args + - name: Open or update the pull request + env: + GH_TOKEN: ${{ github.token }} + run: | + git add benchmarks/results README.md + if git diff --cached --quiet; then + echo "No significant change." + exit 0 + fi + { + echo "Benchmark numbers from ${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY}/actions/runs/${GITHUB_RUN_ID} moved enough to publish:" + echo + cat "$RUNNER_TEMP/changes.md" + echo + echo "Merge to update the README table and the docs." + } > "$RUNNER_TEMP/body.md" + git config user.name "github-actions[bot]" + git config user.email "41898282+github-actions[bot]@users.noreply.github.com" + git switch -c "$BRANCH" + git commit -m "Update benchmark results for ${GITHUB_SHA::9}" + # Each run starts from dev, so an unmerged run is replaced, not stacked. + git push --force origin "$BRANCH" + if [ -z "$(gh pr list --head "$BRANCH" --state open --json number --jq '.[].number')" ]; then + gh pr create --base dev --head "$BRANCH" \ + --title "Update benchmark results" --body-file "$RUNNER_TEMP/body.md" + else + gh pr edit "$BRANCH" --body-file "$RUNNER_TEMP/body.md" + fi diff --git a/README.md b/README.md index bd4493c10a..8ae4f73571 100644 --- a/README.md +++ b/README.md @@ -32,6 +32,17 @@ Built on top of [Plotly.js](https://github.com/plotly/plotly.js), [React](https: To learn more about Dash, read the [extensive announcement letter](https://medium.com/@plotlygraphs/introducing-dash-5ecf7191b503) or [jump in with the user guide](https://plotly.com/dash). +### Performance + +How many people can use one Dash server at once? CI measures it weekly, from +simulated browsers clicking callbacks and streaming against each server setup. + + +No published run yet. + + +Charts, the full numbers and how they are measured: [benchmarks](https://github.com/plotly/dash/tree/dev/benchmarks). + ### Dash OSS & Dash Enterprise With Dash Open Source, Dash apps run on your local laptop or workstation, but cannot be easily accessed by others in your organization. diff --git a/benchmarks/.gitignore b/benchmarks/.gitignore index 60e8dc325d..c25b4bd15b 100644 --- a/benchmarks/.gitignore +++ b/benchmarks/.gitignore @@ -1,5 +1,7 @@ -# Generated benchmark artifacts (baseline.json is committed on purpose) +# Generated benchmark artifacts (baseline.json and results/ are committed on purpose) results.json summary.md *.cpuprofile __pycache__/ +streaming/results.json +callbacks/results.json diff --git a/benchmarks/README.md b/benchmarks/README.md index 2e5888567c..00d20616e6 100644 --- a/benchmarks/README.md +++ b/benchmarks/README.md @@ -39,3 +39,127 @@ scenario( `b` is the browser helper (`b.timed`, `b.render_time`, `b.reload`, `b.state`, `b.graph_time`). Every layout must end with the shared `READY` sentinel so the harness can detect "fully hydrated". Then regenerate `baseline.json`. + +## Streaming load test (`streaming/`) + +How many browsers a server setup can stream to. Each setup (Flask on +gunicorn, FastAPI and Quart on uvicorn, 1 or 4 workers) serves +`streaming/load_app.py`, and simulated browsers (`streaming/client.py`) stream +from it while the count climbs until p95 frame latency passes 1 s or errors +pass 1%. + +```bash +ulimit -n 65536 +python -m benchmarks.streaming.load --redis-url redis://127.0.0.1:6379/0 +python -m benchmarks.streaming.load --setup fastapi-uvicorn-w1 --browsers 100 1000 +``` + +Multi-worker setups share frames through Redis and are skipped without +`--redis-url`. The server and the clients get separate CPUs (`--server-cpus`, +half the machine by default); a point where a client process hit 85% CPU is +flagged `client_bound` and ends that setup's sweep, since past that the numbers +measure the load generator. + +## Callback load test (`callbacks/`) + +Plain callbacks over HTTP vs websocket callbacks, by backend: Flask (HTTP +only), FastAPI and Quart (both), 1 and 4 workers. Simulated browsers click +and wait for the response; the count climbs until p95 round trip passes 1 s +or errors pass 1%. Shares its runner with the streaming test +(`loadkit.py`). + +```bash +python -m benchmarks.callbacks.load +python -m benchmarks.callbacks.load --setup fastapi-uvicorn-w1-ws --browsers 100 1000 +python -m benchmarks.callbacks.load --kind async --work-ms 50 # like a query +``` + +A websocket stays on the worker that accepted it, so with several workers +each browser's callbacks all run on one worker. `--kind sync` (default) runs +plain `def` callbacks; `--work-ms` makes each +one burn that much CPU (or `asyncio.sleep` with `--kind async`). + +## Browser swarm (`swarm/`) + +The simulated clients above speak the wire protocols. The swarm drives real +headless Chromium instead, many users per machine and many machines, against +a deployed app, so the numbers include the renderer, its SharedWorkers and +the browser's own connection limits. Each user clicks a plain HTTP callback, +the same callback over the websocket, or a streaming callback +(`swarm/app.py`), timed in the page from the click to the DOM update. + +Serve the app where it will be tested, the way it would be deployed: + +```bash +SWARM_BACKEND=fastapi SWARM_REDIS_URL=redis://redis:6379/0 \ + uvicorn benchmarks.swarm.app:server --host 0.0.0.0 --port 8050 --workers 4 +``` + +On Flask (no websocket) pass `--scenario http,stream`. + +Try the swarm on one machine first (agents as local processes): + +```bash +pip install playwright psutil && playwright install --only-shell chromium +python -m benchmarks.swarm.run --target http://127.0.0.1:8050 --local 2 \ + --users 10 25 --ramp 10 --duration 20 +``` + +On a cloud, build the agent image (`swarm/Dockerfile`), push it to a +registry, start N machines with Docker that can reach the app, pull the image +on each, and list them in a hosts file (one `user@host` per line). Then from +any machine with SSH access to them: + +```bash +docker build -t REGISTRY/dash-swarm-agent benchmarks/swarm && docker push REGISTRY/dash-swarm-agent +python -m benchmarks.swarm.run --target http://APP:8050 --hosts hosts.txt \ + --image REGISTRY/dash-swarm-agent --users 25 50 100 --out swarm.json +``` + +`--users` is per machine and per step; every agent starts at the same +wall-clock time (`--lead` seconds after the command, NTP keeps cloud clocks +close enough) and reports come back over SSH. Steps stop climbing at 1% +errors or a p95 over `--max-p95` ms (whole streams excluded: they take +frames x interval by design). An agent whose machine passed 85% CPU is +reported as `AGENT-BOUND`: give it fewer users or a bigger machine, its +numbers describe the browsers, not the server. Measured locally at 1 s think +time, a user costs about 0.05 core and up to 170 MB (RSS, an overcount); +start at about 30 users per 4 vCPU / 8 GB machine and watch the flag. + +Reports from any number of agents merge exactly (`swarm/aggregate.py`): + +```bash +python -m benchmarks.swarm.aggregate reports/*.json --out swarm.json +``` + +## Serialization (`serialization/`) + +How long Dash takes to turn a callback response or a layout into JSON, on the +server, for one large figure: alone, next to a small value orjson cannot +serialize (`html.Div`, `pd.Timestamp`), returned as a `go.Figure` object, and +inside a layout. Runs once with orjson and once with its import blocked (issue +#4065). + +```bash +python -m benchmarks.serialization.bench +python -m benchmarks.serialization.bench --case list_float+div --engine json --points 100000 +``` + +## Publishing (`publish.py`) + +`.github/workflows/benchmarks-publish.yml` measures on `dev` weekly (and on +demand). Only when the numbers moved enough to matter (a load test's capacity +changes step, a renderer median moves by over 25% and 5 ms, or a serialization +median by over 25% and 5 ms) does it open a +pull request that commits them to `results/` and the table in the repo's +`README.md`; the pull request lists what moved. `results/` holds the raw numbers +(`*-latest.json`), one summary per run (`*-history.json`), and `report.json`: +every chart as a Plotly figure and the numbers as tables, which the docs page +renders. To update them locally: + +```bash +python -m benchmarks.publish \ + --renderer benchmarks/results.json --streaming benchmarks/streaming/results.json \ + --callbacks benchmarks/callbacks/results.json \ + --serialization benchmarks/serialization/results.json +``` diff --git a/benchmarks/callbacks/__init__.py b/benchmarks/callbacks/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/benchmarks/callbacks/client.py b/benchmarks/callbacks/client.py new file mode 100644 index 0000000000..2fd2b5290e --- /dev/null +++ b/benchmarks/callbacks/client.py @@ -0,0 +1,213 @@ +"""Simulated browsers for the callback load test. + +Run by ``benchmarks.callbacks.load``, several processes at once. Each browser +loads the page once, then loops: click, wait for the callback's response, +think, click again. Over ``http`` every click is a POST to +``/_dash-update-component``, as the renderer sends it (six connections per +host at most, kept alive). Over ``ws`` the browser holds one websocket to +``/_dash-ws-callback`` and sends ``callback_request`` messages on it, as the +renderer's SharedWorker does; replies may arrive batched in a JSON array. + +Prints one JSON line with the round trip of every click made inside the +measure window, however late it finished. +""" +import argparse +import asyncio +import itertools +import json +import random +import sys +import time + +import aiohttp + +UPDATE = "/_dash-update-component" +WS = "/_dash-ws-callback" +TIMEOUT = 30 + + +def _payload(n): + return { + "output": "out.children", + "outputs": {"id": "out", "property": "children"}, + "inputs": [{"id": "btn", "property": "n_clicks", "value": n}], + "changedPropIds": ["btn.n_clicks"], + "state": [], + } + + +class Stats: + def __init__(self, measure_from, measure_to): + self.measure_from = measure_from + self.measure_to = measure_to + self.rtt_ms = [] + self.calls = 0 + self.errors = 0 + self.failed_browsers = 0 + + def in_window(self, t): + return self.measure_from <= t <= self.measure_to + + +class Browser: + _ids = itertools.count() + + def __init__(self, base, stats, transport, think): + self.base = base + self.stats = stats + self.transport = transport + self.think = think + self.renderer_id = f"r{next(self._ids)}" + self.session = aiohttp.ClientSession( + connector=aiohttp.TCPConnector(limit=6), + timeout=aiohttp.ClientTimeout(total=None, sock_connect=30), + ) + self.ws = None + self.pending = {} + self.readers = set() + + async def _http_call(self, n): + async with self.session.post(self.base + UPDATE, json=_payload(n)) as resp: + resp.raise_for_status() + body = await resp.json(content_type=None) + if "response" not in body: + raise ValueError(f"unexpected response {body!r:.200}") + + async def _ws_connect(self): + self.ws = await self.session.ws_connect( + self.base.replace("http", "ws", 1) + WS, + origin=self.base, + max_msg_size=0, + ) + # Each connection gets its own pending requests: a closing one must + # not fail the requests of the connection that replaced it. + self.pending = {} + self.readers.add(asyncio.ensure_future(self._ws_read(self.ws, self.pending))) + + @staticmethod + def _resolve(pending, data): + for item in data if isinstance(data, list) else [data]: + if item.get("type") != "callback_response": + continue + fut = pending.pop(item.get("requestId"), None) + if fut is not None and not fut.done(): + fut.set_result(item.get("payload")) + + @classmethod + async def _ws_read(cls, ws, pending): + try: + async for msg in ws: + if msg.type != aiohttp.WSMsgType.TEXT: + break + cls._resolve(pending, json.loads(msg.data)) + finally: + for fut in pending.values(): + if not fut.done(): + fut.set_exception(ConnectionError("websocket closed")) + pending.clear() + + async def _ws_call(self, n): + request_id = f"{self.renderer_id}-{n}" + fut = asyncio.get_running_loop().create_future() + self.pending[request_id] = fut + await self.ws.send_json( + { + "type": "callback_request", + "requestId": request_id, + "rendererId": self.renderer_id, + "payload": _payload(n), + } + ) + payload = await fut + if not isinstance(payload, dict) or payload.get("status") == "error": + raise ValueError(f"unexpected response {payload!r:.200}") + + async def _ensure_ws(self): + """Connect before timing a click: the renderer connects once and + reuses the socket, so connecting is not part of a round trip. + Returns False (counted as an error) if it can't connect.""" + if self.ws is not None and not self.ws.closed: + return True + try: + await asyncio.wait_for(self._ws_connect(), timeout=TIMEOUT) + return True + except (aiohttp.ClientError, asyncio.TimeoutError): + if self.stats.in_window(time.time()): + self.stats.errors += 1 + await asyncio.sleep(1) + return False + + async def _click(self, call, n): + clicked = time.time() + try: + await asyncio.wait_for(call(n), timeout=TIMEOUT) + except (aiohttp.ClientError, asyncio.TimeoutError, ConnectionError, ValueError): + if self.stats.in_window(clicked): + self.stats.errors += 1 + if self.ws is not None: + await self.ws.close() + await asyncio.sleep(1) + return + if self.stats.in_window(clicked): + self.stats.calls += 1 + self.stats.rtt_ms.append((time.time() - clicked) * 1000) + + async def run(self, start_at, stop_at): + await asyncio.sleep(max(0.0, start_at - time.time())) + call = self._ws_call if self.transport == "ws" else self._http_call + try: + async with self.session.get(self.base + "/") as resp: + resp.raise_for_status() + await resp.read() + n = 0 + while time.time() < stop_at: + n += 1 + if self.transport == "ws" and not await self._ensure_ws(): + continue + await self._click(call, n) + await asyncio.sleep(random.uniform(0, 2 * self.think)) + except (aiohttp.ClientError, asyncio.TimeoutError): + # The page never loaded: this browser never got to click at all. + self.stats.errors += 1 + self.stats.failed_browsers += 1 + finally: + if self.ws is not None: + await self.ws.close() + for reader in self.readers: + reader.cancel() + await self.session.close() + + +async def main(args): + stats = Stats(args.start + args.ramp, args.start + args.ramp + args.duration) + browsers = [ + Browser(args.url, stats, args.transport, args.think) + for _ in range(args.browsers) + ] + await asyncio.gather( + *( + b.run(args.start + args.ramp * i / max(1, args.browsers), stats.measure_to) + for i, b in enumerate(browsers) + ) + ) + json.dump( + { + "rtt_ms": [round(x, 2) for x in stats.rtt_ms], + "calls": stats.calls, + "errors": stats.errors, + "failed_browsers": stats.failed_browsers, + }, + sys.stdout, + ) + + +if __name__ == "__main__": + parser = argparse.ArgumentParser() + parser.add_argument("--url", required=True) + parser.add_argument("--browsers", type=int, required=True) + parser.add_argument("--transport", choices=["http", "ws"], default="http") + parser.add_argument("--think", type=float, default=1.0) + parser.add_argument("--start", type=float, required=True) + parser.add_argument("--ramp", type=float, default=5.0) + parser.add_argument("--duration", type=float, default=20.0) + asyncio.run(main(parser.parse_args())) diff --git a/benchmarks/callbacks/load.py b/benchmarks/callbacks/load.py new file mode 100644 index 0000000000..85b68e6834 --- /dev/null +++ b/benchmarks/callbacks/load.py @@ -0,0 +1,150 @@ +"""Callback load test: plain HTTP callbacks vs websocket callbacks, by backend. + +For each server setup (backend x server x workers x transport) and each +browser count, start the server, run simulated browsers that click and wait +for the callback (``client.py``), and record the round trip, throughput, +errors and CPU. Flask only has HTTP; FastAPI and Quart run both transports. +A setup stops climbing once it saturates (p95 over ``--max-p95`` ms or errors +over 1%). + + python -m benchmarks.callbacks.load --out benchmarks/callbacks/results.json + python -m benchmarks.callbacks.load --setup fastapi-uvicorn-w1-ws --browsers 100 500 + python -m benchmarks.callbacks.load --kind async --work-ms 50 + +Plain callbacks keep no state between requests, so multi-worker setups need +no shared storage. +""" +from __future__ import annotations + +import argparse +import math +import os +import sys + +from benchmarks import loadkit +from benchmarks.loadkit import Setup, pct + +HERE = os.path.dirname(os.path.abspath(__file__)) + + +def _setups(): + servers = [ + ("flask", "gunicorn", "Flask, gunicorn", " x 8 threads", ["http"]), + ("fastapi", "uvicorn", "FastAPI, uvicorn", "", ["http", "ws"]), + ("quart", "uvicorn", "Quart, uvicorn", "", ["http", "ws"]), + ] + setups = [] + for backend, server, label, threads, transports in servers: + for workers in (1, 4): + for transport in transports: + plural = "s" if workers > 1 else "" + setups.append( + Setup( + f"{backend}-{server}-w{workers}-{transport}", + backend, + server, + workers, + f"{label} {workers} worker{plural}{threads}, " + + ("websocket" if transport == "ws" else "HTTP"), + env={"LOAD_TRANSPORT": transport}, + ) + ) + return setups + + +SETUPS = _setups() +SETUPS_BY_NAME = {s.name: s for s in SETUPS} +COUNTS = ("calls", "errors", "failed_browsers") + + +def _failed(report, share): + report.update(errors=share, failed_browsers=share) + + +def run_point(setup, browsers, args): + env = { + "LOAD_BACKEND": setup.backend, + "LOAD_KIND": args.kind, + "LOAD_WORK_MS": str(args.work_ms), + } + with loadkit.serve(setup, HERE, "load_app", env, args.server_cpus) as ( + url, + server, + ): + reports, resources = loadkit.run_clients( + url, + server, + browsers, + "benchmarks.callbacks.client", + [ + "--transport", + setup.env["LOAD_TRANSPORT"], + "--think", + str(args.think), + ], + args, + _failed, + # The last click may hit the client's timeout, its 1 s backoff, + # then a full think. + grace=30 + 1 + 2 * args.think + 10, + ) + rtt = [x for r in reports for x in r.get("rtt_ms", [])] + counts = {k: sum(r.get(k, 0) for r in reports) for k in COUNTS} + attempts = counts["calls"] + counts["errors"] + error_rate = counts["errors"] / attempts if attempts else 1.0 + point = { + "browsers": browsers, + "rtt_p50_ms": pct(rtt, 0.5), + "rtt_p95_ms": pct(rtt, 0.95), + "rtt_p99_ms": pct(rtt, 0.99), + "calls_per_s": round(counts["calls"] / args.duration, 1), + **counts, + "error_rate": round(error_rate, 4), + **resources, + } + # Browsers that never loaded took their load away from the point. + point["saturated"] = bool( + error_rate > 0.01 + or (point["rtt_p95_ms"] or math.inf) > args.max_p95 + or counts["failed_browsers"] + ) + return point + + +def _entry_fields(setup): + return {"transport": setup.env["LOAD_TRANSPORT"]} + + +def _describe(point): + return ( + f"p50 {point['rtt_p50_ms']}ms p95 {point['rtt_p95_ms']}ms" + f" {point['calls_per_s']} calls/s" + ) + + +def main(argv=None): + parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) + parser.add_argument("--setup", nargs="*", choices=list(SETUPS_BY_NAME)) + loadkit.add_common_args(parser, [25, 50, 100, 250, 500, 1000, 2000, 4000]) + parser.add_argument("--kind", choices=["sync", "async"], default="sync") + parser.add_argument("--work-ms", type=float, default=0.0, help="work per callback") + parser.add_argument("--out", default=os.path.join(HERE, "results.json")) + args = parser.parse_args(argv) + loadkit.configure(args) + + setups = [SETUPS_BY_NAME[n] for n in args.setup] if args.setup else SETUPS + results = { + "kind": "callback_load", + "meta": loadkit.machine_info(args), + "params": { + k: getattr(args, k) + for k in ("kind", "work_ms", "think", "ramp", "duration", "max_p95") + }, + "setups": [], + } + loadkit.sweep(results, setups, args, run_point, _describe, _entry_fields) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/benchmarks/callbacks/load_app.py b/benchmarks/callbacks/load_app.py new file mode 100644 index 0000000000..f1ad99b711 --- /dev/null +++ b/benchmarks/callbacks/load_app.py @@ -0,0 +1,53 @@ +"""The app the callback load test serves. + +One button and one output: each click runs a callback that returns the +click count. ``LOAD_TRANSPORT`` picks how the renderer sends it: ``http`` (a +POST per callback) or ``ws`` (``websocket_callbacks=True``, one websocket per +browser). ``LOAD_KIND`` makes the callback a plain ``def`` (``sync``) or an +``async def``, and ``LOAD_WORK_MS`` adds that much work to each call: CPU for +a sync callback, ``asyncio.sleep`` for an async one, like a query. + +Configured from the environment so gunicorn and uvicorn can import it: +``LOAD_BACKEND`` (flask, fastapi, quart) plus the above. +""" +import asyncio +import os +import time + +from dash import Dash, Input, Output, html + +BACKEND = os.environ.get("LOAD_BACKEND", "flask") +TRANSPORT = os.environ.get("LOAD_TRANSPORT", "http") +KIND = os.environ.get("LOAD_KIND", "sync") +WORK = float(os.environ.get("LOAD_WORK_MS", "0")) / 1000 + +app = Dash(__name__, backend=BACKEND, websocket_callbacks=TRANSPORT == "ws") +app.layout = html.Div([html.Button("go", id="btn", n_clicks=0), html.Div(id="out")]) + + +def _burn(seconds): + end = time.perf_counter() + seconds + spins = 0 + while time.perf_counter() < end: + spins += 1 + return spins + + +if KIND == "async": + + @app.callback(Output("out", "children"), Input("btn", "n_clicks")) + async def clicked(n): + if WORK: + await asyncio.sleep(WORK) + return f"clicked {n}" + +else: + + @app.callback(Output("out", "children"), Input("btn", "n_clicks")) + def clicked(n): + if WORK: + _burn(WORK) + return f"clicked {n}" + + +server = app.server diff --git a/benchmarks/loadkit.py b/benchmarks/loadkit.py new file mode 100644 index 0000000000..e389321877 --- /dev/null +++ b/benchmarks/loadkit.py @@ -0,0 +1,424 @@ +"""Shared plumbing for the load tests (``streaming/``, ``callbacks/``). + +Starts a server setup on its own CPUs, runs simulated-browser client +processes against it on the other CPUs, samples CPU and memory during the +measure window, and collects what the clients report. Each load test brings +its own app, client and metrics. +""" +from __future__ import annotations + +import contextlib +import json +import math +import os +import platform +import resource +import socket +import statistics +import subprocess +import sys +import tempfile +import time +from dataclasses import asdict, dataclass, field +from typing import Callable, Dict, List + +import psutil +import requests + +REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +# One Python client process saturates its own CPU at a few hundred browsers. +BROWSERS_PER_CLIENT = 100 + + +@dataclass +class Setup: + name: str + backend: str + server: str + workers: int + label: str + env: Dict[str, str] = field(default_factory=dict) + + @property + def needs_redis(self): + return self.workers > 1 + + +def free_port(): + with socket.socket() as sock: + sock.bind(("127.0.0.1", 0)) + return sock.getsockname()[1] + + +def parse_cpus(spec): + cpus = set() + for part in spec.split(","): + lo, _, hi = part.partition("-") + cpus.update(range(int(lo), int(hi or lo) + 1)) + return sorted(cpus) + + +def configure(args): + """Split the machine between server and clients, raise the fd limit.""" + # Every browser holds sockets in both the server and its client process. + _soft, hard = resource.getrlimit(resource.RLIMIT_NOFILE) + resource.setrlimit(resource.RLIMIT_NOFILE, (hard, hard)) + cores = psutil.cpu_count() + args.server_cpus = ( + parse_cpus(args.server_cpus) if args.server_cpus else list(range(cores // 2)) + ) + args.client_cpus = [ + c for c in range(cores) if c not in args.server_cpus + ] or args.server_cpus + + +def server_cmd(setup, app, port): + py = sys.executable + if setup.server == "gunicorn": + return [ + py, + "-m", + "gunicorn", + f"{app}:server", + "--bind", + f"127.0.0.1:{port}", + "--workers", + str(setup.workers), + "--worker-class", + "gthread", + "--threads", + "8", + "--timeout", + "120", + "--backlog", + "4096", + ] + return [ + py, + "-m", + "uvicorn", + f"{app}:server", + "--host", + "127.0.0.1", + "--port", + str(port), + "--workers", + str(setup.workers), + "--log-level", + "warning", + "--backlog", + "4096", + ] + + +@contextlib.contextmanager +def serve(setup, app_dir, app, env, server_cpus): + """Run ``app_dir/app.py``'s ``server`` under the setup's server and yield + ``(url, process)`` once every worker is up.""" + port = free_port() + env = { + **os.environ, + **setup.env, + **env, + # The server runs from app_dir: without this it would import whichever + # dash is installed, not the checkout under test. + "PYTHONPATH": os.pathsep.join( + p for p in (REPO_ROOT, os.environ.get("PYTHONPATH")) if p + ), + } + proc = subprocess.Popen( # pylint: disable=consider-using-with + server_cmd(setup, app, port), + cwd=app_dir, + env=env, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + start_new_session=True, + ) + with contextlib.suppress(AttributeError): + psutil.Process(proc.pid).cpu_affinity(server_cpus) + url = f"http://127.0.0.1:{port}" + try: + deadline = time.monotonic() + 60 + while time.monotonic() < deadline: + if proc.poll() is not None: + raise RuntimeError(f"{setup.name} exited with {proc.returncode}") + with contextlib.suppress(requests.RequestException): + if requests.get(url, timeout=1).ok: + break + time.sleep(0.3) + else: + raise RuntimeError(f"{setup.name} never came up") + # The first worker answering does not mean the others are up yet; + # wait for all of them so CPU, memory and pinning cover every worker. + # uvicorn runs a single worker in the main process, without forking. + forks = setup.workers if setup.server == "gunicorn" or setup.workers > 1 else 0 + while ( + len(psutil.Process(proc.pid).children(recursive=True)) < forks + and time.monotonic() < deadline + ): + time.sleep(0.3) + time.sleep(1) + for child in psutil.Process(proc.pid).children(recursive=True): + with contextlib.suppress(psutil.Error, AttributeError): + child.cpu_affinity(server_cpus) + yield url, proc + finally: + with contextlib.suppress(ProcessLookupError): + os.killpg(proc.pid, 15) + try: + proc.wait(timeout=15) + except subprocess.TimeoutExpired: + with contextlib.suppress(ProcessLookupError): + os.killpg(proc.pid, 9) + proc.wait() + + +def pct(values, q): + if not values: + return None + values = sorted(values) + return round(values[min(len(values) - 1, math.ceil(q * len(values)) - 1)], 1) + + +def _tree(pid): + proc = psutil.Process(pid) + return [proc] + proc.children(recursive=True) + + +def _start_clients(module, url, browsers, client_args, client_cpus): + n_clients = max(1, math.ceil(browsers / BROWSERS_PER_CLIENT)) + clients = [] + for i in range(n_clients): + share = browsers // n_clients + (1 if i < browsers % n_clients else 0) + # Files, not pipes: nothing reads a pipe until the clients exit, and a + # full pipe would block them forever. + out = tempfile.TemporaryFile() # pylint: disable=consider-using-with + err = tempfile.TemporaryFile() # pylint: disable=consider-using-with + proc = subprocess.Popen( # pylint: disable=consider-using-with + [ + sys.executable, + "-m", + module, + "--url", + url, + "--browsers", + str(share), + *client_args, + ], + cwd=REPO_ROOT, + stdout=out, + stderr=err, + ) + with contextlib.suppress(psutil.Error, AttributeError): + psutil.Process(proc.pid).cpu_affinity(client_cpus) + clients.append((proc, share, out, err)) + return clients + + +def _sample(server_procs, client_procs, n_client_cpus): + server, mem = 0.0, 0 + for p in server_procs: + with contextlib.suppress(psutil.Error): + server += p.cpu_percent(None) + mem += p.memory_info().rss + peak, total = 0.0, 0.0 + for p in client_procs: + with contextlib.suppress(psutil.Error): + cpu = p.cpu_percent(None) + peak = max(peak, cpu) + total += cpu + # Many client processes sharing few CPUs each look idle on their own; + # what matters is how busy the client CPUs are as a whole. + return server / 100, mem, max(peak, total / n_client_cpus) + + +def _collect(clients, on_failure: Callable[[dict, int], None]): + results = [] + for proc, share, out, err in clients: + out.seek(0) + try: + result = json.loads(out.read()) + except ValueError: + result = None + if proc.returncode != 0 or result is None: + err.seek(0) + tail = err.read().decode(errors="replace").strip().splitlines()[-1:] + print(f" client exited {proc.returncode}: {' '.join(tail)}", flush=True) + failed: dict = {} + on_failure(failed, share) + results.append(failed) + else: + results.append(result) + out.close() + err.close() + return results + + +def run_clients( + url, + server, + browsers, + module, + client_args, + args, + on_failure, + grace=90, +): + """Run ``browsers`` simulated browsers (``module``) against ``url``. + + The clients start together, ramp up over ``args.ramp`` seconds, then + measure for ``args.duration``. Returns the clients' JSON reports and the + server/client resource figures sampled inside the measure window. A + crashed client is reported through ``on_failure(report, its_browsers)`` + so its whole share counts as failed and a point can't look healthy with + part of its load missing. + """ + start = time.time() + 2 + measure_from = start + args.ramp + measure_to = measure_from + args.duration + clients = _start_clients( + module, + url, + browsers, + [ + *client_args, + "--start", + str(start), + "--ramp", + str(args.ramp), + "--duration", + str(args.duration), + ], + args.client_cpus, + ) + try: + server_procs = _tree(server.pid) + client_procs = [psutil.Process(c[0].pid) for c in clients] + for p in server_procs + client_procs: + with contextlib.suppress(psutil.Error): + p.cpu_percent(None) + server_cpu: List[float] = [] + client_cpu: List[float] = [] + rss: List[int] = [] + # Work started in the window may run up to the client's own timeout + # past it; anything later is hung. + hard_deadline = measure_to + grace + while any(c[0].poll() is None for c in clients): + if time.time() > hard_deadline: + raise RuntimeError(f"clients still running {grace}s after the window") + time.sleep(1) + if not measure_from <= time.time() <= measure_to: + continue + cores, mem, client = _sample( + server_procs, client_procs, len(args.client_cpus) + ) + server_cpu.append(cores) + rss.append(mem) + client_cpu.append(client) + finally: + for proc, *_ in clients: + if proc.poll() is None: + proc.kill() + proc.wait() + resources = { + "server_cpu_cores": round(statistics.mean(server_cpu), 2) + if server_cpu + else None, + "server_rss_mb": round(max(rss) / 2**20) if rss else None, + "client_cpu_peak_pct": round(max(client_cpu)) if client_cpu else None, + } + # Clients near 100% CPU were the bottleneck, not the server. + resources["client_bound"] = (resources["client_cpu_peak_pct"] or 0) > 85 + return _collect(clients, on_failure), resources + + +def machine_info(args): + cpu = platform.processor() + with contextlib.suppress(OSError), open("/proc/cpuinfo", encoding="utf-8") as f: + for line in f: + if line.startswith("model name"): + cpu = line.split(":", 1)[1].strip() + break + sha = subprocess.run( + ["git", "rev-parse", "--short", "HEAD"], + cwd=REPO_ROOT, + capture_output=True, + text=True, + check=False, + ).stdout.strip() + import dash # pylint: disable=import-outside-toplevel + + return { + "date": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), + "commit": sha, + "dash_version": dash.__version__, + "python": platform.python_version(), + "cpu": cpu, + "cores": psutil.cpu_count(), + "memory_gb": round(psutil.virtual_memory().total / 2**30), + "server_cpus": len(args.server_cpus), + "client_cpus": len(args.client_cpus), + "runner": os.environ.get("BENCH_RUNNER", "local"), + } + + +def sweep(results, setups, args, run_point, describe, entry_fields): + """Climb each setup through ``args.browsers``, saving ``results`` to + ``args.out`` after every point. + + ``entry_fields(setup)`` gives the setup's extra result fields, or None to + skip it; ``describe(point)`` gives the test's own numbers for the log line. + A setup stops climbing once a point saturates or is client-bound: a + saturated server only gets worse, and client-bound numbers are the load + generator's, not the server's. + """ + for setup in setups: + fields = entry_fields(setup) + if fields is None: + continue + entry = { + **{k: v for k, v in asdict(setup).items() if k != "env"}, + **fields, + "points": [], + } + results["setups"].append(entry) + _climb(setup, entry, results, args, run_point, describe) + + +def _climb(setup, entry, results, args, run_point, describe): + for browsers in sorted(args.browsers): + try: + point = run_point(setup, browsers, args) + except RuntimeError as err: + print(f"{setup.name} @ {browsers}: {err}", flush=True) + return + entry["points"].append(point) + _log(setup, browsers, point, describe) + with open(args.out, "w", encoding="utf-8") as f: + json.dump(results, f, indent=1) + if point["saturated"] or point["client_bound"]: + return + + +def _log(setup, browsers, point, describe): + flags = (" SATURATED" if point["saturated"] else "") + ( + " CLIENT-BOUND" if point["client_bound"] else "" + ) + print( + f"{setup.name:24} {browsers:5} browsers {describe(point)}" + f" err {point['error_rate']:.2%} server {point['server_cpu_cores']} cores" + f" client peak {point['client_cpu_peak_pct']}%{flags}", + flush=True, + ) + + +def add_common_args(parser, browsers): + parser.add_argument("--browsers", nargs="*", type=int, default=browsers) + parser.add_argument( + "--think", type=float, default=1.0, help="mean seconds between actions" + ) + parser.add_argument("--ramp", type=float, default=10.0) + parser.add_argument("--duration", type=float, default=20.0) + parser.add_argument("--max-p95", type=float, default=1000.0) + parser.add_argument("--redis-url") + parser.add_argument("--server-cpus", help="e.g. 0-3 (default: half the machine)") diff --git a/benchmarks/publish.py b/benchmarks/publish.py new file mode 100644 index 0000000000..bfe28cdf07 --- /dev/null +++ b/benchmarks/publish.py @@ -0,0 +1,1159 @@ +"""Turn benchmark results into the numbers the docs and the README show. + + python -m benchmarks.publish \\ + --renderer benchmarks/results.json \\ + --streaming benchmarks/streaming/results.json \\ + --callbacks benchmarks/callbacks/results.json \\ + --serialization benchmarks/serialization/results.json + +Writes to ``benchmarks/results/`` (``--dir``), which is committed: + +- ``-latest.json`` and ``-history.json``: the raw numbers, and one + summary per run so trends can be charted +- ``report.json``: every chart as a Plotly figure and the numbers as tables, + in sections; the docs page renders it + +and rewrites the table between the ``benchmarks`` markers in ``README.md`` +(``--readme``). Any input may be omitted; the committed numbers for it stay. +With ``--only-significant`` (CI), an input is committed only if it moves the +published numbers (see ``load_changes`` and ``median_changes``), and nothing +is written when none does. +""" +from __future__ import annotations + +import argparse +import json +import math +import os +import sys +import time +from datetime import datetime, timedelta + +HERE = os.path.dirname(os.path.abspath(__file__)) +REPO_ROOT = os.path.dirname(HERE) +HISTORY_LIMIT = 200 + +# Interactive feel: a p95 frame delivery or callback round trip under this +# counts as "served". +CAPACITY_P95_MS = 250 + +BACKEND_HUE = {"flask": 1, "fastapi": 2, "quart": 3} +WORKER_DASH = {1: "solid", 4: "dash"} +TRANSPORT_HUE = {"http": 1, "ws": 2} +TRANSPORT_DASH = {"http": "solid", "ws": "dash"} +TRANSPORT_NAME = {"http": "HTTP", "ws": "websocket"} +BROWSERS_AXIS = "concurrent browsers" + + +def _load(path, default): + if path and os.path.exists(path): + with open(path, encoding="utf-8") as f: + return json.load(f) + return default + + +def _read(path): + """An input the caller named: a missing file is a mistake, not a skip.""" + if path is None: + return None + with open(path, encoding="utf-8") as f: + return json.load(f) + + +def _dump(path, data): + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w", encoding="utf-8") as f: + json.dump(data, f, indent=1) + f.write("\n") + + +def _append(history, entry, key): + history = [h for h in history if h.get(key) != entry.get(key)] + history.append(entry) + return history[-HISTORY_LIMIT:] + + +# --- streaming -------------------------------------------------------------- + + +def capacity(setup, metric="latency_p95_ms"): + """Highest measured browser count the setup served interactively, and + whether the real limit is higher (the sweep ended, or the clients ran out + of CPU, before the server fell behind).""" + best = 0 + for p in setup["points"]: + if p["client_bound"]: + return best, True + if p["error_rate"] > 0.01 or (p[metric] or 1e9) > CAPACITY_P95_MS: + return best, False + best = p["browsers"] + return best, bool(setup["points"]) + + +def _setup_trace(setup, metric, unit, dash=None): + pts = [ + p for p in setup["points"] if not p["client_bound"] and p[metric] is not None + ] + return { + "name": setup["label"], + "hue": BACKEND_HUE.get(setup["backend"], 1), + "dash": dash or WORKER_DASH.get(setup["workers"], "dot"), + "x": [p["browsers"] for p in pts], + "y": [p[metric] for p in pts], + "unit": unit, + "saturated": [p["saturated"] for p in pts], + } + + +def streaming_charts(run): + meta, params = run["meta"], run["params"] + load = ( + f"each browser streams {params['frames']} frames {int(params['interval'] * 1000)} ms" + f" apart, then waits ~{params['think']:g} s and streams again" + ) + where = _where(meta) + setups = run["setups"] + caps = [(s, *capacity(s)) for s in setups] + caps.sort(key=lambda c: c[1]) + return [ + { + "id": "streaming-capacity", + "title": "Streaming callbacks: browsers served per server setup", + "subtitle": ( + f"Most concurrent browsers with p95 frame latency under {CAPACITY_P95_MS} ms" + f" and under 1% errors; {load}. A + means the sweep stopped before the" + " server did." + ), + "source": where, + "kind": "hbar", + "bars": [ + { + "label": s["label"], + "value": cap, + "text": f"{cap:,}{'+' if capped else ''}", + "hue": 1, + } + for s, cap, capped in caps + ], + "xaxis": {"title": BROWSERS_AXIS}, + }, + { + "id": "streaming-latency", + "title": "Streaming callbacks: p95 frame latency", + "subtitle": f"Time from the callback yielding a frame to the browser reading it; {load}.", + "source": where, + "kind": "lines", + "traces": [_setup_trace(s, "latency_p95_ms", "ms") for s in setups], + "xaxis": {"title": BROWSERS_AXIS, "type": "log"}, + "yaxis": {"title": "p95 latency (ms)", "type": "log"}, + "threshold": CAPACITY_P95_MS, + }, + { + "id": "streaming-first-frame", + "title": "Streaming callbacks: p95 time to first frame", + "subtitle": f"From the click (uplink POST) to the first frame on screen; {load}.", + "source": where, + "kind": "lines", + "traces": [_setup_trace(s, "first_frame_p95_ms", "ms") for s in setups], + "xaxis": {"title": BROWSERS_AXIS, "type": "log"}, + "yaxis": {"title": "p95 time to first frame (ms)", "type": "log"}, + }, + { + "id": "streaming-cpu", + "title": "Streaming callbacks: server CPU", + "subtitle": f"Average CPU used by the whole server process tree; {load}.", + "source": where, + "kind": "lines", + "traces": [_setup_trace(s, "server_cpu_cores", "cores") for s in setups], + "xaxis": {"title": BROWSERS_AXIS, "type": "log"}, + "yaxis": {"title": "CPU (cores)"}, + }, + ] + + +def streaming_table(run): + cols = [ + ("browsers", "browsers"), + ("latency_p50_ms", "p50 ms"), + ("latency_p95_ms", "p95 ms"), + ("latency_p99_ms", "p99 ms"), + ("first_frame_p95_ms", "first frame p95 ms"), + ("frames_per_s", "frames/s"), + ("error_rate", "errors"), + ("server_cpu_cores", "server cores"), + ("server_rss_mb", "server MB"), + ] + rows = [] + for s in run["setups"]: + for p in s["points"]: + rows.append( + [s["label"]] + [_fmt(p.get(k), k) for k, _ in cols] + [_note(p)] + ) + return ["setup"] + [c for _, c in cols] + ["note"], rows + + +def _note(point): + if point["client_bound"]: + return "client-bound" + return "saturated" if point["saturated"] else "" + + +def _fmt(v, key=""): + if v is None: + return "-" + if key == "error_rate": + return f"{v:.1%}" + if isinstance(v, float): + return f"{v:,.1f}" + if isinstance(v, int): + return f"{v:,}" + return str(v) + + +# --- callbacks ------------------------------------------------------------------ + +# Server CPU is compared where every setup still keeps up. +CPU_AT_BROWSERS = 1000 + + +def _where(meta): + return ( + f"Server on {meta['server_cpus']} of {meta['cores']} cores ({meta['cpu']})," + f" Dash {meta['dash_version']} ({meta['commit']}), measured {meta['date'][:10]}" + ) + + +def _callback_load(params): + work = f", each doing {params['work_ms']:g} ms of work" if params["work_ms"] else "" + return ( + f"each browser clicks, waits for the {params['kind']} callback{work}," + f" then waits ~{params['think']:g} s and clicks again" + ) + + +def callbacks_charts(run): + params, where = run["params"], _where(run["meta"]) + load = _callback_load(params) + setups = run["setups"] + caps = sorted(((s, *capacity(s, "rtt_p95_ms")) for s in setups), key=lambda c: c[1]) + cpu = [] + for s in setups: + point = next( + ( + p + for p in s["points"] + if p["browsers"] == CPU_AT_BROWSERS and not p["client_bound"] + ), + None, + ) + if point and point["calls_per_s"] and point["server_cpu_cores"] is not None: + cpu.append((s, point["server_cpu_cores"] / point["calls_per_s"] * 1000)) + cpu.sort(key=lambda c: c[1]) + charts = [ + { + "id": "callbacks-capacity", + "title": "Callbacks: browsers served, HTTP vs websocket", + "subtitle": ( + f"Most concurrent browsers with p95 round trip under {CAPACITY_P95_MS} ms" + f" and under 1% errors; {load}. A + means the sweep stopped before the" + " server did." + ), + "source": where, + "kind": "hbar", + "bars": [ + { + "label": s["label"], + "value": cap, + "text": f"{cap:,}{'+' if capped else ''}", + "hue": TRANSPORT_HUE.get(s["transport"], 1), + } + for s, cap, capped in caps + ], + "legend": [ + {"label": TRANSPORT_NAME[t], "hue": hue} + for t, hue in TRANSPORT_HUE.items() + ], + "xaxis": {"title": BROWSERS_AXIS}, + } + ] + for workers in sorted({s["workers"] for s in setups}): + plural = "s" if workers > 1 else "" + charts.append( + { + "id": f"callbacks-rtt-w{workers}", + "title": f"Callbacks: p95 round trip, {workers} worker{plural}", + "subtitle": ( + "From sending the callback to having its response, HTTP solid," + f" websocket dashed; {load}." + ), + "source": where, + "kind": "lines", + "traces": [ + _setup_trace( + s, + "rtt_p95_ms", + "ms", + dash=TRANSPORT_DASH.get(s["transport"], "dot"), + ) + for s in setups + if s["workers"] == workers + ], + "xaxis": {"title": BROWSERS_AXIS, "type": "log"}, + "yaxis": {"title": "p95 round trip (ms)", "type": "log"}, + "threshold": CAPACITY_P95_MS, + } + ) + if cpu: + charts.append( + { + "id": "callbacks-cpu", + "title": "Callbacks: server CPU per 1,000 calls a second", + "subtitle": ( + f"Server CPU at {CPU_AT_BROWSERS:,} browsers divided by the calls" + f" it served; lower is cheaper; {load}." + ), + "source": where, + "kind": "hbar", + "bars": [ + { + "label": s["label"], + "value": round(cores, 2), + "text": f"{cores:.2f} cores", + "hue": TRANSPORT_HUE.get(s["transport"], 1), + } + for s, cores in cpu + ], + "legend": [ + {"label": TRANSPORT_NAME[t], "hue": hue} + for t, hue in TRANSPORT_HUE.items() + ], + "xaxis": {"title": "cores per 1,000 calls/s", "rangemode": "tozero"}, + } + ) + return charts + + +def callbacks_table(run): + cols = [ + ("browsers", "browsers"), + ("rtt_p50_ms", "p50 ms"), + ("rtt_p95_ms", "p95 ms"), + ("rtt_p99_ms", "p99 ms"), + ("calls_per_s", "calls/s"), + ("error_rate", "errors"), + ("server_cpu_cores", "server cores"), + ("server_rss_mb", "server MB"), + ] + rows = [] + for s in run["setups"]: + for p in s["points"]: + rows.append( + [s["label"]] + [_fmt(p.get(k), k) for k, _ in cols] + [_note(p)] + ) + return ["setup"] + [c for _, c in cols] + ["note"], rows + + +# --- renderer ----------------------------------------------------------------- + + +def _descriptions(): + sys.path.insert(0, REPO_ROOT) + try: + # pylint: disable-next=import-outside-toplevel + from benchmarks.scenarios import SCENARIOS + + return {name: sc.description for name, sc in SCENARIOS.items()} + except Exception as err: # pylint: disable=broad-except + print(f"warning: no scenario descriptions ({err!r})", file=sys.stderr) + return {} + + +def _median_series(history, name): + """One median trend per (outer, inner) key of each run's results.""" + series = {} + for h in history: + for outer, inner in h["results"].items(): + for key, stats in inner.items(): + s = series.setdefault( + name(outer, key), {"x": [], "y": [], "commit": []} + ) + s["x"].append(h["date"]) + s["y"].append(stats["median"]) + s["commit"].append(h["commit"]) + return series + + +def renderer_charts(entry, history): + desc = _descriptions() + rows = [] + for scenario, metrics in entry["results"].items(): + for metric, stats in metrics.items(): + if stats["median"] < 5: + continue + rows.append((f"{scenario} ({metric})", stats, desc.get(scenario, ""))) + rows.sort(key=lambda r: r[1]["median"]) + where = ( + f"Headless Chrome, production bundle, {entry['runner']}; commit" + f" {entry['commit']}, measured {entry['date'][:10]}" + ) + series = _median_series(history, lambda scenario, metric: f"{scenario} ({metric})") + order = [r[0] for r in reversed(rows)] + return [ + { + "id": "renderer-latest", + "title": "Renderer: median time per interaction", + "subtitle": ( + "In-page performance.now() timings: server round trip, patch apply and" + " React render. Hover a bar for p90 and what it measures." + ), + "source": where, + "kind": "hbar", + "bars": [ + { + "label": label, + "value": stats["median"], + "text": f"{stats['median']:,.0f} ms", + "hover": f"{d}
median {stats['median']:,.1f} ms, p90 {stats['p90']:,.1f} ms," + f" growth {stats.get('growth', 1):.2f}x", + "hue": 1, + } + for label, stats, d in rows + ], + "xaxis": {"title": "median ms (log)", "type": "log"}, + }, + { + "id": "renderer-trend", + "title": "Renderer: median over time", + "subtitle": "One point per published run on the dev branch. Pick a scenario.", + "source": where, + "kind": "picker", + "series": {k: series[k] for k in order if k in series}, + "yaxis": {"title": "median ms", "rangemode": "tozero"}, + }, + ] + + +def renderer_table(entry): + desc = _descriptions() + rows = [] + for scenario, metrics in entry["results"].items(): + for metric, s in metrics.items(): + rows.append( + [ + scenario, + metric, + _fmt(s["median"]), + _fmt(s["p90"]), + _fmt(s["max"]), + f"{s.get('growth', 1):.2f}x", + desc.get(scenario, ""), + ] + ) + return [ + "scenario", + "metric", + "median ms", + "p90 ms", + "max ms", + "growth", + "what", + ], rows + + +# --- serialization -------------------------------------------------------------- + +ENGINE_NAME = {"orjson": "with orjson", "json": "without orjson (stdlib json)"} + + +def _figure_only(case): + return case.split("+")[0] + "+none" + + +def _serialization_where(entry): + v = entry["versions"] + return ( + f"{entry['traces']} traces x {entry['points']:,} points, median of" + f" {entry['repeat']}; Dash {v['dash']}, plotly {v['plotly']}, orjson" + f" {v['orjson']}, Python {v['python']}, {entry['runner']}; commit" + f" {entry['commit']}, measured {entry['date'][:10]}" + ) + + +def serialization_charts(entry, history): + labels = entry["labels"] + where = _serialization_where(entry) + charts = [] + for engine, name in ENGINE_NAME.items(): + cases = [c for c, by in entry["results"].items() if engine in by] + if not cases: + continue + bars = [] + for case in reversed(cases): + stats = entry["results"][case][engine] + ref = entry["results"].get(_figure_only(case), {}).get(engine) + ratio = ( + f", {stats['median'] / ref['median']:.1f}x the figure alone" + if ref + else "" + ) + bars.append( + { + "label": labels.get(case, case), + "value": stats["median"], + "text": f"{stats['median']:,.0f} ms", + "hover": f"{labels.get(case, case)}
median {stats['median']:,.1f} ms," + f" min {stats['min']:,.1f} ms{ratio}", + "hue": 1 if case.endswith("+none") else 2, + } + ) + charts.append( + { + "id": f"serialization-{engine}", + "title": f"Serializing a callback response, {name}", + "subtitle": ( + "The same figure alone, then with one small value orjson cannot" + " take as is. Shorter is better." + ), + "source": where, + "kind": "hbar", + "bars": bars, + "legend": [ + {"hue": 1, "label": "figure alone"}, + {"hue": 2, "label": "same figure, plus one non-native value"}, + ], + "xaxis": {"title": "median ms (log)", "type": "log"}, + } + ) + series = _median_series( + history, lambda case, engine: f"{labels.get(case, case)}, {ENGINE_NAME[engine]}" + ) + charts.append( + { + "id": "serialization-trend", + "title": "Serialization: median over time", + "subtitle": "One point per published run on the dev branch. Pick a case.", + "source": where, + "kind": "picker", + "series": series, + "yaxis": {"title": "median ms", "rangemode": "tozero"}, + } + ) + return charts + + +def serialization_table(entry): + rows = [] + for case, by_engine in entry["results"].items(): + row = [entry["labels"].get(case, case)] + for engine in ENGINE_NAME: + stats = by_engine.get(engine) + ref = entry["results"].get(_figure_only(case), {}).get(engine) + row.append(_fmt(stats["median"]) if stats else "-") + row.append( + f"{stats['median'] / ref['median']:.1f}x" if stats and ref else "-" + ) + row.append(_fmt(next(iter(by_engine.values()))["bytes"])) + rows.append(row) + return [ + "case", + "orjson ms", + "orjson vs figure alone", + "stdlib json ms", + "stdlib json vs figure alone", + "bytes", + ], rows + + +# --- report ----------------------------------------------------------------------- + +# A light theme; backgrounds stay transparent so the figures sit on any page. +SERIES = {1: "#2a78d6", 2: "#eb6834", 3: "#1baf7a"} +INK, INK_2, MUTED = "#0b0b0b", "#52514e", "#898781" +GRID, AXIS = "#e1e0d9", "#c3c2b7" +LEGEND = { + "orientation": "h", + "x": 0, + "y": 1.02, + "xanchor": "left", + "yanchor": "bottom", + "font": {"size": 12, "color": INK_2}, +} + + +def _axis(spec, **extra): + spec = spec or {} + log = spec.get("type") == "log" + axis = { + "title": {"text": spec.get("title", ""), "font": {"size": 12, "color": INK_2}}, + "type": spec.get("type", "linear"), + "rangemode": spec.get("rangemode", "normal"), + "gridcolor": GRID, + "linecolor": AXIS, + "zeroline": False, + "tickfont": {"size": 11, "color": MUTED}, + "automargin": True, + } + if log: + # Label the decades only (1, 10, 100, ...). + axis["dtick"] = 1 + axis.update(extra) + return axis + + +def _layout(spec): + return { + "paper_bgcolor": "rgba(0,0,0,0)", + "plot_bgcolor": "rgba(0,0,0,0)", + "font": { + "family": 'system-ui, -apple-system, "Segoe UI", sans-serif', + "color": INK, + }, + "margin": {"l": 8, "r": 16, "t": 8, "b": 8}, + "xaxis": _axis(spec.get("xaxis")), + "yaxis": _axis(spec.get("yaxis")), + "showlegend": False, + } + + +def _marker(hue): + return {"size": 8, "color": SERIES[hue], "line": {"width": 2, "color": "#fff"}} + + +def _lines(spec): + data = [ + { + "type": "scatter", + "mode": "lines+markers", + "name": t["name"], + "x": t["x"], + "y": t["y"], + "line": {"color": SERIES[t["hue"]], "width": 2, "dash": t["dash"]}, + "marker": _marker(t["hue"]), + "hovertemplate": f"%{{x:,}} browsers
%{{y:,.1f}} {t['unit']}" + f"{t['name']}", + } + for t in spec["traces"] + ] + layout = _layout(spec) + # Label the x axis at the measured counts rather than log minor ticks. + xs = sorted({x for t in spec["traces"] for x in t["x"]}) + if xs: + layout["xaxis"].pop("dtick", None) + layout["xaxis"]["tickvals"] = xs + layout["xaxis"]["ticktext"] = [f"{x:,}" for x in xs] + layout.update(showlegend=True, legend=LEGEND, hovermode="closest") + threshold = spec.get("threshold") + if threshold: + log = spec.get("yaxis", {}).get("type") == "log" + layout["shapes"] = [ + { + "type": "line", + "xref": "paper", + "x0": 0, + "x1": 1, + "y0": threshold, + "y1": threshold, + "line": {"color": MUTED, "width": 1}, + } + ] + layout["annotations"] = [ + { + "xref": "paper", + "x": 0, + "yref": "y", + # Annotations on a log axis are placed in log units. + "y": math.log10(threshold) if log else threshold, + "xanchor": "left", + "yanchor": "bottom", + "showarrow": False, + "text": f"{threshold} ms", + "font": {"size": 11, "color": MUTED}, + } + ] + return {"data": data, "layout": layout} + + +def _hbar(spec): + bars = spec["bars"] + names = {item["hue"]: item["label"] for item in spec.get("legend", [])} + data = [] + # One trace per colour so a legend can name them; overlaid, each bar keeps + # its full width on its own row. + for hue in sorted({b.get("hue", 1) for b in bars}): + group = [b for b in bars if b.get("hue", 1) == hue] + data.append( + { + "type": "bar", + "orientation": "h", + "name": names.get(hue, ""), + "y": [b["label"] for b in group], + "x": [b["value"] for b in group], + "text": [b["text"] for b in group], + "textposition": "outside", + "cliponaxis": False, + "textfont": {"size": 12, "color": INK_2}, + "marker": {"color": SERIES[hue]}, + "hovertext": [ + b.get("hover", f"{b['label']}: {b['text']}") for b in group + ], + "hoverinfo": "text", + } + ) + layout = _layout(spec) + layout["yaxis"] = _axis( + {"type": "category"}, + gridcolor="rgba(0,0,0,0)", + tickfont={"size": 12, "color": INK_2}, + categoryorder="array", + categoryarray=[b["label"] for b in bars], + ) + layout.update(bargap=0.35, barmode="overlay") + layout["margin"]["r"] = 64 + if spec.get("legend"): + layout.update(showlegend=True, legend=LEGEND) + return {"data": data, "layout": layout} + + +def _picker(spec): + """One trace per series and a dropdown that shows one at a time.""" + names = list(spec["series"]) + data = [] + for i, name in enumerate(names): + s = spec["series"][name] + data.append( + { + "type": "scatter", + "mode": "lines+markers", + "name": name, + "x": s["x"], + "y": s["y"], + "customdata": s["commit"], + "visible": i == 0, + "line": {"color": SERIES[1], "width": 2}, + "marker": _marker(1), + "hovertemplate": "%{x|%Y-%m-%d}
%{y:,.1f} ms
commit %{customdata}" + "", + } + ) + + def x_range(series): + if len(series["x"]) != 1: + return {"xaxis.autorange": True} + # One run so far: show a week around it, not a millisecond. + day = datetime.fromisoformat(series["x"][0].replace("Z", "+00:00")) + return { + "xaxis.range": [ + (day - timedelta(days=3)).isoformat(), + (day + timedelta(days=3)).isoformat(), + ] + } + + layout = _layout(spec) + layout["xaxis"].update(type="date", tickformat="%b %d") + if names: + first = spec["series"][names[0]] + if len(first["x"]) == 1: + layout["xaxis"]["range"] = x_range(first)["xaxis.range"] + layout["margin"]["t"] = 48 + layout["updatemenus"] = [ + { + "x": 0, + "y": 1.02, + "xanchor": "left", + "yanchor": "bottom", + "buttons": [ + { + "label": name, + "method": "update", + "args": [ + {"visible": [n == name for n in names]}, + x_range(spec["series"][name]), + ], + } + for name in names + ], + } + ] + return {"data": data, "layout": layout} + + +def figure(spec): + """The chart spec as a Plotly figure (``data`` and ``layout``).""" + return {"lines": _lines, "hbar": _hbar, "picker": _picker}[spec["kind"]](spec) + + +def _section(sid, title, intro, charts, table): + header, rows = table + return { + "id": sid, + "title": title, + "intro": intro, + "charts": [ + { + "id": c["id"], + "title": c["title"], + "subtitle": c["subtitle"], + "source": c["source"], + "figure": figure(c), + } + for c in charts + ], + "table": {"header": header, "rows": rows}, + } + + +def report( + streaming, + renderer, + renderer_history, + callbacks=None, + serialization=None, + serialization_history=(), +): + """Every chart as a Plotly figure and the numbers as tables, in sections: + what the docs page renders.""" + sections = [] + if streaming: + sections.append( + _section( + "streaming", + "Streaming callbacks under load", + "Simulated browsers stream from one app, one setup at a time, with the" + " count raised until frames arrive late or fail. Each browser behaves" + " like the renderer: one shared downlink, a long NDJSON response on ASGI" + " and short polls on WSGI. Server and clients run on separate cores of" + " one machine.", + streaming_charts(streaming), + streaming_table(streaming), + ) + ) + if callbacks: + sections.append( + _section( + "callbacks", + "Callbacks under load: HTTP vs websocket", + "The same callback served two ways: a POST per call, or messages on" + " one websocket per browser (FastAPI and Quart; Flask has no" + " websocket). Simulated browsers speak each protocol as the renderer" + " does, and the count is raised until responses come back late or" + " fail. Server and clients run on separate cores of one machine.", + callbacks_charts(callbacks), + callbacks_table(callbacks), + ) + ) + if renderer: + sections.append( + _section( + "renderer", + "Renderer", + "Each scenario runs as a real app in headless Chrome. Lower is better." + " Growth compares late to early operations: near 1x is flat, higher" + " means the cost grows with the page.", + renderer_charts(renderer, renderer_history), + renderer_table(renderer), + ) + ) + if serialization: + sections.append( + _section( + "serialization", + "Serializing callback responses", + "Time for Dash to turn a callback response or a layout into JSON, on" + " the server. orjson serializes plain data in one fast pass, but one" + " value it cannot take (a component, a go.Figure object, a" + " pd.Timestamp) sends the whole response through a slow Python walk" + " (issue #4065). Without orjson, Python's json module does all the" + " work.", + serialization_charts(serialization, serialization_history), + serialization_table(serialization), + ) + ) + return {"sections": sections} + + +# --- README ------------------------------------------------------------------------ + +README_START = "" +README_END = "" +NOT_MEASURED = "-" + + +def _server(setup): + return (setup["backend"], setup["server"], setup["workers"]) + + +def readme_table(streaming, callbacks): + """Browsers each server setup keeps interactive, as a Markdown table.""" + labels, cells = {}, {} + for s in (callbacks or {}).get("setups", []): + key = _server(s) + labels.setdefault(key, s["label"].rsplit(", ", 1)[0]) + cap, capped = capacity(s, "rtt_p95_ms") + cells[key, s["transport"]] = f"{cap:,}{'+' if capped else ''}" + for s in (streaming or {}).get("setups", []): + key = _server(s) + labels.setdefault(key, s["label"]) + cap, capped = capacity(s) + cells[key, "stream"] = f"{cap:,}{'+' if capped else ''}" + if not labels: + return "No published run yet." + backends = list(BACKEND_HUE) + order = sorted( + labels, + key=lambda k: ( + backends.index(k[0]) if k[0] in backends else len(backends), + k[0], + k[2], + ), + ) + lines = [ + "| Server | Callbacks over HTTP | Callbacks over websocket | Streaming callbacks |", + "|---|--:|--:|--:|", + ] + for key in order: + row = [cells.get((key, col), NOT_MEASURED) for col in ("http", "ws", "stream")] + lines.append(f"| {labels[key]} | " + " | ".join(row) + " |") + runs = [r for r in (callbacks, streaming) if r] + where = sorted( + { + f"{r['meta']['date'][:10]}: server on {r['meta']['server_cpus']} cores" + f" ({r['meta']['runner']}), Dash {r['meta']['dash_version']}" + for r in runs + } + ) + lines += [ + "", + f"Concurrent browsers with p95 response under {CAPACITY_P95_MS} ms and under" + " 1% errors; + means the sweep stopped before the server fell behind." + " Flask has no websocket. Measured " + "; ".join(where) + ".", + ] + return "\n".join(lines) + + +def update_readme(path, streaming, callbacks): + with open(path, encoding="utf-8") as f: + text = f.read() + head, start, rest = text.partition(README_START) + _, end, tail = rest.partition(README_END) + if not start or not end: + raise SystemExit(f"{path} has no {README_START} ... {README_END} section") + table = readme_table(streaming, callbacks) + with open(path, "w", encoding="utf-8") as f: + f.write(f"{head}{start}\n{table}\n{end}{tail}") + + +# --- significance --------------------------------------------------------------- + +# With --only-significant, a run is committed only when it moves the numbers +# people read. Load capacity is a step on the sweep's browser counts, so any +# step counts; a median must move by both the ratio and its floor. +MEDIAN_CHANGE = 0.25 +RENDERER_CHANGE_MS = 5.0 +SERIALIZATION_CHANGE_MS = 5.0 + + +def _cap_text(cap): + if cap is None: + return "not measured" + return f"{cap[0]:,}{'+' if cap[1] else ''}" + + +def load_changes(name, old, new, metric): + """What a load run changes in the capacity table, one line per setup.""" + if not old: + return [f"{name}: first published run"] + labels = {s["name"]: s["label"] for s in old["setups"] + new["setups"]} + before = {s["name"]: capacity(s, metric) for s in old["setups"]} + after = {s["name"]: capacity(s, metric) for s in new["setups"]} + changes = [] + for key in labels: + was, now = before.get(key), after.get(key) + if was != now: + changes.append( + f"{name} {labels[key]}: {_cap_text(was)} -> {_cap_text(now)} browsers" + ) + return changes + + +def median_changes(name, old, new, floor_ms): + """Medians that moved by more than the noise allows.""" + if not old: + return [f"{name}: first published run"] + changes = [] + for outer, inner in new.items(): + for key, stats in inner.items(): + was = old["results"].get(outer, {}).get(key) + now = stats["median"] + if was is None: + changes.append(f"{name} {outer} ({key}): new, {now:,.1f} ms") + continue + delta = abs(now - was["median"]) + if delta >= floor_ms and delta > MEDIAN_CHANGE * was["median"]: + changes.append( + f"{name} {outer} ({key}):" f" {was['median']:,.1f} -> {now:,.1f} ms" + ) + return changes + + +SERIALIZATION_SETUP = ("traces", "points", "python", "plotly", "orjson") + + +def serialization_setup_changes(old, meta): + """What else changed between the runs, so a move is not read as Dash's.""" + + def setup(entry): + return {k: entry.get(k, entry["versions"].get(k)) for k in SERIALIZATION_SETUP} + + was, now = setup(old), setup(meta) + return [ + f"serialization measured with {k} {now[k]} (was {was[k]})" + for k in SERIALIZATION_SETUP + if was[k] != now[k] + ] + + +# --- main --------------------------------------------------------------------------- + + +def _publish_load(run, out, name, metric): + """Commit a load run: its numbers, and a capacity summary in its history.""" + _dump(os.path.join(out, f"{name}-latest.json"), run) + history = _load(os.path.join(out, f"{name}-history.json"), []) + summary = { + "date": run["meta"]["date"], + "commit": run["meta"]["commit"], + "runner": run["meta"]["runner"], + "capacity": {s["name"]: capacity(s, metric)[0] for s in run["setups"]}, + } + _dump(os.path.join(out, f"{name}-history.json"), _append(history, summary, "date")) + + +def _publish_medians(name, entry, out): + # The trend only charts medians; full stats stay in -latest. + trend = { + "date": entry["date"], + "commit": entry["commit"], + "runner": entry["runner"], + "results": { + outer: {k: {"median": st["median"]} for k, st in inner.items()} + for outer, inner in entry["results"].items() + }, + } + history = _load(os.path.join(out, f"{name}-history.json"), []) + _dump(os.path.join(out, f"{name}-latest.json"), entry) + _dump(os.path.join(out, f"{name}-history.json"), _append(history, trend, "date")) + + +def _publish_serialization(run, out, stamp, only_significant): + if not run: + return [] + old = _load(os.path.join(out, "serialization-latest.json"), None) + moved = median_changes( + "serialization", old, run["results"], SERIALIZATION_CHANGE_MS + ) + if moved and old: + moved += serialization_setup_changes(old, run["meta"]) + if moved or not only_significant: + _publish_medians( + "serialization", {**stamp, **run["meta"], "results": run["results"]}, out + ) + return moved + + +def _write_outputs(out, readme): + """report.json and the README table, from everything committed in ``out``.""" + + def latest(name): + return _load(os.path.join(out, f"{name}-latest.json"), None) + + def history(name): + return _load(os.path.join(out, f"{name}-history.json"), []) + + streaming, callbacks = latest("streaming"), latest("callbacks") + _dump( + os.path.join(out, "report.json"), + report( + streaming, + latest("renderer"), + history("renderer"), + callbacks, + latest("serialization"), + history("serialization"), + ), + ) + update_readme(readme, streaming, callbacks) + + +def main(argv=None): + parser = argparse.ArgumentParser(description=__doc__.split("\n\n", maxsplit=1)[0]) + parser.add_argument("--dir", default=os.path.join(HERE, "results")) + parser.add_argument("--readme", default=os.path.join(REPO_ROOT, "README.md")) + parser.add_argument("--renderer", help="results.json from benchmarks.run") + parser.add_argument( + "--streaming", help="results.json from benchmarks.streaming.load" + ) + parser.add_argument( + "--callbacks", help="results.json from benchmarks.callbacks.load" + ) + parser.add_argument( + "--serialization", help="results.json from benchmarks.serialization.bench" + ) + parser.add_argument("--commit", default=os.environ.get("GITHUB_SHA", "")[:9]) + parser.add_argument("--date") + parser.add_argument( + "--runner", default=os.environ.get("BENCH_RUNNER", "local machine") + ) + parser.add_argument( + "--only-significant", + action="store_true", + help="commit a run only if it changes the published numbers enough", + ) + parser.add_argument("--changes", help="write the list of changes here (Markdown)") + args = parser.parse_args(argv) + out = args.dir + os.makedirs(out, exist_ok=True) + + changes = [] + for name, path, metric in ( + ("streaming", args.streaming, "latency_p95_ms"), + ("callbacks", args.callbacks, "rtt_p95_ms"), + ): + run = _read(path) + if not run: + continue + old = _load(os.path.join(out, f"{name}-latest.json"), None) + moved = load_changes(name, old, run, metric) + if moved or not args.only_significant: + changes += moved + _publish_load(run, out, name, metric) + + date = args.date or time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) + stamp = {"date": date, "commit": args.commit, "runner": args.runner} + results = _read(args.renderer) + if results: + old = _load(os.path.join(out, "renderer-latest.json"), None) + moved = median_changes("renderer", old, results, RENDERER_CHANGE_MS) + if moved or not args.only_significant: + changes += moved + _publish_medians("renderer", {**stamp, "results": results}, out) + + changes += _publish_serialization( + _read(args.serialization), out, stamp, args.only_significant + ) + + if args.changes: + with open(args.changes, "w", encoding="utf-8") as f: + f.write("".join(f"- {c}\n" for c in changes)) + if args.only_significant and not changes: + print("no significant change; nothing written") + return 0 + + _write_outputs(out, args.readme) + print(f"results written to {out}, README table to {args.readme}") + for change in changes: + print(f" {change}") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/benchmarks/serialization/__init__.py b/benchmarks/serialization/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/benchmarks/serialization/bench.py b/benchmarks/serialization/bench.py new file mode 100644 index 0000000000..fe5d5a36ef --- /dev/null +++ b/benchmarks/serialization/bench.py @@ -0,0 +1,231 @@ +"""Time Dash's JSON serialization of callback responses and layouts. + + python -m benchmarks.serialization.bench --out benchmarks/serialization/results.json + python -m benchmarks.serialization.bench --case list_float+div --engine json + +Every case is one large figure (``--traces`` x ``--points``) serialized with +``dash._utils.to_json``, the way a callback response or the initial layout is. +The figure alone is the reference; the other cases add something small that +orjson cannot serialize natively (a component, a ``go.Figure`` object, a +``pd.Timestamp``) or wrap the figure in a layout. See issue #4065. + +Each engine runs in its own process. ``json`` blocks the orjson import, so +plotly takes its stdlib fallback exactly as on an install without orjson. +""" +from __future__ import annotations + +import argparse +import gc +import json +import os +import platform +import statistics +import subprocess +import sys +import time + +REPO_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +ENGINES = ("orjson", "json") +DATA = { + "numpy": "numpy array", + "list_float": "list of float", + "datetimeindex": "pd.DatetimeIndex", + "list_datetime": "list of datetime", +} +EXTRAS = { + "none": "figure alone", + "div": "+ html.Div", + "figure": "as go.Figure", + "timestamp": "+ pd.Timestamp", + "layout": "in a layout", +} +CASES = [f"{d}+{e}" for d in DATA for e in EXTRAS] + + +def label(case): + data, extra = case.split("+") + return f"{DATA[data]}, {EXTRAS[extra]}" + + +def _figure(data, traces, points): + # pylint: disable=import-outside-toplevel + import datetime + + import numpy as np + import pandas as pd + import plotly.graph_objects as go + + rng = np.random.default_rng(0) + + def trace(): + y = rng.standard_normal(points) + if data == "numpy": + return go.Scatter(y=y) + if data == "list_float": + return go.Scatter(y=y.tolist()) + if data == "datetimeindex": + return go.Scatter(x=pd.date_range("2024", periods=points, freq="min"), y=y) + start = datetime.datetime(2024, 1, 1) + x = [start + datetime.timedelta(minutes=i) for i in range(points)] + return go.Scatter(x=x, y=y) + + return go.Figure([trace() for _ in range(traces)]) + + +def payload(case, traces, points): + """What Dash hands to ``to_json`` for this case.""" + # pylint: disable=import-outside-toplevel + import pandas as pd + from dash import dcc, html + + data, extra = case.split("+") + fig = _figure(data, traces, points) + if extra == "layout": + return html.Div([html.H1("Report"), dcc.Graph(id="graph", figure=fig)]) + response = {"graph": {"figure": fig if extra == "figure" else fig.to_plotly_json()}} + if extra == "div": + response["status"] = {"children": html.Div("ok")} + elif extra == "timestamp": + response["status"] = {"value": pd.Timestamp("2024-01-01")} + return {"multi": True, "response": response} + + +def _versions(): + # pylint: disable=import-outside-toplevel + from importlib.metadata import PackageNotFoundError, version + + out = {"python": platform.python_version()} + for pkg in ("dash", "plotly", "orjson", "numpy", "pandas"): + try: + out[pkg] = version(pkg) + except PackageNotFoundError: + out[pkg] = None + return out + + +def _require_orjson(): + try: + import orjson # noqa: F401 pylint: disable=import-outside-toplevel,unused-import + except ImportError as err: + raise SystemExit("the orjson engine needs orjson installed") from err + + +def worker(engine, cases, traces, points, repeat): + if engine == "json": + sys.modules["orjson"] = None # type: ignore[assignment] + else: + _require_orjson() + # pylint: disable-next=import-outside-toplevel + from dash._utils import to_json + + results = {} + for case in cases: + value = payload(case, traces, points) + size = len(to_json(value)) + times = [] + for _ in range(repeat): + gc.collect() + start = time.perf_counter() + to_json(value) + times.append((time.perf_counter() - start) * 1000) + results[case] = { + "median": statistics.median(times), + "min": min(times), + "max": max(times), + "bytes": size, + } + print( + f" {engine:6} {label(case):40} {results[case]['median']:8.1f} ms", + file=sys.stderr, + ) + # pylint: disable-next=import-outside-toplevel + from _plotly_utils.optional_imports import get_module + + loaded = get_module("orjson", should_load=True) is not None + return {"versions": _versions(), "orjson_loaded": loaded, "results": results} + + +def run(engines, cases, traces, points, repeat): + env = { + **os.environ, + "PYTHONPATH": os.pathsep.join( + p for p in (REPO_ROOT, os.environ.get("PYTHONPATH")) if p + ), + } + by_engine = {} + for engine in engines: + out = subprocess.run( + [ + sys.executable, + "-m", + "benchmarks.serialization.bench", + "--worker", + engine, + "--traces", + str(traces), + "--points", + str(points), + "--repeat", + str(repeat), + "--case", + *cases, + ], + cwd=REPO_ROOT, + env=env, + check=True, + stdout=subprocess.PIPE, + text=True, + ) + by_engine[engine] = json.loads(out.stdout) + versions = by_engine[engines[0]]["versions"] + results = { + case: {engine: run_["results"][case] for engine, run_ in by_engine.items()} + for case in cases + } + return { + "meta": { + "traces": traces, + "points": points, + "repeat": repeat, + "versions": versions, + "orjson_loaded": {e: r["orjson_loaded"] for e, r in by_engine.items()}, + "labels": {case: label(case) for case in cases}, + }, + "results": results, + } + + +def main(argv=None): + parser = argparse.ArgumentParser(description=__doc__.split("\n\n", maxsplit=1)[0]) + parser.add_argument("--engine", nargs="+", choices=ENGINES, default=list(ENGINES)) + parser.add_argument("--case", nargs="+", choices=CASES, default=CASES) + parser.add_argument("--traces", type=int, default=10) + parser.add_argument("--points", type=int, default=20_000) + parser.add_argument("--repeat", type=int, default=7) + parser.add_argument("--out", help="write results here (JSON)") + parser.add_argument("--worker", choices=ENGINES, help=argparse.SUPPRESS) + args = parser.parse_args(argv) + + if args.worker: + json.dump( + worker(args.worker, args.case, args.traces, args.points, args.repeat), + sys.stdout, + ) + return 0 + + data = run(args.engine, args.case, args.traces, args.points, args.repeat) + header = f"{'case':42}" + "".join(f"{e:>10}" for e in args.engine) + print(header) + for case, by_engine in data["results"].items(): + print( + f"{label(case):42}" + + "".join(f"{by_engine[e]['median']:8.1f}ms" for e in args.engine) + ) + if args.out: + with open(args.out, "w", encoding="utf-8") as f: + json.dump(data, f, indent=2) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/benchmarks/streaming/__init__.py b/benchmarks/streaming/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/benchmarks/streaming/client.py b/benchmarks/streaming/client.py new file mode 100644 index 0000000000..10bd238365 --- /dev/null +++ b/benchmarks/streaming/client.py @@ -0,0 +1,247 @@ +"""Simulated browsers for the streaming load test. + +Run by ``benchmarks.streaming.load``, several processes at once: one Python +process saturates its own CPU at a few hundred browsers and then reports its +own slowness as server latency. + +Each browser loads the page once (for its signed ``end_id`` and the server's +stream mode), then loops: click (uplink POST), read its frames off the +downlink until the ``done`` frame, think, click again. The downlink follows +the renderer's ``StreamClient``: one long NDJSON response in ``stream`` mode +(ASGI), or polling with the same pacing and backoff in ``poll`` mode (WSGI). + +Prints one JSON line with the measurements of every stream clicked inside the +measure window, however late it finished: counting by finish time would drop +exactly the slow streams a struggling server produces. +""" +import argparse +import asyncio +import json +import random +import re +import sys +import time + +import aiohttp + +CONFIG_RE = re.compile(r'', re.S) +STAMP_RE = re.compile(r'"(\d{10}\.\d{6})"') +UPDATE = "/_dash-update-component" +# Matches streamClient.ts. +EMPTY_POLLS_BEFORE_BACKOFF = 2 +MAX_BACKOFF_FACTOR = 5 + +CLICK = { + "output": "out.children", + "outputs": {"id": "out", "property": "children"}, + "inputs": [{"id": "btn", "property": "n_clicks", "value": 1}], + "changedPropIds": ["btn.n_clicks"], + "state": [], +} + + +class StreamReset(Exception): + pass + + +class Stats: + def __init__(self, measure_from, measure_to): + self.measure_from = measure_from + self.measure_to = measure_to + self.latency_ms = [] + self.first_frame_ms = [] + self.streams = 0 + self.frames = 0 + self.errors = 0 + self.incomplete = 0 + self.failed_browsers = 0 + + def in_window(self, t): + return self.measure_from <= t <= self.measure_to + + +async def _lines(resp): + buf = b"" + async for chunk in resp.content.iter_any(): + buf += chunk + *lines, buf = buf.split(b"\n") + for line in lines: + if line.strip(): + yield json.loads(line) + if buf.strip(): + yield json.loads(buf) + + +class Browser: + def __init__(self, base, stats, frames, think): + self.base = base + self.stats = stats + self.expected = frames + self.think = think + self.cursor = 0 + # Set from the page config by load_page(). + self.end_id = "" + self.mode = "stream" + self.poll_interval = 0.1 + # Browsers cap connections per host at six. + self.session = aiohttp.ClientSession( + connector=aiohttp.TCPConnector(limit=6), + timeout=aiohttp.ClientTimeout(total=None, sock_connect=30), + ) + + async def load_page(self): + async with self.session.get(self.base + "/") as resp: + text = await resp.text() + config = json.loads(CONFIG_RE.search(text).group(1)) + stream = config.get("stream") or {} + self.end_id = config["end_id"] + self.mode = stream.get("mode", "stream") + self.poll_interval = stream.get("poll_interval", 100) / 1000 + + def _url(self): + return f"{self.base}{UPDATE}?endId={self.end_id}" + + def _frame(self, frame, clicked, state): + now = time.time() + if frame.get("done"): + return True + match = STAMP_RE.search(json.dumps(frame)) + if match: + counted = self.stats.in_window(clicked) + if state["frames"] == 0 and counted: + self.stats.first_frame_ms.append((now - clicked) * 1000) + state["frames"] += 1 + if counted: + self.stats.latency_ms.append((now - float(match.group(1))) * 1000) + self.stats.frames += 1 + return False + + def _envelopes(self, envelope, rid, clicked, state): + if envelope.get("reset"): + # The renderer settles every pending stream on a reset. + self.cursor = 0 + raise StreamReset() + if isinstance(envelope.get("seq"), int): + self.cursor = envelope["seq"] + if envelope.get("rid") != rid: + return False + return self._frame(envelope.get("frame") or {}, clicked, state) + + async def _downlink_stream(self, rid, clicked, state): + # Like the renderer, reconnect from the cursor if the downlink closes + # before the done frame, backing off when it closed with nothing. + while True: + got = False + async with self.session.post( + self._url(), json={"streamDownlink": {"from": self.cursor}} + ) as resp: + resp.raise_for_status() + async for envelope in _lines(resp): + got = True + if self._envelopes(envelope, rid, clicked, state): + return + if not got: + await asyncio.sleep(1) + + async def _downlink_poll(self, rid, clicked, state): + empty = 0 + while True: + async with self.session.post( + self._url(), json={"streamDownlink": {"from": self.cursor}} + ) as resp: + resp.raise_for_status() + envelopes = [e async for e in _lines(resp)] + for envelope in envelopes: + if self._envelopes(envelope, rid, clicked, state): + return + empty = 0 if envelopes else empty + 1 + factor = 1 + if empty > EMPTY_POLLS_BEFORE_BACKOFF: + factor = min( + 2 ** (empty - EMPTY_POLLS_BEFORE_BACKOFF), MAX_BACKOFF_FACTOR + ) + await asyncio.sleep(self.poll_interval * factor) + + async def one_stream(self, n): + rid = f"{id(self)}-{n}" + clicked = time.time() + async with self.session.post( + self._url(), json={**CLICK, "streamConnection": {"requestId": rid}} + ) as resp: + resp.raise_for_status() + await resp.read() + state = {"frames": 0} + if self.mode == "poll": + await self._downlink_poll(rid, clicked, state) + else: + await self._downlink_stream(rid, clicked, state) + if self.stats.in_window(clicked): + self.stats.streams += 1 + if state["frames"] < self.expected: + self.stats.incomplete += 1 + + async def run(self, start_at, stop_at): + await asyncio.sleep(max(0.0, start_at - time.time())) + n = 0 + try: + await asyncio.wait_for(self.load_page(), timeout=30) + while time.time() < stop_at: + n += 1 + clicked = time.time() + try: + await asyncio.wait_for(self.one_stream(n), timeout=60) + except ( + aiohttp.ClientError, + asyncio.TimeoutError, + ValueError, + StreamReset, + ): + if self.stats.in_window(clicked): + self.stats.errors += 1 + await asyncio.sleep(1) + await asyncio.sleep(random.uniform(0, 2 * self.think)) + except (aiohttp.ClientError, asyncio.TimeoutError, AttributeError, ValueError): + # The page never loaded: this browser never got to stream at all. + self.stats.errors += 1 + self.stats.failed_browsers += 1 + finally: + await self.session.close() + + +async def main(args): + start = args.start + ramp_end = start + args.ramp + stats = Stats(ramp_end, ramp_end + args.duration) + browsers = [ + Browser(args.url, stats, args.frames, args.think) for _ in range(args.browsers) + ] + await asyncio.gather( + *( + b.run(start + args.ramp * i / max(1, args.browsers), stats.measure_to) + for i, b in enumerate(browsers) + ) + ) + json.dump( + { + "latency_ms": [round(x, 2) for x in stats.latency_ms], + "first_frame_ms": [round(x, 2) for x in stats.first_frame_ms], + "streams": stats.streams, + "frames": stats.frames, + "errors": stats.errors, + "incomplete": stats.incomplete, + "failed_browsers": stats.failed_browsers, + }, + sys.stdout, + ) + + +if __name__ == "__main__": + parser = argparse.ArgumentParser() + parser.add_argument("--url", required=True) + parser.add_argument("--browsers", type=int, required=True) + parser.add_argument("--frames", type=int, required=True) + parser.add_argument("--think", type=float, default=1.0) + parser.add_argument("--start", type=float, required=True) + parser.add_argument("--ramp", type=float, default=5.0) + parser.add_argument("--duration", type=float, default=20.0) + asyncio.run(main(parser.parse_args())) diff --git a/benchmarks/streaming/load.py b/benchmarks/streaming/load.py new file mode 100644 index 0000000000..f491833814 --- /dev/null +++ b/benchmarks/streaming/load.py @@ -0,0 +1,155 @@ +"""Streaming-callback load test: how many browsers each backend serves. + +For each server setup (backend x server x workers) and each browser count, +start the server, run simulated browsers against it (``client.py``, spread +over several processes) and record frame latency, time to first frame, +errors and CPU. A setup stops climbing once it saturates (p95 latency over +``--max-p95`` ms or errors over 1%), since higher counts only add noise. + + python -m benchmarks.streaming.load --out benchmarks/streaming/results.json + python -m benchmarks.streaming.load --setup fastapi-uvicorn-w1 --browsers 100 500 + +Multi-worker setups need shared storage across processes; pass +``--redis-url`` (they are skipped without it). The server and the clients +are pinned to separate CPUs (``--server-cpus``) so the clients' own load +does not show up as server latency, and each point records the clients' +peak CPU so a client-bound point can be spotted. +""" +from __future__ import annotations + +import argparse +import math +import os +import sys + +from benchmarks import loadkit +from benchmarks.loadkit import Setup, pct + +HERE = os.path.dirname(os.path.abspath(__file__)) + +SETUPS = [ + Setup( + "flask-gunicorn-w1", + "flask", + "gunicorn", + 1, + "Flask, gunicorn 1 worker x 8 threads", + ), + Setup( + "flask-gunicorn-w4", + "flask", + "gunicorn", + 4, + "Flask, gunicorn 4 workers x 8 threads", + ), + Setup("fastapi-uvicorn-w1", "fastapi", "uvicorn", 1, "FastAPI, uvicorn 1 worker"), + Setup("fastapi-uvicorn-w4", "fastapi", "uvicorn", 4, "FastAPI, uvicorn 4 workers"), + Setup("quart-uvicorn-w1", "quart", "uvicorn", 1, "Quart, uvicorn 1 worker"), + Setup("quart-uvicorn-w4", "quart", "uvicorn", 4, "Quart, uvicorn 4 workers"), +] +SETUPS_BY_NAME = {s.name: s for s in SETUPS} +COUNTS = ("streams", "frames", "errors", "incomplete", "failed_browsers") + + +def _failed(report, share): + report.update(errors=share, failed_browsers=share) + + +def run_point(setup, browsers, args): + env = { + "LOAD_BACKEND": setup.backend, + "LOAD_FRAMES": str(args.frames), + "LOAD_INTERVAL": str(args.interval), + "LOAD_SECRET": "load-test", + "LOAD_REDIS_URL": args.redis_url if setup.needs_redis else "", + } + with loadkit.serve(setup, HERE, "load_app", env, args.server_cpus) as ( + url, + server, + ): + # Streams started in the window may run up to the client's 60 s + # stream timeout past it. + reports, resources = loadkit.run_clients( + url, + server, + browsers, + "benchmarks.streaming.client", + ["--frames", str(args.frames), "--think", str(args.think)], + args, + _failed, + ) + lat = [x for r in reports for x in r.get("latency_ms", [])] + first = [x for r in reports for x in r.get("first_frame_ms", [])] + counts = {k: sum(r.get(k, 0) for r in reports) for k in COUNTS} + attempts = counts["streams"] + counts["errors"] + error_rate = ( + (counts["errors"] + counts["incomplete"]) / attempts if attempts else 1.0 + ) + point = { + "browsers": browsers, + "latency_p50_ms": pct(lat, 0.5), + "latency_p95_ms": pct(lat, 0.95), + "latency_p99_ms": pct(lat, 0.99), + "first_frame_p50_ms": pct(first, 0.5), + "first_frame_p95_ms": pct(first, 0.95), + "frames_per_s": round(counts["frames"] / args.duration, 1), + "streams_per_s": round(counts["streams"] / args.duration, 2), + **counts, + "error_rate": round(error_rate, 4), + **resources, + } + point["saturated"] = bool( + error_rate > 0.01 or (point["latency_p95_ms"] or math.inf) > args.max_p95 + ) + return point + + +def _entry_fields(setup, redis_url): + if setup.needs_redis and not redis_url: + print(f"skip {setup.name}: multi-worker needs --redis-url", flush=True) + return None + return {"storage": "redis" if setup.needs_redis else "local"} + + +def _describe(point): + return ( + f"p50 {point['latency_p50_ms']}ms p95 {point['latency_p95_ms']}ms" + f" first {point['first_frame_p50_ms']}ms" + ) + + +def main(argv=None): + parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) + parser.add_argument("--setup", nargs="*", choices=list(SETUPS_BY_NAME)) + loadkit.add_common_args(parser, [25, 50, 100, 250, 500, 1000, 2000, 4000]) + parser.add_argument("--frames", type=int, default=20) + parser.add_argument( + "--interval", type=float, default=0.1, help="seconds between frames" + ) + parser.add_argument("--out", default=os.path.join(HERE, "results.json")) + args = parser.parse_args(argv) + loadkit.configure(args) + + setups = [SETUPS_BY_NAME[n] for n in args.setup] if args.setup else SETUPS + results = { + "kind": "streaming_load", + "meta": loadkit.machine_info(args), + "params": { + k: getattr(args, k) + for k in ("frames", "interval", "think", "ramp", "duration", "max_p95") + }, + "setups": [], + } + loadkit.sweep( + results, + setups, + args, + run_point, + _describe, + lambda setup: _entry_fields(setup, args.redis_url), + ) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/benchmarks/streaming/load_app.py b/benchmarks/streaming/load_app.py new file mode 100644 index 0000000000..e039d590a1 --- /dev/null +++ b/benchmarks/streaming/load_app.py @@ -0,0 +1,46 @@ +"""The app the streaming load test serves. + +One streaming callback that yields ``FRAMES`` frames, ``INTERVAL`` seconds +apart. Each frame carries the server's ``time.time()`` so the client can +measure delivery latency (clients run on the same machine, so the clocks +agree). + +Configured from the environment so gunicorn, uvicorn and hypercorn can all +import it: ``LOAD_BACKEND`` (flask, fastapi, quart), ``LOAD_FRAMES``, +``LOAD_INTERVAL``. ``LOAD_REDIS_URL`` switches shared storage to Redis (needed +across workers), and ``LOAD_SECRET`` gives every worker the same signing secret +so they verify each other's stream tokens. +""" +import asyncio +import os +import time + +from dash import Dash, Input, Output, LocalSharedStorage, RedisSharedStorage, html + +BACKEND = os.environ.get("LOAD_BACKEND", "flask") +FRAMES = int(os.environ.get("LOAD_FRAMES", "20")) +INTERVAL = float(os.environ.get("LOAD_INTERVAL", "0.1")) + +REDIS_URL = os.environ.get("LOAD_REDIS_URL") + +app = Dash( + __name__, + backend=BACKEND, + shared_storage=RedisSharedStorage(url=REDIS_URL) + if REDIS_URL + else LocalSharedStorage, +) +app.server.secret_key = os.environ.get("LOAD_SECRET", "load-test") +app.layout = html.Div([html.Button("go", id="btn", n_clicks=0), html.Div(id="out")]) + + +@app.callback(Output("out", "children"), Input("btn", "n_clicks")) +async def stream(n): + if not n: + return + for _ in range(FRAMES): + yield f"{time.time():.6f}" + await asyncio.sleep(INTERVAL) + + +server = app.server diff --git a/benchmarks/swarm/Dockerfile b/benchmarks/swarm/Dockerfile new file mode 100644 index 0000000000..44b1b58298 --- /dev/null +++ b/benchmarks/swarm/Dockerfile @@ -0,0 +1,13 @@ +# The swarm agent: headless Chromium users driven by Playwright. +# docker build -t dash-swarm-agent benchmarks/swarm +# docker run --rm --init --ipc=host dash-swarm-agent --target http://app:8050 --users 50 +# The image ships the browsers but not the Python package: install it at the +# version that matches the tag. +FROM mcr.microsoft.com/playwright/python:v1.63.0-noble + +RUN python3 -m pip install --no-cache-dir --break-system-packages \ + "playwright==1.63.0" "psutil>=7" +COPY agent.py histogram.py /swarm/ +# The image's unprivileged user. +USER pwuser +ENTRYPOINT ["python3", "/swarm/agent.py"] diff --git a/benchmarks/swarm/__init__.py b/benchmarks/swarm/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/benchmarks/swarm/agent.py b/benchmarks/swarm/agent.py new file mode 100644 index 0000000000..95877f4ba6 --- /dev/null +++ b/benchmarks/swarm/agent.py @@ -0,0 +1,339 @@ +"""One machine's share of a browser swarm: real headless Chromium users. + +Each user is its own browser context (own cookies, cache and SharedWorkers, +like a separate person) in one of ``--browsers`` Chromium processes. A user +loads the page, then loops: click one of the scenarios, wait for its output, +think, click again. Timings are taken inside the page with +``performance.now()``, from just before the click to the DOM update a person +would see, so they include the renderer and need no clock agreement between +machines. Only the start time is shared (``--start-at``, wall clock; NTP is +close enough). + +Scenarios (``--scenario``, picked at random per click): + +- ``http``: a plain callback over HTTP. +- ``ws``: the same callback over the websocket. +- ``stream``: a streaming callback; records first frame and whole stream. + +Writes one JSON report (``--out``) with mergeable histograms; combine the +swarm's reports with ``benchmarks.swarm.aggregate``. The report also carries +the agent machine's CPU during the window: past ~85% the browsers themselves +are the bottleneck and the numbers describe this machine, not the server. + +Needs only ``playwright`` (with Chromium) and ``psutil``, not dash. +""" +import argparse +import asyncio +import contextlib +import json +import math +import os +import random +import socket +import sys +import time + +import psutil +from playwright.async_api import async_playwright + +try: + from .histogram import Histogram +except ImportError: + # Run as a script inside the agent image, where there is no package. + sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + from histogram import Histogram # type: ignore[no-redef] + +ACTION_TIMEOUT = 30 +LOAD_TIMEOUT = 60 + +# Click a scenario's button and resolve once its output shows the result for +# this click: "clicked " for a callback, ": done" for a stream. +CLICK_AND_WAIT = """ +async ([name, n]) => { + const out = document.getElementById(name + '-out'); + const btn = document.getElementById(name + '-btn'); + return await new Promise((resolve) => { + let first = null; + const t0 = performance.now(); + const check = () => { + const text = out.textContent; + if (name === 'stream') { + if (!text.startsWith(n + ':')) return; + const now = performance.now() - t0; + if (first === null) first = now; + if (text.endsWith(' done')) { + obs.disconnect(); + resolve({first, total: now}); + } + } else if (text === 'clicked ' + n) { + obs.disconnect(); + resolve({total: performance.now() - t0}); + } + }; + const obs = new MutationObserver(check); + obs.observe(out, {childList: true, subtree: true, characterData: true}); + btn.click(); + }); +} +""" + + +class Report: + def __init__(self, measure_from, measure_to): + self.measure_from = measure_from + self.measure_to = measure_to + self.hists = {} + self.counts = { + "actions": 0, + "errors": 0, + "page_errors": 0, + "users_failed": 0, + "reloads": 0, + } + self.errors_by_kind = {} + + def in_window(self, t): + return self.measure_from <= t <= self.measure_to + + def add(self, metric, ms): + self.hists.setdefault(metric, Histogram()).add(ms) + + def error(self, kind): + self.counts["errors"] += 1 + self.errors_by_kind[kind] = self.errors_by_kind.get(kind, 0) + 1 + + +class User: + def __init__(self, browser, args, report): + self.browser = browser + self.args = args + self.report = report + self.page = None + self.clicks = {} + + async def load(self): + # The page restarts its click counts whether or not the load succeeds. + self.clicks = {} + await self.page.goto(self.args.target, timeout=LOAD_TIMEOUT * 1000) + await self.page.wait_for_selector( + "#ready", state="attached", timeout=LOAD_TIMEOUT * 1000 + ) + if "ws" in self.args.scenario and not await self.page.evaluate( + "() => !!JSON.parse(" + "document.getElementById('_dash-config').textContent).websocket" + ): + raise RuntimeError("no websocket on this backend: drop the ws scenario") + + async def act(self, scenario): + n = self.clicks.get(scenario, 0) + 1 + self.clicks[scenario] = n + return await asyncio.wait_for( + self.page.evaluate(CLICK_AND_WAIT, [scenario, n]), ACTION_TIMEOUT + ) + + async def _start(self): + """Open this user's context and load the page; None if it failed. + A user that never starts is counted whenever it happens: it takes its + share of the load away from the whole run.""" + context = None + try: + context = await self.browser.new_context() + self.page = await context.new_page() + + def on_page_error(_err): + self.report.counts["page_errors"] += 1 + + self.page.on("pageerror", on_page_error) + started = time.time() + await self.load() + # The first load happens during the ramp, so it is not windowed. + self.report.add("page_load", (time.time() - started) * 1000) + return context + except Exception: # pylint: disable=broad-except + self.report.counts["users_failed"] += 1 + self.report.error("page_load") + if context is not None: + with contextlib.suppress(Exception): + await context.close() + return None + + async def _reload(self): + """Start the page over after an error; False if that failed too.""" + self.report.counts["reloads"] += 1 + attempted = time.time() + try: + await self.load() + return True + except Exception: # pylint: disable=broad-except + if self.report.in_window(attempted): + self.report.error("reload") + await asyncio.sleep(1) + return False + + async def _click(self, scenario): + """One timed click; False if it failed and the page needs a reload.""" + clicked = time.time() + counted = self.report.in_window + try: + result = await self.act(scenario) + except Exception as err: # pylint: disable=broad-except + if counted(clicked): + kind = ( + "timeout" + if isinstance(err, asyncio.TimeoutError) + else type(err).__name__ + ) + self.report.error(f"{scenario}:{kind}") + return False + if counted(clicked): + self.report.counts["actions"] += 1 + if scenario == "stream": + self.report.add("stream_first", result["first"]) + self.report.add("stream_total", result["total"]) + else: + self.report.add(scenario, result["total"]) + return True + + async def run(self, start_at, stop_at): + await asyncio.sleep(max(0.0, start_at - time.time())) + context = await self._start() + if context is None: + return + loaded = True + try: + while time.time() < stop_at: + if not loaded and not await self._reload(): + continue + # Not a security use: just varies what users click. + scenario = random.choice(self.args.scenario) # NOSONAR + loaded = await self._click(scenario) + await asyncio.sleep(random.uniform(0, 2 * self.args.think)) # NOSONAR + finally: + with contextlib.suppress(Exception): + await context.close() + + +async def _watch_cpu(report, stop_at, samples): + """Sample this machine's CPU (browsers included) inside the window.""" + psutil.cpu_percent(None) + while time.time() < stop_at: + await asyncio.sleep(2) + if report.in_window(time.time()): + samples.append(psutil.cpu_percent(None)) + else: + psutil.cpu_percent(None) + + +async def main(args): + start = args.start_at or time.time() + 5 + measure_from = start + args.ramp + stop_at = measure_from + args.duration + report = Report(measure_from, stop_at) + cpu = [] + async with async_playwright() as pw: + browsers = await asyncio.gather( + *( + pw.chromium.launch( + headless=True, args=["--disable-dev-shm-usage", "--no-sandbox"] + ) + for _ in range(args.browsers) + ) + ) + # Past the shared start, this agent's ramp is compressed and the swarm + # no longer starts together; the report says by how much. + late = max(0.0, time.time() - start) + users = [ + User(browsers[i % len(browsers)], args, report) for i in range(args.users) + ] + watcher = asyncio.ensure_future(_watch_cpu(report, stop_at, cpu)) + await asyncio.gather( + *( + u.run(start + args.ramp * i / max(1, args.users), stop_at) + for i, u in enumerate(users) + ) + ) + watcher.cancel() + for browser in browsers: + with contextlib.suppress(Exception): + await browser.close() + + cpu.sort() + out = { + "agent": args.agent_id or socket.gethostname(), + "target": args.target, + "users": args.users, + "browsers": args.browsers, + "scenario": args.scenario, + "think": args.think, + "window": [measure_from, stop_at], + "duration": args.duration, + "late_s": round(late, 2), + "cpus": psutil.cpu_count(), + "cpu_mean_pct": round(sum(cpu) / len(cpu)) if cpu else None, + # p90 of 2 s samples: one spike shouldn't flag an agent as bound. + "cpu_p90_pct": round(cpu[int(0.9 * (len(cpu) - 1))]) if cpu else None, + **report.counts, + "errors_by_kind": report.errors_by_kind, + "histograms": {k: h.to_json() for k, h in report.hists.items()}, + } + return out + + +def write_report(out, path): + text = json.dumps(out) + if path == "-": + print(text) + else: + with open(path, "w", encoding="utf-8") as f: + f.write(text) + summary = { + k: Histogram.from_json(h).summary() for k, h in out["histograms"].items() + } + print( + json.dumps( + { + "agent": out["agent"], + "actions": out["actions"], + "errors": out["errors"], + "late_s": out["late_s"], + "cpu_p90_pct": out["cpu_p90_pct"], + **summary, + } + ), + file=sys.stderr, + ) + + +def parse_args(argv=None): + parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) + parser.add_argument("--target", required=True, help="the app's URL") + parser.add_argument("--users", type=int, default=20) + parser.add_argument( + "--browsers", + type=int, + help="Chromium processes the users share (default: one per 10 users)", + ) + parser.add_argument( + "--scenario", + default="http,ws,stream", + type=lambda s: [x for x in s.split(",") if x], + ) + parser.add_argument("--think", type=float, default=1.0) + parser.add_argument( + "--start-at", + type=float, + help="wall-clock start shared by the swarm (default: in 5 s)", + ) + parser.add_argument("--ramp", type=float, default=30.0) + parser.add_argument("--duration", type=float, default=60.0) + parser.add_argument("--agent-id") + parser.add_argument("--out", default="-") + args = parser.parse_args(argv) + if args.browsers is None: + args.browsers = max(1, math.ceil(args.users / 10)) + return args + + +if __name__ == "__main__": + ARGS = parse_args() + write_report(asyncio.run(main(ARGS)), ARGS.out) diff --git a/benchmarks/swarm/aggregate.py b/benchmarks/swarm/aggregate.py new file mode 100644 index 0000000000..a7d12e7f16 --- /dev/null +++ b/benchmarks/swarm/aggregate.py @@ -0,0 +1,108 @@ +"""Merge a swarm's agent reports into one result. + + python -m benchmarks.swarm.aggregate results/*.json --out swarm.json + +Histograms add up bucket by bucket, so the percentiles are those of every +click in the swarm, not an average of per-agent percentiles. Agents whose +machine ran past 85% CPU (p90 of the window's samples) are flagged: their +browsers were the bottleneck. Agents that got their browsers up after the +shared start are flagged too: their ramp was squeezed and the swarm did not +start together. +""" +import argparse +import json +import os +import sys + +try: + from .histogram import Histogram +except ImportError: + sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + from histogram import Histogram # type: ignore[no-redef] + +COUNTS = ("actions", "errors", "page_errors", "users_failed", "reloads") +AGENT_BOUND_PCT = 85 +LATE_S = 1.0 + + +def aggregate(reports): + hists = {} + counts = dict.fromkeys(COUNTS, 0) + errors_by_kind = {} + for r in reports: + for k in COUNTS: + counts[k] += r.get(k, 0) + for k, n in r.get("errors_by_kind", {}).items(): + errors_by_kind[k] = errors_by_kind.get(k, 0) + n + for name, data in r.get("histograms", {}).items(): + hists.setdefault(name, Histogram()).merge(Histogram.from_json(data)) + durations = {r["duration"] for r in reports} + duration = max(durations) if durations else 0 + attempts = counts["actions"] + counts["errors"] + bound = [ + r["agent"] for r in reports if (r.get("cpu_p90_pct") or 0) > AGENT_BOUND_PCT + ] + late = [r["agent"] for r in reports if (r.get("late_s") or 0) > LATE_S] + return { + "agents": len(reports), + "users": sum(r["users"] for r in reports), + "duration": duration, + **counts, + "error_rate": round(counts["errors"] / attempts, 4) if attempts else None, + "actions_per_s": round(counts["actions"] / duration, 1) if duration else None, + "errors_by_kind": errors_by_kind, + "agent_bound": bound, + "late_agents": late, + "metrics": {k: h.summary() for k, h in sorted(hists.items())}, + "histograms": {k: h.to_json() for k, h in sorted(hists.items())}, + } + + +def table(result): + lines = [ + f"{result['agents']} agents, {result['users']} users, " + f"{result['actions_per_s']} actions/s, errors {result['error_rate']:.2%}" + if result["error_rate"] is not None + else f"{result['agents']} agents, no actions", + f"{'metric':14} {'count':>8} {'p50':>9} {'p95':>9} {'p99':>9} {'max':>9}", + ] + for name, s in result["metrics"].items(): + lines.append( + f"{name:14} {s['count']:>8} {s['p50']:>7}ms {s['p95']:>7}ms " + f"{s['p99']:>7}ms {s['max']:>7}ms" + ) + if result["errors_by_kind"]: + lines.append(f"errors: {result['errors_by_kind']}") + if result["users_failed"]: + lines.append(f"USERS FAILED: {result['users_failed']} never got going") + if result["agent_bound"]: + lines.append( + f"AGENT-BOUND (>{AGENT_BOUND_PCT}% CPU): {', '.join(result['agent_bound'])}" + ) + if result["late_agents"]: + lines.append( + f"LATE (>{LATE_S:.0f}s after the shared start): " + + ", ".join(result["late_agents"]) + ) + return "\n".join(lines) + + +def main(argv=None): + parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) + parser.add_argument("reports", nargs="+") + parser.add_argument("--out") + args = parser.parse_args(argv) + reports = [] + for path in args.reports: + with open(path, encoding="utf-8") as f: + reports.append(json.load(f)) + result = aggregate(reports) + print(table(result)) + if args.out: + with open(args.out, "w", encoding="utf-8") as f: + json.dump(result, f, indent=1) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/benchmarks/swarm/app.py b/benchmarks/swarm/app.py new file mode 100644 index 0000000000..dc228a7c8e --- /dev/null +++ b/benchmarks/swarm/app.py @@ -0,0 +1,93 @@ +"""The app a browser swarm loads. + +One page, three things to click, each measured by the agents in the browser: + +- ``#http-btn`` -> ``#http-out``: a plain callback over an HTTP POST. +- ``#ws-btn`` -> ``#ws-out``: the same callback over the websocket + (``websocket=True``; FastAPI and Quart only). +- ``#stream-btn`` -> ``#stream-out``: a streaming callback, ``SWARM_FRAMES`` + frames ``SWARM_INTERVAL`` seconds apart, the last one ending in ``done``. + +``SWARM_WORK_MS`` adds that much ``asyncio.sleep`` to the plain callbacks, like +a query. Serve it the way a deployment would, for example:: + + SWARM_BACKEND=fastapi SWARM_REDIS_URL=redis://redis:6379/0 \\ + uvicorn benchmarks.swarm.app:server --host 0.0.0.0 --workers 4 + +``SWARM_REDIS_URL`` is needed for streaming with more than one worker or +instance, and ``SWARM_SECRET`` must be the same everywhere the app runs. +""" +import asyncio +import os + +from dash import Dash, Input, Output, LocalSharedStorage, RedisSharedStorage, html + +BACKEND = os.environ.get("SWARM_BACKEND", "fastapi") +REDIS_URL = os.environ.get("SWARM_REDIS_URL") +FRAMES = int(os.environ.get("SWARM_FRAMES", "20")) +INTERVAL = float(os.environ.get("SWARM_INTERVAL", "0.1")) +WORK = float(os.environ.get("SWARM_WORK_MS", "0")) / 1000 +WS = BACKEND != "flask" + +app = Dash( + __name__, + backend=BACKEND, + shared_storage=RedisSharedStorage(url=REDIS_URL) + if REDIS_URL + else LocalSharedStorage, +) +app.server.secret_key = os.environ.get("SWARM_SECRET", "swarm") + + +def _section(name): + return html.Div( + [ + html.Button(name, id=f"{name}-btn", n_clicks=0), + html.Div(id=f"{name}-out"), + ] + ) + + +app.layout = html.Div( + [_section("http"), _section("ws"), _section("stream"), html.Div(id="ready")] +) + + +@app.callback( + Output("http-out", "children"), + Input("http-btn", "n_clicks"), + prevent_initial_call=True, +) +async def http_clicked(n): + if WORK: + await asyncio.sleep(WORK) + return f"clicked {n}" + + +if WS: + + @app.callback( + Output("ws-out", "children"), + Input("ws-btn", "n_clicks"), + prevent_initial_call=True, + websocket=True, + ) + async def ws_clicked(n): + if WORK: + await asyncio.sleep(WORK) + return f"clicked {n}" + + +@app.callback( + Output("stream-out", "children"), + Input("stream-btn", "n_clicks"), + prevent_initial_call=True, +) +async def stream(n): + for i in range(FRAMES): + yield f"{n}:{i}" + (" done" if i == FRAMES - 1 else "") + if i < FRAMES - 1: + await asyncio.sleep(INTERVAL) + + +server = app.server diff --git a/benchmarks/swarm/histogram.py b/benchmarks/swarm/histogram.py new file mode 100644 index 0000000000..37863f3690 --- /dev/null +++ b/benchmarks/swarm/histogram.py @@ -0,0 +1,79 @@ +"""Log-bucketed latency histograms that merge exactly across agents. + +Bucket ``i`` covers ``[GROWTH**i, GROWTH**(i+1))`` milliseconds, so every +recorded value is known to within 2%. Agents ship bucket counts instead of +raw samples: fifty machines' results stay small and add up bucket by bucket. +""" +import math + +GROWTH = 1.02 +_LOG = math.log(GROWTH) + + +class Histogram: + def __init__(self): + self.buckets = {} + self.count = 0 + self.total = 0.0 + self.min = math.inf + self.max = 0.0 + + def add(self, ms): + ms = max(ms, 0.001) + i = math.floor(math.log(ms) / _LOG) + self.buckets[i] = self.buckets.get(i, 0) + 1 + self.count += 1 + self.total += ms + self.min = min(self.min, ms) + self.max = max(self.max, ms) + + def merge(self, other): + for i, n in other.buckets.items(): + self.buckets[i] = self.buckets.get(i, 0) + n + self.count += other.count + self.total += other.total + self.min = min(self.min, other.min) + self.max = max(self.max, other.max) + return self + + def quantile(self, q): + if not self.count: + return None + rank = max(1, math.ceil(q * self.count)) + seen = 0 + for i in sorted(self.buckets): + seen += self.buckets[i] + if seen >= rank: + # Middle of the bucket, clamped to what was actually seen. + mid = GROWTH ** (i + 0.5) + return round(min(max(mid, self.min), self.max), 1) + return round(self.max, 1) + + def summary(self): + return { + "count": self.count, + "mean": round(self.total / self.count, 1) if self.count else None, + "p50": self.quantile(0.5), + "p95": self.quantile(0.95), + "p99": self.quantile(0.99), + "max": round(self.max, 1) if self.count else None, + } + + def to_json(self): + return { + "buckets": {str(i): n for i, n in sorted(self.buckets.items())}, + "count": self.count, + "total": self.total, + "min": self.min if self.count else None, + "max": self.max, + } + + @classmethod + def from_json(cls, data): + h = cls() + h.buckets = {int(i): n for i, n in data["buckets"].items()} + h.count = data["count"] + h.total = data["total"] + h.min = data["min"] if data["min"] is not None else math.inf + h.max = data["max"] + return h diff --git a/benchmarks/swarm/run.py b/benchmarks/swarm/run.py new file mode 100644 index 0000000000..92f10fbd6e --- /dev/null +++ b/benchmarks/swarm/run.py @@ -0,0 +1,243 @@ +"""Run a browser swarm: agents on many machines, one shared start. + +Each step runs every agent with ``--users`` headless browser users against +``--target`` (all agents start at the same wall-clock time and ramp +together), then merges their reports. Steps climb until the swarm sees +errors over 1% or a p95 over ``--max-p95`` ms. + +Agents run either here (``--local N``: N agent processes, for trying it out) +or on remote hosts over SSH (``--hosts hosts.txt``, one ``user@host`` per +line), each in the agent image, which every host must have pulled:: + + python -m benchmarks.swarm.run --target http://10.0.0.5:8050 \\ + --hosts hosts.txt --image ghcr.io/you/dash-swarm-agent \\ + --users 25 50 100 --out swarm-results.json + +Agent reports come back over the SSH session's stdout, so no copying is +needed. Hosts need NTP-synced clocks (the default on cloud VMs) and Docker. +""" +import argparse +import concurrent.futures +import json +import math +import os +import shlex +import subprocess +import sys +import time + +try: + from .aggregate import aggregate, table +except ImportError: + sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + from aggregate import aggregate, table # type: ignore[no-redef] + +HERE = os.path.dirname(os.path.abspath(__file__)) +CONTAINER = "dash-swarm-agent" + + +def _agent_args(args, users, start_at, agent_id): + return [ + "--target", + args.target, + "--users", + str(users), + "--scenario", + ",".join(args.scenario), + "--think", + str(args.think), + "--start-at", + str(start_at), + "--ramp", + str(args.ramp), + "--duration", + str(args.duration), + "--agent-id", + agent_id, + "--out", + "-", + ] + + +def _commands(args, users, start_at): + if args.local: + return [ + ( + f"local-{i}", + [ + sys.executable, + os.path.join(HERE, "agent.py"), + *_agent_args(args, users, start_at, f"local-{i}"), + ], + ) + for i in range(args.local) + ] + commands = [] + for host in _hosts(args.hosts): + docker = [ + "docker", + "run", + "--rm", + "--init", + "--name", + CONTAINER, + "--ipc=host", + "--network=host", + args.image, + *_agent_args(args, users, start_at, host), + ] + commands.append((host, _ssh(host, docker))) + return commands + + +def _hosts(path): + with open(path, encoding="utf-8") as f: + hosts = [h.strip() for h in f if h.strip() and not h.startswith("#")] + if not hosts: + raise SystemExit(f"no hosts in {path}") + return hosts + + +def _ssh(host, command): + return [ + "ssh", + "-o", + "BatchMode=yes", + "-o", + "ConnectTimeout=15", + "-o", + "ServerAliveInterval=15", + host, + " ".join(shlex.quote(a) for a in command), + ] + + +def _run(command, timeout): + host, cmd = command + try: + proc = subprocess.run( + cmd, + stdin=subprocess.DEVNULL, + capture_output=True, + text=True, + timeout=timeout, + check=False, + ) + except subprocess.TimeoutExpired: + if cmd[0] == "ssh": + # Killing ssh leaves the container (and its browsers) running, + # adding load to the next step. + subprocess.run( + _ssh(host, ["docker", "rm", "-f", CONTAINER]), + stdin=subprocess.DEVNULL, + capture_output=True, + timeout=60, + check=False, + ) + return host, None, "timed out" + lines = [line for line in proc.stdout.splitlines() if line.startswith("{")] + if proc.returncode != 0 or not lines: + return host, None, (proc.stderr.strip().splitlines() or ["no output"])[-1] + return host, json.loads(lines[-1]), None + + +def run_step(args, users): + # Long enough for ssh + docker start on every host before anyone begins. + start_at = time.time() + args.lead + commands = _commands(args, users, start_at) + timeout = args.lead + args.ramp + args.duration + 120 + reports, failed = [], [] + with concurrent.futures.ThreadPoolExecutor(len(commands)) as pool: + for host, report, error in pool.map(lambda c: _run(c, timeout), commands): + if report is None: + print(f" {host}: {error}", flush=True) + failed.append(host) + else: + reports.append(report) + result = aggregate(reports) + result["users_per_agent"] = users + result["failed_agents"] = failed + return result + + +def main(argv=None): + parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) + parser.add_argument("--target", required=True) + where = parser.add_mutually_exclusive_group(required=True) + where.add_argument("--local", type=int, help="agent processes on this machine") + where.add_argument("--hosts", help="file with one ssh host per line") + parser.add_argument("--image", default="dash-swarm-agent") + parser.add_argument( + "--users", nargs="+", type=int, default=[10], help="users per agent, per step" + ) + parser.add_argument( + "--scenario", + default="http,ws,stream", + type=lambda s: [x for x in s.split(",") if x], + ) + parser.add_argument("--think", type=float, default=1.0) + parser.add_argument("--ramp", type=float, default=30.0) + parser.add_argument("--duration", type=float, default=60.0) + parser.add_argument("--lead", type=float, default=20.0) + parser.add_argument("--max-p95", type=float, default=1000.0) + parser.add_argument("--out") + args = parser.parse_args(argv) + + steps = [] + for users in args.users: + print(f"--- {users} users per agent", flush=True) + result = run_step(args, users) + steps.append(result) + print(table(result), flush=True) + if args.out: + with open(args.out, "w", encoding="utf-8") as f: + json.dump( + { + "kind": "browser_swarm", + "target": args.target, + "params": { + k: getattr(args, k) + for k in ("scenario", "think", "ramp", "duration") + }, + "steps": steps, + }, + f, + indent=1, + ) + # A whole stream takes frames x interval by design; judge its first + # frame and the plain callbacks. + p95 = max( + ( + m["p95"] or 0 + for name, m in result["metrics"].items() + if name in ("http", "ws", "stream_first") + ), + default=math.inf, + ) + if ( + result["error_rate"] is None + or result["error_rate"] > 0.01 + or p95 > args.max_p95 + ): + print("saturated: stopping", flush=True) + break + # Past these the swarm is not delivering the load it reports, so a + # bigger step would only measure the load generators. + invalid = [ + label + for label, bad in ( + ("failed agents", result["failed_agents"]), + ("failed users", result["users_failed"] > 0.01 * result["users"]), + ("agent-bound", result["agent_bound"]), + ("late agents", result["late_agents"]), + ) + if bad + ] + if invalid: + print(f"load not delivered ({', '.join(invalid)}): stopping", flush=True) + break + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/unit/test_benchmarks_publish.py b/tests/unit/test_benchmarks_publish.py new file mode 100644 index 0000000000..a6c96e8163 --- /dev/null +++ b/tests/unit/test_benchmarks_publish.py @@ -0,0 +1,521 @@ +import json +import math + +import pytest + +from benchmarks import publish + + +def _point(browsers, p95, errors=0.0, client_bound=False): + return { + "browsers": browsers, + "latency_p50_ms": p95 / 2, + "latency_p95_ms": p95, + "latency_p99_ms": p95, + "first_frame_p50_ms": 3.0, + "first_frame_p95_ms": 5.0, + "frames_per_s": 10.0, + "streams_per_s": 1.0, + "streams": 10, + "frames": 200, + "errors": 0, + "incomplete": 0, + "error_rate": errors, + "server_cpu_cores": 0.5, + "server_rss_mb": 100, + "client_cpu_peak_pct": 99 if client_bound else 20, + "saturated": p95 > 1000 or errors > 0.01, + "client_bound": client_bound, + } + + +def _setup(name, points): + return { + "name": name, + "backend": "fastapi", + "server": "uvicorn", + "workers": 1, + "label": name, + "storage": "local", + "points": points, + } + + +def _run(setups): + return { + "kind": "streaming_load", + "meta": { + "date": "2026-10-06T00:00:00Z", + "commit": "abc", + "dash_version": "4.5.0", + "python": "3.12", + "cpu": "test cpu", + "cores": 8, + "memory_gb": 32, + "server_cpus": 4, + "client_cpus": 4, + "runner": "test", + }, + "params": { + "frames": 20, + "interval": 0.1, + "think": 1.0, + "ramp": 1, + "duration": 1, + "max_p95": 1000, + }, + "setups": setups, + } + + +def test_capacity_stops_at_first_slow_or_failing_point(): + setup = _setup("a", [_point(100, 10), _point(500, 400), _point(1000, 10)]) + assert publish.capacity(setup) == (100, False) + assert publish.capacity( + _setup("b", [_point(100, 10), _point(500, 10, errors=0.05)]) + ) == (100, False) + + +def test_capacity_marks_sweeps_that_never_saturated(): + assert publish.capacity(_setup("a", [_point(100, 10), _point(500, 20)])) == ( + 500, + True, + ) + + +def test_capacity_is_a_lower_bound_when_clients_ran_out_of_cpu(): + setup = _setup("a", [_point(100, 10), _point(500, 20, client_bound=True)]) + assert publish.capacity(setup) == (100, True) + + +README = """# Dash + +### Performance + + +No published run yet. + + +### Next +""" + + +def _readme(tmp_path): + path = tmp_path / "README.md" + path.write_text(README) + return path + + +def _renderer_results(tmp_path): + path = tmp_path / "renderer.json" + path.write_text( + json.dumps( + { + "callback_fanout": { + "fanout_ms": { + "n": 3, + "min": 1, + "median": 80.0, + "p90": 90.0, + "max": 95.0, + "growth": 1.0, + } + } + } + ) + ) + return path + + +def test_publish_commits_numbers_history_report_and_readme(tmp_path): + streaming = tmp_path / "streaming.json" + streaming.write_text( + json.dumps(_run([_setup("fastapi-uvicorn-w1", [_point(100, 10)])])) + ) + renderer = _renderer_results(tmp_path) + readme = _readme(tmp_path) + out = tmp_path / "results" + + for date in ("2026-10-01T00:00:00Z", "2026-10-02T00:00:00Z"): + publish.main( + [ + "--dir", + str(out), + "--readme", + str(readme), + "--renderer", + str(renderer), + "--streaming", + str(streaming), + "--date", + date, + ] + ) + + history = json.loads((out / "renderer-history.json").read_text()) + assert [h["date"][:10] for h in history] == ["2026-10-01", "2026-10-02"] + assert history[0]["results"] == {"callback_fanout": {"fanout_ms": {"median": 80.0}}} + latest = json.loads((out / "renderer-latest.json").read_text()) + assert latest["results"]["callback_fanout"]["fanout_ms"]["p90"] == 90.0 + assert json.loads((out / "streaming-latest.json").read_text())["setups"] + sections = json.loads((out / "report.json").read_text())["sections"] + assert [s["id"] for s in sections] == ["streaming", "renderer"] + charts = [c["id"] for s in sections for c in s["charts"]] + assert charts == [ + "streaming-capacity", + "streaming-latency", + "streaming-first-frame", + "streaming-cpu", + "renderer-latest", + "renderer-trend", + ] + for s in sections: + assert s["table"]["rows"] + for c in s["charts"]: + assert c["figure"]["data"] and c["figure"]["layout"] + text = readme.read_text() + assert "| fastapi-uvicorn-w1 | - | - | 100+ |" in text + assert "No published run yet." not in text + assert text.endswith("\n\n### Next\n") + + +def test_lines_figure_marks_the_threshold_in_log_units(): + chart = publish.streaming_charts( + _run([_setup("a", [_point(100, 10), _point(500, 20)])]) + )[1] + fig = publish.figure(chart) + assert [t["x"] for t in fig["data"]] == [[100, 500]] + assert fig["layout"]["xaxis"]["ticktext"] == ["100", "500"] + assert fig["layout"]["shapes"][0]["y0"] == publish.CAPACITY_P95_MS + assert fig["layout"]["annotations"][0]["y"] == pytest.approx( + math.log10(publish.CAPACITY_P95_MS) + ) + + +def test_trend_figure_shows_one_scenario_at_a_time(): + entry = { + "date": "2026-10-01T00:00:00Z", + "commit": "abc", + "runner": "test", + "results": { + "a": {"m": {"median": 10.0, "p90": 12.0, "max": 13.0}}, + "b": {"m": {"median": 20.0, "p90": 22.0, "max": 23.0}}, + }, + } + trend = publish.renderer_charts(entry, [entry])[1] + fig = publish.figure(trend) + # Slowest scenario first. + assert [(t["name"], t["visible"]) for t in fig["data"]] == [ + ("b (m)", True), + ("a (m)", False), + ] + buttons = fig["layout"]["updatemenus"][0]["buttons"] + assert [b["args"][0]["visible"] for b in buttons] == [[True, False], [False, True]] + # A single run gets a week of axis, not a millisecond. + assert "xaxis.range" in buttons[0]["args"][1] + + +def test_named_input_must_exist(tmp_path): + with pytest.raises(FileNotFoundError): + publish.main( + [ + "--dir", + str(tmp_path / "results"), + "--readme", + str(_readme(tmp_path)), + "--streaming", + str(tmp_path / "typo.json"), + ] + ) + + +def test_readme_names_each_run_and_sorts_unknown_backends_last(): + callbacks = _cb_run() + streaming = _run( + [{**_setup("starlette-uvicorn-w1", [_point(100, 10)]), "backend": "starlette"}] + ) + streaming["meta"] = {**streaming["meta"], "date": "2026-10-07T00:00:00Z"} + lines = publish.readme_table(streaming, callbacks).splitlines() + assert lines[-3].startswith("| starlette-uvicorn-w1 |") + assert lines[-1].endswith( + "Measured 2026-10-06: server on 4 cores (test), Dash 4.5.0;" + " 2026-10-07: server on 4 cores (test), Dash 4.5.0." + ) + + +def test_readme_needs_its_markers(tmp_path): + readme = tmp_path / "README.md" + readme.write_text("# Dash\n") + with pytest.raises(SystemExit): + publish.update_readme(str(readme), None, None) + + +def _cb_point(browsers, p95, calls_per_s=100.0, cores=0.2, client_bound=False): + return { + "browsers": browsers, + "rtt_p50_ms": p95 / 2, + "rtt_p95_ms": p95, + "rtt_p99_ms": p95, + "calls_per_s": calls_per_s, + "calls": int(calls_per_s * 20), + "errors": 0, + "failed_browsers": 0, + "error_rate": 0.0, + "server_cpu_cores": cores, + "server_rss_mb": 100, + "client_cpu_peak_pct": 99 if client_bound else 20, + "saturated": p95 > 1000, + "client_bound": client_bound, + } + + +def _cb_run(): + def setup(name, backend, workers, transport, points): + return { + "name": name, + "backend": backend, + "server": "uvicorn", + "workers": workers, + "label": name, + "transport": transport, + "points": points, + } + + run = _run( + [ + setup( + "fastapi-w1-http", + "fastapi", + 1, + "http", + [_cb_point(100, 2), _cb_point(1000, 300, 900, 0.9)], + ), + setup( + "fastapi-w1-ws", + "fastapi", + 1, + "ws", + [_cb_point(100, 1), _cb_point(1000, 5, 1000, 0.3)], + ), + setup("quart-w4-ws", "quart", 4, "ws", [_cb_point(100, 1)]), + ] + ) + run["kind"] = "callback_load" + run["params"] = { + "kind": "sync", + "work_ms": 0.0, + "think": 1.0, + "ramp": 1, + "duration": 1, + "max_p95": 1000, + } + return run + + +def test_callback_charts_split_by_transport_and_workers(): + charts = {c["id"]: c for c in publish.callbacks_charts(_cb_run())} + assert set(charts) == { + "callbacks-capacity", + "callbacks-rtt-w1", + "callbacks-rtt-w4", + "callbacks-cpu", + } + bars = {b["label"]: b for b in charts["callbacks-capacity"]["bars"]} + assert bars["fastapi-w1-http"]["text"] == "100" + assert bars["fastapi-w1-ws"]["text"] == "1,000+" + assert bars["fastapi-w1-ws"]["hue"] != bars["fastapi-w1-http"]["hue"] + w1 = charts["callbacks-rtt-w1"]["traces"] + assert [t["dash"] for t in w1] == ["solid", "dash"] + cpu = {b["label"]: b["value"] for b in charts["callbacks-cpu"]["bars"]} + # Only setups measured at 1,000 browsers are compared. + assert cpu == {"fastapi-w1-ws": 0.3, "fastapi-w1-http": 1.0} + + +def test_readme_table_has_one_row_per_server(): + streaming = _run( + [ + { + **_setup("fastapi-uvicorn-w1", [_point(100, 10), _point(500, 400)]), + "label": "FastAPI, uvicorn 1 worker", + } + ] + ) + callbacks = _cb_run() + for s in callbacks["setups"]: + s["label"] = { + "fastapi-w1-http": "FastAPI, uvicorn 1 worker, HTTP", + "fastapi-w1-ws": "FastAPI, uvicorn 1 worker, websocket", + "quart-w4-ws": "Quart, uvicorn 4 workers, websocket", + }[s["name"]] + lines = publish.readme_table(streaming, callbacks).splitlines() + assert lines[2:4] == [ + "| FastAPI, uvicorn 1 worker | 100 | 1,000+ | 100 |", + "| Quart, uvicorn 4 workers | - | 100+ | - |", + ] + assert lines[-1].endswith( + "Measured 2026-10-06: server on 4 cores (test), Dash 4.5.0." + ) + + +def test_publish_keeps_committed_callbacks_when_not_given(tmp_path): + callbacks = tmp_path / "callbacks.json" + callbacks.write_text(json.dumps(_cb_run())) + out = tmp_path / "results" + readme = _readme(tmp_path) + args = ["--dir", str(out), "--readme", str(readme)] + publish.main([*args, "--callbacks", str(callbacks)]) + publish.main(args) + sections = json.loads((out / "report.json").read_text())["sections"] + assert [s["title"] for s in sections] == ["Callbacks under load: HTTP vs websocket"] + assert "| fastapi-w1-http | 100 | 1,000+ | - |" in readme.read_text() + history = json.loads((out / "callbacks-history.json").read_text()) + assert len(history) == 1 + assert history[0]["capacity"]["fastapi-w1-http"] == 100 + + +def _stats(median): + return {"n": 3, "min": 1, "median": median, "p90": median, "max": median} + + +def test_median_change_must_beat_both_noise_floors(): + old = {"results": {"a": {"m": _stats(100.0)}, "b": {"m": _stats(4.0)}}} + + def changes(new): + return publish.median_changes("renderer", old, new, 5.0) + + # +20 ms but only 20%: noise. + assert changes({"a": {"m": _stats(120.0)}}) == [] + # +100% but only 4 ms: noise. + assert changes({"b": {"m": _stats(8.0)}}) == [] + assert changes({"a": {"m": _stats(130.0)}}) == ["renderer a (m): 100.0 -> 130.0 ms"] + assert changes({"c": {"m": _stats(7.0)}}) == ["renderer c (m): new, 7.0 ms"] + + +def _serialization_run(div_ms): + def stats(median): + return {"median": median, "min": median, "max": median, "bytes": 1000} + + return { + "meta": { + "traces": 10, + "points": 20000, + "repeat": 7, + "versions": { + "python": "3.12.0", + "dash": "4.5.0", + "plotly": "6.7.0", + "orjson": "3.11.0", + }, + "labels": { + "list_float+none": "list of float, figure alone", + "list_float+div": "list of float, + html.Div", + }, + }, + "results": { + "list_float+none": {"orjson": stats(6.0), "json": stats(50.0)}, + "list_float+div": {"orjson": stats(div_ms), "json": stats(51.0)}, + }, + } + + +def test_serialization_section_compares_each_case_to_the_figure_alone(tmp_path): + out = tmp_path / "results" + run = tmp_path / "serialization.json" + changes = tmp_path / "changes.md" + args = ["--dir", str(out), "--readme", str(_readme(tmp_path)), "--only-significant"] + + run.write_text(json.dumps(_serialization_run(40.0))) + publish.main([*args, "--serialization", str(run), "--date", "2026-10-01T00:00:00Z"]) + sections = json.loads((out / "report.json").read_text())["sections"] + assert [s["id"] for s in sections] == ["serialization"] + section = sections[0] + assert [c["id"] for c in section["charts"]] == [ + "serialization-orjson", + "serialization-json", + "serialization-trend", + ] + orjson_bars = section["charts"][0]["figure"]["data"] + assert [t["name"] for t in orjson_bars] == [ + "figure alone", + "same figure, plus one non-native value", + ] + assert orjson_bars[1]["x"] == [40.0] + assert section["table"]["rows"][1][:3] == [ + "list of float, + html.Div", + "40.0", + "6.7x", + ] + latest = json.loads((out / "serialization-latest.json").read_text()) + assert latest["versions"]["plotly"] == "6.7.0" + assert latest["date"] == "2026-10-01T00:00:00Z" + + # A plotly fix that drops the + html.Div case to the figure-alone time. + fixed = _serialization_run(6.5) + fixed["meta"]["versions"]["plotly"] = "6.8.0" + run.write_text(json.dumps(fixed)) + publish.main( + [ + *args, + "--serialization", + str(run), + "--date", + "2026-10-08T00:00:00Z", + "--changes", + str(changes), + ] + ) + assert changes.read_text() == ( + "- serialization list_float+div (orjson): 40.0 -> 6.5 ms\n" + "- serialization measured with plotly 6.8.0 (was 6.7.0)\n" + ) + history = json.loads((out / "serialization-history.json").read_text()) + assert [h["results"]["list_float+div"]["orjson"] for h in history] == [ + {"median": 40.0}, + {"median": 6.5}, + ] + trend = json.loads((out / "report.json").read_text())["sections"][0]["charts"][2] + assert ( + trend["figure"]["data"][0]["name"] == "list of float, figure alone, with orjson" + ) + + +def test_load_change_is_a_capacity_step(): + old = _run([_setup("a", [_point(100, 10), _point(500, 400)])]) + same = _run([_setup("a", [_point(100, 50), _point(500, 900)])]) + assert publish.load_changes("streaming", old, same, "latency_p95_ms") == [] + better = _run([_setup("a", [_point(100, 10), _point(500, 20)])]) + assert publish.load_changes("streaming", old, better, "latency_p95_ms") == [ + "streaming a: 100 -> 500+ browsers" + ] + assert publish.load_changes("streaming", None, old, "latency_p95_ms") == [ + "streaming: first published run" + ] + + +def test_only_significant_runs_are_committed(tmp_path): + out = tmp_path / "results" + readme = _readme(tmp_path) + run = tmp_path / "streaming.json" + changes = tmp_path / "changes.md" + args = ["--dir", str(out), "--readme", str(readme), "--only-significant"] + + run.write_text(json.dumps(_run([_setup("a", [_point(100, 10)])]))) + publish.main([*args, "--streaming", str(run), "--changes", str(changes)]) + assert changes.read_text() == "- streaming: first published run\n" + first = readme.read_text() + + # Slower but still inside the same capacity step: nothing is written. + later = _run([_setup("a", [_point(100, 60)])]) + later["meta"]["date"] = "2026-10-13T00:00:00Z" + run.write_text(json.dumps(later)) + publish.main([*args, "--streaming", str(run), "--changes", str(changes)]) + assert changes.read_text() == "" + assert readme.read_text() == first + assert len(json.loads((out / "streaming-history.json").read_text())) == 1 + + later["setups"][0]["points"].append(_point(500, 20)) + run.write_text(json.dumps(later)) + publish.main([*args, "--streaming", str(run), "--changes", str(changes)]) + assert changes.read_text() == "- streaming a: 100+ -> 500+ browsers\n" + assert "| a | - | - | 500+ |" in readme.read_text() + assert len(json.loads((out / "streaming-history.json").read_text())) == 2 diff --git a/tests/unit/test_benchmarks_serialization.py b/tests/unit/test_benchmarks_serialization.py new file mode 100644 index 0000000000..3537cd4330 --- /dev/null +++ b/tests/unit/test_benchmarks_serialization.py @@ -0,0 +1,34 @@ +import json + +from benchmarks.serialization import bench + + +def test_each_engine_runs_every_case_and_json_never_loads_orjson(tmp_path): + out = tmp_path / "results.json" + cases = [ + "list_float+none", + "list_float+div", + "list_datetime+figure", + "numpy+layout", + ] + bench.main( + [ + "--traces", + "1", + "--points", + "50", + "--repeat", + "1", + "--out", + str(out), + "--case", + *cases, + ] + ) + run = json.loads(out.read_text()) + assert run["meta"]["orjson_loaded"] == {"orjson": True, "json": False} + assert list(run["results"]) == cases + for by_engine in run["results"].values(): + assert set(by_engine) == {"orjson", "json"} + assert by_engine["orjson"]["median"] > 0 + assert run["meta"]["labels"]["list_float+div"] == "list of float, + html.Div" diff --git a/tests/unit/test_benchmarks_swarm.py b/tests/unit/test_benchmarks_swarm.py new file mode 100644 index 0000000000..5c973e220e --- /dev/null +++ b/tests/unit/test_benchmarks_swarm.py @@ -0,0 +1,70 @@ +"""The swarm's histograms must merge to the same percentiles as the raw data.""" +import math +import random + +from benchmarks.swarm.aggregate import aggregate +from benchmarks.swarm.histogram import GROWTH, Histogram + + +def _exact(values, q): + return sorted(values)[math.ceil(q * len(values)) - 1] + + +def test_quantiles_within_bucket_resolution(): + rng = random.Random(1) + values = [rng.lognormvariate(3, 1) for _ in range(5000)] + h = Histogram() + for v in values: + h.add(v) + for q in (0.5, 0.95, 0.99): + assert abs(h.quantile(q) / _exact(values, q) - 1) <= GROWTH - 1 + assert h.count == 5000 + + +def test_merged_agents_match_one_histogram(): + rng = random.Random(2) + parts = [[rng.uniform(1, 500) for _ in range(300)] for _ in range(4)] + whole = Histogram() + reports = [] + for i, part in enumerate(parts): + h = Histogram() + for v in part: + h.add(v) + whole.add(v) + reports.append( + { + "agent": f"a{i}", + "users": 10, + "duration": 60, + "actions": len(part), + "errors": i, + "cpu_p90_pct": 90 if i == 3 else 40, + "late_s": 3 if i == 2 else 0, + "histograms": {"http": h.to_json()}, + } + ) + result = aggregate(reports) + assert result["metrics"]["http"] == whole.summary() + assert result["actions"] == 1200 + assert result["errors"] == 6 + assert result["users"] == 40 + assert result["agent_bound"] == ["a3"] + assert result["late_agents"] == ["a2"] + + +def test_empty_histogram(): + assert Histogram().quantile(0.5) is None + assert Histogram.from_json(Histogram().to_json()).summary()["count"] == 0 + + +def test_quantile_edges(): + one = Histogram() + one.add(7.3) + assert one.summary()["p50"] == one.summary()["p99"] == 7.3 + same = Histogram() + for _ in range(10): + same.add(42.0) + assert same.quantile(0.5) == same.quantile(0.99) == 42.0 + tiny = Histogram() + tiny.add(0) + assert tiny.quantile(0.5) == 0.0