Hi Hou,
Thanks for looking at this, and good to hear the sleep was already on
your list. Answers below; the script is attached.
> what you mean by "updating the hot row" or the hot-counter
> pattern (it sounds like a dependency, but I may be misreading it). I'd
> also be curious about whether the table has an index
The attached repro_parallel_apply.sh takes PGBIN (or PATH), starts a
publisher and a subscriber in a scratch directory, loads the schema and
runs the drains, printing txn/s per drain and a median/speedup summary.
Default run: ./repro_parallel_apply.sh 3
Schema is identical on both nodes; every table has a primary key and no
other index, and replica identity is the default (the PK). The tables
are counters(id int PK, version bigint) with 10,000 rows, rows(id bigint
PK, version bigint) with 4,000,000 rows, events(id bigserial PK, c int)
which grows about 15k rows per drain, and marker(id int PK).
"Hot counter" (hot<H>) is: each transaction either updates one of H rows
in counters (version = version + 1 WHERE id = random(1, H)) or inserts
one row into events, 50/50. You read it right: with small H those
UPDATEs are real dependencies. That is why I also ran "indep": the UPDATE
goes to a row of "rows" that no other transaction ever touches, plus the
same INSERTs. Nothing is shared there, and it regresses the same way: 4
workers 1,392 txn/s vs 1,432 for H=10, and the sweep over H = 1..10,000
was flat (1,506..1,396). So it is not driven by row dependencies but by
transaction size: the waits that pay the sleep are the commit-order ones,
which every transaction has.
> roughly how large the subscriber table is, since both can affect
> replay. Similarly, the configuration of the test PostgreSQL instance
> can make a real difference, so if you could share that too, it helps.
GUCs on both nodes: shared_buffers = 256MB, fsync = off, wal_level =
logical, max_logical_replication_workers = 12, max_worker_processes =
24, max_wal_size = 4GB; everything else default (the subscription keeps
its default synchronous_commit = off). Backlog from 8 pgbench clients,
VACUUM + CHECKPOINT on both nodes before each drain.
Machine: AMD Ryzen AI 9 HX 370 laptop (12 cores / 24 threads: 4 Zen 5
plus 8 Zen 5c), 30 GB RAM, -O2 build. Both nodes and pgbench on the same
host, pinned to the 8 hardware threads of the Zen 5 cores, data
directories on tmpfs, and I discarded runs where those CPUs were not
idle before and after.
> since our own earlier tests (without the pg_usleep()) didn't show a
> regression, we'd like to understand what differs in your environment.
I think my numbers agree with yours. I reran today with a build whose
only change is pg_usleep(1000L) -> pg_usleep(0L), without the logging I
used last time, and with 300k-transaction drains. Speedup with 4 workers
vs serial, first with v27's sleep and then with the sleep removed (K is
disjoint rows updated per transaction):
K=1: 0.0096x then 0.87x
K=10: 0.15x then 1.15x
K=100: 0.50x then 2.13x
K=300: 1.22x then 1.32x (noisy)
K=1000: 1.55x then 1.67x
H=10: 0.0059x then 0.74x
(K=10..300 with the sleep are from the earlier run.) Without the sleep
the crossover drops from 100-300 rows per transaction to between 1 and
10, so only single-row transactions stay below serial. Out of curiosity,
what transaction sizes did your tests use?
This also corrects my previous mail: I wrote that ~1.7x remained with no
sleep at all. That came from the instrumented build (one LOG line per
wait, which only the parallel side pays) and from 0.2 s drains, where a
start-up jitter of ~0.3 s shows up on both sides. The clean residual is
~1.15x (K=1) to ~1.35x (H=10), and I can't tell from this machine how
much of that is CPU contention: publisher, walsender, leader and four
spinning workers share 8 hardware threads.
> For performance test, we normally prefer to test on a more stable
> machine, which could be a more powerful one, so this is worth noting.
The sleep part should not depend on the machine: each wait pays at least
1 ms after its predecessor commits, and a 23.3 s drain had 22,405 of
them (22.4 s, 96% of the wall time), so a faster CPU only shortens the
other 4%. The crossover compares runs on the same machine. The part that
could change with more cores is the residual above; I can rerun the same
script on a larger instance if that would help.
I can also send the raw results and the instrumentation diff.
Regards,
Manu
#!/bin/bash
# repro_parallel_apply.sh -- backlog-drain benchmark for non-streamed parallel
apply (v27 thread)
#
# Starts a publisher and a subscriber on this host, creates the schema, loads
it, and for each
# point <kind>:<workers> times how long the subscriber needs to apply a backlog
of
# transactions with max_parallel_apply_workers_per_subscription = <workers> (0
= serial).
# Prints one DRAIN line per drain and a summary: median txn/s and speedup vs
workers=0.
#
# Usage: PGBIN=/path/to/install/bin ./repro_parallel_apply.sh [rounds]
["points"]
# rounds default 3 (median over rounds; point order is reversed in even
rounds)
# points default "hot10:0 hot10:4 indep:0 indep:4 indepK1000:0 indepK1000:4"
# PGBIN directory with initdb, pg_ctl, psql, pgbench (default: whatever is
in PATH)
# WORKDIR scratch dir for both clusters (default: mktemp -d); removed at
exit unless KEEP=1
# N transactions per drain for hot<H> / indep (default 30000)
# TOT row changes per drain for indepK<K>: TOT/K txns, at most N
(default 600000)
# CLIENTS pgbench clients generating the backlog (default 8)
# STAB 1 (default): VACUUM + CHECKPOINT on both nodes before every drain
# PORT_P / PORT_S default 56351 / 56352 (unix socket only, trust auth)
#
# Workloads (each pgbench script is one transaction):
# hot<H> 50% UPDATE counters SET version = version + 1 WHERE id =
random(1, H)
# 50% INSERT INTO events (new row)
# indep 50% UPDATE rows SET version = version + 1 WHERE id =
nextval(...): a row no
# other txn touches (nextval runs as its own read-only statement:
no xid, not
# replicated), 50% INSERT INTO events
# indepK<K> one UPDATE of K consecutive rows of "rows"; ranges are disjoint
(they start at
# a multiple of K, wrapped modulo 3,999,000), no INSERTs
#
# Method: the subscription is DISABLED and the worker count is set (ALTER
SYSTEM + reload; the
# leader only picks it up when it restarts, hence the DISABLE/ENABLE); the
backlog is committed
# on the publisher, then a marker row; ENABLE; time until the marker is visible
on the
# subscriber (commit order is preserved, so everything before it has been
applied). Polling is
# every 10 ms, so very short (serial) drains are approximate. maxPA = parallel
apply workers
# seen during the drain. Our runs pinned the whole script to 8 CPUs (taskset -c
...).
# Drains shorter than ~0.5 s (serial, or a build without the 1 ms sleep, at
N=30000) show a
# start-up jitter of ~0.3 s on either side; for those use N=300000 (and
TOT=3000000).
set -eu -o pipefail
rounds=${1:-3}
points=${2:-"hot10:0 hot10:4 indep:0 indep:4 indepK1000:0 indepK1000:4"}
N=${N:-30000}; TOT=${TOT:-600000}; CLIENTS=${CLIENTS:-8}; STAB=${STAB:-1}
PP=${PORT_P:-56351}; PS=${PORT_S:-56352}
if [ -n "${PGBIN:-}" ]; then export PATH=$PGBIN:$PATH; fi
for b in initdb pg_ctl postgres psql pgbench; do
command -v $b >/dev/null || { echo "$b not found (set PGBIN)" >&2; exit 1; }
done
R=${WORKDIR:-$(mktemp -d)}; S=$R/sock; mkdir -p "$S"
export PGOPTIONS="-c client_min_messages=warning"
q() { local p=$1; shift; psql -X -q -At -h "$S" -p "$p" -d postgres -v
ON_ERROR_STOP=1 "$@"; }
stop() {
for n in P S; do pg_ctl -D "$R/$n" -m immediate stop >/dev/null 2>&1 || true;
done
if [ "${KEEP:-0}" = 1 ]; then echo "# kept $R"; else rm -rf "$R"; fi
}
trap stop EXIT
for n in P S; do
p=$PP; [ $n = S ] && p=$PS
initdb -A trust -N -D "$R/$n" >/dev/null
cat >>"$R/$n/postgresql.conf" <<EOF
port = $p
unix_socket_directories = '$S'
listen_addresses = ''
shared_buffers = 256MB
fsync = off
wal_level = logical
max_logical_replication_workers = 12
max_worker_processes = 24
max_wal_size = 4GB
EOF
pg_ctl -D "$R/$n" -l "$R/$n.log" -w start >/dev/null
done
echo "# $(postgres --version); $(nproc) CPUs; rounds=$rounds N=$N TOT=$TOT
CLIENTS=$CLIENTS"
for p in $PP $PS; do q $p -c "
CREATE TABLE counters (id int PRIMARY KEY, version bigint NOT NULL);
CREATE TABLE rows (id bigint PRIMARY KEY, version bigint NOT NULL);
CREATE TABLE events (id bigserial PRIMARY KEY, c int);
CREATE TABLE marker (id int PRIMARY KEY)"; done
q $PP -c "
INSERT INTO counters SELECT g, 0 FROM generate_series(1, 10000) g;
INSERT INTO rows SELECT g, 0 FROM generate_series(1, 4000000) g;
CREATE SEQUENCE s_rows; CREATE SEQUENCE s_blk;
CREATE PUBLICATION pub FOR TABLE counters, rows, events, marker"
q $PS -c "CREATE SUBSCRIPTION sub CONNECTION 'host=$S port=$PP dbname=postgres'
PUBLICATION pub"
until [ "$(q $PS -c "SELECT count(*) FROM pg_subscription_rel WHERE srsubstate
<> 'r'")" = 0 ]
do sleep 0.5; done
cat >"$R/hot.pgb" <<'EOF'
\set h random(1, :hot)
UPDATE counters SET version = version + 1 WHERE id = :h;
EOF
cat >"$R/indep.pgb" <<'EOF'
SELECT nextval('s_rows') AS k \gset
UPDATE rows SET version = version + 1 WHERE id = :k;
EOF
cat >"$R/ins.pgb" <<'EOF'
\set h random(1, 10)
INSERT INTO events (c) VALUES (:h);
EOF
cat >"$R/blk.pgb" <<'EOF'
SELECT (nextval('s_blk') * :K) % 3999000 + 1 AS b \gset
UPDATE rows SET version = version + 1 WHERE id BETWEEN :b AND :b + :K - 1;
EOF
mk=0; pts=($points); out=$R/drains.txt; : >"$out"
for r in $(seq "$rounds"); do
order=("${pts[@]}")
if [ $((r % 2)) = 0 ]; then
order=(); for ((i = ${#pts[@]} - 1; i >= 0; i--)); do
order+=("${pts[$i]}"); done
fi
for pt in "${order[@]}"; do
kind=${pt%%:*}; w=${pt##*:}
q $PS -c "ALTER SUBSCRIPTION sub DISABLE"
q $PS -c "ALTER SYSTEM SET max_parallel_apply_workers_per_subscription = $w"
q $PS -c "SELECT pg_reload_conf()" >/dev/null
until [ "$(q $PP -c "SELECT count(*) FROM pg_replication_slots
WHERE slot_name = 'sub' AND active")" = 0 ]; do sleep
0.1; done
if [ "$STAB" = 1 ]; then
for p in $PP $PS; do
q $p -c "VACUUM rows" -c "VACUUM events" -c "VACUUM counters" -c
"CHECKPOINT"
done
fi
ntx=$N
case $kind in
indep) f=(-f "$R/indep.pgb" -f "$R/ins.pgb");;
indepK*) K=${kind#indepK}; ntx=$((TOT / K)); [ $ntx -gt "$N" ] && ntx=$N
f=(-D K="$K" -f "$R/blk.pgb");;
hot*) f=(-D hot="${kind#hot}" -f "$R/hot.pgb" -f "$R/ins.pgb");;
*) echo "unknown kind $kind" >&2; exit 1;;
esac
ntx=$(( (ntx / CLIENTS) * CLIENTS ))
pgbench -n -h "$S" -p $PP -c "$CLIENTS" -j "$CLIENTS" -t $((ntx / CLIENTS))
"${f[@]}" \
postgres >"$R/pgb.out" 2>&1 || { echo "pgbench failed" >&2; cat
"$R/pgb.out" >&2; exit 1; }
mk=$((mk + 1)); q $PP -c "INSERT INTO marker VALUES ($mk)"
t0=$(date +%s.%N); q $PS -c "ALTER SUBSCRIPTION sub ENABLE"; maxpa=0
while [ "$(q $PS -c "SELECT count(*) FROM marker WHERE id = $mk")" != 1 ];
do
a=$(q $PS -c "SELECT count(*) FROM pg_stat_subscription
WHERE worker_type = 'parallel apply'")
[ "$a" -gt $maxpa ] && maxpa=$a
sleep 0.01
done
t1=$(date +%s.%N)
awk -v r="$r" -v k="$kind" -v w="$w" -v a="$t0" -v b="$t1" -v n="$ntx" -v
m=$maxpa 'BEGIN {
printf "DRAIN round=%d kind=%s workers=%d txns=%d secs=%.3f
txn_per_s=%.0f maxPA=%d\n",
r, k, w, n, b - a, n / (b - a), m }' | tee -a "$out"
done
done
ck="SELECT sum(version) FROM rows UNION ALL SELECT sum(version) FROM counters
UNION ALL SELECT count(*) FROM events"
if [ "$(q $PP -c "$ck")" = "$(q $PS -c "$ck")" ]; then echo "# check: P and S
identical"
else echo "# check: P and S DIFFER"; fi
echo "# summary (median over rounds; speedup = median / median at workers=0)"
awk '
{ for (i = 1; i <= NF; i++) { split($i, kv, "="); v[kv[1]] = kv[2] }
key = v["kind"] SUBSEP v["workers"]; c[key]++; x[key, c[key]] =
v["txn_per_s"] + 0
if (!(key in seen)) { seen[key] = 1; keys[++nk] = key } }
function med(key, i, j, t, n) {
n = c[key]
for (i = 1; i <= n; i++) for (j = i + 1; j <= n; j++)
if (x[key, j] < x[key, i]) { t = x[key, i]; x[key, i] = x[key, j]; x[key,
j] = t }
return n % 2 ? x[key, (n + 1) / 2] : (x[key, n / 2] + x[key, n / 2 + 1]) /
2 }
END {
for (i = 1; i <= nk; i++) m[keys[i]] = med(keys[i])
for (i = 1; i <= nk; i++) {
split(keys[i], p, SUBSEP); base = p[1] SUBSEP 0
sp = (base in m) ? sprintf("%.4fx", m[keys[i]] / m[base]) : "n/a"
printf "%-12s workers=%-2s median=%9.0f txn/s n=%d vs serial=%s\n",
p[1], p[2], m[keys[i]], c[keys[i]], sp } }' "$out"
