From 0de9a0d26630b6591f0c11ded47d021929d11590 Mon Sep 17 00:00:00 2001 From: Hou Zhijie 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