From dd45b02709072b03950853e5e9e10dcb152517d9 Mon Sep 17 00:00:00 2001 From: mhenrixon Date: Sun, 5 Jul 2026 23:37:52 +0200 Subject: [PATCH 1/2] feat(bench): execution-mode connection benchmark + DB-pool sizing guidance (#323) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary The "measure first" gate of #323 (elastic pools): a fair, reusable harness that answers "for the same offered concurrency, how many real Postgres connections does threads mode vs async mode actually hold?" — so we can guide users on DB-pool sizing per execution_mode before building any elasticity. The multi-agent design DISPROVED the naive "async saves connections" premise with code evidence: pgbus holds a pgmq connection only for the read/archive SQL round-trip, NOT the job body (executor runs perform_now with zero pgmq conns). So async's connection-density win is real only for I/O-light work; for DB-bound work async is connection-bound like threads. ## What ships - benchmarks/support/execution_mode_harness.rb — reusable, require-able harness. run_cell(mode, pool_size, concurrency, io_profile, job_count, client) -> Result. Fair by construction: identical offered load per pair (assert_fair_pair!), bounded in-flight window = pool capacity (both modes run the same concurrency), real pool checkout (client.pgmq.with_connection), peak_busy = size - available. - benchmarks/execution_modes_bench.rb — 8-cell matrix, rake bench:execution_modes. - docs/performance.md "Execution mode and DB pool sizing" — the guidance, with measured numbers and a sizing table. ## The numbers (measured, local Postgres, 240 jobs @ concurrency 12) async pool=3 io_light : peak_busy=3, ~222 job/s <- 12 fibers on 3 conns, full throughput threads pool=12 io_light: peak_busy=12, ~216 job/s async pool=3 db_bound : peak_busy=3, ~95 job/s <- under-provisioned: ~3.4x throughput collapse async pool=12 db_bound : peak_busy=12, ~324 job/s Key honest finding: NO cell hit pool_timeouts. pool_timeout (5s) dwarfs a 10-30ms checkout, so under-provisioning SERIALISES (throughput collapse), it doesn't error. The guidance says to watch throughput/latency, not error counts. ## Test Coverage - spec/pgbus/execution_mode_harness_spec.rb (16 unit, no DB, runs in CI): Result struct + derivations, IoProfile, summarize percentiles, assert_fair_pair! (anti-rigging), classify_error! tally (never swallow), assert_dedicated_path!, async_yield mode-awareness, PoolSampler busy=size-available + degraded-skip. - spec/integration/execution_mode_harness_spec.rb (3 DB-gated, :integration, auto-skip without PGBUS_DATABASE_URL): run_cell drives real jobs, bigger pool holds more connections (fair pair), async fibers share a small pool on io_light. ## Verification - [x] bundle exec rubocop passes - [x] bundle exec rspec spec/pgbus — 3036 examples, 0 failures - [x] DB-gated specs pass against real Postgres - [x] docs rake lint clean - [x] rake bench:execution_modes runs clean; numbers captured in docs Refs #323 --- CHANGELOG.md | 1 + Rakefile | 8 +- benchmarks/connection_pool_bench.rb | 8 + benchmarks/execution_modes_bench.rb | 163 ++++++++++ benchmarks/support/execution_mode_harness.rb | 293 ++++++++++++++++++ docs/performance.md | 56 ++++ .../execution_mode_harness_spec.rb | 90 ++++++ spec/pgbus/execution_mode_harness_spec.rb | 192 ++++++++++++ 8 files changed, 810 insertions(+), 1 deletion(-) create mode 100644 benchmarks/execution_modes_bench.rb create mode 100644 benchmarks/support/execution_mode_harness.rb create mode 100644 spec/integration/execution_mode_harness_spec.rb create mode 100644 spec/pgbus/execution_mode_harness_spec.rb diff --git a/CHANGELOG.md b/CHANGELOG.md index 8cc086bf..3ad3c8cc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,7 @@ ### Added +- **Execution-mode connection benchmark + DB-pool sizing guidance (`rake bench:execution_modes`).** A fair, reusable harness (`benchmarks/support/execution_mode_harness.rb` + `benchmarks/execution_modes_bench.rb`) that runs the *same* offered load through the threads and async execution pools and measures how many real Postgres connections each actually holds (`peak_busy = pool size − available`, sampled through the real pool checkout), so operators can size the DB pool per `execution_mode`. It settles the "async saves connections" question with data: a job holds a pgmq connection only for the `read_batch` + `archive` round-trip (not the job body), so **async's connection-density win is real only for I/O-light work** (a few connections serve many fibers) — for **DB-bound work async is connection-bound like threads**, and a too-small pool *serialises* (throughput collapse, ~3.4× in the measured cell) rather than erroring, because `pool_timeout` (5 s) dwarfs a typical checkout. New `docs/performance.md` section "Execution mode and DB pool sizing" carries the measured numbers and a sizing table. The harness's fairness/struct/measurement logic is unit-tested with fakes (no DB, runs in CI); the numbers come from a manual `PGBUS_DATABASE_URL` run (run-and-report, never a CI gate). Groundwork for the elastic-pools research in #323; a `pgbus doctor` mode/pool-mismatch hint is deferred to a follow-up. Refs #323. - **Off-thread durable-stream fanout writer (`config.streams_writer_threads`, `config.streams_writer_buffer_limit`) — the full head-of-line cure the #315 deadline only bounded.** The #315 deadline caps each slow-client fanout stall at `streams_fanout_write_deadline_ms` (250ms), but the single dispatcher thread still writes to every client serially, so K slow clients still cost ~`K × 250ms`. Setting `streams_writer_threads > 0` moves the **durable** fanout socket write off the dispatcher into N writer threads (each connection pinned to one worker by `id.hash % N`, preserving per-connection frame order and its per-io mutex). The dispatcher hands off the write and moves on; the pump reports back the highest msg_id it actually committed via an ack queue, and the dispatcher — still the **sole owner** of the read cursor — advances it only on that ack, so a blocked or partial write never advances the cursor past a frame that didn't reach the socket. A failed write posts a `DisconnectMessage` so the dispatcher scrubs the connection deterministically on its own thread (even on an otherwise-quiet stream). Measured (`streams_writer_offload_bench.rb`, dispatcher `handle_durable_wake` wall time, 50 fast + K slow clients @ 50ms): inline scales linearly (K=20 → **1071 ms**) while offload stays flat (K=20 → **0.09 ms**) — fast clients no longer wait behind slow ones. This is a **system-level latency win, not a per-write speedup** (total bytes written are unchanged); the dispatcher does slightly more per-wake bookkeeping under offload, which is why it is **opt-in and default-off** (`streams_writer_threads = 0` = byte-for-byte the pre-#321 inline behavior). **Ephemeral frames always stay inline** regardless of this setting — they have no archive to replay, so they must not risk an async drop (the pump raises if one is ever routed to it; issue #321 B1). `streams_writer_buffer_limit` (default `0` = unbounded) caps a connection's outbound buffer, dropping its oldest durable frame on overflow — safe because durable frames are re-read from the archive on reconnect; it's an OOM guard for a pathologically slow-but-alive client, not a delivery guarantee. Both knobs are dispatcher-internal tuning (not part of the 1.0 public-config surface). Refs #321 (follow-up to #315). - **Bounded dispatcher fanout head-of-line blocking (`config.streams_fanout_write_deadline_ms`, `config.streams_dispatch_queue_limit`).** One dispatcher thread per web-server process fans out every broadcast to every connected SSE client **serially**, writing to each socket in turn. A slow client's write blocked that thread up to the full `streams_write_deadline_ms` (5s) before the client was marked dead — and every connection queued behind it in the same wake waited, so K slow clients could stack a `K × 5s` stall on all other browsers on the worker. Fanout writes now use a separate, shorter `streams_fanout_write_deadline_ms` (default **250 ms**), bounding the worst-case serial stall to `K × 250 ms` (~20× lower); a slow-but-alive client that misses the window is marked dead and reconnects, replaying the gap from the durable archive via its `Last-Event-ID` (or a fresh re-render for ephemeral streams, which have no archive). **Connect-replay** writes keep the full `streams_write_deadline_ms` so a new client catching up a large backlog isn't evicted. The happy path (all-fast clients) is unchanged — the `deadline_ms` threading is a zero-allocation keyword pass-through (bench: 585k vs 583k i/s, within noise; 0 retained/op). `streams_dispatch_queue_limit` (default `0` = unbounded) optionally caps distinct-stream wake backlog: at the cap the Listener drops durable wakes (never ephemeral or connect/disconnect), safe because the next durable wake for a stream re-reads from the min cursor. Both knobs are dispatcher-internal tuning (not part of the 1.0 public-config surface). A per-connection writer offload — moving the socket write off the dispatcher thread entirely — is deferred to a follow-up (it needs an ack/replay path for ephemeral frames first). Refs #315. - **Dedicated streams DB connection pool isolates durable-stream publish + replay from the job pool (`config.streams_pool_size`, `config.streams_pool_timeout`).** Both stream hot paths used to ride on the single job pool sized for worker threads: `Client#send_stream_message` (the broadcast INSERT) drew from `@pgmq`, so under a saturated job pool a broadcast blocked on pool checkout up to `pool_timeout` (~5s) — the browser waited on jobs at the *publish* step; and the dispatcher's per-wake replay read (`read_after` / `stream_current_msg_id` / `stream_oldest_msg_id`) went through `with_raw_connection`, which on the String/Hash config does a **full `PG.connect` + close per call** (TCP + auth + TLS) on the single dispatcher thread inside the fanout loop. Both now use a dedicated `@streams_pgmq` pool (own `PGMQ::Client` → own `connection_pool`), sized independently of worker thread counts via `streams_pool_size` (default 5) / `streams_pool_timeout` (default 5). A new `Client#with_streams_connection` checks out a persistent pooled connection for the replay reads; `Client#streams_pool_stats` exposes its counters alongside `pool_stats`. On the shared-ActiveRecord (Proc) connection path no second pool is created — libpq isn't thread-safe, so `@streams_pgmq` aliases `@pgmq` and streams keep using the single serialized connection. Because every forked process opens its own streams pool on the dedicated path, budget Postgres/PgBouncer `max_connections` for `resolved_pool_size + streams_pool_size` per process (both pools are lazy). Measured (real DB, 50-row stream, `read_after`): pooled replay reads run **~23× faster** than the old fresh-connect-per-call path (5.2k i/s / 191 μs vs 226 i/s / 4.42 ms), removing the per-wake connection-setup cost; this is a client-layer connection-setup win, not end-to-end broadcast-to-browser latency (which is dominated by the PGMQ round-trip + LISTEN/NOTIFY + socket write). Refs #315. diff --git a/Rakefile b/Rakefile index 9a8b2810..3af377cc 100644 --- a/Rakefile +++ b/Rakefile @@ -23,7 +23,8 @@ namespace :bench do bench_dir = "benchmarks" # Benches that need a real PostgreSQL/PGMQ (or boot Puma) — excluded from the # no-DB unit suite that bench:all runs in CI. - db_benches = %w[connection_pool_bench integration_bench streams_bench streams_read_pool_bench].freeze + db_benches = %w[connection_pool_bench integration_bench streams_bench streams_read_pool_bench + execution_modes_bench].freeze # The unit suite is every *_bench.rb that doesn't need a database, derived # from the directory so a new unit bench is picked up automatically (kept in # sync with bench:one, which globs the same files). @@ -63,6 +64,11 @@ namespace :bench do ruby "benchmarks/streams_bench.rb" end + desc "Run execution-mode connection benchmark (threads vs async pool usage; requires PGBUS_DATABASE_URL)" + task :execution_modes do + ruby "benchmarks/execution_modes_bench.rb" + end + desc "Run a single benchmark: rake bench:one[client_bench]" task :one, [:name] do |_t, args| name = args[:name] or abort "Usage: rake bench:one[serialization_bench|client_bench|...]" diff --git a/benchmarks/connection_pool_bench.rb b/benchmarks/connection_pool_bench.rb index b5dec3d0..d6fc3487 100644 --- a/benchmarks/connection_pool_bench.rb +++ b/benchmarks/connection_pool_bench.rb @@ -4,6 +4,14 @@ # Measures peak active Postgres connections under varying concurrency # for ThreadPool vs AsyncPool execution modes. # +# NOTE: this bench opens a RAW PG.connect per task and offers UNEQUAL load per +# mode, so its "peak connections" reflects raw connect behavior, not how a +# shared pool is consumed. For the fair, pool-aware threads-vs-async +# connection-consumption comparison (equal offered load, real pool checkout, +# sizing guidance), use `benchmarks/execution_modes_bench.rb` +# (`rake bench:execution_modes`) — see docs/performance.md "Execution mode and +# DB pool sizing". This bench is kept for its raw peak-connection matrix. +# # REQUIRES: PGBUS_DATABASE_URL environment variable # # Usage: diff --git a/benchmarks/execution_modes_bench.rb b/benchmarks/execution_modes_bench.rb new file mode 100644 index 00000000..7f8ad4be --- /dev/null +++ b/benchmarks/execution_modes_bench.rb @@ -0,0 +1,163 @@ +# frozen_string_literal: true + +# Execution-mode connection benchmark (issue #323). +# +# Answers, with fair measurement, "for the same offered concurrency, how many +# real Postgres connections does threads mode vs async mode actually hold, and +# what throughput/latency does each deliver?" — so we can guide users on DB-pool +# sizing per execution_mode. +# +# THE HONEST FINDING it exists to prove/disprove: pgbus holds a pgmq connection +# only for the read/archive SQL round-trip, NOT the job body, so async does not +# "save connections" by yielding mid-checkout. Async's connection advantage is +# only the fraction of wall time spent OUTSIDE with_connection (io_light). For +# DB-bound work async is connection-bound like threads and a tiny pool STARVES. +# See ExecutionModeHarness for the mechanism. +# +# Requires PGBUS_DATABASE_URL. Run-and-report — never a CI gate. +# PGBUS_DATABASE_URL=postgres://user@host/db bundle exec rake bench:execution_modes + +require "logger" +require "concurrent" +require "pgbus" +require "pg" + +require_relative "support/execution_mode_harness" + +DATABASE_URL = ENV.fetch("PGBUS_DATABASE_URL") do + warn "PGBUS_DATABASE_URL not set. This benchmark requires a real PostgreSQL database." + warn "Example: PGBUS_DATABASE_URL=postgres://user@localhost:5432/pgbus_test bundle exec rake bench:execution_modes" + exit 1 +end + +JOB_COUNT = Integer(ENV.fetch("EM_BENCH_JOBS", "240")) +CONCURRENCY = Integer(ENV.fetch("EM_BENCH_CONCURRENCY", "12")) +MIN_FREE_BACKENDS = 20 + +IO_LIGHT = ExecutionModeHarness::IoProfile.io_light +DB_BOUND = ExecutionModeHarness::IoProfile.db_bound + +# mode, pool_size, io_profile — concurrency + job_count are fixed so offered +# load is identical by construction across every row. +MATRIX = [ + [:threads, 12, IO_LIGHT], # 1 baseline: ~1 conn per busy thread's checkout window + [:async, 12, IO_LIGHT], # 2 async, generously provisioned + [:async, 3, IO_LIGHT], # 3 THE claim: can 3 conns serve 12 fibers? (io-light) + [:threads, 3, IO_LIGHT], # 4 under-provisioned threads -> expect pool_timeouts + [:threads, 12, DB_BOUND], # 5 disproof baseline + [:async, 12, DB_BOUND], # 6 disproof: async peak_busy should ~= threads + [:async, 3, DB_BOUND], # 7 honesty: does 3 starve when DB-bound? + [:async, 6, IO_LIGHT] # 8 the interesting middle +].freeze + +def build_client(pool_size) + config = Pgbus::Configuration.new.tap do |c| + c.database_url = DATABASE_URL + c.queue_prefix = "pgbus_execmode" + c.default_queue = "default" + c.logger = Logger.new(IO::NULL) + c.pool_size = pool_size + c.pool_timeout = 5 + c.stats_enabled = false + end + Pgbus::Client.new(config, schema_ensured: true) +end + +def free_backends + conn = PG.connect(DATABASE_URL) + total = conn.exec("SHOW max_connections").first["max_connections"].to_i + used = conn.exec("SELECT count(*) AS n FROM pg_stat_activity").first["n"].to_i + conn.close + total - used +end + +def preflight! + free = free_backends + return if free >= MIN_FREE_BACKENDS + + warn "Only #{free} Postgres backends free (need >= #{MIN_FREE_BACKENDS}). " \ + "The local box is shared; free some connections and retry." + exit 1 +rescue PG::Error => e + warn "Preflight connection failed: #{e.message}" + exit 1 +end + +def run_row(mode, pool_size, io_profile) + return skip_async_if_missing(mode) if mode == :async && !async_available? + + client = build_client(pool_size) + begin + ExecutionModeHarness.run_cell( + mode: mode, pool_size: pool_size, concurrency: CONCURRENCY, + io_profile: io_profile, job_count: JOB_COUNT, client: client + ) + ensure + client.close + end +rescue PG::ConnectionBad, PGMQ::Errors::ConnectionError => e + warn " skipped #{mode}/pool=#{pool_size}/#{io_profile.label}: #{e.message.lines.first&.strip}" + nil +end + +def async_available? + require "async" + true +rescue LoadError + false +end + +def skip_async_if_missing(_mode) + warn " skipped async row: the `async` gem is not installed (add gem \"async\")" + nil +end + +def print_table(results) + headers = ExecutionModeHarness::Result.headers + widths = headers.map.with_index do |h, i| + [h.length, *results.map { |r| r.to_row[i].to_s.length }].max + end + puts headers.each_with_index.map { |h, i| h.ljust(widths[i]) }.join(" ") + puts widths.map { |w| "-" * w }.join(" ") + results.each do |r| + puts r.to_row.each_with_index.map { |v, i| v.to_s.ljust(widths[i]) }.join(" ") + end +end + +puts "=" * 78 +puts "Execution-mode connection benchmark (issue #323)" +puts "Database: #{DATABASE_URL.sub(%r{//[^@]+@}, "//***@")}" +puts "Offered load per row: #{JOB_COUNT} jobs at concurrency #{CONCURRENCY} (identical for every row)" +puts "=" * 78 + +preflight! + +results = [] +MATRIX.each do |mode, pool_size, io_profile| + print "running #{mode}/pool=#{pool_size}/#{io_profile.label} ... " + $stdout.flush + result = run_row(mode, pool_size, io_profile) + puts result ? "done" : "skipped" + results << result if result + sleep 1 # let connections drain between cells (shared box) +end + +puts +print_table(results) + +puts +puts "Reading this table:" +puts " peak_busy = max(size - available) = live pool checkouts. This is the" +puts " connection cost of the offered load — NOT a throughput number." +puts " pool_timeouts > 0 means the pool was too small for the offered concurrency." +puts +puts "Honest framing:" +puts " * A job holds a pgmq connection only for the read/archive round-trip, not" +puts " the job body. Async does NOT save connections by yielding mid-checkout." +puts " * io_light (work mostly OUTSIDE the checkout): async can share a tiny pool." +puts " * db_bound (work mostly INSIDE the checkout): async is connection-bound" +puts " like threads; a tiny pool starves (pool_timeouts)." +puts " * async DB concurrency is reactor-bound unless a fiber-aware PG driver is" +puts " used — SELECT pg_sleep in with_connection is a blocking libpq call." +puts +puts "Done." diff --git a/benchmarks/support/execution_mode_harness.rb b/benchmarks/support/execution_mode_harness.rb new file mode 100644 index 00000000..fe4e6c5a --- /dev/null +++ b/benchmarks/support/execution_mode_harness.rb @@ -0,0 +1,293 @@ +# frozen_string_literal: true + +require "concurrent" + +# Execution-mode connection-consumption harness (issue #323). +# +# Runs the SAME offered workload through the threads and async execution pools +# and measures how many real Postgres connections each mode actually holds, so +# we can guide users on DB-pool sizing per execution_mode. +# +# THE HONEST FINDING this harness exists to prove/disprove: pgbus holds a pgmq +# connection only for the read/archive SQL round-trip, NOT the job body +# (executor.rb runs `perform_now` with zero pgmq connections). So async does NOT +# "save connections" by yielding mid-checkout — a fiber holds a connection for +# exactly one SQL round-trip, same as a thread. Async's only connection +# advantage is the fraction of wall time spent OUTSIDE `with_connection` +# (io_light). For DB-bound work, async is connection-bound just like threads and +# a tiny pool STARVES (surfacing as pool_timeouts). The io_profile cells prove +# which regime you're in. +# +# Pure, require-able module (no side effects on load, mirroring bench_support). +# The fairness/struct/measurement LOGIC is unit-tested with injected fakes (no +# DB); the actual numbers come from a real DB via run_cell in the bench script. +module ExecutionModeHarness + # A single measured (mode × pool_size × concurrency × io_profile) cell. + Result = Data.define( + :mode, # :threads | :async + :pool_size, # configured pgmq pool_size (the connection ceiling) + :concurrency, # offered concurrency = execution-pool capacity (threads / fibers) + :io_profile, # IoProfile + :job_count, # total jobs offered in the measured phase (SAME for both modes in a fair pair) + :peak_busy, # max(size - available) over samples — the headline metric + :steady_busy, # median(size - available) over samples + :throughput, # jobs/s over the measured phase + :p50, :p95, :p99, # per-job wall latency, ms + :pool_timeouts, # count of pgmq pool-timeout errors (tallied, never swallowed) + :other_errors, # count of any other error (must be 0 in a clean row) + :completed, # jobs finished without error + :wall_seconds + ) do + def under_provisioned? = pool_size < concurrency + + def self.headers + %w[mode pool conc io_profile jobs peak_busy steady_busy thr/s p50 p95 p99 timeouts errs] + end + + def to_row + [mode, pool_size, concurrency, io_profile.label, job_count, + peak_busy, steady_busy, throughput.round(1), + p50.round(1), p95.round(1), p99.round(1), pool_timeouts, other_errors] + end + end + + # Two portions of a job's wall time: + # db_seconds — INSIDE a pooled checkout (SELECT pg_sleep) → holds 1 conn, blocks the fiber + # yield_seconds — OUTSIDE any checkout (app I/O) → holds 0 pgmq conns; a fiber can share its slot here + # db_seconds is kept >= 10ms (above the 5ms sampler interval) so peak_busy + # can't undercount the checkout window. + IoProfile = Data.define(:label, :db_seconds, :yield_seconds) do + def self.io_light = new(label: "io_light", db_seconds: 0.010, yield_seconds: 0.040) + def self.db_bound = new(label: "db_bound", db_seconds: 0.030, yield_seconds: 0.005) + end + + # Raw result of driving `job_count` jobs through a pool. + Outcome = Data.define(:latencies, :elapsed, :finished, :completed, :pool_timeouts, :other_errors) + + # Substring pgmq stamps into a ConnectionPool::TimeoutError message. Replicated + # here (rather than calling Client#pool_timeout_error?, which is private) so the + # harness tally can't drift from production classification. + POOL_TIMEOUT_MARKER = "connection pool timeout" + + # Samples the pool's own counters (client.pool_stats → {size:, available:}). + # busy = size - available is the EXACT live checkout count, race-free (an + # in-memory read of the object under test) and process-scoped (immune to + # foreign backends on a shared Postgres). This is why we sample the pool, not + # pg_stat_activity. + class PoolSampler + def initialize(client, interval_s: 0.005) + @client = client + @interval = interval_s + @busy = Concurrent::Array.new + end + + attr_reader :interval + + # Records one stat sample. A degraded read ({}) is skipped, NOT counted as + # busy: 0 (which would fabricate an idle sample). Public so the primitive is + # unit-testable without a background thread; the harness's sampling thread + # calls it every `interval` seconds during the measured phase. + def record(stats) + return if stats.nil? || stats.empty? + + @busy << (stats[:size] - stats[:available]) + nil + end + + def summary + xs = @busy.to_a + { peak_busy: xs.max || 0, steady_busy: median(xs) } + end + + private + + def median(values) + return 0 if values.empty? + + sorted = values.sort + mid = sorted.length / 2 + sorted.length.odd? ? sorted[mid] : ((sorted[mid - 1] + sorted[mid]) / 2.0).round + end + end + + module_function + + # Runs ONE cell against a real client and returns a Result. `sampler` is + # injectable so specs can drive it deterministically without a DB. + def run_cell(mode:, pool_size:, concurrency:, io_profile:, job_count:, client:, + sampler: nil, warmup_jobs: nil, clock: Process) + assert_dedicated_path!(client) + sampler ||= PoolSampler.new(client) + warmup_jobs ||= pool_size * 2 + pool = Pgbus::ExecutionPools.build(mode: mode, capacity: concurrency) + begin + warmup!(pool: pool, client: client, io_profile: io_profile, warmup_jobs: warmup_jobs) + sampler_thread = start_sampling(sampler, client) + outcome = drive(pool: pool, client: client, io_profile: io_profile, job_count: job_count, clock: clock) + stop_sampling(sampler_thread) + summarize(mode: mode, pool_size: pool_size, concurrency: concurrency, + io_profile: io_profile, job_count: job_count, sampler: sampler, outcome: outcome) + ensure + pool.shutdown + pool.wait_for_termination(15) + end + end + + # Fairness invariant: two Results are only comparable if offered load matches. + # The direct anti-rigging guard against unequal-load comparisons. + def assert_fair_pair!(lhs, rhs) + return if lhs.job_count == rhs.job_count && lhs.io_profile == rhs.io_profile && + lhs.concurrency == rhs.concurrency + + raise ArgumentError, + "unfair comparison: offered load differs " \ + "(#{lhs.mode}=#{lhs.job_count}/#{lhs.concurrency}/#{lhs.io_profile.label} vs " \ + "#{rhs.mode}=#{rhs.job_count}/#{rhs.concurrency}/#{rhs.io_profile.label})" + end + + def assert_dedicated_path!(client) + return unless client.shared_connection? + + raise ArgumentError, + "execution-mode bench requires the dedicated-connection path " \ + "(set config.database_url = PGBUS_DATABASE_URL); the shared-AR path forces " \ + "pool_size:1 and makes the threads-vs-async question meaningless" + end + + # Tally a job error without swallowing it: pool timeouts (the under-provision + # signal) vs everything else. Nothing is dropped — sum of tallies == errors. + def classify_error!(error, timeouts, others) + if error.message.to_s.downcase.include?(POOL_TIMEOUT_MARKER) + timeouts.increment + else + others.increment + Pgbus.logger.debug { "[em-harness] job error #{error.class}: #{error.message}" } if defined?(Pgbus.logger) + end + nil + end + + # Fiber-aware yield. Under async a bare Kernel#sleep parks the WHOLE reactor + # (the exact execution_pool_bench.rb bug); use Async::Task#sleep when a + # scheduler is present. + def async_yield(seconds) + task = (Async::Task.current? if defined?(Async::Task)) + if task + task.sleep(seconds) + else + sleep(seconds) + end + end + + # THE fair unit of work — identical bytes for threads & async, real pooled + # checkout. yield_seconds runs OUTSIDE any checkout (fiber can share its slot); + # db_seconds runs INSIDE a real pooled checkout via the public reader. + def run_job(client, io_profile) + async_yield(io_profile.yield_seconds) if io_profile.yield_seconds.positive? + return unless io_profile.db_seconds.positive? + + client.pgmq.with_connection do |conn| + conn.exec_params("SELECT pg_sleep($1)", [io_profile.db_seconds]) + end + end + + # Posts EXACTLY job_count units through a BOUNDED in-flight window equal to the + # pool capacity, so at most `capacity` jobs run concurrently in BOTH modes — + # the fair definition of "offered concurrency = capacity". This also normalizes + # a real behavioral difference: ThreadPool queues unlimited posts, but + # AsyncPool#post RAISES once `capacity` units are in-flight (async_pool.rb + # reserve_capacity!). Submitting behind a Concurrent::Semaphore sized to the + # capacity keeps both pools at the same concurrency and never overflows async. + # Records per-job wall latency, tallies errors (never swallows). + def drive(pool:, client:, io_profile:, job_count:, clock: Process) + latencies = Concurrent::Array.new + timeouts = Concurrent::AtomicFixnum.new(0) + others = Concurrent::AtomicFixnum.new(0) + completed = Concurrent::AtomicFixnum.new(0) + done = Concurrent::CountDownLatch.new(job_count) + slots = Concurrent::Semaphore.new(pool.capacity) + started = clock.clock_gettime(Process::CLOCK_MONOTONIC) + + job_count.times do + slots.acquire # block the submitter until a slot frees — bounds in-flight + pool.post do + t0 = clock.clock_gettime(Process::CLOCK_MONOTONIC) + run_job(client, io_profile) + latencies << ((clock.clock_gettime(Process::CLOCK_MONOTONIC) - t0) * 1000.0) + completed.increment + rescue StandardError => e + classify_error!(e, timeouts, others) + ensure + slots.release + done.count_down + end + end + + finished = done.wait(deadline_for(job_count, io_profile)) + elapsed = clock.clock_gettime(Process::CLOCK_MONOTONIC) - started + Outcome.new(latencies: latencies.to_a, elapsed: elapsed, finished: finished, + completed: completed.value, pool_timeouts: timeouts.value, other_errors: others.value) + end + + def summarize(mode:, pool_size:, concurrency:, io_profile:, job_count:, sampler:, outcome:) + sorted = outcome.latencies.sort + busy = sampler.summary + Result.new( + mode: mode, pool_size: pool_size, concurrency: concurrency, io_profile: io_profile, + job_count: job_count, peak_busy: busy[:peak_busy], steady_busy: busy[:steady_busy], + throughput: outcome.elapsed.positive? ? (outcome.completed / outcome.elapsed) : 0.0, + p50: percentile(sorted, 0.50), p95: percentile(sorted, 0.95), p99: percentile(sorted, 0.99), + pool_timeouts: outcome.pool_timeouts, other_errors: outcome.other_errors, + completed: outcome.completed, wall_seconds: outcome.elapsed + ) + end + + # Sorted-array percentile (pure; unit-tested against [1..100]). + def percentile(sorted, fraction) + return 0.0 if sorted.empty? + + idx = [(sorted.length * fraction).to_i, sorted.length - 1].min + sorted[idx] + end + + def deadline_for(job_count, io_profile) + per = io_profile.db_seconds + io_profile.yield_seconds + [(job_count * per) + 30, 180].min + end + + # Drives warmup_jobs concurrently so connection_pool opens its working set + # (lazy: first checkout pays TCP/TLS/auth). Warmup latencies are discarded. + def warmup!(pool:, client:, io_profile:, warmup_jobs:) + done = Concurrent::CountDownLatch.new(warmup_jobs) + slots = Concurrent::Semaphore.new(pool.capacity) # bound in-flight (async rejects overflow) + warmup_jobs.times do + slots.acquire + pool.post do + run_job(client, io_profile) + rescue StandardError + nil + ensure + slots.release + done.count_down + end + end + done.wait(60) + end + + # Runs the sampler on its own thread for the measured phase. + def start_sampling(sampler, client) + stop = Concurrent::AtomicBoolean.new(false) + thread = Thread.new do + until stop.true? + sampler.record(client.pool_stats) + sleep(sampler.interval) + end + end + [thread, stop] + end + + def stop_sampling(handle) + thread, stop = handle + stop.make_true + thread&.join(2) + end +end diff --git a/docs/performance.md b/docs/performance.md index 83064c57..b020b44e 100644 --- a/docs/performance.md +++ b/docs/performance.md @@ -38,6 +38,7 @@ rake bench:memory # detailed memory profiling with allocation breakdown rake bench:integration # real PostgreSQL + PGMQ (requires PGBUS_DATABASE_URL) rake bench:streams # real Puma + SSE fan-out (requires PGBUS_DATABASE_URL) rake bench:one[streams_read_pool_bench] # streamer replay-read pool (requires PGBUS_DATABASE_URL) +rake bench:execution_modes # threads vs async DB-connection consumption (requires PGBUS_DATABASE_URL) ``` - **Unit benches** (`benchmarks/*_bench.rb`) isolate gem overhead with a mocked @@ -149,6 +150,61 @@ frame on overflow — safe because durable frames are re-read from the archive o reconnect; it's an OOM guard for a pathologically slow-but-alive client, not a delivery guarantee. +### Execution mode and DB pool sizing (threads vs async) + +**Offered concurrency is not connections held.** A worker's `threads:` setting is +how many jobs run at once; `resolved_pool_size` is how many *DB connections* the +pgmq pool holds. They are not the same number, because a job holds a pgmq +connection only for the `read_batch` + `archive` SQL round-trip — `perform_now` +runs with **zero** pgmq connections checked out (`executor.rb`). If your job does +its own database work, that uses ActiveRecord's *separate* pool, not this one. + +This means the folklore "async saves connections" is only half true, and the +`benchmarks/execution_modes_bench.rb` harness (`rake bench:execution_modes`, +requires `PGBUS_DATABASE_URL`) measures exactly when it holds. It runs the same +offered load (240 jobs at concurrency 12) through both pools and samples +`peak_busy = pool size − available` — the live checkout count — going through the +real pool, so pool *sharing* is what's measured. Representative local numbers: + +| mode | pool | io profile | peak_busy | throughput | note | +|------|------|-----------|-----------|-----------|------| +| threads | 12 | io_light | 12 | ~216 job/s | baseline: ~1 conn per busy thread's checkout window | +| async | 3 | io_light | **3** | ~222 job/s | **12 fibers sustained on 3 connections, full throughput** | +| threads | 3 | io_light | 3 | ~208 job/s | short 10 ms checkouts cycle fast enough that 3 conns serve 12 threads too | +| threads | 12 | db_bound | 12 | ~320 job/s | connection-bound — one conn per concurrent DB call | +| async | 12 | db_bound | 12 | ~324 job/s | async matches threads when the pool is sized right | +| async | **3** | db_bound | 3 | **~95 job/s** | **under-provisioned: throughput collapses ~3.4×** (p50 37 ms → 125 ms) | + +**What the numbers actually say:** + +- **Async's connection-density win is real for I/O-light work** — where a job + spends most of its time *outside* the checkout (HTTP calls, app compute, waits). + There, a handful of connections serves many fibers with no throughput loss. This + is why async workers are auto-sized to a flat `ASYNC_POOL_CONNECTIONS` (3) per + capsule rather than one-per-fiber. +- **For DB-bound work, async is connection-bound just like threads.** A fiber + holds a connection for the whole SQL call (a blocking libpq call does not yield + the reactor), so a too-small async pool doesn't share — it **serialises**, and + throughput collapses. +- **Under-provisioning degrades throughput; it does not necessarily error.** + Because `pool_timeout` (5 s) dwarfs a typical checkout (10–30 ms), a + too-small pool rarely times out — it just serialises work behind the available + connections. Watch `throughput` and p95/p99 latency, not just error counts. + +**Sizing guidance:** + +| mode | recommended `pool_size` | under-provision symptom | over-provision cost | +|------|------------------------|-------------------------|---------------------| +| threads | ≈ `Σ worker threads` (+ dispatcher/scheduler/consumers) — the auto-tuned default | throughput collapse; eventually `pool_timeout` (`enrich_pool_timeout_error`), `available → 0` | wastes Postgres `max_connections`; `warn_if_oversized` fires above 50 | +| async | `ASYNC_POOL_CONNECTIONS` (3) for **I/O-light** work; **≈ your peak concurrent DB calls** for **DB-bound** work — measure with `rake bench:execution_modes` | reactor fibers serialise on checkout → throughput collapse (harder to spot; no error) | forfeits the density win async exists for | + +This is a connection-**density** tool, not a throughput tuner: a smaller pool means +fewer Postgres backends for the same offered load, not faster jobs (PGMQ round-trip +and job I/O dominate wall time). The numbers are single-box, single-process, no +PgBouncer — a per-process, per-io-profile sizing guide, not a production capacity +guarantee. Budget `resolved_pool_size + streams_pool_size` per forked process (see +Capacity planning above). + ### Reading the output `benchmark-ips` reports **i/s** (iterations per second — higher is better). diff --git a/spec/integration/execution_mode_harness_spec.rb b/spec/integration/execution_mode_harness_spec.rb new file mode 100644 index 00000000..b7899df2 --- /dev/null +++ b/spec/integration/execution_mode_harness_spec.rb @@ -0,0 +1,90 @@ +# frozen_string_literal: true + +require_relative "../integration_helper" +require_relative "../../benchmarks/support/execution_mode_harness" + +# DB-gated specs for the execution-mode harness (issue #323): prove run_cell +# drives real jobs through the real pool checkout and reports sane metrics. +# Auto-skips when PGBUS_DATABASE_URL is unset (see integration_helper). These +# are run-and-report, never a CI perf gate. +RSpec.describe "ExecutionModeHarness run_cell", :integration do + let(:database_url) { ENV.fetch("PGBUS_DATABASE_URL") } + # Small offered load so the spec stays fast and connection-modest. + let(:job_count) { 24 } + let(:concurrency) { 4 } + + # A dedicated-path client sized to `pool_size`. Own config so pool_size varies + # per example without touching the shared Pgbus.configuration. + def build_client(pool_size) + config = Pgbus::Configuration.new.tap do |c| + c.database_url = database_url + c.queue_prefix = "pgbus_emspec" + c.default_queue = "default" + c.logger = Logger.new(IO::NULL) + c.pool_size = pool_size + c.pool_timeout = 5 + c.stats_enabled = false + end + Pgbus::Client.new(config, schema_ensured: true) + end + + it "drives a threads cell to completion with no errors and peak_busy <= pool_size" do + client = build_client(4) + result = ExecutionModeHarness.run_cell( + mode: :threads, pool_size: 4, concurrency: concurrency, + io_profile: ExecutionModeHarness::IoProfile.io_light, + job_count: job_count, client: client + ) + + expect(result.completed).to eq(job_count) + expect(result.other_errors).to eq(0) + expect(result.peak_busy).to be <= 4 + expect(result.peak_busy).to be > 0 # it actually held connections + expect(result.throughput).to be > 0 + ensure + client&.close + end + + it "reports higher peak_busy at higher pool_size for the same offered load" do + small = build_client(2) + large = build_client(4) + io = ExecutionModeHarness::IoProfile.db_bound + + r_small = ExecutionModeHarness.run_cell(mode: :threads, pool_size: 2, concurrency: concurrency, + io_profile: io, job_count: job_count, client: small) + r_large = ExecutionModeHarness.run_cell(mode: :threads, pool_size: 4, concurrency: concurrency, + io_profile: io, job_count: job_count, client: large) + + # Same offered load — a fair pair — and the bigger pool holds more connections. + expect { ExecutionModeHarness.assert_fair_pair!(r_small, r_large) }.not_to raise_error + expect(r_large.peak_busy).to be >= r_small.peak_busy + ensure + small&.close + large&.close + end + + it "runs an async cell when the async gem is available (fibers share a small pool on io_light)" do + skip "async gem not installed" unless async_gem_available? + + client = build_client(2) + result = ExecutionModeHarness.run_cell( + mode: :async, pool_size: 2, concurrency: concurrency, + io_profile: ExecutionModeHarness::IoProfile.io_light, + job_count: job_count, client: client + ) + + expect(result.completed).to eq(job_count) + expect(result.other_errors).to eq(0) + # Proves fibers shared: concurrency 4 sustained on a pool of 2. + expect(result.peak_busy).to be <= 2 + ensure + client&.close + end + + def async_gem_available? + require "async" + true + rescue LoadError + false + end +end diff --git a/spec/pgbus/execution_mode_harness_spec.rb b/spec/pgbus/execution_mode_harness_spec.rb new file mode 100644 index 00000000..3b35d43f --- /dev/null +++ b/spec/pgbus/execution_mode_harness_spec.rb @@ -0,0 +1,192 @@ +# frozen_string_literal: true + +require "spec_helper" +require "concurrent" +require_relative "../../benchmarks/support/execution_mode_harness" + +# Unit specs for the execution-mode benchmark harness (issue #323). These run +# WITHOUT a database — the fairness/struct/measurement LOGIC is tested with +# injected fakes. The actual connection/throughput NUMBERS come from a manual +# `rake bench:execution_modes` run against a real Postgres (run-and-report, +# never a CI gate). +RSpec.describe ExecutionModeHarness do + describe ExecutionModeHarness::IoProfile do + it "io_light spends most time OUTSIDE the checkout (fiber-shareable)" do + p = described_class.io_light + expect(p.label).to eq("io_light") + expect(p.yield_seconds).to be > p.db_seconds + end + + it "db_bound spends most time INSIDE the checkout (connection-bound)" do + p = described_class.db_bound + expect(p.label).to eq("db_bound") + expect(p.db_seconds).to be > p.yield_seconds + end + + it "keeps db_seconds at or above 10ms so the 5ms sampler can't undercount the checkout" do + [described_class.io_light, described_class.db_bound].each do |p| + expect(p.db_seconds).to be >= 0.010 + end + end + end + + describe ExecutionModeHarness::Result do + subject(:result) do + described_class.new( + mode: :threads, pool_size: 12, concurrency: 12, + io_profile: ExecutionModeHarness::IoProfile.io_light, + job_count: 240, peak_busy: 9, steady_busy: 4, + throughput: 123.45, p50: 41.0, p95: 60.0, p99: 90.0, + pool_timeouts: 0, other_errors: 0, completed: 240, wall_seconds: 1.94 + ) + end + + it "is under_provisioned? only when pool_size < concurrency" do + expect(result.under_provisioned?).to be false + smaller = result.with(pool_size: 3) + expect(smaller.under_provisioned?).to be true + end + + it "produces a row whose width matches the headers" do + expect(result.to_row.length).to eq(described_class.headers.length) + end + end + + describe ".summarize" do + it "computes throughput as completed / elapsed and correct percentiles" do + latencies = (1..100).map(&:to_f) # sorted 1..100 + outcome = ExecutionModeHarness::Outcome.new( + latencies: latencies, elapsed: 2.0, finished: true, + completed: 100, pool_timeouts: 0, other_errors: 0 + ) + sampler = instance_double(ExecutionModeHarness::PoolSampler, + summary: { peak_busy: 9, steady_busy: 4 }) + + result = described_class.summarize( + mode: :threads, pool_size: 12, concurrency: 12, + io_profile: ExecutionModeHarness::IoProfile.io_light, + job_count: 100, sampler: sampler, outcome: outcome + ) + + expect(result.throughput).to eq(50.0) # 100 / 2.0 + expect(result.p50).to be_within(2.0).of(50.0) + expect(result.p95).to be_within(2.0).of(95.0) + expect(result.p99).to be_within(2.0).of(99.0) + expect(result.peak_busy).to eq(9) + expect(result.steady_busy).to eq(4) + end + end + + describe ".assert_fair_pair!" do + def result(mode:, job_count: 240, io: ExecutionModeHarness::IoProfile.io_light, concurrency: 12) + ExecutionModeHarness::Result.new( + mode: mode, pool_size: 12, concurrency: concurrency, io_profile: io, + job_count: job_count, peak_busy: 1, steady_busy: 1, throughput: 1.0, + p50: 1.0, p95: 1.0, p99: 1.0, pool_timeouts: 0, other_errors: 0, + completed: job_count, wall_seconds: 1.0 + ) + end + + it "passes when offered load (job_count, io_profile, concurrency) matches" do + expect do + described_class.assert_fair_pair!(result(mode: :threads), result(mode: :async)) + end.not_to raise_error + end + + it "raises when job_count differs (the anti-rigging guard)" do + expect do + described_class.assert_fair_pair!(result(mode: :threads, job_count: 20), + result(mode: :async, job_count: 200)) + end.to raise_error(ArgumentError, /unfair/) + end + + it "raises when io_profile differs" do + expect do + described_class.assert_fair_pair!( + result(mode: :threads, io: ExecutionModeHarness::IoProfile.io_light), + result(mode: :async, io: ExecutionModeHarness::IoProfile.db_bound) + ) + end.to raise_error(ArgumentError, /unfair/) + end + end + + describe ".classify_error!" do + let(:timeouts) { Concurrent::AtomicFixnum.new(0) } + let(:others) { Concurrent::AtomicFixnum.new(0) } + + it "tallies a pool-timeout error by its message marker (never swallows)" do + # classify_error! keys off the message substring (POOL_TIMEOUT_MARKER), + # mirroring production's pool_timeout_error?, so the real PGMQ error class + # (not loaded in unit tests) isn't needed — only the marker text. + err = StandardError.new("Connection pool timeout: waited 5s") + described_class.classify_error!(err, timeouts, others) + expect(timeouts.value).to eq(1) + expect(others.value).to eq(0) + end + + it "tallies a non-timeout error as other_errors" do + described_class.classify_error!(StandardError.new("boom"), timeouts, others) + expect(timeouts.value).to eq(0) + expect(others.value).to eq(1) + end + end + + describe ".assert_dedicated_path!" do + it "raises on the shared-AR path (pool_size forced to 1)" do + client = instance_double(Pgbus::Client, shared_connection?: true) + expect { described_class.assert_dedicated_path!(client) } + .to raise_error(ArgumentError, /dedicated-connection path/) + end + + it "passes on the dedicated path" do + client = instance_double(Pgbus::Client, shared_connection?: false) + expect { described_class.assert_dedicated_path!(client) }.not_to raise_error + end + end + + describe ".async_yield" do + it "uses Kernel#sleep when no Async scheduler is present" do + allow(described_class).to receive(:sleep) + described_class.async_yield(0.001) + expect(described_class).to have_received(:sleep).with(0.001) + end + end + + describe ExecutionModeHarness::PoolSampler do + subject(:sampler) { described_class.new(client, interval_s: 0.001) } + + # A client whose pool_stats returns a scripted sequence, so busy = + # size - available and the median math are provable with no DB. + let(:client) do + stats = [ + { size: 12, available: 12, pool_timeout: 5 }, # busy 0 + { size: 12, available: 3, pool_timeout: 5 }, # busy 9 (peak) + { size: 12, available: 8, pool_timeout: 5 } # busy 4 + ] + instance_double(Pgbus::Client).tap do |c| + seq = stats.dup + allow(c).to receive(:pool_stats) { seq.shift || stats.last } + end + end + + it "reports peak_busy and median steady_busy from size - available" do + # Drive the sampler deterministically over exactly the scripted stats. + sampler.record(client.pool_stats) + sampler.record(client.pool_stats) + sampler.record(client.pool_stats) + + summary = sampler.summary + expect(summary[:peak_busy]).to eq(9) + expect(summary[:steady_busy]).to eq(4) # median([0,9,4]) + end + + it "skips a degraded ({}) sample instead of counting it as busy: 0" do + sampler.record({}) + sampler.record({ size: 10, available: 2 }) # busy 8 + + summary = sampler.summary + expect(summary[:peak_busy]).to eq(8) + expect(summary[:steady_busy]).to eq(8) # only the one real sample + end + end +end From 93f2d480f6b0c1c52bc2146d73ce4b71f1ba8647 Mon Sep 17 00:00:00 2001 From: mhenrixon Date: Mon, 6 Jul 2026 06:08:22 +0200 Subject: [PATCH 2/2] =?UTF-8?q?fix(bench):=20address=20#324=20review=20?= =?UTF-8?q?=E2=80=94=20resource=20leaks,=20deprecated=20API,=20flaky=20spe?= =?UTF-8?q?c?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Five CodeRabbit findings on the execution-mode benchmark: - free_backends leaked the raw PG connection if a query raised (conn.close was skipped, and preflight! rescues but doesn't close). Moved close to an ensure. - run_cell leaked the sampler thread if warmup!/drive/summarize raised: the ensure tore down the pool but never stopped the sampler, which then polled pool_stats forever. Hoisted sampler_thread and stop it in the ensure. - Async::Task#sleep is deprecated in async 2.x (the gem's own @deprecated tag) in favor of Kernel#sleep, which the fiber scheduler already intercepts (kernel_sleep hook) and yields to the reactor — verified: two concurrent sleep(0.2) fibers finish in ~201ms, not 400ms. Deleted the over-engineered async_yield method; run_job now just calls sleep. The async cells still share a small pool (async pool=3 io_light: peak_busy=3, 12 fibers) — behavior unchanged, confirmed against a real DB run. - Hardened the flaky integration assertion: replaced the cross-run peak_busy >= peak_busy timing comparison (could flip if the 5ms sampler missed a peak under GC) with deterministic per-pool ceiling invariants (peak_busy between 1 and pool_size). - Made the async DB spec honest: `async` is not a declared dependency (present only transitively via falcon/async-http dev deps), so the skip message now spells out that a SKIP is not a pass and async execution mode requires adding the gem. ## Verification - [x] bundle exec rubocop passes - [x] bundle exec rspec spec/pgbus — 3037 examples, 0 failures - [x] DB-gated harness specs pass; rake bench:execution_modes numbers unchanged --- benchmarks/execution_modes_bench.rb | 3 +- benchmarks/support/execution_mode_harness.rb | 29 ++++++++--------- .../execution_mode_harness_spec.rb | 17 ++++++++-- spec/pgbus/execution_mode_harness_spec.rb | 32 ++++++++++++++++--- 4 files changed, 57 insertions(+), 24 deletions(-) diff --git a/benchmarks/execution_modes_bench.rb b/benchmarks/execution_modes_bench.rb index 7f8ad4be..ecfc2a3f 100644 --- a/benchmarks/execution_modes_bench.rb +++ b/benchmarks/execution_modes_bench.rb @@ -67,8 +67,9 @@ def free_backends conn = PG.connect(DATABASE_URL) total = conn.exec("SHOW max_connections").first["max_connections"].to_i used = conn.exec("SELECT count(*) AS n FROM pg_stat_activity").first["n"].to_i - conn.close total - used +ensure + conn&.close # close even if a query raises (preflight! rescues the error) end def preflight! diff --git a/benchmarks/support/execution_mode_harness.rb b/benchmarks/support/execution_mode_harness.rb index fe4e6c5a..0d22b300 100644 --- a/benchmarks/support/execution_mode_harness.rb +++ b/benchmarks/support/execution_mode_harness.rb @@ -120,14 +120,17 @@ def run_cell(mode:, pool_size:, concurrency:, io_profile:, job_count:, client:, sampler ||= PoolSampler.new(client) warmup_jobs ||= pool_size * 2 pool = Pgbus::ExecutionPools.build(mode: mode, capacity: concurrency) + sampler_thread = nil begin warmup!(pool: pool, client: client, io_profile: io_profile, warmup_jobs: warmup_jobs) sampler_thread = start_sampling(sampler, client) outcome = drive(pool: pool, client: client, io_profile: io_profile, job_count: job_count, clock: clock) - stop_sampling(sampler_thread) summarize(mode: mode, pool_size: pool_size, concurrency: concurrency, io_profile: io_profile, job_count: job_count, sampler: sampler, outcome: outcome) ensure + # Stop the sampler FIRST (else its thread polls pool_stats forever if + # drive/summarize raised), then tear down the pool. + stop_sampling(sampler_thread) if sampler_thread pool.shutdown pool.wait_for_termination(15) end @@ -166,23 +169,17 @@ def classify_error!(error, timeouts, others) nil end - # Fiber-aware yield. Under async a bare Kernel#sleep parks the WHOLE reactor - # (the exact execution_pool_bench.rb bug); use Async::Task#sleep when a - # scheduler is present. - def async_yield(seconds) - task = (Async::Task.current? if defined?(Async::Task)) - if task - task.sleep(seconds) - else - sleep(seconds) - end - end - # THE fair unit of work — identical bytes for threads & async, real pooled - # checkout. yield_seconds runs OUTSIDE any checkout (fiber can share its slot); - # db_seconds runs INSIDE a real pooled checkout via the public reader. + # checkout. yield_seconds is app I/O OUTSIDE any checkout (a fiber can share + # its slot here); db_seconds runs INSIDE a real pooled checkout via the public + # reader. + # + # A plain Kernel#sleep is correct in BOTH modes: under Async's fiber scheduler + # it's intercepted (the kernel_sleep hook) and yields to the reactor, so other + # fibers keep running; under threads it just blocks the thread. (Async::Task#sleep + # is deprecated in favor of Kernel#sleep in async 2.x precisely because of this.) def run_job(client, io_profile) - async_yield(io_profile.yield_seconds) if io_profile.yield_seconds.positive? + sleep(io_profile.yield_seconds) if io_profile.yield_seconds.positive? return unless io_profile.db_seconds.positive? client.pgmq.with_connection do |conn| diff --git a/spec/integration/execution_mode_harness_spec.rb b/spec/integration/execution_mode_harness_spec.rb index b7899df2..0e2d0b35 100644 --- a/spec/integration/execution_mode_harness_spec.rb +++ b/spec/integration/execution_mode_harness_spec.rb @@ -55,16 +55,27 @@ def build_client(pool_size) r_large = ExecutionModeHarness.run_cell(mode: :threads, pool_size: 4, concurrency: concurrency, io_profile: io, job_count: job_count, client: large) - # Same offered load — a fair pair — and the bigger pool holds more connections. + # Same offered load — a fair pair. expect { ExecutionModeHarness.assert_fair_pair!(r_small, r_large) }.not_to raise_error - expect(r_large.peak_busy).to be >= r_small.peak_busy + # Deterministic invariants only — NOT a cross-run timing comparison (which + # would flake if the 5ms sampler missed a peak under GC/contention). A pool + # can never check out more than its size, and a DB-bound load at concurrency + # 4 keeps every slot busy, so each pool saturates to its own ceiling. + expect(r_small.peak_busy).to be_between(1, 2) + expect(r_large.peak_busy).to be_between(1, 4) ensure small&.close large&.close end it "runs an async cell when the async gem is available (fibers share a small pool on io_light)" do - skip "async gem not installed" unless async_gem_available? + # `async` is NOT a runtime dependency of pgbus — async execution mode + # requires the app to add `gem "async"` (AsyncPool#initialize raises a clear + # LoadError otherwise). It's present here transitively (via falcon/async-http + # dev deps), so this runs in this repo; in an environment without it, this + # example skips rather than failing — a SKIP is not a pass, so don't read + # green here as "async is covered" unless the gem is actually loadable. + skip "async gem not available (add `gem \"async\"` to exercise async execution mode)" unless async_gem_available? client = build_client(2) result = ExecutionModeHarness.run_cell( diff --git a/spec/pgbus/execution_mode_harness_spec.rb b/spec/pgbus/execution_mode_harness_spec.rb index 3b35d43f..d1ce37c4 100644 --- a/spec/pgbus/execution_mode_harness_spec.rb +++ b/spec/pgbus/execution_mode_harness_spec.rb @@ -144,11 +144,35 @@ def result(mode:, job_count: 240, io: ExecutionModeHarness::IoProfile.io_light, end end - describe ".async_yield" do - it "uses Kernel#sleep when no Async scheduler is present" do + describe ".run_job" do + # A plain Kernel#sleep is used for the yield portion — under Async's fiber + # scheduler it's intercepted and yields to the reactor; under threads it + # blocks the thread. No mode-specific branch needed. + it "sleeps for the yield portion outside any checkout" do allow(described_class).to receive(:sleep) - described_class.async_yield(0.001) - expect(described_class).to have_received(:sleep).with(0.001) + io = ExecutionModeHarness::IoProfile.new(label: "yield_only", db_seconds: 0.0, yield_seconds: 0.02) + client = instance_double(Pgbus::Client) # never touched: db_seconds is 0 + + described_class.run_job(client, io) + + expect(described_class).to have_received(:sleep).with(0.02) + end + + it "checks out a real pooled connection for the db portion" do + # PG / PGMQ::Client aren't loaded in unit specs (pgmq is stubbed away), so + # use plain doubles for the conn + pgmq; the harness only calls + # #with_connection and #exec_params on them. + allow(described_class).to receive(:sleep) + conn = double("PG::Connection", exec_params: nil) + pgmq = double("PGMQ::Client") + allow(pgmq).to receive(:with_connection).and_yield(conn) + client = instance_double(Pgbus::Client, pgmq: pgmq) + io = ExecutionModeHarness::IoProfile.new(label: "db_only", db_seconds: 0.01, yield_seconds: 0.0) + + described_class.run_job(client, io) + + expect(pgmq).to have_received(:with_connection) + expect(conn).to have_received(:exec_params).with("SELECT pg_sleep($1)", [0.01]) end end