From 344bd7e42608e0a01aaadcbc27b09ee0137b3210 Mon Sep 17 00:00:00 2001 From: Alvaro Herrera Date: Tue, 22 Mar 2022 16:49:40 +0100 Subject: [PATCH v7 2/3] Split LogwrtResult into separate variables Since the schedule of updating each portion is independent, it's not very useful to have them both as a single struct. Before we used atomics for the corresponding XLogCtl struct this made sense because we could use struct assignment, but that's no longer the case. Suggested by Andres Freund. --- src/backend/access/transam/xlog.c | 146 +++++++++++++++--------------- 1 file changed, 71 insertions(+), 75 deletions(-) diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index 6f2eb494fe..de8c3e97a0 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -291,7 +291,8 @@ static bool doPageWrites; * using atomic operations. * * In addition to the shared variable, each backend has a private copy of - * LogwrtResult, each member of which is separately updated when convenient. + * each member of LogwrtResult (LogWriteResult and LogFlushResult), each of + * which is separately updated when convenient. * * The request bookkeeping is simpler: there is a shared XLogCtl->LogwrtRqst * (protected by info_lck), but we don't need to cache any copies of it. @@ -326,12 +327,6 @@ typedef struct XLogwrtRqst XLogRecPtr Flush; /* last byte + 1 to flush */ } XLogwrtRqst; -typedef struct XLogwrtResult -{ - XLogRecPtr Write; /* last byte + 1 written out */ - XLogRecPtr Flush; /* last byte + 1 flushed */ -} XLogwrtResult; - /* * Inserting to WAL is protected by a small fixed number of WAL insertion * locks. To insert to the WAL, you must hold one of the locks - it doesn't @@ -623,7 +618,8 @@ static int UsableBytesInSegment; * Private, possibly out-of-date copy of shared XLogCtl->LogwrtResult. * See discussion above. */ -static XLogwrtResult LogwrtResult = {0, 0}; +static XLogRecPtr LogWriteResult = 0; +static XLogRecPtr LogFlushResult = 0; /* * openLogFile is -1 or a kernel FD for an open log file segment. @@ -1807,7 +1803,7 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, TimeLineID tli, bool opportunistic) * Now that we have the lock, check if someone initialized the page * already. */ - LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); + LogWriteResult = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); while (upto >= XLogCtl->InitializedUpTo || opportunistic) { nextidx = XLogRecPtrToBufIdx(XLogCtl->InitializedUpTo); @@ -1818,7 +1814,7 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, TimeLineID tli, bool opportunistic) * already written out. */ OldPageRqstPtr = XLogCtl->xlblocks[nextidx]; - if (LogwrtResult.Write < OldPageRqstPtr) + if (LogWriteResult < OldPageRqstPtr) { /* * Nope, got work to do. If we just want to pre-initialize as much @@ -1834,11 +1830,11 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, TimeLineID tli, bool opportunistic) SpinLockRelease(&XLogCtl->info_lck); /* - * Before waiting, update LogwrtResult.Write and see if we still need + * Before waiting, update LogWriteResult and see if we still need * to write it or if someone else already did. */ - LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); - if (LogwrtResult.Write < OldPageRqstPtr) + LogWriteResult = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); + if (LogWriteResult < OldPageRqstPtr) { /* * Must acquire write lock. Release WALBufMappingLock first, @@ -1852,8 +1848,8 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, TimeLineID tli, bool opportunistic) LWLockAcquire(WALWriteLock, LW_EXCLUSIVE); - LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); - if (LogwrtResult.Write >= OldPageRqstPtr) + LogWriteResult = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); + if (LogWriteResult >= OldPageRqstPtr) { /* OK, someone wrote it already */ LWLockRelease(WALWriteLock); @@ -2098,7 +2094,7 @@ XLogWrite(XLogwrtRqst WriteRqst, TimeLineID tli, bool flexible) /* * Update local LogwrtResult (caller probably did this already, but...) */ - LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); + LogWriteResult = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); /* * Since successive pages in the xlog cache are consecutively allocated, @@ -2118,9 +2114,9 @@ XLogWrite(XLogwrtRqst WriteRqst, TimeLineID tli, bool flexible) * consider writing. Begin at the buffer containing the next unwritten * page, or last partially written page. */ - curridx = XLogRecPtrToBufIdx(LogwrtResult.Write); + curridx = XLogRecPtrToBufIdx(LogWriteResult); - while (LogwrtResult.Write < WriteRqst.Write) + while (LogWriteResult < WriteRqst.Write) { /* * Make sure we're not ahead of the insert process. This could happen @@ -2129,16 +2125,16 @@ XLogWrite(XLogwrtRqst WriteRqst, TimeLineID tli, bool flexible) */ XLogRecPtr EndPtr = XLogCtl->xlblocks[curridx]; - if (LogwrtResult.Write >= EndPtr) + if (LogWriteResult >= EndPtr) elog(PANIC, "xlog write request %X/%X is past end of log %X/%X", - LSN_FORMAT_ARGS(LogwrtResult.Write), + LSN_FORMAT_ARGS(LogWriteResult), LSN_FORMAT_ARGS(EndPtr)); - /* Advance LogwrtResult.Write to end of current buffer page */ - LogwrtResult.Write = EndPtr; - ispartialpage = WriteRqst.Write < LogwrtResult.Write; + /* Advance LogWriteResult to end of current buffer page */ + LogWriteResult = EndPtr; + ispartialpage = WriteRqst.Write < LogWriteResult; - if (!XLByteInPrevSeg(LogwrtResult.Write, openLogSegNo, + if (!XLByteInPrevSeg(LogWriteResult, openLogSegNo, wal_segment_size)) { /* @@ -2148,7 +2144,7 @@ XLogWrite(XLogwrtRqst WriteRqst, TimeLineID tli, bool flexible) Assert(npages == 0); if (openLogFile >= 0) XLogFileClose(); - XLByteToPrevSeg(LogwrtResult.Write, openLogSegNo, + XLByteToPrevSeg(LogWriteResult, openLogSegNo, wal_segment_size); openLogTLI = tli; @@ -2160,7 +2156,7 @@ XLogWrite(XLogwrtRqst WriteRqst, TimeLineID tli, bool flexible) /* Make sure we have the current logfile open */ if (openLogFile < 0) { - XLByteToPrevSeg(LogwrtResult.Write, openLogSegNo, + XLByteToPrevSeg(LogWriteResult, openLogSegNo, wal_segment_size); openLogTLI = tli; openLogFile = XLogFileOpen(openLogSegNo, tli); @@ -2172,7 +2168,7 @@ XLogWrite(XLogwrtRqst WriteRqst, TimeLineID tli, bool flexible) { /* first of group */ startidx = curridx; - startoffset = XLogSegmentOffset(LogwrtResult.Write - XLOG_BLCKSZ, + startoffset = XLogSegmentOffset(LogWriteResult - XLOG_BLCKSZ, wal_segment_size); } npages++; @@ -2183,7 +2179,7 @@ XLogWrite(XLogwrtRqst WriteRqst, TimeLineID tli, bool flexible) * contiguous in memory), or if we are at the end of the logfile * segment. */ - last_iteration = WriteRqst.Write <= LogwrtResult.Write; + last_iteration = WriteRqst.Write <= LogWriteResult; finishing_seg = !ispartialpage && (startoffset + npages * XLOG_BLCKSZ) >= wal_segment_size; @@ -2274,13 +2270,13 @@ XLogWrite(XLogwrtRqst WriteRqst, TimeLineID tli, bool flexible) /* signal that we need to wakeup walsenders later */ WalSndWakeupRequest(); - LogwrtResult.Flush = LogwrtResult.Write; /* end of page */ + LogFlushResult = LogWriteResult; /* end of page */ if (XLogArchivingActive()) XLogArchiveNotifySeg(openLogSegNo, tli); XLogCtl->lastSegSwitchTime = (pg_time_t) time(NULL); - XLogCtl->lastSegSwitchLSN = LogwrtResult.Flush; + XLogCtl->lastSegSwitchLSN = LogFlushResult; /* * Request a checkpoint if we've consumed too much xlog since @@ -2301,7 +2297,7 @@ XLogWrite(XLogwrtRqst WriteRqst, TimeLineID tli, bool flexible) if (ispartialpage) { /* Only asked to write a partial page */ - LogwrtResult.Write = WriteRqst.Write; + LogWriteResult = WriteRqst.Write; break; } curridx = NextBufIdx(curridx); @@ -2314,13 +2310,13 @@ XLogWrite(XLogwrtRqst WriteRqst, TimeLineID tli, bool flexible) Assert(npages == 0); /* Publish current write result position */ - pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Write, LogwrtResult.Write); + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Write, LogWriteResult); /* * If asked to flush, do so */ - if (LogwrtResult.Flush < WriteRqst.Flush && - LogwrtResult.Flush < LogwrtResult.Write) + if (LogFlushResult < WriteRqst.Flush && + LogFlushResult < LogWriteResult) { /* * Could get here without iterating above loop, in which case we might @@ -2331,12 +2327,12 @@ XLogWrite(XLogwrtRqst WriteRqst, TimeLineID tli, bool flexible) sync_method != SYNC_METHOD_OPEN_DSYNC) { if (openLogFile >= 0 && - !XLByteInPrevSeg(LogwrtResult.Write, openLogSegNo, + !XLByteInPrevSeg(LogWriteResult, openLogSegNo, wal_segment_size)) XLogFileClose(); if (openLogFile < 0) { - XLByteToPrevSeg(LogwrtResult.Write, openLogSegNo, + XLByteToPrevSeg(LogWriteResult, openLogSegNo, wal_segment_size); openLogTLI = tli; openLogFile = XLogFileOpen(openLogSegNo, tli); @@ -2349,11 +2345,11 @@ XLogWrite(XLogwrtRqst WriteRqst, TimeLineID tli, bool flexible) /* signal that we need to wakeup walsenders later */ WalSndWakeupRequest(); - LogwrtResult.Flush = LogwrtResult.Write; + LogFlushResult = LogWriteResult; } /* Publish current flush result position */ - pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Flush, LogwrtResult.Flush); + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Flush, LogFlushResult); /* * Make sure that the shared 'request' values do not fall behind the @@ -2362,10 +2358,10 @@ XLogWrite(XLogwrtRqst WriteRqst, TimeLineID tli, bool flexible) */ { SpinLockAcquire(&XLogCtl->info_lck); - if (XLogCtl->LogwrtRqst.Write < LogwrtResult.Write) - XLogCtl->LogwrtRqst.Write = LogwrtResult.Write; - if (XLogCtl->LogwrtRqst.Flush < LogwrtResult.Flush) - XLogCtl->LogwrtRqst.Flush = LogwrtResult.Flush; + if (XLogCtl->LogwrtRqst.Write < LogWriteResult) + XLogCtl->LogwrtRqst.Write = LogWriteResult; + if (XLogCtl->LogwrtRqst.Flush < LogFlushResult) + XLogCtl->LogwrtRqst.Flush = LogFlushResult; SpinLockRelease(&XLogCtl->info_lck); } } @@ -2381,7 +2377,7 @@ XLogSetAsyncXactLSN(XLogRecPtr asyncXactLSN) XLogRecPtr WriteRqstPtr = asyncXactLSN; bool sleeping; - LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); + LogFlushResult = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); SpinLockAcquire(&XLogCtl->info_lck); sleeping = XLogCtl->WalWriterSleeping; if (XLogCtl->asyncXactLSN < asyncXactLSN) @@ -2399,7 +2395,7 @@ XLogSetAsyncXactLSN(XLogRecPtr asyncXactLSN) WriteRqstPtr -= WriteRqstPtr % XLOG_BLCKSZ; /* if we have already flushed that far, we're done */ - if (WriteRqstPtr <= LogwrtResult.Flush) + if (WriteRqstPtr <= LogFlushResult) return; } @@ -2551,15 +2547,15 @@ XLogFlush(XLogRecPtr record) } /* Quick exit if already known flushed */ - if (record <= LogwrtResult.Flush) + if (record <= LogFlushResult) return; #ifdef WAL_DEBUG if (XLOG_DEBUG) elog(LOG, "xlog flush request %X/%X; write %X/%X; flush %X/%X", LSN_FORMAT_ARGS(record), - LSN_FORMAT_ARGS(LogwrtResult.Write), - LSN_FORMAT_ARGS(LogwrtResult.Flush)); + LSN_FORMAT_ARGS(LogWriteResult), + LSN_FORMAT_ARGS(LogFlushResult)); #endif START_CRIT_SECTION(); @@ -2568,8 +2564,8 @@ XLogFlush(XLogRecPtr record) * Since fsync is usually a horribly expensive operation, we try to * piggyback as much data as we can on each fsync: if we see any more data * entered into the xlog buffer, we'll write and fsync that too, so that - * the final value of LogwrtResult.Flush is as large as possible. This - * gives us some chance of avoiding another fsync immediately after. + * the final value of LogFlushResult is as large as possible. This gives + * us some chance of avoiding another fsync immediately after. */ /* initialize to given target; may increase below */ @@ -2590,8 +2586,8 @@ XLogFlush(XLogRecPtr record) SpinLockRelease(&XLogCtl->info_lck); /* done already? */ - LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); - if (record <= LogwrtResult.Flush) + LogFlushResult = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); + if (record <= LogFlushResult) break; /* @@ -2618,8 +2614,8 @@ XLogFlush(XLogRecPtr record) } /* Got the lock; recheck whether request is satisfied */ - LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); - if (record <= LogwrtResult.Flush) + LogFlushResult = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); + if (record <= LogFlushResult) { LWLockRelease(WALWriteLock); break; @@ -2689,11 +2685,11 @@ XLogFlush(XLogRecPtr record) * calls from bufmgr.c are not within critical sections and so we will not * force a restart for a bad LSN on a data page. */ - if (LogwrtResult.Flush < record) + if (LogFlushResult < record) elog(ERROR, "xlog flush request %X/%X is not satisfied --- flushed only to %X/%X", LSN_FORMAT_ARGS(record), - LSN_FORMAT_ARGS(LogwrtResult.Flush)); + LSN_FORMAT_ARGS(LogFlushResult)); } /* @@ -2749,8 +2745,8 @@ XLogBackgroundFlush(void) WriteRqst.Write -= WriteRqst.Write % XLOG_BLCKSZ; /* if we have already flushed that far, consider async commit records */ - LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); - if (WriteRqst.Write <= LogwrtResult.Flush) + LogFlushResult = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); + if (WriteRqst.Write <= LogFlushResult) { pg_memory_barrier(); SpinLockAcquire(&XLogCtl->info_lck); @@ -2764,12 +2760,12 @@ XLogBackgroundFlush(void) * holding an open file handle to a logfile that's no longer in use, * preventing the file from being deleted. */ - if (WriteRqst.Write <= LogwrtResult.Flush) + if (WriteRqst.Write <= LogFlushResult) { if (openLogFile >= 0) { - LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); - if (!XLByteInPrevSeg(LogwrtResult.Write, openLogSegNo, + LogWriteResult = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); + if (!XLByteInPrevSeg(LogWriteResult, openLogSegNo, wal_segment_size)) { XLogFileClose(); @@ -2784,7 +2780,7 @@ XLogBackgroundFlush(void) */ now = GetCurrentTimestamp(); flushbytes = - WriteRqst.Write / XLOG_BLCKSZ - LogwrtResult.Flush / XLOG_BLCKSZ; + WriteRqst.Write / XLOG_BLCKSZ - LogFlushResult / XLOG_BLCKSZ; if (WalWriterFlushAfter == 0 || lastflush == 0) { @@ -2819,8 +2815,8 @@ XLogBackgroundFlush(void) elog(LOG, "xlog bg flush request write %X/%X; flush: %X/%X, current is write %X/%X; flush %X/%X", LSN_FORMAT_ARGS(WriteRqst.Write), LSN_FORMAT_ARGS(WriteRqst.Flush), - LSN_FORMAT_ARGS(LogwrtResult.Write), - LSN_FORMAT_ARGS(LogwrtResult.Flush)); + LSN_FORMAT_ARGS(LogWriteResult), + LSN_FORMAT_ARGS(LogFlushResult)); #endif START_CRIT_SECTION(); @@ -2828,10 +2824,10 @@ XLogBackgroundFlush(void) /* now wait for any in-progress insertions to finish and get write lock */ WaitXLogInsertionsToFinish(WriteRqst.Write); LWLockAcquire(WALWriteLock, LW_EXCLUSIVE); - LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); - LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); - if (WriteRqst.Write > LogwrtResult.Write || - WriteRqst.Flush > LogwrtResult.Flush) + LogWriteResult = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); + LogFlushResult = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); + if (WriteRqst.Write > LogWriteResult || + WriteRqst.Flush > LogFlushResult) { XLogWrite(WriteRqst, insertTLI, flexible); } @@ -2913,14 +2909,14 @@ XLogNeedsFlush(XLogRecPtr record) } /* Quick exit if already known flushed */ - if (record <= LogwrtResult.Flush) + if (record <= LogFlushResult) return false; /* read LogwrtResult and update local state */ - LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); + LogFlushResult = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); /* check again */ - if (record <= LogwrtResult.Flush) + if (record <= LogFlushResult) return false; return true; @@ -5488,7 +5484,7 @@ StartupXLOG(void) XLogCtl->InitializedUpTo = EndOfLog; } - LogwrtResult.Write = LogwrtResult.Flush = EndOfLog; + LogWriteResult = LogFlushResult = EndOfLog; /* XXX OK to write without WALWriteLock? */ pg_atomic_write_u64(&XLogCtl->LogwrtResult.Write, EndOfLog); @@ -5928,7 +5924,7 @@ GetFlushRecPtr(TimeLineID *insertTLI) { Assert(XLogCtl->SharedRecoveryState == RECOVERY_STATE_DONE); - LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); + LogFlushResult = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); /* * If we're writing and flushing WAL, the time line can't be changing, so @@ -5937,7 +5933,7 @@ GetFlushRecPtr(TimeLineID *insertTLI) if (insertTLI) *insertTLI = XLogCtl->InsertTimeLineID; - return LogwrtResult.Flush; + return LogFlushResult; } /* @@ -9084,9 +9080,9 @@ GetXLogInsertRecPtr(void) XLogRecPtr GetXLogWriteRecPtr(void) { - LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); + LogWriteResult = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); - return LogwrtResult.Write; + return LogWriteResult; } /* -- 2.30.2