v17-0002-Dynamically-adjust-xid-advancement-interval.patch

From 0de9a0d26630b6591f0c11ded47d021929d11590 Mon Sep 17 00:00:00 2001
From: Hou Zhijie <houzj.fnst@cn.fujitsu.com>
Date: Wed, 11 Dec 2024 16:37:00 +0800
Subject: [PATCH v17 2/6] Dynamically adjust xid advancement interval

If no new transaction ID has been assigned since the last advancement, the
interval is doubled. This increase is limited by the
wal_receiver_status_interval if it is not zero, or otherwise restricted to a
maximum of 3 minutes. If a new transaction ID is detected, the interval is
reset to a minimum of 100ms.
---
 src/backend/replication/logical/worker.c | 62 +++++++++++++++++++++++-
 1 file changed, 61 insertions(+), 1 deletion(-)

diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 654078a726..8cd81de908 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -313,8 +313,19 @@ typedef struct RetainConflictInfoData
 	TimestampTz reply_time;		/* when the publisher responds with status */
 	TimestampTz last_recv_time; /* when the last message was received */
 	TimestampTz candidate_xid_time; /* when the candidate_xid is decided */
+	int			xid_advancement_interval;	/* how much time (ms) to wait
+											 * before attempting to advance
+											 * the non-removable transaction
+											 * ID */
 } RetainConflictInfoData;
 
+/*
+ * The minimum (100ms) and maximum (3 minutes) intervals for advancing
+ * non-removable transaction IDs.
+ */
+#define MIN_XID_ADVANCEMENT_INTERVAL 100
+#define MAX_XID_ADVANCEMENT_INTERVAL 180000L
+
 /* errcontext tracker */
 static ApplyErrorCallbackArg apply_error_callback_arg =
 {
@@ -428,6 +439,8 @@ static void send_feedback(XLogRecPtr recvpos, bool force, bool requestReply);
 static void maybe_advance_nonremovable_xid(RetainConflictInfoData *data,
 										   bool status_received);
 static void get_candidate_xid(RetainConflictInfoData *data);
+static void adjust_xid_advancement_interval(RetainConflictInfoData *data,
+											bool new_xid_found);
 static void request_publisher_status(RetainConflictInfoData *data);
 static void wait_for_publisher_status(RetainConflictInfoData *data,
 									  bool status_received);
@@ -3835,6 +3848,13 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
 		else
 			wait_time = NAPTIME_PER_CYCLE;
 
+		/*
+		 * Ensure to wake up when it's possible to attempt advancing the
+		 * non-removable transaction ID.
+		 */
+		if (data.phase == RCI_GET_CANDIDATE_XID && data.xid_advancement_interval)
+			wait_time = Min(wait_time, data.xid_advancement_interval);
+
 		rc = WaitLatchOrSocket(MyLatch,
 							   WL_SOCKET_READABLE | WL_LATCH_SET |
 							   WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
@@ -4129,7 +4149,7 @@ get_candidate_xid(RetainConflictInfoData *data)
 	 * can consider the other interval or a separate GUC if the need arises.
 	 */
 	if (!TimestampDifferenceExceeds(data->candidate_xid_time, now,
-									wal_receiver_status_interval * 1000))
+									data->xid_advancement_interval))
 		return;
 
 	data->candidate_xid_time = now;
@@ -4147,7 +4167,12 @@ get_candidate_xid(RetainConflictInfoData *data)
 	/* Return if the oldest_nonremovable_xid cannot be advanced */
 	if (FullTransactionIdFollowsOrEquals(MyLogicalRepWorker->oldest_nonremovable_xid,
 										 full_xid))
+	{
+		adjust_xid_advancement_interval(data, false);
 		return;
+	}
+
+	adjust_xid_advancement_interval(data, true);
 
 	data->candidate_xid = full_xid;
 	data->phase = RCI_REQUEST_PUBLISHER_STATUS;
@@ -4156,6 +4181,41 @@ get_candidate_xid(RetainConflictInfoData *data)
 	maybe_advance_nonremovable_xid(data, false);
 }
 
+/*
+ * Adjust the interval for advancing non-removable transaction IDs.
+ *
+ * If no new transaction ID has been assigned since the last advancement, the
+ * interval is doubled. This increase is limited by the
+ * wal_receiver_status_interval if it is not zero, or otherwise restricted to a
+ * maximum of 3 minutes. If a new transaction ID is detected, the interval is
+ * reset to a minimum of 100ms.
+ */
+static void
+adjust_xid_advancement_interval(RetainConflictInfoData *data, bool new_xid_found)
+{
+	if (!new_xid_found && data->xid_advancement_interval)
+	{
+		int			max_interval = wal_receiver_status_interval
+			? wal_receiver_status_interval * 1000
+			: MAX_XID_ADVANCEMENT_INTERVAL;
+
+		/*
+		 * No new transaction ID assigned since the last check, so double the
+		 * interval, but not beyond the maximum allowable value.
+		 */
+		data->xid_advancement_interval = Min(data->xid_advancement_interval * 2,
+											 max_interval);
+	}
+	else
+	{
+		/*
+		 * A new transaction ID was found or the interval is not yet
+		 * initialized, so set the interval to the minimum value.
+		 */
+		data->xid_advancement_interval = MIN_XID_ADVANCEMENT_INTERVAL;
+	}
+}
+
 /*
  * Workhorse for the RCI_REQUEST_PUBLISHER_STATUS phase.
  */
-- 
2.30.0.windows.2