From 84a62ac0594f8f0b13c06958cf6e94e2a2723082 Mon Sep 17 00:00:00 2001 From: Alvaro Herrera Date: Fri, 29 Jan 2021 22:34:04 -0300 Subject: [PATCH v2 2/3] atomics in xlog.c --- src/backend/access/transam/xlog.c | 129 ++++++++++++++---------------- 1 file changed, 61 insertions(+), 68 deletions(-) diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index f03bd473e2..8073a92ceb 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -413,7 +413,8 @@ static XLogRecPtr RedoStartLSN = InvalidXLogRecPtr; * which is 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. + * (which use atomic access, so no locks are needed), but we don't need to + * cache any copies of it. * * info_lck is only held long enough to read/update the protected variables, * so it's a plain spinlock. The other locks are held longer (potentially @@ -433,17 +434,19 @@ static XLogRecPtr RedoStartLSN = InvalidXLogRecPtr; *---------- */ -typedef struct XLogwrtRqst +/* Write/Flush position to be used in shared memory */ +typedef struct XLogwrtAtomic { - XLogRecPtr Write; /* last byte + 1 to write out */ - XLogRecPtr Flush; /* last byte + 1 to flush */ -} XLogwrtRqst; + pg_atomic_uint64 Write; /* last byte + 1 of write position */ + pg_atomic_uint64 Flush; /* last byte + 1 of flush position */ +} XLogwrtAtomic; -typedef struct XLogwrtResult +/* Write/Flush position to be used in process-local memory */ +typedef struct XLogwrt { XLogRecPtr Write; /* last byte + 1 written out */ XLogRecPtr Flush; /* last byte + 1 flushed */ -} XLogwrtResult; +} XLogwrt; /* * Inserting to WAL is protected by a small fixed number of WAL insertion @@ -596,8 +599,10 @@ typedef struct XLogCtlData { XLogCtlInsert Insert; + XLogwrtAtomic LogwrtRqst; + XLogwrtAtomic LogwrtResult; + /* Protected by info_lck: */ - XLogwrtRqst LogwrtRqst; XLogRecPtr RedoRecPtr; /* a recent copy of Insert->RedoRecPtr */ FullTransactionId ckptFullXid; /* nextXid of latest checkpoint */ XLogRecPtr asyncXactLSN; /* LSN of newest async commit/abort */ @@ -613,12 +618,6 @@ typedef struct XLogCtlData pg_time_t lastSegSwitchTime; XLogRecPtr lastSegSwitchLSN; - /* - * Protected by info_lck and WALWriteLock (you must hold either lock to - * read it, but both to update) - */ - XLogwrtResult LogwrtResult; - /* * Latest initialized page in the cache (last byte position + 1). * @@ -779,7 +778,7 @@ static int UsableBytesInSegment; * Private, possibly out-of-date copy of shared LogwrtResult. * See discussion above. */ -static XLogwrtResult LogwrtResult = {0, 0}; +static XLogwrt LogwrtResult = {0, 0}; /* * Codes indicating where we got a WAL file from during recovery, or where @@ -910,7 +909,7 @@ static XLogRecPtr XLogGetReplicationSlotMinimumLSN(void); static void AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic); static bool XLogCheckpointNeeded(XLogSegNo new_segno); -static void XLogWrite(XLogwrtRqst WriteRqst, bool flexible); +static void XLogWrite(XLogwrt WriteRqst, bool flexible); static bool InstallXLogFileSegment(XLogSegNo *segno, char *tmppath, bool find_free, XLogSegNo max_segno, bool use_lock); @@ -1168,13 +1167,11 @@ XLogInsertRecord(XLogRecData *rdata, */ if (StartPos / XLOG_BLCKSZ != EndPos / XLOG_BLCKSZ) { - SpinLockAcquire(&XLogCtl->info_lck); /* advance global request to include new block(s) */ - if (XLogCtl->LogwrtRqst.Write < EndPos) - XLogCtl->LogwrtRqst.Write = EndPos; - /* update local result copy while I have the chance */ - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtRqst.Write, EndPos); + /* update local result copy */ + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); } /* @@ -2135,7 +2132,7 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic) XLogCtlInsert *Insert = &XLogCtl->Insert; int nextidx; XLogRecPtr OldPageRqstPtr; - XLogwrtRqst WriteRqst; + XLogwrt WriteRqst; XLogRecPtr NewPageEndPtr = InvalidXLogRecPtr; XLogRecPtr NewPageBeginPtr; XLogPageHeader NewPage; @@ -2166,12 +2163,11 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic) if (opportunistic) break; - /* Before waiting, get info_lck and update LogwrtResult */ - SpinLockAcquire(&XLogCtl->info_lck); - if (XLogCtl->LogwrtRqst.Write < OldPageRqstPtr) - XLogCtl->LogwrtRqst.Write = OldPageRqstPtr; - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + /* Before waiting, update LogwrtResult */ + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtRqst.Write, OldPageRqstPtr); + + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); /* * Now that we have an up-to-date LogwrtResult value, see if we @@ -2191,7 +2187,8 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic) LWLockAcquire(WALWriteLock, LW_EXCLUSIVE); - LogwrtResult = XLogCtl->LogwrtResult; + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); if (LogwrtResult.Write >= OldPageRqstPtr) { /* OK, someone wrote it already */ @@ -2421,7 +2418,7 @@ XLogCheckpointNeeded(XLogSegNo new_segno) * write. */ static void -XLogWrite(XLogwrtRqst WriteRqst, bool flexible) +XLogWrite(XLogwrt WriteRqst, bool flexible) { bool ispartialpage; bool last_iteration; @@ -2438,7 +2435,8 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) /* * Update local LogwrtResult (caller probably did this already, but...) */ - LogwrtResult = XLogCtl->LogwrtResult; + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); /* * Since successive pages in the xlog cache are consecutively allocated, @@ -2675,13 +2673,11 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) * code in a couple of places. */ { - SpinLockAcquire(&XLogCtl->info_lck); - XLogCtl->LogwrtResult = LogwrtResult; - if (XLogCtl->LogwrtRqst.Write < LogwrtResult.Write) - XLogCtl->LogwrtRqst.Write = LogwrtResult.Write; - if (XLogCtl->LogwrtRqst.Flush < LogwrtResult.Flush) - XLogCtl->LogwrtRqst.Flush = LogwrtResult.Flush; - SpinLockRelease(&XLogCtl->info_lck); + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Write, LogwrtResult.Write); + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Flush, LogwrtResult.Flush); + + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtRqst.Write, LogwrtResult.Write); + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtRqst.Flush, LogwrtResult.Flush); } } @@ -2696,8 +2692,9 @@ XLogSetAsyncXactLSN(XLogRecPtr asyncXactLSN) XLogRecPtr WriteRqstPtr = asyncXactLSN; bool sleeping; + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; sleeping = XLogCtl->WalWriterSleeping; if (XLogCtl->asyncXactLSN < asyncXactLSN) XLogCtl->asyncXactLSN = asyncXactLSN; @@ -2856,7 +2853,7 @@ void XLogFlush(XLogRecPtr record) { XLogRecPtr WriteRqstPtr; - XLogwrtRqst WriteRqst; + XLogwrt WriteRqst; /* * During REDO, we are reading not writing WAL. Therefore, instead of @@ -2905,11 +2902,9 @@ XLogFlush(XLogRecPtr record) XLogRecPtr insertpos; /* read LogwrtResult and update local state */ - SpinLockAcquire(&XLogCtl->info_lck); - if (WriteRqstPtr < XLogCtl->LogwrtRqst.Write) - WriteRqstPtr = XLogCtl->LogwrtRqst.Write; - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + WriteRqstPtr = pg_atomic_read_u64(&XLogCtl->LogwrtRqst.Write); + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); /* done already? */ if (record <= LogwrtResult.Flush) @@ -2939,7 +2934,8 @@ XLogFlush(XLogRecPtr record) } /* Got the lock; recheck whether request is satisfied */ - LogwrtResult = XLogCtl->LogwrtResult; + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); if (record <= LogwrtResult.Flush) { LWLockRelease(WALWriteLock); @@ -3044,7 +3040,7 @@ XLogFlush(XLogRecPtr record) bool XLogBackgroundFlush(void) { - XLogwrtRqst WriteRqst; + XLogwrt WriteRqst; bool flexible = true; static TimestampTz lastflush; TimestampTz now; @@ -3055,10 +3051,10 @@ XLogBackgroundFlush(void) return false; /* read LogwrtResult and update local state */ - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - WriteRqst = XLogCtl->LogwrtRqst; - SpinLockRelease(&XLogCtl->info_lck); + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); + WriteRqst.Write = pg_atomic_read_u64(&XLogCtl->LogwrtRqst.Write); + WriteRqst.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtRqst.Flush); /* back off to last completed page boundary */ WriteRqst.Write -= WriteRqst.Write % XLOG_BLCKSZ; @@ -3140,7 +3136,8 @@ XLogBackgroundFlush(void) /* now wait for any in-progress insertions to finish and get write lock */ WaitXLogInsertionsToFinish(WriteRqst.Write); LWLockAcquire(WALWriteLock, LW_EXCLUSIVE); - LogwrtResult = XLogCtl->LogwrtResult; + 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) { @@ -3228,9 +3225,8 @@ XLogNeedsFlush(XLogRecPtr record) return false; /* read LogwrtResult and update local state */ - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); /* check again */ if (record <= LogwrtResult.Flush) @@ -7779,10 +7775,11 @@ StartupXLOG(void) LogwrtResult.Write = LogwrtResult.Flush = EndOfLog; - XLogCtl->LogwrtResult = LogwrtResult; + pg_atomic_write_u64(&XLogCtl->LogwrtResult.Write, LogwrtResult.Write); + pg_atomic_write_u64(&XLogCtl->LogwrtResult.Flush, LogwrtResult.Flush); - XLogCtl->LogwrtRqst.Write = EndOfLog; - XLogCtl->LogwrtRqst.Flush = EndOfLog; + pg_atomic_write_u64(&XLogCtl->LogwrtRqst.Write, EndOfLog); + pg_atomic_write_u64(&XLogCtl->LogwrtRqst.Flush, EndOfLog); /* * Update full_page_writes in shared memory and write an XLOG_FPW_CHANGE @@ -8484,9 +8481,7 @@ GetInsertRecPtr(void) { XLogRecPtr recptr; - SpinLockAcquire(&XLogCtl->info_lck); - recptr = XLogCtl->LogwrtRqst.Write; - SpinLockRelease(&XLogCtl->info_lck); + recptr = pg_atomic_read_u64(&XLogCtl->LogwrtRqst.Write); return recptr; } @@ -8498,9 +8493,8 @@ GetInsertRecPtr(void) XLogRecPtr GetFlushRecPtr(void) { - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); return LogwrtResult.Flush; } @@ -11649,9 +11643,8 @@ GetXLogInsertRecPtr(void) XLogRecPtr GetXLogWriteRecPtr(void) { - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); return LogwrtResult.Write; } -- 2.20.1