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: Fri, 10 Jul 2026 14:01:22 +0500
Message-ID: <CA+UBfa=Xzk=SyKHuuBPCXcUbZSzTXiNoLuqOWxnd6Zwbh9=_Gw@mail.gmail.com> (raw)
In-Reply-To: <CABPTF7Vz+p5dxUbKPxExaAVLujkbJjrpzXsOZbCqKMp_NvL3YA@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>
<CA+UBfamzeXcdEbmhdOdjWn5X_YVt9n8xUpH5ZwmA7S8VWvaoXw@mail.gmail.com>
<CABPTF7Vz+p5dxUbKPxExaAVLujkbJjrpzXsOZbCqKMp_NvL3YA@mail.gmail.com>
Hi
I am attaching a new series of the patch set.
What's changed?
* Rebased: patch v5-0004 is rebased. Just a small fix in meson.build
* Serialization:
Before this we were copying the decoded data and the msg header to a
temp buffer and passing it to the shmem mq via shm_mq_send(). The new
version uses shm_mq_sendv() with shm_mq_iovec instead. This avoids the
extra copy into the temporary buffer, as well as the additional
palloc()/pfree() required to allocate it. The changes would be
reflected in v5-0001 & a small portion (deserialization) of v5-0002.
* CPU resource usage:
New patch v5-0005 added to the patch set. Before this the resource
usage reported at the end of the recovery via pg_rusage_show() in
PerformWalRecovery() only reflected the startup process. Since WAL
decoding is now ofloaded to run on a separate pipeline worker, the CPU
time consumed by that worker was not included in the final report. The
new patch records the pipeline worker's CPU usage before it exits and
combines it with the startup process's CPU usage when the final
recovery statistics are reported. This ensures that the reported user
and system CPU times reflect the total CPU consumed during recovery,
while the elapsed time remains unchanged.
These are mostly refinement changes; the overall implementation and
design of the pipeline remain the same.
Summary of idea: Decoupling the decoding from the startup process and
offloading it to a dedicated worker. Then the startup process simply
receives the decoded record through a shmem queue. This allows the
startup process to spend most of the CPU time in applying the wal
records instead of decoding and reading them. We refer to this
approach as the wal pipeline.
Thanks
Imran Zaheer
Benchmarks: https://drive.google.com/file/d/13FATRT3kjh_y1wWETpYQh4ZXVLNaYU4A/view?usp=sharing
Test script: https://github.com/imranzaheer612/pg-recovery-testing
On Thu, Jun 25, 2026 at 12:47 PM Xuneng Zhou <[email protected]> wrote:
>
> Hi Imran,
>
> On Tue, Jun 23, 2026 at 9:27 PM Imran Zaheer <[email protected]> wrote:
> >
> > 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.
>
> +1 for splitting the patch set into smaller components to make the
> review process smoother.
>
> > * 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]
>
> The perf result looks promising!
>
> > 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.
>
> I haven't had a chance for a meaningful review yet, but expect to do so soon.
>
> --
> Regards,
> Xuneng Zhou
> HighGo Software Co., Ltd.
Attachments:
[text/x-patch] v5-0001-Pipelined-Recovery-Producer-Related-Code.patch (29.1K, ../CA+UBfa=Xzk=SyKHuuBPCXcUbZSzTXiNoLuqOWxnd6Zwbh9=_Gw@mail.gmail.com/2-v5-0001-Pipelined-Recovery-Producer-Related-Code.patch)
download | inline diff:
From 7b6e04cc0328ff66f6ccfd243b73439384d968dd Mon Sep 17 00:00:00 2001
From: Imran Zaheer <[email protected]>
Date: Fri, 10 Jul 2026 06:57:31 +0500
Subject: [PATCH v5 1/5] 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 | 600 ++++++++++++++++++
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, 753 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..424d33094bf
--- /dev/null
+++ b/src/backend/access/transam/xlogpipeline.c
@@ -0,0 +1,600 @@
+/*-------------------------------------------------------------------------
+ *
+ * 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;
+
+/*
+ * Local buffer containing msg header that will be sent together with the
+ * decoded data, to the msg queue
+ */
+static WalRecordMsgHeader msghdr;
+
+/*
+ * 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 *xlogreader, shm_mq_iovec *iov);
+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();
+}
+
+/*
+ * 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 DecodedXLogRecord so the consumer can
+ * reconstruct them.
+ */
+static void
+set_ptrs_to_offsets(DecodedXLogRecord *dec)
+{
+ if (dec->main_data_len > 0)
+ dec->main_data = (char *)((char *)dec->main_data - (char *)dec);
+
+ 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 *)((char *)blk->data - (char *)dec);
+ if (blk->has_image)
+ blk->bkp_image = (char *)((char *)blk->bkp_image - (char *)dec);
+ }
+}
+
+/*
+ * Restore interior pointers from offsets.
+ */
+static void
+reset_offsets_to_ptrs(DecodedXLogRecord *dec)
+{
+
+ 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;
+ }
+}
+
+/*
+ * Producer Function.
+ * Send a decoded WAL record to the consumer
+ */
+bool
+WalPipeline_SendRecord(XLogReaderState *record)
+{
+ Size msglen;
+ shm_mq_result res;
+ shm_mq_iovec iov[2];
+
+
+ if (!producer_mq_handle)
+ return false;
+
+ /* Serialize the decoded data */
+ msglen = serialize_wal_record(record, iov);
+
+ res = shm_mq_sendv(producer_mq_handle, iov, 2, false, true);
+
+ /*
+ * Reset the offsets to exact ptrs because decoded record data still
+ * needed by the producer reader state
+ */
+ reset_offsets_to_ptrs(record->record);
+
+ if (res == SHM_MQ_SUCCESS)
+ {
+ SpinLockAcquire(&WalPipelineShm->mutex);
+ WalPipelineShm->records_sent++;
+ WalPipelineShm->bytes_sent += msglen;
+ SpinLockRelease(&WalPipelineShm->mutex);
+
+ return true;
+ }
+
+ if (res == SHM_MQ_DETACHED)
+ {
+ elog(PANIC, "[walpipeline] producer: consumer detached");
+ return false;
+ }
+
+ /* Some other error */
+ elog(PANIC, "[walpipeline] producer: shm_mq_send failed with result %d", res);
+ 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
+ * shm_mq_iovec, converting interior pointers to relative offsets.
+ *
+ * Data layout:
+ * [WalRecordMsgHeader][DecodedXLogRecord + trailing data]
+ */
+static Size
+serialize_wal_record(XLogReaderState *xlogreader, shm_mq_iovec *iov)
+{
+ DecodedXLogRecord *dec = xlogreader->record;
+ Size payload_size = dec->size;
+
+ /* build header */
+ msghdr.msg_type = WAL_MSG_RECORD;
+ msghdr.readRecPtr = xlogreader->ReadRecPtr;
+ msghdr.endRecPtr = xlogreader->EndRecPtr;
+ msghdr.missingContrecPtr = xlogreader->missingContrecPtr;
+ msghdr.abortedRecPtr = xlogreader->abortedRecPtr;
+ msghdr.overwrittenRecPtr = xlogreader->overwrittenRecPtr;
+ msghdr.decoded_size = payload_size;
+
+ set_ptrs_to_offsets(dec);
+
+ iov[0].data = (char *) &msghdr;
+ iov[0].len = sizeof(WalRecordMsgHeader);
+
+ iov[1].data = (char *) dec;
+ iov[1].len = payload_size;
+
+ return sizeof(WalRecordMsgHeader) + payload_size;;
+}
+
+/*
+ * 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 dff57642ab4..104ae7ab037 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 c0ae4d3f63f..c7a16d99180 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 c9118e71988..13b000f5db3 100644
--- a/src/backend/utils/misc/guc_parameters.dat
+++ b/src/backend/utils/misc/guc_parameters.dat
@@ -3516,6 +3516,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 d7942f50a70..b94a0509ed0 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] v5-0004-Pipelined-Recovery-Add-Tap-test.patch (7.3K, ../CA+UBfa=Xzk=SyKHuuBPCXcUbZSzTXiNoLuqOWxnd6Zwbh9=_Gw@mail.gmail.com/3-v5-0004-Pipelined-Recovery-Add-Tap-test.patch)
download | inline diff:
From 2ecf95da95134014c29de9c843aa9abeff21b4d5 Mon Sep 17 00:00:00 2001
From: Imran Zaheer <[email protected]>
Date: Mon, 22 Jun 2026 12:36:45 +0500
Subject: [PATCH v5 4/5] 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/055_walpipeline.pl | 208 +++++++++++++++++++++++++
2 files changed, 209 insertions(+)
create mode 100644 src/test/recovery/t/055_walpipeline.pl
diff --git a/src/test/recovery/meson.build b/src/test/recovery/meson.build
index ad0d85f4189..979c1213332 100644
--- a/src/test/recovery/meson.build
+++ b/src/test/recovery/meson.build
@@ -63,6 +63,7 @@ tests += {
't/052_checkpoint_segment_missing.pl',
't/053_standby_login_event_trigger.pl',
't/054_unlogged_sequence_promotion.pl',
+ 't/055_walpipeline.pl',
],
},
}
diff --git a/src/test/recovery/t/055_walpipeline.pl b/src/test/recovery/t/055_walpipeline.pl
new file mode 100644
index 00000000000..d682c878c79
--- /dev/null
+++ b/src/test/recovery/t/055_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] v5-0005-Pipeline-Recovery-Report-combined-CPU-rusage-for-.patch (7.0K, ../CA+UBfa=Xzk=SyKHuuBPCXcUbZSzTXiNoLuqOWxnd6Zwbh9=_Gw@mail.gmail.com/4-v5-0005-Pipeline-Recovery-Report-combined-CPU-rusage-for-.patch)
download | inline diff:
From db295eca39586b82967f29bcf8e97e0da7f76d09 Mon Sep 17 00:00:00 2001
From: Imran Zaheer <[email protected]>
Date: Fri, 10 Jul 2026 13:07:59 +0500
Subject: [PATCH v5 5/5] Pipeline Recovery - Report combined CPU rusage for
startup and pipeline workers
The recovery startup process reports CPU and elapsed time at the end of
recovery using pg_rusage_show(). Since WAL decoding now runs in a
separate pipeline worker, the reported CPU usage only reflected the
startup process and omitted the CPU consumed by the pipeline worker.
Record the pipeline worker's CPU usage before it exits and combine it
with the startup process's CPU usage when reporting the final recovery
statistics. Wall-clock time continues to be measured by the startup
process, while user and system CPU time now reflect the total CPU
consumed during recovery.
---
src/backend/access/transam/xlogpipeline.c | 109 ++++++++++++++++++++++
src/backend/access/transam/xlogrecovery.c | 4 +-
src/include/access/xlogpipeline.h | 14 +++
3 files changed, 126 insertions(+), 1 deletion(-)
diff --git a/src/backend/access/transam/xlogpipeline.c b/src/backend/access/transam/xlogpipeline.c
index bbd5ee597d7..559794e46bb 100644
--- a/src/backend/access/transam/xlogpipeline.c
+++ b/src/backend/access/transam/xlogpipeline.c
@@ -48,6 +48,7 @@
#include "tcop/tcopprot.h"
#include "utils/elog.h"
#include "utils/memutils.h"
+#include "utils/pg_rusage.h"
#include "utils/resowner.h"
#include "utils/rel.h"
#include "utils/timeout.h"
@@ -122,6 +123,108 @@ typedef struct XLogPageReadPrivate
} XLogPageReadPrivate;
+/*
+ * Helpers to track and compute rusage
+ */
+
+static struct timeval
+timeval_subtract(struct timeval end, const struct timeval *start)
+{
+ if (end.tv_usec < start->tv_usec)
+ {
+ end.tv_sec--;
+ end.tv_usec += 1000000;
+ }
+
+ end.tv_sec -= start->tv_sec;
+ end.tv_usec -= start->tv_usec;
+
+ return end;
+}
+
+static void
+timeval_add(struct timeval *a, const struct timeval *b)
+{
+ a->tv_sec += b->tv_sec;
+ a->tv_usec += b->tv_usec;
+
+ if (a->tv_usec >= 1000000)
+ {
+ a->tv_sec++;
+ a->tv_usec -= 1000000;
+ }
+}
+
+/*
+ * Called by Producer.
+ *
+ * Compute final rusage delta by the producer proc and save across shmem so
+ * consumer can access it and use to compute final recovery resource usage.
+ * Note that we don't care about the wal clock time val at this point.
+ */
+static PGRUsageDelta
+pipeline_save_rusage(const PGRUsage *ru0)
+{
+ PGRUsage ru1;
+ PGRUsageDelta delta;
+
+ pg_rusage_init(&ru1);
+
+ delta.utime = timeval_subtract(ru1.ru.ru_utime, &ru0->ru.ru_utime);
+ delta.stime = timeval_subtract(ru1.ru.ru_stime, &ru0->ru.ru_stime);
+
+ SpinLockAcquire(&WalPipelineShm->mutex);
+ WalPipelineShm->producer_rusage = delta;
+ SpinLockRelease(&WalPipelineShm->mutex);
+
+ return delta;
+}
+
+/*
+ * Called by Consumer.
+ *
+ * Compute and return string having final resouce usage.
+ *
+ * ru0: initial snapshot of consumer/startup process
+*/
+const char *
+pipeline_final_rusage(const PGRUsage *ru0)
+{
+ static char result[100];
+ PGRUsage ru1;
+ PGRUsageDelta producer_delta;
+
+
+ Assert(!WalPipeline_CheckProducerAlive());
+
+ SpinLockAcquire(&WalPipelineShm->mutex);
+ producer_delta = WalPipelineShm->producer_rusage;
+ SpinLockRelease(&WalPipelineShm->mutex);
+
+ /* get final snapshot to compute delta against */
+ pg_rusage_init(&ru1);
+
+ /* compute consumer delta */
+ ru1.tv = timeval_subtract(ru1.tv, &ru0->tv);
+ ru1.ru.ru_utime = timeval_subtract(ru1.ru.ru_utime, &ru0->ru.ru_utime);
+ ru1.ru.ru_stime = timeval_subtract(ru1.ru.ru_stime, &ru0->ru.ru_stime);
+
+ /* add producer proc delta (user & sys) */
+ timeval_add(&ru1.ru.ru_utime, &producer_delta.utime);
+ timeval_add(&ru1.ru.ru_stime, &producer_delta.stime);
+
+ snprintf(result, sizeof(result),
+ _("CPU: user: %d.%02d s, system: %d.%02d s, elapsed: %d.%02d s"),
+ (int) ru1.ru.ru_utime.tv_sec,
+ (int) ru1.ru.ru_utime.tv_usec / 10000,
+ (int) ru1.ru.ru_stime.tv_sec,
+ (int) ru1.ru.ru_stime.tv_usec / 10000,
+ (int) ru1.tv.tv_sec,
+ (int) ru1.tv.tv_usec / 10000);
+
+ return result;
+}
+
/*
* Register shared memory for WAL Pipeline
*/
@@ -328,6 +431,7 @@ WalPipeline_ProducerMain(Datum main_arg)
bool end_of_wal = false;
uint64 records_sent;
uint64 records_received;
+ PGRUsage ru0;
/*
* Properly accept or ignore signals the postmaster might send us.
@@ -404,6 +508,8 @@ WalPipeline_ProducerMain(Datum main_arg)
/* Handle the signal if we were promoted before the pipeline launch */
promote_signaled = params->promotedBeforeLaunch;
+ pg_rusage_init(&ru0);
+
/* Main decoding loop */
while (true)
{
@@ -455,6 +561,9 @@ WalPipeline_ProducerMain(Datum main_arg)
WalPipeline_WaitForConsumerShutdownRequest();
}
+ /* compute and set rusage delta */
+ pipeline_save_rusage(&ru0);
+
SpinLockAcquire(&WalPipelineShm->mutex);
records_sent = WalPipelineShm->records_sent;
records_received = WalPipelineShm->records_received;
diff --git a/src/backend/access/transam/xlogrecovery.c b/src/backend/access/transam/xlogrecovery.c
index 65c96a4d107..2473be23ee5 100644
--- a/src/backend/access/transam/xlogrecovery.c
+++ b/src/backend/access/transam/xlogrecovery.c
@@ -2008,7 +2008,9 @@ PerformWalRecovery(void)
ereport(LOG,
errmsg("redo done at %X/%08X system usage: %s",
LSN_FORMAT_ARGS(xlogreader->ReadRecPtr),
- pg_rusage_show(&ru0)));
+ wal_pipeline_enabled ?
+ pipeline_final_rusage(&ru0) : pg_rusage_show(&ru0)));
+
xtime = GetLatestXTime();
if (xtime)
ereport(LOG,
diff --git a/src/include/access/xlogpipeline.h b/src/include/access/xlogpipeline.h
index c47cc72121c..aca1e31fa40 100644
--- a/src/include/access/xlogpipeline.h
+++ b/src/include/access/xlogpipeline.h
@@ -20,18 +20,28 @@
#ifndef WAL_PIPELINE_H
#define WAL_PIPELINE_H
+#include <sys/time.h> /* for struct timeval */
+
#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"
+#include "utils/pg_rusage.h"
/*
* Magic number for shared memory TOC
*/
#define PG_WAL_PIPELINE_MAGIC 0x57414C50 /* "WALP" */
+
+typedef struct PGRUsageDelta
+{
+ struct timeval utime;
+ struct timeval stime;
+} PGRUsageDelta;
+
/*
* Message types sent through the pipeline
*/
@@ -129,6 +139,9 @@ typedef struct WalPipelineShmCtl
uint64 records_received;
uint64 bytes_sent;
uint64 bytes_received;
+
+ /* cpu usage delta by the producer */
+ PGRUsageDelta producer_rusage;
} WalPipelineShmCtl;
/* consumer may have to compute prefetecher stats */
@@ -155,6 +168,7 @@ extern bool WalPipeline_CheckProducerAlive(void);
extern bool WalPipeline_IsActive(void);
extern pid_t WalPipeline_GetProducerPid(void);
extern void WalPipeline_WaitForConsumerCatchup(void);
+extern const char *pipeline_final_rusage(const PGRUsage *ru0);
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] v5-0003-Pipelined-Recovery-Decoupling-startup-and-produce.patch (38.0K, ../CA+UBfa=Xzk=SyKHuuBPCXcUbZSzTXiNoLuqOWxnd6Zwbh9=_Gw@mail.gmail.com/5-v5-0003-Pipelined-Recovery-Decoupling-startup-and-produce.patch)
download | inline diff:
From d5cc4d94701af52dd5f1d5c08562f1d0ddc5c27f Mon Sep 17 00:00:00 2001
From: Imran Zaheer <[email protected]>
Date: Mon, 22 Jun 2026 12:33:53 +0500
Subject: [PATCH v5 3/5] 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 0e165736524..65c96a4d107 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 7f011e04990..430f23d5717 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
[text/x-patch] v5-0002-Pipelined-Recovery-Consumer-Related-Code.patch (18.9K, ../CA+UBfa=Xzk=SyKHuuBPCXcUbZSzTXiNoLuqOWxnd6Zwbh9=_Gw@mail.gmail.com/6-v5-0002-Pipelined-Recovery-Consumer-Related-Code.patch)
download | inline diff:
From d80bb1b4fcbdf9b651d9f6a0aed81fb9f92660fa Mon Sep 17 00:00:00 2001
From: Imran Zaheer <[email protected]>
Date: Fri, 10 Jul 2026 06:53:41 +0500
Subject: [PATCH v5 2/5] 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 | 460 ++++++++++++++++++++++
src/backend/access/transam/xlogrecovery.c | 74 +++-
src/include/access/xlogpipeline.h | 18 +
3 files changed, 550 insertions(+), 2 deletions(-)
diff --git a/src/backend/access/transam/xlogpipeline.c b/src/backend/access/transam/xlogpipeline.c
index 424d33094bf..bbd5ee597d7 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;
+
/*
* Local buffer containing msg header that will be sent together with the
* decoded data, to the msg queue
@@ -98,9 +103,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 *xlogreader, shm_mq_iovec *iov);
+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. */
@@ -133,6 +142,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.
@@ -408,6 +584,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)
@@ -444,6 +824,60 @@ serialize_wal_record(XLogReaderState *xlogreader, shm_mq_iovec *iov)
return sizeof(WalRecordMsgHeader) + payload_size;;
}
+/*
+ * 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;
+
+ reset_offsets_to_ptrs(dec);
+
+ /* 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.
@@ -484,6 +918,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 c7a16d99180..0e165736524 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
[application/pdf] recoveries-becnhmark-v05.pdf (35.7K, ../CA+UBfa=Xzk=SyKHuuBPCXcUbZSzTXiNoLuqOWxnd6Zwbh9=_Gw@mail.gmail.com/7-recoveries-becnhmark-v05.pdf)
download
view thread (3+ messages)
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+UBfa=Xzk=SyKHuuBPCXcUbZSzTXiNoLuqOWxnd6Zwbh9=_Gw@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