diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index 0f09add..0f18931 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -170,9 +170,42 @@ HotStandbyState standbyState = STANDBY_DISABLED; static XLogRecPtr LastRec; -/* Local copy of WalRcv->receivedUpto */ -static XLogRecPtr receivedUpto = 0; -static TimeLineID receiveTLI = 0; +/* Local copy of WalRcv->flushedUpto */ +static XLogRecPtr flushedUpto = 0; +static TimeLineID flushedTLI = 0; + +/* NOTES FOR REVIEWERs: + * + * Make WALWriter responsible for flushing received WAL data during recovery. + * At shutdown, WALReceiver flushes WAL just as it did before. WALReceiver + * now spends less time waiting and more time communicating, which should + * improve replication performance overall. + * + * New possibility of being swamped by incoming data that cannot be flushed + * fast enough; that is not addressed in current patch, but some form of + * safety valve would be required. + * + * WALWriter and WALReceiver now need to cooperate via shmem. Startup proc + * is unaffected, it just reads the current position, so it doesn't matter + * who updates it. + * + * Previously the WALReceiver was responsible for both writing and flushing + * WAL files during recovery. As a result some of the variable names were + * called "received" when in fact we meant "flushed", because it was at + * that time the same thing. + * + * WALReceiver now sends status messages earlier than it did before, making + * synchronous_commit = remote_write a more useful option. Some pacing is + * also required, since status messages could be sent more frequently than + * before. + * + * Attention needs to be paid to points where no further data is sent, since + * the WALReceiver must write, then wait for flush and apply. Both + * WalWriter and Startup process must wake the WALReceiver; not checked yet. + * + * WALWriter needs to be started during recovery, but no other behaviour + * changes at the postmaster level, since we do not depend upon it. + */ /* * During recovery, lastFullPageWrites keeps track of full_page_writes that @@ -8278,7 +8311,7 @@ CreateRestartPoint(int flags) * Get the current end of xlog replayed or received, whichever is * later. */ - receivePtr = GetWalRcvWriteRecPtr(NULL, NULL); + receivePtr = GetWalRcvFlushRecPtr(NULL, NULL); replayPtr = GetXLogReplayRecPtr(&replayTLI); endptr = (receivePtr < replayPtr) ? replayPtr : receivePtr; @@ -10173,7 +10206,7 @@ retry: /* See if we need to retrieve more data */ if (readFile < 0 || (readSource == XLOG_FROM_STREAM && - receivedUpto < targetPagePtr + reqLen)) + flushedUpto < targetPagePtr + reqLen)) { if (!WaitForWALToBecomeAvailable(targetPagePtr + reqLen, private->randAccess, @@ -10204,10 +10237,10 @@ retry: */ if (readSource == XLOG_FROM_STREAM) { - if (((targetPagePtr) / XLOG_BLCKSZ) != (receivedUpto / XLOG_BLCKSZ)) + if (((targetPagePtr) / XLOG_BLCKSZ) != (flushedUpto / XLOG_BLCKSZ)) readLen = XLOG_BLCKSZ; else - readLen = receivedUpto % XLogSegSize - targetPageOff; + readLen = flushedUpto % XLogSegSize - targetPageOff; } else readLen = XLOG_BLCKSZ; @@ -10387,7 +10420,7 @@ WaitForWALToBecomeAvailable(XLogRecPtr RecPtr, bool randAccess, curFileTLI = tli; RequestXLogStreaming(tli, ptr, PrimaryConnInfo, PrimarySlotName); - receivedUpto = 0; + flushedUpto = 0; } /* @@ -10535,14 +10568,14 @@ WaitForWALToBecomeAvailable(XLogRecPtr RecPtr, bool randAccess, * XLogReceiptTime will not advance, so the grace time * allotted to conflicting queries will decrease. */ - if (RecPtr < receivedUpto) + if (RecPtr < flushedUpto) havedata = true; else { XLogRecPtr latestChunkStart; - receivedUpto = GetWalRcvWriteRecPtr(&latestChunkStart, &receiveTLI); - if (RecPtr < receivedUpto && receiveTLI == curFileTLI) + flushedUpto = GetWalRcvFlushRecPtr(&latestChunkStart, &flushedTLI); + if (RecPtr < flushedUpto && flushedTLI == curFileTLI) { havedata = true; if (latestChunkStart <= RecPtr) @@ -10568,9 +10601,9 @@ WaitForWALToBecomeAvailable(XLogRecPtr RecPtr, bool randAccess, if (readFile < 0) { if (!expectedTLEs) - expectedTLEs = readTimeLineHistory(receiveTLI); + expectedTLEs = readTimeLineHistory(flushedTLI); readFile = XLogFileRead(readSegNo, PANIC, - receiveTLI, + flushedTLI, XLOG_FROM_STREAM, false); Assert(readFile >= 0); } diff --git a/src/backend/access/transam/xlogfuncs.c b/src/backend/access/transam/xlogfuncs.c index 133143d..216da59 100644 --- a/src/backend/access/transam/xlogfuncs.c +++ b/src/backend/access/transam/xlogfuncs.c @@ -216,7 +216,7 @@ pg_last_xlog_receive_location(PG_FUNCTION_ARGS) { XLogRecPtr recptr; - recptr = GetWalRcvWriteRecPtr(NULL, NULL); + recptr = GetWalRcvFlushRecPtr(NULL, NULL); if (recptr == 0) PG_RETURN_NULL(); diff --git a/src/backend/postmaster/postmaster.c b/src/backend/postmaster/postmaster.c index 5106f52..a21a4f2 100644 --- a/src/backend/postmaster/postmaster.c +++ b/src/backend/postmaster/postmaster.c @@ -1590,7 +1590,8 @@ ServerLoop(void) /* * If no background writer process is running, and we are not in a * state that prevents it, start one. It doesn't matter if this - * fails, we'll just try again later. Likewise for the checkpointer. + * fails, we'll just try again later. Likewise for the checkpointer + * and walwriter. */ if (pmState == PM_RUN || pmState == PM_RECOVERY || pmState == PM_HOT_STANDBY) @@ -1599,17 +1600,11 @@ ServerLoop(void) CheckpointerPID = StartCheckpointer(); if (BgWriterPID == 0) BgWriterPID = StartBackgroundWriter(); + if (WalWriterPID == 0) + WalWriterPID = StartWalWriter(); } /* - * Likewise, if we have lost the walwriter process, try to start a new - * one. But this is needed only in normal operation (else we cannot - * be writing any new WAL). - */ - if (WalWriterPID == 0 && pmState == PM_RUN) - WalWriterPID = StartWalWriter(); - - /* * If we have lost the autovacuum launcher, try to start a new one. We * don't want autovacuum to run in binary upgrade mode because * autovacuum might update relfrozenxid for empty tables before the diff --git a/src/backend/postmaster/walwriter.c b/src/backend/postmaster/walwriter.c index 0826f88..6c75c94 100644 --- a/src/backend/postmaster/walwriter.c +++ b/src/backend/postmaster/walwriter.c @@ -50,6 +50,7 @@ #include "libpq/pqsignal.h" #include "miscadmin.h" #include "postmaster/walwriter.h" +#include "replication/walreceiver.h" #include "storage/bufmgr.h" #include "storage/fd.h" #include "storage/ipc.h" @@ -248,6 +249,7 @@ WalWriterMain(void) { long cur_timeout; int rc; + bool work_done; /* * Advertise whether we might hibernate in this cycle. We do this @@ -285,7 +287,12 @@ WalWriterMain(void) * Do what we're here for; then, if XLogBackgroundFlush() found useful * work to do, reset hibernation counter. */ - if (XLogBackgroundFlush()) + if (RecoveryInProgress()) + work_done = XLogFlushReceived(); + else + work_done = XLogBackgroundFlush(); + + if (work_done) left_till_hibernate = LOOPS_UNTIL_HIBERNATE; else if (left_till_hibernate > 0) left_till_hibernate--; @@ -313,7 +320,6 @@ WalWriterMain(void) } } - /* -------------------------------- * signal handler routines * -------------------------------- diff --git a/src/backend/replication/walreceiver.c b/src/backend/replication/walreceiver.c index c2d4ed3..a668c89 100644 --- a/src/backend/replication/walreceiver.c +++ b/src/backend/replication/walreceiver.c @@ -12,7 +12,7 @@ * in the primary server), and then keeps receiving XLOG records and * writing them to the disk as long as the connection is alive. As XLOG * records are received and flushed to disk, it updates the - * WalRcv->receivedUpto variable in shared memory, to inform the startup + * WalRcv->flushedUpto variable in shared memory, to inform the startup * process of how far it can proceed with XLOG replay. * * If the primary server ends streaming, but doesn't disconnect, walreceiver @@ -56,6 +56,7 @@ #include "replication/walsender.h" #include "storage/ipc.h" #include "storage/pmsignal.h" +#include "storage/proc.h" #include "storage/procarray.h" #include "utils/guc.h" #include "utils/ps_status.h" @@ -136,6 +137,7 @@ static void WalRcvFetchTimeLineHistoryFiles(TimeLineID first, TimeLineID last); static void WalRcvWaitForStartPosition(XLogRecPtr *startpoint, TimeLineID *startpointTLI); static void WalRcvDie(int code, Datum arg); static void XLogWalRcvProcessMsg(unsigned char type, char *buf, Size len); +static void WalRcvUpdateProcessTitle(void); static void XLogWalRcvWrite(char *buf, Size nbytes, XLogRecPtr recptr); static void XLogWalRcvFlush(bool dying); static void XLogWalRcvSendReply(bool force, bool requestReply); @@ -546,6 +548,8 @@ WalReceiverMain(void) errmsg("could not close log segment %s: %m", XLogFileNameP(recvFileTLI, recvSegNo)))); + WalRcvUpdateProcessTitle(); + /* * Create .done file forcibly to prevent the streamed segment from * being archived later. @@ -903,6 +907,8 @@ XLogWalRcvWrite(char *buf, Size nbytes, XLogRecPtr recptr) errmsg("could not close log segment %s: %m", XLogFileNameP(recvFileTLI, recvSegNo)))); + WalRcvUpdateProcessTitle(); + /* * Create .done file forcibly to prevent the streamed segment * from being archived later. @@ -977,48 +983,131 @@ XLogWalRcvWrite(char *buf, Size nbytes, XLogRecPtr recptr) static void XLogWalRcvFlush(bool dying) { - if (LogstreamResult.Flush < LogstreamResult.Write) + /* use volatile pointer to prevent code rearrangement */ + volatile WalRcvData *walrcv = WalRcv; + + if (!dying) + { + /* Update shared-memory status */ + SpinLockAcquire(&walrcv->mutex); + if (walrcv->receivedUpto < LogstreamResult.Write) + { + walrcv->latestChunkStart = walrcv->receivedUpto; + walrcv->receivedUpto = LogstreamResult.Write; + walrcv->receivedTLI = ThisTimeLineID; + } + /* + * Track the progress of the walwriter. We do this + * everytime we receive new data, to avoid an extra + * spinlock cycle for each status message. + */ + LogstreamResult.Flush = walrcv->flushedUpto; + SpinLockRelease(&walrcv->mutex); + + if (ProcGlobal->walwriterLatch) + SetLatch(ProcGlobal->walwriterLatch); + + /* Also let the master know that we made some progress */ + XLogWalRcvSendReply(false, false); + XLogWalRcvSendHSFeedback(false); + } + else if (LogstreamResult.Flush < LogstreamResult.Write) { - /* use volatile pointer to prevent code rearrangement */ - volatile WalRcvData *walrcv = WalRcv; + WalRcvUpdateProcessTitle(); + /* Ignore the walwriter, just do it */ issue_xlog_fsync(recvFile, recvSegNo); LogstreamResult.Flush = LogstreamResult.Write; /* Update shared-memory status */ SpinLockAcquire(&walrcv->mutex); - if (walrcv->receivedUpto < LogstreamResult.Flush) + if (walrcv->flushedUpto < LogstreamResult.Flush) { - walrcv->latestChunkStart = walrcv->receivedUpto; - walrcv->receivedUpto = LogstreamResult.Flush; - walrcv->receivedTLI = ThisTimeLineID; + walrcv->flushedUpto = LogstreamResult.Flush; + walrcv->flushedTLI = walrcv->receivedTLI; } SpinLockRelease(&walrcv->mutex); + /* Signal the walwriter process that new WAL has arrived */ /* Signal the startup process and walsender that new WAL has arrived */ WakeupRecovery(); if (AllowCascadeReplication()) WalSndWakeup(); + } +} - /* Report XLOG streaming progress in PS display */ - if (update_process_title) - { - char activitymsg[50]; +static void +WalRcvUpdateProcessTitle(void) +{ + /* Report XLOG streaming progress in PS display */ + if (update_process_title) + { + char activitymsg[50]; - snprintf(activitymsg, sizeof(activitymsg), "streaming %X/%X", - (uint32) (LogstreamResult.Write >> 32), - (uint32) LogstreamResult.Write); - set_ps_display(activitymsg, false); - } + snprintf(activitymsg, sizeof(activitymsg), "streaming %X/%X", + (uint32) (LogstreamResult.Write >> 32), + (uint32) LogstreamResult.Write); + set_ps_display(activitymsg, false); + } +} - /* Also let the master know that we made some progress */ - if (!dying) - { - XLogWalRcvSendReply(false, false); - XLogWalRcvSendHSFeedback(false); - } +/* + * XLogFlushReceived() is executed only within walwriter + * + * Returns true if WAL file was flushed + */ +bool +XLogFlushReceived(void) +{ + /* use volatile pointer to prevent code rearrangement */ + volatile WalRcvData *walrcv = WalRcv; + bool flush = false; + + /* Update shared-memory status */ + SpinLockAcquire(&walrcv->mutex); + if (walrcv->flushedUpto < walrcv->receivedUpto) + { + LogstreamResult.Write = walrcv->receivedUpto; + if (ThisTimeLineID != walrcv->receivedTLI) + ThisTimeLineID = recvFileTLI = walrcv->receivedTLI; + flush = true; } + SpinLockRelease(&walrcv->mutex); + + if (!flush) + return false; + + /* + * Open file, fsync it and then close it again. + */ + XLByteToPrevSeg(LogstreamResult.Write, recvSegNo); + recvFile = XLogFileOpen(recvSegNo); + issue_xlog_fsync(recvFile, recvSegNo); + if (close(recvFile) != 0) + ereport(ERROR, + (errcode_for_file_access(), + errmsg("could not close log segment %s: %m", + XLogFileNameP(recvFileTLI, recvSegNo)))); + + LogstreamResult.Flush = LogstreamResult.Write; + + /* Update shared-memory status */ + SpinLockAcquire(&walrcv->mutex); + if (walrcv->flushedUpto < LogstreamResult.Flush) + { + walrcv->flushedUpto = LogstreamResult.Flush; + walrcv->flushedTLI = walrcv->receivedTLI; + } + SpinLockRelease(&walrcv->mutex); + + /* Signal the walwriter process that new WAL has arrived */ + /* Signal the startup process and walsender that new WAL has arrived */ + WakeupRecovery(); + if (AllowCascadeReplication()) + WalSndWakeup(); + + return true; } /* @@ -1040,8 +1129,8 @@ XLogWalRcvSendReply(bool force, bool requestReply) static XLogRecPtr writePtr = 0; static XLogRecPtr flushPtr = 0; XLogRecPtr applyPtr; - static TimestampTz sendTime = 0; - TimestampTz now; + static TimestampTz idleTime = 0; + static bool idle = false; /* * If the user doesn't want status to be reported to the master, be sure @@ -1050,29 +1139,39 @@ XLogWalRcvSendReply(bool force, bool requestReply) if (!force && wal_receiver_status_interval <= 0) return; - /* Get current timestamp. */ - now = GetCurrentTimestamp(); - /* - * We can compare the write and flush positions to the last message we - * sent without taking any lock, but the apply position requires a spin - * lock, so we don't check that unless something else has changed or 10 + * We need to grab a spinlock for flush position, but we can avoid + * getting the current timestamp if we are going to send a message + * anyway. Getting the apply position requires a spin lock, so we + * don't check that unless something else has changed or 10 * seconds have passed. This means that the apply log position will * appear, from the master's point of view, to lag slightly, but since * this is only for reporting purposes and only on idle systems, that's * probably OK. */ + LogstreamResult.Flush = GetWalRcvFlushRecPtr(NULL, NULL); if (!force && writePtr == LogstreamResult.Write && flushPtr == LogstreamResult.Flush - && !TimestampDifferenceExceeds(sendTime, now, - wal_receiver_status_interval * 1000)) + && !idle) + { + idleTime = GetCurrentTimestamp(); + idle = true; return; - sendTime = now; + } + + if (idle) + { + TimestampTz now = GetCurrentTimestamp(); + if (!TimestampDifferenceExceeds(idleTime, now, + wal_receiver_status_interval * 1000)) + return; + idle = false; + } /* Construct a new message */ writePtr = LogstreamResult.Write; - flushPtr = LogstreamResult.Flush; + flushPtr = LogstreamResult.Flush; /* updated above */ applyPtr = GetXLogReplayRecPtr(NULL); resetStringInfo(&reply_message); diff --git a/src/backend/replication/walreceiverfuncs.c b/src/backend/replication/walreceiverfuncs.c index 579216a..ed2eab6 100644 --- a/src/backend/replication/walreceiverfuncs.c +++ b/src/backend/replication/walreceiverfuncs.c @@ -268,12 +268,14 @@ RequestXLogStreaming(TimeLineID tli, XLogRecPtr recptr, const char *conninfo, /* * If this is the first startup of walreceiver (on this timeline), - * initialize receivedUpto and latestChunkStart to the starting point. + * initialize all XLogRecPtrs to the starting point. */ if (walrcv->receiveStart == 0 || walrcv->receivedTLI != tli) { walrcv->receivedUpto = recptr; walrcv->receivedTLI = tli; + walrcv->flushedUpto = recptr; + walrcv->flushedTLI = tli; walrcv->latestChunkStart = recptr; } walrcv->receiveStart = recptr; @@ -296,14 +298,14 @@ RequestXLogStreaming(TimeLineID tli, XLogRecPtr recptr, const char *conninfo, * receiveTLI. */ XLogRecPtr -GetWalRcvWriteRecPtr(XLogRecPtr *latestChunkStart, TimeLineID *receiveTLI) +GetWalRcvFlushRecPtr(XLogRecPtr *latestChunkStart, TimeLineID *receiveTLI) { /* use volatile pointer to prevent code rearrangement */ volatile WalRcvData *walrcv = WalRcv; XLogRecPtr recptr; SpinLockAcquire(&walrcv->mutex); - recptr = walrcv->receivedUpto; + recptr = walrcv->flushedUpto; if (latestChunkStart) *latestChunkStart = walrcv->latestChunkStart; if (receiveTLI) diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index 5937cbb..8684098 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -2520,8 +2520,8 @@ GetStandbyFlushRecPtr(void) { XLogRecPtr replayPtr; TimeLineID replayTLI; - XLogRecPtr receivePtr; - TimeLineID receiveTLI; + XLogRecPtr flushPtr; + TimeLineID flushTLI; XLogRecPtr result; /* @@ -2530,14 +2530,14 @@ GetStandbyFlushRecPtr(void) * has streamed, but hasn't been replayed yet. */ - receivePtr = GetWalRcvWriteRecPtr(NULL, &receiveTLI); + flushPtr = GetWalRcvFlushRecPtr(NULL, &flushTLI); replayPtr = GetXLogReplayRecPtr(&replayTLI); ThisTimeLineID = replayTLI; result = replayPtr; - if (receiveTLI == ThisTimeLineID && receivePtr > replayPtr) - result = receivePtr; + if (flushTLI == ThisTimeLineID && flushPtr > replayPtr) + result = flushPtr; return result; } diff --git a/src/include/replication/walreceiver.h b/src/include/replication/walreceiver.h index 7a249f1..ff3312d 100644 --- a/src/include/replication/walreceiver.h +++ b/src/include/replication/walreceiver.h @@ -79,6 +79,16 @@ typedef struct TimeLineID receivedTLI; /* + * flushedUpto-1 is the last byte position that has already been + * flushed, and flushedTLI is the timeline it came from. At the first + * startup of walreceiver, these are set to receiveStart and + * receiveStartTLI. After that, these are updated whenever a process + * flushes the received WAL to disk. + */ + XLogRecPtr flushedUpto; + TimeLineID flushedTLI; + + /* * latestChunkStart is the starting byte position of the current "batch" * of received WAL. It's actually the same as the previous value of * receivedUpto before the last flush to disk. Startup process can use @@ -148,6 +158,7 @@ extern PGDLLIMPORT walrcv_disconnect_type walrcv_disconnect; /* prototypes for functions in walreceiver.c */ extern void WalReceiverMain(void) __attribute__((noreturn)); +extern bool XLogFlushReceived(void); /* prototypes for functions in walreceiverfuncs.c */ extern Size WalRcvShmemSize(void); @@ -157,7 +168,7 @@ extern bool WalRcvStreaming(void); extern bool WalRcvRunning(void); extern void RequestXLogStreaming(TimeLineID tli, XLogRecPtr recptr, const char *conninfo, const char *slotname); -extern XLogRecPtr GetWalRcvWriteRecPtr(XLogRecPtr *latestChunkStart, TimeLineID *receiveTLI); +extern XLogRecPtr GetWalRcvFlushRecPtr(XLogRecPtr *latestChunkStart, TimeLineID *receiveTLI); extern int GetReplicationApplyDelay(void); extern int GetReplicationTransferLatency(void); diff --git a/src/include/storage/proc.h b/src/include/storage/proc.h index 4ad4164..b5fba3a 100644 --- a/src/include/storage/proc.h +++ b/src/include/storage/proc.h @@ -215,11 +215,10 @@ extern PGPROC *PreparedXactProcs; * We set aside some extra PGPROC structures for auxiliary processes, * ie things that aren't full-fledged backends but need shmem access. * - * Background writer, checkpointer and WAL writer run during normal operation. - * Startup process and WAL receiver also consume 2 slots, but WAL writer is - * launched only after startup has exited, so we only need 4 slots. + * Background writer, checkpointer and WAL writer run always. + * Startup process and WAL receiver also consume 2 slots during recovery. */ -#define NUM_AUXILIARY_PROCS 4 +#define NUM_AUXILIARY_PROCS 5 /* configurable options */