public inbox for [email protected]
help / color / mirror / Atom feedFrom: Imran Zaheer <[email protected]>
To: Xuneng Zhou <[email protected]>
Cc: [email protected]
Cc: Zsolt Parragi <[email protected]>
Cc: Jakub Wartak <[email protected]>
Cc: Hayato Kuroda (Fujitsu) <[email protected]>
Cc: pgsql-hackers <[email protected]>
Subject: Re: [WIP] Pipelined Recovery
Date: Tue, 23 Jun 2026 18:27:10 +0500
Message-ID: <CA+UBfamzeXcdEbmhdOdjWn5X_YVt9n8xUpH5ZwmA7S8VWvaoXw@mail.gmail.com> (raw)
In-Reply-To: <CABPTF7XABSSwUPbnS+UE9OyeH-z3ihmdp9tOt3UJ4XcWZkE1DA@mail.gmail.com>
References: <CA+UBfa=vDV8wbmAV0pgrx-FuJh+x8YOW23vJ90Jzr=14rV+9jA@mail.gmail.com>
<OS9PR01MB12149A4E7927072A215AEED69F565A@OS9PR01MB12149.jpnprd01.prod.outlook.com>
<CA+UBfakmkdtauuRsOVXFqhFVJt0nTdEadx94tJn+qG0Pe8Wjfw@mail.gmail.com>
<CAN4CZFM7FV0VTNkujD=Mb7tNa+jkmEfnX7carvj95fY6Tp11FQ@mail.gmail.com>
<CA+UBfamW6NuuMMQTDRPDQ0a9fWN_u2OvjEF98u3CfYKTBcOZMw@mail.gmail.com>
<CA+UBfa=Dv-2tLSEKHJ0YFFH8PCTHxnX9rtVZeV8gd8q1a-GmYA@mail.gmail.com>
<CA+UBfa=PKdShpSTTTSHwXdGPZnm2rGMKPjERNOdS0SG9t9CT3Q@mail.gmail.com>
<CABPTF7WVW2x4XitXttHwCamSZcBn=Q+wLjf+M+MuEbZSAxqdDw@mail.gmail.com>
<CAAAe_zCxg2NTG_i1erLQQr8Wn+6SQ3EMOmp+N4J58Xxb21g2BQ@mail.gmail.com>
<CA+UBfa=qDfWB90w5AsmX4f3PbeeM++GbaoVagd9ff-DKQDLvWA@mail.gmail.com>
<CA+UBfakz7G5FH8PjxWyFLmF+sWdqMVcvQRRM0vURmznafqOjQQ@mail.gmail.com>
<CABPTF7XABSSwUPbnS+UE9OyeH-z3ihmdp9tOt3UJ4XcWZkE1DA@mail.gmail.com>
Hi
I am attaching the new series of patches.
What has changed?
* Rebased
* The patch set is now split into two new patches. This will make the
code easier to understand and review.
* The v4-0003 patch contains code mostly related to keeping the
recovery states synced between the startup process and the pipeline
process. Most of these changes were required to make the streaming
replication work.
* The v4-0002 patch now only contains the consumer code that handles
receiving the decoded records from the shmem queue and moving the redo
loop forward.
* The v4-0004 contains some basic tests to see if the pipeline worker
is functioning as expected. More testing was done by passing
PG_TEST_INITDB_EXTRA_OPTS="-c wal_pipeline=on" before running the
recovery test suite.
* Other than that, the cpu overhead during deserialization is
optimized by skipping multiple copies of the decoded record and
directly passing the pointer to the shmem queue. There is still some
overhead visible during serialization that could be improved at the
producer end.
* Signal handling for the pipeline worker is improved so that
promotion signals are sent to both the startup process and the
producer worker by the postmaster.
You will also find the new benchmarks attached [1] and the pdf report
overview. A simple cpu profiling on the pipelined startup process
shows that the cpu overhead during reading records has now been
removed and offloaded to the producer worker.
Before pipelining:
Around 50% of the cpu time is spent on fetching the wal record. Note that
in this workload pipeline is off so don't worry about the new func
ReceiveRecord(), it's just a wrapper around ReadRecord().
Children Self Command Shared O Symbol
- 98.85% 0.21% postgres postgres [.] PerformWalRecovery
- 98.64% PerformWalRecovery
- 51.00% ReceiveRecord
- 50.78% ReadRecord
- 50.52% XLogPrefetcherReadRecord
- 49.61% XLogPrefetcherNextBlock
+ 25.33% XLogReadAhead
+ 22.32% PrefetchSharedBuffer
+ 0.76% smgropen
- 46.68% ApplyWalRecord
+ 29.23% heap_redo
+ 9.51% heap2_redo
+ 4.74% btree_redo
+ 1.11% xlog_redo
+ 0.80% xact_redo
After Pipelining:
Here the only work needed to be done by the cpu is to get the decoded
record from
the queue. Other times (89.13%) cpu is worried about applying the wal record.
Children Self Command Shared O Symbol
- 98.74% 0.37% postgres postgres [.] PerformWalRecovery
- 98.37% PerformWalRecovery
- 89.13% ApplyWalRecord
+ 56.89% heap_redo
+ 18.28% heap2_redo
+ 8.01% btree_redo
+ 2.02% xlog_redo
+ 1.15% xact_redo
- 7.80% ReceiveRecord
+ 7.63% WalPipeline_ReceiveRecord
If the recovery process is not I/O bound then we would be able to test
this cpu optimization. Doing pgbench on a workload that is fully in
memory shows around 30% performance gains. You can see more
benchmarking details in the attached drive link [1]
Some comments related to attached pdf and benchmarking, it is showing
that we can get more performance advantage out of the pipeline when
most of the workload is running in memory i.e. we have enough shared
buffers configured.
If you want to do some experiments, please be my guest; I would be
happy to see more testing. You can share what performance advantage
you are getting from this. You can also refer to the benchmarking
script that I have been using [2].
Looking forward to your review, comments, etc.
Thanks,
Imran Zaheer
[1]: https://drive.google.com/file/d/13FATRT3kjh_y1wWETpYQh4ZXVLNaYU4A/view?usp=sharing
[2]: https://github.com/imranzaheer612/pg-recovery-testing
On Wed, Apr 22, 2026 at 2:44 PM Xuneng Zhou <[email protected]> wrote:
>
> Hi Henson, Imran,
>
> On Wed, Apr 8, 2026 at 7:14 PM Imran Zaheer <[email protected]> wrote:
> >
> > Hi
> >
> > I am uploading the new version with the following fixes
> >
> > * Rebased version.
> > * Skip serialization of decoded records. As pointed out by Henson,
> > there was no need to serialize the records again
> > for the sh_mq. We can simply pass the continuous bytes with minor
> > pointer fixing to the sh_mq
> >
> > This time I am uploading the benchmarking results to drive and
> > attaching the link here. Otherwise my mail will get holded for
> > moderation (My guess is overall attachment size is greater than 1MB thats why).
> >
> > I am still not sure whether my testing approach is good enough.
> > Because sometimes I am not able to get the same performance
> > improvement
> > with the pgbench builtin scripts as I got with the custom sql scripts.
> > Maybe pgbench is not creating enough WAL to test on
> > or maybe I am missing something.
> >
> > Benchmarks: https://drive.google.com/file/d/1Y4SYVnrFEQRE5T2r87rrTr7SWC9m19Si/view?usp=sharing
> >
> > Thanks & Regards
> > Imran Zaheer
> >
> > Imran Zaheer
> >
> > On Wed, Apr 8, 2026 at 1:46 PM Imran Zaheer <[email protected]> wrote:
> > >
> > > >
> > > > Hi Xuneng, Imran, and everyone,
> > > >
> > >
> > > Hi Henson and Xuneng.
> > >
> > > Thanks for explaining the approaches to Xuneng.
> > >
> > > >
> > > > The two approaches target different bottlenecks. The current patch
> > > > parallelizes WAL decoding, which keeps the redo path single-threaded
> > > > and avoids the Hot Standby visibility problem entirely.
> > > >
> > >
> > > You are right both approaches
> > > target different bottlenecks. Pipeline patch aims to improve overall
> > > cpu throughput
> > > and to save CPU time by offloading the steps we can safely do in parallel with
> > > out causing synchronization problems.
> > >
> > > > One thing I am curious about in the current patch: WAL records are
> > > > already in a serialized format on disk. The producer decodes them and
> > > > then re-serializes into a different custom format for shm_mq. What is
> > > > the advantage of this second serialization format over simply passing
> > > > the raw WAL bytes after CRC validation and letting the consumer decode
> > > > directly? Offloading CRC to a separate core could still improve
> > > > throughput at the cost of higher total CPU usage, without needing the
> > > > custom format.
> > > >
> > >
> > > Thanks. You are right there was no need to serialize the decoded record again.
> > > I was not aware that we already have continuous bytes in memory. In my
> > > next patch
> > > I will remove this extra serialization step.
> > >
> > > > Koichi's approach parallelizes redo (buffer I/O) itself, which attacks
> > > > a larger cost — Jakub's flamegraphs show BufferAlloc ->
> > > > GetVictimBuffer -> FlushBuffer dominating in both p0 and p1 — but at
> > > > the expense of much harder concurrency problems.
> > > >
> > > > Whether the decode pipelining ceiling is high enough, or whether the
> > > > redo parallelization complexity is tractable, seems like the central
> > > > design question for this area.
> > >
> > > I still have to investigate the problem related to `GetVictimBuffer` that
> > > Jakub mentioned. But I was trying that how can we safely offload the work done
> > > by `XLogReadBufferForRedoExtended` to a separate
> > > pipeline worker, or maybe we can try prefetching the buffer header so
> > > the main redo
> > > loop doesn't have to spend time getting the buffer
>
> Thanks for your clarification! I'll try to review this patch later.
>
> --
> Best,
> Xuneng
Attachments:
[text/x-patch] v4-0004-Pipelined-Recovery-Add-Tap-test.patch (7.3K, ../CA+UBfamzeXcdEbmhdOdjWn5X_YVt9n8xUpH5ZwmA7S8VWvaoXw@mail.gmail.com/2-v4-0004-Pipelined-Recovery-Add-Tap-test.patch)
download | inline diff:
From 10027f2e2a6dbda7f49ae3b309a5522add23f25c Mon Sep 17 00:00:00 2001
From: Imran Zaheer <[email protected]>
Date: Mon, 22 Jun 2026 12:36:45 +0500
Subject: [PATCH v4 4/4] Pipelined Recovery - Add Tap test
Some basic tap test to ensure pipeline is working as expected.
Most of the testing is done by running all the tests under the recovery suite
with pipeline on by default i.e. PG_TEST_INITDB_EXTRA_OPTS="-c wal_pipeline=on"
---
src/test/recovery/meson.build | 1 +
src/test/recovery/t/054_walpipeline.pl | 208 +++++++++++++++++++++++++
2 files changed, 209 insertions(+)
create mode 100644 src/test/recovery/t/054_walpipeline.pl
diff --git a/src/test/recovery/meson.build b/src/test/recovery/meson.build
index 9eb8ed11425..251e1822dec 100644
--- a/src/test/recovery/meson.build
+++ b/src/test/recovery/meson.build
@@ -62,6 +62,7 @@ tests += {
't/051_effective_wal_level.pl',
't/052_checkpoint_segment_missing.pl',
't/053_standby_login_event_trigger.pl',
+ 't/054_walpipeline.pl',
],
},
}
diff --git a/src/test/recovery/t/054_walpipeline.pl b/src/test/recovery/t/054_walpipeline.pl
new file mode 100644
index 00000000000..d682c878c79
--- /dev/null
+++ b/src/test/recovery/t/054_walpipeline.pl
@@ -0,0 +1,208 @@
+# Copyright (c) 2025-2026, PostgreSQL Global Development Group
+#
+# Tests for the WAL pipeline feature (wal_pipeline GUC).
+
+use strict;
+use warnings;
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+# ----------
+# Helpers
+# ----------
+
+sub slurp_log
+{
+ my ($node) = @_;
+ open(my $fh, '<', $node->logfile()) or die "Cannot open log: $!";
+ my @lines = <$fh>;
+ close($fh);
+ return @lines;
+}
+
+sub log_matches
+{
+ my ($node, $re) = @_;
+ return grep { /$re/ } slurp_log($node);
+}
+
+
+# ########################################
+# wal_pipeline = on, basic recovery
+# ########################################
+
+my $node1 = PostgreSQL::Test::Cluster->new('p1-recovery');
+$node1->init;
+$node1->start;
+
+$node1->safe_psql('postgres', q{
+ CREATE TABLE t (id serial PRIMARY KEY, v text);
+ INSERT INTO t (v)
+ SELECT md5(i::text) FROM generate_series(1,50000) i;
+});
+
+# generate more WAL
+$node1->safe_psql('postgres', q{
+ INSERT INTO t (v)
+ SELECT md5(i::text) FROM generate_series(1,50000) i;
+});
+
+# crash stop to force WAL recovery
+$node1->stop('immediate');
+
+# restart → recovery happens
+$node1->append_conf('postgresql.conf', "wal_pipeline = on");
+$node1->start;
+
+
+# Producer started
+ok(scalar log_matches($node1, qr/\[walpipeline\] producer: started at/),
+ 'producer started message found in log');
+
+# Pipeline stopped cleanly
+ok(scalar log_matches($node1, qr/\[walpipeline\] shutdown/),
+ 'pipeline stopped message found in log');
+
+# Consumer received shutdown from producer
+ok(scalar log_matches($node1, qr/\[walpipeline\] consumer: received shutdown message/),
+ 'consumer received shutdown message from producer');
+
+# sent == received
+my @exit_lines = log_matches($node1,
+ qr/\[walpipeline\] producer: exiting: sent=\d+ received=\d+/);
+ok(scalar @exit_lines >= 1, 'producer exiting line found in log');
+
+my ($sent, $recv) = $exit_lines[-1] =~ /sent=(\d+) received=(\d+)/;
+ok(defined $sent && $sent > 0, "sent count ($sent) is positive");
+ok(defined $recv && $recv > 0, "received count ($recv) is positive");
+is($sent, $recv, "no records lost in pipeline queue: sent=$sent received=$recv");
+
+# No PANIC
+ok(!(scalar log_matches($node1, qr/\bPANIC\b/)),
+ 'no PANIC messages during pipeline recovery');
+
+# Data integrity
+my $count = $node1->safe_psql('postgres', 'SELECT count(*) FROM t');
+is($count + 0, 100_000, 'all 100000 rows visible after pipeline recovery');
+
+$node1->stop;
+
+# ##############################################################
+# wal_pipeline = off (baseline, no pipeline log messages)
+# ##############################################################
+
+my $node2 = PostgreSQL::Test::Cluster->new('p0-recovery');
+$node2->init;
+$node2->start;
+
+$node2->safe_psql('postgres', q{
+ CREATE TABLE t (id serial PRIMARY KEY, v text);
+ INSERT INTO t (v)
+ SELECT md5(i::text) FROM generate_series(1,50000) i;
+});
+
+# generate more WAL
+$node2->safe_psql('postgres', q{
+ INSERT INTO t (v)
+ SELECT md5(i::text) FROM generate_series(1,50000) i;
+});
+
+# crash stop to force WAL recovery
+$node2->stop('immediate');
+
+# restart → recovery happens
+$node2->append_conf('postgresql.conf', "wal_pipeline = off");
+$node2->start;
+
+ok(!(scalar log_matches($node2, qr/\[walpipeline\] producer: started/)),
+ 'no pipeline log messages when wal_pipeline = off');
+
+my $count2 = $node2->safe_psql('postgres', 'SELECT count(*) FROM t');
+is($count2 + 0, 100_000, 'all rows present after non-pipeline recovery');
+
+$node2->stop;
+
+
+
+# ###################################################################
+# Test pipeline on vs off produce identical data (checksum comparison)
+# ###################################################################
+
+my $primary = PostgreSQL::Test::Cluster->new('primary');
+$primary->init(allows_streaming => 1);
+$primary->start;
+
+$primary->safe_psql('postgres', q{
+ CREATE TABLE t (id serial PRIMARY KEY, v text);
+ INSERT INTO t (v)
+ SELECT md5(i::text) FROM generate_series(1, 30000) i;
+});
+
+$primary->backup('backup3');
+
+$primary->safe_psql('postgres', q{
+ INSERT INTO t (v)
+ SELECT md5(i::text) FROM generate_series(1, 30000) i;
+ UPDATE t SET v = 'x' WHERE id % 10 = 0;
+});
+
+# ensure WAL boundary
+$primary->safe_psql('postgres', 'SELECT pg_switch_wal()');
+my $target_lsn = $primary->safe_psql('postgres', 'SELECT pg_current_wal_lsn()');
+
+my $replica_on = PostgreSQL::Test::Cluster->new('replica_p1');
+$replica_on->init_from_backup($primary, 'backup3',
+ has_streaming => 1);
+$replica_on->append_conf('postgresql.conf', "wal_pipeline = on\n");
+$replica_on->start;
+
+my $replica_off = PostgreSQL::Test::Cluster->new('replica_p0');
+$replica_off->init_from_backup($primary, 'backup3',
+ has_streaming => 1);
+$replica_off->append_conf('postgresql.conf', "wal_pipeline = off\n");
+$replica_off->start;
+
+# wait for replicas to catch up
+$primary->wait_for_catchup($replica_on);
+$primary->wait_for_catchup($replica_off);
+
+my $md5_on = $replica_on->safe_psql('postgres',
+ "SELECT md5(string_agg(id::text||v, ',' ORDER BY id)) FROM t");
+
+my $md5_off = $replica_off->safe_psql('postgres',
+ "SELECT md5(string_agg(id::text||v, ',' ORDER BY id)) FROM t");
+
+is($md5_on, $md5_off,
+ 'table checksum identical between pipeline=on and pipeline=off');
+
+$replica_on->stop;
+$replica_off->stop;
+$primary->stop('fast');
+
+
+
+# #################################
+# Test pipeline when no need of replay
+# #################################
+
+my $node3 = PostgreSQL::Test::Cluster->new('p1-small-replay');
+$node3->init;
+$node3->start;
+
+# crash stop to force WAL recovery
+$node3->stop('immediate');
+
+# restart → recovery happens
+$node3->append_conf('postgresql.conf', "wal_pipeline = on");
+$node3->start;
+
+ok(scalar log_matches($node3, qr/\[walpipeline\] producer: exiting: sent=0 received=0/),
+ 'pipeline producer sent zero records');
+
+ok((scalar log_matches($node3, qr/redo done at/)),
+ 'pipeline redo done even with tiny replay');
+
+$node3->stop;
+
+done_testing();
\ No newline at end of file
--
2.34.1
[text/x-patch] v4-0001-Pipelined-Recovery-Producer-Related-Code.patch (28.4K, ../CA+UBfamzeXcdEbmhdOdjWn5X_YVt9n8xUpH5ZwmA7S8VWvaoXw@mail.gmail.com/3-v4-0001-Pipelined-Recovery-Producer-Related-Code.patch)
download | inline diff:
From 5885605a6579e517f664422b763ec683464852d5 Mon Sep 17 00:00:00 2001
From: Imran Zaheer <[email protected]>
Date: Mon, 22 Jun 2026 12:39:49 +0500
Subject: [PATCH v4 1/4] Pipelined Recovery - Producer Related Code
This includes the producer specific code for the producer-consumer
architecture for WAL replay that separates WAL decoding from the startup
process, enabling parallel processing between differemt steps of replay.
The producer includes a background worker that reads and decodes WAL records,
then send them to the startup process for the redo. IPC happens via shared
memory message queues (shm_mq), allowing the decoder to run ahead of
the apply process.
This will give some performance improvements in recoveries for CPU-bound
workloads.
New GUC: wal_pipeline (default: off)
Author: Imran Zaheer <[email protected]>
Idea by: Ants Aasma <[email protected]>
---
src/backend/access/transam/Makefile | 1 +
src/backend/access/transam/meson.build | 1 +
src/backend/access/transam/xlogpipeline.c | 567 ++++++++++++++++++
src/backend/access/transam/xlogprefetcher.c | 3 +-
src/backend/access/transam/xlogrecovery.c | 8 +-
src/backend/postmaster/bgworker.c | 5 +
src/backend/postmaster/postmaster.c | 12 +-
src/backend/utils/misc/guc_parameters.dat | 15 +
src/backend/utils/misc/postgresql.conf.sample | 2 +
src/include/access/xlog.h | 2 +
src/include/access/xlogpipeline.h | 105 ++++
src/include/access/xlogrecovery.h | 4 +
src/include/storage/subsystemlist.h | 1 +
13 files changed, 720 insertions(+), 6 deletions(-)
create mode 100644 src/backend/access/transam/xlogpipeline.c
create mode 100644 src/include/access/xlogpipeline.h
diff --git a/src/backend/access/transam/Makefile b/src/backend/access/transam/Makefile
index a32f473e0a2..ba0bf343769 100644
--- a/src/backend/access/transam/Makefile
+++ b/src/backend/access/transam/Makefile
@@ -32,6 +32,7 @@ OBJS = \
xlogbackup.o \
xlogfuncs.o \
xloginsert.o \
+ xlogpipeline.o \
xlogprefetcher.o \
xlogreader.o \
xlogrecovery.o \
diff --git a/src/backend/access/transam/meson.build b/src/backend/access/transam/meson.build
index 06aadc7f315..be37b40581d 100644
--- a/src/backend/access/transam/meson.build
+++ b/src/backend/access/transam/meson.build
@@ -20,6 +20,7 @@ backend_sources += files(
'xlogbackup.c',
'xlogfuncs.c',
'xloginsert.c',
+ 'xlogpipeline.c',
'xlogprefetcher.c',
'xlogrecovery.c',
'xlogstats.c',
diff --git a/src/backend/access/transam/xlogpipeline.c b/src/backend/access/transam/xlogpipeline.c
new file mode 100644
index 00000000000..ab9e6eccc15
--- /dev/null
+++ b/src/backend/access/transam/xlogpipeline.c
@@ -0,0 +1,567 @@
+/*-------------------------------------------------------------------------
+ *
+ * xlogpipeline.c
+ * WAL replay pipeline implementation
+ *
+ * This module implements a producer-consumer pipeline for WAL replay.
+ * The producer (background worker) reads and decodes WAL records in parallel
+ * with the consumer (startup process) that applies them.
+ *
+ * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ * src/backend/access/transam/xlogpipeline.c
+ *
+ *-------------------------------------------------------------------------
+ */
+#include "postgres.h"
+
+#include <fcntl.h>
+#include <unistd.h>
+
+#include "access/heapam_xlog.h"
+#include "access/rmgr.h"
+#include "access/xlog.h"
+#include "access/xlogpipeline.h"
+#include "access/xlogprefetcher.h"
+#include "access/xlogreader.h"
+#include "access/xlogrecord.h"
+#include "access/xlogrecovery.h"
+#include "access/xlogutils.h"
+#include "access/xlog_internal.h"
+#include "miscadmin.h"
+#include "pgstat.h"
+#include "postmaster/bgworker.h"
+#include "postmaster/startup.h"
+#include "storage/bufmgr.h"
+#include "storage/dsm.h"
+#include "storage/ipc.h"
+#include "storage/lwlock.h"
+#include "storage/md.h"
+#include "storage/proc.h"
+#include "storage/procarray.h"
+#include "storage/procsignal.h"
+#include "storage/shm_mq.h"
+#include "storage/shm_toc.h"
+#include "storage/smgr.h"
+#include "storage/subsystems.h"
+#include "tcop/tcopprot.h"
+#include "utils/elog.h"
+#include "utils/memutils.h"
+#include "utils/resowner.h"
+#include "utils/rel.h"
+#include "utils/timeout.h"
+
+
+/*
+ * Convert values of GUCs measured in megabytes to bytes
+ */
+#define MBToBytes(mbvar) (mbvar * 1024 * 1024)
+
+/*
+ * Waiting for consumer before exiting gracefully.
+ */
+#define MAX_SHUTDOWN_WAIT_ITERS 1000 /* 1000 * 10ms = 10 seconds */
+
+
+/* Global shared memory control structure */
+WalPipelineShmCtl *WalPipelineShm = NULL;
+
+static void WalPipelineShmemRequest(void *arg);
+static void WalPipelineShmemInit(void *arg);
+
+const ShmemCallbacks WalPipelineShmemCallbacks = {
+ .request_fn = WalPipelineShmemRequest,
+ .init_fn = WalPipelineShmemInit,
+};
+
+XLogPrefetcher *xlogprefetcher_pipelined = NULL;
+
+/* Local state for producer */
+static dsm_segment *producer_dsm_seg = NULL;
+static shm_mq *producer_mq = NULL;
+static shm_mq_handle *producer_mq_handle = NULL;
+
+/*
+ * Flags set by interrupt handlers for later service in the redo loop.
+ */
+static volatile sig_atomic_t got_SIGHUP = false;
+static volatile sig_atomic_t promote_signaled = false;
+
+/* Signal handlers */
+static void PipelineBgwSigHupHandler(SIGNAL_ARGS);
+static void PipelineProcTriggerHandler(SIGNAL_ARGS);
+
+static void wal_pipeline_cleanup_callback(int code, Datum arg);
+static Size serialize_wal_record(XLogReaderState *record, char **buffer);
+static void cleanup_producer_resources(void);
+
+/* copied from xlogrecovery.c */
+/* Parameters passed down from ReadRecord to the XLogPageRead callback. */
+typedef struct XLogPageReadPrivate
+{
+ int emode;
+ bool fetching_ckpt; /* are we fetching a checkpoint record? */
+ bool randAccess;
+ TimeLineID replayTLI;
+} XLogPageReadPrivate;
+
+
+/*
+ * Register shared memory for WAL Pipeline
+ */
+static void
+WalPipelineShmemRequest(void *arg)
+{
+ ShmemRequestStruct(.name = "WAL Pipeline Ctl",
+ .size = sizeof(WalPipelineShmCtl),
+ .ptr = (void **) &WalPipelineShm,
+ );
+}
+
+static void
+WalPipelineShmemInit(void *arg)
+{
+ memset(WalPipelineShm, 0, sizeof(WalPipelineShmCtl));
+
+ SpinLockInit(&WalPipelineShm->mutex);
+}
+
+
+/*
+ * Producer Function.
+ * Main loop for the producer background worker.
+ */
+void
+WalPipeline_ProducerMain(Datum main_arg)
+{
+ dsm_handle handle = DatumGetUInt32(main_arg);
+ dsm_segment *seg;
+ shm_toc *toc;
+ WalPipelineParams *params;
+ XLogReaderState *xlogreader;
+ XLogPageReadPrivate *private;
+ XLogRecord *record;
+ TimeLineID replayTLI = 0;
+ bool end_of_wal = false;
+ uint64 records_sent;
+ uint64 records_received;
+
+ /*
+ * Properly accept or ignore signals the postmaster might send us.
+ */
+ pqsignal(SIGHUP, PipelineBgwSigHupHandler); /* reload config file */
+ pqsignal(SIGUSR2, PipelineProcTriggerHandler);
+
+ /* Register cleanup callback */
+ before_shmem_exit(wal_pipeline_cleanup_callback, (Datum) 0);
+
+ seg = dsm_attach(handle);
+ if (seg == NULL)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("[walpipeline] producer: could not map dynamic shared memory segment")));
+
+ toc = shm_toc_attach(PG_WAL_PIPELINE_MAGIC, dsm_segment_address(seg));
+ if (toc == NULL)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("[walpipeline] producer: bad magic number in dynamic shared memory segment")));
+
+ /* Lookup params and queue */
+ params = shm_toc_lookup(toc, 1, false);
+ producer_mq = shm_toc_lookup(toc, 2, false);
+
+ /* Set up producer side of queue */
+ producer_dsm_seg = seg;
+ shm_mq_set_sender(producer_mq, MyProc);
+ producer_mq_handle = shm_mq_attach(producer_mq, seg, NULL);
+
+ SpinLockAcquire(&WalPipelineShm->mutex);
+ WalPipelineShm->producer_pid = MyProcPid;
+ SpinLockRelease(&WalPipelineShm->mutex);
+
+ /* DSM is now attached, so safe to unblock the signals */
+ BackgroundWorkerUnblockSignals();
+
+ /* Set up WAL reading processor */
+ private = palloc0(sizeof(XLogPageReadPrivate));
+ xlogreader =
+ XLogReaderAllocate(wal_segment_size, NULL,
+ XL_ROUTINE(.page_read = &XLogPageRead,
+ .segment_open = NULL,
+ .segment_close = wal_segment_close),
+ private);
+
+ if (!xlogreader)
+ ereport(ERROR,
+ (errcode(ERRCODE_OUT_OF_MEMORY),
+ errmsg("out of memory"),
+ errdetail("Failed while allocating a WAL reading processor.")));
+ xlogreader->system_identifier = GetSystemIdentifier();
+
+ /*
+ * Set the WAL decode buffer size. This limits how far ahead we can read
+ * in the WAL.
+ */
+ XLogReaderSetDecodeBuffer(xlogreader, NULL, wal_decode_buffer_size);
+
+ /* Init some important globals before starting */
+ replayTLI = params->ReplayTLI;
+ WalPipeline_ImportRecoveryState(params);
+
+ /* Reinit the WAL prefetcher. */
+ xlogprefetcher_pipelined = XLogPrefetcherAllocate(xlogreader);
+
+
+ elog(LOG, "[walpipeline] producer: started at %X/%X, TLI %u",
+ LSN_FORMAT_ARGS(params->NextRecPtr), replayTLI);
+
+ XLogPrefetcherBeginRead(xlogprefetcher_pipelined, params->NextRecPtr);
+
+ /* Handle the signal if we were promoted before the pipeline launch */
+ promote_signaled = params->promotedBeforeLaunch;
+
+ /* Main decoding loop */
+ while (true)
+ {
+ bool shutdown_requested;
+
+ /* Check if consumer requested to stop decoding */
+ SpinLockAcquire(&WalPipelineShm->mutex);
+ shutdown_requested = WalPipelineShm->shutdown_requested;
+ SpinLockRelease(&WalPipelineShm->mutex);
+
+ if (shutdown_requested)
+ {
+ elog(DEBUG1, "[walpipeline] producer: got shutdown request from the consumer.");
+ break;
+ }
+
+ /* Read next WAL record */
+ record = ReadRecord(xlogprefetcher_pipelined, LOG, false, replayTLI);
+
+ if (record == NULL)
+ {
+ end_of_wal = true;
+ elog(DEBUG1, "[walpipeline] producer: reached end of WAL");
+ break;
+ }
+
+ /*
+ * Successfully decoded a record. Send it to the consumer.
+ */
+ if (!WalPipeline_SendRecord(xlogreader))
+ {
+ elog(WARNING, "[walpipeline] producer: failed to send record, queue full or detached");
+ break;
+ }
+
+ /* Update our position for monitoring */
+ SpinLockAcquire(&WalPipelineShm->mutex);
+ WalPipelineShm->producer_lsn = xlogreader->EndRecPtr;
+ SpinLockRelease(&WalPipelineShm->mutex);
+
+ CHECK_FOR_INTERRUPTS();
+ }
+
+
+ if (end_of_wal)
+ {
+ /* Notify consumer we need to exit as no more records found */
+ WalPipeline_SendShutdown();
+ WalPipeline_WaitForConsumerShutdownRequest();
+ }
+
+ SpinLockAcquire(&WalPipelineShm->mutex);
+ records_sent = WalPipelineShm->records_sent;
+ records_received = WalPipelineShm->records_received;
+ SpinLockRelease(&WalPipelineShm->mutex);
+
+ elog(LOG, "[walpipeline] producer: exiting: sent=" UINT64_FORMAT " received=" UINT64_FORMAT,
+ records_sent, records_received);
+
+ /* Cleanup */
+ pfree(private);
+ XLogReaderFree(xlogreader);
+ XLogPrefetcherFree(xlogprefetcher_pipelined);
+ DisownRecoveryWakeupLatch();
+ cleanup_producer_resources();
+}
+
+/*
+ * Producer Function.
+ * Send a decoded WAL record to the consumer
+ */
+bool
+WalPipeline_SendRecord(XLogReaderState *record)
+{
+ char *buffer = NULL;
+ Size msglen;
+ shm_mq_result res;
+
+
+ if (!producer_mq_handle)
+ return false;
+
+ /* Serialize the record */
+ msglen = serialize_wal_record(record, &buffer);
+
+ res = shm_mq_send(producer_mq_handle, msglen, buffer, false, true);
+
+ if (res == SHM_MQ_SUCCESS)
+ {
+ SpinLockAcquire(&WalPipelineShm->mutex);
+ WalPipelineShm->records_sent++;
+ WalPipelineShm->bytes_sent += msglen;
+ SpinLockRelease(&WalPipelineShm->mutex);
+
+ pfree(buffer);
+ return true;
+ }
+
+ if (res == SHM_MQ_DETACHED)
+ {
+ elog(PANIC, "[walpipeline] producer: consumer detached");
+ pfree(buffer);
+ return false;
+ }
+
+ /* Some other error */
+ elog(PANIC, "[walpipeline] producer: shm_mq_send failed with result %d", res);
+ pfree(buffer);
+ return false;
+}
+
+/*
+ * Producer Function.
+ * Send shutdown message to consumer
+ */
+bool
+WalPipeline_SendShutdown(void)
+{
+ WalRecordMsgHeader hdr;
+ shm_mq_result res;
+
+ if (!producer_mq_handle)
+ return false;
+
+ hdr.msg_type = WAL_MSG_SHUTDOWN;
+ hdr.endRecPtr = InvalidXLogRecPtr;
+
+ res = shm_mq_send(producer_mq_handle, sizeof(hdr), &hdr, false, true);
+ return (res == SHM_MQ_SUCCESS);
+}
+
+
+/*
+ * serialize_wal_record (Producer)
+ *
+ * Pack a WalRecordMsgHeader followed by the DecodedXLogRecord into a
+ * contiguous buffer, converting interior pointers to relative offsets.
+ *
+ * Output buffer layout:
+ * [WalRecordMsgHeader][DecodedXLogRecord + trailing data]
+ */
+static Size
+serialize_wal_record(XLogReaderState *xlogreader, char **outbuf)
+{
+ DecodedXLogRecord *dec = xlogreader->record;
+ DecodedXLogRecord *dec_copy;
+ WalRecordMsgHeader hdr;
+ Size payload_size = dec->size;
+ Size total = sizeof(WalRecordMsgHeader) + payload_size;
+ char *buf = palloc(total);
+
+ /* build header */
+ hdr.msg_type = WAL_MSG_RECORD;
+ hdr.readRecPtr = xlogreader->ReadRecPtr;
+ hdr.endRecPtr = xlogreader->EndRecPtr;
+ hdr.missingContrecPtr = xlogreader->missingContrecPtr;
+ hdr.abortedRecPtr = xlogreader->abortedRecPtr;
+ hdr.overwrittenRecPtr = xlogreader->overwrittenRecPtr;
+ hdr.decoded_size = payload_size;
+
+ memcpy(buf, &hdr, sizeof(hdr));
+ memcpy(buf + sizeof(hdr), dec, payload_size);
+
+ /*
+ * Fix up the interior pointers: main_data and each block's data/bkp_image
+ * are absolute addresses in the producer. Convert them to byte offsets
+ * from the start of the copied DecodedXLogRecord so the consumer can
+ * reconstruct them.
+ */
+ dec_copy = (DecodedXLogRecord *)(buf + sizeof(hdr));
+
+ if (dec_copy->main_data_len > 0)
+ dec_copy->main_data = (char *)((char *)dec->main_data - (char *)dec);
+
+ for (int i = 0; i <= dec_copy->max_block_id; i++)
+ {
+ DecodedBkpBlock *blk = &dec_copy->blocks[i];
+ if (!blk->in_use)
+ continue;
+ if (blk->has_data)
+ blk->data = (char *)((char *)dec->blocks[i].data - (char *)dec);
+ if (blk->has_image)
+ blk->bkp_image = (char *)((char *)dec->blocks[i].bkp_image - (char *)dec);
+ }
+
+ *outbuf = buf;
+ return total;
+}
+
+/*
+ * We need to put some assertion that only pipeline worker should be touching
+ * the specific code.
+ */
+bool AmWalPipeline(void)
+{
+ if (MyBackendType == B_BG_WORKER && MyBgworkerEntry)
+ {
+ if (strncmp(MyBgworkerEntry->bgw_name, "wal pipeline", 12) == 0)
+ return true;
+ }
+
+ return false;
+}
+
+/*
+ * Clean up producer-side resources
+ */
+static void
+cleanup_producer_resources(void)
+{
+ if (producer_mq_handle)
+ {
+ shm_mq_detach(producer_mq_handle);
+ producer_mq_handle = NULL;
+ }
+
+ if (producer_dsm_seg)
+ {
+ dsm_detach(producer_dsm_seg);
+ producer_dsm_seg = NULL;
+ }
+
+ producer_mq = NULL;
+
+ SpinLockAcquire(&WalPipelineShm->mutex);
+ WalPipelineShm->producer_pid = 0;
+ SpinLockRelease(&WalPipelineShm->mutex);
+}
+
+
+/*
+ * Cleanup callback for process exit
+ */
+static void
+wal_pipeline_cleanup_callback(int code, Datum arg)
+{
+ pid_t mypid = MyProcPid;
+ bool is_producer = false;
+
+ if (WalPipelineShm)
+ {
+ SpinLockAcquire(&WalPipelineShm->mutex);
+ is_producer = (WalPipelineShm->producer_pid == mypid);
+ SpinLockRelease(&WalPipelineShm->mutex);
+ }
+
+ if (is_producer)
+ cleanup_producer_resources();
+ else
+ cleanup_consumer_resources();
+}
+
+/* --------------------------------
+ * signal handler routines
+ * --------------------------------
+ */
+
+ /* SIGUSR2: set flag to finish recovery */
+static void
+PipelineProcTriggerHandler(SIGNAL_ARGS)
+{
+ promote_signaled = true;
+ WakeupRecovery();
+}
+
+/* SIGHUP: set flag to re-read config file at next convenient time */
+static void
+PipelineBgwSigHupHandler(SIGNAL_ARGS)
+{
+ got_SIGHUP = true;
+ WakeupRecovery();
+}
+
+/*
+ * Re-read the config file.
+ *
+ * If one of the critical walreceiver options has changed, flag xlogrecovery.c
+ * to restart it.
+ */
+static void
+PipelineRereadConfig(void)
+{
+ char *conninfo = pstrdup(PrimaryConnInfo);
+ char *slotname = pstrdup(PrimarySlotName);
+ bool tempSlot = wal_receiver_create_temp_slot;
+ bool conninfoChanged;
+ bool slotnameChanged;
+ bool tempSlotChanged = false;
+
+ ProcessConfigFile(PGC_SIGHUP);
+
+ conninfoChanged = strcmp(conninfo, PrimaryConnInfo) != 0;
+ slotnameChanged = strcmp(slotname, PrimarySlotName) != 0;
+
+ /*
+ * wal_receiver_create_temp_slot is used only when we have no slot
+ * configured. We do not need to track this change if it has no effect.
+ */
+ if (!slotnameChanged && strcmp(PrimarySlotName, "") == 0)
+ tempSlotChanged = tempSlot != wal_receiver_create_temp_slot;
+ pfree(conninfo);
+ pfree(slotname);
+
+ if (conninfoChanged || slotnameChanged || tempSlotChanged)
+ StartupRequestWalReceiverRestart();
+}
+
+bool
+IsPromoteSignaledPipeline(void)
+{
+ return promote_signaled;
+}
+
+void
+ResetPromoteSignaledPipeline(void)
+{
+ promote_signaled = false;
+}
+
+/*
+ * Process any requests or signals received recently.
+ */
+void
+ProcessPipelineBgwInterrupts(void)
+{
+
+ bool shutdown_requested;
+
+ if (got_SIGHUP)
+ {
+ got_SIGHUP = false;
+ PipelineRereadConfig();
+ }
+
+ SpinLockAcquire(&WalPipelineShm->mutex);
+ shutdown_requested = WalPipelineShm->shutdown_requested;
+ SpinLockRelease(&WalPipelineShm->mutex);
+
+ if (shutdown_requested)
+ proc_exit(0);
+
+ CHECK_FOR_INTERRUPTS();
+}
\ No newline at end of file
diff --git a/src/backend/access/transam/xlogprefetcher.c b/src/backend/access/transam/xlogprefetcher.c
index 83a3f97a57c..ac91fdf078f 100644
--- a/src/backend/access/transam/xlogprefetcher.c
+++ b/src/backend/access/transam/xlogprefetcher.c
@@ -27,6 +27,7 @@
#include "postgres.h"
+#include "access/xlogpipeline.h"
#include "access/xlogprefetcher.h"
#include "access/xlogreader.h"
#include "catalog/pg_control.h"
@@ -355,7 +356,7 @@ XLogPrefetchReconfigure(void)
static inline void
XLogPrefetchIncrement(pg_atomic_uint64 *counter)
{
- Assert(AmStartupProcess() || !IsUnderPostmaster);
+ Assert(AmStartupProcess() || AmWalPipeline() || !IsUnderPostmaster);
pg_atomic_write_u64(counter, pg_atomic_read_u64(counter) + 1);
}
diff --git a/src/backend/access/transam/xlogrecovery.c b/src/backend/access/transam/xlogrecovery.c
index 4d61795b483..6de4a75c6c2 100644
--- a/src/backend/access/transam/xlogrecovery.c
+++ b/src/backend/access/transam/xlogrecovery.c
@@ -3104,7 +3104,7 @@ ConfirmRecoveryPaused(void)
* (emode must be either PANIC, LOG). In standby mode, retries until a valid
* record is available.
*/
-static XLogRecord *
+XLogRecord *
ReadRecord(XLogPrefetcher *xlogprefetcher, int emode,
bool fetching_ckpt, TimeLineID replayTLI)
{
@@ -3112,7 +3112,7 @@ ReadRecord(XLogPrefetcher *xlogprefetcher, int emode,
XLogReaderState *xlogreader = XLogPrefetcherGetReader(xlogprefetcher);
XLogPageReadPrivate *private = (XLogPageReadPrivate *) xlogreader->private_data;
- Assert(AmStartupProcess() || !IsUnderPostmaster);
+ Assert(AmStartupProcess() || AmWalPipeline() || !IsUnderPostmaster);
/* Pass through parameters to XLogPageRead */
private->fetching_ckpt = fetching_ckpt;
@@ -3273,7 +3273,7 @@ ReadRecord(XLogPrefetcher *xlogprefetcher, int emode,
* XLogPageRead() to try fetching the record from another source, or to
* sleep and retry.
*/
-static int
+int
XLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr, int reqLen,
XLogRecPtr targetRecPtr, char *readBuf)
{
@@ -3285,7 +3285,7 @@ XLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr, int reqLen,
int r;
instr_time io_start;
- Assert(AmStartupProcess() || !IsUnderPostmaster);
+ Assert(AmStartupProcess() || AmWalPipeline()|| !IsUnderPostmaster);
XLByteToSeg(targetPagePtr, targetSegNo, wal_segment_size);
targetPageOff = XLogSegmentOffset(targetPagePtr, wal_segment_size);
diff --git a/src/backend/postmaster/bgworker.c b/src/backend/postmaster/bgworker.c
index 2e4acad4f00..24b188f1af4 100644
--- a/src/backend/postmaster/bgworker.c
+++ b/src/backend/postmaster/bgworker.c
@@ -13,6 +13,7 @@
#include "postgres.h"
#include "access/parallel.h"
+#include "access/xlogpipeline.h"
#include "commands/repack.h"
#include "libpq/pqsignal.h"
#include "miscadmin.h"
@@ -166,6 +167,10 @@ static const struct
{
.fn_name = "DataChecksumsWorkerMain",
.fn_addr = DataChecksumsWorkerMain
+ },
+ {
+ .fn_name = "WalPipeline_ProducerMain",
+ .fn_addr = WalPipeline_ProducerMain
}
};
diff --git a/src/backend/postmaster/postmaster.c b/src/backend/postmaster/postmaster.c
index 90c7c4528e8..bc0239c0cff 100644
--- a/src/backend/postmaster/postmaster.c
+++ b/src/backend/postmaster/postmaster.c
@@ -91,6 +91,7 @@
#include "access/xlog.h"
#include "access/xlog_internal.h"
+#include "access/xlogpipeline.h"
#include "access/xlogrecovery.h"
#include "common/file_perm.h"
#include "common/pg_prng.h"
@@ -3936,12 +3937,21 @@ process_pm_pmsignal(void)
CheckPromoteSignal())
{
/*
- * Tell startup process to finish recovery.
+ * Tell startup process to finish recovery. Incase pipeline is enabled
+ * also signal the pipeline worker.
*
* Leave the promote signal file in place and let the Startup process
* do the unlink.
*/
signal_child(StartupPMChild, SIGUSR2);
+
+ if (wal_pipeline_enabled && WalPipeline_IsActive())
+ {
+ pid_t pipeline_bgw = WalPipeline_GetProducerPid();
+
+ if (pipeline_bgw != InvalidPid)
+ signal_child(FindPostmasterChildByPid(pipeline_bgw), SIGUSR2);
+ }
}
}
diff --git a/src/backend/utils/misc/guc_parameters.dat b/src/backend/utils/misc/guc_parameters.dat
index afaa058b046..5055b0f4f35 100644
--- a/src/backend/utils/misc/guc_parameters.dat
+++ b/src/backend/utils/misc/guc_parameters.dat
@@ -3508,6 +3508,21 @@
boot_val => 'false',
},
+{ name => 'wal_pipeline', type => 'bool', context => 'PGC_POSTMASTER', group => 'WAL_RECOVERY',
+ short_desc => 'Use parallel workers to speedup recovery.',
+ variable => 'wal_pipeline_enabled',
+ boot_val => 'false',
+},
+
+{ name => 'wal_pipeline_queue_size', type => 'int', context => 'PGC_POSTMASTER', group => 'WAL_RECOVERY',
+ short_desc => 'Size of the shared memory queue used by the WAL pipeline.',
+ flags => 'GUC_UNIT_MB',
+ variable => 'wal_pipeline_mq_size_mb',
+ boot_val => '128',
+ min => '1',
+ max => '1024',
+},
+
{ name => 'wal_receiver_create_temp_slot', type => 'bool', context => 'PGC_SIGHUP', group => 'REPLICATION_STANDBY',
short_desc => 'Sets whether a WAL receiver should create a temporary replication slot if no permanent slot is configured.',
variable => 'wal_receiver_create_temp_slot',
diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample
index ac38cddaaf9..4dc79db1b05 100644
--- a/src/backend/utils/misc/postgresql.conf.sample
+++ b/src/backend/utils/misc/postgresql.conf.sample
@@ -288,6 +288,8 @@
#recovery_prefetch = try # prefetch pages referenced in the WAL?
#wal_decode_buffer_size = 512kB # lookahead window used for prefetching
# (change requires restart)
+#wal_pipeline = off # decode in parallel
+#wal_pipeline_queue_size = 128MB
# - Archiving -
diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h
index 4dd98624204..9979e064325 100644
--- a/src/include/access/xlog.h
+++ b/src/include/access/xlog.h
@@ -40,6 +40,8 @@ extern PGDLLIMPORT int min_wal_size_mb;
extern PGDLLIMPORT int max_wal_size_mb;
extern PGDLLIMPORT int wal_keep_size_mb;
extern PGDLLIMPORT int max_slot_wal_keep_size_mb;
+extern PGDLLIMPORT int wal_pipeline_mq_size_mb;
+extern PGDLLIMPORT bool wal_pipeline_enabled;
extern PGDLLIMPORT int XLOGbuffers;
extern PGDLLIMPORT int XLogArchiveTimeout;
extern PGDLLIMPORT int wal_retrieve_retry_interval;
diff --git a/src/include/access/xlogpipeline.h b/src/include/access/xlogpipeline.h
new file mode 100644
index 00000000000..333090632b4
--- /dev/null
+++ b/src/include/access/xlogpipeline.h
@@ -0,0 +1,105 @@
+/*-------------------------------------------------------------------------
+ *
+ * xlogpipeline.h
+ * WAL replay pipeline for parallel recovery
+ *
+ * This module implements a producer-consumer pipeline for WAL replay:
+ * - Producer: background worker that reads and decodes WAL records
+ * - Consumer: startup process: core redo loop
+ *
+ * The pipeline uses shared memory queues (shm_mq) to pass decoded WAL
+ * records from producer to consumer, enabling parallelism while
+ * maintaining sequential replay semantics.
+ *
+ * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group
+ *
+ * src/include/access/xlogpipeline.h
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef WAL_PIPELINE_H
+#define WAL_PIPELINE_H
+
+#include "access/xlogreader.h"
+#include "access/xlogrecovery.h"
+#include "access/xlogutils.h"
+#include "storage/dsm.h"
+#include "storage/shm_mq.h"
+#include "storage/spin.h"
+
+/*
+ * Magic number for shared memory TOC
+ */
+#define PG_WAL_PIPELINE_MAGIC 0x57414C50 /* "WALP" */
+
+/*
+ * Message types sent through the pipeline
+ */
+typedef enum WalMsgType
+{
+ WAL_MSG_INVALID = 0,
+ WAL_MSG_RECORD, /* Decoded WAL record */
+ WAL_MSG_SHUTDOWN, /* Graceful shutdown request */
+} WalMsgType;
+
+/* Wire header for a serialized WAL message */
+typedef struct WalRecordMsgHeader
+{
+ WalMsgType msg_type; /* WAL_MSG_RECORD etc */
+ uint32 decoded_size; /* byte length of the payload that follows */
+ XLogRecPtr readRecPtr; /* XLogReaderState->ReadRecPtr */
+ XLogRecPtr endRecPtr; /* XLogReaderState->EndRecPtr */
+ XLogRecPtr missingContrecPtr; /* XLogReaderState->missingContrecPtr */
+ XLogRecPtr abortedRecPtr; /* XLogReaderState->abortedRecPtr */
+ XLogRecPtr overwrittenRecPtr; /* XLogReaderState->overwrittenRecPtr */
+} WalRecordMsgHeader;
+
+/*
+ * Shared memory control structure for the WAL pipeline
+ */
+typedef struct WalPipelineShmCtl
+{
+ /* Lifecycle management */
+ slock_t mutex;
+ bool initialized;
+ bool shutdown_requested;
+
+ /* Producer state */
+ pid_t producer_pid;
+ XLogRecPtr producer_lsn; /* Last LSN read by producer */
+
+ /* Consumer state */
+ pid_t consumer_pid;
+ XLogRecPtr consumer_lsn; /* Last LSN recieved by consumer */
+ XLogRecPtr applied_lsn; /* Last LSN applied by consumer */
+
+ /* Queue handles */
+ dsm_handle dsm_seg_handle;
+ shm_mq_handle *producer_mq_handle;
+ shm_mq_handle *consumer_mq_handle;
+
+ /* Statistics */
+ uint64 records_sent;
+ uint64 records_received;
+ uint64 bytes_sent;
+ uint64 bytes_received;
+} WalPipelineShmCtl;
+
+/* consumer may have to compute prefetecher stats */
+extern PGDLLIMPORT XLogPrefetcher *xlogprefetcher_pipelined;
+
+
+/* Producer functions (called by background worker) */
+extern void WalPipeline_ProducerMain(Datum main_arg);
+extern bool WalPipeline_SendRecord(XLogReaderState *record);
+extern bool WalPipeline_SendShutdown(void);
+extern bool AmWalPipeline(void);
+
+
+extern void ProcessPipelineBgwInterrupts(void);
+extern bool IsPromoteSignaledPipeline(void);
+extern void ResetPromoteSignaledPipeline(void);
+/* Global shared memory pointer */
+extern WalPipelineShmCtl *WalPipelineShm;
+
+#endif /* WAL_PIPELINE_H */
\ No newline at end of file
diff --git a/src/include/access/xlogrecovery.h b/src/include/access/xlogrecovery.h
index ba7750dca0b..81ac984a904 100644
--- a/src/include/access/xlogrecovery.h
+++ b/src/include/access/xlogrecovery.h
@@ -217,6 +217,10 @@ extern void GetXLogReceiptTime(TimestampTz *rtime, bool *fromStream);
extern TimestampTz GetLatestXTime(void);
extern TimestampTz GetCurrentChunkReplayStartTime(void);
extern XLogRecPtr GetCurrentReplayRecPtr(TimeLineID *replayEndTLI);
+extern XLogRecord *ReadRecord(XLogPrefetcher *xlogprefetcher, int emode,
+ bool fetching_ckpt, TimeLineID replayTLI);
+extern int XLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr, int reqLen,
+ XLogRecPtr targetRecPtr, char *readBuf);
extern bool PromoteIsTriggered(void);
extern bool CheckPromoteSignal(void);
diff --git a/src/include/storage/subsystemlist.h b/src/include/storage/subsystemlist.h
index 9ad619080be..e9ff6de9a1a 100644
--- a/src/include/storage/subsystemlist.h
+++ b/src/include/storage/subsystemlist.h
@@ -42,6 +42,7 @@ PG_SHMEM_SUBSYSTEM(MultiXactShmemCallbacks)
PG_SHMEM_SUBSYSTEM(BufferManagerShmemCallbacks)
PG_SHMEM_SUBSYSTEM(StrategyCtlShmemCallbacks)
PG_SHMEM_SUBSYSTEM(BufTableShmemCallbacks)
+PG_SHMEM_SUBSYSTEM(WalPipelineShmemCallbacks)
/* lock manager */
PG_SHMEM_SUBSYSTEM(LockManagerShmemCallbacks)
--
2.34.1
[text/x-patch] v4-0002-Pipelined-Recovery-Consumer-Related-Code.patch (19.2K, ../CA+UBfamzeXcdEbmhdOdjWn5X_YVt9n8xUpH5ZwmA7S8VWvaoXw@mail.gmail.com/4-v4-0002-Pipelined-Recovery-Consumer-Related-Code.patch)
download | inline diff:
From 6c86a9d9ec37d8c96e23909955a4ad0e352dcea9 Mon Sep 17 00:00:00 2001
From: Imran Zaheer <[email protected]>
Date: Mon, 22 Jun 2026 12:30:30 +0500
Subject: [PATCH v4 2/4] Pipelined Recovery - Consumer Related Code
This includes the consumer-specific code for the producer-consumer
architecture for WAL replay that separates WAL decoding from the
recovery process, enabling parallel processing between different steps
of replay.
The startup process act as consumer and will initiate the pipeline before
starting the redo loop inside PerformWalRecovery() and will connect to
the shm mq to receive the decoded records directly. Subsequently,
it will apply those records and keep on fetching new decoded records in
the main redo loop.
Finally, it will receive a shutdown message from the producer
(based on the consistency checks, the producer will take care of whether
to stop decoding or not). The consumer will exit the loop and later will
ask to fully shutdown the pipeline workers in FinishWalRecovery().
Author: Imran Zaheer <[email protected]>
Idea by: Ants Aasma <[email protected]>
---
src/backend/access/transam/xlogpipeline.c | 476 ++++++++++++++++++++++
src/backend/access/transam/xlogrecovery.c | 74 +++-
src/include/access/xlogpipeline.h | 18 +
3 files changed, 566 insertions(+), 2 deletions(-)
diff --git a/src/backend/access/transam/xlogpipeline.c b/src/backend/access/transam/xlogpipeline.c
index ab9e6eccc15..c48d2768a9a 100644
--- a/src/backend/access/transam/xlogpipeline.c
+++ b/src/backend/access/transam/xlogpipeline.c
@@ -82,6 +82,11 @@ static dsm_segment *producer_dsm_seg = NULL;
static shm_mq *producer_mq = NULL;
static shm_mq_handle *producer_mq_handle = NULL;
+/* Local state for consumer */
+static dsm_segment *consumer_dsm_seg = NULL;
+static shm_mq *consumer_mq = NULL;
+static shm_mq_handle *consumer_mq_handle = NULL;
+
/*
* Flags set by interrupt handlers for later service in the redo loop.
*/
@@ -92,9 +97,13 @@ static volatile sig_atomic_t promote_signaled = false;
static void PipelineBgwSigHupHandler(SIGNAL_ARGS);
static void PipelineProcTriggerHandler(SIGNAL_ARGS);
+/* Forward declarations */
static void wal_pipeline_cleanup_callback(int code, Datum arg);
static Size serialize_wal_record(XLogReaderState *record, char **buffer);
+static DecodedXLogRecord *deserialize_wal_record(const char *buffer, Size len, XLogReaderState *startup_reader);
static void cleanup_producer_resources(void);
+static void cleanup_consumer_resources(void);
+static void WalPipeline_WaitForConsumerShutdownRequest(void);
/* copied from xlogrecovery.c */
/* Parameters passed down from ReadRecord to the XLogPageRead callback. */
@@ -127,6 +136,173 @@ WalPipelineShmemInit(void *arg)
SpinLockInit(&WalPipelineShm->mutex);
}
+/*
+ * Called by Consumer.
+ *
+ * Initialize and start the WAL pipeline. This will be called by the startup
+ * process (consumer) as a request to start the pipeline.
+ */
+void
+WalPipeline_Start(WalPipelineParams *params)
+{
+ BackgroundWorkerHandle *handle;
+ BackgroundWorker worker;
+ shm_toc_estimator e;
+ WalPipelineParams *shared_params;
+ dsm_segment *seg;
+ shm_toc *toc;
+ shm_mq *mq;
+ Size queue_size;
+ Size segsize;
+ pid_t pid;
+
+ if (wal_pipeline_mq_size_mb <= 0)
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("invalid wal_pipeline_mq_size_mb")));
+
+ queue_size = MBToBytes(wal_pipeline_mq_size_mb);
+
+ /* Set init flag */
+ SpinLockAcquire(&WalPipelineShm->mutex);
+ if (WalPipelineShm->initialized)
+ {
+ SpinLockRelease(&WalPipelineShm->mutex);
+ return;
+ }
+ WalPipelineShm->initialized = true;
+ SpinLockRelease(&WalPipelineShm->mutex);
+
+ /*
+ * Estimate how much shared memory we need.
+ *
+ * We need one key to register the location of the WalPipelineParams, and
+ * we need 1 key to track the location of the message queue.
+ */
+ shm_toc_initialize_estimator(&e);
+ shm_toc_estimate_chunk(&e, sizeof(WalPipelineParams));
+ shm_toc_estimate_chunk(&e, queue_size);
+ shm_toc_estimate_keys(&e, 2);
+ segsize = shm_toc_estimate(&e);
+
+ /* Create the shared memory segment and establish a table of contents. */
+ seg = dsm_create(segsize, 0);
+ dsm_pin_segment(seg);
+ toc = shm_toc_create(PG_WAL_PIPELINE_MAGIC, dsm_segment_address(seg),
+ segsize);
+
+ /* Setup arguments to be passed to the producer */
+ shared_params = shm_toc_allocate(toc, sizeof(WalPipelineParams));
+ shm_toc_insert(toc, 1, shared_params);
+ *shared_params = *params;
+
+ /* Setup the message queue */
+ mq = shm_mq_create(shm_toc_allocate(toc, queue_size), queue_size);
+ shm_toc_insert(toc, 2, mq);
+
+ /* update shared state */
+ SpinLockAcquire(&WalPipelineShm->mutex);
+ WalPipelineShm->dsm_seg_handle = dsm_segment_handle(seg);
+ WalPipelineShm->consumer_pid = MyProcPid;
+ SpinLockRelease(&WalPipelineShm->mutex);
+
+ /* Set up consumer side of the queue */
+ consumer_dsm_seg = seg;
+ consumer_mq = mq;
+ shm_mq_set_receiver(consumer_mq, MyProc);
+ consumer_mq_handle = shm_mq_attach(consumer_mq, seg, NULL);
+
+ /* Register background worker */
+ memset(&worker, 0, sizeof(worker));
+ worker.bgw_flags = BGWORKER_SHMEM_ACCESS;
+ worker.bgw_start_time = BgWorkerStart_PostmasterStart;
+ worker.bgw_restart_time = BGW_NEVER_RESTART;
+ sprintf(worker.bgw_library_name, "postgres");
+ sprintf(worker.bgw_function_name, "WalPipeline_ProducerMain");
+ snprintf(worker.bgw_name, BGW_MAXLEN, "wal pipeline producer");
+ snprintf(worker.bgw_type, BGW_MAXLEN, "wal pipeline producer");
+ worker.bgw_main_arg = UInt32GetDatum(dsm_segment_handle(seg));
+ worker.bgw_notify_pid = MyProcPid;
+
+ if (!RegisterDynamicBackgroundWorker(&worker, &handle))
+ goto fail;
+
+ if (WaitForBackgroundWorkerStartup(handle, &pid) != BGWH_STARTED)
+ goto fail;
+
+ /* Register cleanup callback */
+ before_shmem_exit(wal_pipeline_cleanup_callback, (Datum) 0);
+
+ ereport(LOG, (errmsg("[walpipeline] started.")));
+ return;
+
+fail:
+ SpinLockAcquire(&WalPipelineShm->mutex);
+ WalPipelineShm->initialized = false;
+ SpinLockRelease(&WalPipelineShm->mutex);
+
+ cleanup_consumer_resources();
+
+ ereport(ERROR,
+ (errcode(ERRCODE_INSUFFICIENT_RESOURCES),
+ errmsg("could not start pipeline background worker"),
+ errhint("More details may be available in the server log.")));
+}
+
+/*
+ * Request producer shutdown.
+ * This is called by the consumer when it no longer needs records.
+ */
+static void
+WalPipeline_RequestShutdown(void)
+{
+ if (!WalPipelineShm)
+ return;
+
+ SpinLockAcquire(&WalPipelineShm->mutex);
+ WalPipelineShm->shutdown_requested = true;
+ SpinLockRelease(&WalPipelineShm->mutex);
+}
+
+/*
+ * Consumer Function.
+ * Stop the WAL pipeline. This be called be the startup process
+ * (consumer). This will only be called and the time to recovery shutdown.
+ * This function will also wait until the pipeline workers
+ * are exited.
+ */
+void
+WalPipeline_Stop(void)
+{
+ if (!WalPipelineShm || !WalPipelineShm->initialized)
+ return;
+
+ /* Ask producer to stop */
+ WalPipeline_RequestShutdown();
+
+ /* Wait for producer to exit (max 10 seconds) */
+ for (int i = 0; i < 100; i++)
+ {
+ bool producer_alive;
+
+ SpinLockAcquire(&WalPipelineShm->mutex);
+ producer_alive = (WalPipelineShm->producer_pid != 0);
+ SpinLockRelease(&WalPipelineShm->mutex);
+
+ if (!producer_alive)
+ break;
+
+ pg_usleep(100000); /* 100 ms */
+ }
+
+ cleanup_consumer_resources();
+
+ SpinLockAcquire(&WalPipelineShm->mutex);
+ WalPipelineShm->initialized = false;
+ SpinLockRelease(&WalPipelineShm->mutex);
+
+ elog(LOG, "[walpipeline] shutdown");
+}
/*
* Producer Function.
@@ -353,6 +529,210 @@ WalPipeline_SendShutdown(void)
return (res == SHM_MQ_SUCCESS);
}
+/*
+ * Consumer Function.
+ * Receive and deserialize a WAL record from the producer
+ */
+DecodedXLogRecord *
+WalPipeline_ReceiveRecord(XLogReaderState *startup_reader)
+{
+ shm_mq_result res;
+ Size nbytes;
+ void *data;
+ WalRecordMsgHeader *hdr;
+ DecodedXLogRecord *record;
+
+ if (!consumer_mq_handle)
+ return NULL;
+
+ /* Receive message from queue */
+ res = shm_mq_receive(consumer_mq_handle, &nbytes, &data, false);
+
+ if (res != SHM_MQ_SUCCESS)
+ elog(ERROR, "[walpipeline] consumer: failed to receive record");
+
+ hdr = (WalRecordMsgHeader *) data;
+
+ /* Handle different message types */
+ switch (hdr->msg_type)
+ {
+ case WAL_MSG_RECORD:
+ record = deserialize_wal_record((char *) data, nbytes, startup_reader);
+
+ /* Update statistics */
+ SpinLockAcquire(&WalPipelineShm->mutex);
+ WalPipelineShm->records_received++;
+ WalPipelineShm->bytes_received += nbytes;
+ WalPipelineShm->consumer_lsn = hdr->endRecPtr;
+ SpinLockRelease(&WalPipelineShm->mutex);
+
+ return record;
+
+ case WAL_MSG_SHUTDOWN:
+ elog(LOG, "[walpipeline] consumer: received shutdown message from the producer");
+ return NULL;
+
+ default:
+ elog(PANIC, "[walpipeline] consumer: unknown message type: %d",
+ hdr->msg_type);
+ return NULL;
+ }
+}
+
+/*
+ * Consumer Function.
+ * Check if producer is still running
+ */
+bool
+WalPipeline_CheckProducerAlive(void)
+{
+ pid_t pid;
+ bool alive;
+
+ SpinLockAcquire(&WalPipelineShm->mutex);
+ pid = WalPipelineShm->producer_pid;
+ SpinLockRelease(&WalPipelineShm->mutex);
+
+ if (pid == 0)
+ return false;
+
+ alive = (kill(pid, 0) == 0);
+
+ if (!alive)
+ {
+ SpinLockAcquire(&WalPipelineShm->mutex);
+ WalPipelineShm->producer_pid = 0;
+ SpinLockRelease(&WalPipelineShm->mutex);
+ }
+
+ return alive;
+}
+
+/*
+ * Consumer Function.
+ * Check if pipeline is active
+ */
+bool
+WalPipeline_IsActive(void)
+{
+ bool active;
+
+ if (!WalPipelineShm)
+ return false;
+
+ SpinLockAcquire(&WalPipelineShm->mutex);
+ active = WalPipelineShm->initialized && !WalPipelineShm->shutdown_requested;
+ SpinLockRelease(&WalPipelineShm->mutex);
+
+ return active;
+}
+
+/*
+ * Consumer Function.
+ * Check if pid pipeline decoder worker
+ */
+pid_t
+WalPipeline_GetProducerPid(void)
+{
+ pid_t pid;
+
+ SpinLockAcquire(&WalPipelineShm->mutex);
+ pid = WalPipelineShm->producer_pid;
+ SpinLockRelease(&WalPipelineShm->mutex);
+
+ if (!(pid > 0))
+ pid = InvalidPid;
+
+ return pid;
+}
+
+/*
+ * Producer Function.
+ * Producer may can exit without waiting for the consumer, but its better to
+ * wait until consumer request shutdown. This way log messages will show
+ * no of records_sent & records_received records equal to each other.
+ */
+static void
+WalPipeline_WaitForConsumerShutdownRequest(void)
+{
+ int iters = 0;
+
+ while (true)
+ {
+ bool shutdown_requested;
+
+ SpinLockAcquire(&WalPipelineShm->mutex);
+ shutdown_requested = WalPipelineShm->shutdown_requested;
+ SpinLockRelease(&WalPipelineShm->mutex);
+
+ if (shutdown_requested)
+ break;
+
+ if (++iters >= MAX_SHUTDOWN_WAIT_ITERS)
+ {
+ elog(WARNING,
+ "[walpipeline] producer: timed out waiting for consumer "
+ "to acknowledge shutdown, exiting anyway");
+ break;
+ }
+
+ /* Allow SIGTERM / SIGHUP to interrupt the wait */
+ ProcessPipelineBgwInterrupts();
+
+ pg_usleep(10000); /* sleep 10ms */
+ }
+}
+
+/*
+ * Consumer Function.
+ * Wait unless last sent record by the pipeline is applied by the
+ * startup process.
+ */
+void
+WalPipeline_WaitForConsumerCatchup(void)
+{
+ XLogRecPtr producer_lsn;
+ XLogRecPtr consumer_lsn;
+
+ for (;;)
+ {
+ SpinLockAcquire(&WalPipelineShm->mutex);
+ producer_lsn = WalPipelineShm->producer_lsn;
+ consumer_lsn = WalPipelineShm->applied_lsn;
+ SpinLockRelease(&WalPipelineShm->mutex);
+
+ if (producer_lsn == consumer_lsn)
+ return;
+
+ CHECK_FOR_INTERRUPTS();
+
+ /* short sleep to avoid busy looping */
+ pg_usleep(50); /* 50 microseconds */
+ }
+}
+
+/*
+ * Consumer Function.
+ * Get pipeline statistics
+ */
+void
+WalPipeline_GetStats(uint64 *records_sent, uint64 *records_received,
+ XLogRecPtr *producer_lsn, XLogRecPtr *consumer_lsn)
+{
+ SpinLockAcquire(&WalPipelineShm->mutex);
+
+ if (records_sent)
+ *records_sent = WalPipelineShm->records_sent;
+ if (records_received)
+ *records_received = WalPipelineShm->records_received;
+ if (producer_lsn)
+ *producer_lsn = WalPipelineShm->producer_lsn;
+ if (consumer_lsn)
+ *consumer_lsn = WalPipelineShm->consumer_lsn;
+
+ SpinLockRelease(&WalPipelineShm->mutex);
+}
+
/*
* serialize_wal_record (Producer)
@@ -411,6 +791,76 @@ serialize_wal_record(XLogReaderState *xlogreader, char **outbuf)
return total;
}
+/*
+ * deserialize_wal_record (Consumer)
+ *
+ * Unpack a buffer produced by serialize_wal_record, restore interior
+ * offsets to pointers, and attach the record to the startup reader.
+ *
+ * Data Layout:
+ * [WalRecordMsgHeader][DecodedXLogRecord + trailing data]
+ */
+DecodedXLogRecord *
+deserialize_wal_record(const char *buf, Size len,
+ XLogReaderState *startup_reader)
+{
+ WalRecordMsgHeader hdr;
+ DecodedXLogRecord *dec;
+ const char *payload;
+
+ if (len < sizeof(hdr))
+ return NULL;
+
+ memcpy(&hdr, buf, sizeof(hdr));
+
+ if (hdr.decoded_size != len - sizeof(hdr))
+ return NULL;
+
+
+ /*
+ * We don't have to copy the data to some local buffer. The decoded blks
+ * are continous bytes in memory and hence we can just point to its location
+ * in the sh_mq
+ */
+ payload = buf + sizeof(hdr);
+ dec = (DecodedXLogRecord *) payload;
+
+ /*
+ * Restore interior pointers from offsets.
+ * Offset 0 means the original pointer was NULL.
+ */
+ if (dec->main_data_len > 0)
+ dec->main_data = (char *)dec + (ptrdiff_t)dec->main_data;
+
+ for (int i = 0; i <= dec->max_block_id; i++)
+ {
+ DecodedBkpBlock *blk = &dec->blocks[i];
+ if (!blk->in_use)
+ continue;
+ if (blk->has_data)
+ blk->data = (char *)dec + (ptrdiff_t)blk->data;
+ if (blk->has_image)
+ blk->bkp_image = (char *)dec + (ptrdiff_t)blk->bkp_image;
+ }
+
+ /* clear the queue link — it belongs to the producer's queue */
+ dec->next = NULL;
+
+ /* Attach to reader, only updating the public parameters */
+ startup_reader->record = dec;
+ startup_reader->ReadRecPtr = dec->lsn;
+ startup_reader->DecodeRecPtr = dec->lsn;
+ startup_reader->EndRecPtr = dec->next_lsn;
+ startup_reader->NextRecPtr = dec->next_lsn;
+ startup_reader->decode_queue_head = dec;
+ startup_reader->decode_queue_tail = dec;
+ startup_reader->missingContrecPtr = hdr.missingContrecPtr;
+ startup_reader->abortedRecPtr = hdr.abortedRecPtr;
+ startup_reader->overwrittenRecPtr = hdr.overwrittenRecPtr;
+
+ return dec;
+}
+
/*
* We need to put some assertion that only pipeline worker should be touching
* the specific code.
@@ -451,6 +901,32 @@ cleanup_producer_resources(void)
SpinLockRelease(&WalPipelineShm->mutex);
}
+/*
+ * Clean up consumer-side resources
+ */
+static void
+cleanup_consumer_resources(void)
+{
+ if (consumer_mq_handle)
+ {
+ shm_mq_detach(consumer_mq_handle);
+ consumer_mq_handle = NULL;
+ }
+
+ if (consumer_dsm_seg)
+ {
+ dsm_unpin_segment(dsm_segment_handle(consumer_dsm_seg));
+ dsm_detach(consumer_dsm_seg);
+ consumer_dsm_seg = NULL;
+ }
+
+ consumer_mq = NULL;
+
+ SpinLockAcquire(&WalPipelineShm->mutex);
+ WalPipelineShm->consumer_pid = 0;
+ WalPipelineShm->dsm_seg_handle = DSM_HANDLE_INVALID;
+ SpinLockRelease(&WalPipelineShm->mutex);
+}
/*
* Cleanup callback for process exit
diff --git a/src/backend/access/transam/xlogrecovery.c b/src/backend/access/transam/xlogrecovery.c
index 6de4a75c6c2..9cc28c4a717 100644
--- a/src/backend/access/transam/xlogrecovery.c
+++ b/src/backend/access/transam/xlogrecovery.c
@@ -1600,7 +1600,57 @@ ShutdownWalRecovery(void)
* it, but let's do it for the sake of tidiness.
*/
if (ArchiveRecoveryRequested)
- DisownLatch(&XLogRecoveryCtl->recoveryWakeupLatch);
+
+/*
+ * Get next record for redo.
+ *
+ * Use the pipeline if enabled for parallel decoding and receive decoded
+ * records from a shared queue, else read it directly.
+ *
+ * Parallel decoding helps us offloading some load from the CPU and hence
+ * boosting the recovery process.
+ */
+static XLogRecord *
+ReceiveRecord(XLogPrefetcher *xlogprefetcher, int emode,
+ bool fetching_ckpt, TimeLineID replayTLI,
+ XLogReaderState **localreader)
+{
+
+ XLogRecord *record = NULL;
+ XLogReaderState *reader = *localreader;
+ DecodedXLogRecord *decoded_record = NULL;
+
+ /*
+ * If pipeline not enabled read the record directly
+ */
+ if (!wal_pipeline_enabled)
+ {
+ record = ReadRecord(xlogprefetcher, emode, fetching_ckpt, replayTLI);
+ return record;
+ }
+
+ Assert (WalPipeline_IsActive());
+
+ /*
+ * Get the record from the pipeline message queue
+ */
+ decoded_record = WalPipeline_ReceiveRecord(reader);
+
+ if (decoded_record)
+ {
+ record = &decoded_record->header;
+ return record;
+ }
+ else
+ {
+ /*
+ * We will end up here only when pipeline couldn't read more
+ * records and have sent a shutdown msg. We will acknowldge this
+ * and will trigger request to stop more pipelined decoding.
+ */
+ WalPipeline_Stop();
+ return NULL;
+ }
}
/*
@@ -1615,6 +1665,12 @@ PerformWalRecovery(void)
bool reachedRecoveryTarget = false;
TimeLineID replayTLI;
+ /*
+ * standalone backend may exist in case of pg_rewind.
+ */
+ if (!IsUnderPostmaster)
+ wal_pipeline_enabled = false;
+
/*
* Initialize shared variables for tracking progress of WAL replay, as if
* we had just replayed the record before the REDO location (or the
@@ -1694,6 +1750,17 @@ PerformWalRecovery(void)
InRedo = true;
+ if(wal_pipeline_enabled)
+ {
+ /*
+ * Startup proc parameters that pipeline should also be aware of.
+ */
+ WalPipelineParams *params = palloc0(sizeof(WalPipelineParams));
+
+ params->ReplayTLI = replayTLI;
+ WalPipeline_Start(params);
+ }
+
RmgrStartup();
ereport(LOG,
@@ -1802,7 +1869,7 @@ PerformWalRecovery(void)
}
/* Else, try to fetch the next WAL record */
- record = ReadRecord(xlogprefetcher, LOG, false, replayTLI);
+ record = ReceiveRecord(xlogprefetcher, LOG, false, replayTLI, &xlogreader);
} while (record != NULL);
/*
@@ -1811,6 +1878,9 @@ PerformWalRecovery(void)
if (reachedRecoveryTarget)
{
+ if (wal_pipeline_enabled)
+ WalPipeline_Stop();
+
if (!reachedConsistency)
ereport(FATAL,
(errmsg("requested recovery stop point is before consistent recovery point")));
diff --git a/src/include/access/xlogpipeline.h b/src/include/access/xlogpipeline.h
index 333090632b4..a8995d5c1a7 100644
--- a/src/include/access/xlogpipeline.h
+++ b/src/include/access/xlogpipeline.h
@@ -88,11 +88,29 @@ typedef struct WalPipelineShmCtl
/* consumer may have to compute prefetecher stats */
extern PGDLLIMPORT XLogPrefetcher *xlogprefetcher_pipelined;
+/*
+ * Public API functions
+ */
+
+/* Start/stop the pipeline */
+extern void WalPipeline_Start(WalPipelineParams *params);
+extern void WalPipeline_Stop(void);
/* Producer functions (called by background worker) */
extern void WalPipeline_ProducerMain(Datum main_arg);
extern bool WalPipeline_SendRecord(XLogReaderState *record);
extern bool WalPipeline_SendShutdown(void);
+
+/* Consumer functions (called by startup process) */
+extern DecodedXLogRecord *WalPipeline_ReceiveRecord(XLogReaderState *startup_reader);
+extern bool WalPipeline_CheckProducerAlive(void);
+
+/* Status and monitoring */
+extern bool WalPipeline_IsActive(void);
+extern pid_t WalPipeline_GetProducerPid(void);
+extern void WalPipeline_WaitForConsumerCatchup(void);
+extern void WalPipeline_GetStats(uint64 *records_sent, uint64 *records_received,
+ XLogRecPtr *producer_lsn, XLogRecPtr *consumer_lsn);
extern bool AmWalPipeline(void);
--
2.34.1
[text/x-patch] v4-0003-Pipelined-Recovery-Decoupling-startup-and-produce.patch (38.0K, ../CA+UBfamzeXcdEbmhdOdjWn5X_YVt9n8xUpH5ZwmA7S8VWvaoXw@mail.gmail.com/5-v4-0003-Pipelined-Recovery-Decoupling-startup-and-produce.patch)
download | inline diff:
From 10414785dc5c7661da2f39fe3db78b335478dfe0 Mon Sep 17 00:00:00 2001
From: Imran Zaheer <[email protected]>
Date: Mon, 22 Jun 2026 12:33:53 +0500
Subject: [PATCH v4 3/4] Pipelined Recovery - Decoupling startup and producer
states.
Before this patch, most states/variables were statically defined
under xlogrecovery.c as they would only be needed by the startup process.
However, to implement pipelining, the producer (parallel decoder)
should also be aware of these static states. Some of these states stays
constant throughout decoding/recovery, while others may change during
recovery.
To address this, WalPipelineParams contains a set of all local states
that the startup process had before it started the pipeline. These
states are passed to the producer, which updates them accordingly
before entering the decoding loop. This is done using
WalPipeline_ExportRecoveryState() and WalPipeline_ImportRecoveryState().
WalPipeline_ExportRecoveryState --> saves all the local startup proc states
WalPipeline_ImportRecoveryState --> apply those states to the producer context
To manage the states that may keep changing during the recovery, we
added them to the shared memory in XLogRecoveryCtlData. Now whenever
the startup proc updates any of such local states the shared state
is also upated accordingly, so that the pipeline stay aware of this
update during the decoding.
---
src/backend/access/transam/xlog.c | 15 +-
src/backend/access/transam/xlogrecovery.c | 518 ++++++++++++++++++++--
src/backend/storage/ipc/standby.c | 1 +
src/include/access/xlogpipeline.h | 46 ++
src/include/access/xlogrecovery.h | 85 ++++
5 files changed, 629 insertions(+), 36 deletions(-)
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index a81912b7441..8cdc1119987 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -59,6 +59,7 @@
#include "access/xlog_internal.h"
#include "access/xlogarchive.h"
#include "access/xloginsert.h"
+#include "access/xlogpipeline.h"
#include "access/xlogreader.h"
#include "access/xlogrecovery.h"
#include "access/xlogutils.h"
@@ -6269,6 +6270,12 @@ StartupXLOG(void)
ProcArrayApplyRecoveryInfo(&running);
}
+
+ /*
+ * Finaly update the shared standby state, so that pipeline worker
+ * stay consistent with the startup process.
+ */
+ SetSharedHotStandbyState();
}
/*
@@ -8925,6 +8932,12 @@ xlog_redo(XLogReaderState *record)
running.xids = xids;
ProcArrayApplyRecoveryInfo(&running);
+
+ /*
+ * Update the shared standby state, so that pipeline worker
+ * stay consistent with the startup process.
+ */
+ SetSharedHotStandbyState();
}
/* ControlFile->checkPointCopy always tracks the latest ckpt XID */
@@ -10149,7 +10162,7 @@ GetOldestRestartPoint(XLogRecPtr *oldrecptr, TimeLineID *oldtli)
void
XLogShutdownWalRcv(void)
{
- Assert(AmStartupProcess() || !IsUnderPostmaster);
+ Assert(AmStartupProcess() || AmWalPipeline() || !IsUnderPostmaster);
ShutdownWalRcv();
ResetInstallXLogFileSegmentActive();
diff --git a/src/backend/access/transam/xlogrecovery.c b/src/backend/access/transam/xlogrecovery.c
index 9cc28c4a717..ba367ec7b63 100644
--- a/src/backend/access/transam/xlogrecovery.c
+++ b/src/backend/access/transam/xlogrecovery.c
@@ -35,6 +35,7 @@
#include "access/xact.h"
#include "access/xlog_internal.h"
#include "access/xlogarchive.h"
+#include "access/xlogpipeline.h"
#include "access/xlogprefetcher.h"
#include "access/xlogreader.h"
#include "access/xlogrecovery.h"
@@ -100,6 +101,8 @@ int recovery_min_apply_delay = 0;
char *PrimaryConnInfo = NULL;
char *PrimarySlotName = NULL;
bool wal_receiver_create_temp_slot = false;
+bool wal_pipeline_enabled = false;
+int wal_pipeline_mq_size_mb = 128;
/*
* recoveryTargetTimeLineGoal: what the user requested, if any
@@ -206,17 +209,6 @@ typedef struct XLogPageReadPrivate
/* flag to tell XLogPageRead that we have started replaying */
static bool InRedo = false;
-/*
- * Codes indicating where we got a WAL file from during recovery, or where
- * to attempt to get one.
- */
-typedef enum
-{
- XLOG_FROM_ANY = 0, /* request to read WAL from any source */
- XLOG_FROM_ARCHIVE, /* restored using restore_command */
- XLOG_FROM_PG_WAL, /* existing file in pg_wal */
- XLOG_FROM_STREAM, /* streamed from primary */
-} XLogSource;
/* human-readable names for XLogSources, for debugging output */
static const char *const xlogSourceNames[] = {"any", "archive", "pg_wal", "stream"};
@@ -365,12 +357,6 @@ static void recoveryPausesHere(bool endOfRecovery);
static bool recoveryApplyDelay(XLogReaderState *record);
static void ConfirmRecoveryPaused(void);
-static XLogRecord *ReadRecord(XLogPrefetcher *xlogprefetcher,
- int emode, bool fetching_ckpt,
- TimeLineID replayTLI);
-
-static int XLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr,
- int reqLen, XLogRecPtr targetRecPtr, char *readBuf);
static XLogPageReadResult WaitForWALToBecomeAvailable(XLogRecPtr RecPtr,
bool randAccess,
bool fetching_ckpt,
@@ -392,6 +378,7 @@ static bool HotStandbyActiveInReplay(void);
static void SetCurrentChunkStartTime(TimestampTz xtime);
static void SetLatestXTime(TimestampTz xtime);
+static void WalPipeline_ExportRecoveryState(WalPipelineParams *params);
/*
* Register shared memory for WAL recovery
@@ -412,9 +399,27 @@ XLogRecoveryShmemInit(void *arg)
SpinLockInit(&XLogRecoveryCtl->info_lck);
InitSharedLatch(&XLogRecoveryCtl->recoveryWakeupLatch);
+ InitSharedLatch(&XLogRecoveryCtl->recoveryApplyDelayLatch);
ConditionVariableInit(&XLogRecoveryCtl->recoveryNotPausedCV);
}
+/*
+ * We may not be able to share expectedTLEs list across the sharedmemory.
+ * For now just trigger the startup process (consumer) to
+ * reread the timelinehistory file whenever pipeline updates the value for
+ * expectedTLEs. So the consumer proc will update expectedTLEs locally.
+ */
+static void
+PipelineConsumerExpectedTLEsUpdateTLI(TimeLineID recoveryTargetTLI)
+{
+ if (wal_pipeline_enabled)
+ {
+ SpinLockAcquire(&XLogRecoveryCtl->info_lck);
+ XLogRecoveryCtl->expectedTLEsUpdateTLI = recoveryTargetTLI;
+ SpinLockRelease(&XLogRecoveryCtl->info_lck);
+ }
+}
+
/*
* A thin wrapper to enable StandbyMode and do other preparatory work as
* needed.
@@ -429,8 +434,12 @@ EnableStandbyMode(void)
* standby as it will always be in recovery unless promoted. We disable
* startup progress timeout in standby mode to avoid calling
* startup_progress_timeout_handler() unnecessarily.
+ *
+ * This function could also be called from the pipeline worker.
+ * But timeout could only be dsiabled by the startup process.
*/
- disable_startup_progress_timeout();
+ if (!AmWalPipeline())
+ disable_startup_progress_timeout();
}
/*
@@ -490,7 +499,10 @@ InitWalRecovery(ControlFileData *ControlFile, bool *wasShutdown_ptr,
* recovery, if required.
*/
if (ArchiveRecoveryRequested)
+ {
+ OwnLatch(&XLogRecoveryCtl->recoveryApplyDelayLatch);
OwnLatch(&XLogRecoveryCtl->recoveryWakeupLatch);
+ }
/*
* Set the WAL reading processor now, as it will be needed when reading
@@ -962,6 +974,14 @@ InitWalRecovery(ControlFileData *ControlFile, bool *wasShutdown_ptr,
minRecoveryPointTLI = 0;
}
+ /* update shared state. */
+ if (wal_pipeline_enabled)
+ {
+ SpinLockAcquire(&XLogRecoveryCtl->info_lck);
+ XLogRecoveryCtl->InArchiveRecovery = InArchiveRecovery;
+ SpinLockRelease(&XLogRecoveryCtl->info_lck);
+ }
+
/*
* Start recovery assuming that the final record isn't lost.
*/
@@ -973,6 +993,12 @@ InitWalRecovery(ControlFileData *ControlFile, bool *wasShutdown_ptr,
*haveTblspcMap_ptr = haveTblspcMap;
}
+void DisownRecoveryWakeupLatch(void)
+{
+ if (ArchiveRecoveryRequested)
+ DisownLatch(&XLogRecoveryCtl->recoveryWakeupLatch);
+}
+
/*
* See if there are any recovery signal files and if so, set state for
* recovery.
@@ -1402,6 +1428,55 @@ read_tablespace_map(List **tablespaces)
return true;
}
+static void
+ResetStatesIfPipelined(void)
+{
+ if (wal_pipeline_enabled)
+ {
+ /*
+ * Invalidate ptrs pointing the pipeline shared message queue.
+ *
+ * We do this because instead of pointing to the decode_buffer (the
+ * normal i.e. pipeline-off) reader would be pointing to the
+ * sh_mq (ring buffer) to get the next decoded record. And we cannot
+ * continue like this for future xlog read i.e. FinishWalRecovery().
+ *
+ * See under deserialize_wal_record() for more details.
+ */
+ xlogreader->record = NULL;
+ xlogreader->decode_queue_head = NULL;
+ xlogreader->decode_queue_tail = NULL;
+
+ /*
+ * Invalidate contents of internal buffer before read attempt.
+ *
+ * This is needed because we were reading from the pipelined reader
+ * so far and updating the local buffer public states accordingly.
+ * So better to invalidate the any cached state that was there
+ * before the pipeline started, and force a reread for the future
+ * xlog reads i.e. refetching the last record in FinishWalRecovery().
+ */
+ xlogreader->readLen = 0;
+ xlogreader->seg.ws_segno = 0;
+ xlogreader->segoff = 0;
+
+ /*
+ * Update the startup proc local state with the pipeline state,
+ * because pipeline was doing the main decoding and hence have the more
+ * updated information.
+ */
+ SpinLockAcquire(&XLogRecoveryCtl->info_lck);
+ InArchiveRecovery = XLogRecoveryCtl->InArchiveRecovery;
+ missingContrecPtr = XLogRecoveryCtl->missingContrecPtr;
+ abortedRecPtr = XLogRecoveryCtl->abortedRecPtr;
+
+ /* only update if found a new timeline */
+ if (recoveryTargetTLI < XLogRecoveryCtl->recoveryTargetTLI)
+ recoveryTargetTLI = XLogRecoveryCtl->recoveryTargetTLI;
+ SpinLockRelease(&XLogRecoveryCtl->info_lck);
+ }
+}
+
/*
* Finish WAL recovery.
*
@@ -1421,6 +1496,8 @@ FinishWalRecovery(void)
TimeLineID lastRecTLI;
XLogRecPtr endOfLog;
+ ResetStatesIfPipelined();
+
/*
* Kill WAL receiver, if it's still running, before we continue to write
* the startup checkpoint and aborted-contrecord records. It will trump
@@ -1478,6 +1555,7 @@ FinishWalRecovery(void)
lastRec = XLogRecoveryCtl->lastReplayedReadRecPtr;
lastRecTLI = XLogRecoveryCtl->lastReplayedTLI;
}
+
XLogPrefetcherBeginRead(xlogprefetcher, lastRec);
(void) ReadRecord(xlogprefetcher, PANIC, false, lastRecTLI);
endOfLog = xlogreader->EndRecPtr;
@@ -1600,6 +1678,17 @@ ShutdownWalRecovery(void)
* it, but let's do it for the sake of tidiness.
*/
if (ArchiveRecoveryRequested)
+ {
+ DisownLatch(&XLogRecoveryCtl->recoveryApplyDelayLatch);
+
+ /*
+ * Only disown the latch if we (startup process) were the owner
+ * i.e. pipeline disabled.
+ */
+ if (!wal_pipeline_enabled)
+ DisownLatch(&XLogRecoveryCtl->recoveryWakeupLatch);
+ }
+}
/*
* Get next record for redo.
@@ -1758,6 +1847,7 @@ PerformWalRecovery(void)
WalPipelineParams *params = palloc0(sizeof(WalPipelineParams));
params->ReplayTLI = replayTLI;
+ WalPipeline_ExportRecoveryState(params);
WalPipeline_Start(params);
}
@@ -1953,7 +2043,9 @@ static void
ApplyWalRecord(XLogReaderState *xlogreader, XLogRecord *record, TimeLineID *replayTLI)
{
ErrorContextCallback errcallback;
+ bool wakeupWalRcvr = false;
bool switchedTLI = false;
+ bool pipeline_enabled_standby = false;
/* Setup error traceback support for ereport() */
errcallback.callback = rm_redo_error_callback;
@@ -2054,6 +2146,22 @@ ApplyWalRecord(XLogReaderState *xlogreader, XLogRecord *record, TimeLineID *repl
XLogRecoveryCtl->lastReplayedReadRecPtr = xlogreader->ReadRecPtr;
XLogRecoveryCtl->lastReplayedEndRecPtr = xlogreader->EndRecPtr;
XLogRecoveryCtl->lastReplayedTLI = *replayTLI;
+
+ /*
+ * As the pipeline (producer) was running way ahead of the startup proc
+ * (consumer), see if the producer asked to wakeup the wal_reciever by
+ * updating the value of `WakeupWalRcvrRecPtr`.
+ */
+ if (XLogRecoveryCtl->WakeupWalRcvrRecPtr != InvalidXLogRecPtr &&
+ xlogreader->EndRecPtr >= XLogRecoveryCtl->WakeupWalRcvrRecPtr)
+ {
+ Assert(wal_pipeline_enabled);
+
+ XLogRecoveryCtl->WakeupWalRcvrRecPtr = InvalidXLogRecPtr;
+ wakeupWalRcvr = true;
+ }
+
+ pipeline_enabled_standby = XLogRecoveryCtl->stanbyEnabled;
SpinLockRelease(&XLogRecoveryCtl->info_lck);
/* ------
@@ -2085,8 +2193,11 @@ ApplyWalRecord(XLogReaderState *xlogreader, XLogRecord *record, TimeLineID *repl
* If rm_redo called XLogRequestWalReceiverReply, then we wake up the
* receiver so that it notices the updated lastReplayedEndRecPtr and sends
* a reply to the primary.
+ *
+ * Also wakeup in case requested by the pipeline decoder before waiting
+ * for more wal.
*/
- if (doRequestWalReceiverReply)
+ if (doRequestWalReceiverReply || wakeupWalRcvr)
{
doRequestWalReceiverReply = false;
WalRcvRequestApplyReply();
@@ -2107,6 +2218,16 @@ ApplyWalRecord(XLogReaderState *xlogreader, XLogRecord *record, TimeLineID *repl
/* Reset the prefetcher. */
XLogPrefetchReconfigure();
}
+
+ /*
+ * Conusmer should also enable the standby if pipline have.
+ */
+ if (pipeline_enabled_standby)
+ EnableStandbyMode();
+
+ SpinLockAcquire(&WalPipelineShm->mutex);
+ WalPipelineShm->applied_lsn = xlogreader->EndRecPtr;
+ SpinLockRelease(&WalPipelineShm->mutex);
}
/*
@@ -2232,12 +2353,27 @@ CheckRecoveryConsistency(void)
Assert(InArchiveRecovery);
- /*
- * assume that we are called in the startup process, and hence don't need
- * a lock to read lastReplayedEndRecPtr
- */
- lastReplayedEndRecPtr = XLogRecoveryCtl->lastReplayedEndRecPtr;
- lastReplayedTLI = XLogRecoveryCtl->lastReplayedTLI;
+
+ if (AmStartupProcess())
+ {
+ /*
+ * assume that we are called in the startup process, and hence don't need
+ * a lock to read lastReplayedEndRecPtr
+ */
+ lastReplayedEndRecPtr = XLogRecoveryCtl->lastReplayedEndRecPtr;
+ lastReplayedTLI = XLogRecoveryCtl->lastReplayedTLI;
+ }
+ else
+ {
+ /*
+ * We could be in the pipeline worker, so update the shared states.
+ */
+ SpinLockAcquire(&XLogRecoveryCtl->info_lck);
+ lastReplayedEndRecPtr = XLogRecoveryCtl->lastReplayedEndRecPtr;
+ lastReplayedTLI = XLogRecoveryCtl->lastReplayedTLI;
+ standbyState = XLogRecoveryCtl->standbyState;
+ SpinLockRelease(&XLogRecoveryCtl->info_lck);
+ }
/*
* Have we reached the point where our base backup was completed?
@@ -2424,12 +2560,28 @@ static void
checkTimeLineSwitch(XLogRecPtr lsn, TimeLineID newTLI, TimeLineID prevTLI,
TimeLineID replayTLI)
{
+
+
/* Check that the record agrees on what the current (old) timeline is */
if (prevTLI != replayTLI)
ereport(PANIC,
(errmsg("unexpected previous timeline ID %u (current timeline ID %u) in checkpoint record",
prevTLI, replayTLI)));
+
+ /* Pipeline may have updated the expectedTLEs */
+ if (wal_pipeline_enabled)
+ {
+ TimeLineID targetTLI;
+
+ SpinLockAcquire(&XLogRecoveryCtl->info_lck);
+ targetTLI = XLogRecoveryCtl->expectedTLEsUpdateTLI;
+ SpinLockRelease(&XLogRecoveryCtl->info_lck);
+
+ if (targetTLI)
+ expectedTLEs = readTimeLineHistory(targetTLI);
+ }
+
/*
* The new timeline better be in the list of timelines we expect to see,
* according to the timeline history. It should also not decrease.
@@ -2971,8 +3123,13 @@ getRecoveryStopReason(void)
static void
recoveryPausesHere(bool endOfRecovery)
{
- /* Don't pause unless users can connect! */
- if (!LocalHotStandbyActive)
+ /*
+ * Don't pause unless users can connect!
+ *
+ * As this function could be called from the pipeline process so better
+ * to rely on the shared state then checking LocalHotStandbyActive.
+ */
+ if (!HotStandbyActive())
return;
/* Don't pause after standby promotion has been triggered */
@@ -3077,7 +3234,7 @@ recoveryApplyDelay(XLogReaderState *record)
while (true)
{
- ResetLatch(&XLogRecoveryCtl->recoveryWakeupLatch);
+ ResetLatch(&XLogRecoveryCtl->recoveryApplyDelayLatch);
/* This might change recovery_min_apply_delay. */
ProcessStartupProcInterrupts();
@@ -3102,7 +3259,7 @@ recoveryApplyDelay(XLogReaderState *record)
elog(DEBUG2, "recovery apply delay %ld milliseconds", msecs);
- (void) WaitLatch(&XLogRecoveryCtl->recoveryWakeupLatch,
+ (void) WaitLatch(&XLogRecoveryCtl->recoveryApplyDelayLatch,
WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
msecs,
WAIT_EVENT_RECOVERY_APPLY_DELAY);
@@ -3217,6 +3374,17 @@ ReadRecord(XLogPrefetcher *xlogprefetcher, int emode,
{
abortedRecPtr = xlogreader->abortedRecPtr;
missingContrecPtr = xlogreader->missingContrecPtr;
+
+ /*
+ * Also update the shared state if necessary
+ */
+ if (wal_pipeline_enabled)
+ {
+ SpinLockAcquire(&XLogRecoveryCtl->info_lck);
+ XLogRecoveryCtl->abortedRecPtr = abortedRecPtr;
+ XLogRecoveryCtl->missingContrecPtr = missingContrecPtr;
+ SpinLockRelease(&XLogRecoveryCtl->info_lck);
+ }
}
if (readFile >= 0)
@@ -3284,9 +3452,31 @@ ReadRecord(XLogPrefetcher *xlogprefetcher, int emode,
if (!InArchiveRecovery && ArchiveRecoveryRequested &&
!fetching_ckpt)
{
+ /*
+ * Wait for the startup process to apply the last sent record
+ * by the pipeline, otherwise we will fail the consistency
+ * check as all the records decoded by the pipeline have not
+ * arrived/consumed by the consumer (statup proc) yet.
+ */
+ if (wal_pipeline_enabled && AmWalPipeline())
+ WalPipeline_WaitForConsumerCatchup();
+
ereport(DEBUG1,
(errmsg_internal("reached end of WAL in pg_wal, entering archive recovery")));
InArchiveRecovery = true;
+
+ if (wal_pipeline_enabled)
+ {
+ /* also update the shared state */
+ SpinLockAcquire(&XLogRecoveryCtl->info_lck);
+ XLogRecoveryCtl->InArchiveRecovery = InArchiveRecovery;
+
+ /* update startup proc (consumer) about the standbymode */
+ if (StandbyModeRequested)
+ XLogRecoveryCtl->stanbyEnabled = true;
+ SpinLockRelease(&XLogRecoveryCtl->info_lck);
+ }
+
if (StandbyModeRequested)
EnableStandbyMode();
@@ -3783,7 +3973,8 @@ WaitForWALToBecomeAvailable(XLogRecPtr RecPtr, bool randAccess,
LSN_FORMAT_ARGS(RecPtr));
/* Do background tasks that might benefit us later. */
- KnownAssignedTransactionIdsIdleMaintenance();
+ if (AmStartupProcess())
+ KnownAssignedTransactionIdsIdleMaintenance();
(void) WaitLatch(&XLogRecoveryCtl->recoveryWakeupLatch,
WL_LATCH_SET | WL_TIMEOUT |
@@ -3795,6 +3986,8 @@ WaitForWALToBecomeAvailable(XLogRecPtr RecPtr, bool randAccess,
/* Handle interrupt signals of startup process */
ProcessStartupProcInterrupts();
+ if (wal_pipeline_enabled)
+ ProcessPipelineBgwInterrupts();
}
last_fail_time = now;
currentSource = XLOG_FROM_ARCHIVE;
@@ -3820,6 +4013,15 @@ WaitForWALToBecomeAvailable(XLogRecPtr RecPtr, bool randAccess,
xlogSourceNames[oldSource], xlogSourceNames[currentSource],
lastSourceFailed ? "failure" : "success");
+ if (wal_pipeline_enabled)
+ {
+ /* also update the shared state */
+ SpinLockAcquire(&XLogRecoveryCtl->info_lck);
+ XLogRecoveryCtl->currentSource = currentSource;
+ pendingWalRcvRestart = XLogRecoveryCtl->pendingWalRcvRestart;
+ SpinLockRelease(&XLogRecoveryCtl->info_lck);
+ }
+
/*
* We've now handled possible failure. Try to read from the chosen
* source.
@@ -3894,6 +4096,14 @@ WaitForWALToBecomeAvailable(XLogRecPtr RecPtr, bool randAccess,
}
pendingWalRcvRestart = false;
+ if (wal_pipeline_enabled)
+ {
+ /* also update the shared state */
+ SpinLockAcquire(&XLogRecoveryCtl->info_lck);
+ XLogRecoveryCtl->pendingWalRcvRestart = false;
+ SpinLockRelease(&XLogRecoveryCtl->info_lck);
+ }
+
/*
* Launch walreceiver if needed.
*
@@ -3971,6 +4181,15 @@ WaitForWALToBecomeAvailable(XLogRecPtr RecPtr, bool randAccess,
if (latestChunkStart <= RecPtr)
{
XLogReceiptTime = GetCurrentTimestamp();
+
+ if (wal_pipeline_enabled)
+ {
+ /* also update the shared state */
+ SpinLockAcquire(&XLogRecoveryCtl->info_lck);
+ XLogRecoveryCtl->XLogReceiptTime = XLogReceiptTime;
+ SpinLockRelease(&XLogRecoveryCtl->info_lck);
+ }
+
SetCurrentChunkStartTime(XLogReceiptTime);
}
}
@@ -3999,7 +4218,12 @@ WaitForWALToBecomeAvailable(XLogRecPtr RecPtr, bool randAccess,
if (readFile < 0)
{
if (!expectedTLEs)
+ {
expectedTLEs = readTimeLineHistory(recoveryTargetTLI);
+ PipelineConsumerExpectedTLEsUpdateTLI(recoveryTargetTLI);
+ }
+
+
readFile = XLogFileRead(readSegNo, receiveTLI,
XLOG_FROM_STREAM, false);
Assert(readFile >= 0);
@@ -4009,6 +4233,13 @@ WaitForWALToBecomeAvailable(XLogRecPtr RecPtr, bool randAccess,
/* just make sure source info is correct... */
readSource = XLOG_FROM_STREAM;
XLogReceiptSource = XLOG_FROM_STREAM;
+ if (wal_pipeline_enabled)
+ {
+ /* also update the shared state */
+ SpinLockAcquire(&XLogRecoveryCtl->info_lck);
+ XLogRecoveryCtl->XLogReceiptSource = XLogReceiptSource;
+ SpinLockRelease(&XLogRecoveryCtl->info_lck);
+ }
return XLREAD_SUCCESS;
}
break;
@@ -4046,15 +4277,40 @@ WaitForWALToBecomeAvailable(XLogRecPtr RecPtr, bool randAccess,
*/
if (!streaming_reply_sent)
{
- WalRcvRequestApplyReply();
- streaming_reply_sent = true;
+ if (wal_pipeline_enabled && AmWalPipeline())
+ {
+ /*
+ * In case of pipeline enabled, we cannot just call
+ * WalRcvRequestApplyReply() directly as the
+ * consumer (startup proc) running behined the
+ * pipeline producer and haven't actually replayed
+ * all the wal received from the wal_receiver yet.
+ *
+ * So in order to make wakeup consistent, note the
+ * lsn and consumer will only wakeup walrcvr when
+ * the lsn is replayed.
+ */
+ SpinLockAcquire(&XLogRecoveryCtl->info_lck);
+ XLogRecoveryCtl->WakeupWalRcvrRecPtr = flushedUpto;
+ SpinLockRelease(&XLogRecoveryCtl->info_lck);
+ streaming_reply_sent = true;
+ }
+ else
+ {
+ WalRcvRequestApplyReply();
+ streaming_reply_sent = true;
+ }
}
/* Do any background tasks that might benefit us later. */
- KnownAssignedTransactionIdsIdleMaintenance();
+ if (AmStartupProcess())
+ KnownAssignedTransactionIdsIdleMaintenance();
/* Update pg_stat_recovery_prefetch before sleeping. */
- XLogPrefetcherComputeStats(xlogprefetcher);
+ if (AmWalPipeline())
+ XLogPrefetcherComputeStats(xlogprefetcher_pipelined);
+ else
+ XLogPrefetcherComputeStats(xlogprefetcher);
/*
* Wait for more WAL to arrive, when we will be woken
@@ -4085,6 +4341,8 @@ WaitForWALToBecomeAvailable(XLogRecPtr RecPtr, bool randAccess,
* process.
*/
ProcessStartupProcInterrupts();
+ if (wal_pipeline_enabled)
+ ProcessPipelineBgwInterrupts();
}
return XLREAD_FAIL; /* not reached */
@@ -4250,6 +4508,18 @@ rescanLatestTimeLine(TimeLineID replayTLI, XLogRecPtr replayLSN)
list_free_deep(expectedTLEs);
expectedTLEs = newExpectedTLEs;
+ /* Update shared state */
+ if (wal_pipeline_enabled)
+ {
+ PipelineConsumerExpectedTLEsUpdateTLI(newtarget);
+
+ /* XXX May be we can combine the spin locks with about call. */
+ SpinLockAcquire(&XLogRecoveryCtl->info_lck);
+ XLogRecoveryCtl->recoveryTargetTLI = recoveryTargetTLI;
+ SpinLockRelease(&XLogRecoveryCtl->info_lck);
+ }
+
+
/*
* As in StartupXLOG(), try to ensure we have all the history files
* between the old target and new target in pg_wal.
@@ -4338,6 +4608,15 @@ XLogFileRead(XLogSegNo segno, TimeLineID tli,
if (source != XLOG_FROM_STREAM)
XLogReceiptTime = GetCurrentTimestamp();
+ if (wal_pipeline_enabled)
+ {
+ /* also update the shared state */
+ SpinLockAcquire(&XLogRecoveryCtl->info_lck);
+ XLogRecoveryCtl->XLogReceiptTime = XLogReceiptTime;
+ XLogRecoveryCtl->XLogReceiptSource = XLogReceiptSource;
+ SpinLockRelease(&XLogRecoveryCtl->info_lck);
+ }
+
return fd;
}
if (errno != ENOENT || !notfoundOk) /* unexpected failure? */
@@ -4422,7 +4701,10 @@ XLogFileReadAnyTLI(XLogSegNo segno, XLogSource source)
{
elog(DEBUG1, "got WAL segment from archive");
if (!expectedTLEs)
+ {
expectedTLEs = tles;
+ PipelineConsumerExpectedTLEsUpdateTLI(recoveryTargetTLI);
+ }
return fd;
}
}
@@ -4433,7 +4715,10 @@ XLogFileReadAnyTLI(XLogSegNo segno, XLogSource source)
if (fd != -1)
{
if (!expectedTLEs)
+ {
expectedTLEs = tles;
+ PipelineConsumerExpectedTLEsUpdateTLI(recoveryTargetTLI);
+ }
return fd;
}
}
@@ -4455,16 +4740,52 @@ XLogFileReadAnyTLI(XLogSegNo segno, XLogSource source)
void
StartupRequestWalReceiverRestart(void)
{
+
+ /*
+ * currentSource is also defined as pipeline shared state variable.
+ * Update the state before procedding.
+ */
+ if (wal_pipeline_enabled)
+ {
+ SpinLockAcquire(&XLogRecoveryCtl->info_lck);
+ currentSource = XLogRecoveryCtl->currentSource;
+ SpinLockRelease(&XLogRecoveryCtl->info_lck);
+ }
+
if (currentSource == XLOG_FROM_STREAM && WalRcvRunning())
{
ereport(LOG,
(errmsg("WAL receiver process shutdown requested")));
pendingWalRcvRestart = true;
+
+ /*
+ * pendingWalRcvRestart is also defined as pipeline shared state variable.
+ * Update the state before procedding.
+ */
+ if (wal_pipeline_enabled)
+ {
+ SpinLockAcquire(&XLogRecoveryCtl->info_lck);
+ XLogRecoveryCtl->pendingWalRcvRestart = pendingWalRcvRestart;
+ SpinLockRelease(&XLogRecoveryCtl->info_lck);
+ }
}
}
+/*
+ * standbyState is also defined as a shared state. Pipeline worker can also
+ * update its value, so always confirm the shared state before procedding.
+ */
+void
+SetSharedHotStandbyState(void)
+{
+ SpinLockAcquire(&XLogRecoveryCtl->info_lck);
+ XLogRecoveryCtl->standbyState = standbyState;
+ SpinLockRelease(&XLogRecoveryCtl->info_lck);
+}
+
+
/*
* Has a standby promotion already been triggered?
*
@@ -4516,11 +4837,18 @@ CheckForStandbyTrigger(void)
if (LocalPromoteIsTriggered)
return true;
- if (IsPromoteSignaled() && CheckPromoteSignal())
+ /*
+ * Check if the startup process was signaled for promotion. In case we are
+ * calling this function from the pipeline, we need to check the promotion
+ * signals for the pipeline worker instead.
+ */
+ if ((IsPromoteSignaledPipeline() || IsPromoteSignaled()) &&
+ CheckPromoteSignal())
{
ereport(LOG, (errmsg("received promote request")));
RemovePromoteSignalFiles();
ResetPromoteSignaled();
+ ResetPromoteSignaledPipeline();
SetPromoteIsTriggered();
return true;
}
@@ -4559,6 +4887,7 @@ void
WakeupRecovery(void)
{
SetLatch(&XLogRecoveryCtl->recoveryWakeupLatch);
+ SetLatch(&XLogRecoveryCtl->recoveryApplyDelayLatch);
}
/*
@@ -4728,6 +5057,17 @@ GetXLogReceiptTime(TimestampTz *rtime, bool *fromStream)
*/
Assert(InRecovery);
+ /*
+ * If pipeline enabled, get the updated state.
+ */
+ if (wal_pipeline_enabled)
+ {
+ SpinLockAcquire(&XLogRecoveryCtl->info_lck);
+ XLogReceiptTime = XLogRecoveryCtl->XLogReceiptTime;
+ XLogReceiptSource = XLogRecoveryCtl->XLogReceiptSource;
+ SpinLockRelease(&XLogRecoveryCtl->info_lck);
+ }
+
*rtime = XLogReceiptTime;
*fromStream = (XLogReceiptSource == XLOG_FROM_STREAM);
}
@@ -4813,6 +5153,114 @@ RecoveryRequiresIntParameter(const char *param_name, int currValue, int minValue
}
}
+/*
+ * Called by the startup process before launching the pipeline worker.
+ * Exports recovery state variables into WalPipelineParams so the
+ * producer background worker can initialize itself with the same
+ * environment.
+ */
+static void
+WalPipeline_ExportRecoveryState(WalPipelineParams *params)
+{
+ /*
+ * In order to start decoding through the pipeline,
+ * these variables should be saved and then restored later.
+ */
+ params->NextRecPtr = xlogreader->NextRecPtr;
+ params->recoveryTargetTLI = recoveryTargetTLI;
+ params->StandbyModeRequested = StandbyModeRequested;
+ params->StandbyMode = StandbyMode;
+ params->ArchiveRecoveryRequested = ArchiveRecoveryRequested;
+ params->InArchiveRecovery = InArchiveRecovery;
+ params->minRecoveryPointTLI = minRecoveryPointTLI;
+ params->minRecoveryPoint = minRecoveryPoint;
+ params->InRedo = InRedo;
+ params->currentSource = currentSource;
+ params->lastSourceFailed = lastSourceFailed;
+ params->pendingWalRcvRestart = pendingWalRcvRestart;
+ params->RedoStartTLI = RedoStartTLI;
+ params->CheckPointLoc = CheckPointLoc;
+ params->CheckPointTLI = CheckPointTLI;
+ params->RedoStartLSN = RedoStartLSN;
+ params->standbyState = standbyState;
+ params->flushedUpto = flushedUpto;
+ params->receiveTLI = receiveTLI;
+ params->abortedRecPtr = abortedRecPtr;
+ params->missingContrecPtr = missingContrecPtr;
+ params->backupEndRequired = backupEndRequired;
+ params->backupStartPoint = backupStartPoint;
+ params->backupEndPoint = backupEndPoint;
+ params->curFileTLI = curFileTLI;
+
+ /*
+ * Saving promotion signal now, otherwise we won't be able hanlde this
+ * signal becuase prodcuer process not started yet and startup proc won't
+ * be responsible for decoding.
+ */
+ params->promotedBeforeLaunch = IsPromoteSignaled() && CheckPromoteSignal();
+
+ /*
+ * The pipeline will do the waiting in this case startup proc should disown
+ * the latch.
+ */
+ DisownRecoveryWakeupLatch();
+
+ /*
+ * Update shared state before starting.
+ */
+ SpinLockAcquire(&XLogRecoveryCtl->info_lck);
+ XLogRecoveryCtl->InArchiveRecovery = InArchiveRecovery;
+ XLogRecoveryCtl->pendingWalRcvRestart = pendingWalRcvRestart;
+ XLogRecoveryCtl->abortedRecPtr = abortedRecPtr;
+ XLogRecoveryCtl->missingContrecPtr = missingContrecPtr;
+ XLogRecoveryCtl->currentSource = currentSource;
+ XLogRecoveryCtl->standbyState = standbyState;
+ XLogRecoveryCtl->XLogReceiptSource = XLogReceiptSource;
+ XLogRecoveryCtl->XLogReceiptTime = XLogReceiptTime;
+ SpinLockRelease(&XLogRecoveryCtl->info_lck);
+}
+
+/*
+ * Called by the producer background worker on startup.
+ * Imports recovery state from WalPipelineParams into the worker's
+ * own local variables, mirroring the environment the startup process
+ * had when it launched the pipeline.
+ */
+void
+WalPipeline_ImportRecoveryState(WalPipelineParams *params)
+{
+ StandbyMode = params->StandbyMode;
+ StandbyModeRequested = params->StandbyModeRequested;
+ ArchiveRecoveryRequested = params->ArchiveRecoveryRequested;
+ InArchiveRecovery = params->InArchiveRecovery;
+ recoveryTargetTLI = params->recoveryTargetTLI;
+ minRecoveryPointTLI = params->minRecoveryPointTLI;
+ minRecoveryPoint = params->minRecoveryPoint;
+ currentSource = params->currentSource;
+ lastSourceFailed = params->lastSourceFailed;
+ pendingWalRcvRestart = params->pendingWalRcvRestart;
+ RedoStartTLI = params->RedoStartTLI;
+ RedoStartLSN = params->RedoStartLSN;
+ standbyState = params->standbyState;
+ CheckPointLoc = params->CheckPointLoc;
+ CheckPointTLI = params->CheckPointTLI;
+ flushedUpto = params->flushedUpto;
+ receiveTLI = params->receiveTLI;
+ abortedRecPtr = params->abortedRecPtr;
+ missingContrecPtr = params->missingContrecPtr;
+ InRedo = params->InRedo;
+ backupEndRequired = params->backupEndRequired;
+ backupStartPoint = params->backupStartPoint;
+ backupEndPoint = params->backupEndPoint;
+ curFileTLI = params->curFileTLI;
+ InRecovery = true;
+
+ /*
+ * As pipeline will be reading the wal, so better to own the latch to wait at.
+ */
+ if (ArchiveRecoveryRequested)
+ OwnLatch(&XLogRecoveryCtl->recoveryWakeupLatch);
+}
/*
* GUC check_hook for primary_slot_name
diff --git a/src/backend/storage/ipc/standby.c b/src/backend/storage/ipc/standby.c
index de9092fdf5b..f7f31a916a7 100644
--- a/src/backend/storage/ipc/standby.c
+++ b/src/backend/storage/ipc/standby.c
@@ -1197,6 +1197,7 @@ standby_redo(XLogReaderState *record)
running.xids = xlrec->xids;
ProcArrayApplyRecoveryInfo(&running);
+ SetSharedHotStandbyState();
/*
* The startup process currently has no convenient way to schedule
diff --git a/src/include/access/xlogpipeline.h b/src/include/access/xlogpipeline.h
index a8995d5c1a7..c47cc72121c 100644
--- a/src/include/access/xlogpipeline.h
+++ b/src/include/access/xlogpipeline.h
@@ -54,6 +54,52 @@ typedef struct WalRecordMsgHeader
XLogRecPtr overwrittenRecPtr; /* XLogReaderState->overwrittenRecPtr */
} WalRecordMsgHeader;
+/*
+ * Parameters passed from StartupXLOG (consumer side)
+ * to the WAL pipeline producer background worker.
+ */
+typedef struct WalPipelineParams
+{
+ bool StandbyMode;
+ bool StandbyModeRequested;
+ bool ArchiveRecoveryRequested;
+ bool InArchiveRecovery;
+ bool InRedo;
+ bool lastSourceFailed;
+ bool pendingWalRcvRestart;
+ bool backupEndRequired;
+ bool promotedBeforeLaunch;
+
+ TimeLineID RedoStartTLI;
+ TimeLineID CheckPointTLI;
+ TimeLineID recoveryTargetTLI;
+ TimeLineID minRecoveryPointTLI;
+ TimeLineID ReplayTLI;
+ TimeLineID receiveTLI;
+
+ XLogRecPtr backupStartPoint;
+ XLogRecPtr backupEndPoint;
+ XLogRecPtr CheckPointLoc;
+ XLogRecPtr RedoStartLSN;
+ XLogRecPtr NextRecPtr;
+ XLogRecPtr minRecoveryPoint;
+ XLogRecPtr flushedUpto;
+ XLogRecPtr abortedRecPtr;
+ XLogRecPtr missingContrecPtr;
+
+ int readFile;
+ XLogSegNo readSegNo;
+ uint32 readOff;
+ uint32 readLen;
+ XLogSource readSource;
+ TimeLineID curFileTLI;
+
+
+ HotStandbyState standbyState;
+ XLogSource currentSource;
+
+} WalPipelineParams;
+
/*
* Shared memory control structure for the WAL pipeline
*/
diff --git a/src/include/access/xlogrecovery.h b/src/include/access/xlogrecovery.h
index 81ac984a904..45357a25cad 100644
--- a/src/include/access/xlogrecovery.h
+++ b/src/include/access/xlogrecovery.h
@@ -11,7 +11,9 @@
#ifndef XLOGRECOVERY_H
#define XLOGRECOVERY_H
+#include "access/xlogprefetcher.h"
#include "access/xlogreader.h"
+#include "access/xlogutils.h"
#include "catalog/pg_control.h"
#include "lib/stringinfo.h"
#include "storage/condition_variable.h"
@@ -60,6 +62,17 @@ typedef enum RecoveryPauseState
RECOVERY_PAUSED, /* recovery is paused */
} RecoveryPauseState;
+/* Codes indicating where we got a WAL file from during recovery, or where
+ * to attempt to get one.
+ */
+typedef enum
+{
+ XLOG_FROM_ANY = 0, /* request to read WAL from any source */
+ XLOG_FROM_ARCHIVE, /* restored using restore_command */
+ XLOG_FROM_PG_WAL, /* existing file in pg_wal */
+ XLOG_FROM_STREAM, /* streamed from primary */
+} XLogSource;
+
/*
* Shared-memory state for WAL recovery.
*/
@@ -94,6 +107,14 @@ typedef struct XLogRecoveryCtlData
*/
Latch recoveryWakeupLatch;
+ /*
+ * In case pipeline enabled we will need two latches. One that can be used
+ * by the pipeline for WAL waiting and other that can be used by the
+ * startup process for the apply delay. Before this we had only one latch
+ * for both cases.
+ */
+ Latch recoveryApplyDelayLatch;
+
/*
* Last record successfully replayed.
*/
@@ -121,6 +142,65 @@ typedef struct XLogRecoveryCtlData
ConditionVariable recoveryNotPausedCV;
slock_t info_lck; /* locks shared variables shown above */
+
+ /* ------------------------------------------------------------------
+ * Variables use for IPC between pipeline and the startup proc.
+ * These are also the static variables in xlogrecovery.c but there state
+ * keep on changing. So we added them as the shared states so that both
+ * the pipeline and the startup proc stay synced if any of these state
+ * changes.
+ * ------------------------------------------------------------------
+ */
+
+ /*
+ * Pipeline could be waiting for the startup process to catchup with the
+ * decoder. This could happend when no wait wal is available from the
+ * current resource and now pipline have change the wal srouce
+ * i.e enabling standby if requested.
+ */
+ bool pipeline_waiting;
+ bool InArchiveRecovery;
+ bool pendingWalRcvRestart;
+ bool stanbyEnabled;
+
+ /*
+ * The target TLI for which expectedTLEs should be recomputed by the
+ * consumer
+ */
+ TimeLineID expectedTLEsUpdateTLI;
+
+ /*
+ * We also export the recoveryTargetTLI as a WalPipelineParams. But other
+ * than passing the initial state, the recoveryTargetTLI can also change
+ * it state during the decoding by the prodcuer (see rescanLatestTimeLine()).
+ *
+ * This mean at the end of the recovery, the startup process should aware
+ * of any such state changes done by the producer. To handle this
+ * ResetStatesIfPipelined() will update the startup local recoveryTargetTLI
+ * with updated one, on FinishWalRecovery().
+ */
+ TimeLineID recoveryTargetTLI;
+
+ /*
+ * Normaly we wakeup walrcvr after specific records have been applied, as
+ * decoding and apllying are sequential so we wakeup after enough records
+ * decoded read.
+ *
+ * But in case of pipeline reads (decoded records) could be ahead of the
+ * consumer apply loop. We cannot wakeup wal rcvr based on how much records
+ * decoded, so we tell consumer to wakeup after only after a specific lsn
+ * (WakeupWalRcvrRecPtr set by the pipeline) has beed replayed.
+ */
+ XLogRecPtr WakeupWalRcvrRecPtr;
+
+ XLogRecPtr abortedRecPtr;
+ XLogRecPtr missingContrecPtr;
+
+ XLogSource currentSource;
+ XLogSource XLogReceiptSource;
+
+ HotStandbyState standbyState;
+ TimestampTz XLogReceiptTime;
} XLogRecoveryCtlData;
extern PGDLLIMPORT XLogRecoveryCtlData *XLogRecoveryCtl;
@@ -205,6 +285,8 @@ typedef struct
bool recovery_signal_file_found;
} EndOfWalRecoveryInfo;
+struct WalPipelineParams; /* forward declaration */
+
extern EndOfWalRecoveryInfo *FinishWalRecovery(void);
extern void ShutdownWalRecovery(void);
extern void RemovePromoteSignalFiles(void);
@@ -225,11 +307,14 @@ extern int XLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr, i
extern bool PromoteIsTriggered(void);
extern bool CheckPromoteSignal(void);
extern void WakeupRecovery(void);
+extern void DisownRecoveryWakeupLatch(void);
+extern void SetSharedHotStandbyState(void);
extern void StartupRequestWalReceiverRestart(void);
extern void XLogRequestWalReceiverReply(void);
extern void RecoveryRequiresIntParameter(const char *param_name, int currValue, int minValue);
+extern void WalPipeline_ImportRecoveryState(struct WalPipelineParams *params);
extern void xlog_outdesc(StringInfo buf, XLogReaderState *record);
--
2.34.1
[application/pdf] recoveries-becnhmark-v04.pdf (35.7K, ../CA+UBfamzeXcdEbmhdOdjWn5X_YVt9n8xUpH5ZwmA7S8VWvaoXw@mail.gmail.com/6-recoveries-becnhmark-v04.pdf)
download
view thread (12+ messages) latest in thread
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected], [email protected], [email protected], [email protected], [email protected], [email protected]
Subject: Re: [WIP] Pipelined Recovery
In-Reply-To: <CA+UBfamzeXcdEbmhdOdjWn5X_YVt9n8xUpH5ZwmA7S8VWvaoXw@mail.gmail.com>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox