diff --git a/src/gausskernel/storage/access/transam/xlog.cpp b/src/gausskernel/storage/access/transam/xlog.cpp index 7ae1cec49..fc31a2e05 100644 --- a/src/gausskernel/storage/access/transam/xlog.cpp +++ b/src/gausskernel/storage/access/transam/xlog.cpp @@ -535,6 +535,7 @@ static void XLogSelfFlushWithoutStatus(int numHitsOnStartPage, XLogRecPtr CurrPo static void XLogArchiveNotify(const char *xlog); static void XLogArchiveNotifySeg(XLogSegNo segno); static bool XLogArchiveCheckDone(const char *xlog); +static bool HasBeenArchivedOnHaMode(const char* xlog); static bool XLogArchiveIsBusy(const char *xlog); static bool XLogArchiveIsReady(const char *xlog); static void XLogArchiveCleanup(const char *xlog); @@ -2436,6 +2437,14 @@ static bool XLogArchiveCheckDone(const char *xlog) struct stat stat_buf; errno_t errorno = EOK; + /* Only the primary need this step to check this xlog has been archived on standby. + * If this xlog has been archived by all standby which start up the archive thread, + * we should check weather the primary (if primary start up the archive s) + */ + if (!HasBeenArchivedOnHaMode(xlog)) { + return false; + } + /* Always deletable if archiving is off or in recovery process. Archiving is always disabled on standbys. */ if (!XLogArchivingActive() || RecoveryInProgress()) { return true; @@ -2467,6 +2476,59 @@ static bool XLogArchiveCheckDone(const char *xlog) return false; } +static bool HasBeenArchivedOnHaMode(const char* xlog) +{ + load_server_mode(); + if (!t_thrd.xlog_cxt.server_mode == PRIMARY_MODE && !t_thrd.xlog_cxt.server_mode == STANDBY_MODE) { + return true; + } + int mode = t_thrd.xlog_cxt.server_mode; + XLogRecPtr minium_lsn = PG_UINT64_MAX; + for (int i = 0; mode == PRIMARY_MODE && i < g_instance.attr.attr_storage.max_wal_senders; i++) { + /* use volatile pointer to prevent code rearrangement */ + volatile WalSnd* walsnd = &t_thrd.walsender_cxt.WalSndCtl->walsnds[i]; + if (walsnd == NULL) { + continue; + } + + SpinLockAcquire(&walsnd->mutex); + if (IsValidArchiverStandby((WalSnd*)walsnd) && + walsnd->arch_task_last_lsn != 0 && walsnd->arch_task_last_lsn < minium_lsn) { + minium_lsn = walsnd->arch_task_last_lsn; + } + SpinLockRelease(&walsnd->mutex); + } + + if (mode == STANDBY_MODE) { + XLogRecPtr target_lsn = g_instance.archive_standby_cxt.archive_task.targetLsn; + XLogRecPtr start_point = g_instance.archive_standby_cxt.standby_archive_start_point; + minium_lsn = (target_lsn == 0) ? start_point : target_lsn; + } + + if (minium_lsn == PG_UINT64_MAX) { + return true; + } + char minium_archived_xlog_name[MAXFNAMELEN]; + XLogSegNo xlogSegno = 0; + XLByteToSeg(minium_lsn, xlogSegno); + XLogFileName(minium_archived_xlog_name, DEFAULT_TIMELINE_ID, xlogSegno); + + if (mode == STANDBY_MODE) { + /* + * targetLsn or start_point may be the xlog to be archived. Therefore, + * the value of targetLsn or start_point must be smaller than minium_archived_xlog_name + * to ensure that the xlog has been archived. + */ + return strcmp(xlog, minium_archived_xlog_name) < 0; + } + + /* + * if the input xlog name is smaller than or equal to the minium_archived_xlog_name, + * we can confirm that the xlog has been archived by all standby nodes. + */ + return strcmp(xlog, minium_archived_xlog_name) <= 0; +} + /* * XLogArchiveIsBusy * diff --git a/src/gausskernel/storage/replication/syncrep.cpp b/src/gausskernel/storage/replication/syncrep.cpp index 99720b66f..a2fdf69b1 100644 --- a/src/gausskernel/storage/replication/syncrep.cpp +++ b/src/gausskernel/storage/replication/syncrep.cpp @@ -626,7 +626,8 @@ bool SyncRepGetSyncRecPtr(XLogRecPtr *receivePtr, XLogRecPtr *writePtr, XLogRecP */ if ((!(*am_sync) && check_am_sync) || t_thrd.syncrep_cxt.SyncRepConfig == NULL || (!t_thrd.walsender_cxt.WalSndCtl->most_available_sync && - list_length(sync_standbys) < t_thrd.syncrep_cxt.SyncRepConfig->num_sync)) { + list_length(sync_standbys) < t_thrd.syncrep_cxt.SyncRepConfig->num_sync) || + (t_thrd.walsender_cxt.WalSndCtl->most_available_sync && list_length(sync_standbys) == 0)) { list_free(sync_standbys); return false; } diff --git a/src/gausskernel/storage/replication/walsender.cpp b/src/gausskernel/storage/replication/walsender.cpp index 6987d6761..91b8e5a38 100644 --- a/src/gausskernel/storage/replication/walsender.cpp +++ b/src/gausskernel/storage/replication/walsender.cpp @@ -3130,6 +3130,36 @@ static int WalSndLoop(WalSndSendDataCallback send_data) /* Check for input from the client */ ProcessRepliesIfAny(); + /* Only the primary can send the archive lsn to standby */ + load_server_mode(); + if (t_thrd.xlog_cxt.server_mode == PRIMARY_MODE && IsValidArchiverStandby(t_thrd.walsender_cxt.MyWalSnd)) { + XLogRecPtr receivePtr; + XLogRecPtr writePtr; + XLogRecPtr flushPtr; + XLogRecPtr replayPtr; + bool amSync = false; + bool got_recptr = false; + got_recptr = SyncRepGetSyncRecPtr(&receivePtr, &writePtr, &flushPtr, &replayPtr, &amSync, false); + if (got_recptr) { + ArchiveXlogOnStandby(flushPtr); + } else { + if (t_thrd.syncrep_cxt.SyncRepConfig == NULL || + (t_thrd.walsender_cxt.WalSndCtl->most_available_sync && + list_length(SyncRepGetSyncStandbys(&amSync)) == 0)) { + ArchiveXlogOnStandby(t_thrd.walsender_cxt.MyWalSnd->flush); + } else { + ereport(WARNING, (errcode(ERRCODE_WARNING), + errmsg("ArchiveXlogOnStandby failed when call SyncRepGetSyncRecPtr"))); + } + } + } + + volatile unsigned int *standby_archive_flag = &t_thrd.walsender_cxt.MyWalSnd->standby_archive_flag; + if (unlikely(pg_atomic_read_u32(standby_archive_flag) == 1)) { + WalSndSendArchiveLsn2Standby(t_thrd.walsender_cxt.MyWalSnd->arch_task_lsn); + pg_atomic_write_u32(standby_archive_flag, 0); + } + /* Walsender first startup, send a keepalive to standby, no need reply. */ if (first_startup) { WalSndKeepalive(false);