From e0e7b14b9c4f3987224b2d0e8d3cb4be6e02003d Mon Sep 17 00:00:00 2001 From: Heikki Linnakangas Date: Fri, 21 Mar 2025 16:13:38 +0300 Subject: [PATCH v5 2/3] Apply walrcv_shutdown_deblocking_v2-2.patch From https://www.postgresql.org/message-id/20240123.172410.1596193222420636986.horikyota.ntt%40gmail.com Plus couple of fixes from review: - use of AmWalReceiverProcess - use of die --- .../libpqwalreceiver/libpqwalreceiver.c | 4 +- src/backend/replication/walreceiver.c | 41 +++---------------- src/backend/tcop/postgres.c | 5 +++ 3 files changed, 12 insertions(+), 38 deletions(-) diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c index 5755ab2f072..d51624ef3da 100644 --- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c +++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c @@ -820,7 +820,7 @@ libpqrcv_PQgetResult(PGconn *streamConn) if (rc & WL_LATCH_SET) { ResetLatch(MyLatch); - ProcessWalRcvInterrupts(); + CHECK_FOR_INTERRUPTS(); } /* Consume whatever data is available from the socket */ @@ -1172,7 +1172,7 @@ libpqrcv_processTuples(PGresult *pgres, WalRcvExecResult *walres, { char *cstrs[MaxTupleAttributeNumber]; - ProcessWalRcvInterrupts(); + CHECK_FOR_INTERRUPTS(); /* Do the allocations in temporary context. */ oldcontext = MemoryContextSwitchTo(rowcontext); diff --git a/src/backend/replication/walreceiver.c b/src/backend/replication/walreceiver.c index 2e5dd6deb2c..b51a6d06b21 100644 --- a/src/backend/replication/walreceiver.c +++ b/src/backend/replication/walreceiver.c @@ -71,6 +71,7 @@ #include "storage/proc.h" #include "storage/procarray.h" #include "storage/procsignal.h" +#include "tcop/tcopprot.h" #include "utils/acl.h" #include "utils/builtins.h" #include "utils/guc.h" @@ -145,38 +146,6 @@ static void XLogWalRcvSendHSFeedback(bool immed); static void ProcessWalSndrMessage(XLogRecPtr walEnd, TimestampTz sendTime); static void WalRcvComputeNextWakeup(WalRcvWakeupReason reason, TimestampTz now); -/* - * Process any interrupts the walreceiver process may have received. - * This should be called any time the process's latch has become set. - * - * Currently, only SIGTERM is of interest. We can't just exit(1) within the - * SIGTERM signal handler, because the signal might arrive in the middle of - * some critical operation, like while we're holding a spinlock. Instead, the - * signal handler sets a flag variable as well as setting the process's latch. - * We must check the flag (by calling ProcessWalRcvInterrupts) anytime the - * latch has become set. Operations that could block for a long time, such as - * reading from a remote server, must pay attention to the latch too; see - * libpqrcv_PQgetResult for example. - */ -void -ProcessWalRcvInterrupts(void) -{ - /* - * Although walreceiver interrupt handling doesn't use the same scheme as - * regular backends, call CHECK_FOR_INTERRUPTS() to make sure we receive - * any incoming signals on Win32, and also to make sure we process any - * barrier events. - */ - CHECK_FOR_INTERRUPTS(); - - if (ShutdownRequestPending) - { - ereport(FATAL, - (errcode(ERRCODE_ADMIN_SHUTDOWN), - errmsg("terminating walreceiver process due to administrator command"))); - } -} - /* Main entry point for walreceiver process */ void @@ -280,7 +249,7 @@ WalReceiverMain(const void *startup_data, size_t startup_data_len) pqsignal(SIGHUP, SignalHandlerForConfigReload); /* set flag to read config * file */ pqsignal(SIGINT, SIG_IGN); - pqsignal(SIGTERM, SignalHandlerForShutdownRequest); /* request shutdown */ + pqsignal(SIGTERM, die); /* request shutdown */ /* SIGQUIT handler was already set up by InitPostmasterChild */ pqsignal(SIGALRM, SIG_IGN); pqsignal(SIGPIPE, SIG_IGN); @@ -459,7 +428,7 @@ WalReceiverMain(const void *startup_data, size_t startup_data_len) errmsg("cannot continue WAL streaming, recovery has already ended"))); /* Process any requests or signals received recently */ - ProcessWalRcvInterrupts(); + CHECK_FOR_INTERRUPTS(); if (ConfigReloadPending) { @@ -555,7 +524,7 @@ WalReceiverMain(const void *startup_data, size_t startup_data_len) if (rc & WL_LATCH_SET) { ResetLatch(MyLatch); - ProcessWalRcvInterrupts(); + CHECK_FOR_INTERRUPTS(); if (walrcv->force_reply) { @@ -704,7 +673,7 @@ WalRcvWaitForStartPosition(XLogRecPtr *startpoint, TimeLineID *startpointTLI) { ResetLatch(MyLatch); - ProcessWalRcvInterrupts(); + CHECK_FOR_INTERRUPTS(); SpinLockAcquire(&walrcv->mutex); Assert(walrcv->walRcvState == WALRCV_RESTARTING || diff --git a/src/backend/tcop/postgres.c b/src/backend/tcop/postgres.c index 0554a4ae3c7..c7d3b25d3f2 100644 --- a/src/backend/tcop/postgres.c +++ b/src/backend/tcop/postgres.c @@ -58,6 +58,7 @@ #include "replication/logicallauncher.h" #include "replication/logicalworker.h" #include "replication/slot.h" +#include "replication/walreceiver.h" #include "replication/walsender.h" #include "rewrite/rewriteHandler.h" #include "storage/bufmgr.h" @@ -3311,6 +3312,10 @@ ProcessInterrupts(void) */ proc_exit(1); } + else if (AmWalReceiverProcess()) + ereport(FATAL, + (errcode(ERRCODE_ADMIN_SHUTDOWN), + errmsg("terminating walreceiver process due to administrator command"))); else if (AmBackgroundWorkerProcess()) ereport(FATAL, (errcode(ERRCODE_ADMIN_SHUTDOWN), -- 2.43.0