From ed5990361c5266744ca888403b5001526bfe1ed7 Mon Sep 17 00:00:00 2001 From: Hayato Kuroda 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