From b5acfe1ae3af8a03bbbfbee795c01ce9344a44a8 Mon Sep 17 00:00:00 2001
From: Thomas Munro <thomas.munro@gmail.com>
Date: Thu, 27 Jan 2022 21:43:17 +1300
Subject: [PATCH v4 2/2] Suppress useless wakeups in walreceiver.

Instead of waking up 10 times per second to check for various timeout
conditions, keep track of when we next have periodic work to do.

Reviewed-by: Kyotaro Horiguchi <horikyota.ntt@gmail.com>
Discussion: https://postgr.es/m/CA%2BhUKGJGhX4r2LPUE3Oy9BX71Eum6PBcS8L3sJpScR9oKaTVaA%40mail.gmail.com
---
 src/backend/replication/walreceiver.c | 203 ++++++++++++++++----------
 1 file changed, 125 insertions(+), 78 deletions(-)

diff --git a/src/backend/replication/walreceiver.c b/src/backend/replication/walreceiver.c
index 89985c54cf..d6f81e5499 100644
--- a/src/backend/replication/walreceiver.c
+++ b/src/backend/replication/walreceiver.c
@@ -95,8 +95,6 @@ bool		hot_standby_feedback;
 static WalReceiverConn *wrconn = NULL;
 WalReceiverFunctionsType *WalReceiverFunctions = NULL;
 
-#define NAPTIME_PER_CYCLE 100	/* max sleep time between cycles (100ms) */
-
 /*
  * These variables are used similarly to openLogFile/SegNo,
  * but for walreceiver to write the XLOG. recvFileTLI is the TimeLineID
@@ -116,12 +114,25 @@ static struct
 	XLogRecPtr	Flush;			/* last byte + 1 flushed in the standby */
 }			LogstreamResult;
 
+/*
+ * Reasons to wake up and perform periodic tasks.
+ */
+typedef enum WalRcvWakeupReason
+{
+	WALRCV_WAKEUP_TERMINATE,
+	WALRCV_WAKEUP_PING,
+	WALRCV_WAKEUP_REPLY,
+	WALRCV_WAKEUP_HSFEEDBACK,
+	NUM_WALRCV_WAKEUPS
+} WalRcvWakeupReason;
+
 /*
  * A struct to keep track of non-shared state.
  */
 typedef struct WalRcvInfo
 {
 	TimeLineID	startpointTLI;
+	TimestampTz	wakeup[NUM_WALRCV_WAKEUPS];
 } WalRcvInfo;
 
 static StringInfoData reply_message;
@@ -137,9 +148,13 @@ static void XLogWalRcvWrite(WalRcvInfo *state, char *buf, Size nbytes,
 							XLogRecPtr recptr);
 static void XLogWalRcvFlush(WalRcvInfo *state, bool dying);
 static void XLogWalRcvClose(WalRcvInfo *state, XLogRecPtr recptr);
-static void XLogWalRcvSendReply(bool force, bool requestReply);
-static void XLogWalRcvSendHSFeedback(bool immed);
+static void XLogWalRcvSendReply(WalRcvInfo *state, TimestampTz now, bool force,
+								bool requestReply);
+static void XLogWalRcvSendHSFeedback(WalRcvInfo *state, TimestampTz now,
+									 bool immed);
 static void ProcessWalSndrMessage(XLogRecPtr walEnd, TimestampTz sendTime);
+static void WalRcvComputeNextWakeup(WalRcvInfo *state,
+									WalRcvWakeupReason reason, TimestampTz now);
 
 /*
  * Process any interrupts the walreceiver process may have received.
@@ -186,9 +201,7 @@ WalReceiverMain(void)
 	TimeLineID	primaryTLI;
 	bool		first_stream;
 	WalRcvData *walrcv = WalRcv;
-	TimestampTz last_recv_timestamp;
-	TimestampTz starttime;
-	bool		ping_sent;
+	TimestampTz now;
 	char	   *err;
 	char	   *sender_host = NULL;
 	int			sender_port = 0;
@@ -200,7 +213,7 @@ WalReceiverMain(void)
 	 */
 	Assert(walrcv != NULL);
 
-	starttime = GetCurrentTimestamp();
+	now = GetCurrentTimestamp();
 
 	/*
 	 * Mark walreceiver as running in shared memory.
@@ -256,7 +269,7 @@ WalReceiverMain(void)
 
 	/* Initialise to a sanish value */
 	walrcv->lastMsgSendTime =
-		walrcv->lastMsgReceiptTime = walrcv->latestWalEndTime = starttime;
+		walrcv->lastMsgReceiptTime = walrcv->latestWalEndTime = now;
 
 	/* Report the latch to use to awaken this process */
 	walrcv->latch = &MyProc->procLatch;
@@ -422,9 +435,10 @@ WalReceiverMain(void)
 			initStringInfo(&reply_message);
 			initStringInfo(&incoming_message);
 
-			/* Initialize the last recv timestamp */
-			last_recv_timestamp = GetCurrentTimestamp();
-			ping_sent = false;
+			/* Initialize nap wakeup times. */
+			now = GetCurrentTimestamp();
+			for (int i = 0; i < NUM_WALRCV_WAKEUPS; ++i)
+				WalRcvComputeNextWakeup(&state, i, now);
 
 			/* Loop until end-of-streaming or error */
 			for (;;)
@@ -434,6 +448,8 @@ WalReceiverMain(void)
 				bool		endofwal = false;
 				pgsocket	wait_fd = PGINVALID_SOCKET;
 				int			rc;
+				TimestampTz	nextWakeup;
+				int			nap;
 
 				/*
 				 * Exit walreceiver if we're not in recovery. This should not
@@ -451,11 +467,15 @@ WalReceiverMain(void)
 				{
 					ConfigReloadPending = false;
 					ProcessConfigFile(PGC_SIGHUP);
-					XLogWalRcvSendHSFeedback(true);
+					now = GetCurrentTimestamp();
+					for (int i = 0; i < NUM_WALRCV_WAKEUPS; ++i)
+						WalRcvComputeNextWakeup(&state, i, now);
+					XLogWalRcvSendHSFeedback(&state, now, true);
 				}
 
 				/* See if we can read data immediately */
 				len = walrcv_receive(wrconn, &buf, &wait_fd);
+				now = GetCurrentTimestamp();
 				if (len != 0)
 				{
 					/*
@@ -467,11 +487,15 @@ WalReceiverMain(void)
 						if (len > 0)
 						{
 							/*
-							 * Something was received from primary, so reset
-							 * timeout
+							 * Something was received from primary, so adjust
+							 * the ping and terminate wakeup times.
 							 */
-							last_recv_timestamp = GetCurrentTimestamp();
-							ping_sent = false;
+							WalRcvComputeNextWakeup(&state,
+													WALRCV_WAKEUP_TERMINATE,
+													now);
+							WalRcvComputeNextWakeup(&state,
+													WALRCV_WAKEUP_PING,
+													now);
 							XLogWalRcvProcessMsg(&state, buf[0], &buf[1],
 												 len - 1);
 						}
@@ -488,10 +512,11 @@ WalReceiverMain(void)
 							break;
 						}
 						len = walrcv_receive(wrconn, &buf, &wait_fd);
+						now = GetCurrentTimestamp();
 					}
 
 					/* Let the primary know that we received some data. */
-					XLogWalRcvSendReply(false, false);
+					XLogWalRcvSendReply(&state, now, false, false);
 
 					/*
 					 * If we've written some records, flush them to disk and
@@ -505,6 +530,12 @@ WalReceiverMain(void)
 				if (endofwal)
 					break;
 
+				/* Find the soonest wakeup time, to limit our nap. */
+				nextWakeup = INT64_MAX;
+				for (int i = 0; i < NUM_WALRCV_WAKEUPS; ++i)
+					nextWakeup = Min(state.wakeup[i], nextWakeup);
+				nap = Max(0, (nextWakeup - now + 999) / 1000);
+
 				/*
 				 * Ideally we would reuse a WaitEventSet object repeatedly
 				 * here to avoid the overheads of WaitLatchOrSocket on epoll
@@ -521,8 +552,9 @@ WalReceiverMain(void)
 									   WL_EXIT_ON_PM_DEATH | WL_SOCKET_READABLE |
 									   WL_TIMEOUT | WL_LATCH_SET,
 									   wait_fd,
-									   NAPTIME_PER_CYCLE,
+									   nap,
 									   WAIT_EVENT_WAL_RECEIVER_MAIN);
+				now = GetCurrentTimestamp();
 				if (rc & WL_LATCH_SET)
 				{
 					ResetLatch(MyLatch);
@@ -538,7 +570,7 @@ WalReceiverMain(void)
 						 */
 						walrcv->force_reply = false;
 						pg_memory_barrier();
-						XLogWalRcvSendReply(true, false);
+						XLogWalRcvSendReply(&state, now, true, false);
 					}
 				}
 				if (rc & WL_TIMEOUT)
@@ -558,38 +590,23 @@ WalReceiverMain(void)
 					 * Check if time since last receive from primary has
 					 * reached the configured limit.
 					 */
-					if (wal_receiver_timeout > 0)
-					{
-						TimestampTz now = GetCurrentTimestamp();
-						TimestampTz timeout;
-
-						timeout =
-							TimestampTzPlusMilliseconds(last_recv_timestamp,
-														wal_receiver_timeout);
-
-						if (now >= timeout)
-							ereport(ERROR,
-									(errcode(ERRCODE_CONNECTION_FAILURE),
-									 errmsg("terminating walreceiver due to timeout")));
+					if (now >= state.wakeup[WALRCV_WAKEUP_TERMINATE])
+						ereport(ERROR,
+								(errcode(ERRCODE_CONNECTION_FAILURE),
+								 errmsg("terminating walreceiver due to timeout")));
 
-						/*
-						 * We didn't receive anything new, for half of
-						 * receiver replication timeout. Ping the server.
-						 */
-						if (!ping_sent)
-						{
-							timeout = TimestampTzPlusMilliseconds(last_recv_timestamp,
-																  (wal_receiver_timeout / 2));
-							if (now >= timeout)
-							{
-								requestReply = true;
-								ping_sent = true;
-							}
-						}
+					/*
+					 * We didn't receive anything new, for half of receiver
+					 * replication timeout. Ping the server.
+					 */
+					if (now >= state.wakeup[WALRCV_WAKEUP_PING])
+					{
+						requestReply = true;
+						state.wakeup[WALRCV_WAKEUP_PING] = INT64_MAX;
 					}
 
-					XLogWalRcvSendReply(requestReply, requestReply);
-					XLogWalRcvSendHSFeedback(false);
+					XLogWalRcvSendReply(&state, now, requestReply, requestReply);
+					XLogWalRcvSendHSFeedback(&state, now, false);
 				}
 			}
 
@@ -872,7 +889,7 @@ XLogWalRcvProcessMsg(WalRcvInfo *state, unsigned char type, char *buf, Size len)
 
 				/* If the primary requested a reply, send one immediately */
 				if (replyRequested)
-					XLogWalRcvSendReply(true, false);
+					XLogWalRcvSendReply(state, GetCurrentTimestamp(), true, false);
 				break;
 			}
 		default:
@@ -1010,8 +1027,10 @@ XLogWalRcvFlush(WalRcvInfo *state, bool dying)
 		/* Also let the primary know that we made some progress */
 		if (!dying)
 		{
-			XLogWalRcvSendReply(false, false);
-			XLogWalRcvSendHSFeedback(false);
+			TimestampTz now = GetCurrentTimestamp();
+
+			XLogWalRcvSendReply(state, now, false, false);
+			XLogWalRcvSendHSFeedback(state, now, false);
 		}
 	}
 }
@@ -1077,13 +1096,12 @@ XLogWalRcvClose(WalRcvInfo *state, XLogRecPtr recptr)
  * wal_receiver_timeout.
  */
 static void
-XLogWalRcvSendReply(bool force, bool requestReply)
+XLogWalRcvSendReply(WalRcvInfo *state, TimestampTz now, bool force,
+					bool requestReply)
 {
 	static XLogRecPtr writePtr = 0;
 	static XLogRecPtr flushPtr = 0;
 	XLogRecPtr	applyPtr;
-	static TimestampTz sendTime = 0;
-	TimestampTz now;
 
 	/*
 	 * If the user doesn't want status to be reported to the primary, be sure
@@ -1092,9 +1110,6 @@ XLogWalRcvSendReply(bool force, bool requestReply)
 	if (!force && wal_receiver_status_interval <= 0)
 		return;
 
-	/* Get current timestamp. */
-	now = GetCurrentTimestamp();
-
 	/*
 	 * We can compare the write and flush positions to the last message we
 	 * sent without taking any lock, but the apply position requires a spin
@@ -1107,10 +1122,11 @@ XLogWalRcvSendReply(bool force, bool requestReply)
 	if (!force
 		&& writePtr == LogstreamResult.Write
 		&& flushPtr == LogstreamResult.Flush
-		&& !TimestampDifferenceExceeds(sendTime, now,
-									   wal_receiver_status_interval * 1000))
+		&& now < state->wakeup[WALRCV_WAKEUP_REPLY])
 		return;
-	sendTime = now;
+
+	/* Make sure we wake up when it's time to send another reply. */
+	WalRcvComputeNextWakeup(state, WALRCV_WAKEUP_REPLY, now);
 
 	/* Construct a new message */
 	writePtr = LogstreamResult.Write;
@@ -1122,7 +1138,7 @@ XLogWalRcvSendReply(bool force, bool requestReply)
 	pq_sendint64(&reply_message, writePtr);
 	pq_sendint64(&reply_message, flushPtr);
 	pq_sendint64(&reply_message, applyPtr);
-	pq_sendint64(&reply_message, GetCurrentTimestamp());
+	pq_sendint64(&reply_message, now);
 	pq_sendbyte(&reply_message, requestReply ? 1 : 0);
 
 	/* Send it */
@@ -1146,16 +1162,14 @@ XLogWalRcvSendReply(bool force, bool requestReply)
  * send a feedback message explicitly setting InvalidTransactionId).
  */
 static void
-XLogWalRcvSendHSFeedback(bool immed)
+XLogWalRcvSendHSFeedback(WalRcvInfo *state, TimestampTz now, bool immed)
 {
-	TimestampTz now;
 	FullTransactionId nextFullXid;
 	TransactionId nextXid;
 	uint32		xmin_epoch,
 				catalog_xmin_epoch;
 	TransactionId xmin,
 				catalog_xmin;
-	static TimestampTz sendTime = 0;
 
 	/* initially true so we always send at least one feedback message */
 	static bool primary_has_standby_xmin = true;
@@ -1168,19 +1182,12 @@ XLogWalRcvSendHSFeedback(bool immed)
 		!primary_has_standby_xmin)
 		return;
 
-	/* Get current timestamp. */
-	now = GetCurrentTimestamp();
+	/* Send feedback at most once per wal_receiver_status_interval. */
+	if (!immed && now < state->wakeup[WALRCV_WAKEUP_HSFEEDBACK])
+		return;
 
-	if (!immed)
-	{
-		/*
-		 * Send feedback at most once per wal_receiver_status_interval.
-		 */
-		if (!TimestampDifferenceExceeds(sendTime, now,
-										wal_receiver_status_interval * 1000))
-			return;
-		sendTime = now;
-	}
+	/* Make sure we wake up when it's time to send feedback again. */
+	WalRcvComputeNextWakeup(state, WALRCV_WAKEUP_HSFEEDBACK, now);
 
 	/*
 	 * If Hot Standby is not yet accepting connections there is nothing to
@@ -1228,7 +1235,7 @@ XLogWalRcvSendHSFeedback(bool immed)
 	/* Construct the message and send it. */
 	resetStringInfo(&reply_message);
 	pq_sendbyte(&reply_message, 'h');
-	pq_sendint64(&reply_message, GetCurrentTimestamp());
+	pq_sendint64(&reply_message, now);
 	pq_sendint32(&reply_message, xmin);
 	pq_sendint32(&reply_message, xmin_epoch);
 	pq_sendint32(&reply_message, catalog_xmin);
@@ -1291,6 +1298,46 @@ ProcessWalSndrMessage(XLogRecPtr walEnd, TimestampTz sendTime)
 	}
 }
 
+/*
+ * Compute the next wakeup time for a given wakeup reason.  Can be called to
+ * initialize a wakeup time, to adjust it for the next wakeup, or to
+ * reinitialize it when GUCs have changed.
+ */
+static void
+WalRcvComputeNextWakeup(WalRcvInfo *state, WalRcvWakeupReason reason,
+						TimestampTz now)
+{
+	switch (reason)
+	{
+	case WALRCV_WAKEUP_TERMINATE:
+		if (wal_receiver_timeout <= 0)
+			state->wakeup[reason] = INT64_MAX;
+		else
+			state->wakeup[reason] = now + wal_receiver_timeout * 1000;
+		break;
+	case WALRCV_WAKEUP_PING:
+		if (wal_receiver_timeout <= 0)
+			state->wakeup[reason] = INT64_MAX;
+		else
+			state->wakeup[reason] = now + (wal_receiver_timeout / 2) * 1000;
+		break;
+	case WALRCV_WAKEUP_HSFEEDBACK:
+		if (!hot_standby_feedback || wal_receiver_status_interval <= 0)
+			state->wakeup[reason] = INT64_MAX;
+		else
+			state->wakeup[reason] = now + wal_receiver_status_interval * 1000000;
+		break;
+	case WALRCV_WAKEUP_REPLY:
+		if (wal_receiver_status_interval <= 0)
+			state->wakeup[reason] = INT64_MAX;
+		else
+			state->wakeup[reason] = now + wal_receiver_status_interval * 1000000;
+		break;
+	default:
+		break;
+	}
+}
+
 /*
  * Wake up the walreceiver main loop.
  *
-- 
2.25.1

