diff --git a/src/gausskernel/process/postmaster/pgarch.cpp b/src/gausskernel/process/postmaster/pgarch.cpp index 97949ae1..28817f47 100644 --- a/src/gausskernel/process/postmaster/pgarch.cpp +++ b/src/gausskernel/process/postmaster/pgarch.cpp @@ -109,6 +109,7 @@ typedef bool(*doArchive)(XLogRecPtr); static void pgarch_ArchiverObsCopyLoop(XLogRecPtr flushPtr, doArchive fun); static void archKill(int code, Datum arg); static void initLastTaskLsn(); +static void initArchiveCxt(); AlarmCheckResult DataInstArchChecker(Alarm* alarm, AlarmAdditionalParam* additionalParam) { @@ -227,6 +228,7 @@ NON_EXEC_STATIC void PgArchiverMain() init_ps_display("archiver process", "", "", ""); setObsArchLatch(&t_thrd.arch.mainloop_latch); initLastTaskLsn(); + initArchiveCxt(); pgarch_MainLoop(); gs_thread_exit(0); @@ -689,14 +691,28 @@ static void pgarch_ArchiverObsCopyLoop(XLogRecPtr flushPtr, doArchive fun) } uint32 size = isObsSlot() ? OBS_XLOG_SLICE_BLOCK_SIZE : NAS_XLOG_FILE_SIZE; - targetLsn = Min(t_thrd.arch.pitr_task_last_lsn + size - (t_thrd.arch.pitr_task_last_lsn % size) - 1, - flushPtr); + if (flushPtr == InvalidXLogRecPtr) { + targetLsn = t_thrd.arch.pitr_task_last_lsn + size - (t_thrd.arch.pitr_task_last_lsn % size) - 1; + } else { + targetLsn = Min(t_thrd.arch.pitr_task_last_lsn + size - (t_thrd.arch.pitr_task_last_lsn % size) - 1, + flushPtr); + } /* The previous slice has been archived, switch to the next. */ if (t_thrd.arch.pitr_task_last_lsn == targetLsn) { targetLsn = Min(targetLsn + size, flushPtr); } + if (!XlogFileIsExisted(t_thrd.proc_cxt.DataDir, targetLsn, DEFAULT_TIMELINE_ID)) { + ereport(WARNING, + (errmsg("transaction log file \"%X/%X\" does not existed, it may be archived by the old primary.", + (uint32)(targetLsn >> 32), (uint32)(targetLsn)))); + gettimeofday(&tv, NULL); + t_thrd.arch.last_arch_time = TIME_GET_MILLISEC(tv); + t_thrd.arch.pitr_task_last_lsn = targetLsn; + continue; + } + if (fun(targetLsn) == false) { ereport(WARNING, (errmsg("transaction log file \"%X/%X\" could not be archived: try again", @@ -1102,14 +1118,14 @@ static void initLastTaskLsn() gettimeofday(&tv,NULL); XLogRecPtr targetLsn; t_thrd.arch.last_arch_time = TIME_GET_MILLISEC(tv); - ReplicationSlot* obs_archive_slot = getObsReplicationSlot(); + volatile ReplicationSlot* obs_archive_slot = getObsReplicationSlot(); if (obs_archive_slot != NULL && !IsServerModeStandby()) { - ArchiveXlogMessage obs_archive_info; - if (obs_replication_get_last_xlog(&obs_archive_info) == 0) { - t_thrd.arch.pitr_task_last_lsn = obs_archive_info.targetLsn; + XLogRecPtr targetLsn = obs_archive_slot->data.restart_lsn; + if (targetLsn != InvalidXLogRecPtr) { + t_thrd.arch.pitr_task_last_lsn = targetLsn; advanceObsSlot(t_thrd.arch.pitr_task_last_lsn); ereport(LOG, - (errmsg("initLastTaskLsn update lsn to %X/%X from obs", (uint32)(t_thrd.arch.pitr_task_last_lsn >> 32), + (errmsg("initLastTaskLsn update lsn to %X/%X from slot", (uint32)(t_thrd.arch.pitr_task_last_lsn >> 32), (uint32)(t_thrd.arch.pitr_task_last_lsn)))); } else { targetLsn = GetFlushRecPtr(); @@ -1128,3 +1144,12 @@ static void initLastTaskLsn() (uint32)(t_thrd.arch.pitr_task_last_lsn)))); } } + +/* + * The global variables related to archiving tasks must be initialized before entering the mainloop. + */ +static void initArchiveCxt() { + g_instance.archive_obs_cxt.pitr_finish_result = false; + g_instance.archive_obs_cxt.archive_task.targetLsn = InvalidXLogRecPtr; + g_instance.archive_obs_cxt.pitr_task_status = PITR_TASK_NONE; +} diff --git a/src/gausskernel/process/threadpool/knl_thread.cpp b/src/gausskernel/process/threadpool/knl_thread.cpp index 411316e5..f43458a4 100644 --- a/src/gausskernel/process/threadpool/knl_thread.cpp +++ b/src/gausskernel/process/threadpool/knl_thread.cpp @@ -626,7 +626,9 @@ static void knl_t_arch_init(knl_t_arch_context* arch) arch->last_sigterm_time = 0; arch->pitr_task_last_lsn = 0; arch->task_wait_interval = 1000; + arch->advance_slot_wait_interval = 1000; arch->last_arch_time = 0; + arch->last_advance_slot_time = 0; arch->sync_walsender_idx = -1; } diff --git a/src/gausskernel/storage/replication/slot.cpp b/src/gausskernel/storage/replication/slot.cpp index ef32eaea..3ba3997c 100644 --- a/src/gausskernel/storage/replication/slot.cpp +++ b/src/gausskernel/storage/replication/slot.cpp @@ -52,6 +52,7 @@ #include "postmaster/postmaster.h" #include "utils/builtins.h" +#define TIME_GET_MILLISEC(t) (((long)(t).tv_sec * 1000) + ((long)(t).tv_usec) / 1000) extern bool PMstateIsRun(void); @@ -1903,6 +1904,7 @@ ReplicationSlot *getObsReplicationSlot() g_instance.archive_obs_cxt.archive_slot->archive_obs->obs_sk = pstrdup_ext(slot->archive_obs->obs_sk); g_instance.archive_obs_cxt.archive_slot->archive_obs->obs_prefix = pstrdup_ext(slot->archive_obs->obs_prefix); g_instance.archive_obs_cxt.archive_slot->archive_obs->media_type = slot->archive_obs->media_type; + g_instance.archive_obs_cxt.archive_slot->data.restart_lsn = slot->data.restart_lsn; MemoryContextSwitchTo(curr); SpinLockRelease(&slot->mutex); *slot_idx = slotno; @@ -1924,12 +1926,20 @@ void advanceObsSlot(XLogRecPtr restart_pos) SpinLockAcquire(&slot->mutex); if (slot->in_use == true && slot->archive_obs != NULL) { slot->data.restart_lsn = restart_pos; + SpinLockRelease(&slot->mutex); + struct timeval tv; + gettimeofday(&tv, NULL); + long diff = TIME_GET_MILLISEC(tv) - t_thrd.arch.last_advance_slot_time; + if (!RecoveryInProgress() && diff > t_thrd.arch.advance_slot_wait_interval) { + log_slot_advance(&slot->data); + t_thrd.arch.last_advance_slot_time = TIME_GET_MILLISEC(tv); + } } else { ereport(WARNING, (errcode_for_file_access(), errmsg("slot idx not valid, obs slot %X/%X not advance ", (uint32)(restart_pos >> 32), (uint32)(restart_pos)))); + SpinLockRelease(&slot->mutex); } - SpinLockRelease(&slot->mutex); } } diff --git a/src/gausskernel/storage/replication/slotfuncs.cpp b/src/gausskernel/storage/replication/slotfuncs.cpp index c6a1375f..c55b58b5 100644 --- a/src/gausskernel/storage/replication/slotfuncs.cpp +++ b/src/gausskernel/storage/replication/slotfuncs.cpp @@ -849,9 +849,11 @@ void redo_slot_advance(const ReplicationSlotPersistentData *slotInfo) * If logical replication slot is active on the current standby, the current * standby notify the primary to advance the logical replication slot. * Thus, we do not redo the slot_advance log. + * for logical replication slot, the Slot->data.database is VALID. */ #ifndef ENABLE_MULTIPLE_NODES - if (IsReplicationSlotActive(NameStr(slotInfo->name))) { + if (IsReplicationSlotActive(NameStr(slotInfo->name)) && t_thrd.slot_cxt.MyReplicationSlot != NULL && + t_thrd.slot_cxt.MyReplicationSlot->data.database != InvalidOid) { return; } #endif diff --git a/src/gausskernel/storage/replication/walsender.cpp b/src/gausskernel/storage/replication/walsender.cpp index 40361406..c0636f8b 100644 --- a/src/gausskernel/storage/replication/walsender.cpp +++ b/src/gausskernel/storage/replication/walsender.cpp @@ -3078,7 +3078,9 @@ static int WalSndLoop(WalSndSendDataCallback send_data) } volatile unsigned int *pitr_archive_flag = &t_thrd.walsender_cxt.MyWalSnd->archive_flag; - /* standby can't parse the archive message when the standby is building, + + /* + * standby can't parse the archive message when the standby is building, * so we can't send the archive message. */ if (unlikely(pg_atomic_read_u32(pitr_archive_flag) == 1) && diff --git a/src/include/knl/knl_thread.h b/src/include/knl/knl_thread.h index f4732cbd..00c94da4 100644 --- a/src/include/knl/knl_thread.h +++ b/src/include/knl/knl_thread.h @@ -1188,9 +1188,12 @@ typedef struct knl_t_arch_context { XLogRecPtr pitr_task_last_lsn; /* millsecond */ int task_wait_interval; + long advance_slot_wait_interval; int sync_walsender_idx; /* for standby millsecond*/ long last_arch_time; + /* for advance slot millsecond*/ + long last_advance_slot_time; } knl_t_arch_context; /* Maximum length of a timezone name (not including trailing null) */