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

xlog lock less modify

This commit is contained in:
jeffee 2020-12-14 14:25:48 +08:00
parent 7b706fbf10
commit 5d0218bec4
40 changed files with 1514 additions and 712 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
src/common/backend/catalog/storage.cpp
View File

@ -601,7 +601,7 @@ void RelationTruncate(Relation rel, BlockNumber nblocks)
* contain entries for the non-existent heap pages.
*/
if (fsm || vm)
XLogFlush(lsn);
XLogWaitFlush(lsn);
}
if (!RELATION_IS_GLOBAL_TEMP(rel)) {
/* Lock RelFileNode to control concurrent with Catchup Thread */
@ -673,7 +673,7 @@ void PartitionTruncate(Relation parent, Partition part, BlockNumber nblocks)
* contain entries for the non-existent heap pages.
*/
if (fsm || vm)
XLogFlush(lsn);
XLogWaitFlush(lsn);
}
/* Lock RelFileNode to control concurrent with Catchup Thread */
@ -910,7 +910,7 @@ void XLogBlockSmgrRedoTruncate(RelFileNode rnode, BlockNumber blkno, XLogRecPtr
* after truncation, but that would leave a small window where the
* WAL-first rule could be violated.
*/
XLogFlush(lsn);
XLogWaitFlush(lsn);
LockRelFileNode(rnode, AccessExclusiveLock);
smgrtruncate(reln, MAIN_FORKNUM, blkno);

4
src/common/backend/pgxc_single/barrier/barrier.cpp
View File

@ -149,7 +149,7 @@ void ProcessCreateBarrierExecute(const char* id)
rdata[0].next = NULL;
recptr = XLogInsert(RM_BARRIER_ID, XLOG_BARRIER_CREATE, rdata);
XLogFlush(recptr);
XLogWaitFlush(recptr);
}
pq_beginmessage(&buf, 'b');
@ -525,7 +525,7 @@ static void ExecuteBarrier(const char* id)
rdata[0].next = NULL;
recptr = XLogInsert(RM_BARRIER_ID, XLOG_BARRIER_CREATE, rdata);
XLogFlush(recptr);
XLogWaitFlush(recptr);
}
}

2
src/common/backend/utils/cache/relmapper.cpp vendored
View File

@ -749,7 +749,7 @@ static void write_relmap_file(bool shared, RelMapFile* newmap, bool write_wal, b
lsn = XLogInsert(RM_RELMAP_ID, XLOG_RELMAP_UPDATE);
/* As always, WAL must hit the disk before the data update does */
XLogFlush(lsn);
XLogWaitFlush(lsn);
}
errno = 0;

69
src/common/backend/utils/misc/guc.cpp
View File

@ -6531,6 +6531,23 @@ static void init_configure_names_int()
NULL,
NULL
},
{
{
"wal_writer_cpu",
PGC_POSTMASTER,
WAL_SETTINGS,
gettext_noop("Sets the binding CPU number for the WAL writer thread."),
NULL,
GUC_NOT_IN_SAMPLE
},
&g_instance.attr.attr_storage.wal_writer_cpu,
-1,
-1,
1023,
NULL,
NULL,
NULL
},
{
{
"advance_xlog_file_num",
@ -6548,6 +6565,41 @@ static void init_configure_names_int()
NULL,
NULL
},
{
{
"xlog_flush_uplimit",
PGC_POSTMASTER,
WAL_SETTINGS,
gettext_noop("Sets the maximum bytes that a xlog flush can write."),
NULL,
GUC_NOT_IN_SAMPLE
},
&g_instance.attr.attr_storage.xlog_flush_uplimit,
INT_MAX,
100,
INT_MAX,
NULL,
NULL,
NULL
},
{
{
"wal_file_init_num",
PGC_POSTMASTER,
WAL_SETTINGS,
gettext_noop("Sets the number of xlog segment files that WAL writer auxiliary thread "
"creates at one time."),
NULL,
GUC_NOT_IN_SAMPLE
},
&g_instance.attr.attr_storage.wal_file_init_num,
10,
1,
INT_MAX,
NULL,
NULL,
NULL
},
{
{
"checkpoint_wait_timeout",
@ -9944,6 +9996,23 @@ static void init_configure_names_int64()
NULL,
NULL
},
{
{
"xlog_idle_flushes_before_sleep",
PGC_POSTMASTER,
WAL_SETTINGS,
gettext_noop("Number of idle xlog flushes before xlog flusher goes to sleep."),
NULL,
GUC_NOT_IN_SAMPLE
},
&g_instance.attr.attr_storage.xlog_idle_flushes_before_sleep,
INT64CONST(500000000),
INT64CONST(0),
INT64CONST(0x7FFFFFFFFFFFFFF),
NULL,
NULL,
NULL
},
/* End-of-list marker */
{
{

4
src/common/port/gs_thread.cpp
View File

@ -293,6 +293,10 @@ static void check_boot_name_internel(AuxProcType aux_thread_type, char** name_th
*name_thread = "WalWriter";
break;
}
case WalWriterAuxiliaryProcess: {
*name_thread = "WalWriterAuxiliary";
break;
}
case WalReceiverProcess: {
*name_thread = "WalReceiver";
break;

2
src/gausskernel/process/postmaster/Makefile
View File

@ -32,7 +32,7 @@ ifneq "$(MAKECMDGOALS)" "clean"
endif
endif
OBJS = autovacuum.o bgwriter.o fork_process.o pgarch.o pgstat.o postmaster.o gaussdb_version.o\
startup.o syslogger.o walwriter.o checkpointer.o pgaudit.o alarmchecker.o bgworker.o\
startup.o syslogger.o walwriter.o walwriterauxiliary.o checkpointer.o pgaudit.o alarmchecker.o bgworker.o\
twophasecleaner.o aiocompleter.o fencedudf.o lwlockmonitor.o cbmwriter.o remoteservice.o pagewriter.o\
$(top_builddir)/src/lib/config/libconfig.a

4
src/gausskernel/process/postmaster/pagewriter.cpp
View File

@ -332,7 +332,7 @@ uint64 get_dirty_page_queue_tail()
#if defined(__x86_64__) || defined(__aarch64__)
uint128_u compare;
compare = atomic_compare_and_swap_u128((uint128_u*)&g_instance.ckpt_cxt_ctl->dirty_page_queue_reclsn);
// return the dirty page queue tail
/* return the dirty page queue tail */
return compare.u64[1];
#else
return 0;
@ -601,7 +601,7 @@ static void ckpt_pagewriter_main_thread_flush_dirty_page()
dw_perform(requested_flush_num, NULL, &g_instance.ckpt_cxt_ctl->page_writer_procs.thrd_dw_cxt);
CurrBytePos = GetXLogInsertEndRecPtr();
XLogFlush(CurrBytePos);
XLogWaitFlush(CurrBytePos);
g_instance.ckpt_cxt_ctl->page_writer_xlog_flush_loc = t_thrd.xlog_cxt.LogwrtResult->Flush;
divide_dirty_page_to_thread(requested_flush_num);

24
src/gausskernel/process/postmaster/pgstat.cpp
View File

@ -2507,6 +2507,30 @@ static THR_LOCAL char* BackendNspRelnameBuffer = NULL;
PgBackendStatus* PgBackendStatusArray = NULL;
THR_LOCAL XLogStat_Collect *XLogStat_shared = NULL;
Size XLogStatShmemSize(void)
{
return sizeof(XLogStat_Collect);
}
void XLogStatShmemInit(void)
{
bool found;
errno_t rc;
XLogStat_shared = (XLogStat_Collect *)ShmemInitStruct("XLogStat", XLogStatShmemSize(), &found);
XLogStat_shared->remoteFlushWaitCount = 0;
if (!IsUnderPostmaster) {
Assert(!found);
rc = memset_s(XLogStat_shared, XLogStatShmemSize(), 0, XLogStatShmemSize());
securec_check(rc, "\0", "\0");
} else {
Assert(found);
}
}
/*
* ---------
* pgstat_fetch_waitcount() -

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

@ -137,6 +137,7 @@
#include "postmaster/twophasecleaner.h"
#include "postmaster/licensechecker.h"
#include "postmaster/walwriter.h"
#include "postmaster/walwriterauxiliary.h"
#include "postmaster/lwlockmonitor.h"
#include "replication/walreceiver.h"
#include "replication/datareceiver.h"
@ -2958,6 +2959,10 @@ static int ServerLoop(void)
g_instance.pid_cxt.WalWriterPID, pmState, t_thrd.postmaster_cxt.HaShmData->current_mode)));
}
if (g_instance.pid_cxt.WalWriterAuxiliaryPID == 0 && pmState == PM_RUN) {
g_instance.pid_cxt.WalWriterAuxiliaryPID = initialize_util_thread(WALWRITERAUXILIARY);
}
/*
* let cbm writer thread exit if enable_cbm_track gus is switched off
*/
@ -4337,6 +4342,9 @@ static void SIGHUP_handler(SIGNAL_ARGS)
if (g_instance.pid_cxt.WalWriterPID != 0)
signal_child(g_instance.pid_cxt.WalWriterPID, SIGHUP);
if (g_instance.pid_cxt.WalWriterAuxiliaryPID != 0)
signal_child(g_instance.pid_cxt.WalWriterAuxiliaryPID, SIGHUP);
if (g_instance.pid_cxt.WalRcvWriterPID != 0)
signal_child(g_instance.pid_cxt.WalRcvWriterPID, SIGHUP);
@ -4627,6 +4635,9 @@ static void pmdie(SIGNAL_ARGS)
if (g_instance.pid_cxt.WalWriterPID != 0)
signal_child(g_instance.pid_cxt.WalWriterPID, SIGTERM);
if (g_instance.pid_cxt.WalWriterAuxiliaryPID != 0)
signal_child(g_instance.pid_cxt.WalWriterAuxiliaryPID, SIGTERM);
if (ENABLE_THREAD_POOL && (pmState == PM_RECOVERY || pmState == PM_STARTUP)) {
/*
* Although there is not connections from client at PM_RECOVERY and PM_STARTUP
@ -4806,6 +4817,9 @@ static void ProcessDemoteRequest(void)
if (g_instance.pid_cxt.WalWriterPID != 0)
signal_child(g_instance.pid_cxt.WalWriterPID, SIGTERM);
if (g_instance.pid_cxt.WalWriterAuxiliaryPID != 0)
signal_child(g_instance.pid_cxt.WalWriterAuxiliaryPID, SIGTERM);
if (g_instance.pid_cxt.CBMWriterPID != 0) {
Assert(!dummyStandbyMode);
signal_child(g_instance.pid_cxt.CBMWriterPID, SIGTERM);
@ -4941,6 +4955,9 @@ static void ProcessDemoteRequest(void)
if (g_instance.pid_cxt.WalWriterPID != 0)
signal_child(g_instance.pid_cxt.WalWriterPID, SIGTERM);
if (g_instance.pid_cxt.WalWriterAuxiliaryPID != 0)
signal_child(g_instance.pid_cxt.WalWriterAuxiliaryPID, SIGTERM);
pmState = PM_WAIT_BACKENDS;
}
break;
@ -5130,6 +5147,9 @@ static void reaper(SIGNAL_ARGS)
if (g_instance.pid_cxt.WalWriterPID == 0)
g_instance.pid_cxt.WalWriterPID = initialize_util_thread(WALWRITER);
if (g_instance.pid_cxt.WalWriterAuxiliaryPID == 0)
g_instance.pid_cxt.WalWriterAuxiliaryPID = initialize_util_thread(WALWRITERAUXILIARY);
if (g_instance.pid_cxt.CBMWriterPID == 0 && !dummyStandbyMode &&
u_sess->attr.attr_storage.enable_cbm_tracking)
g_instance.pid_cxt.CBMWriterPID = initialize_util_thread(CBMWRITER);
@ -5352,6 +5372,20 @@ static void reaper(SIGNAL_ARGS)
continue;
}
/*
* Was it the wal file creator? Normal exit can be ignored; we'll start a
* new one at the next iteration of the postmaster's main loop, if
* necessary. Any other exit condition is treated as a crash.
*/
if (pid == g_instance.pid_cxt.WalWriterAuxiliaryPID) {
g_instance.pid_cxt.WalWriterAuxiliaryPID = 0;
if (!EXIT_STATUS_0(exitstatus))
HandleChildCrash(pid, exitstatus, _("WAL file creator process"));
continue;
}
/*
* Was it the wal receiver? If exit status is zero (normal) or one
@ -5683,6 +5717,8 @@ static const char* GetProcName(ThreadId pid)
return "checkpointer process";
else if (pid == g_instance.pid_cxt.WalWriterPID)
return "WAL writer process";
else if (pid == g_instance.pid_cxt.WalWriterAuxiliaryPID)
return "WAL file creator process";
else if (pid == g_instance.pid_cxt.WalReceiverPID)
return "WAL receiver process";
else if (pid == g_instance.pid_cxt.WalRcvWriterPID)
@ -6129,7 +6165,7 @@ static void PostmasterStateMachine(void)
g_instance.pid_cxt.DataReceiverPID == 0 && g_instance.pid_cxt.DataRcvWriterPID == 0 &&
g_instance.pid_cxt.BgWriterPID == 0 &&
(g_instance.pid_cxt.CheckpointerPID == 0 || !g_instance.fatal_error) &&
g_instance.pid_cxt.WalWriterPID == 0 && g_instance.pid_cxt.AutoVacPID == 0 &&
g_instance.pid_cxt.WalWriterPID == 0 && g_instance.pid_cxt.WalWriterAuxiliaryPID == 0 && g_instance.pid_cxt.AutoVacPID == 0 &&
g_instance.pid_cxt.WLMCollectPID == 0 && g_instance.pid_cxt.WLMMonitorPID == 0 &&
g_instance.pid_cxt.WLMArbiterPID == 0 && g_instance.pid_cxt.CPMonitorPID == 0 &&
g_instance.pid_cxt.PgJobSchdPID == 0 && g_instance.pid_cxt.CBMWriterPID == 0 &&
@ -6256,6 +6292,7 @@ static void PostmasterStateMachine(void)
Assert(g_instance.pid_cxt.BgWriterPID == 0);
Assert(g_instance.pid_cxt.CheckpointerPID == 0);
Assert(g_instance.pid_cxt.WalWriterPID == 0);
Assert(g_instance.pid_cxt.WalWriterAuxiliaryPID == 0);
Assert(g_instance.pid_cxt.AutoVacPID == 0);
Assert(g_instance.pid_cxt.PgJobSchdPID == 0);
Assert(g_instance.pid_cxt.CBMWriterPID == 0);
@ -9921,6 +9958,9 @@ static void SetAuxType()
case WALWRITER:
t_thrd.bootstrap_cxt.MyAuxProcType = WalWriterProcess;
break;
case WALWRITERAUXILIARY:
t_thrd.bootstrap_cxt.MyAuxProcType = WalWriterAuxiliaryProcess;
break;
case WALRECEIVER:
t_thrd.bootstrap_cxt.MyAuxProcType = WalReceiverProcess;
break;
@ -10157,6 +10197,13 @@ int GaussDbAuxiliaryThreadMain(knl_thread_arg* arg)
proc_exit(1); /* should never return */
break;
case WALWRITERAUXILIARY:
/* don't set signals, walwriterauxiliary has its own agenda */
InitXLOGAccess();
WalWriterAuxiliaryMain();
proc_exit(1); /* should never return */
break;
case WALRECEIVER:
/* don't set signals, walreceiver has its own agenda */
WalReceiverMain();
@ -10374,6 +10421,7 @@ int GaussDbThreadMain(knl_thread_arg* arg)
case BGWRITER:
case CHECKPOINT_THREAD:
case WALWRITER:
case WALWRITERAUXILIARY:
case WALRECEIVER:
case WALRECWRITE:
case DATARECIVER:
@ -10660,6 +10708,7 @@ static GaussdbThreadEntry GaussdbThreadEntryGate[] = {GaussDbThreadMain<MASTER>,
GaussDbThreadMain<SNAPSHOT_WORKER>,
GaussDbThreadMain<CHECKPOINT_THREAD>,
GaussDbThreadMain<WALWRITER>,
GaussDbThreadMain<WALWRITERAUXILIARY>,
GaussDbThreadMain<WALRECEIVER>,
GaussDbThreadMain<WALRECWRITE>,
GaussDbThreadMain<DATARECIVER>,

126
src/gausskernel/process/postmaster/walwriter.cpp
View File

@ -63,6 +63,9 @@
#include "gssignal/gs_signal.h"
THR_LOCAL int WalWriterDelay = 0;
THR_LOCAL int SLEEP_TIME_OUT_MS = 300; /* WAL writer sleep timeout in millisecond. */
/*
* Number of do-nothing loops before lengthening the delay time, and the
* multiplier to apply to WalWriterDelay when we do decide to hibernate.
@ -95,12 +98,31 @@ void WalWriterMain(void)
MemoryContext walwriter_context;
int left_till_hibernate;
bool hibernating = false;
sigset_t oldSigMask;
sigset_t old_sig_mask;
bool wrote_something = true;
long times_wrote_nothing = 0;
struct timespec time_to_wait;
int sleep_times_counter = 0;
int time_out_counter = 0;
knl_thread_set_name("WalWriter");
g_instance.wal_cxt.isWalWriterUp = true;
ereport(LOG, (errmsg("WalWriter started")));
if (g_instance.attr.attr_storage.wal_writer_cpu >= 0) {
cpu_set_t walWriterSet;
CPU_ZERO(&walWriterSet);
CPU_SET(g_instance.attr.attr_storage.wal_writer_cpu, &walWriterSet);
int rc = sched_setaffinity(0, sizeof(cpu_set_t), &walWriterSet);
if (rc == -1) {
ereport(WARNING,
(errmsg("Failed to schedule WalWriter on wal_writer_cpu, sched_setaffinity() set errno as %d.",
errno)));
}
}
/*
* Properly accept or ignore signals the postmaster might send us
*
@ -160,9 +182,8 @@ void WalWriterMain(void)
if (sigsetjmp(local_sigjmp_buf, 1) != 0) {
gstrace_tryblock_exit(true, oldTryCounter);
// We need restore the signal mask of current thread
//
pthread_sigmask(SIG_SETMASK, &oldSigMask, NULL);
/* We need restore the signal mask of current thread. */
pthread_sigmask(SIG_SETMASK, &old_sig_mask, NULL);
/* Since not using PG_TRY, must reset error stack by hand */
t_thrd.log_cxt.error_context_stack = NULL;
@ -257,22 +278,8 @@ void WalWriterMain(void)
* Loop forever
*/
for (;;) {
long cur_timeout;
int rc;
/*
* Advertise whether we might hibernate in this cycle. We do this
* before resetting the latch to ensure that any async commits will
* see the flag set if they might possibly need to wake us up, and
* that we won't miss any signal they send us. (If we discover work
* to do in the last cycle before we would hibernate, the global flag
* will be set unnecessarily, but little harm is done.) But avoid
* touching the global flag if it doesn't need to change.
*/
if (hibernating != (bool)(left_till_hibernate <= 1)) {
hibernating = (left_till_hibernate <= 1);
SetWalWriterSleeping(hibernating);
}
long cur_timeout = WalWriterDelay;
int rc = 0;
/* Clear any already-pending wakeups */
ResetLatch(&t_thrd.proc->procLatch);
@ -295,35 +302,64 @@ void WalWriterMain(void)
/* execute callbacks (i.e. write data from MOT) */
CallWALCallback();
/*
* Do what we're here for; then, if XLogBackgroundFlush() found useful
* work to do, reset hibernation counter.
*/
if (XLogBackgroundFlush()) {
left_till_hibernate = LOOPS_UNTIL_HIBERNATE;
} else if (left_till_hibernate > 0) {
left_till_hibernate--;
}
wrote_something = XLogBackgroundFlush();
/*
* Sleep until we are signaled or WalWriterDelay has elapsed. If we
* haven't done anything useful for quite some time, lengthen the
* sleep time so as to reduce the server's idle power consumption.
*/
if (left_till_hibernate > 0) {
cur_timeout = u_sess->attr.attr_storage.WalWriterDelay; /* in ms */
} else {
cur_timeout = u_sess->attr.attr_storage.WalWriterDelay * HIBERNATE_FACTOR;
if (!wrote_something && ++times_wrote_nothing > g_instance.attr.attr_storage.xlog_idle_flushes_before_sleep) {
/*
* Wait for the first entry after last flushed entry to be updated
*/
int lastFlushedEntry = g_instance.wal_cxt.lastWalStatusEntryFlushed;
WalInsertStatusEntry *pCriticalEntry = &g_instance.wal_cxt.walInsertStatusTable[GET_NEXT_STATUS_ENTRY(lastFlushedEntry)];
if (g_instance.wal_cxt.isWalWriterUp && pCriticalEntry->status == WAL_NOT_COPIED) {
sleep_times_counter++;
pthread_mutex_lock(&g_instance.wal_cxt.criticalEntryMutex);
g_instance.wal_cxt.isWalWriterSleeping = true;
while (pCriticalEntry->status == WAL_NOT_COPIED && !t_thrd.walwriter_cxt.shutdown_requested) {
(void)clock_gettime(CLOCK_REALTIME, &time_to_wait);
time_to_wait.tv_nsec += 1000000 * SLEEP_TIME_OUT_MS;
if (time_to_wait.tv_nsec >= 1000000000) {
time_to_wait.tv_nsec -= 1000000000;
time_to_wait.tv_sec += 1;
}
int res = pthread_cond_timedwait(&g_instance.wal_cxt.criticalEntryCV,
&g_instance.wal_cxt.criticalEntryMutex, &time_to_wait);
if (res == 0) {
/*
* We should not break out here because we may be notified by an
* entry after the critiacal entry. We must check again if critical
* entry status is WAL_NOT_COPIED.
*/
continue;
} else if (res == ETIMEDOUT) {
time_out_counter++;
} else {
ereport(WARNING, (errmsg("WAL writer pthread_cond_timedwait returned error code = %d.",
errno)));
}
CHECK_FOR_INTERRUPTS();
}
g_instance.wal_cxt.isWalWriterSleeping = false;
pthread_mutex_unlock(&g_instance.wal_cxt.criticalEntryMutex);
time_out_counter = 0;
}
times_wrote_nothing = 0;
}
pgstat_report_activity(STATE_IDLE, NULL);
rc = WaitLatch(&t_thrd.proc->procLatch, WL_LATCH_SET | WL_TIMEOUT | WL_POSTMASTER_DEATH, cur_timeout);
if (cur_timeout > 0) {
rc = WaitLatch(&t_thrd.proc->procLatch, WL_LATCH_SET | WL_TIMEOUT | WL_POSTMASTER_DEATH, cur_timeout);
}
/* r
* Emergency bailout if postmaster has died. This is to avoid the
* necessity for manual cleanup of all postmaster children.
*/
if (rc & WL_POSTMASTER_DEATH) {
g_instance.wal_cxt.isWalWriterUp = false;
pg_memory_barrier();
/* Stop WalWriterAuxiliary from waiting. */
PGSemaphoreReset(&g_instance.wal_cxt.walInitSegLock->l.sem);
PGSemaphoreUnlock(&g_instance.wal_cxt.walInitSegLock->l.sem);
gs_thread_exit(1);
}
}
@ -343,6 +379,12 @@ static void wal_quickdie(SIGNAL_ARGS)
{
gs_signal_setmask(&t_thrd.libpq_cxt.BlockSig, NULL);
g_instance.wal_cxt.isWalWriterUp = false;
pg_memory_barrier();
/* Stop WalWriterAuxiliary from waiting. */
PGSemaphoreReset(&g_instance.wal_cxt.walInitSegLock->l.sem);
PGSemaphoreUnlock(&g_instance.wal_cxt.walInitSegLock->l.sem);
/*
* We DO NOT want to run proc_exit() callbacks -- we're here because
* shared memory may be corrupted, so we don't want to try to clean up our
@ -388,6 +430,12 @@ static void WalShutdownHandler(SIGNAL_ARGS)
SetLatch(&t_thrd.proc->procLatch);
errno = save_errno;
g_instance.wal_cxt.isWalWriterUp = false;
pg_memory_barrier();
/* Stop WalWriterAuxiliary from waiting. */
PGSemaphoreReset(&g_instance.wal_cxt.walInitSegLock->l.sem);
PGSemaphoreUnlock(&g_instance.wal_cxt.walInitSegLock->l.sem);
}
/* SIGUSR1: used for latch wakeups */

329
src/gausskernel/process/postmaster/walwriterauxiliary.cpp Normal file
View File

@ -0,0 +1,329 @@
/* -------------------------------------------------------------------------
*
* walwriterauxiliary.cpp
*
* The WAL writer auxiliary background process. It creates and zeros xlog segment
* files in advance so that walwriterauxiliary thread can directly flush xlog record
* into well-prepared files.
* The walwriterauxiliary is started by the postmaster as soon as the startup subprocess
* finishes. It remains alive until the postmaster commands it to terminate.
* Normal termination is by SIGTERM, which instructs the walwriterauxiliary to exit(0).
* Emergency termination is by SIGQUIT; like any backend, the walwriterauxiliary will
* simply abort and exit on SIGQUIT.
*
*
* If the walwriterauxiliary exits unexpectedly, the postmaster treats that the same
* as a backend crash: shared memory may be corrupted, so remaining backends
* should be killed by SIGQUIT and then a recovery cycle started.
*
*
* Portions Copyright (c) 1996-2012, PostgreSQL Global Development Group
*
*
* IDENTIFICATION
* src/backend/postmaster/walwriterauxiliary.cpp
*
* -------------------------------------------------------------------------
*/
#include <signal.h>
#include <sys/time.h>
#include <time.h>
#include <unistd.h>
#include <utmpx.h>
#include <errno.h>
#ifdef __USE_NUMA
#include <numa.h>
#endif
#include "postgres.h"
#include "access/xlog.h"
#include "libpq/pqsignal.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "postmaster/walwriterauxiliary.h"
#include "storage/bufmgr.h"
#include "storage/ipc.h"
#include "storage/lwlock.h"
#include "storage/proc.h"
#include "storage/smgr.h"
#include "utils/guc.h"
#include "utils/hsearch.h"
#include "utils/memutils.h"
#include "utils/resowner.h"
#include "gssignal/gs_signal.h"
/*
* GUC parameters
*/
THR_LOCAL int WalWriterAuxiliaryDelay = 1000; /* (in ms) */
/*
* Number of do-nothing loops before lengthening the delay time, and the
* multiplier to apply to WalWriterDelay when we do decide to hibernate.
* (Perhaps these need to be configurable?)
*/
#define LOOPS_UNTIL_HIBERNATE 50
#define HIBERNATE_FACTOR 25
/* Signal handlers */
static void walwriterauxiliary_quickdie(SIGNAL_ARGS);
static void WalwriterauxiliarySigHupHandler(SIGNAL_ARGS);
static void WalwriterauxiliaryShutdownHandler(SIGNAL_ARGS);
static void walwriterauxiliary_sigusr1_handler(SIGNAL_ARGS);
/*
* Main entry point for walwriterauxiliary process
*
* This is invoked from AuxiliaryProcessMain, which has already created the
* basic execution environment, but not enabled signals yet.
*/
void WalWriterAuxiliaryMain(void)
{
sigjmp_buf local_sigjmp_buf;
MemoryContext walwriterauxiliary_context;
sigset_t old_sig_mask;
t_thrd.role = WALWRITERAUXILIARY;
ereport(LOG, (errmsg("walwriterauxiliary started")));
/*
* Properly accept or ignore signals the postmaster might send us
*
* We have no particular use for SIGINT at the moment, but seems
* reasonable to treat like SIGTERM.
*
* Reset some signals that are accepted by postmaster but not here.
*/
(void)gspqsignal(SIGHUP, WalwriterauxiliarySigHupHandler); /* set flag to read config file */
(void)gspqsignal(SIGINT, WalwriterauxiliaryShutdownHandler); /* request shutdown */
(void)gspqsignal(SIGTERM, WalwriterauxiliaryShutdownHandler); /* request shutdown */
(void)gspqsignal(SIGQUIT, walwriterauxiliary_quickdie); /* hard crash time */
(void)gspqsignal(SIGALRM, SIG_IGN);
(void)gspqsignal(SIGPIPE, SIG_IGN);
(void)gspqsignal(SIGUSR1, walwriterauxiliary_sigusr1_handler);
(void)gspqsignal(SIGUSR2, SIG_IGN); /* not used */
(void)gspqsignal(SIGCHLD, SIG_DFL);
(void)gspqsignal(SIGTTIN, SIG_DFL);
(void)gspqsignal(SIGTTOU, SIG_DFL);
(void)gspqsignal(SIGCONT, SIG_DFL);
(void)gspqsignal(SIGWINCH, SIG_DFL);
/* We allow SIGQUIT (quickdie) at all times */
sigdelset(&t_thrd.libpq_cxt.BlockSig, SIGQUIT);
/*
* Create a resource owner to keep track of our resources (not clear that
* we need this, but may as well have one).
*/
t_thrd.utils_cxt.CurrentResourceOwner = ResourceOwnerCreate(NULL, "Wal Writer Auxiliary");
/*
* Create a memory context that we will do all our work in. We do this so
* that we can reset the context during error recovery and thereby avoid
* possible memory leaks. Formerly this code just ran in
* t_thrd.top_mem_cxt, but resetting that would be a really bad idea.
*/
walwriterauxiliary_context = AllocSetContextCreate(t_thrd.top_mem_cxt,
"Wal Writer Auxiliary",
ALLOCSET_DEFAULT_MINSIZE,
ALLOCSET_DEFAULT_INITSIZE,
ALLOCSET_DEFAULT_MAXSIZE);
MemoryContextSwitchTo(walwriterauxiliary_context);
/*
* If an exception is encountered, processing resumes here.
*
* This code is heavily based on bgwriter.c, q.v.
*/
if (sigsetjmp(local_sigjmp_buf, 1) != 0) {
/* We need restore the signal mask of current thread */
pthread_sigmask(SIG_SETMASK, &old_sig_mask, NULL);
/* Since not using PG_TRY, must reset error stack by hand */
t_thrd.log_cxt.error_context_stack = NULL;
/* Prevent interrupts while cleaning up */
HOLD_INTERRUPTS();
/* Report the error to the server log */
EmitErrorReport();
/* abort async io, must before LWlock release */
AbortAsyncListIO();
/*
* These operations are really just a minimal subset of
* AbortTransaction(). We don't have very many resources to worry
* about in walwriterauxiliary, but we do have LWLocks, and perhaps buffers?
*/
LWLockReleaseAll();
pgstat_report_waitevent(WAIT_EVENT_END);
AbortBufferIO();
UnlockBuffers();
/* buffer pins are released here: */
ResourceOwnerRelease(t_thrd.utils_cxt.CurrentResourceOwner, RESOURCE_RELEASE_BEFORE_LOCKS, false, true);
/* we needn't bother with the other ResourceOwnerRelease phases */
AtEOXact_Buffers(false);
AtEOXact_SMgr();
AtEOXact_Files();
AtEOXact_HashTables(false);
/*
* Now return to normal top-level context and clear ErrorContext for
* next time.
*/
MemoryContextSwitchTo(walwriterauxiliary_context);
FlushErrorState();
/* Flush any leaked data in the top-level context */
MemoryContextResetAndDeleteChildren(walwriterauxiliary_context);
/* Now we can allow interrupts again */
RESUME_INTERRUPTS();
/*
* Sleep at least 1 second after any error. A write error is likely
* to be repeated, and we don't want to be filling the error logs as
* fast as we can.
*/
pg_usleep(1000000L);
/*
* Close all open files after any error. This is helpful on Windows,
* where holding deleted files open causes various strange errors.
* It's not clear we need it elsewhere, but shouldn't hurt.
*/
smgrcloseall();
}
/* We can now handle ereport(ERROR) */
t_thrd.log_cxt.PG_exception_stack = &local_sigjmp_buf;
/*
* Unblock signals (they were blocked when the postmaster forked us)
*/
gs_signal_setmask(&t_thrd.libpq_cxt.UnBlockSig, NULL);
(void)gs_signal_unblock_sigusr2();
/*
* Advertise our latch that backends can use to wake us up while we're
* sleeping.
*/
g_instance.proc_base->walwriterauxiliaryLatch = &t_thrd.proc->procLatch;
pgstat_report_appname("Wal Writer Auxiliary");
pgstat_report_activity(STATE_IDLE, NULL);
/*
* Loop forever
*/
for (;;) {
long cur_timeout = 0;
int rc = 0;
if (g_instance.wal_cxt.isWalWriterUp) {
PGSemaphoreLock(&g_instance.wal_cxt.walInitSegLock->l.sem, true);
}
/* Clear any already-pending wakeups */
ResetLatch(&t_thrd.proc->procLatch);
/*
* Process any requests or signals received recently.
*/
if (t_thrd.walwriterauxiliary_cxt.got_SIGHUP) {
t_thrd.walwriterauxiliary_cxt.got_SIGHUP = false;
ProcessConfigFile(PGC_SIGHUP);
}
if (t_thrd.walwriterauxiliary_cxt.shutdown_requested) {
/* Normal exit from the walwriterauxiliary is here. */
proc_exit(0); /* done */
}
XLogMultiFileInit(g_instance.attr.attr_storage.wal_file_init_num);
if (cur_timeout > 0) {
rc = WaitLatch(&t_thrd.proc->procLatch, WL_LATCH_SET | WL_TIMEOUT | WL_POSTMASTER_DEATH, cur_timeout);
}
/*
* Emergency bailout if postmaster has died. This is to avoid the
* necessity for manual cleanup of all postmaster children.
*/
if (rc & WL_POSTMASTER_DEATH) {
gs_thread_exit(1);
}
}
}
/* --------------------------------
* signal handler routines
* --------------------------------
*/
/*
* wal_quickdie() occurs when signalled SIGQUIT by the postmaster.
*
* Some backend has bought the farm,
* so we need to stop what we're doing and exit.
*/
static void walwriterauxiliary_quickdie(SIGNAL_ARGS)
{
gs_signal_setmask(&t_thrd.libpq_cxt.BlockSig, NULL);
/*
* We DO NOT want to run proc_exit() callbacks -- we're here because
* shared memory may be corrupted, so we don't want to try to clean up our
* transaction. Just nail the windows shut and get out of town. Now that
* there's an atexit callback to prevent third-party code from breaking
* things by calling exit() directly, we have to reset the callbacks
* explicitly to make this work as intended.
*/
on_exit_reset();
/*
* Note we do exit(2) not exit(0). This is to force the postmaster into a
* system reset cycle if some idiot DBA sends a manual SIGQUIT to a random
* backend. This is necessary precisely because we don't clean up our
* shared memory state. (The "dead man switch" mechanism in pmsignal.c
* should ensure the postmaster sees this as a crash, too, but no harm in
* being doubly sure.)
*/
exit(2);
}
/* SIGHUP: set flag to re-read config file at next convenient time */
static void WalwriterauxiliarySigHupHandler(SIGNAL_ARGS)
{
int save_errno = errno;
t_thrd.walwriterauxiliary_cxt.got_SIGHUP = true;
if (t_thrd.proc)
SetLatch(&t_thrd.proc->procLatch);
errno = save_errno;
}
/* SIGTERM: set flag to exit normally */
static void WalwriterauxiliaryShutdownHandler(SIGNAL_ARGS)
{
int save_errno = errno;
t_thrd.walwriterauxiliary_cxt.shutdown_requested = true;
if (t_thrd.proc)
SetLatch(&t_thrd.proc->procLatch);
errno = save_errno;
}
/* SIGUSR1: used for latch wakeups */
static void walwriterauxiliary_sigusr1_handler(SIGNAL_ARGS)
{
int save_errno = errno;
latch_sigusr1_handler();
errno = save_errno;
}

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

@ -340,6 +340,27 @@ static void knl_g_bgworker_init(knl_g_bgworker_context* bgworker_cxt)
bgworker_cxt->have_crashed_worker = false;
}
static void knl_g_wal_init(knl_g_wal_context *wal_cxt)
{
wal_cxt->walInsertStatusTable = NULL;
wal_cxt->walFlushWaitLock = NULL;
wal_cxt->walBufferInitWaitLock = NULL;
wal_cxt->walInitSegLock = NULL;
wal_cxt->isWalWriterUp = false;
wal_cxt->flushResult = InvalidXLogRecPtr;
wal_cxt->sentResult = InvalidXLogRecPtr;
wal_cxt->flushResultMutex = PTHREAD_MUTEX_INITIALIZER;
wal_cxt->flushResultCV = (pthread_cond_t)PTHREAD_COND_INITIALIZER;
wal_cxt->XLogFlusherCPU = 0;
wal_cxt->isWalWriterSleeping = false;
wal_cxt->criticalEntryMutex = PTHREAD_MUTEX_INITIALIZER;
wal_cxt->criticalEntryCV = (pthread_cond_t)PTHREAD_COND_INITIALIZER;
wal_cxt->XLogFlushWaitTime = 0.0;
wal_cxt->globalEndPosSegNo = InvalidXLogSegPtr;
wal_cxt->walWaitFlushCount3 = 0;
wal_cxt->lastWalStatusEntryFlushed = -1;
}
static void knl_g_mot_init(knl_g_mot_context* mot_cxt)
{
mot_cxt->jitExecMode = JitExec::JIT_EXEC_MODE_INVALID;
@ -395,6 +416,7 @@ void knl_instance_init()
knl_g_numa_init(&g_instance.numa_cxt);
knl_g_mot_init(&g_instance.mot_cxt);
knl_g_bgworker_init(&g_instance.bgworker_cxt);
knl_g_wal_init(&g_instance.wal_cxt);
MemoryContextSwitchTo(old_cxt);

7
src/gausskernel/process/threadpool/knl_thread.cpp
View File

@ -951,6 +951,12 @@ static void knl_t_walwriter_init(knl_t_walwriter_context* walwriter_cxt)
walwriter_cxt->shutdown_requested = false;
}
static void knl_t_walwriterauxiliary_init(knl_t_walwriterauxiliary_context *walwriterauxiliary_cxt)
{
walwriterauxiliary_cxt->got_SIGHUP = false;
walwriterauxiliary_cxt->shutdown_requested = false;
}
static void knl_t_poolcleaner_init(knl_t_poolcleaner_context* poolcleaner_cxt)
{
poolcleaner_cxt->shutdown_requested;
@ -1525,6 +1531,7 @@ void knl_thread_init(knl_thread_role role)
knl_t_walreceiverfuncs_init(&t_thrd.walreceiverfuncs_cxt);
knl_t_walsender_init(&t_thrd.walsender_cxt);
knl_t_walwriter_init(&t_thrd.walwriter_cxt);
knl_t_walwriterauxiliary_init(&t_thrd.walwriterauxiliary_cxt);
knl_t_catchup_init(&t_thrd.catchup_cxt);
knl_t_wlm_init(&t_thrd.wlm_cxt);
knl_t_xact_init(&t_thrd.xact_cxt);

2
src/gausskernel/storage/access/nbtree/nbtinsert.cpp
View File

@ -1605,7 +1605,7 @@ static void _bt_insert_parent(Relation rel, Buffer buf, Buffer rbuf, BTStack sta
/* Check for error only after writing children */
if (pbuf == InvalidBuffer) {
XLogFlush(t_thrd.xlog_cxt.LogwrtResult->Write);
XLogWaitFlush(t_thrd.xlog_cxt.LogwrtResult->Write);
ereport(ERROR,
(errcode(ERRCODE_INDEX_CORRUPTED),
errmsg("failed to re-find parent key in index \"%s\" for split pages %u/%u",

2
src/gausskernel/storage/access/transam/clog.cpp
View File

@ -1094,7 +1094,7 @@ static void WriteTruncateXlogRec(int64 pageno)
XLogBeginInsert();
XLogRegisterData((char*)(&pageno), sizeof(int64));
recptr = XLogInsert(RM_CLOG_ID, CLOG_TRUNCATE);
XLogFlush(recptr);
XLogWaitFlush(recptr);
}
/*

2
src/gausskernel/storage/access/transam/double_write.cpp
View File

@ -1087,7 +1087,7 @@ static void dw_flush(dw_context_t* dw_ctx, XLogRecPtr latest_lsn, ThrdDwCxt* thr
dw_assemble_batch(dw_ctx, offset_page, file_head->head.dwn);
if (!XLogRecPtrIsInvalid(latest_lsn)) {
XLogFlush(latest_lsn);
XLogWaitFlush(latest_lsn);
}
pgstat_report_waitevent(WAIT_EVENT_DW_WRITE);
dw_pwrite_file(dw_ctx->fd, dw_ctx->buf, (pages_to_write * BLCKSZ), (offset_page * BLCKSZ));

2
src/gausskernel/storage/access/transam/slru.cpp
View File

@ -744,7 +744,7 @@ static bool SlruPhysicalWritePage(SlruCtl ctl, int64 pageno, int slotno, SlruFlu
* section anyway, but let's make sure.
*/
START_CRIT_SECTION();
XLogFlush(max_lsn);
XLogWaitFlush(max_lsn);
END_CRIT_SECTION();
}
}

16
src/gausskernel/storage/access/transam/twophase.cpp
View File

@ -1673,20 +1673,14 @@ void EndPrepare(GlobalTransaction gxact)
}
gxact->prepare_end_lsn = XLogInsert(RM_XACT_ID, XLOG_XACT_PREPARE);
XLogFlush(gxact->prepare_end_lsn);
if (u_sess->attr.attr_storage.guc_synchronous_commit < SYNCHRONOUS_COMMIT_REMOTE_RECEIVE) {
XLogWaitFlush(gxact->prepare_end_lsn);
}
/* If we crash now, we have prepared: WAL replay will fix things */
/* Store record's start location to read that later on Commit */
gxact->prepare_start_lsn = t_thrd.xlog_cxt.ProcLastRecPtr;
/*
* Wake up all walsenders to send WAL up to the PREPARE record immediately
* if replication is enabled
*/
if (g_instance.attr.attr_storage.max_wal_senders > 0) {
WalSndWakeup();
}
/*
* Mark the prepared transaction as valid. As soon as xact.c marks
* MyPgXact as not running our XID (which it will do immediately after
@ -3137,7 +3131,7 @@ static void RecordTransactionCommitPrepared(TransactionId xid, int nchildren, Tr
* a contradiction)
*/
/* Flush XLOG to disk */
XLogFlush(recptr);
XLogWaitFlush(recptr);
/*
* Wake up all walsenders to send WAL up to the COMMIT PREPARED record
@ -3220,7 +3214,7 @@ static void RecordTransactionAbortPrepared(TransactionId xid, int nchildren, Tra
}
/* Always flush, since we're about to remove the 2PC state file */
XLogFlush(recptr);
XLogWaitFlush(recptr);
/*
* Wake up all walsenders to send WAL up to the ABORT PREPARED record

8
src/gausskernel/storage/access/transam/xact.cpp
View File

@ -1721,11 +1721,8 @@ static TransactionId RecordTransactionCommit(void)
* We do not sleep if u_sess->attr.attr_storage.enableFsync is not turned on, nor if there are
* fewer than u_sess->attr.attr_storage.CommitSiblings other backends with active transactions.
*/
if (u_sess->attr.attr_storage.CommitDelay > 0 && u_sess->attr.attr_storage.enableFsync &&
MinimumActiveBackends(u_sess->attr.attr_storage.CommitSiblings))
pg_usleep(u_sess->attr.attr_storage.CommitDelay);
XLogFlush(t_thrd.xlog_cxt.XactLastRecEnd);
/* Wait for local flush only when we don't wait for the remote server */
XLogWaitFlush(t_thrd.xlog_cxt.XactLastRecEnd);
/* Now we may update the CLOG, if we wrote a COMMIT record above */
if (markXidCommitted) {
@ -1744,7 +1741,6 @@ static TransactionId RecordTransactionCommit(void)
* Report the latest async commit LSN, so that the WAL writer knows to
* flush this commit.
*/
XLogSetAsyncXactLSN(t_thrd.xlog_cxt.XactLastRecEnd);
/*
* We must not immediately update the CLOG, since we didn't flush the

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

File diff suppressed because it is too large Load Diff

4
src/gausskernel/storage/buffer/bufmgr.cpp
View File

@ -1347,7 +1347,7 @@ void PageListBackWrite(uint32* buf_list, int32 nbufs, uint32 flags = 0, SMgrRela
* changes they describe do.
*/
recptr = BufferGetLSN(bufHdr);
XLogFlush(recptr);
XLogWaitFlush(recptr);
/*
* Now it's safe to write the buffer. The io_in_progress
@ -4111,7 +4111,7 @@ void FlushBuffer(void* buf, SMgrRelation reln, ReadBufferMethod flushmethod)
*/
logicalpage = PageIsLogical((Block)bufferinfo.pageinfo.page);
XLogFlush(bufferinfo.lsn, logicalpage);
XLogWaitFlush(bufferinfo.lsn);
/*
* Now it's safe to write buffer to disk. Note that no one else should

9
src/gausskernel/storage/ipc/ipci.cpp
View File

@ -135,6 +135,7 @@ void CreateSharedMemoryAndSemaphores(bool makePrivate, int port)
size = add_size(size, PredicateLockShmemSize());
size = add_size(size, ProcGlobalShmemSize());
size = add_size(size, XLOGShmemSize());
size = add_size(size, XLogStatShmemSize());
size = add_size(size, CLOGShmemSize());
size = add_size(size, CSNLOGShmemSize());
size = add_size(size, TwoPhaseShmemSize());
@ -198,6 +199,7 @@ void CreateSharedMemoryAndSemaphores(bool makePrivate, int port)
*/
numSemas = ProcGlobalSemas();
numSemas += SpinlockSemas();
numSemas += XLogSemas();
#ifdef ENABLED_DEBUG_SYNC
numSemas += 1; /* For debug sync handling */
@ -241,6 +243,7 @@ void CreateSharedMemoryAndSemaphores(bool makePrivate, int port)
* Set up xlog, clog, and buffers
*/
XLOGShmemInit();
XLogStatShmemInit();
dw_shmem_init();
{
@ -298,6 +301,12 @@ void CreateSharedMemoryAndSemaphores(bool makePrivate, int port)
}
ReplicationSlotsShmemInit();
WalSndShmemInit();
/*
* Set up WAL semaphores. This must be done after WalSndShmemInit().
*/
if (!IsUnderPostmaster) {
InitWalSemaphores();
}
WalRcvShmemInit();
DataSndShmemInit();
DataRcvShmemInit();

3
src/gausskernel/storage/lmgr/lwlock.cpp
View File

@ -150,6 +150,9 @@ static const char *BuiltinTrancheNames[] = {
"MultiXactMember Ctl",
"OldSerXid SLRU Ctl",
"WALInsertLock",
"WALFlushWaitLock",
"WALBufferInitLock",
"WALInitSegLock",
"DoubleWriteLock",
"LWTRANCHE_ACCOUNT_TABLE",
"GeneralExtendedLock",

1
src/gausskernel/storage/lmgr/proc.cpp
View File

@ -248,6 +248,7 @@ void InitProcGlobal(void)
g_instance.proc_base->startupProcPid = 0;
g_instance.proc_base->startupBufferPinWaitBufId = -1;
g_instance.proc_base->walwriterLatch = NULL;
g_instance.proc_base->walwriterauxiliaryLatch = NULL;
g_instance.proc_base->checkpointerLatch = NULL;
g_instance.proc_base->cbmwriterLatch = NULL;
pg_atomic_init_u32(&g_instance.proc_base->procArrayGroupFirst, INVALID_PGPROCNO);

2
src/gausskernel/storage/replication/logical/logical.cpp
View File

@ -254,7 +254,7 @@ LogicalDecodingContext* CreateInitDecodingContext(const char* plugin, List* outp
flushptr = LogStandbySnapshot();
/* and make sure it's fsynced to disk */
XLogFlush(flushptr);
XLogWaitFlush(flushptr);
} else
slot->data.restart_lsn = GetRedoRecPtr();

10
src/gausskernel/storage/replication/slotfuncs.cpp
View File

@ -54,7 +54,7 @@ void log_slot_create(const ReplicationSlotPersistentData* slotInfo)
XLogRegisterData((char*)&xlrec, SizeOfSlotHeader);
recptr = XLogInsert(RM_SLOT_ID, XLOG_SLOT_CREATE);
XLogFlush(recptr);
XLogWaitFlush(recptr);
if (g_instance.attr.attr_storage.max_wal_senders > 0)
WalSndWakeup();
@ -81,7 +81,7 @@ void log_slot_advance(const ReplicationSlotPersistentData* slotInfo)
XLogRegisterData((char*)&xlrec, SizeOfSlotHeader);
Ptr = XLogInsert(RM_SLOT_ID, XLOG_SLOT_ADVANCE);
XLogFlush(Ptr);
XLogWaitFlush(Ptr);
if (g_instance.attr.attr_storage.max_wal_senders > 0)
WalSndWakeup();
END_CRIT_SECTION();
@ -101,7 +101,7 @@ void log_slot_drop(const char* name)
XLogRegisterData((char*)&xlrec, SizeOfSlotHeader);
Ptr = XLogInsert(RM_SLOT_ID, XLOG_SLOT_DROP);
XLogFlush(Ptr);
XLogWaitFlush(Ptr);
if (g_instance.attr.attr_storage.max_wal_senders > 0)
WalSndWakeup();
END_CRIT_SECTION();
@ -124,7 +124,7 @@ void LogCheckSlot()
XLogRegisterData((char*)LogicalSlot, size);
recptr = XLogInsert(RM_SLOT_ID, XLOG_SLOT_CHECK);
XLogFlush(recptr);
XLogWaitFlush(recptr);
if (g_instance.attr.attr_storage.max_wal_senders > 0)
WalSndWakeup();
@ -892,7 +892,7 @@ void write_term_log(uint32 term)
XLogRegisterData((char*)&term, sizeof(uint32));
recptr = XLogInsert(RM_SLOT_ID, XLOG_TERM_LOG);
XLogFlush(recptr);
XLogWaitFlush(recptr);
if (g_instance.attr.attr_storage.max_wal_senders > 0) {
WalSndWakeup();
}

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

@ -2874,7 +2874,7 @@ static int WalSndLoop(WalSndSendDataCallback send_data)
* work to do, continue to loop.
*/
if (XLogNeedsFlush(WriteRqstPtr)) {
XLogFlush(WriteRqstPtr);
XLogWaitFlush(WriteRqstPtr);
ereport(LOG,
(errmsg("the latest WAL flush to %X/%X.", (uint32)(WriteRqstPtr >> 32), (uint32)WriteRqstPtr)));
} else {

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

@ -22,7 +22,9 @@
#include "access/parallel_recovery/redo_item.h"
#include "knl/knl_instance.h"
#include "access/htup.h"
#include "storage/lwlock.h"
#include <time.h>
/* Sync methods */
#define SYNC_METHOD_FSYNC 0
#define SYNC_METHOD_FDATASYNC 1
@ -134,6 +136,57 @@ extern const int DemoteModeNum;
#define DemoteModeDesc(mode) (((mode) > 0 && (mode) < DemoteModeNum) ? DemoteModeDescs[(mode)] : DemoteModeDescs[0])
typedef struct {
LWLock* lock;
PGSemaphoreData sem;
} WALFlushWaitLock;
typedef struct {
LWLock* lock;
PGSemaphoreData sem;
} WALInitSegLock;
typedef struct {
LWLock* lock;
PGSemaphoreData sem;
} WALBufferInitWaitLock;
#define WAL_INSERT_STATUS_ENTRIES 4194304
#define WAL_NOT_COPIED 0
#define WAL_COPIED 1
#define WAL_COPY_SUSPEND (-1)
/* (ientry + 1) % WAL_INSERT_STATUS_ENTRIES */
#define GET_NEXT_STATUS_ENTRY(ientry) ((ientry + 1) & (WAL_INSERT_STATUS_ENTRIES - 1))
#define GET_STATUS_ENTRY_INDEX(ientry) ientry
struct WalInsertStatusEntry {
/* The end LSN of the record corresponding to this entry */
uint64 endLSN;
/* The log record counter of the record corresponding to this entry */
int32 LRC;
/* WAL copy status: "0" - not copied; "1" - copied */
uint32 status;
};
struct WALFlushWaitLockPadded {
WALFlushWaitLock l;
char padding[PG_CACHE_LINE_SIZE];
};
struct WALBufferInitWaitLockPadded {
WALBufferInitWaitLock l;
char padding[PG_CACHE_LINE_SIZE];
};
struct WALInitSegLockPadded {
WALInitSegLock l;
char padding[PG_CACHE_LINE_SIZE];
};
/*
* OR-able request flag bits for checkpoints. The "cause" bits are used only
* for logging purposes. Note: the flags must be defined so that it's
@ -249,8 +302,10 @@ typedef struct XLogwrtResult {
XLogRecPtr Flush; /* last byte + 1 flushed */
} XLogwrtResult;
extern void XLogMultiFileInit(int advance_xlog_file_num);
extern XLogRecPtr XLogInsertRecord(struct XLogRecData* rdata, XLogRecPtr fpw_lsn, bool isupgrade = false);
extern void XLogFlush(XLogRecPtr record, bool LogicalPage = false);
extern void XLogWaitFlush(XLogRecPtr recptr);
extern void XLogWaitBufferInit(XLogRecPtr recptr);
extern void UpdateMinRecoveryPoint(XLogRecPtr lsn, bool force);
extern bool XLogBackgroundFlush(void);
extern bool XLogNeedsFlush(XLogRecPtr RecPtr);
@ -305,6 +360,8 @@ extern void SetThisTimeID(uint64 timelineID);
extern Size XLOGShmemSize(void);
extern void XLOGShmemInit(void);
extern void BootStrapXLOG(void);
extern int XLogSemas(void);
extern void InitWalSemaphores(void);
extern void StartupXLOG(void);
extern void ShutdownXLOG(int code, Datum arg);
extern void InitXLOGAccess(void);

14
src/include/access/xlog_internal.h
View File

@ -219,6 +219,20 @@ typedef struct RmgrData {
bool (*rm_safe_restartpoint)(void);
} RmgrData;
/*
* New XLogCtlInsert Structure.
*/
struct Combined128 {
uint64 currentBytePos;
uint32 byteSize;
int32 LRC;
};
union Union128 {
uint128_u value;
struct Combined128 struct128;
};
extern const RmgrData RmgrTable[];
/*

1
src/include/gs_thread.h
View File

@ -85,6 +85,7 @@ typedef enum knl_thread_role {
SNAPSHOT_WORKER,
CHECKPOINT_THREAD,
WALWRITER,
WALWRITERAUXILIARY,
WALRECEIVER,
WALRECWRITE,
DATARECIVER,

4
src/include/knl/knl_guc/knl_instance_attr_storage.h
View File

@ -62,7 +62,11 @@ typedef struct knl_instance_attr_storage {
int max_prepared_xacts;
int max_locks_per_xact;
int max_predicate_locks_per_xact;
int64 xlog_idle_flushes_before_sleep;
int num_xloginsert_locks;
int wal_writer_cpu;
int xlog_flush_uplimit;
int wal_file_init_num;
int XLOGbuffers;
int max_wal_senders;
int max_replication_slots;

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

@ -115,6 +115,7 @@ typedef struct knl_g_pid_context {
ThreadId* PageWriterPID;
ThreadId CheckpointerPID;
ThreadId WalWriterPID;
ThreadId WalWriterAuxiliaryPID;
ThreadId WalReceiverPID;
ThreadId WalRcvWriterPID;
ThreadId DataReceiverPID;
@ -548,6 +549,39 @@ typedef struct knl_g_mot_context {
JitExec::JitExecMode jitExecMode;
} knl_g_mot_context;
typedef struct WalInsertStatusEntry WALInsertStatusEntry;
typedef struct WALFlushWaitLockPadded WALFlushWaitLockPadded;
typedef struct WALBufferInitWaitLockPadded WALBufferInitWaitLockPadded;
typedef struct WALInitSegLockPadded WALInitSegLockPadded;
typedef struct knl_g_wal_context {
/* Start address of WAL insert status table that contains WAL_INSERT_STATUS_ENTRIES entries */
WALInsertStatusEntry* walInsertStatusTable;
WALFlushWaitLockPadded* walFlushWaitLock;
WALBufferInitWaitLockPadded* walBufferInitWaitLock;
WALInitSegLockPadded* walInitSegLock;
volatile bool isWalWriterUp;
XLogRecPtr flushResult;
XLogRecPtr sentResult;
pthread_mutex_t flushResultMutex;
pthread_cond_t flushResultCV;
int XLogFlusherCPU;
volatile bool isWalWriterSleeping;
pthread_mutex_t criticalEntryMutex;
pthread_cond_t criticalEntryCV;
volatile uint32 walWaitFlushCount3;
volatile double XLogFlushWaitTime;
/*
* walWaitFlushCount3 and XLogFlushWaitTime are only for xlog statistics use.
* We need this variable as well as XLogStat_shared->xlogFlushWaitTime because
* we want to only update XLogStat_shared->xlogFlushWaitTime together with other
* fields of XLogStat_shared so that we can probably get a consistent snapshot
* of statistics data without resorting to locks.
*/
volatile XLogSegNo globalEndPosSegNo; /* Global variable for init xlog segment files. */
int lastWalStatusEntryFlushed;
} knl_g_wal_context;
typedef struct knl_instance_context {
knl_virtual_role role;
volatile int status;
@ -600,6 +634,7 @@ typedef struct knl_instance_context {
MemoryContext error_context;
MemoryContext signal_context;
MemoryContext increCheckPoint_context;
MemoryContext wal_context;
MemoryContext account_context;
knl_instance_attr_t attr;
@ -615,6 +650,7 @@ typedef struct knl_instance_context {
knl_g_bgwriter_context bgwriter_cxt;
struct knl_g_dw_context dw_cxt;
knl_g_shmem_context shmem_cxt;
knl_g_wal_context wal_cxt;
knl_g_executor_context exec_cxt;
knl_g_heartbeat_context heartbeat_cxt;
knl_g_rto_context rto_cxt;

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

@ -1784,6 +1784,11 @@ typedef struct knl_t_walwriter_context {
volatile sig_atomic_t shutdown_requested;
} knl_t_walwriter_context;
typedef struct knl_t_walwriterauxiliary_context {
volatile sig_atomic_t got_SIGHUP;
volatile sig_atomic_t shutdown_requested;
} knl_t_walwriterauxiliary_context;
typedef struct knl_t_poolcleaner_context {
volatile sig_atomic_t shutdown_requested;
} knl_t_poolcleaner_context;
@ -2872,6 +2877,7 @@ typedef struct knl_thrd_context {
knl_t_walreceiver_context walreceiver_cxt;
knl_t_walreceiverfuncs_context walreceiverfuncs_cxt;
knl_t_walwriter_context walwriter_cxt;
knl_t_walwriterauxiliary_context walwriterauxiliary_cxt;
knl_t_catchup_context catchup_cxt;
knl_t_wlmthrd_context wlm_cxt;
knl_t_xact_context xact_cxt;

2
src/include/miscadmin.h
View File

@ -311,6 +311,7 @@ typedef enum {
BgWriterProcess,
CheckpointerProcess,
WalWriterProcess,
WalWriterAuxiliaryProcess,
WalReceiverProcess,
WalRcvWriterProcess,
DataReceiverProcess,
@ -351,6 +352,7 @@ typedef enum {
#define AmMulitBackgroundWriterProcess() (t_thrd.bootstrap_cxt.MyAuxProcType == MultiBgWriterProcess)
#define AmCheckpointerProcess() (t_thrd.bootstrap_cxt.MyAuxProcType == CheckpointerProcess)
#define AmWalWriterProcess() (t_thrd.bootstrap_cxt.MyAuxProcType == WalWriterProcess)
#define AmWalWriterAuxiliaryProcess() (t_thrd.bootstrap_cxt.MyAuxProcType == WalWriterAuxiliaryProcess)
#define AmWalReceiverProcess() (t_thrd.bootstrap_cxt.MyAuxProcType == WalReceiverProcess)
#define AmWalReceiverWriterProcess() (t_thrd.bootstrap_cxt.MyAuxProcType == WalRcvWriterProcess)
#define AmDataReceiverProcess() (t_thrd.bootstrap_cxt.MyAuxProcType == DataReceiverProcess)

25
src/include/pgstat.h
View File

@ -2441,6 +2441,31 @@ extern void resetBadBlockStat();
extern bool CalcSQLRowStatCounter(
PgStat_TableCounts* last_total_counter, PgStat_TableCounts* current_sql_table_counter);
extern void GetCurrentTotalTableCounter(PgStat_TableCounts* total_table_counter);
typedef struct XLogStat_Collect {
double entryScanTime;
double IOTime;
double memsetTime;
double entryUpdateTime;
uint64 writeBytes;
uint64 scanEntryCount;
uint64 writeSomethingCount;
uint64 flushWaitCount;
double xlogFlushWaitTime;
uint32 walAuxWakeNum;
XLogRecPtr writeRqstPtr;
XLogRecPtr minCopiedPtr;
double IONotificationTime;
double sendBufferTime;
double memsetNotificationTime;
uint32 remoteFlushWaitCount;
} XLogStat_Collect;
extern THR_LOCAL XLogStat_Collect *XLogStat_shared;
extern void XLogStatShmemInit(void);
extern Size XLogStatShmemSize(void);
extern bool CheckUserExist(Oid userId, bool removeCount);
void pgstat_init_sql_rt_info_array(knl_g_stat_context* stat_cxt);
#endif /* PGSTAT_H */

15
src/include/postmaster/walwriterauxiliary.h Normal file
View File

@ -0,0 +1,15 @@
/* -------------------------------------------------------------------------
*
* walwriterauxiliary.h
* Exports from postmaster/walwriterauxiliary.c.
*
* src/include/postmaster/walwriterauxiliary.h
*
* -------------------------------------------------------------------------
*/
#ifndef _WALWRITERAUXILIARY_H
#define _WALWRITERAUXILIARY_H
extern void WalWriterAuxiliaryMain(void);
#endif /* _WALWRITERAUXILIARY_H */

3
src/include/storage/lwlock.h
View File

@ -138,6 +138,9 @@ enum BuiltinTrancheIds {
LWTRANCHE_MULTIXACTMEMBER_CTL,
LWTRANCHE_OLDSERXID_SLRU_CTL,
LWTRANCHE_WAL_INSERT,
LWTRANCHE_WAL_FLUSH_WAIT,
LWTRANCHE_WAL_BUFFER_INIT_WAIT,
LWTRANCHE_WAL_INIT_SEG,
LWTRANCHE_DOUBLE_WRITE,
LWTRANCHE_ACCOUNT_TABLE,
LWTRANCHE_EXTEND, // For general 3rd plugin

2
src/include/storage/proc.h
View File

@ -314,6 +314,8 @@ typedef struct PROC_HDR {
pg_atomic_uint32 clogGroupFirst;
/* WALWriter process's latch */
Latch* walwriterLatch;
/* WALWriterAuxiliary process's latch */
Latch* walwriterauxiliaryLatch;
/* Checkpointer process's latch */
Latch* checkpointerLatch;
/* BCMWriter process's latch */

10
src/include/utils/atomic.h
View File

@ -114,6 +114,16 @@ static inline bool gs_compare_and_swap_64(int64* dest, int64 oldval, int64 newva
return __sync_bool_compare_and_swap(dest, oldval, newval);
}
static inline uint32 gs_compare_and_swap_u32(volatile uint32* ptr, uint32 oldval, uint32 newval)
{
return (uint32)__sync_val_compare_and_swap(ptr, oldval, newval);
}
static inline uint64 gs_compare_and_swap_u64(volatile uint64* ptr, uint64 oldval, uint64 newval)
{
return (uint64)__sync_val_compare_and_swap(ptr, oldval, newval);
}
/*
* @Description: Atomic init in a 32-bit address.
* @IN ptr: int32 pointer

2
src/test/regress/expected/backtrace_log.out
View File

@ -8,7 +8,7 @@ select count(*) from pg_node_env;
select count(*) from pg_os_threads;
count
-------
12
13
(1 row)
-- test backtrace output to log