Pooler soak: warm backends first; judge progress rows that outlive their call

test/harness/pooler_soak/samples.py

@@ -4,7 +4,8 @@
 context totals `pg_log_backend_memory_contexts` writes to the server
 log), pg_warren's own accounting (`warren.backends()`'s private,
 transient and mapped bytes per backend; `warren.shared_cache()`), the
-activity and progress rows, the DSM segments the cluster's processes map,
+activity and progress rows (and the progress rows that outlived their
+call: `OUTLIVED_PROGRESS_ROWS`), the DSM segments the cluster's processes map,
 the catalog's rows and table sizes, and each pooler's `SHOW POOLS` and
 `SHOW STATS`.
 
@@ -25,9 +26,19 @@
             "coalesce(b.mapped_bytes, 0) FROM pg_stat_activity a "
             "LEFT JOIN warren.backends() b ON b.pid = a.pid AND b.kind = 'backend' "
             "WHERE a.backend_type = 'client backend' AND a.usename = ANY ('{{{roles}}}')")
+# A progress row is a call in progress: its frames end with the call, or
+# with the statement or transaction around it (DESIGN.md §5.7). One that
+# outlived its call is on a backend that has since gone idle: its command
+# began before the backend's state last changed, to a state other than
+# running a statement. A call that begins after `pg_stat_activity`'s
+# snapshot shows the backend idle began after that change, so a row the
+# two reads see apart is not counted.
+OUTLIVED_PROGRESS_ROWS = ("SELECT count(*) FROM warren.progress() p JOIN pg_stat_activity a USING (pid) "
+                          "WHERE a.state <> 'active' AND p.command_start < a.state_change")
 CLUSTER_METRICS = {
     "activity_rows": "SELECT count(*) FROM warren.backends() WHERE kind = 'backend'",
     "progress_rows": "SELECT count(*) FROM warren.progress()",
+    "outlived_progress_rows": OUTLIVED_PROGRESS_ROWS,
     "shared_bytes": "SELECT coalesce(sum(bytes), 0) FROM warren.shared_cache() WHERE kind = 'cluster'",
     "shared_entries": "SELECT count(*) FROM warren.shared_cache() WHERE kind IN ('version', 'vector')",
     "shared_pinned": "SELECT count(*) FROM warren.shared_cache() WHERE kind IN ('version', 'vector') AND pins > 0",

test/harness/pooler_soak/settled.py

@@ -3,9 +3,11 @@
 exit should have freed, and nothing in the catalog that leaked.
 
 `after_clients`, with the poolers still up and their server backends
-idle: the session pool's backends hold no transient vector (its
-`DISCARD ALL` reset each one when its client left); the transaction
-pool's are reported (no reset runs between transactions). `after_poolers`,
+idle: no progress row outlived its call (a call a client abandoned may
+still run, until its backend sees the client gone); the session pool's
+backends hold no transient vector (its `DISCARD ALL` reset each one when
+its client left); the transaction pool's are reported (no reset runs
+between transactions). `after_poolers`,
 once the poolers have closed their server connections: no activity row
 of a backend that has gone, no progress row, nothing pinned or mapped in
 the shared cache, no staged version, no segment or chunk without its
@@ -18,15 +20,16 @@
 import time
 
 from .fixture import GRAPHS
+from .samples import OUTLIVED_PROGRESS_ROWS
 from .workload import PERSISTED_NAMES_PER_CLIENT
 
 SETTLE_SECONDS = 15
 RETAIN = {"g": 3, "t": 2}
 
 
-def until(admin, sql, seconds=SETTLE_SECONDS):
-    """Polls `sql`, a count, until it is 0 or `seconds` pass; its last value."""
-    deadline = time.monotonic() + seconds
+def until(admin, sql):
+    """Polls `sql`, a count, until it is 0 or `SETTLE_SECONDS` pass; its last value."""
+    deadline = time.monotonic() + SETTLE_SECONDS
     while (value := admin.run(sql).value()) != "0" and time.monotonic() < deadline:
         time.sleep(0.2)
     return value
@@ -35,6 +38,8 @@ def until(admin, sql, seconds=SETTLE_SECONDS):
 def after_clients(admin, roles):
     """Failures, and a report of the transient bytes the transaction pool's backends keep."""
     failures, report = [], {}
+    if (left := until(admin, OUTLIVED_PROGRESS_ROWS)) != "0":
+        failures.append(f"{left} progress rows outlived their call once the clients stopped")
     for mode, role in roles.items():
         sql = (f"SELECT coalesce(sum(b.transient_bytes), 0) FROM warren.backends() b JOIN pg_stat_activity a "
                f"USING (pid) WHERE b.kind = 'backend' AND a.usename = '{role}'")

test/harness/pooler_soak/verdict.py

@@ -11,12 +11,15 @@
 replaces its backends. A backend that a pool started later has a
 warm-up of its own, as long as the run's from its first sample. Per
 cluster (`CLUSTER_TOLERANCES`): the shared
-cache's bytes and entries, the DSM segments mapped, the activity and
-progress rows, the versions (promoted and staged), segments, vectors
-and chunks, and the catalog tables' sizes. The tolerances leave room for
-what varies without leaking: version sizes as the source changes,
-staged versions in flight, the dead tuples autovacuum has not yet
-reclaimed. Other series (RSS, PSS, the pools' counters) are reported."""
+cache's bytes and entries, the DSM segments mapped, the activity rows,
+the progress rows that outlived their call (none allowed), the versions
+(promoted and staged), segments, vectors and chunks, and the catalog
+tables' sizes. The tolerances leave room for what varies without
+leaking: version sizes as the source changes, staged versions in
+flight, the dead tuples autovacuum has not yet reclaimed. Other series
+(RSS, PSS, the pools' counters, and all progress rows, which count the
+calls in flight as a sample is taken, a gauge of the load) are
+reported."""
 
 import statistics
 from collections import defaultdict
@@ -37,7 +40,7 @@
     "dsm_segments": Tolerance(4, 0.25),
     "dsm_bytes": Tolerance(16 * MIB, 0.25),
     "activity_rows": Tolerance(4, 0),
-    "progress_rows": Tolerance(2, 0),
+    "outlived_progress_rows": Tolerance(0, 0),
     "promoted_versions": Tolerance(2, 0),
     "staged_versions": Tolerance(2, 0),
     "segments": Tolerance(8, 0.25),

test/harness/pooler_soak/warm.py

@@ -0,0 +1,74 @@
+"""Warming each pool's server backends before the pooler soak's clock
+starts (DESIGN.md §13 M11), so that the steady state begins at every
+backend's working set.
+
+A backend's anonymous memory is a high-water mark: glibc keeps what a
+call's peak took, so it rises in steps the first few times the backend
+runs each kind of call (a refresh about 4 MB, then less) and then stays
+flat. Only one client in a pool writes, and the pooler decides which
+backend serves it (session pooling keeps it on one backend for its
+connection, transaction pooling hands out its last released connection
+first), so without this a backend may first serve a write well into the
+steady window, and its warm-up would look like growth. `warm` opens as
+many sessions as the pool has server connections, each pinned to its own
+backend by an open transaction, and runs every kind of call on each,
+`PASSES` times, before the transaction commits."""
+
+import random
+
+from .fixture import edit_sql
+from .workload import READS, traversal_sql
+
+PASSES = 3
+
+
+def calls(rng, nodes, name):
+    """One pass of every kind of call the soak's clients make, in one
+    transaction; `name` is the persisted vector it writes and drops."""
+    graph_reads = [traversal_sql(graph, kind, rng.randint(1, nodes), rng.randint(1, nodes))
+                   for graph in ("g", "t") for kind in READS]
+    current = "(SELECT (warren.info('g') ->> 'current_version')::bigint)"
+    return [edit_sql(rng, nodes), "SELECT warren.refresh('g')", "SELECT warren.refresh('t')",
+            "SELECT warren.drop_version('g', warren.refresh('g', promote := false))", "SELECT warren.compact('g')",
+            *graph_reads,
+            f"SELECT warren.pagerank('g', '{name}', max_iter := 10, version := {current})",
+            f"SELECT warren.vector_reduce('g', '{name}', 'sum', {current})",
+            f"SELECT warren.drop_vector('g', '{name}', {current})",
+            f"SELECT warren.vector_const('g', '{name}_x', 1, persist := false, version := {current})",
+            f"SELECT warren.spmv('g', '{name}_x', '{name}_y', weighted := true, persist := false, "
+            f"version := {current})",
+            *(f"SELECT warren.drop_vector('g', '{name}_{v}', {current})" for v in "xy"),
+            "SELECT (SELECT count(*) FROM warren.cache()), (SELECT count(*) FROM warren.backends()), "
+            "(SELECT count(*) FROM warren.progress()), (SELECT count(*) FROM warren.shared_cache())"]
+
+
+def warm(connect, pool_size, nodes, seed):
+    """Runs every kind of call `PASSES` times on each of a pool's
+    `pool_size` server backends, through sessions `connect()` opens; the
+    pids of the backends warmed."""
+    rng = random.Random(f"{seed}-warm")
+    sessions = []
+    try:
+        pids = set()
+        for _ in range(pool_size):
+            sessions.append(connect())
+            _run(sessions[-1], "BEGIN")
+            pids.add(_run(sessions[-1], "SELECT pg_backend_pid()").value())
+        if len(pids) != pool_size:
+            raise RuntimeError(f"warming: {len(pids)} server backends for {pool_size} sessions")
+        for number, session in enumerate(sessions):
+            for _ in range(PASSES):
+                for sql in calls(rng, nodes, f"warm{number}"):
+                    _run(session, sql)
+            _run(session, "COMMIT")
+        return pids
+    finally:
+        for session in sessions:
+            session.close()
+
+
+def _run(session, sql):
+    done = session.run(sql)
+    if not done.ok:
+        raise RuntimeError(f"warming: {sql[:200]}: {done.sqlstate} {done.message}")
+    return done

test/pooler_soak.py

@@ -6,7 +6,9 @@
 mode (default: transaction and session at once, each its own login role
 and pool of --pool-size server connections for --clients client
 connections). It first probes what each pool mode keeps of a client's
-session state (harness/pooler_soak/probes.py), then runs the clients'
+session state (harness/pooler_soak/probes.py), warms every server
+connection of each pool with every kind of call
+(harness/pooler_soak/warm.py), then runs the clients'
 mix of reads, compute, persisted vectors, introspection, timeouts,
 cancels, disconnects in transactions and writes that retire versions
 (harness/pooler_soak/workload.py) for --minutes or --hours, sampling
@@ -35,12 +37,13 @@
 import threading
 import time
 from contextlib import ExitStack, nullcontext
+from functools import partial
 from pathlib import Path
 
 from harness import extension, server_log, shared_cache
 from harness.cluster import Cluster
 from harness.pooler import MODES, PGBOUNCER, Pooler
-from harness.pooler_soak import probes, report, settled, verdict
+from harness.pooler_soak import probes, report, settled, verdict, warm
 from harness.pooler_soak.fixture import DATABASE, GRAPHS, role, setup_sql
 from harness.pooler_soak.samples import Sampler
 from harness.pooler_soak.workload import Chaos, Tally, Workload
@@ -147,6 +150,8 @@ def soak(args, out):
                                 roles[mode]) for i, mode in enumerate(args.modes)}
                 sampler = Sampler(cluster, admin, pools, args.context_every)
                 dsm_before = sampler.dsm_count()
+                for pooler, role_name in pools.values():
+                    warm.warm(partial(Session, pooler, DATABASE, user=role_name), args.pool_size, args.nodes, args.seed)
                 rows = run_clients(args, cluster, pools, sampler, tally, out)
                 failures, kept = settled.after_clients(admin, roles)
             failures += settled.after_poolers(admin, roles, args.clients * len(args.modes), dsm_before,

test/tests/test_pooler_soak_settled.py

@@ -0,0 +1,23 @@
+"""What the pooler soak checks once its clients have stopped."""
+
+from harness.pooler_soak.samples import OUTLIVED_PROGRESS_ROWS
+from harness.pooler_soak.settled import after_clients
+from harness.session import Outcome
+
+
+class Answers:
+    """Answers each statement with the value of the first key it contains, else 0."""
+
+    def __init__(self, values):
+        self.values = values
+
+    def run(self, sql):
+        value = next((value for key, value in self.values.items() if key in sql), "0")
+        return Outcome([[value]], "00000", "")
+
+
+def test_a_progress_row_left_once_the_clients_have_stopped_fails(monkeypatch):
+    monkeypatch.setattr("harness.pooler_soak.settled.SETTLE_SECONDS", 0)
+    failures, _ = after_clients(Answers({OUTLIVED_PROGRESS_ROWS: "2"}), {"transaction": "soak_transaction"})
+    assert failures == ["2 progress rows outlived their call once the clients stopped"]
+    assert after_clients(Answers({}), {"transaction": "soak_transaction"}) == ([], {"transaction": 0})

test/tests/test_pooler_soak_verdict.py

@@ -67,3 +67,22 @@ def test_a_backend_born_late_has_its_own_warm_up():
     judged = {(j.metric, j.subject): j.growth for j in judge_rows(rows, warmup=60)}
     assert judged[("anonymous", "transaction/9")].points == 34, "its first 60 s are its warm-up"
     assert not judged[("anonymous", "transaction/9")].grew
+
+
+def test_calls_in_flight_are_reported_not_judged():
+    """`progress_rows` counts the calls running as each sample is taken: a
+    gauge of the load, around 5 calls in the CI soak, whose robust slope
+    over 18 samples exceeded a bound of 2 in up to one run in twenty
+    with no trend at all (CI run 40390992808: 2.6 over 85 s at a level of
+    5, both halves over 2)."""
+    counts = [2, 5, 3, 6, 2, 4, 5, 4, 6, 3, 6, 5, 7, 4, 6, 5, 9, 8]
+    rows = [(30 + 5 * i, "progress_rows", "cluster", count) for i, count in enumerate(counts)]
+    assert [j.metric for j in judge_rows(rows, warmup=30)] == []
+
+
+def test_progress_rows_that_outlive_their_call_grow_from_the_first():
+    left = [(t, "outlived_progress_rows", "cluster", t // 100) for t in range(0, 600, 10)]
+    none = [(t, "outlived_progress_rows", "cluster", 0) for t in range(0, 600, 10)]
+    assert ("outlived_progress_rows", "cluster") in names(judge_rows(left, warmup=60))
+    assert names(judge_rows(none, warmup=60)) == set()
+    assert CLUSTER_TOLERANCES["outlived_progress_rows"].allowed(0) == 0

test/tests/test_pooler_soak_warm.py

@@ -0,0 +1,50 @@
+"""How the pooler soak warms each pool's server backends before its clock starts."""
+
+import itertools
+
+import pytest
+
+from harness.pooler_soak.warm import PASSES, warm
+from harness.session import Outcome
+
+
+class Backend:
+    """A pooled session pinned to one server backend by its open transaction."""
+
+    pids = itertools.count(100)
+
+    def __init__(self, distinct=True):
+        self.pid = str(next(self.pids)) if distinct else "99"
+        self.ran, self.closed = [], False
+
+    def run(self, sql):
+        self.ran.append(sql)
+        value = self.pid if sql == "SELECT pg_backend_pid()" else "7"
+        return Outcome([[value]], "00000", "")
+
+    def close(self):
+        self.closed = True
+
+
+def test_every_server_backend_runs_every_kind_of_call_in_its_own_transaction():
+    backends = []
+
+    def connect():
+        backends.append(Backend())
+        return backends[-1]
+
+    warmed = warm(connect, 3, nodes=100, seed=1)
+    assert warmed == {b.pid for b in backends} and len(backends) == 3
+    for backend in backends:
+        assert backend.ran[:2] == ["BEGIN", "SELECT pg_backend_pid()"] and backend.ran[-1] == "COMMIT"
+        calls = " ".join(backend.ran)
+        for call in ("warren.refresh('g')", "warren.refresh('t')", "promote := false", "warren.drop_version",
+                     "warren.compact('g')", "warren.pagerank", "warren.spmv", "warren.progress()"):
+            assert calls.count(call) == PASSES, call
+        assert calls.count("warren.bfs('g'") == calls.count("warren.bfs('t'") == PASSES
+        assert backend.closed
+
+
+def test_a_pool_that_does_not_give_each_session_its_own_backend_is_refused():
+    with pytest.raises(RuntimeError, match="2 server backends for 3 sessions"):
+        warm(iter([Backend(), Backend(distinct=False), Backend(distinct=False)]).__next__, 3, nodes=100, seed=1)