From db295eca39586b82967f29bcf8e97e0da7f76d09 Mon Sep 17 00:00:00 2001 From: Imran Zaheer 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 /* 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