v21-0006-Make-launcher-wake-up-more-frequently.patch

From ed5990361c5266744ca888403b5001526bfe1ed7 Mon Sep 17 00:00:00 2001
From: Hayato Kuroda <kuroda.hayato@fujitsu.com>
Date: Fri, 10 Jan 2025 10:27:14 +0000
Subject: [PATCH v21 6/7] Make launcher wake up more frequently

---
 src/backend/replication/logical/launcher.c | 22 ++++++++++++++++++----
 src/backend/replication/logical/worker.c   |  3 +++
 src/include/replication/logicallauncher.h  |  1 +
 3 files changed, 22 insertions(+), 4 deletions(-)

diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index b3b57daf19..5c4ab223c7 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -52,6 +52,13 @@
  */
 #define MIN_NAPTIME_PER_SLOT_UPDATE 200
 
+/*
+ * Max sleep time (10s) between cycles update the xmin; If one of the
+ * subscriptions has retain_conflict_info = true, this will be the max sleep
+ * time of the cycle.
+ */
+#define MAX_NAPTIME_FOR_SLOT_UPDATE 10000L
+
 /* GUC variables */
 int			max_logical_replication_workers = 4;
 int			max_sync_workers_per_subscription = 2;
@@ -97,7 +104,6 @@ static dshash_table *last_start_times = NULL;
 static bool on_commit_launcher_wakeup = false;
 
 
-static void ApplyLauncherWakeup(void);
 static void logicalrep_launcher_onexit(int code, Datum arg);
 static void logicalrep_worker_onexit(int code, Datum arg);
 static void logicalrep_worker_detach(void);
@@ -1117,7 +1123,14 @@ ApplyLauncherWakeupAtCommit(void)
 		on_commit_launcher_wakeup = true;
 }
 
-static void
+/*
+ * Request wakeup of the launcher immediately.
+ *
+ * Apart from ApplyLauncherWakeupAtCommit(), this does not wait committing
+ * current transactions. Should be used when oldest_nonremovable_xid of apply
+ * workers is updated.
+ */
+void
 ApplyLauncherWakeup(void)
 {
 	if (LogicalRepCtx->launcher_pid != 0)
@@ -1440,19 +1453,20 @@ advance_conflict_slot_xmin(FullTransactionId new_xmin)
  * Update the sleep time before the next slot update.
  *
  * If there is no slot activity, the wait time between sync cycles will double
- * (up to a maximum of 3 minutes). If there is some slot activity, the wait
+ * (up to a maximum of 10 seconds). If there is some slot activity, the wait
  * time between sync cycles is reset to the minimum (200ms).
  */
 static void
 compute_slot_update_naptime(bool slot_updated, long *sleep_time)
 {
+
 	if (!slot_updated)
 	{
 		/*
 		 * The slot was not updated, so double the sleep time, but not beyond
 		 * the maximum allowable value.
 		 */
-		*sleep_time = Min(*sleep_time * 2, DEFAULT_NAPTIME_PER_CYCLE);
+		*sleep_time = Min(*sleep_time * 2, MAX_NAPTIME_FOR_SLOT_UPDATE);
 	}
 	else
 	{
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index bea9895363..5fa382ab3e 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -4358,6 +4358,9 @@ wait_for_local_flush(RetainConflictInfoData *data)
 		 LSN_FORMAT_ARGS(data->remote_lsn),
 		 XidFromFullTransactionId(data->candidate_xid));
 
+	/* Notify launcher to update the xmin of the conflict slot */
+	ApplyLauncherWakeup();
+
 	/*
 	 * Reset all data fields except those used to determine the timing for the
 	 * next round of transaction ID advancement.
diff --git a/src/include/replication/logicallauncher.h b/src/include/replication/logicallauncher.h
index 82b202f330..7b29f1814d 100644
--- a/src/include/replication/logicallauncher.h
+++ b/src/include/replication/logicallauncher.h
@@ -25,6 +25,7 @@ extern void ApplyLauncherShmemInit(void);
 extern void ApplyLauncherForgetWorkerStartTime(Oid subid);
 
 extern void ApplyLauncherWakeupAtCommit(void);
+extern void ApplyLauncherWakeup(void);
 extern void AtEOXact_ApplyLauncher(bool isCommit);
 
 extern bool IsLogicalLauncher(void);
-- 
2.43.0