Eao3piq4e
/
openGauss-server
forked from huawei/openGauss-server
0
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

rollback standby archive

This commit is contained in:
luozihao 2021-10-19 09:31:58 +08:00
parent aedb48e7e1
commit d31aeb6bbf
19 changed files with 34 additions and 854 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/common/backend/utils/init/globals.cpp
View File

@ -59,7 +59,7 @@ bool open_join_children = true;
bool will_shutdown = false; bool will_shutdown = false;
/* hard-wired binary version number */ /* hard-wired binary version number */
const uint32 GRAND_VERSION_NUM = 92299; const uint32 GRAND_VERSION_NUM = 92298;
const uint32 MATVIEW_VERSION_NUM = 92213; const uint32 MATVIEW_VERSION_NUM = 92213;
const uint32 PARTIALPUSH_VERSION_NUM = 92087; const uint32 PARTIALPUSH_VERSION_NUM = 92087;

176
src/gausskernel/process/postmaster/pgarch.cpp
View File

@ -47,7 +47,6 @@
#include "storage/latch.h" #include "storage/latch.h"
#include "storage/pg_shmem.h" #include "storage/pg_shmem.h"
#include "storage/pmsignal.h" #include "storage/pmsignal.h"
#include "storage/proc.h"
#include "utils/guc.h" #include "utils/guc.h"
#include "utils/ps_status.h" #include "utils/ps_status.h"
@ -60,7 +59,6 @@
#include "pgxc/pgxc.h" #include "pgxc/pgxc.h"
#include "replication/obswalreceiver.h" #include "replication/obswalreceiver.h"
#include "replication/walreceiver.h" #include "replication/walreceiver.h"
#include "access/xlogreader.h"
/* ---------- /* ----------
* Timer definitions. * Timer definitions.
* ---------- * ----------
@ -91,20 +89,6 @@
#define ARCHIVE_BUF_SIZE (1024 * 1024) #define ARCHIVE_BUF_SIZE (1024 * 1024)
/*
* Timeout interval for sending the status of the archive thread on the standby node
*/
#define UPDATE_STATUS_WAIT 3000
/* the status of archiver on standby */
#define ARCHIVE_ON true
#define ARCHIVE_OFF false
#define STANDBY_ARCHIVE_XLOG_UPGRADE_VERSION 92299
/* the interval of sending archive status is 1s. */
#define SEND_ARCHIVE_STATUS_INTERVAL 1000000L
NON_EXEC_STATIC void PgArchiverMain(); NON_EXEC_STATIC void PgArchiverMain();
static void pgarch_exit(SIGNAL_ARGS); static void pgarch_exit(SIGNAL_ARGS);
static void ArchSigHupHandler(SIGNAL_ARGS); static void ArchSigHupHandler(SIGNAL_ARGS);
@ -117,7 +101,6 @@ static bool pgarch_archiveXlog(char* xlog);
static bool pgarch_readyXlog(char* xlog, int xlog_length); static bool pgarch_readyXlog(char* xlog, int xlog_length);
static void pgarch_archiveDone(const char* xlog); static void pgarch_archiveDone(const char* xlog);
static void pgarch_archiveRoachForPitrStandby(); static void pgarch_archiveRoachForPitrStandby();
static void CreateArchiveReadyFile();
static bool pgarch_archiveRoachForPitrMaster(XLogRecPtr targetLsn); static bool pgarch_archiveRoachForPitrMaster(XLogRecPtr targetLsn);
static bool pgarch_archiveRoachForCoordinator(XLogRecPtr targetLsn); static bool pgarch_archiveRoachForCoordinator(XLogRecPtr targetLsn);
static WalSnd* pgarch_chooseWalsnd(XLogRecPtr targetLsn); static WalSnd* pgarch_chooseWalsnd(XLogRecPtr targetLsn);
@ -125,11 +108,6 @@ typedef bool(*doArchive)(XLogRecPtr);
static void pgarch_ArchiverObsCopyLoop(XLogRecPtr flushPtr, doArchive fun); static void pgarch_ArchiverObsCopyLoop(XLogRecPtr flushPtr, doArchive fun);
static void archKill(int code, Datum arg); static void archKill(int code, Datum arg);
static void initLastTaskLsn(); static void initLastTaskLsn();
static bool SendArchiveThreadStatusInternal(bool is_archive_activited);
static void SendArchiveThreadStatus(bool status);
static bool NotifyPrimaryArchiverActived(ThreadId* last_walrcv_pid, bool* sent_archive_status);
static void NotifyPrimaryArchiverShutdown(bool sent_archive_status);
static void ChangeArchiveTaskStatus2Done(const char* xlog);
AlarmCheckResult DataInstArchChecker(Alarm* alarm, AlarmAdditionalParam* additionalParam) AlarmCheckResult DataInstArchChecker(Alarm* alarm, AlarmAdditionalParam* additionalParam)
{ {
@ -241,7 +219,6 @@ NON_EXEC_STATIC void PgArchiverMain()
*/ */
init_ps_display("archiver process", "", "", ""); init_ps_display("archiver process", "", "", "");
setObsArchLatch(&t_thrd.arch.mainloop_latch); setObsArchLatch(&t_thrd.arch.mainloop_latch);
SetStandbyArchLatch(&t_thrd.arch.mainloop_latch);
initLastTaskLsn(); initLastTaskLsn();
pgarch_MainLoop(); pgarch_MainLoop();
@ -337,8 +314,6 @@ static void pgarch_MainLoop(void)
gettimeofday(&last_copy_time, NULL); gettimeofday(&last_copy_time, NULL);
bool time_to_stop = false; bool time_to_stop = false;
doArchive fun = NULL; doArchive fun = NULL;
bool sent_archive_status = false;
ThreadId last_walrcv_pid = 0;
/* /*
* We run the copy loop immediately upon entry, in case there are * We run the copy loop immediately upon entry, in case there are
@ -370,7 +345,6 @@ static void pgarch_MainLoop(void)
t_thrd.arch.got_SIGHUP = false; t_thrd.arch.got_SIGHUP = false;
ProcessConfigFile(PGC_SIGHUP); ProcessConfigFile(PGC_SIGHUP);
if (!XLogArchivingActive()) { if (!XLogArchivingActive()) {
NotifyPrimaryArchiverShutdown(sent_archive_status);
ereport(LOG, (errmsg("PgArchiver exit"))); ereport(LOG, (errmsg("PgArchiver exit")));
return; return;
} }
@ -393,26 +367,12 @@ static void pgarch_MainLoop(void)
} }
load_server_mode(); load_server_mode();
if (IsServerModeStandby()) { if (IsServerModeStandby()) {
/*
* this step was used by standby to notify primary the archive thread is actived
*/
if (obs_archive_slot == NULL && t_thrd.proc->workingVersionNum >= STANDBY_ARCHIVE_XLOG_UPGRADE_VERSION &&
!NotifyPrimaryArchiverActived(&last_walrcv_pid, &sent_archive_status)) {
pg_usleep(SEND_ARCHIVE_STATUS_INTERVAL);
continue;
}
/* if we should do pitr archive, for standby */ /* if we should do pitr archive, for standby */
volatile unsigned int *pitr_task_status = &g_instance.archive_obs_cxt.pitr_task_status; volatile unsigned int *pitr_task_status = &g_instance.archive_obs_cxt.pitr_task_status;
if (unlikely(pg_atomic_read_u32(pitr_task_status) == PITR_TASK_GET)) { if (unlikely(pg_atomic_read_u32(pitr_task_status) == PITR_TASK_GET)) {
pgarch_archiveRoachForPitrStandby(); pgarch_archiveRoachForPitrStandby();
pg_atomic_write_u32(pitr_task_status, PITR_TASK_DONE); pg_atomic_write_u32(pitr_task_status, PITR_TASK_DONE);
} }
/* if we should do archive by standby, we should create a .ready file*/
volatile unsigned int* arch_task_status = &g_instance.archive_standby_cxt.arch_task_status;
if (unlikely(pg_atomic_read_u32(arch_task_status) == ARCH_TASK_GET)) {
CreateArchiveReadyFile();
}
} }
/* Do what we're here for */ /* Do what we're here for */
@ -466,7 +426,7 @@ static void pgarch_MainLoop(void)
wait_interval = t_thrd.arch.task_wait_interval; wait_interval = t_thrd.arch.task_wait_interval;
last_time = t_thrd.arch.last_arch_time; last_time = t_thrd.arch.last_arch_time;
} else { } else {
wait_interval = IsServerModeStandby() ? t_thrd.arch.task_wait_interval : PGARCH_AUTOWAKE_INTERVAL; wait_interval = PGARCH_AUTOWAKE_INTERVAL;
last_time = TIME_GET_MILLISEC(last_copy_time); last_time = TIME_GET_MILLISEC(last_copy_time);
} }
gettimeofday(&curtime, NULL); gettimeofday(&curtime, NULL);
@ -504,15 +464,6 @@ static void pgarch_MainLoop(void)
* SIGUSR2. * SIGUSR2.
*/ */
} while (PostmasterIsAlive() && XLogArchivingActive() && !time_to_stop); } while (PostmasterIsAlive() && XLogArchivingActive() && !time_to_stop);
/*
* This step don't need to check the version number, because we have checked the version in the loop.
*
* If the version number is bigger than STANDBY_ARCHIVE_XLOG_UPGRADE_VERSION and we notify the primary successfully
* in the loop, the sent_archive_status will be set to true. We don't need to send the archive status when
* the sent_archive_status is false, so we don't need to check the version number in this step.
*/
NotifyPrimaryArchiverShutdown(sent_archive_status);
} }
/* /*
@ -566,9 +517,6 @@ static void pgarch_ArchiverCopyLoop(void)
if (pgarch_archiveXlog(xlog)) { if (pgarch_archiveXlog(xlog)) {
/* successful */ /* successful */
pgarch_archiveDone(xlog); pgarch_archiveDone(xlog);
/* archive xlog on standby success, we need to change the status of archive task. */
ChangeArchiveTaskStatus2Done(xlog);
break; /* out of inner retry loop */ break; /* out of inner retry loop */
} else { } else {
if (++failures >= NUM_ARCHIVE_RETRIES) { if (++failures >= NUM_ARCHIVE_RETRIES) {
@ -987,43 +935,6 @@ static void pgarch_archiveRoachForPitrStandby()
} }
} }
static void CreateArchiveReadyFile()
{
XLogRecPtr targetLsn = g_instance.archive_standby_cxt.archive_task.targetLsn;
if (!XlogFileIsExisted(t_thrd.proc_cxt.DataDir, targetLsn, DEFAULT_TIMELINE_ID)) {
ereport(WARNING, (errmsg("CreateArchiveReadyFile: the %X/%X is not exists, skipping",
(uint32)(targetLsn >> 32), (uint32)(targetLsn))));
volatile unsigned int* arch_task_status = &g_instance.archive_standby_cxt.arch_task_status;
if (likely(pg_atomic_read_u32(arch_task_status) == ARCH_TASK_GET)) {
pg_memory_barrier();
g_instance.archive_standby_cxt.arch_finish_result = true;
pg_atomic_write_u32(arch_task_status, ARCH_TASK_DONE);
}
return;
}
char xlogfname[MAXFNAMELEN];
ereport(LOG, (errmsg("CreateArchiveReadyFile %X/%X", (uint32)(targetLsn >> 32), (uint32)(targetLsn))));
XLogSegNo xlogSegno = 0;
XLByteToSeg(targetLsn, xlogSegno);
if (xlogSegno == InvalidXLogSegPtr) {
g_instance.archive_standby_cxt.arch_finish_result = false;
ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE), errmsg("Invalid Lsn: %lu", targetLsn)));
}
XLogFileName(xlogfname, DEFAULT_TIMELINE_ID, xlogSegno);
char tempPath[PATH_MAX] = {0};
char srcPath[PATH_MAX + 1] = {0};
int rc = snprintf_s(tempPath, PATH_MAX, PATH_MAX - 1, XLOGDIR "/%s", xlogfname);
securec_check_ss(rc, "\0", "\0");
char* retVal = realpath(tempPath, srcPath);
if (retVal == NULL) {
ereport(WARNING, (errmsg_internal("realpath src %s failed:%m\n", tempPath)));
} else {
XLogArchiveNotify(xlogfname);
}
}
/* /*
* pgarch_archiveRoachForPitrMaster * pgarch_archiveRoachForPitrMaster
* choose a walsender to send archive command * choose a walsender to send archive command
@ -1181,88 +1092,3 @@ static void initLastTaskLsn()
(uint32)(t_thrd.arch.pitr_task_last_lsn)))); (uint32)(t_thrd.arch.pitr_task_last_lsn))));
} }
} }
/*
* This function is used to send the status of archiver
*/
static void SendArchiveThreadStatus(bool status)
{
if (!WalRcvIsOnline()) {
return;
}
while(!SendArchiveThreadStatusInternal(status)) {
ereport(WARNING,
(errcode(ERRCODE_WARNING),
errmsg("Notifing primary to update archive status is failed.")));
pg_usleep(SEND_ARCHIVE_STATUS_INTERVAL);
}
ereport(LOG, (errmsg("Notifing primary to update archive status is success.")));
}
static bool SendArchiveThreadStatusInternal(bool is_archive_activited)
{
ResetLatch(&t_thrd.arch.mainloop_latch);
g_instance.archive_standby_cxt.archive_enabled = is_archive_activited;
(void)gs_signal_send(g_instance.pid_cxt.WalReceiverPID, SIGUSR2);
g_instance.archive_standby_cxt.arch_latch = &t_thrd.arch.mainloop_latch;
int rc = WaitLatch(&t_thrd.arch.mainloop_latch,
WL_LATCH_SET | WL_TIMEOUT | WL_POSTMASTER_DEATH,
(long)(UPDATE_STATUS_WAIT));
if (rc & WL_POSTMASTER_DEATH) {
gs_thread_exit(1);
}
if (rc & WL_TIMEOUT) {
return false;
}
return true;
}
static bool NotifyPrimaryArchiverActived(ThreadId* last_walrcv_pid, bool* sent_archive_status)
{
if (WalRcvIsOnline()) {
/* last_walrcv_pid is used to check whether the primary node is switched over. */
if (!(*sent_archive_status) || *last_walrcv_pid != g_instance.pid_cxt.WalReceiverPID) {
SendArchiveThreadStatus(ARCHIVE_ON);
*sent_archive_status = true;
*last_walrcv_pid = g_instance.pid_cxt.WalReceiverPID;
return true;
}
if (*sent_archive_status) {
return true;
}
}
*sent_archive_status = false;
return false;
}
static void NotifyPrimaryArchiverShutdown(bool sent_archive_status)
{
if (sent_archive_status) {
SendArchiveThreadStatus(ARCHIVE_OFF);
}
}
static void ChangeArchiveTaskStatus2Done(const char* xlog)
{
if (IsServerModeStandby()) {
char fname[MAXFNAMELEN];
int segno = 0;
XLogRecPtr targetLsn = g_instance.archive_standby_cxt.archive_task.targetLsn;
XLByteToSeg(targetLsn, segno);
XLogFileName(fname, DEFAULT_TIMELINE_ID, segno);
if (strcmp(xlog, fname) != 0) {
/*
* The archived xlog is not the xlog required by the archive task.
* Therefore, the archived xlog cannot be returned to the primary.
*/
ereport(LOG, (errmsg("\"%s\" is not archived target, no reply is sent", xlog)));
return;
}
volatile unsigned int *arch_task_status = &g_instance.archive_standby_cxt.arch_task_status;
if (unlikely(pg_atomic_read_u32(arch_task_status) == ARCH_TASK_GET)) {
pg_memory_barrier();
g_instance.archive_standby_cxt.arch_finish_result = true;
pg_atomic_write_u32(arch_task_status, ARCH_TASK_DONE);
}
}
}

13
src/gausskernel/process/postmaster/postmaster.cpp
View File

@ -313,6 +313,10 @@ char g_bbox_dump_path[1024] = {0};
} \ } \
} while (0) } while (0)
#define WalRcvIsOnline() \
((g_instance.pid_cxt.WalReceiverPID != 0 && t_thrd.walreceiverfuncs_cxt.WalRcv && \
t_thrd.walreceiverfuncs_cxt.WalRcv->isRuning))
#define DataRcvIsOnline() \ #define DataRcvIsOnline() \
((IS_DN_DUMMY_STANDYS_MODE() ? (g_instance.pid_cxt.DataReceiverPID != 0 && t_thrd.datareceiver_cxt.DataRcv && \ ((IS_DN_DUMMY_STANDYS_MODE() ? (g_instance.pid_cxt.DataReceiverPID != 0 && t_thrd.datareceiver_cxt.DataRcv && \
t_thrd.datareceiver_cxt.DataRcv->isRuning) \ t_thrd.datareceiver_cxt.DataRcv->isRuning) \
@ -2791,6 +2795,7 @@ static int ServerLoop(void)
} }
} }
} }
/* If we have lost the stats collector, try to start a new one */ /* If we have lost the stats collector, try to start a new one */
if (g_instance.pid_cxt.PgStatPID == 0 && (pmState == PM_RUN || pmState == PM_HOT_STANDBY) && !dummyStandbyMode) if (g_instance.pid_cxt.PgStatPID == 0 && (pmState == PM_RUN || pmState == PM_HOT_STANDBY) && !dummyStandbyMode)
g_instance.pid_cxt.PgStatPID = pgstat_start(); g_instance.pid_cxt.PgStatPID = pgstat_start();
@ -4867,6 +4872,7 @@ static void reaper(SIGNAL_ARGS)
#define LOOPHEADER() (exitstatus = (long)(intptr_t)status) #define LOOPHEADER() (exitstatus = (long)(intptr_t)status)
gs_signal_setmask(&t_thrd.libpq_cxt.BlockSig, NULL); gs_signal_setmask(&t_thrd.libpq_cxt.BlockSig, NULL);
ReplicationSlot *obs_slot = NULL;
ereport(DEBUG4, (errmsg_internal("reaping dead processes"))); ereport(DEBUG4, (errmsg_internal("reaping dead processes")));
for (;;) { for (;;) {
@ -5412,8 +5418,13 @@ static void reaper(SIGNAL_ARGS)
LogChildExit(LOG, _("archiver process"), pid, exitstatus); LogChildExit(LOG, _("archiver process"), pid, exitstatus);
if (XLogArchivingActive()) { if (XLogArchivingActive()) {
if (pmState == PM_RUN || pmState == PM_HOT_STANDBY || pmState == PM_RECOVERY) { if (pmState == PM_RUN) {
g_instance.pid_cxt.PgArchPID = pgarch_start(); g_instance.pid_cxt.PgArchPID = pgarch_start();
}else if (pmState == PM_HOT_STANDBY) {
obs_slot = getObsReplicationSlot();
if (obs_slot != NULL) {
g_instance.pid_cxt.PgArchPID = pgarch_start();
}
} }
} }
continue; continue;

19
src/gausskernel/process/threadpool/knl_instance.cpp
View File

@ -490,24 +490,6 @@ static void knl_g_archive_obs_init(knl_g_archive_obs_context *archive_obs_cxt)
INSTANCE_GET_MEM_CXT_GROUP(MEMORY_CONTEXT_STORAGE), sizeof(ReplicationSlot)); INSTANCE_GET_MEM_CXT_GROUP(MEMORY_CONTEXT_STORAGE), sizeof(ReplicationSlot));
} }
static void knl_g_archive_standby_init(knl_g_archive_standby_context* archive_standby_cxt)
{
Assert(archive_standby_cxt != NULL);
errno_t rc = memset_s(archive_standby_cxt, sizeof(knl_g_archive_standby_context), 0, sizeof(knl_g_archive_standby_context));
securec_check(rc, "\0", "\0");
archive_standby_cxt->arch_task_status = 0;
archive_standby_cxt->arch_finish_result = false;
archive_standby_cxt->need_to_send_archive_status = false;
/* we don't need to use this parameter, but we should init it. */
archive_standby_cxt->archive_task.sub_term = -1;
archive_standby_cxt->archive_task.term = 0;
archive_standby_cxt->archive_task.targetLsn = 0;
archive_standby_cxt->archive_enabled = false;
archive_standby_cxt->arch_latch = NULL;
}
#ifdef ENABLE_MOT #ifdef ENABLE_MOT
static void knl_g_mot_init(knl_g_mot_context* mot_cxt) static void knl_g_mot_init(knl_g_mot_context* mot_cxt)
{ {
@ -592,7 +574,6 @@ void knl_instance_init()
knl_g_wal_init(&g_instance.wal_cxt); knl_g_wal_init(&g_instance.wal_cxt);
knl_g_oid_nodename_cache_init(&g_instance.oid_nodename_cache); knl_g_oid_nodename_cache_init(&g_instance.oid_nodename_cache);
knl_g_archive_obs_init(&g_instance.archive_obs_cxt); knl_g_archive_obs_init(&g_instance.archive_obs_cxt);
knl_g_archive_standby_init(&g_instance.archive_standby_cxt);
knl_g_hypo_init(&g_instance.hypo_cxt); knl_g_hypo_init(&g_instance.hypo_cxt);
} }

75
src/gausskernel/storage/access/transam/xlog.cpp
View File

@ -532,8 +532,9 @@ static void XLogFlushCore(XLogRecPtr WriteRqstPtr);
static void XLogSelfFlush(void); static void XLogSelfFlush(void);
static void XLogSelfFlushWithoutStatus(int numHitsOnStartPage, XLogRecPtr CurrPos, int currLRC); static void XLogSelfFlushWithoutStatus(int numHitsOnStartPage, XLogRecPtr CurrPos, int currLRC);
static void XLogArchiveNotify(const char *xlog);
static void XLogArchiveNotifySeg(XLogSegNo segno); static void XLogArchiveNotifySeg(XLogSegNo segno);
static bool HasBeenArchivedOnHaMode(const char* xlog); static bool XLogArchiveCheckDone(const char *xlog);
static bool XLogArchiveIsBusy(const char *xlog); static bool XLogArchiveIsBusy(const char *xlog);
static bool XLogArchiveIsReady(const char *xlog); static bool XLogArchiveIsReady(const char *xlog);
static void XLogArchiveCleanup(const char *xlog); static void XLogArchiveCleanup(const char *xlog);
@ -2325,7 +2326,7 @@ static uint64 XLogRecPtrToBytePos(XLogRecPtr ptr)
* and the archiver then knows to archive XLOGDIR/0000000100000001000000C6, * and the archiver then knows to archive XLOGDIR/0000000100000001000000C6,
* then when complete, rename it to 0000000100000001000000C6.done * then when complete, rename it to 0000000100000001000000C6.done
*/ */
void XLogArchiveNotify(const char *xlog) static void XLogArchiveNotify(const char *xlog)
{ {
char archiveStatusPath[MAXPGPATH]; char archiveStatusPath[MAXPGPATH];
FILE *fd = NULL; FILE *fd = NULL;
@ -2430,20 +2431,12 @@ void XLogArchiveForceDone(const char *xlog)
* The reason we do things this way is so that if the original attempt to * The reason we do things this way is so that if the original attempt to
* create <XLOG>.ready fails, we'll retry during subsequent checkpoints. * create <XLOG>.ready fails, we'll retry during subsequent checkpoints.
*/ */
bool XLogArchiveCheckDone(const char *xlog) static bool XLogArchiveCheckDone(const char *xlog)
{ {
char archiveStatusPath[MAXPGPATH]; char archiveStatusPath[MAXPGPATH];
struct stat stat_buf; struct stat stat_buf;
errno_t errorno = EOK; 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. */ /* Always deletable if archiving is off or in recovery process. Archiving is always disabled on standbys. */
if (!XLogArchivingActive() || RecoveryInProgress()) { if (!XLogArchivingActive() || RecoveryInProgress()) {
return true; return true;
@ -2475,66 +2468,6 @@ bool XLogArchiveCheckDone(const char *xlog)
return false; return false;
} }
static bool HasBeenArchivedOnHaMode(const char* xlog)
{
/*
* Generally, the validity of the xlog transferred from the upper layer has been verified.
* Therefore, if the length of the xlog name transferred is greater than the standard length
* of the xlog name (24 characters), the transferred file is a .backup file.
* Therefore, if the xlog name contains more than 24 characters, return true directly.
*/
if (strlen(xlog) > XLOG_NAME_LENGTH) {
return true;
}
load_server_mode();
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 && XLogArchivingActive()) {
XLogRecPtr target_lsn = g_instance.archive_standby_cxt.archive_task.targetLsn;
minium_lsn = 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 * XLogArchiveIsBusy
* *

6
src/gausskernel/storage/replication/syncrep.cpp
View File

@ -631,14 +631,10 @@ bool SyncRepGetSyncRecPtr(XLogRecPtr *receivePtr, XLogRecPtr *writePtr, XLogRecP
* or there are not enough synchronous standbys. * or there are not enough synchronous standbys.
* but in a particular scenario, when most_available_sync is true, primary only wait the alive sync standbys * but in a particular scenario, when most_available_sync is true, primary only wait the alive sync standbys
* if list_length(sync_standbys) doesn't satisfy t_thrd.syncrep_cxt.SyncRepConfig->num_sync. * if list_length(sync_standbys) doesn't satisfy t_thrd.syncrep_cxt.SyncRepConfig->num_sync.
*
* All synchronous standbys are allowed to disconnect from the host
* only when the maximum available mode is on
*/ */
if ((!(*am_sync) && check_am_sync) || t_thrd.syncrep_cxt.SyncRepConfig == NULL || if ((!(*am_sync) && check_am_sync) || t_thrd.syncrep_cxt.SyncRepConfig == NULL ||
(!t_thrd.walsender_cxt.WalSndCtl->most_available_sync && (!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); list_free(sync_standbys);
return false; return false;
} }

5
src/gausskernel/storage/replication/walrcvwriter.cpp
View File

@ -140,6 +140,8 @@ static void XLogWalRcvWrite(WalRcvCtlBlock *walrcb, char *buf, Size nbytes, XLog
* would otherwise have to reopen this file to fsync it later * would otherwise have to reopen this file to fsync it later
*/ */
if (recvFile >= 0) { if (recvFile >= 0) {
char xlogfname[MAXFNAMELEN];
/* /*
* XLOG segment files will be re-read by recovery in startup * XLOG segment files will be re-read by recovery in startup
* process soon, so we don't advise the OS to release cache * process soon, so we don't advise the OS to release cache
@ -150,15 +152,12 @@ static void XLogWalRcvWrite(WalRcvCtlBlock *walrcb, char *buf, Size nbytes, XLog
errmsg("could not close log file %s: %m", errmsg("could not close log file %s: %m",
XLogFileNameP(t_thrd.xlog_cxt.ThisTimeLineID, recvSegNo)))); XLogFileNameP(t_thrd.xlog_cxt.ThisTimeLineID, recvSegNo))));
#ifdef ENABLE_MULTIPLE_NODES
/* /*
* Create .done file forcibly to prevent the restored segment from * Create .done file forcibly to prevent the restored segment from
* being archived again later. * being archived again later.
*/ */
char xlogfname[MAXFNAMELEN];
XLogFileName(xlogfname, recvFileTLI, recvSegNo); XLogFileName(xlogfname, recvFileTLI, recvSegNo);
XLogArchiveForceDone(xlogfname); XLogArchiveForceDone(xlogfname);
#endif
} }
recvFile = -1; recvFile = -1;

225
src/gausskernel/storage/replication/walreceiver.cpp
View File

@ -87,8 +87,6 @@ bool wal_catchup = false;
#define TEMP_CONF_FILE "postgresql.conf.bak" #define TEMP_CONF_FILE "postgresql.conf.bak"
#define MAX_PATH 256
const char *g_reserve_param[RESERVE_SIZE] = { const char *g_reserve_param[RESERVE_SIZE] = {
"application_name", "application_name",
"archive_command", "archive_command",
@ -169,7 +167,6 @@ static void WalRcvSigHupHandler(SIGNAL_ARGS);
static void WalRcvShutdownHandler(SIGNAL_ARGS); static void WalRcvShutdownHandler(SIGNAL_ARGS);
static void WalRcvQuickDieHandler(SIGNAL_ARGS); static void WalRcvQuickDieHandler(SIGNAL_ARGS);
static void sigusr1_handler(SIGNAL_ARGS); static void sigusr1_handler(SIGNAL_ARGS);
static void sigusr2_handler(SIGNAL_ARGS);
static void ConfigFileTimer(void); static void ConfigFileTimer(void);
static bool ProcessConfigFileMessage(char *buf, Size len); static bool ProcessConfigFileMessage(char *buf, Size len);
static void firstSynchStandbyFile(void); static void firstSynchStandbyFile(void);
@ -178,12 +175,8 @@ static TimestampTz GetHeartbeatLastReplyTimestamp();
static bool WalRecCheckTimeOut(TimestampTz nowtime, TimestampTz last_recv_timestamp, bool ping_sent); static bool WalRecCheckTimeOut(TimestampTz nowtime, TimestampTz last_recv_timestamp, bool ping_sent);
static void WalRcvRefreshPercentCountStartLsn(XLogRecPtr currentMaxLsn, XLogRecPtr currentDoneLsn); static void WalRcvRefreshPercentCountStartLsn(XLogRecPtr currentMaxLsn, XLogRecPtr currentDoneLsn);
static void ProcessArchiveXlogMessage(const ArchiveXlogMessage* archive_xlog_message); static void ProcessArchiveXlogMessage(const ArchiveXlogMessage* archive_xlog_message);
static void ProcessStandbyArchiveXlogMessage(const ArchiveXlogMessage* archive_xlog_message);
static void WalRecvSendArchiveXlogResponse(); static void WalRecvSendArchiveXlogResponse();
static void WalRecvSendArchiveXlogResult2Standby();
static void SendArchiveStatus(bool status);
static void ProcessArchiveStatusResponse(ArchiveStatusResponseMessage* response);
static XLogRecPtr InitArchiveStartPoint();
void ProcessWalRcvInterrupts(void) void ProcessWalRcvInterrupts(void)
{ {
/* /*
@ -306,15 +299,6 @@ void setObsArchLatch(const Latch* latch)
SpinLockRelease(&walrcv->mutex); SpinLockRelease(&walrcv->mutex);
} }
void SetStandbyArchLatch(const Latch* latch)
{
/* use volatile pointer to prevent code rearrangement */
volatile WalRcvData *walrcv = t_thrd.walreceiverfuncs_cxt.WalRcv;
SpinLockAcquire(&walrcv->mutex);
walrcv->arch_latch = (Latch *)latch;
SpinLockRelease(&walrcv->mutex);
}
static void wakeupObsArchLatch() static void wakeupObsArchLatch()
{ {
/* use volatile pointer to prevent code rearrangement */ /* use volatile pointer to prevent code rearrangement */
@ -326,17 +310,6 @@ static void wakeupObsArchLatch()
SpinLockRelease(&walrcv->mutex); SpinLockRelease(&walrcv->mutex);
} }
static void wakeupArchLatch()
{
/* use volatile pointer to prevent code rearrangement */
volatile WalRcvData *walrcv = t_thrd.walreceiverfuncs_cxt.WalRcv;
SpinLockAcquire(&walrcv->mutex);
if (walrcv->obsArchLatch != NULL) {
SetLatch(walrcv->arch_latch);
}
SpinLockRelease(&walrcv->mutex);
}
void RefuseConnect() void RefuseConnect()
{ {
WalRcvData *walrcv = t_thrd.walreceiverfuncs_cxt.WalRcv; WalRcvData *walrcv = t_thrd.walreceiverfuncs_cxt.WalRcv;
@ -404,20 +377,6 @@ void WalRcvrProcessData(TimestampTz *last_recv_timestamp, bool *ping_sent)
pg_atomic_write_u32(pitr_task_status, PITR_TASK_NONE); pg_atomic_write_u32(pitr_task_status, PITR_TASK_NONE);
} }
/* send archive status to primary */
if (g_instance.archive_standby_cxt.need_to_send_archive_status) {
g_instance.archive_standby_cxt.need_to_send_archive_status = false;
SendArchiveStatus(g_instance.archive_standby_cxt.archive_enabled);
}
/* response the result of archive to primary */
volatile unsigned int* arch_task_status = &g_instance.archive_standby_cxt.arch_task_status;
if (unlikely(pg_atomic_read_u32(arch_task_status) == ARCH_TASK_DONE)) {
WalRecvSendArchiveXlogResult2Standby();
pg_memory_barrier();
pg_atomic_write_u32(arch_task_status, ARCH_TASK_NONE);
}
if (!WalRcvWriterInProgress()) if (!WalRcvWriterInProgress())
ereport(FATAL, (errmsg("terminating walreceiver process due to the death of walrcvwriter"))); ereport(FATAL, (errmsg("terminating walreceiver process due to the death of walrcvwriter")));
@ -593,7 +552,7 @@ void WalReceiverMain(void)
(void)gspqsignal(SIGALRM, SIG_IGN); (void)gspqsignal(SIGALRM, SIG_IGN);
(void)gspqsignal(SIGPIPE, SIG_IGN); (void)gspqsignal(SIGPIPE, SIG_IGN);
(void)gspqsignal(SIGUSR1, sigusr1_handler); (void)gspqsignal(SIGUSR1, sigusr1_handler);
(void)gspqsignal(SIGUSR2, sigusr2_handler); (void)gspqsignal(SIGUSR2, SIG_IGN);
/* Reset some signals that are accepted by postmaster but not here */ /* Reset some signals that are accepted by postmaster but not here */
(void)gspqsignal(SIGCHLD, SIG_DFL); (void)gspqsignal(SIGCHLD, SIG_DFL);
@ -918,18 +877,6 @@ static void sigusr1_handler(SIGNAL_ARGS)
errno = save_errno; errno = save_errno;
} }
static void sigusr2_handler(SIGNAL_ARGS) {
/* get sigusr2 */
int save_errno = errno;
gs_signal_setmask(&t_thrd.libpq_cxt.BlockSig, NULL);
/*
* sending archive status to the primary in this step
*/
g_instance.archive_standby_cxt.need_to_send_archive_status = true;
gs_signal_setmask(&t_thrd.libpq_cxt.UnBlockSig, NULL);
errno = save_errno;
}
/* Wal receiver is shut down? */ /* Wal receiver is shut down? */
bool WalRcvIsShutdown(void) bool WalRcvIsShutdown(void)
{ {
@ -1085,25 +1032,6 @@ static void XLogWalRcvProcessMsg(unsigned char type, char *buf, Size len)
ProcessArchiveXlogMessage(&archiveXLogMessage); ProcessArchiveXlogMessage(&archiveXLogMessage);
break; break;
} }
case 'n' : /* process the archive task message sent by primary */
{
ArchiveXlogMessage archiveXLogMessage;
CHECK_MSG_SIZE(len, ArchiveXlogMessage, "invalid ArchiveXlogMessage message received from primary");
/* memcpy is required here for alignment reasons */
errorno = memcpy_s(&archiveXLogMessage, sizeof(ArchiveXlogMessage), buf, sizeof(ArchiveXlogMessage));
securec_check(errorno, "\0", "\0");
ProcessStandbyArchiveXlogMessage(&archiveXLogMessage);
break;
}
case 'S': /* send the status of the archive thread on standby */
{
ArchiveStatusResponseMessage response;
CHECK_MSG_SIZE(len, ArchiveStatusResponseMessage, "invalid ArchiveStatusResponseMessage message received from primary");
errorno = memcpy_s(&response, sizeof(ArchiveStatusResponseMessage), buf, sizeof(ArchiveStatusResponseMessage));
securec_check(errorno, "\0", "\0");
ProcessArchiveStatusResponse(&response);
break;
}
default: default:
ereport(ERROR, (errcode(ERRCODE_PROTOCOL_VIOLATION), ereport(ERROR, (errcode(ERRCODE_PROTOCOL_VIOLATION),
errmsg_internal("invalid replication message type %c", type))); errmsg_internal("invalid replication message type %c", type)));
@ -1950,36 +1878,6 @@ static void ProcessArchiveXlogMessage(const ArchiveXlogMessage* archive_xlog_mes
wakeupObsArchLatch(); wakeupObsArchLatch();
} }
/*
* Process ProcessStandbyArchiveXlogMessage received from primary sender, message type is 'n'.
*/
static void ProcessStandbyArchiveXlogMessage(const ArchiveXlogMessage* archive_xlog_message)
{
ereport(LOG, (errmsg("ProcessStandbyArchiveXlogMessage: get archive xlog message :%X/%X",
(uint32)(archive_xlog_message->targetLsn >> 32), (uint32)(archive_xlog_message->targetLsn))));
errno_t errorno = EOK;
volatile unsigned int* arch_task_status = &g_instance.archive_standby_cxt.arch_task_status;
unsigned int expected = ARCH_TASK_NONE;
int failed_times = 0;
while (pg_atomic_compare_exchange_u32(arch_task_status, &expected, ARCH_TASK_GET) == false) {
/* some task arrived before last task done if expected not equal to NONE */
expected = ARCH_TASK_NONE;
pg_usleep(ARCHIVE_XLOG_DELAY); // sleep 0.01s
if (failed_times++ >= GET_ARCHIVE_XLOG_RETRY_MAX) {
ereport(WARNING, (errmsg("get archive xlog message :%X/%X, but not finished",
(uint32)(archive_xlog_message->targetLsn >> 32),
(uint32)(archive_xlog_message->targetLsn))));
return;
}
}
errorno = memcpy_s(&g_instance.archive_standby_cxt.archive_task,
sizeof(ArchiveXlogMessage) + 1,
archive_xlog_message,
sizeof(ArchiveXlogMessage));
securec_check(errorno, "\0", "\0");
wakeupArchLatch();
}
/* /*
* Send switchover request message to primary, indicating the current time. * Send switchover request message to primary, indicating the current time.
*/ */
@ -2016,30 +1914,19 @@ static void XLogWalRcvSendSwitchRequest(void)
*/ */
static void WalRecvSendArchiveXlogResponse() static void WalRecvSendArchiveXlogResponse()
{ {
char buf[sizeof(ArchiveXlogResponseMessage) + 1]; char buf[sizeof(ArchiveXlogResponseMeeeage) + 1];
ArchiveXlogResponseMessage reply; ArchiveXlogResponseMeeeage reply;
errno_t errorno = EOK; errno_t errorno = EOK;
reply.pitr_result = g_instance.archive_obs_cxt.pitr_finish_result; reply.pitr_result = g_instance.archive_obs_cxt.pitr_finish_result;
reply.targetLsn = g_instance.archive_obs_cxt.archive_task.targetLsn; reply.targetLsn = g_instance.archive_obs_cxt.archive_task.targetLsn;
XLogSegNo segno = 0;
char lastoff[MAXFNAMELEN];
XLByteToSeg(reply.targetLsn, segno);
XLogFileName(lastoff, reply.targetLsn, segno);
if (!reply.pitr_result) {
reply.archive_result = ARCHIVE_FAILED;
} else if (XLogArchiveCheckDone(lastoff)) {
reply.archive_result = ARCHIVE_SUCCESS;
} else {
reply.archive_result = ARCHIVE_SKIP;
}
buf[0] = 'a'; buf[0] = 'a';
errorno = memcpy_s(&buf[1], errorno = memcpy_s(&buf[1],
sizeof(ArchiveXlogResponseMessage), sizeof(ArchiveXlogResponseMeeeage),
&reply, &reply,
sizeof(ArchiveXlogResponseMessage)); sizeof(ArchiveXlogResponseMeeeage));
securec_check(errorno, "\0", "\0"); securec_check(errorno, "\0", "\0");
libpqrcv_send(buf, sizeof(ArchiveXlogResponseMessage) + 1); libpqrcv_send(buf, sizeof(ArchiveXlogResponseMeeeage) + 1);
ereport(LOG, ereport(LOG,
(errmsg("WalRecvSendArchiveXlogResponse %d %X/%X", reply.pitr_result, (errmsg("WalRecvSendArchiveXlogResponse %d %X/%X", reply.pitr_result,
@ -2047,30 +1934,6 @@ static void WalRecvSendArchiveXlogResponse()
} }
/*
* Send archive xlog result message to primary.
*/
static void WalRecvSendArchiveXlogResult2Standby()
{
char buf[sizeof(ArchiveXlogResponseMessage) + 1];
ArchiveXlogResponseMessage reply;
errno_t errorno = EOK;
reply.pitr_result = g_instance.archive_standby_cxt.arch_finish_result;
reply.targetLsn = g_instance.archive_standby_cxt.archive_task.targetLsn;
buf[0] = 'n';
errorno = memcpy_s(&buf[1],
sizeof(ArchiveXlogResponseMessage),
&reply,
sizeof(ArchiveXlogResponseMessage));
securec_check(errorno, "\0", "\0");
libpqrcv_send(buf, sizeof(ArchiveXlogResponseMessage) + 1);
const char* archive_result_string = ((reply.pitr_result) ? "success" : "fail");
ereport(LOG,
(errmsg("WalRecvSendArchiveXlogResult2Standby archive_result:%s archive_lsn:%X/%X", archive_result_string,
(uint32)(reply.targetLsn >> 32), (uint32)(reply.targetLsn))));
}
/* /*
* process switchover response message from primary. * process switchover response message from primary.
*/ */
@ -2878,77 +2741,3 @@ static void WalRcvRefreshPercentCountStartLsn(XLogRecPtr currentMaxLsn, XLogRecP
WalRcvSetPercentCountStartLsn(currentDoneLsn); WalRcvSetPercentCountStartLsn(currentDoneLsn);
} }
} }
/*
* SendArchiveStatus
* This function is used to notify the primary of
* whether archiving is enabled for standby.
*
* If the value of status is true, archiving is enabled for the standby node.
* Otherwise, archiving is disabled.
*
* The targetLsn is the minimum lsn which is not archived.
*/
static void SendArchiveStatus(bool status)
{
char buf[sizeof(ArchiveStatusMessage) + 1];
ArchiveStatusMessage message;
errno_t errorno = EOK;
message.is_archive_activied = status;
XLogRecPtr start_point = InitArchiveStartPoint();
const char* status_string = status ? "on" : "off";
ereport(LOG, (errmsg("the archive thread status is %s, and the start point of lsn is %X/%X",
status_string, (uint32)(start_point >> 32), (uint32)start_point)));
message.startLsn = start_point;
buf[0] = 'S';
errorno = memcpy_s(&buf[1],
sizeof(ArchiveStatusMessage),
&message,
sizeof(ArchiveStatusMessage));
securec_check(errorno, "\0", "\0");
volatile unsigned int* arch_task_status = &g_instance.archive_standby_cxt.arch_task_status;
pg_atomic_write_u32(arch_task_status, ARCH_TASK_NONE);
libpqrcv_send(buf, sizeof(ArchiveStatusMessage) + 1);
}
static void ProcessArchiveStatusResponse(ArchiveStatusResponseMessage* response)
{
ereport(LOG, (errmsg("get updating archive status response")));
bool is_success = response->is_set_status_success;
if (is_success) {
SetLatch(g_instance.archive_standby_cxt.arch_latch);
} else {
ereport(WARNING,
(errcode(ERRCODE_WARNING),
errmsg("it is failed to notify the primary that updating the archive status of standby")));
}
}
/*
* InitArchiveStartPoint is used to find a start point which is the standby should check
*/
static XLogRecPtr InitArchiveStartPoint()
{
XLogRecPtr flushPtr = t_thrd.walreceiverfuncs_cxt.WalRcv->walRcvCtlBlock->flushPtr;
XLogRecPtr targetLsn = flushPtr - XLogSegSize;
/*
* Start from the previous xlog file of the flushptr.
* If the xlog file exists, it traverses backwards until the xlog file is archived.
* The next file is the oldest xlog file that has not been archived.
*/
while (XlogFileIsExisted(t_thrd.proc_cxt.DataDir, targetLsn, DEFAULT_TIMELINE_ID)) {
int errorno = 0;
XLogSegNo segno = 0;
char lastoff[MAXFNAMELEN];
XLByteToSeg(targetLsn, segno);
errorno = snprintf_s(lastoff, MAXFNAMELEN, MAXFNAMELEN - 1, "%08X%08X%08X", t_thrd.xlog_cxt.ThisTimeLineID,
(uint32)((segno) / XLogSegmentsPerXLogId), (uint32)((segno) % XLogSegmentsPerXLogId));
if (XLogArchiveCheckDone(lastoff)) {
return targetLsn + XLogSegSize;
}
targetLsn -= XLogSegSize;
}
return targetLsn + XLogSegSize;
}

256
src/gausskernel/storage/replication/walsender.cpp
View File

@ -116,8 +116,6 @@ long g_logical_slot_sleep_time = 0;
#define AmWalSenderToDummyStandby() (t_thrd.walsender_cxt.MyWalSnd->sendRole == SNDROLE_PRIMARY_DUMMYSTANDBY) #define AmWalSenderToDummyStandby() (t_thrd.walsender_cxt.MyWalSnd->sendRole == SNDROLE_PRIMARY_DUMMYSTANDBY)
#define AmWalSenderOnDummyStandby() (t_thrd.walsender_cxt.MyWalSnd->sendRole == SNDROLE_DUMMYSTANDBY_STANDBY) #define AmWalSenderOnDummyStandby() (t_thrd.walsender_cxt.MyWalSnd->sendRole == SNDROLE_DUMMYSTANDBY_STANDBY)
#define TIME_GET_MILLISEC(t) (((long)(t).tv_sec * 1000) + ((long)(t).tv_usec) / 1000)
#define WAIT_FOR_ARCHIVE_TIME 10000L
/* /*
* calculate catchup late every 1000ms * calculate catchup late every 1000ms
@ -219,14 +217,7 @@ static void WalSndSetPercentCountStartLsn(XLogRecPtr startLsn);
static void WalSndRefreshPercentCountStartLsn(XLogRecPtr currentMaxLsn, XLogRecPtr currentDoneLsn); static void WalSndRefreshPercentCountStartLsn(XLogRecPtr currentMaxLsn, XLogRecPtr currentDoneLsn);
static void set_xlog_location(ServerMode local_role, XLogRecPtr* sndWrite, XLogRecPtr* sndFlush, XLogRecPtr* sndReplay); static void set_xlog_location(ServerMode local_role, XLogRecPtr* sndWrite, XLogRecPtr* sndFlush, XLogRecPtr* sndReplay);
static void ProcessArchiveFeedbackMessage(void); static void ProcessArchiveFeedbackMessage(void);
static void ProcessStandbyArchiveFeedbackMessage(void);
static void WalSndArchiveXlog(XLogRecPtr targetLsn, int sub_term); static void WalSndArchiveXlog(XLogRecPtr targetLsn, int sub_term);
static void WalSndSendArchiveLsn2Standby(XLogRecPtr targetLsn);
static void ArchiveXlogOnStandby(XLogRecPtr targetLsn);
static void SendLsn2Standby(XLogRecPtr targetLsn);
static void CheckStandbyFinishArchive(XLogRecPtr targetLsn);
static void ProcessArchiveStatusMessage();
static void ResponseArchiveStatusMessage();
static void CalCatchupRate(); static void CalCatchupRate();
char *DataDir = "."; char *DataDir = ".";
@ -2108,15 +2099,6 @@ static void ProcessStandbyMessage(void)
case 'a': case 'a':
ProcessArchiveFeedbackMessage(); ProcessArchiveFeedbackMessage();
break; break;
case 'n':
ProcessStandbyArchiveFeedbackMessage();
break;
case 'S':
ProcessArchiveStatusMessage();
break;
default: default:
ereport(COMMERROR, ereport(COMMERROR,
(errcode(ERRCODE_PROTOCOL_VIOLATION), errmsg("unexpected message type \"%d\"", msgtype))); (errcode(ERRCODE_PROTOCOL_VIOLATION), errmsg("unexpected message type \"%d\"", msgtype)));
@ -2553,9 +2535,9 @@ static void ProcessStandbySwitchRequestMessage(void)
static void ProcessArchiveFeedbackMessage(void) static void ProcessArchiveFeedbackMessage(void)
{ {
volatile WalSnd *walsnd = t_thrd.walsender_cxt.MyWalSnd; volatile WalSnd *walsnd = t_thrd.walsender_cxt.MyWalSnd;
ArchiveXlogResponseMessage reply; ArchiveXlogResponseMeeeage reply;
/* Decipher the reply message */ /* Decipher the reply message */
pq_copymsgbytes(t_thrd.walsender_cxt.reply_message, (char*)&reply, sizeof(ArchiveXlogResponseMessage)); pq_copymsgbytes(t_thrd.walsender_cxt.reply_message, (char*)&reply, sizeof(ArchiveXlogResponseMeeeage));
ereport(LOG, ereport(LOG,
(errmsg("ProcessArchiveFeedbackMessage %d %X/%X", reply.pitr_result, (errmsg("ProcessArchiveFeedbackMessage %d %X/%X", reply.pitr_result,
(uint32)(reply.targetLsn >> 32), (uint32)(reply.targetLsn)))); (uint32)(reply.targetLsn >> 32), (uint32)(reply.targetLsn))));
@ -2574,59 +2556,6 @@ static void ProcessArchiveFeedbackMessage(void)
SetLatch(walsnd->arch_latch); SetLatch(walsnd->arch_latch);
} }
/*
* Process the feedback to check is the standby archive successful
*/
static void ProcessStandbyArchiveFeedbackMessage(void)
{
volatile WalSnd *walsnd = t_thrd.walsender_cxt.MyWalSnd;
ArchiveXlogResponseMessage reply;
/* Decipher the reply message */
pq_copymsgbytes(t_thrd.walsender_cxt.reply_message, (char*)&reply, sizeof(ArchiveXlogResponseMessage));
if (reply.pitr_result && reply.archive_result == ARCHIVE_SKIP) {
ereport(WARNING,
(errmsg("ProcessArchiveFeedbackMessage: %X/%X has been removed in the standby, skip it.",
(uint32)(reply.targetLsn >> 32), (uint32)(reply.targetLsn))));
} else {
ereport(LOG,
(errmsg("ProcessArchiveFeedbackMessage %d %X/%X", reply.pitr_result,
(uint32)(reply.targetLsn >> 32), (uint32)(reply.targetLsn))));
}
walsnd->arch_finish_result = reply.pitr_result;
walsnd->archive_target_lsn = reply.targetLsn;
}
static void ProcessArchiveStatusMessage()
{
volatile WalSnd *walsnd = t_thrd.walsender_cxt.MyWalSnd;
ArchiveStatusMessage message;
pq_copymsgbytes(t_thrd.walsender_cxt.reply_message, (char*)&message, sizeof(ArchiveStatusMessage));
walsnd->is_start_archive = message.is_archive_activied;
if (message.startLsn == 0) {
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) {
walsnd->arch_task_last_lsn = flushPtr;
} else {
ereport(ERROR,
(errmsg("ProcessArchiveStatusMessage failed when call SyncRepGetSyncRecPtr")));
}
}
if (walsnd->arch_task_last_lsn < message.startLsn) {
walsnd->arch_task_last_lsn = message.startLsn;
}
ereport(LOG,
(errmsg("ProcessArchiveStatusMessage: reset last task lsn to %X/%X",
(uint32)(walsnd->arch_task_last_lsn >> 32), (uint32)(walsnd->arch_task_last_lsn))));
ResponseArchiveStatusMessage();
}
/* /*
* Count the limit for sleep_count, it is based on sleep time. * Count the limit for sleep_count, it is based on sleep time.
*/ */
@ -3203,42 +3132,6 @@ static int WalSndLoop(WalSndSendDataCallback send_data)
/* Check for input from the client */ /* Check for input from the client */
ProcessRepliesIfAny(); ProcessRepliesIfAny();
#ifdef ENABLE_MULTIPLE_NODES
/* 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;
List* sync_standbys = SyncRepGetSyncStandbys(&amSync);
int standby_nums = list_length(sync_standbys);
list_free(sync_standbys);
got_recptr = SyncRepGetSyncRecPtr(&receivePtr, &writePtr, &flushPtr, &replayPtr, &amSync, false);
if (got_recptr) {
ArchiveXlogOnStandby(flushPtr);
} else if (t_thrd.syncrep_cxt.SyncRepConfig == NULL ||
u_sess->attr.attr_storage.guc_synchronous_commit <= SYNCHRONOUS_COMMIT_LOCAL_FLUSH ||
(t_thrd.walsender_cxt.WalSndCtl->most_available_sync && standby_nums == 0)) {
/*
* This step is used to deal with the situation that synchronous standbys are not set.
*/
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);
}
#endif
/* Walsender first startup, send a keepalive to standby, no need reply. */ /* Walsender first startup, send a keepalive to standby, no need reply. */
if (first_startup) { if (first_startup) {
WalSndKeepalive(false); WalSndKeepalive(false);
@ -3710,12 +3603,6 @@ static void InitWalSnd(void)
walsnd->replSender = false; walsnd->replSender = false;
walsnd->peer_role = UNKNOWN_MODE; walsnd->peer_role = UNKNOWN_MODE;
walsnd->peer_state = NORMAL_STATE; walsnd->peer_state = NORMAL_STATE;
walsnd->is_start_archive = false;
walsnd->archive_target_lsn = 0;
walsnd->arch_task_last_lsn = 0;
walsnd->arch_finish_result = false;
walsnd->has_sent_arch_lsn = false;
walsnd->last_send_lsn_time = 0;
walsnd->channel_get_replc = 0; walsnd->channel_get_replc = 0;
rc = memset_s((void *)&walsnd->receive, sizeof(XLogRecPtr), 0, sizeof(XLogRecPtr)); rc = memset_s((void *)&walsnd->receive, sizeof(XLogRecPtr), 0, sizeof(XLogRecPtr));
securec_check(rc, "", ""); securec_check(rc, "", "");
@ -5159,31 +5046,6 @@ static void WalSndArchiveXlog(XLogRecPtr targetLsn, int sub_term)
(void)pq_putmessage_noblock('d', t_thrd.walsender_cxt.output_xlog_message, sizeof(ArchiveXlogMessage) + 1); (void)pq_putmessage_noblock('d', t_thrd.walsender_cxt.output_xlog_message, sizeof(ArchiveXlogMessage) + 1);
} }
/*
* send archive lsn to standby
*/
static void WalSndSendArchiveLsn2Standby(XLogRecPtr targetLsn)
{
ArchiveXlogMessage archive_message;
errno_t errorno = EOK;
ereport(LOG,
(errmsg("WalSndSendArchiveLsn2Standby %X/%X", (uint32)(targetLsn >> 32), (uint32)(targetLsn))));
archive_message.targetLsn = targetLsn;
/* Prepend with the message type and send it. */
t_thrd.walsender_cxt.output_xlog_message[0] = 'n';
errorno = memcpy_s(t_thrd.walsender_cxt.output_xlog_message + 1,
sizeof(ArchiveXlogMessage) + WS_MAX_SEND_SIZE,
&archive_message,
sizeof(ArchiveXlogMessage));
securec_check(errorno, "\0", "\0");
(void)pq_putmessage_noblock('d', t_thrd.walsender_cxt.output_xlog_message, sizeof(ArchiveXlogMessage) + 1);
struct timeval tv;
gettimeofday(&tv, NULL);
t_thrd.walsender_cxt.MyWalSnd->last_send_lsn_time = TIME_GET_MILLISEC(tv);
}
/* /*
* This isn't currently used for anything. Monitoring tools might be * This isn't currently used for anything. Monitoring tools might be
* interested in the future, and we'll need something like this in the * interested in the future, and we'll need something like this in the
@ -5623,106 +5485,6 @@ XLogSegNo WalGetSyncCountWindow(void)
return (XLogSegNo)(uint32)u_sess->attr.attr_storage.wal_keep_segments; return (XLogSegNo)(uint32)u_sess->attr.attr_storage.wal_keep_segments;
} }
static void ArchiveXlogOnStandby(XLogRecPtr flushLsn)
{
/* Check whether the active node is connected to the standby node. */
volatile WalSnd* walsnd = t_thrd.walsender_cxt.MyWalSnd;
/*
* Check whether the size of the newly archived xlog file is greater than that of the last archived xlog file.
* If the size is greater than the xlog size, the system sends an archive LSN to the standby node for archiving.
*/
if ((flushLsn - walsnd->arch_task_last_lsn ) > XLogSegSize) {
XLogRecPtr targetLsn;
targetLsn = Min(walsnd->arch_task_last_lsn + XLogSegSize -
(walsnd->arch_task_last_lsn % XLogSegSize) - 1,
flushLsn);
if (walsnd->arch_task_last_lsn == targetLsn) {
targetLsn = Min(targetLsn + XLogSegSize, flushLsn);
}
if (!walsnd->has_sent_arch_lsn) {
SendLsn2Standby(targetLsn);
} else {
CheckStandbyFinishArchive(targetLsn);
}
}
}
/*
* SendLsn2Standby
*
* Sending the lsn which is need to be archived by standby.
* It will return true if send archive lsn successful.
*/
static void SendLsn2Standby(XLogRecPtr targetLsn)
{
/* use volatile pointer to prevent code rearrangement */
volatile WalSnd* walsnd = t_thrd.walsender_cxt.MyWalSnd;
if (walsnd == NULL) {
/* send failed, walsnd is null */
return;
}
ereport(LOG,
(errmsg("the lsn which is ready to be sent is: \"%X/%X\"",
(uint32)(targetLsn >> 32), (uint32)(targetLsn))));
walsnd->has_sent_arch_lsn = true;
if (!XLogRecPtrIsInvalid(walsnd->flush) && XLByteLE(targetLsn, walsnd->flush)) {
walsnd->arch_task_lsn = targetLsn;
pg_atomic_write_u32(&walsnd->standby_archive_flag, 1);
}
}
/*
* CheckStandbyFinishArchive
*
* check the targetLsn and g_instance.archive_obs_cxt.archive_task.targetLsn for deal message with wrong order
*/
static void CheckStandbyFinishArchive(XLogRecPtr targetLsn)
{
struct timeval tv;
gettimeofday(&tv, NULL);
volatile WalSnd* walsnd = t_thrd.walsender_cxt.MyWalSnd;
long time_diff = (long)TIME_GET_MILLISEC(tv) - walsnd->last_send_lsn_time;
if (walsnd->arch_finish_result == false && time_diff > WAIT_FOR_ARCHIVE_TIME) {
walsnd->has_sent_arch_lsn = false;
ereport(WARNING,
(errcode(ERRCODE_WARNING),
errmsg("transaction xlog file \"%X/%X\" could not be archived: try again",
(uint32)(targetLsn >> 32), (uint32)(targetLsn))));
return;
}
if (walsnd->arch_finish_result == true) {
/* reset result flag */
if (XLByteEQ(walsnd->archive_target_lsn, targetLsn)) {
walsnd->arch_finish_result = false;
walsnd->arch_task_last_lsn = targetLsn;
ereport(LOG, (errmsg("the archive time is %.2lf seconds,last archive lsn change to \"%X/%X\"",
((double)time_diff / 1000), (uint32)(targetLsn >> 32), (uint32)(targetLsn))));
}
walsnd->has_sent_arch_lsn = false;
}
}
static void ResponseArchiveStatusMessage()
{
char msgbuf[sizeof(ArchiveStatusResponseMessage) + 1];
msgbuf[0] = 'S';
ArchiveStatusResponseMessage response;
volatile WalSnd *walsnd = t_thrd.walsender_cxt.MyWalSnd;
errno_t errorno = EOK;
if (walsnd == NULL)
return;
ereport(LOG,(errmsg("sending archive status response message")));
/* Prepend with the message type and send it. */
response.is_set_status_success = true;
errorno = memcpy_s(&msgbuf[1], sizeof(ArchiveStatusResponseMessage),
&response, sizeof(ArchiveStatusResponseMessage));
securec_check(errorno, "\0", "\0");
(void)pq_putmessage_noblock('d', msgbuf, sizeof(ArchiveStatusResponseMessage) + 1);
}
/* /*
* Calculate catchup rate of standby to estimate how long * Calculate catchup rate of standby to estimate how long
* the standby will be caught up with primary. * the standby will be caught up with primary.
@ -5753,17 +5515,3 @@ static void CalCatchupRate() {
} }
SpinLockRelease(&walsnd->mutex); SpinLockRelease(&walsnd->mutex);
} }
/* Check whether the standby node corresponding to the walsnd is a valid standby node for xlog archiving. */
bool IsValidArchiverStandby(WalSnd* walsnd)
{
if (walsnd == NULL) {
return false;
}
if (walsnd->pid != 0 && ((walsnd->sendRole & SNDROLE_PRIMARY_STANDBY) == walsnd->sendRole) &&
walsnd->is_start_archive) {
return true;
} else {
return false;
}
}

4
src/include/access/xlog.h
View File

@ -474,10 +474,6 @@ bool CheckForForceFinishRedoTrigger(TermFileData *term_file);
extern XLogRecPtr XlogRemoveSegPrimary; extern XLogRecPtr XlogRemoveSegPrimary;
void XLogArchiveNotify(const char *xlog);
bool XLogArchiveCheckDone(const char *xlog);
/* File path names (all relative to $PGDATA) */ /* File path names (all relative to $PGDATA) */
#define BACKUP_LABEL_FILE "backup_label" #define BACKUP_LABEL_FILE "backup_label"
#define DISABLE_CONN_FILE "disable_conn_file" #define DISABLE_CONN_FILE "disable_conn_file"

20
src/include/knl/knl_instance.h
View File

@ -737,25 +737,6 @@ typedef struct knl_g_archive_obs_context {
int sync_walsender_term; int sync_walsender_term;
} knl_g_archive_obs_context; } knl_g_archive_obs_context;
typedef struct knl_g_archive_standby_context {
/*
* walreceiver set when get task from walsender
* 0 for no task
* 1 walreceive get task from walsender and set it for archive thread
* 2 archive thread set when task is done
*/
volatile unsigned int arch_task_status;
/* archive thread set when archive done*/
bool arch_finish_result;
/* for standby */
ArchiveXlogMessage archive_task;
bool need_to_send_archive_status;
bool archive_enabled;
Latch* arch_latch;
} knl_g_archive_standby_context;
#ifdef ENABLE_MOT #ifdef ENABLE_MOT
typedef struct knl_g_mot_context { typedef struct knl_g_mot_context {
JitExec::JitExecMode jitExecMode; JitExec::JitExecMode jitExecMode;
@ -861,7 +842,6 @@ typedef struct knl_instance_context {
knl_g_barrier_creator_context barrier_creator_cxt; knl_g_barrier_creator_context barrier_creator_cxt;
knl_g_oid_nodename_mapping_cache oid_nodename_cache; knl_g_oid_nodename_mapping_cache oid_nodename_cache;
knl_g_archive_obs_context archive_obs_cxt; knl_g_archive_obs_context archive_obs_cxt;
knl_g_archive_standby_context archive_standby_cxt;
struct HTAB* ngroup_hash_table; struct HTAB* ngroup_hash_table;
knl_g_hypo_context hypo_cxt; knl_g_hypo_context hypo_cxt;
} knl_instance_context; } knl_instance_context;

1
src/include/knl/knl_thread.h
View File

@ -51,7 +51,6 @@
#include "knl/knl_guc.h" #include "knl/knl_guc.h"
#include "knl/knl_session.h" #include "knl/knl_session.h"
#include "nodes/pg_list.h" #include "nodes/pg_list.h"
#include "replication/walprotocol.h"
#include "storage/lock/s_lock.h" #include "storage/lock/s_lock.h"
#include "utils/palloc.h" #include "utils/palloc.h"
#include "storage/latch.h" #include "storage/latch.h"

5
src/include/postmaster/postmaster.h
View File

@ -51,11 +51,6 @@ typedef enum ReplicationType {
#define IS_DN_DUMMY_STANDYS_MODE() (g_instance.attr.attr_storage.replication_type == RT_WITH_DUMMY_STANDBY) #define IS_DN_DUMMY_STANDYS_MODE() (g_instance.attr.attr_storage.replication_type == RT_WITH_DUMMY_STANDBY)
#define IS_DN_WITHOUT_STANDBYS_MODE() (g_instance.attr.attr_storage.replication_type == RT_WITHOUT_STANDBY) #define IS_DN_WITHOUT_STANDBYS_MODE() (g_instance.attr.attr_storage.replication_type == RT_WITHOUT_STANDBY)
#define WalRcvIsOnline() \
((g_instance.pid_cxt.WalReceiverPID != 0 && t_thrd.walreceiverfuncs_cxt.WalRcv && \
t_thrd.walreceiverfuncs_cxt.WalRcv->isRuning))
/* /*
* We use a simple state machine to control startup, shutdown, and * We use a simple state machine to control startup, shutdown, and
* crash recovery (which is rather like shutdown followed by startup). * crash recovery (which is rather like shutdown followed by startup).

31
src/include/replication/walprotocol.h
View File

@ -142,32 +142,13 @@ typedef struct ArchiveXlogMessage {
} ArchiveXlogMessage; } ArchiveXlogMessage;
/* /*
* Refence :ArchiveXlogResponseMessage * Refence :ArchiveXlogResponseMeeeage
*/ */
typedef struct ArchiveXlogResponseMessage { typedef struct ArchiveXlogResponseMeeeage {
bool pitr_result; bool pitr_result;
XLogRecPtr targetLsn; XLogRecPtr targetLsn;
unsigned int archive_result; } ArchiveXlogResponseMeeeage;
} ArchiveXlogResponseMessage;
/* the result type of archive task */
typedef enum {
ARCHIVE_SUCCESS = 0,
ARCHIVE_SKIP,
ARCHIVE_FAILED
} ARCH_RESULT_STATUS;
/*
* Refence :ArchiveStatusMessage
*/
typedef struct ArchiveStatusMessage {
bool is_archive_activied;
XLogRecPtr startLsn;
} ArchiveStatusMessage;
typedef struct ArchiveStatusResponseMessage {
bool is_set_status_success;
} ArchiveStatusResponseMessage;
/* /*
* Keepalive message from primary (message type 'k'). (lowercase k) * Keepalive message from primary (message type 'k'). (lowercase k)
* This is wrapped within a CopyData message at the FE/BE protocol level. * This is wrapped within a CopyData message at the FE/BE protocol level.
@ -192,12 +173,6 @@ typedef enum {
PITR_TASK_DONE PITR_TASK_DONE
} PITR_TASK_STATUS; } PITR_TASK_STATUS;
typedef enum {
ARCH_TASK_NONE = 0,
ARCH_TASK_GET,
ARCH_TASK_DONE
} ARCH_TASK_STATUS;
/* /*
* switchover response message from primary (message type 'p'). This is wrapped within * switchover response message from primary (message type 'p'). This is wrapped within
* a CopyData message at the FE/BE protocol level. * a CopyData message at the FE/BE protocol level.

6
src/include/replication/walreceiver.h
View File

@ -193,11 +193,6 @@ typedef struct WalRcvData {
char lastRecoveredBarrierId[MAX_BARRIER_ID_LENGTH]; char lastRecoveredBarrierId[MAX_BARRIER_ID_LENGTH];
XLogRecPtr lastRecoveredBarrierLSN; XLogRecPtr lastRecoveredBarrierLSN;
Latch* obsArchLatch; Latch* obsArchLatch;
bool archive_enabled;
Latch* arch_latch;
bool arch_finish_result;
volatile unsigned int arch_task_status;
ArchiveXlogMessage archive_task;
} WalRcvData; } WalRcvData;
typedef struct WalReceiverFunc { typedef struct WalReceiverFunc {
@ -272,7 +267,6 @@ extern void set_failover_host_conninfo_for_dummy(const char *remote_host, int re
extern void get_failover_host_conninfo_for_dummy(int *repl); extern void get_failover_host_conninfo_for_dummy(int *repl);
extern void set_wal_rcv_write_rec_ptr(XLogRecPtr rec_ptr); extern void set_wal_rcv_write_rec_ptr(XLogRecPtr rec_ptr);
extern void setObsArchLatch(const Latch* latch); extern void setObsArchLatch(const Latch* latch);
extern void SetStandbyArchLatch(const Latch* latch);
static inline void WalRcvCtlAcquireExitLock(void) static inline void WalRcvCtlAcquireExitLock(void)

2
src/include/replication/walsender.h
View File

@ -60,8 +60,6 @@ extern bool WalSndAllInProgress(int type);
extern bool WalSndQuorumInProgress(int type); extern bool WalSndQuorumInProgress(int type);
extern XLogSegNo WalGetSyncCountWindow(void); extern XLogSegNo WalGetSyncCountWindow(void);
bool IsValidArchiverStandby(WalSnd* walsnd);
/* /*
* Remember that we want to wakeup walsenders later * Remember that we want to wakeup walsenders later
* *

12
src/include/replication/walsender_private.h
View File

@ -120,18 +120,6 @@ typedef struct WalSnd {
LogCtrlData log_ctrl; LogCtrlData log_ctrl;
unsigned int archive_flag; unsigned int archive_flag;
Latch* arch_latch; Latch* arch_latch;
bool is_start_archive;
unsigned int standby_archive_flag;
XLogRecPtr archive_target_lsn;
XLogRecPtr arch_task_last_lsn;
bool arch_finish_result;
/*
* has_sent_arch_lsn indicates whether the walsnd has sent the archive location,
* and last_send_time is used to record the time when the archive location is sent last time.
*/
bool has_sent_arch_lsn;
long last_send_lsn_time;
/* /*
* lastCalTime is last time calculating catchupRate, and lastCalWrite * lastCalTime is last time calculating catchupRate, and lastCalWrite

3
src/test/ha/ha_schedule_single
View File

@ -51,5 +51,4 @@ llt_single/sql_llt_coverage
llt_single/sequence_llt_coverage llt_single/sequence_llt_coverage
llt_single/temp_table_stop llt_single/temp_table_stop
llt_single/text_search llt_single/text_search
llt_single/xlog_redo llt_single/xlog_redo
llt_single/archive

27
src/test/ha/testcase/llt_single/archive.sh
View File

@ -1,27 +0,0 @@
#!/bin/sh
source ./standby_env.sh
function test_1()
{
check_instance
stop_standby
gs_guc set -D $standby_data_dir -c "archive_mode = on"
gs_guc set -D $standby_data_dir -c "archive_dest = $standby_data_dir/archive_directory"
start_standby
#create table1
gsql -d $db -p $dn1_primary_port -c "DROP TABLE if exists mpp_test1; CREATE TABLE mpp_test1(id INT);"
for i in $(seq 1 10000)
do
gsql -d $db -p $dn1_primary_port -c "insert into mpp_test1 values($i);"
done
stop_primary
stop_standby
cp $standby_data_dir/archive_directory/* $primary_data_dir/pg_xlog
cp $standby_data_dir/archive_directory/* $standby_data_dir/pg_xlog
start_primary
start_standby
check_standby_startup
}