From d80bb1b4fcbdf9b651d9f6a0aed81fb9f92660fa Mon Sep 17 00:00:00 2001 From: Imran Zaheer 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 Idea by: Ants Aasma --- 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