diff --git a/.gitignore b/.gitignore index f5da5e1..bd5b96e 100644 --- a/.gitignore +++ b/.gitignore @@ -11,3 +11,4 @@ build/ htmlcov/ .env .playwright-cli/ +bench/results-quick.json diff --git a/3rdparty_licenses.md b/3rdparty_licenses.md index 1850aa4..7c193ac 100644 --- a/3rdparty_licenses.md +++ b/3rdparty_licenses.md @@ -27,16 +27,34 @@ license is listed separately below. | https://hub.docker.com/_/postgres | postgres (container image) | PostgreSQL | https://www.postgresql.org/about/licence/ | | https://hub.docker.com/_/python | python (container image, demo only) | PSF-2.0 | https://docs.python.org/3/license.html | | https://github.com/astral-sh/uv | uv (container image, build stage of the demo) | MIT OR Apache-2.0 | https://github.com/astral-sh/uv/blob/main/LICENSE-MIT | +| https://pypi.org/project/celery/ | celery (benchmark comparison only, `bench` group) | BSD-3-Clause | https://github.com/celery/celery/blob/main/LICENSE | +| https://pypi.org/project/redis/ | redis-py (benchmark comparison only, `bench` group) | MIT | https://github.com/redis/redis-py/blob/master/LICENSE | +| https://hub.docker.com/r/valkey/valkey | valkey (container image, benchmark comparison only) | BSD-3-Clause | https://github.com/valkey-io/valkey/blob/unstable/COPYING | ## Development-only transitive dependencies -These are installed only by `uv sync --all-extras` for local development and CI. They are -not runtime dependencies, are not imported by pgqueue and are not distributed with it -(neither in the package nor in the demo image). +The `pathspec` row is installed only by `uv sync --all-extras` for local development and +CI. The rest belong to the optional `bench` dependency group (`uv sync --group bench`), +which only `make bench` uses for the Celery comparison. None of them are runtime +dependencies, none are imported by pgqueue and none are distributed with it (neither in +the package nor in the demo image). | source | name | license type | license link | |---|---|---|---| | https://pypi.org/project/pathspec/ | pathspec 1.1.1 (transitive, via mypy; dev/CI only, not distributed) | MPL-2.0 | https://github.com/cpburnz/python-pathspec/blob/master/LICENSE | +| https://pypi.org/project/kombu/ | kombu (transitive, via celery; benchmark only) | BSD-3-Clause | https://github.com/celery/kombu/blob/main/LICENSE | +| https://pypi.org/project/amqp/ | amqp (transitive, via kombu; benchmark only) | BSD-3-Clause | https://github.com/celery/py-amqp/blob/main/LICENSE | +| https://pypi.org/project/billiard/ | billiard (transitive, via celery; benchmark only) | BSD-3-Clause | https://github.com/celery/billiard/blob/main/LICENSE.txt | +| https://pypi.org/project/vine/ | vine (transitive, via celery; benchmark only) | BSD-3-Clause | https://github.com/celery/vine/blob/main/LICENSE | +| https://pypi.org/project/click-didyoumean/ | click-didyoumean (transitive, via celery; benchmark only) | MIT | https://github.com/click-contrib/click-didyoumean/blob/master/LICENSE | +| https://pypi.org/project/click-plugins/ | click-plugins (transitive, via celery; benchmark only) | BSD-3-Clause | https://github.com/click-contrib/click-plugins/blob/master/LICENSE.txt | +| https://pypi.org/project/click-repl/ | click-repl (transitive, via celery; benchmark only) | MIT | https://github.com/click-contrib/click-repl/blob/main/LICENSE | +| https://pypi.org/project/prompt-toolkit/ | prompt-toolkit (transitive, via click-repl; benchmark only) | BSD-3-Clause | https://github.com/prompt-toolkit/python-prompt-toolkit/blob/main/LICENSE | +| https://pypi.org/project/wcwidth/ | wcwidth (transitive, via prompt-toolkit; benchmark only) | MIT | https://github.com/jquast/wcwidth/blob/master/LICENSE | +| https://pypi.org/project/python-dateutil/ | python-dateutil (transitive, via celery; benchmark only) | Apache-2.0 AND BSD-3-Clause | https://github.com/dateutil/dateutil/blob/master/LICENSE | +| https://pypi.org/project/six/ | six (transitive, via python-dateutil; benchmark only) | MIT | https://github.com/benjaminp/six/blob/main/LICENSE | +| https://pypi.org/project/tzlocal/ | tzlocal (transitive, via celery; benchmark only) | MIT | https://github.com/regebro/tzlocal/blob/master/LICENSE.txt | +| https://pypi.org/project/tzdata/ | tzdata (transitive, via celery and kombu; benchmark only) | Apache-2.0 | https://github.com/python/tzdata/blob/master/LICENSE | ## Notes diff --git a/CHANGELOG.md b/CHANGELOG.md index 824d017..ab6cab6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- `make bench`: a reproducible benchmark with a 1/2/4/8 worker sweep, enqueue and pickup latency + percentiles, payload sizes, repeated runs and a Celery + Valkey comparison; results, charts + and caveats in `docs/benchmarks.md`. It replaces the earlier single-run numbers. - Dashboard favicon set (SVG, ICO, 32 px PNG, Apple touch icon) and `theme-color` tags; the admin app answers `/favicon.ico`. - Dashboard rebuilt: live updates over server-sent events (polling fallback), throughput diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..36cf1dc --- /dev/null +++ b/Makefile @@ -0,0 +1,32 @@ +# Benchmark helpers. Needs Docker, uv and a Linux or macOS shell. +# +# make bench start Postgres and Valkey, run the full benchmark, draw the chart +# make bench-quick the same with tiny numbers, to check the setup works +# make bench-chart redraw docs/benchmarks.png from bench/results.json +# make bench-down stop the benchmark containers +# +# Override the ports if the defaults are taken, e.g. `make bench PG_PORT=55461`. +PG_PORT ?= 55432 +REDIS_PORT ?= 56379 +export PGQUEUE_BENCH_DSN ?= postgresql://pgqueue:pgqueue@localhost:$(PG_PORT)/pgqueue_bench +export BENCH_REDIS_URL ?= redis://localhost:$(REDIS_PORT)/0 +export PGQUEUE_DSN ?= $(PGQUEUE_BENCH_DSN) +BENCH = uv run --group bench python -m bench.bench + +.PHONY: bench bench-quick bench-up bench-down bench-chart + +bench-up: + PG_PORT=$(PG_PORT) REDIS_PORT=$(REDIS_PORT) docker compose --profile bench up -d --wait postgres valkey + +bench-down: + docker compose --profile bench stop postgres valkey + +bench: bench-up + $(BENCH) $(ARGS) + $(MAKE) bench-chart + +bench-quick: bench-up + $(BENCH) --quick --out bench/results-quick.json $(ARGS) + +bench-chart: + uv run --group bench python -m bench.chart diff --git a/README.md b/README.md index e593009..f626240 100644 --- a/README.md +++ b/README.md @@ -400,49 +400,51 @@ What happens when… ## Benchmarks -Measured with [`bench/bench.py`](bench/bench.py) on the author's laptop, so read them as an -order of magnitude, not a promise: - -- **CPU** 13th Gen Intel Core i5-13500H (12 cores / 16 threads), **RAM** 14 GiB, NVMe SSD -- PostgreSQL 17.11 in Docker with the **default configuration** (`shared_buffers = 128MB`, - `synchronous_commit = on`) on the same machine as the workers and the load generator -- Python 3.12.13, Linux 7.0; the machine was not idle (load average ≈ 5 from other - processes during the run) -- Workers are separate processes started through the real CLI; the handler is a no-op, so - this measures the queue's own overhead (claim, lease, completion), not your work - -| Scenario | Jobs | Time | Throughput | -|---|---:|---:|---:| -| `enqueue()`, one client, one job per transaction | 5,000 | 10.2 s | **490 jobs/s** | -| `enqueue()`, 10 concurrent clients | 20,000 | 13.9 s | **1,438 jobs/s** | -| `enqueue_many()`, batches of 1,000 | 100,000 | 2.5 s | **40,496 jobs/s** | -| drain a 50k backlog, 1 worker × 32 slots | 50,000 | 34.8 s | **1,436 jobs/s** | -| drain a 50k backlog, 2 workers × 32 slots | 50,000 | 19.0 s | **2,628 jobs/s** | -| drain a 50k backlog, 4 workers × 32 slots | 50,000 | 12.0 s | **4,185 jobs/s** | -| drain a 50k backlog, 8 workers × 32 slots | 50,000 | 7.6 s | **6,625 jobs/s** | - -Pickup latency (enqueue → handler starts) for a steady 100 jobs/s into idle workers: - -| Wake-up mechanism | p50 | p95 | p99 | -|---|---:|---:|---:| -| LISTEN/NOTIFY (+ 1 s polling fallback) | **4.1 ms** | 6.7 ms | 12.5 ms | -| polling only, 1 s interval | 483 ms | 932 ms | 991 ms | - -Observations: - -- Single-job enqueue is bound by commit latency (an fsync per transaction). Batching with - `enqueue_many` is ~80× faster; enqueueing inside a transaction you are committing anyway - costs nothing extra. -- Throughput scales nearly linearly up to 4 worker processes and keeps growing at 8; the - ceiling on this machine is Postgres's per-row UPDATE + commit cost (two transactions per - job: claim and completion). +`make bench` runs a reproducible benchmark against a real PostgreSQL and, next to it, +Celery on Valkey (the BSD-licensed Redis fork) with the same no-op handler, the same +2 KiB JSON payload and 32 job slots per worker process. Workers are separate processes +started through the real CLI. Each cell is repeated 5 times. The full method, every table, +the SKIP LOCKED explanation and the caveats are in [`docs/benchmarks.md`](docs/benchmarks.md); +raw per-run numbers are in [`bench/results.json`](bench/results.json). + +Read the numbers with these caveats first: one 13th Gen Intel i5-13500H laptop (16 threads, +14 GiB RAM) running Postgres 17 with default settings, Valkey, the workers and the load +generator together, **and busy with unrelated work during the runs** (1-minute load average +between 2 and 38). Run-to-run spread is large, so cells are the median with the minimum and +maximum run in brackets. + +| Backlog of 20,000 jobs, jobs/s | pgqueue | Celery + Valkey | +|---:|---:|---:| +| 1 worker process | 825 (725-1,399) | 349 (280-499) | +| 2 worker processes | 1,702 (1,427-2,528) | 722 (574-1,268) | +| 4 worker processes | 3,915 (2,444-5,475) | 1,765 (990-3,354) | +| 8 worker processes | 4,161 (2,589-8,142) | 3,422 (1,683-6,747) | + +![Backlog throughput by number of worker processes](docs/benchmarks-throughput.svg) + +What this does and does not show: + +- Within each run the two systems were measured seconds apart under the same load, and + pgqueue was faster in every run at 1, 2 and 4 workers and in 4 of 5 runs at 8 workers. + The absolute numbers moved by a factor of two to three with how busy the machine was. +- Enqueue is where Postgres pays for durability: a single `enqueue()` call (one fsynced + commit) took a median of 4.4 ms at 184 jobs/s for one client, against 1.7 ms and 566 + jobs/s for a Celery publish to a Valkey without persistence. `enqueue_many` inserted + 1,000 jobs per call at 6,188 jobs/s (median; range 5,181 to 14,792). +- Pickup latency (enqueue to handler start, 200 jobs/s into idle workers) had a p50 of + 4.5 to 7.3 ms for pgqueue and 3.5 to 4.5 ms for Celery depending on the worker count; + pgqueue's p99 was 15 to 38 ms, Celery's 6 to 11 ms. +- Not measured: a real handler workload, other Celery pools (prefork, gevent), RQ, separate + database and worker hosts, tuned Postgres, Valkey with persistence, tables of millions + of rows. - The benchmark caught a real bug: the first version of the claim query filtered with `queue = ANY($1)`, which cannot use the index for ordering, and degraded to a sequential scan + sort of the whole backlog (22 ms per claim at 50k jobs). Walking the index per queue through a `LATERAL` subquery brought that to ~0.1 ms and doubled - throughput. Raw numbers are in [`bench/results.json`](bench/results.json). + throughput. -Reproduce: `docker compose up -d postgres && uv run python -m bench.bench`. +Reproduce: `make bench` (needs Docker and uv; about an hour), then `make bench-chart` to +redraw the charts. `make bench-quick` checks the setup in a few minutes. ## Design decisions and trade-offs diff --git a/bench/bench.py b/bench/bench.py index e253bde..58a6880 100644 --- a/bench/bench.py +++ b/bench/bench.py @@ -1,13 +1,23 @@ -"""Measure pgqueue throughput and latency against a real Postgres. - - docker compose up -d postgres - uv run python -m bench.bench # full run, prints a Markdown report - uv run python -m bench.bench --quick # smaller numbers, for a smoke test - -Every scenario runs against a freshly truncated ``pgqueue_bench`` database. -Workers are separate OS processes started with the real CLI, exactly as in -production, with a no-op handler so that the numbers measure the queue's -own overhead (claim, heartbeat, completion), not handler work. +"""Reproducible benchmark: pgqueue on PostgreSQL, optionally next to Celery on Redis. + + make bench # full run: ~30 min, writes bench/results.json + make bench-quick # smoke test with tiny numbers, results are meaningless + uv run --group bench python -m bench.bench --runs 3 --skip-celery + +What is measured (all on one machine, workers are separate OS processes): + +* enqueue: latency per call and jobs/s for one client, ten concurrent clients and + ``enqueue_many`` (pgqueue only, Celery has no bulk publish) +* drain: a backlog is loaded first, then 1/2/4/8 worker processes (32 slots each) empty + it. Throughput is taken from the completion timestamps of the 20th to the 95th + percentile job, so worker start-up stagger and the straggling tail are excluded. +* pickup: jobs arrive at a fixed rate into idle workers; latency is the time from + enqueue to the first line of the handler. +* payload size: the drain scenario at 4 workers with 256 B, 2 KiB and 16 KiB payloads. + +The handler is a no-op, so the numbers describe the queue, not your work. Every +scenario is repeated ``--runs`` times, interleaving the systems, and the raw +per-run numbers are kept in the results file. """ from __future__ import annotations @@ -20,33 +30,53 @@ import statistics import subprocess import sys +import threading import time -from collections.abc import Awaitable, Callable -from dataclasses import asdict, dataclass +from concurrent.futures import ThreadPoolExecutor from pathlib import Path +from typing import Any from urllib.parse import urlsplit, urlunsplit import asyncpg -from bench.tasks import Noop, app, noop +from bench.payload import SIZES, make_payload +from bench.tasks import Email, app, noop from pgqueue.migrate import upgrade ROOT = Path(__file__).resolve().parent.parent DSN = os.environ.get( "PGQUEUE_BENCH_DSN", "postgresql://pgqueue:pgqueue@localhost:55432/pgqueue_bench" ) +REDIS_URL = os.environ.get("BENCH_REDIS_URL", "redis://localhost:56379/0") +SLOTS = 32 # concurrent jobs per worker process, for both systems +WINDOW = (0.20, 0.95) # share of the backlog that defines the steady-state window + +Record = dict[str, Any] + + +def summarize(samples_ms: list[float]) -> dict[str, float]: + q = statistics.quantiles(samples_ms, n=100, method="inclusive") + return { + "p50": q[49], + "p95": q[94], + "p99": q[98], + "max": max(samples_ms), + "n": float(len(samples_ms)), + } + +def steady_rate(finished: list[float]) -> float: + """Jobs/s between the 20th and the 95th percentile completion.""" + finished = sorted(finished) + lo, hi = int(len(finished) * WINDOW[0]), int(len(finished) * WINDOW[1]) + return (hi - lo) / (finished[hi] - finished[lo]) -@dataclass -class Result: - scenario: str - jobs: int - seconds: float - detail: str = "" - @property - def rate(self) -> float: - return self.jobs / self.seconds +def log(msg: str) -> None: + print(f"[{time.strftime('%H:%M:%S')}] {msg}", file=sys.stderr, flush=True) + + +# -- pgqueue side --------------------------------------------------------------------------------- async def prepare_database() -> None: @@ -61,216 +91,351 @@ async def prepare_database() -> None: await asyncio.to_thread(upgrade, DSN) -async def reset() -> None: +async def pg_reset() -> None: await app.pool.execute("TRUNCATE pgqueue_jobs, pgqueue_workers RESTART IDENTITY") await app.pool.execute("VACUUM ANALYZE pgqueue_jobs") -async def timed(fn: Callable[[], Awaitable[object]]) -> float: - start = time.perf_counter() - await fn() - return time.perf_counter() - start - - -# -- enqueue ------------------------------------------------------------------------------------ - - -async def bench_enqueue_sequential(n: int) -> Result: - await reset() - - async def run() -> None: - for i in range(n): - await noop.enqueue(Noop(n=i)) - - return Result("enqueue, one at a time (1 client)", n, await timed(run)) - - -async def bench_enqueue_concurrent(n: int, clients: int) -> Result: - await reset() - - async def client(offset: int) -> None: - for i in range(offset, n, clients): - await noop.enqueue(Noop(n=i)) - - async def run() -> None: - await asyncio.gather(*(client(c) for c in range(clients))) - - return Result(f"enqueue, one at a time ({clients} concurrent clients)", n, await timed(run)) - - -async def bench_enqueue_bulk(n: int, batch: int) -> Result: - await reset() - - async def run() -> None: - for start in range(0, n, batch): - await noop.enqueue_many([Noop(n=i) for i in range(start, min(n, start + batch))]) - - return Result(f"enqueue_many, batches of {batch}", n, await timed(run)) - - -# -- processing --------------------------------------------------------------------------------- - - -def start_workers(count: int, concurrency: int, *extra: str) -> list[subprocess.Popen[bytes]]: - # 8 pooled connections per worker (+1 for LISTEN) is plenty for no-op jobs and - # keeps 8 workers under Postgres' default max_connections of 100. +def pg_workers(count: int, *extra: str) -> list[subprocess.Popen[bytes]]: + # 8 pooled connections per worker (+1 for LISTEN) keeps 8 workers well under + # Postgres' default max_connections of 100. env = {**os.environ, "PGQUEUE_DSN": DSN, "BENCH_POOL_SIZE": "8"} - return [ - subprocess.Popen( - [ - sys.executable, - "-m", - "pgqueue", - "--log-level", - "WARNING", - "worker", - "bench.tasks:app", - "--queues", - "bench", - "--concurrency", - str(concurrency), - "--no-periodic", - *extra, - ], - cwd=ROOT, - env=env, - ) - for _ in range(count) - ] + cmd = [sys.executable, "-m", "pgqueue", "--log-level", "WARNING", "worker", "bench.tasks:app"] + cmd += ["--queues", "bench", "--concurrency", str(SLOTS), "--no-periodic", *extra] + return [subprocess.Popen(cmd, cwd=ROOT, env=env) for _ in range(count)] -def stop_workers(procs: list[subprocess.Popen[bytes]]) -> None: +def stop(procs: list[subprocess.Popen[bytes]]) -> None: for proc in procs: proc.terminate() for proc in procs: - proc.wait(timeout=60) + try: + proc.wait(timeout=90) + except subprocess.TimeoutExpired: + proc.kill() + proc.wait() -async def wait_registered(count: int) -> None: +async def pg_wait_registered(count: int) -> None: + deadline = time.monotonic() + 60 while await app.pool.fetchval("SELECT count(*) FROM pgqueue_workers") < count: + if time.monotonic() > deadline: + raise TimeoutError("workers did not register") await asyncio.sleep(0.05) -async def remaining() -> int: - value: int = await app.pool.fetchval( - "SELECT count(*) FROM pgqueue_jobs WHERE status <> 'succeeded'" - ) - return value +async def pg_wait_done(total: int, timeout: float) -> None: + deadline = time.monotonic() + timeout + while await app.pool.fetchval("SELECT count(*) FROM pgqueue_jobs WHERE status <> 'succeeded'"): + if time.monotonic() > deadline: + raise TimeoutError("backlog did not drain") + await asyncio.sleep(0.1) -async def bench_drain(n: int, workers: int, concurrency: int) -> Result: - """Pre-load n jobs, then measure how fast ``workers`` processes drain them.""" - await reset() +async def pg_preload(n: int, size: int) -> None: for first in range(0, n, 5000): - await noop.enqueue_many([Noop(n=i) for i in range(first, min(n, first + 5000))]) - procs = start_workers(workers, concurrency) + batch = [Email(**make_payload(i, size)) for i in range(first, min(n, first + 5000))] + await noop.enqueue_many(batch) + + +async def pg_enqueue(n: int, size: int, clients: int, bulk: int) -> Record: + await pg_reset() + payloads = [Email(**make_payload(i, size)) for i in range(n)] + lat: list[float] = [] + + async def one(offset: int) -> None: + for i in range(offset, n, clients): + t = time.perf_counter() + await noop.enqueue(payloads[i]) + lat.append((time.perf_counter() - t) * 1000) + + start = time.perf_counter() + if bulk: + for first in range(0, n, bulk): + t = time.perf_counter() + await noop.enqueue_many(payloads[first : first + bulk]) + lat.append((time.perf_counter() - t) * 1000) + else: + await asyncio.gather(*(one(c) for c in range(clients))) + seconds = time.perf_counter() - start + return {"jobs": n, "seconds": seconds, "rate": n / seconds, "latency_ms": summarize(lat)} + + +async def pg_drain(n: int, size: int, workers: int) -> Record: + await pg_reset() + await pg_preload(n, size) + procs = pg_workers(workers) try: - await wait_registered(workers) - start = time.perf_counter() - while await remaining(): - await asyncio.sleep(0.02) - seconds = time.perf_counter() - start + await pg_wait_registered(workers) + await pg_wait_done(n, timeout=600) finally: - stop_workers(procs) - return Result( - f"process backlog: {workers} worker process(es) x {concurrency} concurrency", - n, - seconds, + stop(procs) + rows = await app.pool.fetch( + "SELECT extract(epoch FROM finished_at) AS f FROM pgqueue_jobs WHERE status = 'succeeded'" ) + return {"jobs": n, "rate": steady_rate([float(r["f"]) for r in rows])} -async def bench_latency(rate: int, seconds: int, *, listen: bool) -> Result: - """Steady trickle of jobs into idle workers: how long from enqueue to start?""" - await reset() - extra = () if listen else ("--no-listen",) - procs = start_workers(2, 16, "--poll-interval", "1.0", *extra) +async def pg_pickup(rate: int, seconds: int, size: int, workers: int) -> Record: + await pg_reset() + procs = pg_workers(workers) total = rate * seconds + payloads = [Email(**make_payload(i, size)) for i in range(total)] try: - await wait_registered(2) + await pg_wait_registered(workers) await asyncio.sleep(1) interval = 1 / rate start = time.perf_counter() - for i in range(total): - await noop.enqueue(Noop(n=i)) + for i, payload in enumerate(payloads): + await noop.enqueue(payload) delay = start + (i + 1) * interval - time.perf_counter() if delay > 0: await asyncio.sleep(delay) - while await remaining(): - await asyncio.sleep(0.05) + await pg_wait_done(total, timeout=120) finally: - stop_workers(procs) + stop(procs) rows = await app.pool.fetch( - "SELECT extract(epoch FROM started_at - created_at) * 1000 AS pickup_ms FROM pgqueue_jobs" + "SELECT extract(epoch FROM started_at - created_at) * 1000 AS ms FROM pgqueue_jobs" ) - pickup = sorted(float(r["pickup_ms"]) for r in rows) - q = statistics.quantiles(pickup, n=100) - detail = f"pickup latency p50 {q[49]:.1f} ms, p95 {q[94]:.1f} ms, p99 {q[98]:.1f} ms" - mode = "LISTEN/NOTIFY + 1 s polling" if listen else "polling only (1 s interval)" - return Result(f"latency at {rate} jobs/s, {mode}", total, seconds, detail) + return {"jobs": total, "latency_ms": summarize([float(r["ms"]) for r in rows])} + + +# -- Celery side ---------------------------------------------------------------------------------- + + +class CeleryBench: + def __init__(self) -> None: + import redis + from celery import Celery + + from bench import celery_tasks + + self.redis = redis.Redis.from_url(REDIS_URL) + self.tasks = celery_tasks + self.app: Celery = celery_tasks.app + self.noop = celery_tasks.noop + self.local = threading.local() + + def reset(self) -> None: + self.redis.flushall() + + def workers(self, count: int) -> list[subprocess.Popen[bytes]]: + env = {**os.environ, "BENCH_REDIS_URL": REDIS_URL} + procs = [ + subprocess.Popen( + [ + *[sys.executable, "-m", "celery", "-A", "bench.celery_tasks", "worker"], + *["-P", "threads", "-c", str(SLOTS), "-Q", "bench", "-l", "WARNING"], + *["--without-gossip", "--without-mingle", "-n", f"w{i}@%h"], + ], + cwd=ROOT, + env=env, + stdout=subprocess.DEVNULL, # the start-up banner + ) + for i in range(count) + ] + deadline = time.monotonic() + 60 + while len(self.app.control.ping(timeout=1.0)) < count: + if time.monotonic() > deadline: + stop(procs) + raise TimeoutError("celery workers did not start") + return procs + + def send(self, i: int, size: int) -> None: + payload = make_payload(i, size, sent_at=time.time()) + self.noop.apply_async(args=[payload], queue="bench") + + def wait_done(self, total: int, timeout: float) -> None: + deadline = time.monotonic() + timeout + while self.redis.llen("bench:finished") < total: + if time.monotonic() > deadline: + raise TimeoutError("celery backlog did not drain") + time.sleep(0.1) + + def enqueue(self, n: int, size: int, clients: int) -> Record: + self.reset() + lat: list[float] = [] + + def timed(i: int) -> None: + t = time.perf_counter() + self.send(i, size) + lat.append((time.perf_counter() - t) * 1000) + start = time.perf_counter() + if clients == 1: + for i in range(n): + timed(i) + else: + with ThreadPoolExecutor(clients) as pool: + list(pool.map(timed, range(n))) + seconds = time.perf_counter() - start + return {"jobs": n, "seconds": seconds, "rate": n / seconds, "latency_ms": summarize(lat)} -# -- report --------------------------------------------------------------------------------------- + def drain(self, n: int, size: int, workers: int) -> Record: + self.reset() + with ThreadPoolExecutor(8) as pool: + list(pool.map(lambda i: self.send(i, size), range(n))) + procs = self.workers(workers) + try: + self.wait_done(n, timeout=600) + finally: + stop(procs) + finished = [float(v) for v in self.redis.lrange("bench:finished", 0, -1)] + return {"jobs": n, "rate": steady_rate(finished)} + + def pickup(self, rate: int, seconds: int, size: int, workers: int) -> Record: + self.reset() + procs = self.workers(workers) + total = rate * seconds + try: + time.sleep(1) + interval = 1 / rate + start = time.perf_counter() + for i in range(total): + self.send(i, size) + delay = start + (i + 1) * interval - time.perf_counter() + if delay > 0: + time.sleep(delay) + self.wait_done(total, timeout=120) + finally: + stop(procs) + ms = [float(v) * 1000 for v in self.redis.lrange("bench:pickup_s", 0, -1)] + return {"jobs": total, "latency_ms": summarize(ms)} -def hardware() -> dict[str, str]: - def run(cmd: str) -> str: - try: - # Force English output: lscpu and free are localised. - env = {**os.environ, "LC_ALL": "C"} - return subprocess.run( - cmd, shell=True, capture_output=True, text=True, check=False, env=env - ).stdout.strip() - except OSError: # pragma: no cover - return "unknown" - - cpu = run("lscpu | sed -n 's/^Model name:[ \\t]*//p'") or platform.processor() - return { - "cpu": f"{cpu} ({os.cpu_count()} logical CPUs)", - "memory": run("free -h | awk '/^Mem/ {print $2}'") + " RAM", - "disk": "NVMe SSD; Postgres 17 in Docker (default config) on the same machine", +# -- orchestration -------------------------------------------------------------------------------- + + +def sh(cmd: str) -> str: + env = {**os.environ, "LC_ALL": "C"} + out = subprocess.run(cmd, shell=True, capture_output=True, text=True, check=False, env=env) + return out.stdout.strip() + + +async def environment(celery: bool) -> dict[str, str]: + pg = await asyncpg.connect(DSN) + try: + version = await pg.fetchval("SHOW server_version") + settings = { + k: await pg.fetchval(f"SHOW {k}") + for k in ("shared_buffers", "synchronous_commit", "fsync", "max_connections") + } + finally: + await pg.close() + info = { + "cpu": sh("lscpu | sed -n 's/^Model name:[ \\t]*//p'") or platform.processor(), + "logical_cpus": str(os.cpu_count()), + "memory": sh("free -h | awk '/^Mem/ {print $2}'"), "os": f"{platform.system()} {platform.release()}", "python": platform.python_version(), - "postgres": run("docker exec pgqueue-postgres-1 postgres --version 2>/dev/null"), - "load_before": run("cat /proc/loadavg | cut -d' ' -f1-3"), + "postgres": f"{version} {settings}", + "load_average_at_start": sh("cut -d' ' -f1-3 /proc/loadavg"), } + if celery: + import celery as celery_pkg + import redis + r = redis.Redis.from_url(REDIS_URL) + info["broker"] = ( + f"{r.info('server').get('server_name', 'redis')} {r.info('server')['redis_version']}" + ) + info["celery"] = f"celery {celery_pkg.__version__}, redis-py {redis.__version__}" + info["persistence"] = ( + str(r.config_get("save")) + " appendonly=" + str(r.config_get("appendonly")) + ) + return info + + +async def main(args: argparse.Namespace) -> None: + quick = args.quick + runs = 1 if quick else args.runs + jobs = 2_000 if quick else args.jobs + rate = args.rate + secs = 3 if quick else args.seconds + sweep = [int(w) for w in args.workers.split(",")] + size = SIZES[args.payload] + celery = None if args.skip_celery else CeleryBench() -async def main(quick: bool) -> None: await prepare_database() await app.connect() - hw = hardware() - scale = 0.2 if quick else 1.0 - results: list[Result] = [] + env = await environment(celery is not None) + records: list[Record] = [] + + def add(system: str, scenario: str, run: int, result: Record, **kw: Any) -> None: + rec = { + "system": system, + "scenario": scenario, + "run": run, + **kw, + **result, + "load_1min": sh("cut -d' ' -f1 /proc/loadavg"), + } + records.append(rec) + short = {k: (round(v, 1) if isinstance(v, float) else v) for k, v in result.items()} + log(f"{system:8} {scenario:8} run {run} {kw} {short}") + try: - results.append(await bench_enqueue_sequential(int(5_000 * scale))) - results.append(await bench_enqueue_concurrent(int(20_000 * scale), clients=10)) - results.append(await bench_enqueue_bulk(int(100_000 * scale), batch=1000)) - for workers, concurrency in [(1, 32), (2, 32), (4, 32), (8, 32)]: - results.append(await bench_drain(int(50_000 * scale), workers, concurrency)) - results.append(await bench_latency(100, 10 if not quick else 3, listen=True)) - results.append(await bench_latency(100, 10 if not quick else 3, listen=False)) + for run in range(1, runs + 1): + load = sh("cut -d' ' -f1 /proc/loadavg") + log(f"run {run}/{runs} (load average 1 min: {load})") + for name, clients, bulk in [("single", 1, 0), ("ten", 10, 0), ("bulk", 1, 1000)]: + n = jobs // 4 if name != "bulk" else jobs * 2 + kw = {"payload": size, "variant": name} + add("pgqueue", "enqueue", run, await pg_enqueue(n, size, clients, bulk), **kw) + if celery and name != "bulk": + add("celery", "enqueue", run, celery.enqueue(n, size, clients), **kw) + for w in sweep: + add("pgqueue", "drain", run, await pg_drain(jobs, size, w), workers=w, payload=size) + if celery: + add( + "celery", "drain", run, celery.drain(jobs, size, w), workers=w, payload=size + ) + for w in sweep: + kw = {"workers": w, "payload": size, "arrival_rate": rate} + add("pgqueue", "pickup", run, await pg_pickup(rate, secs, size, w), **kw) + if celery: + add("celery", "pickup", run, celery.pickup(rate, secs, size, w), **kw) + for label, psize in SIZES.items(): + n = jobs // 2 + kw = {"workers": 4, "payload": psize, "label": label} + add("pgqueue", "drain-size", run, await pg_drain(n, psize, 4), **kw) + if celery: + add("celery", "drain-size", run, celery.drain(n, psize, 4), **kw) finally: + env["load_average_at_end"] = sh("cut -d' ' -f1-3 /proc/loadavg") await app.close() - - print("\n## Environment\n") - for key, value in hw.items(): - print(f"- **{key}**: {value}") - print("\n## Results\n") - print("| Scenario | Jobs | Seconds | Jobs/s | Notes |") - print("|---|---:|---:|---:|---|") - for r in results: - print(f"| {r.scenario} | {r.jobs:,} | {r.seconds:.2f} | {r.rate:,.0f} | {r.detail} |") - out = ROOT / "bench" / "results.json" - out.write_text( - json.dumps( - {"environment": hw, "results": [asdict(r) | {"rate": r.rate} for r in results]}, - indent=2, + out = Path(args.out) + out.write_text( + json.dumps( + { + "environment": env, + "parameters": { + "runs": runs, + "jobs": jobs, + "slots_per_worker": SLOTS, + "arrival_rate": rate, + "pickup_seconds": secs, + "workers": sweep, + "payload_bytes": size, + "quick": quick, + }, + "records": records, + }, + indent=1, + ) ) - ) - print(f"\nRaw numbers written to {out.relative_to(ROOT)}") + log(f"raw numbers written to {out}") if __name__ == "__main__": - parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument("--quick", action="store_true") - asyncio.run(main(parser.parse_args().quick)) + parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) + parser.add_argument("--quick", action="store_true", help="tiny smoke run") + parser.add_argument("--runs", type=int, default=5) + parser.add_argument("--jobs", type=int, default=20_000, help="backlog size for drain runs") + parser.add_argument("--rate", type=int, default=200, help="arrivals/s for the pickup runs") + parser.add_argument("--seconds", type=int, default=15, help="length of a pickup run") + parser.add_argument("--workers", default="1,2,4,8") + parser.add_argument("--payload", choices=list(SIZES), default="medium") + parser.add_argument("--skip-celery", action="store_true") + parser.add_argument("--out", default=str(ROOT / "bench" / "results.json")) + asyncio.run(main(parser.parse_args())) diff --git a/bench/celery_tasks.py b/bench/celery_tasks.py new file mode 100644 index 0000000..f2e075a --- /dev/null +++ b/bench/celery_tasks.py @@ -0,0 +1,38 @@ +"""Celery + Redis counterpart of bench.tasks, for the comparison runs. + +The task does the minimum a benchmark needs: it records when it started and finished in +Redis with one pipelined round trip. pgqueue gets the same information from its own job +rows at no extra cost, so Celery is charged one extra Redis call per task here; the +write-up says so. +""" + +from __future__ import annotations + +import os +import time +from typing import Any + +import redis +from celery import Celery + +BROKER = os.environ.get("BENCH_REDIS_URL", "redis://localhost:56379/0") + +app = Celery("bench", broker=BROKER) +app.conf.update( + task_ignore_result=True, + task_default_queue="bench", + broker_connection_retry_on_startup=True, +) + +_redis = redis.Redis.from_url(BROKER) + + +def _noop(payload: dict[str, Any]) -> None: + started = time.time() + pipe = _redis.pipeline(transaction=False) + pipe.rpush("bench:pickup_s", started - payload["sent_at"]) + pipe.rpush("bench:finished", time.time()) + pipe.execute() + + +noop = app.task(name="bench.noop")(_noop) diff --git a/bench/chart.py b/bench/chart.py new file mode 100644 index 0000000..bd39077 --- /dev/null +++ b/bench/chart.py @@ -0,0 +1,249 @@ +"""Draw docs/benchmarks-*.svg from bench/results.json using only the standard library. + + make bench-chart + +Points are medians over the repeated runs; the whiskers span the minimum to the maximum +run, so the spread is visible in the picture and not only in the tables. +""" + +from __future__ import annotations + +import json +import statistics +from dataclasses import dataclass, field +from html import escape +from pathlib import Path +from typing import Any + +ROOT = Path(__file__).resolve().parent.parent +RESULTS = ROOT / "bench" / "results.json" +DOCS = ROOT / "docs" + +# Okabe-Ito colours (colour-blind safe); every series also has its own marker and dash. +BLUE, ORANGE, GREEN, PURPLE, GREY, RED = ( + "#0072b2", + "#d55e00", + "#009e73", + "#8e5a9b", + "#6b7280", + "#b3261e", +) +INK, MUTED, GRID, BG = "#1f2328", "#59636e", "#d8dee4", "#ffffff" +FONT = "system-ui, -apple-system, 'Segoe UI', Roboto, sans-serif" +MARKERS = ("circle", "square", "diamond", "triangle", "cross") + + +@dataclass +class Series: + label: str + color: str + marker: str + dash: str = "" + points: list[tuple[float, float, float, float]] = field(default_factory=list) # x, med, lo, hi + + +def _marker(kind: str, x: float, y: float, color: str) -> str: + s = 4.5 + common = f'fill="{color}" stroke="{BG}" stroke-width="1.5"' + if kind == "square": + return f'' + if kind == "diamond": + pts = f"{x},{y - s - 1} {x + s + 1},{y} {x},{y + s + 1} {x - s - 1},{y}" + return f'' + if kind == "triangle": + pts = f"{x},{y - s - 1} {x + s + 1},{y + s} {x - s - 1},{y + s}" + return f'' + if kind == "cross": + return ( + f'' + ) + return f'' + + +def _nice_max(value: float) -> float: + step = 10 ** (len(str(int(value))) - 1) + for mult in (1, 2, 2.5, 5, 10): + if value <= mult * step: + return float(mult * step) + return float(10 * step) + + +def line_chart( # noqa: PLR0917 + title: str, + subtitle: str, + xlabels: list[str], + series: list[Series], + ylabel: str, + note: str, + *, + fmt: str = "{:,.0f}", +) -> str: + w, h = 860, 480 + left, right, top, bottom = 78, 190, 78, 70 + pw, ph = w - left - right, h - top - bottom + ymax = _nice_max(max(p[3] for s in series for p in s.points) * 1.05) + n = len(xlabels) + + def px(i: float) -> float: + return left + pw * (i + 0.5) / n + + def py(v: float) -> float: + return top + ph * (1 - v / ymax) + + out = [ + f'', + f'{escape(title)}{escape(subtitle)}', + f'', + f'' + f"{escape(title)}", + f'{escape(subtitle)}', + ] + for k in range(5): + v = ymax * k / 4 + y = py(v) + out.append( + f'' + ) + out.append( + f'{fmt.format(v)}' + ) + out.append( + f'{escape(ylabel)}' + ) + for i, lab in enumerate(xlabels): + out.append( + f'{escape(lab)}' + ) + out.append( + f'worker processes' + ) + for s in series: + pts = [(px(x), py(m), py(lo), py(hi)) for x, m, lo, hi in s.points] + dash = f' stroke-dasharray="{s.dash}"' if s.dash else "" + if len(pts) > 1: + path = " ".join(f"{x:.1f},{y:.1f}" for x, y, _, _ in pts) + out.append( + f'' + ) + for x, y, ylo, yhi in pts: + if abs(ylo - yhi) > 1: + out.append( + f'' + ) + out.append(_marker(s.marker, x, y, s.color)) + ly = top + 4 + for s in series: + lx = left + pw + 24 + dash = f' stroke-dasharray="{s.dash}"' if s.dash else "" + out.append( + f'' + ) + out.append(_marker(s.marker, lx + 11, ly, s.color)) + out.append( + f'{escape(s.label)}' + ) + ly += 22 + out.append(f'{escape(note)}') + out.append("") + return "\n".join(out) + "\n" + + +def aggregate( + records: list[dict[str, Any]], system: str, scenario: str, key: str, workers: list[int] +) -> list[tuple[float, float, float, float]]: + """(x index, median, min, max) of ``key`` over runs, one point per worker count.""" + out = [] + for i, w in enumerate(workers): + vals = [ + _get(r, key) + for r in records + if r["system"] == system + and r["scenario"] == scenario + and r.get("workers") == w + and "label" not in r + ] + if vals: + out.append((float(i), statistics.median(vals), min(vals), max(vals))) + return out + + +def _get(rec: dict[str, Any], key: str) -> float: + value: Any = rec + for part in key.split("."): + value = value[part] + return float(value) + + +def build(results: dict[str, Any]) -> dict[str, str]: + params, env, records = results["parameters"], results["environment"], results["records"] + workers = list(params["workers"]) + labels = [str(w) for w in workers] + runs = params["runs"] + kib = params["payload_bytes"] / 1024 + note = ( + f"Median of {runs} run{'s' if runs != 1 else ''}, whiskers = min to max. " + f"{env['cpu'].split(' (')[0]}; PostgreSQL, Valkey, workers and load generator share it." + ) + systems = [ + ("pgqueue", "pgqueue (PostgreSQL)", BLUE, "circle", ""), + ("celery", "Celery (Valkey)", ORANGE, "triangle", "6 4"), + ] + present = {r["system"] for r in records} + drain = [ + Series(label, color, marker, dash, aggregate(records, s, "drain", "rate", workers)) + for s, label, color, marker, dash in systems + if s in present + ] + charts = { + "benchmarks-throughput.svg": line_chart( + "Backlog throughput by worker processes", + f"Jobs per second draining a pre-loaded backlog, no-op handler, {kib:g} KiB payload, " + f"{params['slots_per_worker']} slots per worker", + labels, + drain, + "jobs per second", + note, + ) + } + lat: list[Series] = [] + for s, label, color, marker, _dash in systems: + if s not in present: + continue + short = label.split(" (")[0] + for q, qcolor, qmarker in ( + ("p50", color, marker), + ("p99", GREY if s == "pgqueue" else RED, "square"), + ): + pts = aggregate(records, s, "pickup", f"latency_ms.{q}", workers) + lat.append(Series(f"{short} {q}", qcolor, qmarker, "" if q == "p50" else "2 3", pts)) + charts["benchmarks-latency.svg"] = line_chart( + "Pickup latency by worker processes", + f"Enqueue to handler start at a steady {params['arrival_rate']} jobs/s, " + f"{kib:g} KiB payload (lower is better)", + labels, + lat, + "milliseconds", + note, + fmt="{:,.1f}", + ) + return charts + + +def main() -> None: + results = json.loads(RESULTS.read_text()) + for name, svg in build(results).items(): + (DOCS / name).write_text(svg) + print(f"wrote docs/{name}") + + +if __name__ == "__main__": + main() diff --git a/bench/payload.py b/bench/payload.py new file mode 100644 index 0000000..1a8779f --- /dev/null +++ b/bench/payload.py @@ -0,0 +1,77 @@ +"""Deterministic job payloads of a chosen serialized size. + +The shape imitates a typical background job (a transactional e-mail or a webhook +delivery): a few identifiers, short strings and a free-text body. The body is made of +words drawn from a fixed vocabulary with a fixed seed, so every run, and both queue +systems, see the same bytes. Real text compresses somewhat; so does this. +""" + +from __future__ import annotations + +import json +import random +from typing import Any + +SIZES: dict[str, int] = {"small": 256, "medium": 2048, "large": 16384} + +_VOCAB = [ + "order", + "invoice", + "customer", + "shipment", + "warehouse", + "confirmation", + "payment", + "account", + "reminder", + "statement", + "subscription", + "renewal", + "address", + "delivery", + "tracking", + "refund", + "receipt", + "update", + "notification", + "profile", + "password", + "session", + "device", + "invoice", + "quarterly", + "summary", + "report", + "attachment", + "support", + "ticket", + "priority", + "escalated", + "resolved", + "pending", + "review", + "approval", +] + + +def make_payload(i: int, size: int, *, sent_at: float | None = None) -> dict[str, Any]: + """A JSON-serializable payload that is about ``size`` bytes once serialized.""" + rng = random.Random(i) + doc: dict[str, Any] = { + "n": i, + "kind": "email.transactional", + "recipient": f"user{i % 10_000}@example.com", + "subject": f"Update {i} for your account", + "body": "", + } + if sent_at is not None: + doc["sent_at"] = sent_at + overhead = len(json.dumps(doc)) + words: list[str] = [] + length = 0 + while length < size - overhead: + word = rng.choice(_VOCAB) + words.append(word) + length += len(word) + 1 + doc["body"] = " ".join(words)[: max(0, size - overhead)] + return doc diff --git a/bench/report.py b/bench/report.py new file mode 100644 index 0000000..b13455c --- /dev/null +++ b/bench/report.py @@ -0,0 +1,97 @@ +"""Print the Markdown tables of docs/benchmarks.md from bench/results.json. + + uv run python -m bench.report + +Every cell is the median over the runs, with the minimum and maximum run in brackets. +""" + +from __future__ import annotations + +import json +import statistics +from pathlib import Path +from typing import Any + +RESULTS = Path(__file__).resolve().parent.parent / "bench" / "results.json" +NAMES = {"pgqueue": "pgqueue", "celery": "Celery + Valkey"} + + +def get(rec: dict[str, Any], key: str) -> float: + value: Any = rec + for part in key.split("."): + value = value[part] + return float(value) + + +def cell(recs: list[dict[str, Any]], key: str, fmt: str) -> str: + if not recs: + return "n/a" + vals = [get(r, key) for r in recs] + med, lo, hi = statistics.median(vals), min(vals), max(vals) + return f"{fmt.format(med)} ({fmt.format(lo)}-{fmt.format(hi)})" + + +def select(records: list[dict[str, Any]], **where: Any) -> list[dict[str, Any]]: + return [r for r in records if all(r.get(k) == v for k, v in where.items())] + + +def main() -> None: + data = json.loads(RESULTS.read_text()) + recs, params = data["records"], data["parameters"] + systems = [s for s in NAMES if select(recs, system=s)] + workers = params["workers"] + print(f"Runs per cell: {params['runs']}. Cells are median (min-max).\n") + + print("### Enqueue\n") + print("| Variant | System | jobs/s | p50 ms | p95 ms | p99 ms |") + print("|---|---|---:|---:|---:|---:|") + for variant, label in ( + ("single", "1 client, 1 job per call"), + ("ten", "10 concurrent clients"), + ("bulk", "`enqueue_many`, 1,000 per call (latency is per call)"), + ): + for s in systems: + r = select(recs, system=s, scenario="enqueue", variant=variant) + if r: + cols = [cell(r, "rate", "{:,.0f}")] + cols += [cell(r, f"latency_ms.{q}", "{:,.1f}") for q in ("p50", "p95", "p99")] + print(f"| {label} | {NAMES[s]} | " + " | ".join(cols) + " |") + + print("\n### Backlog throughput (jobs/s)\n") + print("| Worker processes | " + " | ".join(NAMES[s] for s in systems) + " |") + print("|---:|" + "---:|" * len(systems)) + for w in workers: + row = [ + cell(select(recs, system=s, scenario="drain", workers=w), "rate", "{:,.0f}") + for s in systems + ] + print(f"| {w} | " + " | ".join(row) + " |") + + print(f"\n### Pickup latency at {params['arrival_rate']} jobs/s (ms)\n") + print("| Worker processes | System | p50 | p95 | p99 | max |") + print("|---:|---|---:|---:|---:|---:|") + for w in workers: + for s in systems: + r = select(recs, system=s, scenario="pickup", workers=w) + cols = [cell(r, f"latency_ms.{q}", "{:,.1f}") for q in ("p50", "p95", "p99", "max")] + print(f"| {w} | {NAMES[s]} | " + " | ".join(cols) + " |") + + print("\n### Payload size, 4 workers, backlog throughput (jobs/s)\n") + print("| Payload | " + " | ".join(NAMES[s] for s in systems) + " |") + print("|---|" + "---:|" * len(systems)) + for label, size in (("small", 256), ("medium", 2048), ("large", 16384)): + row = [ + cell(select(recs, system=s, scenario="drain-size", label=label), "rate", "{:,.0f}") + for s in systems + ] + print(f"| {size:,} B | " + " | ".join(row) + " |") + + loads = [float(r["load_1min"]) for r in recs] + print( + f"\nOne-minute load average sampled at each measurement: " + f"min {min(loads):.1f}, median {statistics.median(loads):.1f}, max {max(loads):.1f}." + ) + + +if __name__ == "__main__": + main() diff --git a/bench/results.json b/bench/results.json index bb45b19..9c49fbb 100644 --- a/bench/results.json +++ b/bench/results.json @@ -1,77 +1,1892 @@ { - "environment": { - "cpu": "13th Gen Intel(R) Core(TM) i5-13500H (16 logical CPUs)", - "memory": "14Gi RAM", - "disk": "NVMe SSD; Postgres 17 in Docker (default config) on the same machine", - "os": "Linux 7.0.0-38-generic", - "python": "3.12.13", - "postgres": "postgres (PostgreSQL) 17.11", - "load_before": "5.89 4.33 4.17" - }, - "results": [ - { - "scenario": "enqueue, one at a time (1 client)", - "jobs": 5000, - "seconds": 10.196898877999047, - "detail": "", - "rate": 490.34515883922916 - }, - { - "scenario": "enqueue, one at a time (10 concurrent clients)", - "jobs": 20000, - "seconds": 13.912013039000158, - "detail": "", - "rate": 1437.606473192134 - }, - { - "scenario": "enqueue_many, batches of 1000", - "jobs": 100000, - "seconds": 2.469376367997029, - "detail": "", - "rate": 40496.05450833419 - }, - { - "scenario": "process backlog: 1 worker process(es) x 32 concurrency", - "jobs": 50000, - "seconds": 34.82572885800255, - "detail": "", - "rate": 1435.7201310522057 - }, - { - "scenario": "process backlog: 2 worker process(es) x 32 concurrency", - "jobs": 50000, - "seconds": 19.02412557900243, - "detail": "", - "rate": 2628.2416919696266 - }, - { - "scenario": "process backlog: 4 worker process(es) x 32 concurrency", - "jobs": 50000, - "seconds": 11.947290708001674, - "detail": "", - "rate": 4185.049248572532 - }, - { - "scenario": "process backlog: 8 worker process(es) x 32 concurrency", - "jobs": 50000, - "seconds": 7.547685925001133, - "detail": "", - "rate": 6624.546979939748 - }, - { - "scenario": "latency at 100 jobs/s, LISTEN/NOTIFY + 1 s polling", - "jobs": 1000, - "seconds": 10, - "detail": "pickup latency p50 4.1 ms, p95 6.7 ms, p99 12.5 ms", - "rate": 100.0 - }, - { - "scenario": "latency at 100 jobs/s, polling only (1 s interval)", - "jobs": 1000, - "seconds": 10, - "detail": "pickup latency p50 482.8 ms, p95 931.8 ms, p99 991.0 ms", - "rate": 100.0 - } + "environment": { + "cpu": "13th Gen Intel(R) Core(TM) i5-13500H", + "logical_cpus": "16", + "memory": "14Gi", + "os": "Linux 7.0.0-38-generic", + "python": "3.12.13", + "postgres": "17.11 {'shared_buffers': '128MB', 'synchronous_commit': 'on', 'fsync': 'on', 'max_connections': '100'}", + "load_average_at_start": "38.62 28.42 17.01", + "broker": "valkey 7.2.4", + "celery": "celery 5.6.3, redis-py 8.1.0", + "persistence": "{'save': ''} appendonly={'appendonly': 'no'}", + "load_average_at_end": "4.66 4.80 7.46" + }, + "parameters": { + "runs": 5, + "jobs": 20000, + "slots_per_worker": 32, + "arrival_rate": 200, + "pickup_seconds": 15, + "workers": [ + 1, + 2, + 4, + 8 ], - "date": "2026-10-10" + "payload_bytes": 2048, + "quick": false + }, + "records": [ + { + "system": "pgqueue", + "scenario": "enqueue", + "run": 1, + "payload": 2048, + "variant": "single", + "jobs": 5000, + "seconds": 45.08915296100167, + "rate": 110.89141559888206, + "latency_ms": { + "p50": 5.584774498856859, + "p95": 19.54631574844825, + "p99": 66.51129696903809, + "max": 427.9835440029274, + "n": 5000.0 + }, + "load_1min": "37.73" + }, + { + "system": "celery", + "scenario": "enqueue", + "run": 1, + "payload": 2048, + "variant": "single", + "jobs": 5000, + "seconds": 8.862754460998985, + "rate": 564.1586960355002, + "latency_ms": { + "p50": 1.6649319986754563, + "p95": 2.4693350449524587, + "p99": 4.84756686273613, + "max": 315.6045110008563, + "n": 5000.0 + }, + "load_1min": "36.85" + }, + { + "system": "pgqueue", + "scenario": "enqueue", + "run": 1, + "payload": 2048, + "variant": "ten", + "jobs": 5000, + "seconds": 28.581017406002502, + "rate": 174.9412880924916, + "latency_ms": { + "p50": 51.301703500939766, + "p95": 100.73975665363832, + "p99": 153.63346406302298, + "max": 574.3456950003747, + "n": 5000.0 + }, + "load_1min": "32.20" + }, + { + "system": "celery", + "scenario": "enqueue", + "run": 1, + "payload": 2048, + "variant": "ten", + "jobs": 5000, + "seconds": 10.650425011001062, + "rate": 469.4648330780592, + "latency_ms": { + "p50": 18.792613998812158, + "p95": 37.44688920414774, + "p99": 61.80480238464952, + "max": 696.4193099993281, + "n": 5000.0 + }, + "load_1min": "28.11" + }, + { + "system": "pgqueue", + "scenario": "enqueue", + "run": 1, + "payload": 2048, + "variant": "bulk", + "jobs": 40000, + "seconds": 7.72024922700075, + "rate": 5181.179884724998, + "latency_ms": { + "p50": 190.38571499913814, + "p95": 229.7530514515529, + "p99": 245.02151536231395, + "max": 251.8819960023393, + "n": 40.0 + }, + "load_1min": "25.36" + }, + { + "system": "pgqueue", + "scenario": "drain", + "run": 1, + "workers": 1, + "payload": 2048, + "jobs": 20000, + "rate": 825.2743848281829, + "load_1min": "22.02" + }, + { + "system": "celery", + "scenario": "drain", + "run": 1, + "workers": 1, + "payload": 2048, + "jobs": 20000, + "rate": 307.091147584142, + "load_1min": "20.19" + }, + { + "system": "pgqueue", + "scenario": "drain", + "run": 1, + "workers": 2, + "payload": 2048, + "jobs": 20000, + "rate": 1426.7025988550924, + "load_1min": "19.97" + }, + { + "system": "celery", + "scenario": "drain", + "run": 1, + "workers": 2, + "payload": 2048, + "jobs": 20000, + "rate": 573.8004467968121, + "load_1min": "17.39" + }, + { + "system": "pgqueue", + "scenario": "drain", + "run": 1, + "workers": 4, + "payload": 2048, + "jobs": 20000, + "rate": 2469.361782380255, + "load_1min": "14.28" + }, + { + "system": "celery", + "scenario": "drain", + "run": 1, + "workers": 4, + "payload": 2048, + "jobs": 20000, + "rate": 990.1651978208564, + "load_1min": "15.43" + }, + { + "system": "pgqueue", + "scenario": "drain", + "run": 1, + "workers": 8, + "payload": 2048, + "jobs": 20000, + "rate": 2588.8792343892146, + "load_1min": "19.10" + }, + { + "system": "celery", + "scenario": "drain", + "run": 1, + "workers": 8, + "payload": 2048, + "jobs": 20000, + "rate": 1858.7040055432692, + "load_1min": "16.65" + }, + { + "system": "pgqueue", + "scenario": "pickup", + "run": 1, + "workers": 1, + "payload": 2048, + "arrival_rate": 200, + "jobs": 3000, + "latency_ms": { + "p50": 7.343, + "p95": 17.358800000000002, + "p99": 56.35592, + "max": 145.712, + "n": 3000.0 + }, + "load_1min": "14.11" + }, + { + "system": "celery", + "scenario": "pickup", + "run": 1, + "workers": 1, + "payload": 2048, + "arrival_rate": 200, + "jobs": 3000, + "latency_ms": { + "p50": 5.315542221069336, + "p95": 53.596627712249756, + "p99": 87.71420240402222, + "max": 137.86053657531738, + "n": 3000.0 + }, + "load_1min": "13.86" + }, + { + "system": "pgqueue", + "scenario": "pickup", + "run": 1, + "workers": 2, + "payload": 2048, + "arrival_rate": 200, + "jobs": 3000, + "latency_ms": { + "p50": 5.207000000000001, + "p95": 14.30675, + "p99": 52.77924, + "max": 262.881, + "n": 3000.0 + }, + "load_1min": "15.25" + }, + { + "system": "celery", + "scenario": "pickup", + "run": 1, + "workers": 2, + "payload": 2048, + "arrival_rate": 200, + "jobs": 3000, + "latency_ms": { + "p50": 4.410028457641602, + "p95": 8.52506160736084, + "p99": 10.952956676483154, + "max": 20.374298095703125, + "n": 3000.0 + }, + "load_1min": "13.99" + }, + { + "system": "pgqueue", + "scenario": "pickup", + "run": 1, + "workers": 4, + "payload": 2048, + "arrival_rate": 200, + "jobs": 3000, + "latency_ms": { + "p50": 4.472, + "p95": 8.7764, + "p99": 15.189280000000002, + "max": 49.061, + "n": 3000.0 + }, + "load_1min": "18.60" + }, + { + "system": "celery", + "scenario": "pickup", + "run": 1, + "workers": 4, + "payload": 2048, + "arrival_rate": 200, + "jobs": 3000, + "latency_ms": { + "p50": 4.246354103088379, + "p95": 8.142495155334473, + "p99": 10.999796390533447, + "max": 17.753124237060547, + "n": 3000.0 + }, + "load_1min": "17.36" + }, + { + "system": "pgqueue", + "scenario": "pickup", + "run": 1, + "workers": 8, + "payload": 2048, + "arrival_rate": 200, + "jobs": 3000, + "latency_ms": { + "p50": 4.4544999999999995, + "p95": 7.4091499999999995, + "p99": 10.244090000000002, + "max": 28.264, + "n": 3000.0 + }, + "load_1min": "23.00" + }, + { + "system": "celery", + "scenario": "pickup", + "run": 1, + "workers": 8, + "payload": 2048, + "arrival_rate": 200, + "jobs": 3000, + "latency_ms": { + "p50": 4.472017288208008, + "p95": 8.271324634552002, + "p99": 11.251399517059326, + "max": 26.91960334777832, + "n": 3000.0 + }, + "load_1min": "18.93" + }, + { + "system": "pgqueue", + "scenario": "drain-size", + "run": 1, + "workers": 4, + "payload": 256, + "label": "small", + "jobs": 10000, + "rate": 3498.9614549285675, + "load_1min": "18.85" + }, + { + "system": "celery", + "scenario": "drain-size", + "run": 1, + "workers": 4, + "payload": 256, + "label": "small", + "jobs": 10000, + "rate": 1028.8071280978897, + "load_1min": "17.19" + }, + { + "system": "pgqueue", + "scenario": "drain-size", + "run": 1, + "workers": 4, + "payload": 2048, + "label": "medium", + "jobs": 10000, + "rate": 3643.3321141515503, + "load_1min": "18.53" + }, + { + "system": "celery", + "scenario": "drain-size", + "run": 1, + "workers": 4, + "payload": 2048, + "label": "medium", + "jobs": 10000, + "rate": 1035.9953899817565, + "load_1min": "17.69" + }, + { + "system": "pgqueue", + "scenario": "drain-size", + "run": 1, + "workers": 4, + "payload": 16384, + "label": "large", + "jobs": 10000, + "rate": 3132.2093842973727, + "load_1min": "15.35" + }, + { + "system": "celery", + "scenario": "drain-size", + "run": 1, + "workers": 4, + "payload": 16384, + "label": "large", + "jobs": 10000, + "rate": 984.2144138493258, + "load_1min": "19.98" + }, + { + "system": "pgqueue", + "scenario": "enqueue", + "run": 2, + "payload": 2048, + "variant": "single", + "jobs": 5000, + "seconds": 24.20208390300104, + "rate": 206.5937801075057, + "latency_ms": { + "p50": 4.386684999190038, + "p95": 7.335558452177793, + "p99": 9.644859913314576, + "max": 24.75346599385375, + "n": 5000.0 + }, + "load_1min": "17.52" + }, + { + "system": "celery", + "scenario": "enqueue", + "run": 2, + "payload": 2048, + "variant": "single", + "jobs": 5000, + "seconds": 14.38342478100094, + "rate": 347.62235532419913, + "latency_ms": { + "p50": 2.87959200431942, + "p95": 4.34535954882449, + "p99": 5.902515582929482, + "max": 8.470221000607125, + "n": 5000.0 + }, + "load_1min": "16.85" + }, + { + "system": "pgqueue", + "scenario": "enqueue", + "run": 2, + "payload": 2048, + "variant": "ten", + "jobs": 5000, + "seconds": 13.218141149001895, + "rate": 378.2680139088658, + "latency_ms": { + "p50": 25.435657498746878, + "p95": 28.870913250284502, + "p99": 33.9327484772366, + "max": 403.73103599995375, + "n": 5000.0 + }, + "load_1min": "15.11" + }, + { + "system": "celery", + "scenario": "enqueue", + "run": 2, + "payload": 2048, + "variant": "ten", + "jobs": 5000, + "seconds": 11.309189368999796, + "rate": 442.11833729707973, + "latency_ms": { + "p50": 16.215782503422815, + "p95": 57.99709665225237, + "p99": 89.74461567711842, + "max": 765.4800780001096, + "n": 5000.0 + }, + "load_1min": "13.65" + }, + { + "system": "pgqueue", + "scenario": "enqueue", + "run": 2, + "payload": 2048, + "variant": "bulk", + "jobs": 40000, + "seconds": 6.464510794001399, + "rate": 6187.629857021374, + "latency_ms": { + "p50": 153.09613199860905, + "p95": 244.95854519955174, + "p99": 258.40662220864033, + "max": 266.74574299977394, + "n": 40.0 + }, + "load_1min": "12.02" + }, + { + "system": "pgqueue", + "scenario": "drain", + "run": 2, + "workers": 1, + "payload": 2048, + "jobs": 20000, + "rate": 725.4323605376657, + "load_1min": "11.77" + }, + { + "system": "celery", + "scenario": "drain", + "run": 2, + "workers": 1, + "payload": 2048, + "jobs": 20000, + "rate": 279.97636674101545, + "load_1min": "10.39" + }, + { + "system": "pgqueue", + "scenario": "drain", + "run": 2, + "workers": 2, + "payload": 2048, + "jobs": 20000, + "rate": 1701.982613279956, + "load_1min": "11.97" + }, + { + "system": "celery", + "scenario": "drain", + "run": 2, + "workers": 2, + "payload": 2048, + "jobs": 20000, + "rate": 721.9893430575147, + "load_1min": "7.89" + }, + { + "system": "pgqueue", + "scenario": "drain", + "run": 2, + "workers": 4, + "payload": 2048, + "jobs": 20000, + "rate": 2443.879146019566, + "load_1min": "7.27" + }, + { + "system": "celery", + "scenario": "drain", + "run": 2, + "workers": 4, + "payload": 2048, + "jobs": 20000, + "rate": 1205.9949418590977, + "load_1min": "10.56" + }, + { + "system": "pgqueue", + "scenario": "drain", + "run": 2, + "workers": 8, + "payload": 2048, + "jobs": 20000, + "rate": 3892.883482023774, + "load_1min": "13.64" + }, + { + "system": "celery", + "scenario": "drain", + "run": 2, + "workers": 8, + "payload": 2048, + "jobs": 20000, + "rate": 1683.2564164182097, + "load_1min": "20.19" + }, + { + "system": "pgqueue", + "scenario": "pickup", + "run": 2, + "workers": 1, + "payload": 2048, + "arrival_rate": 200, + "jobs": 3000, + "latency_ms": { + "p50": 6.158, + "p95": 15.108899999999998, + "p99": 152.05146, + "max": 276.808, + "n": 3000.0 + }, + "load_1min": "19.72" + }, + { + "system": "celery", + "scenario": "pickup", + "run": 2, + "workers": 1, + "payload": 2048, + "arrival_rate": 200, + "jobs": 3000, + "latency_ms": { + "p50": 3.840327262878418, + "p95": 8.82648229598999, + "p99": 11.942691802978516, + "max": 21.192312240600586, + "n": 3000.0 + }, + "load_1min": "20.29" + }, + { + "system": "pgqueue", + "scenario": "pickup", + "run": 2, + "workers": 2, + "payload": 2048, + "arrival_rate": 200, + "jobs": 3000, + "latency_ms": { + "p50": 4.343, + "p95": 8.4795, + "p99": 58.6489, + "max": 165.783, + "n": 3000.0 + }, + "load_1min": "21.15" + }, + { + "system": "celery", + "scenario": "pickup", + "run": 2, + "workers": 2, + "payload": 2048, + "arrival_rate": 200, + "jobs": 3000, + "latency_ms": { + "p50": 3.606081008911133, + "p95": 6.586277484893799, + "p99": 8.820066452026367, + "max": 15.003204345703125, + "n": 3000.0 + }, + "load_1min": "20.27" + }, + { + "system": "pgqueue", + "scenario": "pickup", + "run": 2, + "workers": 4, + "payload": 2048, + "arrival_rate": 200, + "jobs": 3000, + "latency_ms": { + "p50": 4.532, + "p95": 9.49195, + "p99": 15.620459999999998, + "max": 45.115, + "n": 3000.0 + }, + "load_1min": "20.29" + }, + { + "system": "celery", + "scenario": "pickup", + "run": 2, + "workers": 4, + "payload": 2048, + "arrival_rate": 200, + "jobs": 3000, + "latency_ms": { + "p50": 3.50797176361084, + "p95": 5.929052829742432, + "p99": 8.112523555755615, + "max": 13.669013977050781, + "n": 3000.0 + }, + "load_1min": "18.90" + }, + { + "system": "pgqueue", + "scenario": "pickup", + "run": 2, + "workers": 8, + "payload": 2048, + "arrival_rate": 200, + "jobs": 3000, + "latency_ms": { + "p50": 5.4835, + "p95": 12.36155, + "p99": 17.50569, + "max": 29.213, + "n": 3000.0 + }, + "load_1min": "19.23" + }, + { + "system": "celery", + "scenario": "pickup", + "run": 2, + "workers": 8, + "payload": 2048, + "arrival_rate": 200, + "jobs": 3000, + "latency_ms": { + "p50": 4.563808441162109, + "p95": 9.957492351531982, + "p99": 14.362025260925293, + "max": 28.88655662536621, + "n": 3000.0 + }, + "load_1min": "22.38" + }, + { + "system": "pgqueue", + "scenario": "drain-size", + "run": 2, + "workers": 4, + "payload": 256, + "label": "small", + "jobs": 10000, + "rate": 2318.7867377223224, + "load_1min": "23.87" + }, + { + "system": "celery", + "scenario": "drain-size", + "run": 2, + "workers": 4, + "payload": 256, + "label": "small", + "jobs": 10000, + "rate": 1159.339103356094, + "load_1min": "26.01" + }, + { + "system": "pgqueue", + "scenario": "drain-size", + "run": 2, + "workers": 4, + "payload": 2048, + "label": "medium", + "jobs": 10000, + "rate": 2329.100608076651, + "load_1min": "25.17" + }, + { + "system": "celery", + "scenario": "drain-size", + "run": 2, + "workers": 4, + "payload": 2048, + "label": "medium", + "jobs": 10000, + "rate": 1045.4916602035116, + "load_1min": "25.18" + }, + { + "system": "pgqueue", + "scenario": "drain-size", + "run": 2, + "workers": 4, + "payload": 16384, + "label": "large", + "jobs": 10000, + "rate": 2599.9615013736934, + "load_1min": "20.40" + }, + { + "system": "celery", + "scenario": "drain-size", + "run": 2, + "workers": 4, + "payload": 16384, + "label": "large", + "jobs": 10000, + "rate": 992.3490402814351, + "load_1min": "15.07" + }, + { + "system": "pgqueue", + "scenario": "enqueue", + "run": 3, + "payload": 2048, + "variant": "single", + "jobs": 5000, + "seconds": 34.49851580500399, + "rate": 144.93377130371368, + "latency_ms": { + "p50": 5.66179100133013, + "p95": 14.014201795362169, + "p99": 20.782233447243925, + "max": 541.6902500001015, + "n": 5000.0 + }, + "load_1min": "11.23" + }, + { + "system": "celery", + "scenario": "enqueue", + "run": 3, + "payload": 2048, + "variant": "single", + "jobs": 5000, + "seconds": 8.841205862998322, + "rate": 565.5337153640655, + "latency_ms": { + "p50": 1.7083099992305506, + "p95": 2.79489669483155, + "p99": 4.212858436003444, + "max": 10.707467001338955, + "n": 5000.0 + }, + "load_1min": "10.57" + }, + { + "system": "pgqueue", + "scenario": "enqueue", + "run": 3, + "payload": 2048, + "variant": "ten", + "jobs": 5000, + "seconds": 19.933400506000908, + "rate": 250.83527512000578, + "latency_ms": { + "p50": 38.08796299927053, + "p95": 52.84257829807757, + "p99": 100.80564835472615, + "max": 492.93822600157, + "n": 5000.0 + }, + "load_1min": "12.33" + }, + { + "system": "celery", + "scenario": "enqueue", + "run": 3, + "payload": 2048, + "variant": "ten", + "jobs": 5000, + "seconds": 7.887904863004223, + "rate": 633.8818845864829, + "latency_ms": { + "p50": 12.610375499207294, + "p95": 36.40313139658247, + "p99": 52.02377141395118, + "max": 263.5603800008539, + "n": 5000.0 + }, + "load_1min": "13.58" + }, + { + "system": "pgqueue", + "scenario": "enqueue", + "run": 3, + "payload": 2048, + "variant": "bulk", + "jobs": 40000, + "seconds": 6.934848261997104, + "rate": 5767.970471567429, + "latency_ms": { + "p50": 170.0141290020838, + "p95": 221.04751035149093, + "p99": 256.4960592490388, + "max": 260.2016149976407, + "n": 40.0 + }, + "load_1min": "14.44" + }, + { + "system": "pgqueue", + "scenario": "drain", + "run": 3, + "workers": 1, + "payload": 2048, + "jobs": 20000, + "rate": 741.5935921056312, + "load_1min": "15.74" + }, + { + "system": "celery", + "scenario": "drain", + "run": 3, + "workers": 1, + "payload": 2048, + "jobs": 20000, + "rate": 348.83336808476884, + "load_1min": "17.27" + }, + { + "system": "pgqueue", + "scenario": "drain", + "run": 3, + "workers": 2, + "payload": 2048, + "jobs": 20000, + "rate": 1675.3718797694773, + "load_1min": "18.24" + }, + { + "system": "celery", + "scenario": "drain", + "run": 3, + "workers": 2, + "payload": 2048, + "jobs": 20000, + "rate": 662.4256528215456, + "load_1min": "11.96" + }, + { + "system": "pgqueue", + "scenario": "drain", + "run": 3, + "workers": 4, + "payload": 2048, + "jobs": 20000, + "rate": 3914.6534252165925, + "load_1min": "11.47" + }, + { + "system": "celery", + "scenario": "drain", + "run": 3, + "workers": 4, + "payload": 2048, + "jobs": 20000, + "rate": 1764.5011301271961, + "load_1min": "9.06" + }, + { + "system": "pgqueue", + "scenario": "drain", + "run": 3, + "workers": 8, + "payload": 2048, + "jobs": 20000, + "rate": 4160.502414186186, + "load_1min": "8.88" + }, + { + "system": "celery", + "scenario": "drain", + "run": 3, + "workers": 8, + "payload": 2048, + "jobs": 20000, + "rate": 3422.487819796637, + "load_1min": "8.82" + }, + { + "system": "pgqueue", + "scenario": "pickup", + "run": 3, + "workers": 1, + "payload": 2048, + "arrival_rate": 200, + "jobs": 3000, + "latency_ms": { + "p50": 10.039499999999999, + "p95": 21.312400000000004, + "p99": 38.15044, + "max": 186.268, + "n": 3000.0 + }, + "load_1min": "7.44" + }, + { + "system": "celery", + "scenario": "pickup", + "run": 3, + "workers": 1, + "payload": 2048, + "arrival_rate": 200, + "jobs": 3000, + "latency_ms": { + "p50": 3.507852554321289, + "p95": 4.242002964019775, + "p99": 5.659053325653076, + "max": 15.669584274291992, + "n": 3000.0 + }, + "load_1min": "6.13" + }, + { + "system": "pgqueue", + "scenario": "pickup", + "run": 3, + "workers": 2, + "payload": 2048, + "arrival_rate": 200, + "jobs": 3000, + "latency_ms": { + "p50": 6.327000000000001, + "p95": 14.788200000000002, + "p99": 22.34689, + "max": 166.036, + "n": 3000.0 + }, + "load_1min": "5.92" + }, + { + "system": "celery", + "scenario": "pickup", + "run": 3, + "workers": 2, + "payload": 2048, + "arrival_rate": 200, + "jobs": 3000, + "latency_ms": { + "p50": 3.754854202270508, + "p95": 6.616246700286865, + "p99": 7.644095420837402, + "max": 14.815330505371094, + "n": 3000.0 + }, + "load_1min": "6.09" + }, + { + "system": "pgqueue", + "scenario": "pickup", + "run": 3, + "workers": 4, + "payload": 2048, + "arrival_rate": 200, + "jobs": 3000, + "latency_ms": { + "p50": 6.5104999999999995, + "p95": 14.164350000000002, + "p99": 20.85005, + "max": 255.088, + "n": 3000.0 + }, + "load_1min": "7.05" + }, + { + "system": "celery", + "scenario": "pickup", + "run": 3, + "workers": 4, + "payload": 2048, + "arrival_rate": 200, + "jobs": 3000, + "latency_ms": { + "p50": 3.7631988525390625, + "p95": 5.845975875854492, + "p99": 7.403628826141357, + "max": 12.630224227905273, + "n": 3000.0 + }, + "load_1min": "5.97" + }, + { + "system": "pgqueue", + "scenario": "pickup", + "run": 3, + "workers": 8, + "payload": 2048, + "arrival_rate": 200, + "jobs": 3000, + "latency_ms": { + "p50": 7.151500000000001, + "p95": 20.5187, + "p99": 61.96403, + "max": 200.181, + "n": 3000.0 + }, + "load_1min": "9.09" + }, + { + "system": "celery", + "scenario": "pickup", + "run": 3, + "workers": 8, + "payload": 2048, + "arrival_rate": 200, + "jobs": 3000, + "latency_ms": { + "p50": 4.999041557312012, + "p95": 12.883663177490234, + "p99": 20.632174015045166, + "max": 40.76695442199707, + "n": 3000.0 + }, + "load_1min": "12.64" + }, + { + "system": "pgqueue", + "scenario": "drain-size", + "run": 3, + "workers": 4, + "payload": 256, + "label": "small", + "jobs": 10000, + "rate": 2563.7229062124393, + "load_1min": "20.38" + }, + { + "system": "celery", + "scenario": "drain-size", + "run": 3, + "workers": 4, + "payload": 256, + "label": "small", + "jobs": 10000, + "rate": 3466.183224889127, + "load_1min": "17.19" + }, + { + "system": "pgqueue", + "scenario": "drain-size", + "run": 3, + "workers": 4, + "payload": 2048, + "label": "medium", + "jobs": 10000, + "rate": 3014.8124813295, + "load_1min": "16.21" + }, + { + "system": "celery", + "scenario": "drain-size", + "run": 3, + "workers": 4, + "payload": 2048, + "label": "medium", + "jobs": 10000, + "rate": 3403.045913720026, + "load_1min": "13.76" + }, + { + "system": "pgqueue", + "scenario": "drain-size", + "run": 3, + "workers": 4, + "payload": 16384, + "label": "large", + "jobs": 10000, + "rate": 1489.1287280372983, + "load_1min": "11.47" + }, + { + "system": "celery", + "scenario": "drain-size", + "run": 3, + "workers": 4, + "payload": 16384, + "label": "large", + "jobs": 10000, + "rate": 1921.1655508774245, + "load_1min": "9.79" + }, + { + "system": "pgqueue", + "scenario": "enqueue", + "run": 4, + "payload": 2048, + "variant": "single", + "jobs": 5000, + "seconds": 27.13840025999525, + "rate": 184.24077882624889, + "latency_ms": { + "p50": 4.1135624996968545, + "p95": 11.65304429414391, + "p99": 18.08264980893, + "max": 61.1318070004927, + "n": 5000.0 + }, + "load_1min": "7.84" + }, + { + "system": "celery", + "scenario": "enqueue", + "run": 4, + "payload": 2048, + "variant": "single", + "jobs": 5000, + "seconds": 2.9728898409957765, + "rate": 1681.865211098854, + "latency_ms": { + "p50": 0.5128524971951265, + "p95": 0.927027744546649, + "p99": 1.1197004620044027, + "max": 3.1106969981919974, + "n": 5000.0 + }, + "load_1min": "7.37" + }, + { + "system": "pgqueue", + "scenario": "enqueue", + "run": 4, + "payload": 2048, + "variant": "ten", + "jobs": 5000, + "seconds": 22.525252996994823, + "rate": 221.97308952166125, + "latency_ms": { + "p50": 30.761746998905437, + "p95": 145.85516980623652, + "p99": 266.00960627532913, + "max": 504.965855005139, + "n": 5000.0 + }, + "load_1min": "7.85" + }, + { + "system": "celery", + "scenario": "enqueue", + "run": 4, + "payload": 2048, + "variant": "ten", + "jobs": 5000, + "seconds": 5.4819459520003875, + "rate": 912.0848771184029, + "latency_ms": { + "p50": 7.774089001031825, + "p95": 29.739558155779378, + "p99": 46.220146818232024, + "max": 249.2208199982997, + "n": 5000.0 + }, + "load_1min": "7.46" + }, + { + "system": "pgqueue", + "scenario": "enqueue", + "run": 4, + "payload": 2048, + "variant": "bulk", + "jobs": 40000, + "seconds": 3.541166109003825, + "rate": 11295.714114707967, + "latency_ms": { + "p50": 86.51759999702335, + "p95": 106.21537289698608, + "p99": 142.07703389125527, + "max": 164.0927870030282, + "n": 40.0 + }, + "load_1min": "7.03" + }, + { + "system": "pgqueue", + "scenario": "drain", + "run": 4, + "workers": 1, + "payload": 2048, + "jobs": 20000, + "rate": 1085.7065487329562, + "load_1min": "5.75" + }, + { + "system": "celery", + "scenario": "drain", + "run": 4, + "workers": 1, + "payload": 2048, + "jobs": 20000, + "rate": 483.1513060312756, + "load_1min": "3.55" + }, + { + "system": "pgqueue", + "scenario": "drain", + "run": 4, + "workers": 2, + "payload": 2048, + "jobs": 20000, + "rate": 2193.0171114045147, + "load_1min": "3.92" + }, + { + "system": "celery", + "scenario": "drain", + "run": 4, + "workers": 2, + "payload": 2048, + "jobs": 20000, + "rate": 1201.2801499033976, + "load_1min": "3.41" + }, + { + "system": "pgqueue", + "scenario": "drain", + "run": 4, + "workers": 4, + "payload": 2048, + "jobs": 20000, + "rate": 4068.2656057050654, + "load_1min": "3.36" + }, + { + "system": "celery", + "scenario": "drain", + "run": 4, + "workers": 4, + "payload": 2048, + "jobs": 20000, + "rate": 3354.2605324597284, + "load_1min": "3.08" + }, + { + "system": "pgqueue", + "scenario": "drain", + "run": 4, + "workers": 8, + "payload": 2048, + "jobs": 20000, + "rate": 5664.7663027829085, + "load_1min": "5.73" + }, + { + "system": "celery", + "scenario": "drain", + "run": 4, + "workers": 8, + "payload": 2048, + "jobs": 20000, + "rate": 6747.352330724353, + "load_1min": "4.87" + }, + { + "system": "pgqueue", + "scenario": "pickup", + "run": 4, + "workers": 1, + "payload": 2048, + "arrival_rate": 200, + "jobs": 3000, + "latency_ms": { + "p50": 6.0975, + "p95": 14.7863, + "p99": 19.58638, + "max": 54.889, + "n": 3000.0 + }, + "load_1min": "3.93" + }, + { + "system": "celery", + "scenario": "pickup", + "run": 4, + "workers": 1, + "payload": 2048, + "arrival_rate": 200, + "jobs": 3000, + "latency_ms": { + "p50": 2.7168989181518555, + "p95": 3.2401204109191895, + "p99": 3.6842989921569824, + "max": 13.327360153198242, + "n": 3000.0 + }, + "load_1min": "3.35" + }, + { + "system": "pgqueue", + "scenario": "pickup", + "run": 4, + "workers": 2, + "payload": 2048, + "arrival_rate": 200, + "jobs": 3000, + "latency_ms": { + "p50": 3.8835, + "p95": 10.295049999999998, + "p99": 20.794640000000005, + "max": 273.911, + "n": 3000.0 + }, + "load_1min": "3.60" + }, + { + "system": "celery", + "scenario": "pickup", + "run": 4, + "workers": 2, + "payload": 2048, + "arrival_rate": 200, + "jobs": 3000, + "latency_ms": { + "p50": 2.800464630126953, + "p95": 3.2862067222595215, + "p99": 3.7076759338378906, + "max": 6.489753723144531, + "n": 3000.0 + }, + "load_1min": "3.00" + }, + { + "system": "pgqueue", + "scenario": "pickup", + "run": 4, + "workers": 4, + "payload": 2048, + "arrival_rate": 200, + "jobs": 3000, + "latency_ms": { + "p50": 3.79, + "p95": 5.4211, + "p99": 10.428310000000002, + "max": 98.603, + "n": 3000.0 + }, + "load_1min": "3.05" + }, + { + "system": "celery", + "scenario": "pickup", + "run": 4, + "workers": 4, + "payload": 2048, + "arrival_rate": 200, + "jobs": 3000, + "latency_ms": { + "p50": 2.7223825454711914, + "p95": 3.3565640449523926, + "p99": 3.9217662811279297, + "max": 7.006168365478516, + "n": 3000.0 + }, + "load_1min": "2.34" + }, + { + "system": "pgqueue", + "scenario": "pickup", + "run": 4, + "workers": 8, + "payload": 2048, + "arrival_rate": 200, + "jobs": 3000, + "latency_ms": { + "p50": 3.836, + "p95": 9.85255, + "p99": 13.44224, + "max": 236.1, + "n": 3000.0 + }, + "load_1min": "4.18" + }, + { + "system": "celery", + "scenario": "pickup", + "run": 4, + "workers": 8, + "payload": 2048, + "arrival_rate": 200, + "jobs": 3000, + "latency_ms": { + "p50": 2.0548105239868164, + "p95": 3.3158063888549805, + "p99": 3.6653661727905273, + "max": 7.615566253662109, + "n": 3000.0 + }, + "load_1min": "3.57" + }, + { + "system": "pgqueue", + "scenario": "drain-size", + "run": 4, + "workers": 4, + "payload": 256, + "label": "small", + "jobs": 10000, + "rate": 6369.74010590153, + "load_1min": "3.57" + }, + { + "system": "celery", + "scenario": "drain-size", + "run": 4, + "workers": 4, + "payload": 256, + "label": "small", + "jobs": 10000, + "rate": 3736.738391084554, + "load_1min": "3.61" + }, + { + "system": "pgqueue", + "scenario": "drain-size", + "run": 4, + "workers": 4, + "payload": 2048, + "label": "medium", + "jobs": 10000, + "rate": 4872.676759008726, + "load_1min": "3.89" + }, + { + "system": "celery", + "scenario": "drain-size", + "run": 4, + "workers": 4, + "payload": 2048, + "label": "medium", + "jobs": 10000, + "rate": 3395.2207003329136, + "load_1min": "3.83" + }, + { + "system": "pgqueue", + "scenario": "drain-size", + "run": 4, + "workers": 4, + "payload": 16384, + "label": "large", + "jobs": 10000, + "rate": 5484.933784534009, + "load_1min": "3.75" + }, + { + "system": "celery", + "scenario": "drain-size", + "run": 4, + "workers": 4, + "payload": 16384, + "label": "large", + "jobs": 10000, + "rate": 2910.794568502165, + "load_1min": "3.27" + }, + { + "system": "pgqueue", + "scenario": "enqueue", + "run": 5, + "payload": 2048, + "variant": "single", + "jobs": 5000, + "seconds": 18.10742730199854, + "rate": 276.1297845690173, + "latency_ms": { + "p50": 3.203447999112541, + "p95": 4.869761601366918, + "p99": 11.2776507774106, + "max": 159.1634470023564, + "n": 5000.0 + }, + "load_1min": "3.21" + }, + { + "system": "celery", + "scenario": "enqueue", + "run": 5, + "payload": 2048, + "variant": "single", + "jobs": 5000, + "seconds": 2.9161221949980245, + "rate": 1714.6057900373366, + "latency_ms": { + "p50": 0.5053759996371809, + "p95": 1.0290260968758957, + "p99": 1.4984722039662302, + "max": 5.52466900262516, + "n": 5000.0 + }, + "load_1min": "3.19" + }, + { + "system": "pgqueue", + "scenario": "enqueue", + "run": 5, + "payload": 2048, + "variant": "ten", + "jobs": 5000, + "seconds": 13.930296312006249, + "rate": 358.9299098893251, + "latency_ms": { + "p50": 27.764914997533197, + "p95": 32.50564004993066, + "p99": 42.038412287147366, + "max": 139.36303799710004, + "n": 5000.0 + }, + "load_1min": "3.01" + }, + { + "system": "celery", + "scenario": "enqueue", + "run": 5, + "payload": 2048, + "variant": "ten", + "jobs": 5000, + "seconds": 4.855084426999383, + "rate": 1029.8482086520955, + "latency_ms": { + "p50": 7.09863800148014, + "p95": 25.529733846269664, + "p99": 42.25982627023768, + "max": 139.9783020024188, + "n": 5000.0 + }, + "load_1min": "3.01" + }, + { + "system": "pgqueue", + "scenario": "enqueue", + "run": 5, + "payload": 2048, + "variant": "bulk", + "jobs": 40000, + "seconds": 2.7041835950003588, + "rate": 14791.895074710967, + "latency_ms": { + "p50": 66.05182099883677, + "p95": 75.09466004848946, + "p99": 81.53902892052429, + "max": 83.13275300315581, + "n": 40.0 + }, + "load_1min": "2.85" + }, + { + "system": "pgqueue", + "scenario": "drain", + "run": 5, + "workers": 1, + "payload": 2048, + "jobs": 20000, + "rate": 1398.7215264581648, + "load_1min": "3.43" + }, + { + "system": "celery", + "scenario": "drain", + "run": 5, + "workers": 1, + "payload": 2048, + "jobs": 20000, + "rate": 499.2201708228078, + "load_1min": "3.04" + }, + { + "system": "pgqueue", + "scenario": "drain", + "run": 5, + "workers": 2, + "payload": 2048, + "jobs": 20000, + "rate": 2527.7023693714336, + "load_1min": "3.44" + }, + { + "system": "celery", + "scenario": "drain", + "run": 5, + "workers": 2, + "payload": 2048, + "jobs": 20000, + "rate": 1267.8315052941882, + "load_1min": "3.52" + }, + { + "system": "pgqueue", + "scenario": "drain", + "run": 5, + "workers": 4, + "payload": 2048, + "jobs": 20000, + "rate": 5475.173831349194, + "load_1min": "3.96" + }, + { + "system": "celery", + "scenario": "drain", + "run": 5, + "workers": 4, + "payload": 2048, + "jobs": 20000, + "rate": 2537.7084165343013, + "load_1min": "4.25" + }, + { + "system": "pgqueue", + "scenario": "drain", + "run": 5, + "workers": 8, + "payload": 2048, + "jobs": 20000, + "rate": 8142.137886836757, + "load_1min": "4.07" + }, + { + "system": "celery", + "scenario": "drain", + "run": 5, + "workers": 8, + "payload": 2048, + "jobs": 20000, + "rate": 5898.2492160144, + "load_1min": "5.22" + }, + { + "system": "pgqueue", + "scenario": "pickup", + "run": 5, + "workers": 1, + "payload": 2048, + "arrival_rate": 200, + "jobs": 3000, + "latency_ms": { + "p50": 9.539, + "p95": 20.44395, + "p99": 27.238259999999997, + "max": 50.666, + "n": 3000.0 + }, + "load_1min": "4.80" + }, + { + "system": "celery", + "scenario": "pickup", + "run": 5, + "workers": 1, + "payload": 2048, + "arrival_rate": 200, + "jobs": 3000, + "latency_ms": { + "p50": 3.3731460571289062, + "p95": 5.09103536605835, + "p99": 5.770459175109863, + "max": 7.795095443725586, + "n": 3000.0 + }, + "load_1min": "4.01" + }, + { + "system": "pgqueue", + "scenario": "pickup", + "run": 5, + "workers": 2, + "payload": 2048, + "arrival_rate": 200, + "jobs": 3000, + "latency_ms": { + "p50": 5.014500000000001, + "p95": 12.465549999999999, + "p99": 19.3887, + "max": 63.199, + "n": 3000.0 + }, + "load_1min": "5.72" + }, + { + "system": "celery", + "scenario": "pickup", + "run": 5, + "workers": 2, + "payload": 2048, + "arrival_rate": 200, + "jobs": 3000, + "latency_ms": { + "p50": 2.4083852767944336, + "p95": 4.565095901489258, + "p99": 5.966072082519531, + "max": 8.34345817565918, + "n": 3000.0 + }, + "load_1min": "4.72" + }, + { + "system": "pgqueue", + "scenario": "pickup", + "run": 5, + "workers": 4, + "payload": 2048, + "arrival_rate": 200, + "jobs": 3000, + "latency_ms": { + "p50": 5.398, + "p95": 10.129, + "p99": 12.63306, + "max": 28.3, + "n": 3000.0 + }, + "load_1min": "5.43" + }, + { + "system": "celery", + "scenario": "pickup", + "run": 5, + "workers": 4, + "payload": 2048, + "arrival_rate": 200, + "jobs": 3000, + "latency_ms": { + "p50": 3.052234649658203, + "p95": 5.911779403686523, + "p99": 6.440732479095459, + "max": 10.005712509155273, + "n": 3000.0 + }, + "load_1min": "4.51" + }, + { + "system": "pgqueue", + "scenario": "pickup", + "run": 5, + "workers": 8, + "payload": 2048, + "arrival_rate": 200, + "jobs": 3000, + "latency_ms": { + "p50": 5.810999999999999, + "p95": 13.385250000000001, + "p99": 17.707310000000003, + "max": 90.209, + "n": 3000.0 + }, + "load_1min": "6.13" + }, + { + "system": "celery", + "scenario": "pickup", + "run": 5, + "workers": 8, + "payload": 2048, + "arrival_rate": 200, + "jobs": 3000, + "latency_ms": { + "p50": 4.0923357009887695, + "p95": 6.211698055267334, + "p99": 6.686742305755615, + "max": 8.281230926513672, + "n": 3000.0 + }, + "load_1min": "5.34" + }, + { + "system": "pgqueue", + "scenario": "drain-size", + "run": 5, + "workers": 4, + "payload": 256, + "label": "small", + "jobs": 10000, + "rate": 2997.8103418914275, + "load_1min": "5.34" + }, + { + "system": "celery", + "scenario": "drain-size", + "run": 5, + "workers": 4, + "payload": 256, + "label": "small", + "jobs": 10000, + "rate": 3163.05426233278, + "load_1min": "5.75" + }, + { + "system": "pgqueue", + "scenario": "drain-size", + "run": 5, + "workers": 4, + "payload": 2048, + "label": "medium", + "jobs": 10000, + "rate": 3260.9717188351497, + "load_1min": "5.61" + }, + { + "system": "celery", + "scenario": "drain-size", + "run": 5, + "workers": 4, + "payload": 2048, + "label": "medium", + "jobs": 10000, + "rate": 2281.2304132437093, + "load_1min": "5.61" + }, + { + "system": "pgqueue", + "scenario": "drain-size", + "run": 5, + "workers": 4, + "payload": 16384, + "label": "large", + "jobs": 10000, + "rate": 2931.023494566609, + "load_1min": "5.26" + }, + { + "system": "celery", + "scenario": "drain-size", + "run": 5, + "workers": 4, + "payload": 16384, + "label": "large", + "jobs": 10000, + "rate": 2714.6834962830817, + "load_1min": "4.66" + } + ] } \ No newline at end of file diff --git a/bench/tasks.py b/bench/tasks.py index 06db2c7..99b23cd 100644 --- a/bench/tasks.py +++ b/bench/tasks.py @@ -1,4 +1,4 @@ -"""Benchmark tasks: a no-op handler, so we measure the queue, not the work.""" +"""pgqueue benchmark task: a no-op handler, so we measure the queue, not the work.""" from __future__ import annotations @@ -14,10 +14,14 @@ ) -class Noop(BaseModel): +class Email(BaseModel): n: int + kind: str + recipient: str + subject: str + body: str @app.task(name="bench.noop", queue="bench") -async def noop(payload: Noop) -> None: +async def noop(payload: Email) -> None: return None diff --git a/compose.yaml b/compose.yaml index 7bcb8ba..f6bcad9 100644 --- a/compose.yaml +++ b/compose.yaml @@ -22,7 +22,7 @@ services: POSTGRES_PASSWORD: pgqueue POSTGRES_DB: pgqueue ports: - - "55432:5432" + - "${PG_PORT:-55432}:5432" healthcheck: test: ["CMD-SHELL", "pg_isready -U pgqueue -d pgqueue"] interval: 2s @@ -67,5 +67,19 @@ services: PGQUEUE_DSN: postgresql://pgqueue:pgqueue@postgres:5432/pgqueue PRODUCER_RATE: "15" + # Only for `make bench`: the broker for the Celery comparison. Valkey is the + # BSD-3-Clause fork of Redis; persistence is off, as in a typical speed-first setup. + valkey: + image: valkey/valkey:8-alpine + profiles: ["bench"] + command: ["valkey-server", "--save", "", "--appendonly", "no"] + ports: + - "${REDIS_PORT:-56379}:6379" + healthcheck: + test: ["CMD", "valkey-cli", "ping"] + interval: 2s + timeout: 3s + retries: 30 + volumes: pgdata: diff --git a/docs/benchmarks-latency.svg b/docs/benchmarks-latency.svg new file mode 100644 index 0000000..c3474bf --- /dev/null +++ b/docs/benchmarks-latency.svg @@ -0,0 +1,71 @@ + +Pickup latency by worker processesEnqueue to handler start at a steady 200 jobs/s, 2 KiB payload (lower is better) + +Pickup latency by worker processes +Enqueue to handler start at a steady 200 jobs/s, 2 KiB payload (lower is better) + +0.0 + +50.0 + +100.0 + +150.0 + +200.0 +milliseconds +1 +2 +4 +8 +worker processes + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +pgqueue p50 + + +pgqueue p99 + + +Celery p50 + + +Celery p99 +Median of 5 runs, whiskers = min to max. 13th Gen Intel(R) Core(TM) i5-13500H; PostgreSQL, Valkey, workers and load generator share it. + diff --git a/docs/benchmarks-throughput.svg b/docs/benchmarks-throughput.svg new file mode 100644 index 0000000..ac2f2aa --- /dev/null +++ b/docs/benchmarks-throughput.svg @@ -0,0 +1,47 @@ + +Backlog throughput by worker processesJobs per second draining a pre-loaded backlog, no-op handler, 2 KiB payload, 32 slots per worker + +Backlog throughput by worker processes +Jobs per second draining a pre-loaded backlog, no-op handler, 2 KiB payload, 32 slots per worker + +0 + +2,500 + +5,000 + +7,500 + +10,000 +jobs per second +1 +2 +4 +8 +worker processes + + + + + + + + + + + + + + + + + + + + +pgqueue (PostgreSQL) + + +Celery (Valkey) +Median of 5 runs, whiskers = min to max. 13th Gen Intel(R) Core(TM) i5-13500H; PostgreSQL, Valkey, workers and load generator share it. + diff --git a/docs/benchmarks.md b/docs/benchmarks.md new file mode 100644 index 0000000..c54a9d8 --- /dev/null +++ b/docs/benchmarks.md @@ -0,0 +1,189 @@ +# Benchmarks + +What the numbers in the README come from, how to reproduce them, and how far to trust them. +Short version: they were taken on a laptop that was busy with other work, the run-to-run +spread is large, and the honest summary is a range, not a point. + +Run it yourself with `make bench` (about an hour with the defaults; `make bench-quick` +checks the setup in a few minutes and produces meaningless numbers). Everything below is +generated from [`bench/results.json`](../bench/results.json) by `python -m bench.report` +and the charts by `python -m bench.chart`; no number in this file is typed in by hand. + +## Setup + +| | | +|---|---| +| Machine | 13th Gen Intel Core i5-13500H, 16 logical CPUs, 14 GiB RAM, NVMe SSD, Linux 7.0 | +| PostgreSQL | 17.11 (`postgres:17-alpine`) in Docker, default configuration: `shared_buffers = 128MB`, `synchronous_commit = on`, `fsync = on` | +| Broker for the comparison | Valkey 8 (`valkey/valkey:8-alpine`, BSD-3-Clause fork of Redis; it reports itself as 7.2.4 for protocol compatibility), persistence off (`--save "" --appendonly no`) | +| Celery | 5.6.3 with redis-py 8.1.0, thread pool, defaults otherwise (prefetch multiplier 4, no result backend, `ignore_result`) | +| Python | 3.12.13 | +| Topology | Postgres or Valkey, the workers and the load generator all run on the same machine, workers as separate OS processes | +| Handler | no-op. This measures the queue, not your work | +| Job payload | JSON object of about 2 KiB (identifiers, short strings and a free-text body, deterministic). A 256 B and a 16 KiB variant are in the payload table | +| Concurrency | 32 job slots per worker process in both systems (asyncio tasks for pgqueue, threads for Celery) | +| Repetitions | 5 runs of every cell, the two systems interleaved inside each run | + +Cells below are the **median over the 5 runs, with the minimum and maximum run in +brackets**. Raw per-run numbers, the load average at each measurement and the environment +are in `bench/results.json`. + +### What is timed + +* **Enqueue**: wall-clock time around each `enqueue()` call (pgqueue) or `apply_async()` + call (Celery), measured by the client. The call returns after the job is committed in + Postgres (with an fsync) or after the broker accepted the message into memory. + `enqueue_many` inserts 1,000 jobs in one statement; Celery has no bulk publish, so it is + not in that row. +* **Backlog throughput**: a backlog of 20,000 jobs is loaded first, then N worker + processes start and drain it. The rate is taken between the 20th and the 95th + percentile completion timestamp, so staggered worker start-up and the straggling tail do + not count. pgqueue reads `finished_at` from its job rows; the Celery task records its own + finish time in Valkey. +* **Pickup latency**: jobs arrive at a fixed 200 per second for 15 seconds (3,000 jobs) + into idle workers, and the latency is the time from enqueue to the first line of the + handler. pgqueue uses `started_at - created_at` (database clock, LISTEN/NOTIFY on, 1 s + polling as fallback); Celery compares `time.time()` in the producer and in the task, on + the same machine. + +The Celery task makes one pipelined Redis call to record its timestamps, which pgqueue does +not need because its job rows already hold them. Celery is charged that extra round trip. + +## Results + +### Enqueue + + +| Variant | System | jobs/s | p50 ms | p95 ms | p99 ms | +|---|---|---:|---:|---:|---:| +| 1 client, 1 job per call | pgqueue | 184 (111-276) | 4.4 (3.2-5.7) | 11.7 (4.9-19.5) | 18.1 (9.6-66.5) | +| 1 client, 1 job per call | Celery + Valkey | 566 (348-1,715) | 1.7 (0.5-2.9) | 2.5 (0.9-4.3) | 4.2 (1.1-5.9) | +| 10 concurrent clients | pgqueue | 251 (175-378) | 30.8 (25.4-51.3) | 52.8 (28.9-145.9) | 100.8 (33.9-266.0) | +| 10 concurrent clients | Celery + Valkey | 634 (442-1,030) | 12.6 (7.1-18.8) | 36.4 (25.5-58.0) | 52.0 (42.3-89.7) | +| `enqueue_many`, 1,000 per call (latency is per call) | pgqueue | 6,188 (5,181-14,792) | 153.1 (66.1-190.4) | 221.0 (75.1-245.0) | 245.0 (81.5-258.4) | + + +### Backlog throughput (jobs/s) + + +| Worker processes | pgqueue | Celery + Valkey | +|---:|---:|---:| +| 1 | 825 (725-1,399) | 349 (280-499) | +| 2 | 1,702 (1,427-2,528) | 722 (574-1,268) | +| 4 | 3,915 (2,444-5,475) | 1,765 (990-3,354) | +| 8 | 4,161 (2,589-8,142) | 3,422 (1,683-6,747) | + + +![Backlog throughput by number of worker processes, pgqueue and Celery](benchmarks-throughput.svg) + +### Pickup latency at 200 jobs/s (ms) + + +| Worker processes | System | p50 | p95 | p99 | max | +|---:|---|---:|---:|---:|---:| +| 1 | pgqueue | 7.3 (6.1-10.0) | 17.4 (14.8-21.3) | 38.2 (19.6-152.1) | 145.7 (50.7-276.8) | +| 1 | Celery + Valkey | 3.5 (2.7-5.3) | 5.1 (3.2-53.6) | 5.8 (3.7-87.7) | 15.7 (7.8-137.9) | +| 2 | pgqueue | 5.0 (3.9-6.3) | 12.5 (8.5-14.8) | 22.3 (19.4-58.6) | 166.0 (63.2-273.9) | +| 2 | Celery + Valkey | 3.6 (2.4-4.4) | 6.6 (3.3-8.5) | 7.6 (3.7-11.0) | 14.8 (6.5-20.4) | +| 4 | pgqueue | 4.5 (3.8-6.5) | 9.5 (5.4-14.2) | 15.2 (10.4-20.9) | 49.1 (28.3-255.1) | +| 4 | Celery + Valkey | 3.5 (2.7-4.2) | 5.9 (3.4-8.1) | 7.4 (3.9-11.0) | 12.6 (7.0-17.8) | +| 8 | pgqueue | 5.5 (3.8-7.2) | 12.4 (7.4-20.5) | 17.5 (10.2-62.0) | 90.2 (28.3-236.1) | +| 8 | Celery + Valkey | 4.5 (2.1-5.0) | 8.3 (3.3-12.9) | 11.3 (3.7-20.6) | 26.9 (7.6-40.8) | + + +![Pickup latency by number of worker processes, pgqueue and Celery](benchmarks-latency.svg) + +### Payload size, 4 workers, backlog throughput (jobs/s) + + +| Payload | pgqueue | Celery + Valkey | +|---|---:|---:| +| 256 B | 2,998 (2,319-6,370) | 3,163 (1,029-3,737) | +| 2,048 B | 3,261 (2,329-4,873) | 2,281 (1,036-3,403) | +| 16,384 B | 2,931 (1,489-5,485) | 1,921 (984-2,911) | + +Load average (1 minute) sampled at each measurement: min 2.3, median 10.4, max 37.7. + + +## How to read this + +**The spread is as important as the median.** The laptop was also running unrelated work +(other benchmarks and containers) during these runs: the 1-minute load average was 38 when +the first run started and 4 to 5 when the last one finished, on a 16-thread CPU. The runs +that executed on a quieter machine are visibly faster. For example, 8 pgqueue workers +drained the backlog at 2,589, 3,893, 4,161, 5,665 and 8,142 jobs/s in runs 1 to 5, with +load averages between 19 and 4 at the time. A median of 4,161 jobs/s is therefore +a middle value of a noisy sample, not a property of the queue. Treat the figures as an +order of magnitude: low thousands of jobs per second per Postgres on a laptop, no-op +handler, default Postgres settings. + +**Comparisons are more trustworthy than absolute numbers**, because both systems ran +inside the same run, a few seconds apart, under similar load. In the backlog scenario +pgqueue was faster than Celery in 5 of 5 runs at 1, 2 and 4 workers (1.2 to 2.8 times) and +in 4 of 5 runs at 8 workers (0.84 to 2.3 times; run 4 went to Celery, 6,747 against +5,665 jobs/s). With only 5 runs per cell and +that variance this is a pattern, not a significance test. + +**Enqueue favours Celery, as it should.** A Valkey publish with persistence off is a memory +write; a pgqueue `enqueue()` is a durable commit. The gap on single-job enqueue is what +durability costs. If you enqueue inside a transaction you are committing anyway, the job +rides on that commit and adds one INSERT. If you can batch, `enqueue_many` amortizes the +commit across the batch. + +**Pickup latency is low for both at this rate.** With idle workers the median is a few +milliseconds for both systems. Comparing the two within each run, pgqueue's median was +between 1.0 and 2.9 times Celery's. The tails differ more: pgqueue's p99 is tens of milliseconds, with single +outliers in the hundreds under load. Do not read the 1-worker row as a scaling effect; the +runs with a loaded machine dominate the maximum values. + +**Celery on one worker is slow here** (hundreds of jobs/s) because one process with a thread +pool is limited by the GIL, while an asyncio worker spends its time waiting on the +database. A prefork pool, or the gevent pool, would change this comparison. It was not +tried, so there is no number for it. + +## Why more workers help, and where SKIP LOCKED contention comes in + +A pgqueue worker claims up to as many jobs as it has free slots with one statement +(`UPDATE ... FROM (SELECT ... FOR UPDATE SKIP LOCKED)`, see `store.py`). Without +`SKIP LOCKED`, two workers that select the same head of the queue would serialize: the +second one blocks until the first commits and then finds the rows taken. With it, a +claimer passes over rows another transaction has locked and takes the next ones in index +order, so N workers receive disjoint rows and never wait for each other. + +The remaining cost of contention is skipping, not waiting. While other workers hold their +batches, the head of the ready-index is full of locked entries, and a claimer has to step +over them to find free rows. The more workers and slots, the more locked entries sit at the +head. Besides that, all workers insert and update rows of the same table and index pages, +and every completion is its own UPDATE and commit. In this benchmark the backlog throughput +went from 825 jobs/s with one worker to roughly 4,000 with four, but the step from four to +eight workers added little at the median (3,915 to 4,161 jobs/s) while the best runs kept +growing; with a noisy machine and a 16-thread CPU shared with Postgres itself, a CPU or +disk limit and the claim-side skipping cannot be told apart from these runs. This +benchmark did not instrument lock skipping, so the contention explanation above describes +how the mechanism works, not a measured cause of this curve. If you want to measure it, +compare `pg_stat_activity` wait events and the claim statement time (`pg_stat_statements`) +across worker counts on a dedicated machine. + +## Caveats + +* **One machine, one configuration.** Postgres, Valkey, the workers and the load generator + compete for the same 16 threads and the same disk. On separate hosts the network adds + latency and the CPU contention disappears. Neither is modeled here. +* **Noisy host.** See above. The load averages are stored with every measurement. +* **Five runs.** Enough to show the spread, not enough for confidence intervals. +* **No-op handler.** Real jobs spend time in their handler, which moves the bottleneck away + from the queue. For handlers that take tens of milliseconds or longer, the queue is + unlikely to be the limit at these rates. +* **Default Postgres.** `shared_buffers = 128MB` and the default WAL settings. Tuning + would likely move the numbers, and it was not tried. +* **Durability is not equal.** pgqueue commits every job with an fsync; Valkey runs without + persistence. A Valkey with `appendonly yes` would be slower to enqueue; it was not + measured here. +* **Celery configuration.** One configuration (thread pool, 32 threads per process, + defaults otherwise). Other pools and settings were not measured. RQ was not benchmarked. +* **Steady-state window.** Backlog throughput excludes the first 20% and the last 5% of + the completions. It is not the time to drain the whole backlog. +* **The dataset is small.** The 20,000-job backlog stays in memory; a queue table of tens + of millions of rows, bloat, autovacuum and retention were not tested. +* **Single-job enqueue measures a sequential client.** The "10 concurrent clients" row shares + one asyncpg pool (pgqueue) or ten threads (Celery); both are bound by the same machine. diff --git a/pyproject.toml b/pyproject.toml index 2a20917..c44178a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -33,6 +33,11 @@ admin = [ pgqueue = "pgqueue.cli:main" [dependency-groups] +# Only for the Celery comparison in `make bench`; not needed by the library or CI. +bench = [ + "celery>=5.6,<6", + "redis>=8.1,<9", +] dev = [ "pytest>=9.1,<10", "pytest-asyncio>=1.4,<2", @@ -62,8 +67,9 @@ ignore = ["PLR0913", "PLR2004", "PLR0912", "PLR0915"] [tool.ruff.lint.per-file-ignores] "tests/**" = ["PLR2004", "S101"] -# Benchmarks poll the database on purpose: that is what they measure. -"bench/**" = ["ASYNC110"] +# Benchmarks poll the database on purpose (that is what they measure) and import the +# optional Celery group lazily so that pgqueue-only runs do not need it. +"bench/**" = ["ASYNC110", "ASYNC109", "ASYNC240", "PLC0415"] [tool.mypy] python_version = "3.12" @@ -72,7 +78,7 @@ plugins = ["pydantic.mypy"] files = ["src", "tests", "examples", "bench", "demo"] [[tool.mypy.overrides]] -module = ["asyncpg", "asyncpg.*", "prometheus_client", "prometheus_client.*"] +module = ["asyncpg", "asyncpg.*", "prometheus_client", "prometheus_client.*", "celery", "celery.*", "redis", "redis.*"] ignore_missing_imports = true [tool.pytest.ini_options] diff --git a/uv.lock b/uv.lock index 35f2102..5da2d53 100644 --- a/uv.lock +++ b/uv.lock @@ -23,6 +23,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/3f/27/78a89b55b0904d222183164e079b4ca56208e94eff1d35ad1f1ad5be9b06/alembic-1.20.0-py3-none-any.whl", hash = "sha256:77eb101048d95f982c0353e9233404889dcd7a6fc244c107836c0e2fc9cf7d9d", size = 268719, upload-time = "2026-09-11T19:09:12.88Z" }, ] +[[package]] +name = "amqp" +version = "5.4.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "vine" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/b2/a7/787f067b237d4b3bfefa8d67ead64a7d807feafe85ad516afb284fb1587f/amqp-5.4.0.tar.gz", hash = "sha256:aaa33987dcb6a7893955d3b4f537c5d4755addce1009929e63e5d18c1b51a0a7", size = 131733, upload-time = "2026-09-19T08:39:40.147Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/07/99/6b965aa99a32630c1c61ce57c9dcf81cec6fec354274a87f14525302b7f7/amqp-5.4.0-py3-none-any.whl", hash = "sha256:5d931d8442a403b6d4a5725a868a55edaa40734eae37d282b16be9fa710bd8ee", size = 51856, upload-time = "2026-09-19T08:39:38.404Z" }, +] + [[package]] name = "annotated-doc" version = "0.0.5" @@ -158,6 +170,35 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/3c/d7/8fb3044eaef08a310acfe23dae9a8e2e07d305edc29a53497e52bc76eca7/asyncpg-0.31.0-cp314-cp314t-win_amd64.whl", hash = "sha256:bd4107bb7cdd0e9e65fae66a62afd3a249663b844fa34d479f6d5b3bef9c04c3", size = 706062, upload-time = "2025-11-24T23:26:44.086Z" }, ] +[[package]] +name = "billiard" +version = "4.3.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/60/a9/b1c91c6866717e68b12331222f294e20df4b7674f58c14639ecd0166549c/billiard-4.3.0.tar.gz", hash = "sha256:3ddeaf82c0373e076c7b450c2fdc7ded6d73981bcfd931da1cdd0c0a03468dc5", size = 160930, upload-time = "2026-09-17T08:38:04.925Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/60/60/ade6d336983b82fbc330d0d0371f945249b448152852952f643787a3288d/billiard-4.3.0-py3-none-any.whl", hash = "sha256:be3e51fb6202df38e1d67ff0aa3bfa4cdc1838a3c47c36373cc3220c6b787c2a", size = 88116, upload-time = "2026-09-17T08:38:03.233Z" }, +] + +[[package]] +name = "celery" +version = "5.6.3" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "billiard" }, + { name = "click" }, + { name = "click-didyoumean" }, + { name = "click-plugins" }, + { name = "click-repl" }, + { name = "kombu" }, + { name = "python-dateutil" }, + { name = "tzlocal" }, + { name = "vine" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/e8/b4/a1233943ab5c8ea05fb877a88a0a0622bf47444b99e4991a8045ac37ea1d/celery-5.6.3.tar.gz", hash = "sha256:177006bd2054b882e9f01be59abd8529e88879ef50d7918a7050c5a9f4e12912", size = 1742243, upload-time = "2026-03-26T12:14:51.76Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/cf/c9/6eccdda96e098f7ae843162db2d3c149c6931a24fda69fe4ab84d0027eb5/celery-5.6.3-py3-none-any.whl", hash = "sha256:0808f42f80909c4d5833202360ffafb2a4f83f4d8e23e1285d926610e9a7afa6", size = 451235, upload-time = "2026-03-26T12:14:49.491Z" }, +] + [[package]] name = "click" version = "8.5.0" @@ -167,6 +208,44 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/58/50/6c0d534c5f134586a8e1ba4e330569e32f057e33372ae556463212fb4cd3/click-8.5.0-py3-none-any.whl", hash = "sha256:255bc9599cf7748b4b1a446ccc735421bd08a2ae529a8b88597d3de5664ee360", size = 125251, upload-time = "2026-08-26T13:33:12.928Z" }, ] +[[package]] +name = "click-didyoumean" +version = "0.3.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "click" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/30/ce/217289b77c590ea1e7c24242d9ddd6e249e52c795ff10fac2c50062c48cb/click_didyoumean-0.3.1.tar.gz", hash = "sha256:4f82fdff0dbe64ef8ab2279bd6aa3f6a99c3b28c05aa09cbfc07c9d7fbb5a463", size = 3089, upload-time = "2024-03-24T08:22:07.499Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/1b/5b/974430b5ffdb7a4f1941d13d83c64a0395114503cc357c6b9ae4ce5047ed/click_didyoumean-0.3.1-py3-none-any.whl", hash = "sha256:5c4bb6007cfea5f2fd6583a2fb6701a22a41eb98957e63d0fac41c10e7c3117c", size = 3631, upload-time = "2024-03-24T08:22:06.356Z" }, +] + +[[package]] +name = "click-plugins" +version = "1.1.1.2" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "click" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/c3/a4/34847b59150da33690a36da3681d6bbc2ec14ee9a846bc30a6746e5984e4/click_plugins-1.1.1.2.tar.gz", hash = "sha256:d7af3984a99d243c131aa1a828331e7630f4a88a9741fd05c927b204bcf92261", size = 8343, upload-time = "2025-06-25T00:47:37.555Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/3d/9a/2abecb28ae875e39c8cad711eb1186d8d14eab564705325e77e4e6ab9ae5/click_plugins-1.1.1.2-py2.py3-none-any.whl", hash = "sha256:008d65743833ffc1f5417bf0e78e8d2c23aab04d9745ba817bd3e71b0feb6aa6", size = 11051, upload-time = "2025-06-25T00:47:36.731Z" }, +] + +[[package]] +name = "click-repl" +version = "0.4.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "click" }, + { name = "prompt-toolkit" }, + { name = "typing-extensions" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/8e/f5/19bce5e62ee97138752eeef0b91bfa46d880725cc8ec00aa00daa2685410/click_repl-0.4.0.tar.gz", hash = "sha256:52f8962ff0fdae8105d568123b144951257b6bfcfb44ff6f043379c307d0a8ac", size = 15941, upload-time = "2026-09-17T07:22:01.768Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/e0/ae/63d94d251d2e1a7bf3794ea47cb1c99fe6de055b732ea79d7bcdabef8ff5/click_repl-0.4.0-py3-none-any.whl", hash = "sha256:fee730650f34204f1bd41b8fab46f6b51dda019b4a35ed89ab8553ba31f7d6d3", size = 14976, upload-time = "2026-09-17T07:22:00.429Z" }, +] + [[package]] name = "colorama" version = "0.4.6" @@ -298,6 +377,21 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/62/a1/3d680cbfd5f4b8f15abc1d571870c5fc3e594bb582bc3b64ea099db13e56/jinja2-3.1.6-py3-none-any.whl", hash = "sha256:85ece4451f492d0c13c5dd7c13a64681a86afae63a5f347908daf103ce6d2f67", size = 134899, upload-time = "2025-03-05T20:05:00.369Z" }, ] +[[package]] +name = "kombu" +version = "5.6.2" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "amqp" }, + { name = "packaging" }, + { name = "tzdata" }, + { name = "vine" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/b6/a5/607e533ed6c83ae1a696969b8e1c137dfebd5759a2e9682e26ff1b97740b/kombu-5.6.2.tar.gz", hash = "sha256:8060497058066c6f5aed7c26d7cd0d3b574990b09de842a8c5aaed0b92cc5a55", size = 472594, upload-time = "2025-12-29T20:30:07.779Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/fb/0f/834427d8c03ff1d7e867d3db3d176470c64871753252b21b4f4897d1fa45/kombu-5.6.2-py3-none-any.whl", hash = "sha256:efcfc559da324d41d61ca311b0c64965ea35b4c55cc04ee36e55386145dace93", size = 214219, upload-time = "2025-12-29T20:30:05.74Z" }, +] + [[package]] name = "librt" version = "0.15.0" @@ -575,6 +669,10 @@ admin = [ ] [package.dev-dependencies] +bench = [ + { name = "celery" }, + { name = "redis" }, +] dev = [ { name = "mypy" }, { name = "pytest" }, @@ -596,6 +694,10 @@ requires-dist = [ provides-extras = ["admin"] [package.metadata.requires-dev] +bench = [ + { name = "celery", specifier = ">=5.6,<6" }, + { name = "redis", specifier = ">=8.1,<9" }, +] dev = [ { name = "mypy", specifier = ">=2.3.1,<2.4" }, { name = "pytest", specifier = ">=9.1,<10" }, @@ -621,6 +723,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/eb/a3/b69efbf4143b5b9859b977770bbbabcc2796b702fa69dc40271e45cd5a56/prometheus_client-0.26.0-py3-none-any.whl", hash = "sha256:fa93d06737aa02bacd05794768508bb97d2fbee28cb3bca04eaae92f0ca953d6", size = 64494, upload-time = "2026-07-24T19:36:40.854Z" }, ] +[[package]] +name = "prompt-toolkit" +version = "3.0.53" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "wcwidth" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/7d/ea/39b988c938f75cb75d7045b5c69f8bfed47ee2152c8837fb403de29d6fb8/prompt_toolkit-3.0.53.tar.gz", hash = "sha256:9ec8a0ad96d5c56148b3f914aa79c1564c3fde5d2e6b876e7bc327e353cf8fa6", size = 435492, upload-time = "2026-07-26T20:56:14.758Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/54/6f/84908cad2d6aa5144abcf7b42709fe4fdb459bc640ec7ac5786e7693dabc/prompt_toolkit-3.0.53-py3-none-any.whl", hash = "sha256:01c0891d7f9237d5e339f7d3e42cdae80b7534abb1c7c0e3352efba6231492f2", size = 392288, upload-time = "2026-07-26T20:56:12.512Z" }, +] + [[package]] name = "pydantic" version = "2.13.5" @@ -749,6 +863,27 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/03/e2/08a497ef684b88559c9cc5f4ad53a37e7b99e727094a86d6ea32536d5d3c/pytest_asyncio-1.4.0-py3-none-any.whl", hash = "sha256:933ca923a23075a87fb7070c0ec272a6848489824d887c85c812670932835aa1", size = 16930, upload-time = "2026-05-26T09:56:02.576Z" }, ] +[[package]] +name = "python-dateutil" +version = "2.9.0.post0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "six" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/66/c0/0c8b6ad9f17a802ee498c46e004a0eb49bc148f2fd230864601a86dcf6db/python-dateutil-2.9.0.post0.tar.gz", hash = "sha256:37dd54208da7e1cd875388217d5e00ebd4179249f90fb72437e91a35459a0ad3", size = 342432, upload-time = "2024-03-01T18:36:20.211Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/ec/57/56b9bcc3c9c6a792fcbaf139543cee77261f3651ca9da0c93f5c1221264b/python_dateutil-2.9.0.post0-py2.py3-none-any.whl", hash = "sha256:a8b2bc7bffae282281c8140a97d3aa9c14da0b136dfe83f850eea9a5f7470427", size = 229892, upload-time = "2024-03-01T18:36:18.57Z" }, +] + +[[package]] +name = "redis" +version = "8.1.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/a8/99/604f0b666d4c616d891cf77ebb9db6bb21601344c051aebf1b72b9ff915f/redis-8.1.0.tar.gz", hash = "sha256:6e1a19beef9225c83efd689c7e6b7da2d5215b1f42cd13b7fc3714d0a09c7b25", size = 5254356, upload-time = "2026-07-30T08:51:00.269Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/66/9d/c5731f6e3608663d4d3656fd8d3aecee8b509c3082818f5a13eae925baea/redis-8.1.0-py3-none-any.whl", hash = "sha256:a4fe1aac3d3b3cc791d4b3d5931c5a956045dc951ee74d1c913ee3ac4d2ee9fb", size = 560618, upload-time = "2026-07-30T08:50:58.497Z" }, +] + [[package]] name = "ruff" version = "0.16.9" @@ -774,6 +909,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/51/60/5fb1a39dbb5ae314d5f59bc7348a63c1d5c20f3cd83914c4b5cb0be31d2d/ruff-0.16.9-py3-none-win_arm64.whl", hash = "sha256:ed1a252039200f57a59eebc063b54beabea67bfbaaca0eeaa7f54b5fbcda2284", size = 10458649, upload-time = "2026-09-24T20:37:46.882Z" }, ] +[[package]] +name = "six" +version = "1.17.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/94/e7/b2c673351809dca68a0e064b6af791aa332cf192da575fd474ed7d6f16a2/six-1.17.0.tar.gz", hash = "sha256:ff70335d468e7eb6ec65b95b99d3a2836546063f63acc5171de367e834932a81", size = 34031, upload-time = "2024-12-04T17:35:28.174Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/b7/ce/149a00dd41f10bc29e5921b496af8b574d8413afcd5e30dfa0ed46c2cc5e/six-1.17.0-py2.py3-none-any.whl", hash = "sha256:4721f391ed90541fddacab5acf947aa0d3dc7d27b2e1e8eda2be8970586c3274", size = 11050, upload-time = "2024-12-04T17:35:26.475Z" }, +] + [[package]] name = "sqlalchemy" version = "2.1.1" @@ -885,6 +1029,27 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/67/81/4add07e5172b7ac40d8ed5ff580409a7801a4fe26d529bdd915401dabfbe/typing_inspection-0.4.4-py3-none-any.whl", hash = "sha256:65b8397ba37ccbce054456aaccddfc91e6e3083c92824df348d96ca832f3f147", size = 14750, upload-time = "2026-08-12T12:37:24.648Z" }, ] +[[package]] +name = "tzdata" +version = "2026.4" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/e4/31/3d74fa778a63b98b7374323befcc0be5ab3bd94afd4096a0124e7379152c/tzdata-2026.4.tar.gz", hash = "sha256:f1b8bd365d8d210c55353f4d7f8d6d8561c0ba50d704b700d195a9424bba0d79", size = 199350, upload-time = "2026-09-12T12:56:03.251Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/f9/bc/8737e8d54cf51106118039b83f485a4783112fab49ea9d044b234978a46e/tzdata-2026.4-py2.py3-none-any.whl", hash = "sha256:c2169a8b0a7a5e9674da5a135ccdfb2b3e671b333ed9fed17b41f73c34476e81", size = 347494, upload-time = "2026-09-12T12:56:01.67Z" }, +] + +[[package]] +name = "tzlocal" +version = "5.4.4" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "tzdata", marker = "sys_platform == 'win32'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/81/5b/879b2f932adfa7a053c360d50bc896c977fa6426109185f7c12ebdd0cb9d/tzlocal-5.4.4.tar.gz", hash = "sha256:8dbb8660838688a7b6ba4fed31d18dedf842afb4d47ca050d6d891c2c15f3be4", size = 31170, upload-time = "2026-06-29T08:03:40.026Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/9e/a4/017a7a6cbe387d961a688ec31364ae60a5c4e22c96ae9921b79a947c855d/tzlocal-5.4.4-py3-none-any.whl", hash = "sha256:aae09f0126a8a86fa736be266eb4a471380d26a0de3bc14844e7821fee3e2a15", size = 18115, upload-time = "2026-06-29T08:03:38.666Z" }, +] + [[package]] name = "uvicorn" version = "0.54.0" @@ -897,3 +1062,40 @@ sdist = { url = "https://files.pythonhosted.org/packages/da/34/30e9280707135d2cf wheels = [ { url = "https://files.pythonhosted.org/packages/38/0c/b54a4fdd7f90a3af8b02ebc9ce6712c2c208b7926a2f7bad95c33ebbe943/uvicorn-0.54.0-py3-none-any.whl", hash = "sha256:505bdb0f318731d45f1f712071fc781a8981f6847a31c902c9f5e652d4f67faf", size = 87427, upload-time = "2026-09-25T06:52:35.829Z" }, ] + +[[package]] +name = "vine" +version = "5.1.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/bd/e4/d07b5f29d283596b9727dd5275ccbceb63c44a1a82aa9e4bfd20426762ac/vine-5.1.0.tar.gz", hash = "sha256:8b62e981d35c41049211cf62a0a1242d8c1ee9bd15bb196ce38aefd6799e61e0", size = 48980, upload-time = "2023-11-05T08:46:53.857Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/03/ff/7c0c86c43b3cbb927e0ccc0255cb4057ceba4799cd44ae95174ce8e8b5b2/vine-5.1.0-py3-none-any.whl", hash = "sha256:40fdf3c48b2cfe1c38a49e9ae2da6fda88e4794c810050a728bd7413811fb1dc", size = 9636, upload-time = "2023-11-05T08:46:51.205Z" }, +] + +[[package]] +name = "wcwidth" +version = "0.9.1" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/dc/ac/3a943d2792c9bb368aaa8b50121c0f778460ba2d7fbdc0a0366201d9e761/wcwidth-0.9.1.tar.gz", hash = "sha256:5823209b0d43af322ce698c689380d7c15ca31fa8e6e3be8459f27031bef0af5", size = 952410, upload-time = "2026-09-23T20:23:49.127Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/48/7c/130de33c0a7f6efecee28fef5711fc1f8d55ac33885a6b85f4f6689004ba/wcwidth-0.9.1-cp310-abi3-macosx_10_9_x86_64.whl", hash = "sha256:10b00ba23482e352f874d2e8135e7ace9da838646c7dd800566246bbd46125ff", size = 604219, upload-time = "2026-09-23T20:23:22.071Z" }, + { url = "https://files.pythonhosted.org/packages/33/7c/c7bb03da54de7ac5a08753e186b963f91847b04ed8ae5be136a4d5f52b75/wcwidth-0.9.1-cp310-abi3-macosx_11_0_arm64.whl", hash = "sha256:40d936d72c9bdc10df43f93a8be502bc5024b487259139f66a328722c07f34a9", size = 609038, upload-time = "2026-09-23T20:23:24.941Z" }, + { url = "https://files.pythonhosted.org/packages/0b/71/f57effdf895c2d6d49333f1d6833a9f5438725c55a04ed3124a1c95bb3aa/wcwidth-0.9.1-cp310-abi3-manylinux1_i686.manylinux2014_i686.manylinux_2_17_i686.manylinux_2_5_i686.whl", hash = "sha256:69bb970cf5652b88cfdb9d3fdd1764fc15e5f8ad531643e7bb3e894cd969740e", size = 761464, upload-time = "2026-09-23T20:23:32.425Z" }, + { url = "https://files.pythonhosted.org/packages/4f/16/15b03d8fb7a747d0d2cf4a33da7b1b3e092493c58780f6589d5b1e9e3b49/wcwidth-0.9.1-cp310-abi3-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:b5da43d6967668982e44a52fb551967d293f86d26cd86036ac95bbdd34394ed9", size = 770483, upload-time = "2026-09-23T20:23:28.128Z" }, + { url = "https://files.pythonhosted.org/packages/e2/a0/834886b30e3b885a5a96e8358f1b3917bc0bac43ae4178d77f0c210d079b/wcwidth-0.9.1-cp310-abi3-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:747fb724223f417a17541a95a17c1dac3a8ef9a0cf41684950f0eab191a35f65", size = 768443, upload-time = "2026-09-23T20:23:33.875Z" }, + { url = "https://files.pythonhosted.org/packages/a2/ae/1b7597e0132ebc2c920632f39d8c1c61881345c9b480c0a95eb91e7460fa/wcwidth-0.9.1-cp310-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:6e1272b7986cefe79783737e38bdb9eaae0682b333c7bd132024441193dd5ce7", size = 781171, upload-time = "2026-09-23T20:23:29.46Z" }, + { url = "https://files.pythonhosted.org/packages/fb/bb/32149e3b481953f4ef35f38f166e2532886d33ae54ccdd71393f821dfad9/wcwidth-0.9.1-cp310-abi3-musllinux_1_2_i686.whl", hash = "sha256:fe021c4d8de9d36c31a0cb41d0d2546dadd3b0708a301f1a0c66ce200851831f", size = 781576, upload-time = "2026-09-23T20:23:35.504Z" }, + { url = "https://files.pythonhosted.org/packages/d6/72/1ee3dff67697bda93bc56ec9f7b3b89609cd64b9abb328512a2f22eb9e47/wcwidth-0.9.1-cp310-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:708158c082364af442f9983de7b6ec9ac0d2e1b825ada25f0911b1f138d55405", size = 783526, upload-time = "2026-09-23T20:23:36.937Z" }, + { url = "https://files.pythonhosted.org/packages/ce/0a/e94f19a60f6127bcf3dfcf223ca05a37650d57b37b5ed623597c842ce7ca/wcwidth-0.9.1-cp310-abi3-win32.whl", hash = "sha256:356376852357b8fca71fe5415808ec421679e04b4a98eb7c9cb6a7984b911a05", size = 592136, upload-time = "2026-09-23T20:23:43.774Z" }, + { url = "https://files.pythonhosted.org/packages/3c/ff/d884d2ec7dcdc86cda742a86c7160bc726dbccae76d7ae1f9ac833c26680/wcwidth-0.9.1-cp310-abi3-win_amd64.whl", hash = "sha256:991d1c8834f548e9c1f16432075ee84638e122312556bbf1ed595ea8fffc4673", size = 596738, upload-time = "2026-09-23T20:23:44.989Z" }, + { url = "https://files.pythonhosted.org/packages/fa/93/16e30d617b937272a2310a6e3c1f9d8f94984e19ad4a96eb935628e3e6b6/wcwidth-0.9.1-cp310-abi3-win_arm64.whl", hash = "sha256:61bd7aef9cafb6cb77a37a169998d7928ce82a51522146d11f60db9e7d1cb43a", size = 594700, upload-time = "2026-09-23T20:23:41.013Z" }, + { url = "https://files.pythonhosted.org/packages/0b/fb/bf93868205814f03d63e5b91757cf5bee99e024ce951c2e91fdeaab49c59/wcwidth-0.9.1-cp314-cp314t-macosx_10_15_x86_64.whl", hash = "sha256:dc10e262c3ac0abbfd0a2a51e45a848b1b7f500b21ff512630277973ba25674d", size = 605391, upload-time = "2026-09-23T20:23:23.602Z" }, + { url = "https://files.pythonhosted.org/packages/59/bb/dd3ae682f8ab13b1bffd8d52320d2f7becc16e3200580015f52124636e3a/wcwidth-0.9.1-cp314-cp314t-macosx_11_0_arm64.whl", hash = "sha256:bcb9ed4a367cc025bf1092ac679a156759605182d60b7ed3c28f7221a42ddf55", size = 609851, upload-time = "2026-09-23T20:23:26.583Z" }, + { url = "https://files.pythonhosted.org/packages/56/e7/4f3a86953ab1f59eac15585795f7753da99370798f75b134d3cbf252b243/wcwidth-0.9.1-cp314-cp314t-manylinux1_i686.manylinux2014_i686.manylinux_2_17_i686.manylinux_2_5_i686.whl", hash = "sha256:0d68a30d504c68cfdff2a5f804675c1e7ab4c0bbe878024c8b680ec5579cde67", size = 774751, upload-time = "2026-09-23T20:23:38.473Z" }, + { url = "https://files.pythonhosted.org/packages/48/0c/a456332bb581ebf3eb1a7dedb1492747b15772ce82c0717cc05387a64a17/wcwidth-0.9.1-cp314-cp314t-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:9f1636c5075ffd5c2e835b4561874f6e5dd2bbeba3c6c2c99f067d0d16883af8", size = 785798, upload-time = "2026-09-23T20:23:30.998Z" }, + { url = "https://files.pythonhosted.org/packages/1b/6e/a35de089370d6d872c5424d234e00a8580e62c1bff518ff583750e7b15f7/wcwidth-0.9.1-cp314-cp314t-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:c4cead196551112cb8f43cdd1f80c235ef2456e34b1a9955c424537ec99961b2", size = 782933, upload-time = "2026-09-23T20:23:39.672Z" }, + { url = "https://files.pythonhosted.org/packages/ed/25/e39b6140d05f3aa8e2a6cbb2d679cba5767de7ac0a4860d1da246b6c36b3/wcwidth-0.9.1-cp314-cp314t-win32.whl", hash = "sha256:0665ee822ea04e25801e6e82e5407528be0863e88a17b0e7ca038843a4a3ac0c", size = 599829, upload-time = "2026-09-23T20:23:46.221Z" }, + { url = "https://files.pythonhosted.org/packages/12/ad/307b55d8e0a9484c48e8f78f40e9b9a4a9af0c9e2e1c5209ad6d5a09c77c/wcwidth-0.9.1-cp314-cp314t-win_amd64.whl", hash = "sha256:eab587e18e7cadf1a750b0098fc8bebfb62c125eb9306f2268f0443a282a3d78", size = 604398, upload-time = "2026-09-23T20:23:47.672Z" }, + { url = "https://files.pythonhosted.org/packages/36/f1/69566660f633e2a93df7dad7353761562e9f71bf8a5a7b1eb7f34793b45d/wcwidth-0.9.1-cp314-cp314t-win_arm64.whl", hash = "sha256:03cfca3dcbffa86564290fe3c9978a6191ba003e8ced7f7dbda315fcb3fbe725", size = 602291, upload-time = "2026-09-23T20:23:42.345Z" }, + { url = "https://files.pythonhosted.org/packages/96/a2/f06f2e0be4895e5943d358755b51968ace85774cf0a4cc42b4dafb793832/wcwidth-0.9.1-py3-none-any.whl", hash = "sha256:e0c3a1c45c5b9550c6919a4449e95f5b177f6e165786376981db8f1addae9b21", size = 300656, upload-time = "2026-09-23T20:23:20.674Z" }, +]