Wal flush

bench/walrcvflusher/scripts/cascade_setup.sh

@@ -0,0 +1,88 @@
+#!/bin/bash
+# cascade_setup.sh — initialize 3-node cascade: primary → replica1 → replica2
+# Usage:  ./cascade_setup.sh [throttle]
+
+source "$(dirname "$0")/common.sh"
+
+mkdir -p "$WORK_DIR" "$RESULTS_DIR"
+stop_all
+rm -rf "$PRIMARY_DATA" "$STANDBY_DATA" "$CASC1_DATA" "$CASC2_DATA"
+mkdir -p "$PRIMARY_DATA" "$CASC1_DATA" "$CASC2_DATA"
+
+# ── Primary ───────────────────────────────────────────────────────────────────
+log "Initializing primary..."
+initdb -D "$PRIMARY_DATA" --auth=trust --wal-segsize="$WAL_SEG_SIZE" 2>&1 | tail -1
+
+cat >> "$PRIMARY_DATA/postgresql.conf" <<EOF
+$(common_conf)
+port = $PRIMARY_PORT
+wal_keep_size = 4GB
+wal_sender_timeout = 60s
+synchronous_commit = off
+synchronous_standby_names = ''
+EOF
+
+chmod 0700 "$PRIMARY_DATA"
+pg_ctl -D "$PRIMARY_DATA" -l "$WORK_DIR/primary.log" start -w
+psql_p "CREATE ROLE repl WITH REPLICATION LOGIN;" 2>/dev/null || true
+psql_p "SELECT pg_create_physical_replication_slot('$SLOT_CASC1');" 2>/dev/null || true
+
+# ── Replica1 (samurai) — standby of primary, walsender for replica2 ────────────
+log "Taking basebackup for cascade replica1..."
+pg_basebackup -h localhost -p "$PRIMARY_PORT" -U repl \
+    -D "$CASC1_DATA" -X stream -S "$SLOT_CASC1" -c fast --progress 2>&1 | tail -1
+
+cat >> "$CASC1_DATA/postgresql.conf" <<EOF
+$(common_conf)
+port = $CASC1_PORT
+hot_standby = on
+hot_standby_feedback = on
+wal_receiver_status_interval = 1s
+wal_receiver_timeout = 60s
+EOF
+
+chmod 0700 "$CASC1_DATA"
+
+touch "$CASC1_DATA/standby.signal"
+cat > "$CASC1_DATA/postgresql.auto.conf" <<EOF
+primary_conninfo = 'host=localhost port=$PRIMARY_PORT user=repl application_name=casc1'
+primary_slot_name = '$SLOT_CASC1'
+EOF
+
+# Start replica1 (optionally with throttle)
+if [ "${1:-}" = "throttle" ]; then
+    start_node "$CASC1_DATA" "$WORK_DIR/casc1.log" throttle
+else
+    start_node "$CASC1_DATA" "$WORK_DIR/casc1.log"
+fi
+wait_for_streaming "$PRIMARY_PORT" "casc1"
+
+# Create slot for replica2 on replica1
+psql_c1 "SELECT pg_create_physical_replication_slot('$SLOT_CASC2');" 2>/dev/null || true
+
+# ── Replica2 (stubble) — standby of replica1 ──────────────────────────────────
+log "Taking basebackup for cascade replica2..."
+pg_basebackup -h localhost -p "$CASC1_PORT" -U repl \
+    -D "$CASC2_DATA" -X stream -S "$SLOT_CASC2" -c fast --progress 2>&1 | tail -1
+
+cat >> "$CASC2_DATA/postgresql.conf" <<EOF
+$(common_conf)
+port = $CASC2_PORT
+hot_standby = on
+hot_standby_feedback = on
+wal_receiver_status_interval = 1s
+wal_receiver_timeout = 60s
+EOF
+
+chmod 0700 "$CASC2_DATA"
+
+touch "$CASC2_DATA/standby.signal"
+cat > "$CASC2_DATA/postgresql.auto.conf" <<EOF
+primary_conninfo = 'host=localhost port=$CASC1_PORT user=repl application_name=casc2'
+primary_slot_name = '$SLOT_CASC2'
+EOF
+
+start_node "$CASC2_DATA" "$WORK_DIR/casc2.log"
+wait_for_streaming "$CASC1_PORT" "casc2"
+
+log "Cascade setup complete: primary:$PRIMARY_PORT → casc1:$CASC1_PORT → casc2:$CASC2_PORT"

bench/walrcvflusher/scripts/collect_metrics.sh

@@ -0,0 +1,83 @@
+#!/bin/bash
+# collect_metrics.sh — standalone metric collection (run separately from bench)
+# Usage:  ./collect_metrics.sh <label> <duration_sec>
+# Collects pg_stat_io, wait events, lag, and fsync timing from standby
+
+source "$(dirname "$0")/common.sh"
+
+LABEL="${1:?Usage: collect_metrics.sh <label> <duration_sec>}"
+DURATION="${2:-60}"
+OUT="$RESULTS_DIR/${LABEL}"
+mkdir -p "$OUT"
+
+log "Collecting metrics for ${DURATION}s..."
+
+# ── pg_stat_io (full snapshot) ────────────────────────────────────────────────
+psql -p "$STANDBY_PORT" -d postgres -c "
+SELECT backend_type, object, context,
+       reads, writes, writebacks, fsyncs, extends,
+       op_bytes, read_bytes, write_bytes, writeback_bytes,
+       reads_time, writes_time, writebacks_time, fsyncs_time,
+       stats_reset
+FROM pg_stat_io
+ORDER BY backend_type, object, context;" > "$OUT/pg_stat_io_full.txt" 2>&1
+
+# ── Wait events over duration ─────────────────────────────────────────────────
+log "Sampling wait events for ${DURATION}s..."
+( for i in $(seq 1 $((DURATION / 2))); do
+    psql -p "$STANDBY_PORT" -d postgres -At -c "
+      SELECT clock_timestamp()||','||pid||','||backend_type||','||
+             coalesce(wait_event_type,'')||','||coalesce(wait_event,'')
+      FROM pg_stat_activity
+      WHERE backend_type IN ('walreceiver','walrcvflusher','startup')"
+    sleep 2
+  done ) > "$OUT/wait_events_timeseries.csv" 2>/dev/null
+
+# ── Wait event summary ────────────────────────────────────────────────────────
+psql -p "$STANDBY_PORT" -d postgres -c "
+SELECT backend_type, wait_event_type, wait_event, count(*)
+FROM pg_stat_activity
+WHERE backend_type IN ('walreceiver','walrcvflusher','startup')
+GROUP BY 1,2,3 ORDER BY 1,4 DESC;" > "$OUT/wait_events_summary.txt" 2>&1
+
+# ── fsync timing per backend type ─────────────────────────────────────────────
+psql -p "$STANDBY_PORT" -d postgres -c "
+SELECT backend_type,
+       fsyncs,
+       round(fsyncs_time::numeric / greatest(fsyncs,1), 3) AS avg_fsync_ms,
+       fsyncs_time AS total_fsync_ms,
+       writes,
+       round(writes_time::numeric / greatest(writes,1), 3) AS avg_write_ms
+FROM pg_stat_io
+WHERE object = 'wal'
+  AND backend_type IN ('walreceiver','walrcvflusher','walsender')
+ORDER BY backend_type;" > "$OUT/fsync_timing.txt" 2>&1
+
+# ── Lag snapshot ──────────────────────────────────────────────────────────────
+psql -p "$PRIMARY_PORT" -d postgres -c "
+SELECT application_name, state, sync_state,
+       sent_lsn, write_lsn, flush_lsn, replay_lsn,
+       write_lag, flush_lag, replay_lag,
+       pg_current_wal_lsn() AS current_lsn
+FROM pg_stat_replication;" > "$OUT/pg_stat_replication.txt" 2>&1
+
+# ── WAL receiver status ───────────────────────────────────────────────────────
+psql_s "SELECT status, receive_start_lsn, receive_start_tli,
+        written_lsn, flushed_lsn, received_lsn,
+        latest_end_lsn, latest_end_time,
+        conninfo
+FROM pg_stat_wal_receiver;" > "$OUT/wal_receiver.txt" 2>&1
+
+# ── pg_test_fsync characterization ────────────────────────────────────────────
+if command -v pg_test_fsync &>/dev/null; then
+    log "Running pg_test_fsync (5s per test)..."
+    pg_test_fsync --secs-per-test=5 > "$OUT/pg_test_fsync.txt" 2>&1
+fi
+
+# ── pgbench log (if exists) ────────────────────────────────────────────────────
+if [ -f "$OUT/pgbench.log" ]; then
+    TPS=$(grep "tps =" "$OUT/pgbench.log" 2>/dev/null | head -1 || true)
+    log "pgbench result: $TPS"
+fi
+
+log "Metrics collected in $OUT/"

bench/walrcvflusher/scripts/common.sh

@@ -0,0 +1,152 @@
+#!/bin/bash
+# common.sh — shared configuration and helper functions for walrcvflusher benchmarks
+# Source this from other scripts: source "$(dirname "$0")/common.sh"
+
+set -euo pipefail
+
+# ── Paths ────────────────────────────────────────────────────────────────────
+BENCH_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+SCRIPTS_DIR="$BENCH_ROOT/scripts"
+SQL_DIR="$BENCH_ROOT/sql"
+RESULTS_DIR="$BENCH_ROOT/results"
+
+# PostgreSQL install prefix — adjust to your build
+PGINST="${PGINST:-$(pwd)/pginst}"
+export PATH="$PGINST/bin:$PATH"
+export PGHOST=localhost
+
+# ── Topology ports ───────────────────────────────────────────────────────────
+export PRIMARY_PORT="${PRIMARY_PORT:-55432}"
+export STANDBY_PORT="${STANDBY_PORT:-55433}"
+export CASC1_PORT="${CASC1_PORT:-55434}"   # replica1 (samurai) in cascade
+export CASC2_PORT="${CASC2_PORT:-55435}"   # replica2 (stubble) in cascade
+
+# ── Data directories ─────────────────────────────────────────────────────────
+export WORK_DIR="${WORK_DIR:-$BENCH_ROOT/work}"
+export PRIMARY_DATA="$WORK_DIR/primary"
+export STANDBY_DATA="$WORK_DIR/standby"
+export CASC1_DATA="$WORK_DIR/casc1"
+export CASC2_DATA="$WORK_DIR/casc2"
+
+# ── Benchmark parameters ─────────────────────────────────────────────────────
+export RUNS="${RUNS:-5}"            # repetitions per scenario
+export DURATION="${DURATION:-120}"   # seconds per pgbench run
+export WARMUP="${WARMUP:-30}"        # warmup seconds
+export WAL_SEG_SIZE="${WAL_SEG_SIZE:-16}"  # MB
+
+# ── fsync throttle ──────────────────────────────────────────────────────────
+export THROTTLE_SO="$WORK_DIR/fsync_throttle.so"
+export THROTTLE_MS="${THROTTLE_MS:-0}"  # 0 = no throttle
+
+# ── Replication slot names ───────────────────────────────────────────────────
+SLOT_STANDBY="standby1_slot"
+SLOT_CASC1="casc1_slot"
+SLOT_CASC2="casc2_slot"
+
+# ── Helper functions ─────────────────────────────────────────────────────────
+
+log()   { echo "[$(date '+%H:%M:%S')] $*" >&2; }
+fail()  { log "ERROR: $*"; exit 1; }
+
+# psql wrappers
+psql_p()  { local sql="$1"; shift; if [ "$sql" = "-f" ]; then psql -p "$PRIMARY_PORT"  -d postgres -At -f "$1" "${@:2}"; else psql -p "$PRIMARY_PORT"  -d postgres -At -c "$sql" "$@"; fi; }
+psql_s()  { local sql="$1"; shift; if [ "$sql" = "-f" ]; then psql -p "$STANDBY_PORT"  -d postgres -At -f "$1" "${@:2}"; else psql -p "$STANDBY_PORT"  -d postgres -At -c "$sql" "$@"; fi; }
+psql_c1() { local sql="$1"; shift; if [ "$sql" = "-f" ]; then psql -p "$CASC1_PORT"    -d postgres -At -f "$1" "${@:2}"; else psql -p "$CASC1_PORT"    -d postgres -At -c "$sql" "$@"; fi; }
+psql_c2() { local sql="$1"; shift; if [ "$sql" = "-f" ]; then psql -p "$CASC2_PORT"    -d postgres -At -f "$1" "${@:2}"; else psql -p "$CASC2_PORT"    -d postgres -At -c "$sql" "$@"; fi; }
+
+# Wrappers that accept a single SQL string via -c
+psql_pc() { psql -p "$PRIMARY_PORT"  -d postgres -At -c "$1"; }
+psql_sc() { psql -p "$STANDBY_PORT"  -d postgres -At -c "$1"; }
+
+# Start a node, optionally with LD_PRELOAD throttle.
+# When throttle is requested, start postgres directly (not via pg_ctl) so
+# LD_PRELOAD is inherited by the walreceiver/flusher child processes.
+# pg_ctl internally clears LD_PRELOAD in the forked postgres process.
+start_node() {
+    local data_dir="$1"
+    local log_file="$2"
+    local mode="${3:-}"
+
+    if [ "$mode" = "throttle" ]; then
+        log "Starting node (throttled, THROTTLE_MS=$THROTTLE_MS): $data_dir"
+        # Start postgres directly — pg_ctl clears LD_PRELOAD.
+        LD_PRELOAD="$THROTTLE_SO" THROTTLE_MS="$THROTTLE_MS" \
+            postgres -D "$data_dir" -c logging_collector=on \
+            >> "$log_file" 2>&1 &
+        # Wait for postgres to be ready (up to 60s)
+        local pid_file="$data_dir/postmaster.pid"
+        for i in $(seq 1 60); do
+            [ -f "$pid_file" ] && break
+            sleep 1
+        done
+        [ -f "$pid_file" ] || fail "Server did not start: $data_dir"
+    else
+        log "Starting node: $data_dir"
+        pg_ctl -D "$data_dir" -l "$log_file" start -w
+    fi
+}
+
+stop_node() {
+    local data_dir="$1"
+    log "Stopping node: $data_dir"
+    pg_ctl -D "$data_dir" stop -m fast -w 2>/dev/null || true
+}
+
+stop_all() {
+    stop_node "$CASC2_DATA"
+    stop_node "$CASC1_DATA"
+    stop_node "$STANDBY_DATA"
+    stop_node "$PRIMARY_DATA"
+}
+
+# Common postgresql.conf additions for benchmarking
+common_conf() {
+    cat <<'CONF'
+listen_addresses = 'localhost'
+shared_buffers = 1GB
+max_connections = 200
+max_wal_size = 16GB
+min_wal_size = 1GB
+checkpoint_timeout = 1h
+autovacuum = off
+track_io_timing = on
+track_wal_io_timing = on
+logging_collector = on
+log_min_messages = warning
+wal_level = replica
+max_wal_senders = 10
+max_replication_slots = 10
+CONF
+}
+
+# Wait for replication to be connected and streaming
+# Timeout: 180s (increased from 60s for slow-disk throttle scenarios)
+wait_for_streaming() {
+    local port="$1"
+    local app_name="$2"
+    local timeout="${3:-180}"
+    log "Waiting for standby '$app_name' to start streaming (timeout ${timeout}s)..."
+    for i in $(seq 1 "$timeout"); do
+        local state
+        state=$(psql -p "$port" -d postgres -At -c \
+            "SELECT state FROM pg_stat_replication WHERE application_name='$app_name'" 2>/dev/null || true)
+        [ "$state" = "streaming" ] && { log "Streaming connected."; return 0; }
+        sleep 1
+    done
+    fail "Standby '$app_name' did not start streaming within ${timeout}s"
+}
+
+# Reset IO stats on a node
+reset_io_stats() {
+    local port="$1"
+    psql -p "$port" -d postgres -c "SELECT pg_stat_reset_shared('io');" >/dev/null 2>&1
+}
+
+# Compute WAL receive rate (bytes/sec) from start/end LSN
+calc_recv_rate() {
+    local start_lsn="$1"
+    local end_lsn="$2"
+    local elapsed="$3"
+    psql -p "$STANDBY_PORT" -d postgres -At -c \
+        "SELECT (pg_wal_lsn_diff('$end_lsn','$start_lsn')::bigint / ${elapsed}::numeric)::bigint"
+}

bench/walrcvflusher/scripts/compare.sh

@@ -0,0 +1,58 @@
+#!/bin/bash
+# compare.sh — compare baseline vs new results from ALL.csv
+# Usage:  ./compare.sh
+# Reads results/baseline/ALL.csv and results/new/ALL.csv
+
+source "$(dirname "$0")/common.sh"
+
+BASELINE_CSV="$RESULTS_DIR/baseline/ALL.csv"
+NEW_CSV="$RESULTS_DIR/new/ALL.csv"
+
+if [ ! -f "$BASELINE_CSV" ] || [ ! -f "$NEW_CSV" ]; then
+    fail "Missing ALL.csv files. Run ./matrix.sh baseline and ./matrix.sh new first."
+fi
+
+echo "══════════════════════════════════════════════════════════════════════════"
+echo "  walrcvflusher benchmark comparison: baseline vs new"
+echo "══════════════════════════════════════════════════════════════════════════"
+printf "%-40s %15s %15s %10s %10s\n" "Scenario" "baseline B/s" "new B/s" "ratio" "verdict"
+echo "─────────────────────────────────────────────────────────────────────────"
+
+# Read scenarios from new CSV (they should match baseline)
+while IFS=, read -r label bps delta elapsed; do
+    # Extract scenario base name (strip _runN suffix)
+    base=$(echo "$label" | sed 's/_run[0-9]*//')
+
+    # Skip header
+    [ "$label" = "label" ] && continue
+
+    # Find matching baseline entry (same scenario, average across runs)
+    baseline_bps=$(grep "^${base}_run" "$BASELINE_CSV" 2>/dev/null | cut -d, -f2 | \
+                   awk '{sum+=$1; n++} END {if(n>0) printf "%.0f", sum/n}')
+    new_bps=$(grep "^${base}_run" "$NEW_CSV" 2>/dev/null | cut -d, -f2 | \
+              awk '{sum+=$1; n++} END {if(n>0) printf "%.0f", sum/n}')
+
+    if [ -z "$baseline_bps" ] || [ -z "$new_bps" ]; then
+        continue
+    fi
+
+    ratio=$(echo "scale=2; $new_bps / $baseline_bps" | bc -l)
+    verdict="OK"
+    if (( $(echo "$ratio > 1.05" | bc -l) )); then
+        verdict="WIN"
+    elif (( $(echo "$ratio < 0.95" | bc -l) )); then
+        verdict="REGRESSION"
+    fi
+
+    printf "%-40s %15s %15s %10s %10s\n" "$base" "$baseline_bps" "$new_bps" "${ratio}x" "$verdict"
+done < "$NEW_CSV" | sort -u
+
+echo "─────────────────────────────────────────────────────────────────────────"
+echo ""
+echo "pg_stat_io verification (new build):"
+echo "  Check results/new/*/pg_stat_io.txt for:"
+echo "    walreceiver:  writes>0, fsyncs=0  (writes WAL, no fsync)"
+echo "    walrcvflusher: fsyncs>0, writes=0 (fsyncs WAL, no writes)"
+echo ""
+echo "  On baseline (no flusher): walreceiver should have writes>0 AND fsyncs>0"
+echo "  and there should be NO walrcvflusher row."