diff --git a/benchmarks/__main__.py b/benchmarks/__main__.py index c019ae9..7c3ab5e 100644 --- a/benchmarks/__main__.py +++ b/benchmarks/__main__.py @@ -11,8 +11,9 @@ import os import pathlib import sys +import typing -from benchmarks.config import DEFAULT_DSN, ExtendConfig, RunConfig +from benchmarks.config import DEFAULT_DSN, ExtendConfig, RunConfig, VacuumVariant from benchmarks.extend import ExtendResult, format_extend, run_extend, to_json from benchmarks.report import compare, format_markdown, format_table, to_baseline from benchmarks.workload import RunResult, make_engine, run_consumer, run_producer @@ -126,17 +127,35 @@ async def _run_extend_sweep(dsn: str, points: list[ExtendConfig]) -> list[Extend results = [] for cfg in points: result = await run_extend(dsn, cfg) - sys.stdout.write(f"{cfg.name}: {result.committed} pings in {result.wall_seconds:.1f}s\n") + sys.stdout.write(f"{cfg.name}/{cfg.vacuum}: {result.committed} pings in {result.wall_seconds:.1f}s\n") sys.stdout.flush() results.append(result) return results -def _extend(dsn: str, *, seconds: float | None, only: list[str] | None, json_path: str | None) -> int: - """Run the extend sweep and print its tables; nothing is gated.""" - points = [p for p in _EXTEND_POINTS if not only or p.name in only] +def _extend_points( + *, + only: list[str] | None, + seconds: float | None, + vacuum: list[VacuumVariant] | None, + sample_interval: float | None, +) -> list[ExtendConfig]: + """Return the selected points, each crossed with every requested vacuum variant.""" + overrides: dict[str, float] = {} if seconds is not None: - points = [dataclasses.replace(p, seconds=seconds) for p in points] + overrides["seconds"] = seconds + if sample_interval is not None: + overrides["sample_interval"] = sample_interval + return [ + dataclasses.replace(p, **overrides, vacuum=variant) + for p in _EXTEND_POINTS + if not only or p.name in only + for variant in (vacuum or [p.vacuum]) + ] + + +def _extend(dsn: str, *, points: list[ExtendConfig], json_path: str | None) -> int: + """Run the extend sweep and print its tables; nothing is gated.""" results = asyncio.run(_run_extend_sweep(dsn, points)) sys.stdout.write("\n" + format_extend(results) + "\n") if json_path is not None: @@ -168,6 +187,8 @@ def _build_parser() -> argparse.ArgumentParser: extend.add_argument("--seconds", type=float) extend.add_argument("--only", nargs="+", choices=[p.name for p in _EXTEND_POINTS]) extend.add_argument("--json") + extend.add_argument("--vacuum", nargs="+", choices=typing.get_args(VacuumVariant)) + extend.add_argument("--sample-interval", type=float) return parser @@ -179,7 +200,10 @@ def main(argv: list[str] | None = None) -> int: if args.command == "check": return _check(dsn, markdown=args.markdown) if args.command == "extend": - return _extend(dsn, seconds=args.seconds, only=args.only, json_path=args.json) + points = _extend_points( + only=args.only, seconds=args.seconds, vacuum=args.vacuum, sample_interval=args.sample_interval + ) + return _extend(dsn, points=points, json_path=args.json) return _run(dsn, args.messages, args.repeats, write_baseline=args.write_baseline) diff --git a/benchmarks/config.py b/benchmarks/config.py index 9fa7bf8..727b0d9 100644 --- a/benchmarks/config.py +++ b/benchmarks/config.py @@ -1,6 +1,7 @@ """Workload configuration for a single benchmark run.""" import dataclasses +import typing # Matches the payload size in Tsvettsikh's «50 оттенков Transactional Outbox» @@ -10,6 +11,10 @@ # The DSN the harness runs against. Overridden by POSTGRES_DSN in compose. DEFAULT_DSN = "postgresql+asyncpg://outbox:outbox@localhost:5432/outbox" +# How an extend point's table is vacuumed: no reloptions, outbox_autovacuum_ddl's defaults, +# or those defaults with autovacuum's cost throttle turned off. +VacuumVariant = typing.Literal["untuned", "tuned", "unthrottled"] + # Tags the harness' own connections so the probe can assert it owns the database. APPLICATION_NAME = "faststream-outbox-bench" @@ -49,8 +54,17 @@ class ExtendConfig: pool_size: int = 20 max_fetch_interval: float = 1.0 queue: str = "timeouts" + vacuum: VacuumVariant = "tuned" + # Seconds between table-health samples taken while the load runs. + sample_interval: float = 10.0 def __post_init__(self) -> None: + if self.vacuum not in typing.get_args(VacuumVariant): + msg = f"{self.name}: vacuum must be one of {typing.get_args(VacuumVariant)}, got {self.vacuum!r}" + raise ValueError(msg) + if self.sample_interval <= 0: + msg = f"{self.name}: sample_interval must be positive, got {self.sample_interval}" + raise ValueError(msg) if self.ping_interval >= self.timeout_seconds: msg = f"{self.name}: each identity pings every {self.ping_interval:.2f}s, not before its timeout" raise ValueError(msg) diff --git a/benchmarks/extend.py b/benchmarks/extend.py index 70cf005..dd37455 100644 --- a/benchmarks/extend.py +++ b/benchmarks/extend.py @@ -10,9 +10,12 @@ task on a pool of ``pool_size`` connections, so pool waits show up once Postgres falls behind. A point stops offering load once the backlog passes five seconds of pings and is reported as saturated. -The table carries the recommended autovacuum settings, unlike the gated sweep's tables, because dead -tuples and their cleanup are part of what this measures. Statement, tuple and WAL counters come from -the same probe as the gated sweep. Nothing here is gated: every number depends on timing. +The table's autovacuum settings follow the point's ``vacuum`` variant, ``outbox_autovacuum_ddl``'s +defaults unless it says otherwise, because dead tuples and their cleanup are part of what this measures. +Every ``sample_interval`` seconds a separate connection records table and index size, dead tuples, +autovacuum runs and the fetch statement's cumulative counters, so a long run shows whether they settle. +Statement, tuple and WAL counters come from the same probe as the gated sweep. Nothing here is gated: +every number depends on timing. """ import asyncio @@ -50,6 +53,24 @@ """, ) +# Mentions pg_stat, so the probe's NOT ILIKE '%pg_stat%' filter drops it from the statement counters. +_SAMPLE_SQL = text( + """ + SELECT pg_relation_size(t.relid) AS table_bytes, + pg_indexes_size(t.relid) AS index_bytes, + t.n_dead_tup, t.n_live_tup, t.autovacuum_count, + coalesce(f.calls, 0) AS fetch_calls, + coalesce(f.total_ms, 0) AS fetch_ms + FROM pg_stat_user_tables t, + LATERAL ( + SELECT sum(s.calls) AS calls, sum(s.total_exec_time) AS total_ms + FROM pg_stat_statements s + WHERE s.query NOT ILIKE '%pg_stat%' AND lower(split_part(btrim(s.query), ' ', 1)) = 'with' + ) f + WHERE t.relname = :table AND t.schemaname = current_schema() + """, +) + _VACUUM_SQL = text( """ SELECT t.autovacuum_count, t.n_live_tup @@ -84,6 +105,20 @@ class _Tally: silent_fired: asyncio.Event = dataclasses.field(default_factory=asyncio.Event) +@dataclasses.dataclass(frozen=True) +class Sample: + """Table health *elapsed* seconds after sampling began; the fetch counters are cumulative.""" + + elapsed: float + table_bytes: int + index_bytes: int + dead_tuples: int + live_tuples: int + autovacuum_count: int + fetch_calls: int + fetch_ms: float + + @dataclasses.dataclass(frozen=True) class ExtendResult: """One extend point: client-side timings, expiry outcomes and the probe's catalog deltas.""" @@ -106,6 +141,7 @@ class ExtendResult: autovacuum_count: int live_tuples: int probe: ProbeResult + samples: tuple[Sample, ...] = () def percentiles(values: typing.Sequence[float]) -> dict[str, float]: @@ -120,6 +156,27 @@ def _rank(q: float) -> float: return {"p50": _rank(0.50), "p95": _rank(0.95), "p99": _rank(0.99), "max": ordered[-1]} +def autovacuum_ddl(cfg: ExtendConfig, table_name: str) -> str | None: + """Return the ``ALTER TABLE`` that sets up *cfg*'s vacuum variant, or ``None`` to leave the table untuned.""" + if cfg.vacuum == "untuned": + return None + return outbox_autovacuum_ddl(table_name, vacuum_cost_delay=0 if cfg.vacuum == "unthrottled" else None) + + +def _sample_at(samples: typing.Sequence[Sample], elapsed: float) -> Sample: + """Return the last sample taken at or before *elapsed*, or the first one.""" + return next((s for s in reversed(samples) if s.elapsed <= elapsed), samples[0]) + + +def fetch_mean_ms(samples: typing.Sequence[Sample], start: float, end: float) -> float: + """Mean fetch statement time between the samples at *start* and *end*; 0 when nothing ran.""" + if not samples: + return 0.0 + first, last = _sample_at(samples, start), _sample_at(samples, end) + calls = last.fetch_calls - first.fetch_calls + return (last.fetch_ms - first.fetch_ms) / calls if calls else 0.0 + + def _body(identity: int, deadline: dt.datetime) -> dict[str, typing.Any]: return {"identity": identity, "deadline": deadline.isoformat()} @@ -129,7 +186,8 @@ async def _create_table(engine: AsyncEngine, cfg: ExtendConfig) -> tuple[MetaDat table = make_outbox_table(metadata, table_name=f"bench_extend_{uuid.uuid4().hex[:10]}") async with engine.begin() as conn: await conn.run_sync(metadata.create_all) - await conn.execute(text(outbox_autovacuum_ddl(table.name))) + if (ddl := autovacuum_ddl(cfg, table.name)) is not None: + await conn.execute(text(ddl)) start = utcnow() rows = [] for identity in range(cfg.timers): @@ -227,25 +285,52 @@ def _spawn() -> None: return started +async def _sample_until(stop: asyncio.Event, engine: AsyncEngine, table: Table, interval: float) -> list[Sample]: + """Sample table health every *interval* seconds, plus once more after *stop* is set.""" + samples: list[Sample] = [] + started = time.perf_counter() + async with _autocommit(engine) as conn: + while True: + row = (await conn.execute(_SAMPLE_SQL, {"table": table.name})).one() + samples.append( + Sample( + elapsed=time.perf_counter() - started, + table_bytes=int(row.table_bytes), + index_bytes=int(row.index_bytes), + dead_tuples=int(row.n_dead_tup), + live_tuples=int(row.n_live_tup), + autovacuum_count=int(row.autovacuum_count), + fetch_calls=int(row.fetch_calls), + fetch_ms=float(row.fetch_ms), + ), + ) + if stop.is_set(): + return samples + with contextlib.suppress(TimeoutError): + await asyncio.wait_for(stop.wait(), timeout=interval) + + async def run_extend(dsn: str, cfg: ExtendConfig) -> ExtendResult: """Run one extend point against a fresh table and return its measurements.""" engine = make_engine(dsn, pool_size=cfg.pool_size) subscriber_engine = make_engine(dsn) + sampler_engine = make_engine(dsn, pool_size=1) try: await ensure_extension(engine) metadata, table = await _create_table(engine, cfg) try: - return await _measure(engine, subscriber_engine, table, cfg) + return await _measure(engine, subscriber_engine, sampler_engine, table, cfg) finally: async with engine.begin() as conn: await conn.run_sync(metadata.drop_all) finally: + await sampler_engine.dispose() await subscriber_engine.dispose() await engine.dispose() async def _measure( - engine: AsyncEngine, subscriber_engine: AsyncEngine, table: Table, cfg: ExtendConfig + engine: AsyncEngine, subscriber_engine: AsyncEngine, sampler_engine: AsyncEngine, table: Table, cfg: ExtendConfig ) -> ExtendResult: tally = _Tally() publisher = OutboxBroker(outbox_table=table, logger=None) @@ -263,12 +348,16 @@ async def _expire(body: dict[str, typing.Any]) -> None: await _checkpoint(engine) await assert_owns_database(engine) + stop_sampling = asyncio.Event() async with probe(engine, table.name) as sink: await expirer.start() + sampler = asyncio.create_task(_sample_until(stop_sampling, sampler_engine, table, cfg.sample_interval)) try: started = await _offer_load(engine=engine, publisher=publisher, cfg=cfg, tally=tally) await _wait_for_silent_expiries(cfg, tally, started) finally: + stop_sampling.set() + samples = await sampler await expirer.stop() await subscriber_engine.dispose() wall = (tally.last_commit or time.perf_counter()) - started @@ -296,6 +385,7 @@ async def _expire(body: dict[str, typing.Any]) -> None: autovacuum_count=int(vacuum.autovacuum_count), live_tuples=int(vacuum.n_live_tup), probe=sink[0], + samples=tuple(samples), ) @@ -306,8 +396,27 @@ async def _wait_for_silent_expiries(cfg: ExtendConfig, tally: _Tally, started: f await asyncio.wait_for(tally.silent_fired.wait(), timeout=max(0.0, give_up - time.perf_counter())) +def _label(cfg: ExtendConfig) -> str: + return f"{cfg.name}/{cfg.vacuum}" + + +def _drift_row(r: ExtendResult) -> str: + samples = r.samples + first, last = samples[0], samples[-1] + mb = 2**20 + table_peak = max(s.table_bytes for s in samples) + index_peak = max(s.index_bytes for s in samples) + early = fetch_mean_ms(samples, first.elapsed, first.elapsed + 60.0) + late = fetch_mean_ms(samples, last.elapsed - 60.0, last.elapsed) + return ( + f"| {_label(r.config)} | {first.table_bytes / mb:.1f} / {last.table_bytes / mb:.1f} / {table_peak / mb:.1f} " + f"| {first.index_bytes / mb:.1f} / {last.index_bytes / mb:.1f} / {index_peak / mb:.1f} " + f"| {max(s.dead_tuples for s in samples):,} | {last.autovacuum_count} | {early:.3f} / {late:.3f} |" + ) + + def format_extend(results: typing.Sequence[ExtendResult]) -> str: - """Render the extend points as two markdown tables: the write path, then table and expiry health.""" + """Render the extend points as markdown tables: the write path, table and expiry health, then drift.""" write = [ ( "| point | offered/s | committed/s | commits/s | ping p50 / p95 / p99 ms | pool wait p99 ms " @@ -329,7 +438,7 @@ def format_extend(results: typing.Sequence[ExtendResult]) -> str: insert_ms = r.statement_ms.get("insert", (0.0, 0.0)) fetch_ms = r.statement_ms.get("with", (0.0, 0.0)) errors = ", ".join(f"{k} {v}" for k, v in sorted(r.errors.items())) or "0" - label = f"{cfg.name}{' (saturated)' if r.saturated else ''}" + label = f"{_label(cfg)}{' (saturated)' if r.saturated else ''}" offered_rate = r.offered / max(r.offered_seconds, 1e-9) write.append( f"| {label} | {offered_rate:,.0f} | {r.committed / span:,.0f} | {r.transactions / span:,.0f} " @@ -338,12 +447,24 @@ def format_extend(results: typing.Sequence[ExtendResult]) -> str: ) hot = 100.0 * p.tup_hot_upd / p.tup_upd if p.tup_upd else 0.0 health.append( - f"| {cfg.name} | {p.wal_bytes / pings:,.0f} | {p.wal_records / pings:.2f} | {p.wal_fpi:,} | {hot:.0f} " + f"| {_label(cfg)} | {p.wal_bytes / pings:,.0f} | {p.wal_records / pings:.2f} | {p.wal_fpi:,} | {hot:.0f} " f"| {p.dead_tup:,} | {r.autovacuum_count} | {p.index_bytes / 2**20:.1f} " f"| {fetch_ms[0]:.3f} / {fetch_ms[1]:.1f} | {r.expired_silent} / {cfg.silent_timers} " f"| {r.false_expiries} | {r.lateness['p50']:.2f} / {r.lateness['max']:.2f} | {r.leased} |", ) - return "\n".join([*write, "", *health]) + tables = [*write, "", *health] + sampled = [r for r in results if r.samples] + if sampled: + tables += [ + "", + ( + "| point | table MB start / end / peak | index MB start / end / peak | dead tuples peak " + "| autovacuums | fetch mean ms first / last minute |" + ), + "|---|---|---|---|---|---|", + *(_drift_row(r) for r in sampled), + ] + return "\n".join(tables) def to_json(results: typing.Sequence[ExtendResult]) -> str: diff --git a/tests/test_benchmarks.py b/tests/test_benchmarks.py index 1125e52..c720dc5 100644 --- a/tests/test_benchmarks.py +++ b/tests/test_benchmarks.py @@ -9,9 +9,9 @@ import pytest -from benchmarks.__main__ import _build_parser +from benchmarks.__main__ import _build_parser, _extend_points from benchmarks.config import ExtendConfig, RunConfig -from benchmarks.extend import ExtendResult, format_extend, percentiles +from benchmarks.extend import ExtendResult, Sample, autovacuum_ddl, fetch_mean_ms, format_extend, percentiles from benchmarks.probes import ProbeResult from benchmarks.report import EXACT_KEYS, compare, format_markdown, format_table, normalize, to_baseline from benchmarks.workload import RunResult @@ -360,9 +360,21 @@ def test_extend_config_rejects_a_run_too_short_for_silent_timers_to_fire() -> No ExtendConfig(name="p", rate=1_000, timers=2_000, seconds=10.0) -def test_format_extend_reports_saturation_and_per_ping_wal() -> None: - result = ExtendResult( - config=ExtendConfig(name="r1000-b1", rate=1_000, timers=2_000), +def _sample(elapsed: float, *, table_mb: float = 1.0, index_mb: float = 1.0, dead: int = 0, **kw: float) -> Sample: + fields: dict[str, float] = {"live_tuples": 2_000, "autovacuum_count": 0, "fetch_calls": 0, "fetch_ms": 0.0} + fields.update(kw) + return Sample( + elapsed=elapsed, + table_bytes=int(table_mb * 2**20), + index_bytes=int(index_mb * 2**20), + dead_tuples=dead, + **fields, + ) + + +def _extend_result(*, samples: tuple[Sample, ...] = (), **cfg_kwargs: object) -> ExtendResult: + return ExtendResult( + config=ExtendConfig(name="r1000-b1", rate=1_000, timers=2_000, **cfg_kwargs), # ty: ignore[invalid-argument-type] wall_seconds=30.0, offered=30_000, offered_seconds=30.0, @@ -380,14 +392,106 @@ def test_format_extend_reports_saturation_and_per_ping_wal() -> None: autovacuum_count=1, live_tuples=2_000, probe=_probe(insert_calls=30_000, wal_bytes=18_000_000, wal_records=199_000, tup_upd=30_000, tup_hot_upd=0), + samples=samples, ) - table = format_extend([result]) - assert "| r1000-b1 (saturated) | 1,000 | 1,000 | 1,000 | 2.0 / 4.0 / 17.0 |" in table + + +def test_format_extend_reports_saturation_and_per_ping_wal() -> None: + table = format_extend([_extend_result()]) + assert "| r1000-b1/tuned (saturated) | 1,000 | 1,000 | 1,000 | 2.0 / 4.0 / 17.0 |" in table assert "TimeoutError 5" in table - assert "| r1000-b1 | 600 | 6.63 |" in table + assert "| r1000-b1/tuned | 600 | 6.63 |" in table assert "| 100 / 100 | 0 | 0.98 / 1.02 | 2 |" in table def test_extend_accepts_its_flags() -> None: args = _build_parser().parse_args(["extend", "--seconds", "12", "--only", "r1000-b1", "--json", "out.json"]) assert (args.seconds, args.only, args.json) == (12.0, ["r1000-b1"], "out.json") + + +def test_extend_config_rejects_an_unknown_vacuum_variant() -> None: + with pytest.raises(ValueError, match="vacuum"): + ExtendConfig(name="p", rate=1_000, timers=2_000, vacuum="off") # ty: ignore[invalid-argument-type] + + +def test_extend_config_rejects_a_non_positive_sample_interval() -> None: + with pytest.raises(ValueError, match="sample_interval"): + ExtendConfig(name="p", rate=1_000, timers=2_000, sample_interval=0.0) + + +@pytest.mark.parametrize( + ("variant", "expected"), + [ + ("untuned", None), + ( + "tuned", + ( + "ALTER TABLE t SET (autovacuum_vacuum_scale_factor = 0, autovacuum_vacuum_threshold = 1000, " + "autovacuum_vacuum_insert_scale_factor = 0, autovacuum_vacuum_insert_threshold = 1000)" + ), + ), + ( + "unthrottled", + ( + "ALTER TABLE t SET (autovacuum_vacuum_scale_factor = 0, autovacuum_vacuum_threshold = 1000, " + "autovacuum_vacuum_insert_scale_factor = 0, autovacuum_vacuum_insert_threshold = 1000, " + "autovacuum_vacuum_cost_delay = 0)" + ), + ), + ], +) +def test_autovacuum_ddl_per_variant(variant: str, expected: str | None) -> None: + cfg = ExtendConfig(name="p", rate=1_000, timers=2_000, vacuum=variant) # ty: ignore[invalid-argument-type] + assert autovacuum_ddl(cfg, "t") == expected + + +def test_fetch_mean_ms_divides_the_window_deltas() -> None: + samples = [ + _sample(0.0, fetch_calls=0, fetch_ms=0.0), + _sample(60.0, fetch_calls=100, fetch_ms=20.0), + _sample(120.0, fetch_calls=300, fetch_ms=120.0), + ] + assert fetch_mean_ms(samples, 0.0, 60.0) == pytest.approx(0.2) + assert fetch_mean_ms(samples, 60.0, 120.0) == pytest.approx(0.5) + + +def test_fetch_mean_ms_is_zero_without_fetches_in_the_window() -> None: + assert fetch_mean_ms([_sample(0.0), _sample(60.0)], 0.0, 60.0) == 0.0 + assert fetch_mean_ms([], 0.0, 60.0) == 0.0 + + +def test_format_extend_reports_drift_over_the_run() -> None: + samples = ( + _sample(0.0, table_mb=1.0, index_mb=0.5, dead=0, fetch_calls=0, fetch_ms=0.0), + _sample(60.0, table_mb=3.0, index_mb=2.0, dead=50_000, autovacuum_count=1, fetch_calls=100, fetch_ms=25.0), + _sample(120.0, table_mb=2.5, index_mb=2.5, dead=10_000, autovacuum_count=2, fetch_calls=200, fetch_ms=125.0), + ) + table = format_extend([_extend_result(vacuum="unthrottled", samples=samples)]) + assert "| r1000-b1/unthrottled | 1.0 / 2.5 / 3.0 | 0.5 / 2.5 / 2.5 | 50,000 | 2 | 0.250 / 1.000 |" in table + + +def test_format_extend_skips_drift_without_samples() -> None: + assert "table MB" not in format_extend([_extend_result()]) + + +def test_extend_accepts_vacuum_variants_and_sample_interval() -> None: + args = _build_parser().parse_args(["extend", "--vacuum", "untuned", "unthrottled", "--sample-interval", "5"]) + assert (args.vacuum, args.sample_interval) == (["untuned", "unthrottled"], 5.0) + + +def test_extend_points_cross_points_with_vacuum_variants() -> None: + points = _extend_points( + only=["r1000-b1", "r3000-b1"], seconds=900.0, vacuum=["untuned", "tuned"], sample_interval=5.0 + ) + assert [(p.name, p.vacuum, p.seconds, p.sample_interval) for p in points] == [ + ("r1000-b1", "untuned", 900.0, 5.0), + ("r1000-b1", "tuned", 900.0, 5.0), + ("r3000-b1", "untuned", 900.0, 5.0), + ("r3000-b1", "tuned", 900.0, 5.0), + ] + + +def test_extend_points_default_to_every_point_tuned() -> None: + points = _extend_points(only=None, seconds=None, vacuum=None, sample_interval=None) + assert {p.vacuum for p in points} == {"tuned"} + assert len(points) == 8