public inbox for [email protected]help / color / mirror / Atom feed
[PATCH 2/2] make LogwrtResult atomic 25+ messages / 3 participants [nested] [flat]
* [PATCH 2/2] make LogwrtResult atomic @ 2021-02-02 17:03 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 25+ messages in thread From: Alvaro Herrera @ 2021-02-02 17:03 UTC (permalink / raw) --- src/backend/access/transam/xlog.c | 66 +++++++++++++++---------------- 1 file changed, 32 insertions(+), 34 deletions(-) diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index f03bd473e2..10802bd56f 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -405,12 +405,9 @@ static XLogRecPtr RedoStartLSN = InvalidXLogRecPtr; * These structs are identical but are declared separately to indicate their * slightly different functions. * - * To read XLogCtl->LogwrtResult, you must hold either info_lck or - * WALWriteLock. To update it, you need to hold both locks. The point of - * this arrangement is that the value can be examined by code that already - * holds WALWriteLock without needing to grab info_lck as well. In addition - * to the shared variable, each backend has a private copy of LogwrtResult, - * which is updated when convenient. + * XLogCtl->LogwrtResult is read and written 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. * * 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. @@ -433,6 +430,12 @@ static XLogRecPtr RedoStartLSN = InvalidXLogRecPtr; *---------- */ +typedef struct XLogwrtAtomic +{ + pg_atomic_uint64 Write; /* last byte + 1 of write position */ + pg_atomic_uint64 Flush; /* last byte + 1 of flush position */ +} XLogwrtAtomic; + typedef struct XLogwrtRqst { XLogRecPtr Write; /* last byte + 1 to write out */ @@ -596,6 +599,7 @@ typedef struct XLogCtlData { XLogCtlInsert Insert; + XLogwrtAtomic LogwrtResult; /* Protected by info_lck: */ XLogwrtRqst LogwrtRqst; XLogRecPtr RedoRecPtr; /* a recent copy of Insert->RedoRecPtr */ @@ -613,12 +617,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). * @@ -1172,9 +1170,10 @@ XLogInsertRecord(XLogRecData *rdata, /* 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); + /* update local result copy */ + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); } /* @@ -2170,13 +2169,13 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic) SpinLockAcquire(&XLogCtl->info_lck); if (XLogCtl->LogwrtRqst.Write < OldPageRqstPtr) XLogCtl->LogwrtRqst.Write = OldPageRqstPtr; - LogwrtResult = XLogCtl->LogwrtResult; SpinLockRelease(&XLogCtl->info_lck); /* * Now that we have an up-to-date LogwrtResult value, 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) { /* @@ -2191,7 +2190,7 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic) LWLockAcquire(WALWriteLock, LW_EXCLUSIVE); - LogwrtResult = XLogCtl->LogwrtResult; + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); if (LogwrtResult.Write >= OldPageRqstPtr) { /* OK, someone wrote it already */ @@ -2438,7 +2437,7 @@ 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); /* * Since successive pages in the xlog cache are consecutively allocated, @@ -2636,7 +2635,6 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) */ if (LogwrtResult.Flush < WriteRqst.Flush && LogwrtResult.Flush < LogwrtResult.Write) - { /* * Could get here without iterating above loop, in which case we might @@ -2675,8 +2673,10 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) * code in a couple of places. */ { + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Write, LogwrtResult.Write); + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Flush, LogwrtResult.Flush); + pg_memory_barrier(); SpinLockAcquire(&XLogCtl->info_lck); - XLogCtl->LogwrtResult = LogwrtResult; if (XLogCtl->LogwrtRqst.Write < LogwrtResult.Write) XLogCtl->LogwrtRqst.Write = LogwrtResult.Write; if (XLogCtl->LogwrtRqst.Flush < LogwrtResult.Flush) @@ -2696,8 +2696,8 @@ XLogSetAsyncXactLSN(XLogRecPtr asyncXactLSN) XLogRecPtr WriteRqstPtr = asyncXactLSN; bool sleeping; + 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; @@ -2908,10 +2908,10 @@ XLogFlush(XLogRecPtr record) SpinLockAcquire(&XLogCtl->info_lck); if (WriteRqstPtr < XLogCtl->LogwrtRqst.Write) WriteRqstPtr = XLogCtl->LogwrtRqst.Write; - LogwrtResult = XLogCtl->LogwrtResult; SpinLockRelease(&XLogCtl->info_lck); /* done already? */ + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); if (record <= LogwrtResult.Flush) break; @@ -2939,7 +2939,7 @@ XLogFlush(XLogRecPtr record) } /* Got the lock; recheck whether request is satisfied */ - LogwrtResult = XLogCtl->LogwrtResult; + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); if (record <= LogwrtResult.Flush) { LWLockRelease(WALWriteLock); @@ -3056,7 +3056,6 @@ XLogBackgroundFlush(void) /* read LogwrtResult and update local state */ SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; WriteRqst = XLogCtl->LogwrtRqst; SpinLockRelease(&XLogCtl->info_lck); @@ -3064,8 +3063,10 @@ 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) { + pg_memory_barrier(); SpinLockAcquire(&XLogCtl->info_lck); WriteRqst.Write = XLogCtl->asyncXactLSN; SpinLockRelease(&XLogCtl->info_lck); @@ -3081,6 +3082,7 @@ XLogBackgroundFlush(void) { if (openLogFile >= 0) { + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); if (!XLByteInPrevSeg(LogwrtResult.Write, openLogSegNo, wal_segment_size)) { @@ -3140,7 +3142,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 +3231,7 @@ XLogNeedsFlush(XLogRecPtr record) return false; /* read LogwrtResult and update local state */ - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); /* check again */ if (record <= LogwrtResult.Flush) @@ -7779,7 +7780,8 @@ StartupXLOG(void) LogwrtResult.Write = LogwrtResult.Flush = EndOfLog; - XLogCtl->LogwrtResult = LogwrtResult; + pg_atomic_write_u64(&XLogCtl->LogwrtResult.Write, EndOfLog); + pg_atomic_write_u64(&XLogCtl->LogwrtResult.Flush, EndOfLog); XLogCtl->LogwrtRqst.Write = EndOfLog; XLogCtl->LogwrtRqst.Flush = EndOfLog; @@ -8498,9 +8500,7 @@ GetInsertRecPtr(void) XLogRecPtr GetFlushRecPtr(void) { - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); return LogwrtResult.Flush; } @@ -11649,9 +11649,7 @@ 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); return LogwrtResult.Write; } -- 2.20.1 --0OAP2g/MAC+5xKAE-- ^ permalink raw reply [nested|flat] 25+ messages in thread
* [PATCH 2/2] make LogwrtResult atomic @ 2021-02-02 17:03 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 25+ messages in thread From: Alvaro Herrera @ 2021-02-02 17:03 UTC (permalink / raw) --- src/backend/access/transam/xlog.c | 66 +++++++++++++++---------------- 1 file changed, 32 insertions(+), 34 deletions(-) diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index f03bd473e2..10802bd56f 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -405,12 +405,9 @@ static XLogRecPtr RedoStartLSN = InvalidXLogRecPtr; * These structs are identical but are declared separately to indicate their * slightly different functions. * - * To read XLogCtl->LogwrtResult, you must hold either info_lck or - * WALWriteLock. To update it, you need to hold both locks. The point of - * this arrangement is that the value can be examined by code that already - * holds WALWriteLock without needing to grab info_lck as well. In addition - * to the shared variable, each backend has a private copy of LogwrtResult, - * which is updated when convenient. + * XLogCtl->LogwrtResult is read and written 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. * * 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. @@ -433,6 +430,12 @@ static XLogRecPtr RedoStartLSN = InvalidXLogRecPtr; *---------- */ +typedef struct XLogwrtAtomic +{ + pg_atomic_uint64 Write; /* last byte + 1 of write position */ + pg_atomic_uint64 Flush; /* last byte + 1 of flush position */ +} XLogwrtAtomic; + typedef struct XLogwrtRqst { XLogRecPtr Write; /* last byte + 1 to write out */ @@ -596,6 +599,7 @@ typedef struct XLogCtlData { XLogCtlInsert Insert; + XLogwrtAtomic LogwrtResult; /* Protected by info_lck: */ XLogwrtRqst LogwrtRqst; XLogRecPtr RedoRecPtr; /* a recent copy of Insert->RedoRecPtr */ @@ -613,12 +617,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). * @@ -1172,9 +1170,10 @@ XLogInsertRecord(XLogRecData *rdata, /* 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); + /* update local result copy */ + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); } /* @@ -2170,13 +2169,13 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic) SpinLockAcquire(&XLogCtl->info_lck); if (XLogCtl->LogwrtRqst.Write < OldPageRqstPtr) XLogCtl->LogwrtRqst.Write = OldPageRqstPtr; - LogwrtResult = XLogCtl->LogwrtResult; SpinLockRelease(&XLogCtl->info_lck); /* * Now that we have an up-to-date LogwrtResult value, 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) { /* @@ -2191,7 +2190,7 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic) LWLockAcquire(WALWriteLock, LW_EXCLUSIVE); - LogwrtResult = XLogCtl->LogwrtResult; + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); if (LogwrtResult.Write >= OldPageRqstPtr) { /* OK, someone wrote it already */ @@ -2438,7 +2437,7 @@ 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); /* * Since successive pages in the xlog cache are consecutively allocated, @@ -2636,7 +2635,6 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) */ if (LogwrtResult.Flush < WriteRqst.Flush && LogwrtResult.Flush < LogwrtResult.Write) - { /* * Could get here without iterating above loop, in which case we might @@ -2675,8 +2673,10 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) * code in a couple of places. */ { + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Write, LogwrtResult.Write); + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Flush, LogwrtResult.Flush); + pg_memory_barrier(); SpinLockAcquire(&XLogCtl->info_lck); - XLogCtl->LogwrtResult = LogwrtResult; if (XLogCtl->LogwrtRqst.Write < LogwrtResult.Write) XLogCtl->LogwrtRqst.Write = LogwrtResult.Write; if (XLogCtl->LogwrtRqst.Flush < LogwrtResult.Flush) @@ -2696,8 +2696,8 @@ XLogSetAsyncXactLSN(XLogRecPtr asyncXactLSN) XLogRecPtr WriteRqstPtr = asyncXactLSN; bool sleeping; + 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; @@ -2908,10 +2908,10 @@ XLogFlush(XLogRecPtr record) SpinLockAcquire(&XLogCtl->info_lck); if (WriteRqstPtr < XLogCtl->LogwrtRqst.Write) WriteRqstPtr = XLogCtl->LogwrtRqst.Write; - LogwrtResult = XLogCtl->LogwrtResult; SpinLockRelease(&XLogCtl->info_lck); /* done already? */ + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); if (record <= LogwrtResult.Flush) break; @@ -2939,7 +2939,7 @@ XLogFlush(XLogRecPtr record) } /* Got the lock; recheck whether request is satisfied */ - LogwrtResult = XLogCtl->LogwrtResult; + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); if (record <= LogwrtResult.Flush) { LWLockRelease(WALWriteLock); @@ -3056,7 +3056,6 @@ XLogBackgroundFlush(void) /* read LogwrtResult and update local state */ SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; WriteRqst = XLogCtl->LogwrtRqst; SpinLockRelease(&XLogCtl->info_lck); @@ -3064,8 +3063,10 @@ 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) { + pg_memory_barrier(); SpinLockAcquire(&XLogCtl->info_lck); WriteRqst.Write = XLogCtl->asyncXactLSN; SpinLockRelease(&XLogCtl->info_lck); @@ -3081,6 +3082,7 @@ XLogBackgroundFlush(void) { if (openLogFile >= 0) { + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); if (!XLByteInPrevSeg(LogwrtResult.Write, openLogSegNo, wal_segment_size)) { @@ -3140,7 +3142,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 +3231,7 @@ XLogNeedsFlush(XLogRecPtr record) return false; /* read LogwrtResult and update local state */ - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); /* check again */ if (record <= LogwrtResult.Flush) @@ -7779,7 +7780,8 @@ StartupXLOG(void) LogwrtResult.Write = LogwrtResult.Flush = EndOfLog; - XLogCtl->LogwrtResult = LogwrtResult; + pg_atomic_write_u64(&XLogCtl->LogwrtResult.Write, EndOfLog); + pg_atomic_write_u64(&XLogCtl->LogwrtResult.Flush, EndOfLog); XLogCtl->LogwrtRqst.Write = EndOfLog; XLogCtl->LogwrtRqst.Flush = EndOfLog; @@ -8498,9 +8500,7 @@ GetInsertRecPtr(void) XLogRecPtr GetFlushRecPtr(void) { - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); return LogwrtResult.Flush; } @@ -11649,9 +11649,7 @@ 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); return LogwrtResult.Write; } -- 2.20.1 --0OAP2g/MAC+5xKAE-- ^ permalink raw reply [nested|flat] 25+ messages in thread
* [PATCH 2/2] make LogwrtResult atomic @ 2021-02-02 17:03 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 25+ messages in thread From: Alvaro Herrera @ 2021-02-02 17:03 UTC (permalink / raw) --- src/backend/access/transam/xlog.c | 66 +++++++++++++++---------------- 1 file changed, 32 insertions(+), 34 deletions(-) diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index f03bd473e2..10802bd56f 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -405,12 +405,9 @@ static XLogRecPtr RedoStartLSN = InvalidXLogRecPtr; * These structs are identical but are declared separately to indicate their * slightly different functions. * - * To read XLogCtl->LogwrtResult, you must hold either info_lck or - * WALWriteLock. To update it, you need to hold both locks. The point of - * this arrangement is that the value can be examined by code that already - * holds WALWriteLock without needing to grab info_lck as well. In addition - * to the shared variable, each backend has a private copy of LogwrtResult, - * which is updated when convenient. + * XLogCtl->LogwrtResult is read and written 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. * * 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. @@ -433,6 +430,12 @@ static XLogRecPtr RedoStartLSN = InvalidXLogRecPtr; *---------- */ +typedef struct XLogwrtAtomic +{ + pg_atomic_uint64 Write; /* last byte + 1 of write position */ + pg_atomic_uint64 Flush; /* last byte + 1 of flush position */ +} XLogwrtAtomic; + typedef struct XLogwrtRqst { XLogRecPtr Write; /* last byte + 1 to write out */ @@ -596,6 +599,7 @@ typedef struct XLogCtlData { XLogCtlInsert Insert; + XLogwrtAtomic LogwrtResult; /* Protected by info_lck: */ XLogwrtRqst LogwrtRqst; XLogRecPtr RedoRecPtr; /* a recent copy of Insert->RedoRecPtr */ @@ -613,12 +617,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). * @@ -1172,9 +1170,10 @@ XLogInsertRecord(XLogRecData *rdata, /* 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); + /* update local result copy */ + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); } /* @@ -2170,13 +2169,13 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic) SpinLockAcquire(&XLogCtl->info_lck); if (XLogCtl->LogwrtRqst.Write < OldPageRqstPtr) XLogCtl->LogwrtRqst.Write = OldPageRqstPtr; - LogwrtResult = XLogCtl->LogwrtResult; SpinLockRelease(&XLogCtl->info_lck); /* * Now that we have an up-to-date LogwrtResult value, 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) { /* @@ -2191,7 +2190,7 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic) LWLockAcquire(WALWriteLock, LW_EXCLUSIVE); - LogwrtResult = XLogCtl->LogwrtResult; + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); if (LogwrtResult.Write >= OldPageRqstPtr) { /* OK, someone wrote it already */ @@ -2438,7 +2437,7 @@ 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); /* * Since successive pages in the xlog cache are consecutively allocated, @@ -2636,7 +2635,6 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) */ if (LogwrtResult.Flush < WriteRqst.Flush && LogwrtResult.Flush < LogwrtResult.Write) - { /* * Could get here without iterating above loop, in which case we might @@ -2675,8 +2673,10 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) * code in a couple of places. */ { + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Write, LogwrtResult.Write); + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Flush, LogwrtResult.Flush); + pg_memory_barrier(); SpinLockAcquire(&XLogCtl->info_lck); - XLogCtl->LogwrtResult = LogwrtResult; if (XLogCtl->LogwrtRqst.Write < LogwrtResult.Write) XLogCtl->LogwrtRqst.Write = LogwrtResult.Write; if (XLogCtl->LogwrtRqst.Flush < LogwrtResult.Flush) @@ -2696,8 +2696,8 @@ XLogSetAsyncXactLSN(XLogRecPtr asyncXactLSN) XLogRecPtr WriteRqstPtr = asyncXactLSN; bool sleeping; + 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; @@ -2908,10 +2908,10 @@ XLogFlush(XLogRecPtr record) SpinLockAcquire(&XLogCtl->info_lck); if (WriteRqstPtr < XLogCtl->LogwrtRqst.Write) WriteRqstPtr = XLogCtl->LogwrtRqst.Write; - LogwrtResult = XLogCtl->LogwrtResult; SpinLockRelease(&XLogCtl->info_lck); /* done already? */ + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); if (record <= LogwrtResult.Flush) break; @@ -2939,7 +2939,7 @@ XLogFlush(XLogRecPtr record) } /* Got the lock; recheck whether request is satisfied */ - LogwrtResult = XLogCtl->LogwrtResult; + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); if (record <= LogwrtResult.Flush) { LWLockRelease(WALWriteLock); @@ -3056,7 +3056,6 @@ XLogBackgroundFlush(void) /* read LogwrtResult and update local state */ SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; WriteRqst = XLogCtl->LogwrtRqst; SpinLockRelease(&XLogCtl->info_lck); @@ -3064,8 +3063,10 @@ 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) { + pg_memory_barrier(); SpinLockAcquire(&XLogCtl->info_lck); WriteRqst.Write = XLogCtl->asyncXactLSN; SpinLockRelease(&XLogCtl->info_lck); @@ -3081,6 +3082,7 @@ XLogBackgroundFlush(void) { if (openLogFile >= 0) { + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); if (!XLByteInPrevSeg(LogwrtResult.Write, openLogSegNo, wal_segment_size)) { @@ -3140,7 +3142,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 +3231,7 @@ XLogNeedsFlush(XLogRecPtr record) return false; /* read LogwrtResult and update local state */ - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); /* check again */ if (record <= LogwrtResult.Flush) @@ -7779,7 +7780,8 @@ StartupXLOG(void) LogwrtResult.Write = LogwrtResult.Flush = EndOfLog; - XLogCtl->LogwrtResult = LogwrtResult; + pg_atomic_write_u64(&XLogCtl->LogwrtResult.Write, EndOfLog); + pg_atomic_write_u64(&XLogCtl->LogwrtResult.Flush, EndOfLog); XLogCtl->LogwrtRqst.Write = EndOfLog; XLogCtl->LogwrtRqst.Flush = EndOfLog; @@ -8498,9 +8500,7 @@ GetInsertRecPtr(void) XLogRecPtr GetFlushRecPtr(void) { - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); return LogwrtResult.Flush; } @@ -11649,9 +11649,7 @@ 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); return LogwrtResult.Write; } -- 2.20.1 --0OAP2g/MAC+5xKAE-- ^ permalink raw reply [nested|flat] 25+ messages in thread
* [PATCH v5 1/2] make LogwrtResult atomic @ 2021-02-02 17:03 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 25+ messages in thread From: Alvaro Herrera @ 2021-02-02 17:03 UTC (permalink / raw) --- src/backend/access/transam/xlog.c | 192 +++++++++++++++--------------- src/include/port/atomics.h | 29 +++++ src/tools/pgindent/typedefs.list | 1 + 3 files changed, 123 insertions(+), 99 deletions(-) diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index 221e4cb34f..225ac8bc2d 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -382,16 +382,14 @@ static XLogRecPtr RedoStartLSN = InvalidXLogRecPtr; * * LogwrtRqst indicates a byte position that we need to write and/or fsync * the log up to (all records before that point must be written or fsynced). - * LogwrtResult indicates the byte positions we have already written/fsynced. + * LogWriteResult, LogFlushResult indicate the byte positions we have already + * written/fsynced. * These structs are identical but are declared separately to indicate their * slightly different functions. * - * To read XLogCtl->LogwrtResult, you must hold either info_lck or - * WALWriteLock. To update it, you need to hold both locks. The point of - * this arrangement is that the value can be examined by code that already - * holds WALWriteLock without needing to grab info_lck as well. In addition - * to the shared variable, each backend has a private copy of LogwrtResult, - * which is updated when convenient. + * XLogCtl->LogwrtResult is read and written 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. * * 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. @@ -414,18 +412,18 @@ static XLogRecPtr RedoStartLSN = InvalidXLogRecPtr; *---------- */ +typedef struct XLogwrtAtomic +{ + pg_atomic_uint64 Write; /* last byte + 1 of write position */ + pg_atomic_uint64 Flush; /* last byte + 1 of flush position */ +} XLogwrtAtomic; + typedef struct XLogwrtRqst { XLogRecPtr Write; /* last byte + 1 to write out */ 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 @@ -577,6 +575,7 @@ typedef struct XLogCtlData { XLogCtlInsert Insert; + XLogwrtAtomic LogwrtResult; /* uses atomics */ /* Protected by info_lck: */ XLogwrtRqst LogwrtRqst; XLogRecPtr RedoRecPtr; /* a recent copy of Insert->RedoRecPtr */ @@ -594,12 +593,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). * @@ -770,10 +763,11 @@ static ControlFileData *ControlFile = NULL; static int UsableBytesInSegment; /* - * Private, possibly out-of-date copy of shared LogwrtResult. + * 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; /* * Codes indicating where we got a WAL file from during recovery, or where @@ -1201,8 +1195,6 @@ XLogInsertRecord(XLogRecData *rdata, /* 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); } @@ -2174,6 +2166,7 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, TimeLineID tli, bool opportunistic) * Now that we have the lock, check if someone initialized the page * already. */ + LogWriteResult = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); while (upto >= XLogCtl->InitializedUpTo || opportunistic) { nextidx = XLogRecPtrToBufIdx(XLogCtl->InitializedUpTo); @@ -2184,7 +2177,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 @@ -2193,18 +2186,18 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, TimeLineID tli, bool opportunistic) if (opportunistic) break; - /* Before waiting, get info_lck and update LogwrtResult */ + /* Advance shared memory write request position */ SpinLockAcquire(&XLogCtl->info_lck); if (XLogCtl->LogwrtRqst.Write < OldPageRqstPtr) XLogCtl->LogwrtRqst.Write = OldPageRqstPtr; - LogwrtResult = XLogCtl->LogwrtResult; SpinLockRelease(&XLogCtl->info_lck); /* - * Now that we have an up-to-date LogwrtResult value, see if we - * still need to write it or if someone else already did. + * Before waiting, update LogWriteResult and see if we still need + * to write it or if someone else already did. */ - if (LogwrtResult.Write < OldPageRqstPtr) + LogWriteResult = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); + if (LogWriteResult < OldPageRqstPtr) { /* * Must acquire write lock. Release WALBufMappingLock first, @@ -2218,8 +2211,8 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, TimeLineID tli, bool opportunistic) LWLockAcquire(WALWriteLock, LW_EXCLUSIVE); - LogwrtResult = XLogCtl->LogwrtResult; - if (LogwrtResult.Write >= OldPageRqstPtr) + LogWriteResult = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); + if (LogWriteResult >= OldPageRqstPtr) { /* OK, someone wrote it already */ LWLockRelease(WALWriteLock); @@ -2476,7 +2469,7 @@ XLogWrite(XLogwrtRqst WriteRqst, TimeLineID tli, bool flexible) /* * Update local LogwrtResult (caller probably did this already, but...) */ - LogwrtResult = XLogCtl->LogwrtResult; + LogWriteResult = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); /* * Since successive pages in the xlog cache are consecutively allocated, @@ -2496,9 +2489,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 @@ -2507,16 +2500,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)) { /* @@ -2526,7 +2519,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; @@ -2538,7 +2531,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); @@ -2550,7 +2543,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++; @@ -2561,7 +2554,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; @@ -2652,13 +2645,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 @@ -2679,7 +2672,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); @@ -2691,12 +2684,14 @@ XLogWrite(XLogwrtRqst WriteRqst, TimeLineID tli, bool flexible) Assert(npages == 0); + /* Publish current write result position */ + 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 @@ -2707,12 +2702,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); @@ -2725,23 +2720,23 @@ 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, LogFlushResult); + /* - * Update shared-memory status - * - * We make sure that the shared 'request' values do not fall behind the + * Make sure that the shared 'request' values do not fall behind the * 'result' values. This is not absolutely essential, but it saves some * 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; + if (XLogCtl->LogwrtRqst.Write < LogWriteResult) + XLogCtl->LogwrtRqst.Write = LogWriteResult; + if (XLogCtl->LogwrtRqst.Flush < LogFlushResult) + XLogCtl->LogwrtRqst.Flush = LogFlushResult; SpinLockRelease(&XLogCtl->info_lck); } } @@ -2757,8 +2752,8 @@ XLogSetAsyncXactLSN(XLogRecPtr asyncXactLSN) XLogRecPtr WriteRqstPtr = asyncXactLSN; bool sleeping; + LogFlushResult = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; sleeping = XLogCtl->WalWriterSleeping; if (XLogCtl->asyncXactLSN < asyncXactLSN) XLogCtl->asyncXactLSN = asyncXactLSN; @@ -2775,7 +2770,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; } @@ -2931,15 +2926,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(); @@ -2948,8 +2943,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 */ @@ -2967,11 +2962,11 @@ XLogFlush(XLogRecPtr record) SpinLockAcquire(&XLogCtl->info_lck); if (WriteRqstPtr < XLogCtl->LogwrtRqst.Write) WriteRqstPtr = XLogCtl->LogwrtRqst.Write; - LogwrtResult = XLogCtl->LogwrtResult; SpinLockRelease(&XLogCtl->info_lck); /* done already? */ - if (record <= LogwrtResult.Flush) + LogFlushResult = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); + if (record <= LogFlushResult) break; /* @@ -2998,8 +2993,8 @@ XLogFlush(XLogRecPtr record) } /* Got the lock; recheck whether request is satisfied */ - LogwrtResult = XLogCtl->LogwrtResult; - if (record <= LogwrtResult.Flush) + LogFlushResult = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); + if (record <= LogFlushResult) { LWLockRelease(WALWriteLock); break; @@ -3069,11 +3064,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)); } /* @@ -3122,7 +3117,6 @@ XLogBackgroundFlush(void) /* read LogwrtResult and update local state */ SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; WriteRqst = XLogCtl->LogwrtRqst; SpinLockRelease(&XLogCtl->info_lck); @@ -3130,8 +3124,10 @@ XLogBackgroundFlush(void) WriteRqst.Write -= WriteRqst.Write % XLOG_BLCKSZ; /* if we have already flushed that far, consider async commit records */ - if (WriteRqst.Write <= LogwrtResult.Flush) + LogFlushResult = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); + if (WriteRqst.Write <= LogFlushResult) { + pg_memory_barrier(); SpinLockAcquire(&XLogCtl->info_lck); WriteRqst.Write = XLogCtl->asyncXactLSN; SpinLockRelease(&XLogCtl->info_lck); @@ -3143,11 +3139,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) { - if (!XLByteInPrevSeg(LogwrtResult.Write, openLogSegNo, + LogWriteResult = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); + if (!XLByteInPrevSeg(LogWriteResult, openLogSegNo, wal_segment_size)) { XLogFileClose(); @@ -3162,7 +3159,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) { @@ -3197,8 +3194,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(); @@ -3206,9 +3203,10 @@ 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; - 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); } @@ -3290,16 +3288,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 */ - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + LogFlushResult = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); /* check again */ - if (record <= LogwrtResult.Flush) + if (record <= LogFlushResult) return false; return true; @@ -8086,9 +8082,11 @@ StartupXLOG(void) XLogCtl->InitializedUpTo = EndOfLog; } - LogwrtResult.Write = LogwrtResult.Flush = EndOfLog; + LogWriteResult = LogFlushResult = EndOfLog; - XLogCtl->LogwrtResult = LogwrtResult; + /* XXX OK to write without WALWriteLock? */ + pg_atomic_write_u64(&XLogCtl->LogwrtResult.Write, EndOfLog); + pg_atomic_write_u64(&XLogCtl->LogwrtResult.Flush, EndOfLog); XLogCtl->LogwrtRqst.Write = EndOfLog; XLogCtl->LogwrtRqst.Flush = EndOfLog; @@ -8750,9 +8748,7 @@ GetFlushRecPtr(TimeLineID *insertTLI) { Assert(XLogCtl->SharedRecoveryState == RECOVERY_STATE_DONE); - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + LogFlushResult = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); /* * If we're writing and flushing WAL, the time line can't be changing, @@ -8761,7 +8757,7 @@ GetFlushRecPtr(TimeLineID *insertTLI) if (insertTLI) *insertTLI = XLogCtl->InsertTimeLineID; - return LogwrtResult.Flush; + return LogFlushResult; } /* @@ -12041,11 +12037,9 @@ GetXLogInsertRecPtr(void) XLogRecPtr GetXLogWriteRecPtr(void) { - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + LogWriteResult = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); - return LogwrtResult.Write; + return LogWriteResult; } /* diff --git a/src/include/port/atomics.h b/src/include/port/atomics.h index 856338f161..4ff9ce9864 100644 --- a/src/include/port/atomics.h +++ b/src/include/port/atomics.h @@ -519,6 +519,35 @@ pg_atomic_sub_fetch_u64(volatile pg_atomic_uint64 *ptr, int64 sub_) return pg_atomic_sub_fetch_u64_impl(ptr, sub_); } +/* + * Monotonically advance the given variable using only atomic operations until + * it's at least the target value. + * + * Full barrier semantics (even when value is unchanged). + */ +static inline void +pg_atomic_monotonic_advance_u64(volatile pg_atomic_uint64 *ptr, uint64 target_) +{ + uint64 currval; + +#ifndef PG_HAVE_ATOMIC_U64_SIMULATION + AssertPointerAlignment(ptr, 8); +#endif + + currval = pg_atomic_read_u64(ptr); + if (currval >= target_) + { + pg_memory_barrier(); + return; + } + + while (currval < target_) + { + if (pg_atomic_compare_exchange_u64(ptr, &currval, target_)) + break; + } +} + #undef INSIDE_ATOMICS_H #endif /* ATOMICS_H */ diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index da6ac8ed83..edc9bfa5ce 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -2943,6 +2943,7 @@ XLogRecordBuffer XLogRedoAction XLogSegNo XLogSource +XLogwrtAtomic XLogwrtResult XLogwrtRqst XPVIV -- 2.30.2 --fl2voahem2nkfart Content-Type: text/x-diff; charset=utf-8 Content-Disposition: attachment; filename="v5-0002-more-atomics.patch" ^ permalink raw reply [nested|flat] 25+ messages in thread
* [PATCH 2/2] make LogwrtResult atomic @ 2021-02-02 17:03 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 25+ messages in thread From: Alvaro Herrera @ 2021-02-02 17:03 UTC (permalink / raw) --- src/backend/access/transam/xlog.c | 66 +++++++++++++++---------------- 1 file changed, 32 insertions(+), 34 deletions(-) diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index f03bd473e2..10802bd56f 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -405,12 +405,9 @@ static XLogRecPtr RedoStartLSN = InvalidXLogRecPtr; * These structs are identical but are declared separately to indicate their * slightly different functions. * - * To read XLogCtl->LogwrtResult, you must hold either info_lck or - * WALWriteLock. To update it, you need to hold both locks. The point of - * this arrangement is that the value can be examined by code that already - * holds WALWriteLock without needing to grab info_lck as well. In addition - * to the shared variable, each backend has a private copy of LogwrtResult, - * which is updated when convenient. + * XLogCtl->LogwrtResult is read and written 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. * * 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. @@ -433,6 +430,12 @@ static XLogRecPtr RedoStartLSN = InvalidXLogRecPtr; *---------- */ +typedef struct XLogwrtAtomic +{ + pg_atomic_uint64 Write; /* last byte + 1 of write position */ + pg_atomic_uint64 Flush; /* last byte + 1 of flush position */ +} XLogwrtAtomic; + typedef struct XLogwrtRqst { XLogRecPtr Write; /* last byte + 1 to write out */ @@ -596,6 +599,7 @@ typedef struct XLogCtlData { XLogCtlInsert Insert; + XLogwrtAtomic LogwrtResult; /* Protected by info_lck: */ XLogwrtRqst LogwrtRqst; XLogRecPtr RedoRecPtr; /* a recent copy of Insert->RedoRecPtr */ @@ -613,12 +617,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). * @@ -1172,9 +1170,10 @@ XLogInsertRecord(XLogRecData *rdata, /* 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); + /* update local result copy */ + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); } /* @@ -2170,13 +2169,13 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic) SpinLockAcquire(&XLogCtl->info_lck); if (XLogCtl->LogwrtRqst.Write < OldPageRqstPtr) XLogCtl->LogwrtRqst.Write = OldPageRqstPtr; - LogwrtResult = XLogCtl->LogwrtResult; SpinLockRelease(&XLogCtl->info_lck); /* * Now that we have an up-to-date LogwrtResult value, 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) { /* @@ -2191,7 +2190,7 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic) LWLockAcquire(WALWriteLock, LW_EXCLUSIVE); - LogwrtResult = XLogCtl->LogwrtResult; + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); if (LogwrtResult.Write >= OldPageRqstPtr) { /* OK, someone wrote it already */ @@ -2438,7 +2437,7 @@ 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); /* * Since successive pages in the xlog cache are consecutively allocated, @@ -2636,7 +2635,6 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) */ if (LogwrtResult.Flush < WriteRqst.Flush && LogwrtResult.Flush < LogwrtResult.Write) - { /* * Could get here without iterating above loop, in which case we might @@ -2675,8 +2673,10 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) * code in a couple of places. */ { + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Write, LogwrtResult.Write); + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Flush, LogwrtResult.Flush); + pg_memory_barrier(); SpinLockAcquire(&XLogCtl->info_lck); - XLogCtl->LogwrtResult = LogwrtResult; if (XLogCtl->LogwrtRqst.Write < LogwrtResult.Write) XLogCtl->LogwrtRqst.Write = LogwrtResult.Write; if (XLogCtl->LogwrtRqst.Flush < LogwrtResult.Flush) @@ -2696,8 +2696,8 @@ XLogSetAsyncXactLSN(XLogRecPtr asyncXactLSN) XLogRecPtr WriteRqstPtr = asyncXactLSN; bool sleeping; + 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; @@ -2908,10 +2908,10 @@ XLogFlush(XLogRecPtr record) SpinLockAcquire(&XLogCtl->info_lck); if (WriteRqstPtr < XLogCtl->LogwrtRqst.Write) WriteRqstPtr = XLogCtl->LogwrtRqst.Write; - LogwrtResult = XLogCtl->LogwrtResult; SpinLockRelease(&XLogCtl->info_lck); /* done already? */ + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); if (record <= LogwrtResult.Flush) break; @@ -2939,7 +2939,7 @@ XLogFlush(XLogRecPtr record) } /* Got the lock; recheck whether request is satisfied */ - LogwrtResult = XLogCtl->LogwrtResult; + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); if (record <= LogwrtResult.Flush) { LWLockRelease(WALWriteLock); @@ -3056,7 +3056,6 @@ XLogBackgroundFlush(void) /* read LogwrtResult and update local state */ SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; WriteRqst = XLogCtl->LogwrtRqst; SpinLockRelease(&XLogCtl->info_lck); @@ -3064,8 +3063,10 @@ 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) { + pg_memory_barrier(); SpinLockAcquire(&XLogCtl->info_lck); WriteRqst.Write = XLogCtl->asyncXactLSN; SpinLockRelease(&XLogCtl->info_lck); @@ -3081,6 +3082,7 @@ XLogBackgroundFlush(void) { if (openLogFile >= 0) { + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); if (!XLByteInPrevSeg(LogwrtResult.Write, openLogSegNo, wal_segment_size)) { @@ -3140,7 +3142,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 +3231,7 @@ XLogNeedsFlush(XLogRecPtr record) return false; /* read LogwrtResult and update local state */ - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); /* check again */ if (record <= LogwrtResult.Flush) @@ -7779,7 +7780,8 @@ StartupXLOG(void) LogwrtResult.Write = LogwrtResult.Flush = EndOfLog; - XLogCtl->LogwrtResult = LogwrtResult; + pg_atomic_write_u64(&XLogCtl->LogwrtResult.Write, EndOfLog); + pg_atomic_write_u64(&XLogCtl->LogwrtResult.Flush, EndOfLog); XLogCtl->LogwrtRqst.Write = EndOfLog; XLogCtl->LogwrtRqst.Flush = EndOfLog; @@ -8498,9 +8500,7 @@ GetInsertRecPtr(void) XLogRecPtr GetFlushRecPtr(void) { - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); return LogwrtResult.Flush; } @@ -11649,9 +11649,7 @@ 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); return LogwrtResult.Write; } -- 2.20.1 --0OAP2g/MAC+5xKAE-- ^ permalink raw reply [nested|flat] 25+ messages in thread
* [PATCH 2/2] make LogwrtResult atomic @ 2021-02-02 17:03 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 25+ messages in thread From: Alvaro Herrera @ 2021-02-02 17:03 UTC (permalink / raw) --- src/backend/access/transam/xlog.c | 66 +++++++++++++++---------------- 1 file changed, 32 insertions(+), 34 deletions(-) diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index f03bd473e2..10802bd56f 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -405,12 +405,9 @@ static XLogRecPtr RedoStartLSN = InvalidXLogRecPtr; * These structs are identical but are declared separately to indicate their * slightly different functions. * - * To read XLogCtl->LogwrtResult, you must hold either info_lck or - * WALWriteLock. To update it, you need to hold both locks. The point of - * this arrangement is that the value can be examined by code that already - * holds WALWriteLock without needing to grab info_lck as well. In addition - * to the shared variable, each backend has a private copy of LogwrtResult, - * which is updated when convenient. + * XLogCtl->LogwrtResult is read and written 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. * * 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. @@ -433,6 +430,12 @@ static XLogRecPtr RedoStartLSN = InvalidXLogRecPtr; *---------- */ +typedef struct XLogwrtAtomic +{ + pg_atomic_uint64 Write; /* last byte + 1 of write position */ + pg_atomic_uint64 Flush; /* last byte + 1 of flush position */ +} XLogwrtAtomic; + typedef struct XLogwrtRqst { XLogRecPtr Write; /* last byte + 1 to write out */ @@ -596,6 +599,7 @@ typedef struct XLogCtlData { XLogCtlInsert Insert; + XLogwrtAtomic LogwrtResult; /* Protected by info_lck: */ XLogwrtRqst LogwrtRqst; XLogRecPtr RedoRecPtr; /* a recent copy of Insert->RedoRecPtr */ @@ -613,12 +617,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). * @@ -1172,9 +1170,10 @@ XLogInsertRecord(XLogRecData *rdata, /* 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); + /* update local result copy */ + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); } /* @@ -2170,13 +2169,13 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic) SpinLockAcquire(&XLogCtl->info_lck); if (XLogCtl->LogwrtRqst.Write < OldPageRqstPtr) XLogCtl->LogwrtRqst.Write = OldPageRqstPtr; - LogwrtResult = XLogCtl->LogwrtResult; SpinLockRelease(&XLogCtl->info_lck); /* * Now that we have an up-to-date LogwrtResult value, 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) { /* @@ -2191,7 +2190,7 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic) LWLockAcquire(WALWriteLock, LW_EXCLUSIVE); - LogwrtResult = XLogCtl->LogwrtResult; + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); if (LogwrtResult.Write >= OldPageRqstPtr) { /* OK, someone wrote it already */ @@ -2438,7 +2437,7 @@ 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); /* * Since successive pages in the xlog cache are consecutively allocated, @@ -2636,7 +2635,6 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) */ if (LogwrtResult.Flush < WriteRqst.Flush && LogwrtResult.Flush < LogwrtResult.Write) - { /* * Could get here without iterating above loop, in which case we might @@ -2675,8 +2673,10 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) * code in a couple of places. */ { + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Write, LogwrtResult.Write); + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Flush, LogwrtResult.Flush); + pg_memory_barrier(); SpinLockAcquire(&XLogCtl->info_lck); - XLogCtl->LogwrtResult = LogwrtResult; if (XLogCtl->LogwrtRqst.Write < LogwrtResult.Write) XLogCtl->LogwrtRqst.Write = LogwrtResult.Write; if (XLogCtl->LogwrtRqst.Flush < LogwrtResult.Flush) @@ -2696,8 +2696,8 @@ XLogSetAsyncXactLSN(XLogRecPtr asyncXactLSN) XLogRecPtr WriteRqstPtr = asyncXactLSN; bool sleeping; + 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; @@ -2908,10 +2908,10 @@ XLogFlush(XLogRecPtr record) SpinLockAcquire(&XLogCtl->info_lck); if (WriteRqstPtr < XLogCtl->LogwrtRqst.Write) WriteRqstPtr = XLogCtl->LogwrtRqst.Write; - LogwrtResult = XLogCtl->LogwrtResult; SpinLockRelease(&XLogCtl->info_lck); /* done already? */ + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); if (record <= LogwrtResult.Flush) break; @@ -2939,7 +2939,7 @@ XLogFlush(XLogRecPtr record) } /* Got the lock; recheck whether request is satisfied */ - LogwrtResult = XLogCtl->LogwrtResult; + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); if (record <= LogwrtResult.Flush) { LWLockRelease(WALWriteLock); @@ -3056,7 +3056,6 @@ XLogBackgroundFlush(void) /* read LogwrtResult and update local state */ SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; WriteRqst = XLogCtl->LogwrtRqst; SpinLockRelease(&XLogCtl->info_lck); @@ -3064,8 +3063,10 @@ 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) { + pg_memory_barrier(); SpinLockAcquire(&XLogCtl->info_lck); WriteRqst.Write = XLogCtl->asyncXactLSN; SpinLockRelease(&XLogCtl->info_lck); @@ -3081,6 +3082,7 @@ XLogBackgroundFlush(void) { if (openLogFile >= 0) { + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); if (!XLByteInPrevSeg(LogwrtResult.Write, openLogSegNo, wal_segment_size)) { @@ -3140,7 +3142,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 +3231,7 @@ XLogNeedsFlush(XLogRecPtr record) return false; /* read LogwrtResult and update local state */ - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); /* check again */ if (record <= LogwrtResult.Flush) @@ -7779,7 +7780,8 @@ StartupXLOG(void) LogwrtResult.Write = LogwrtResult.Flush = EndOfLog; - XLogCtl->LogwrtResult = LogwrtResult; + pg_atomic_write_u64(&XLogCtl->LogwrtResult.Write, EndOfLog); + pg_atomic_write_u64(&XLogCtl->LogwrtResult.Flush, EndOfLog); XLogCtl->LogwrtRqst.Write = EndOfLog; XLogCtl->LogwrtRqst.Flush = EndOfLog; @@ -8498,9 +8500,7 @@ GetInsertRecPtr(void) XLogRecPtr GetFlushRecPtr(void) { - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); return LogwrtResult.Flush; } @@ -11649,9 +11649,7 @@ 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); return LogwrtResult.Write; } -- 2.20.1 --0OAP2g/MAC+5xKAE-- ^ permalink raw reply [nested|flat] 25+ messages in thread
* [PATCH 2/2] make LogwrtResult atomic @ 2021-02-02 17:03 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 25+ messages in thread From: Alvaro Herrera @ 2021-02-02 17:03 UTC (permalink / raw) --- src/backend/access/transam/xlog.c | 66 +++++++++++++++---------------- 1 file changed, 32 insertions(+), 34 deletions(-) diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index f03bd473e2..10802bd56f 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -405,12 +405,9 @@ static XLogRecPtr RedoStartLSN = InvalidXLogRecPtr; * These structs are identical but are declared separately to indicate their * slightly different functions. * - * To read XLogCtl->LogwrtResult, you must hold either info_lck or - * WALWriteLock. To update it, you need to hold both locks. The point of - * this arrangement is that the value can be examined by code that already - * holds WALWriteLock without needing to grab info_lck as well. In addition - * to the shared variable, each backend has a private copy of LogwrtResult, - * which is updated when convenient. + * XLogCtl->LogwrtResult is read and written 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. * * 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. @@ -433,6 +430,12 @@ static XLogRecPtr RedoStartLSN = InvalidXLogRecPtr; *---------- */ +typedef struct XLogwrtAtomic +{ + pg_atomic_uint64 Write; /* last byte + 1 of write position */ + pg_atomic_uint64 Flush; /* last byte + 1 of flush position */ +} XLogwrtAtomic; + typedef struct XLogwrtRqst { XLogRecPtr Write; /* last byte + 1 to write out */ @@ -596,6 +599,7 @@ typedef struct XLogCtlData { XLogCtlInsert Insert; + XLogwrtAtomic LogwrtResult; /* Protected by info_lck: */ XLogwrtRqst LogwrtRqst; XLogRecPtr RedoRecPtr; /* a recent copy of Insert->RedoRecPtr */ @@ -613,12 +617,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). * @@ -1172,9 +1170,10 @@ XLogInsertRecord(XLogRecData *rdata, /* 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); + /* update local result copy */ + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); } /* @@ -2170,13 +2169,13 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic) SpinLockAcquire(&XLogCtl->info_lck); if (XLogCtl->LogwrtRqst.Write < OldPageRqstPtr) XLogCtl->LogwrtRqst.Write = OldPageRqstPtr; - LogwrtResult = XLogCtl->LogwrtResult; SpinLockRelease(&XLogCtl->info_lck); /* * Now that we have an up-to-date LogwrtResult value, 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) { /* @@ -2191,7 +2190,7 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic) LWLockAcquire(WALWriteLock, LW_EXCLUSIVE); - LogwrtResult = XLogCtl->LogwrtResult; + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); if (LogwrtResult.Write >= OldPageRqstPtr) { /* OK, someone wrote it already */ @@ -2438,7 +2437,7 @@ 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); /* * Since successive pages in the xlog cache are consecutively allocated, @@ -2636,7 +2635,6 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) */ if (LogwrtResult.Flush < WriteRqst.Flush && LogwrtResult.Flush < LogwrtResult.Write) - { /* * Could get here without iterating above loop, in which case we might @@ -2675,8 +2673,10 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) * code in a couple of places. */ { + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Write, LogwrtResult.Write); + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Flush, LogwrtResult.Flush); + pg_memory_barrier(); SpinLockAcquire(&XLogCtl->info_lck); - XLogCtl->LogwrtResult = LogwrtResult; if (XLogCtl->LogwrtRqst.Write < LogwrtResult.Write) XLogCtl->LogwrtRqst.Write = LogwrtResult.Write; if (XLogCtl->LogwrtRqst.Flush < LogwrtResult.Flush) @@ -2696,8 +2696,8 @@ XLogSetAsyncXactLSN(XLogRecPtr asyncXactLSN) XLogRecPtr WriteRqstPtr = asyncXactLSN; bool sleeping; + 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; @@ -2908,10 +2908,10 @@ XLogFlush(XLogRecPtr record) SpinLockAcquire(&XLogCtl->info_lck); if (WriteRqstPtr < XLogCtl->LogwrtRqst.Write) WriteRqstPtr = XLogCtl->LogwrtRqst.Write; - LogwrtResult = XLogCtl->LogwrtResult; SpinLockRelease(&XLogCtl->info_lck); /* done already? */ + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); if (record <= LogwrtResult.Flush) break; @@ -2939,7 +2939,7 @@ XLogFlush(XLogRecPtr record) } /* Got the lock; recheck whether request is satisfied */ - LogwrtResult = XLogCtl->LogwrtResult; + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); if (record <= LogwrtResult.Flush) { LWLockRelease(WALWriteLock); @@ -3056,7 +3056,6 @@ XLogBackgroundFlush(void) /* read LogwrtResult and update local state */ SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; WriteRqst = XLogCtl->LogwrtRqst; SpinLockRelease(&XLogCtl->info_lck); @@ -3064,8 +3063,10 @@ 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) { + pg_memory_barrier(); SpinLockAcquire(&XLogCtl->info_lck); WriteRqst.Write = XLogCtl->asyncXactLSN; SpinLockRelease(&XLogCtl->info_lck); @@ -3081,6 +3082,7 @@ XLogBackgroundFlush(void) { if (openLogFile >= 0) { + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); if (!XLByteInPrevSeg(LogwrtResult.Write, openLogSegNo, wal_segment_size)) { @@ -3140,7 +3142,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 +3231,7 @@ XLogNeedsFlush(XLogRecPtr record) return false; /* read LogwrtResult and update local state */ - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); /* check again */ if (record <= LogwrtResult.Flush) @@ -7779,7 +7780,8 @@ StartupXLOG(void) LogwrtResult.Write = LogwrtResult.Flush = EndOfLog; - XLogCtl->LogwrtResult = LogwrtResult; + pg_atomic_write_u64(&XLogCtl->LogwrtResult.Write, EndOfLog); + pg_atomic_write_u64(&XLogCtl->LogwrtResult.Flush, EndOfLog); XLogCtl->LogwrtRqst.Write = EndOfLog; XLogCtl->LogwrtRqst.Flush = EndOfLog; @@ -8498,9 +8500,7 @@ GetInsertRecPtr(void) XLogRecPtr GetFlushRecPtr(void) { - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); return LogwrtResult.Flush; } @@ -11649,9 +11649,7 @@ 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); return LogwrtResult.Write; } -- 2.20.1 --0OAP2g/MAC+5xKAE-- ^ permalink raw reply [nested|flat] 25+ messages in thread
* [PATCH 2/2] make LogwrtResult atomic @ 2021-02-02 17:03 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 25+ messages in thread From: Alvaro Herrera @ 2021-02-02 17:03 UTC (permalink / raw) --- src/backend/access/transam/xlog.c | 66 +++++++++++++++---------------- 1 file changed, 32 insertions(+), 34 deletions(-) diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index f03bd473e2..10802bd56f 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -405,12 +405,9 @@ static XLogRecPtr RedoStartLSN = InvalidXLogRecPtr; * These structs are identical but are declared separately to indicate their * slightly different functions. * - * To read XLogCtl->LogwrtResult, you must hold either info_lck or - * WALWriteLock. To update it, you need to hold both locks. The point of - * this arrangement is that the value can be examined by code that already - * holds WALWriteLock without needing to grab info_lck as well. In addition - * to the shared variable, each backend has a private copy of LogwrtResult, - * which is updated when convenient. + * XLogCtl->LogwrtResult is read and written 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. * * 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. @@ -433,6 +430,12 @@ static XLogRecPtr RedoStartLSN = InvalidXLogRecPtr; *---------- */ +typedef struct XLogwrtAtomic +{ + pg_atomic_uint64 Write; /* last byte + 1 of write position */ + pg_atomic_uint64 Flush; /* last byte + 1 of flush position */ +} XLogwrtAtomic; + typedef struct XLogwrtRqst { XLogRecPtr Write; /* last byte + 1 to write out */ @@ -596,6 +599,7 @@ typedef struct XLogCtlData { XLogCtlInsert Insert; + XLogwrtAtomic LogwrtResult; /* Protected by info_lck: */ XLogwrtRqst LogwrtRqst; XLogRecPtr RedoRecPtr; /* a recent copy of Insert->RedoRecPtr */ @@ -613,12 +617,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). * @@ -1172,9 +1170,10 @@ XLogInsertRecord(XLogRecData *rdata, /* 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); + /* update local result copy */ + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); } /* @@ -2170,13 +2169,13 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic) SpinLockAcquire(&XLogCtl->info_lck); if (XLogCtl->LogwrtRqst.Write < OldPageRqstPtr) XLogCtl->LogwrtRqst.Write = OldPageRqstPtr; - LogwrtResult = XLogCtl->LogwrtResult; SpinLockRelease(&XLogCtl->info_lck); /* * Now that we have an up-to-date LogwrtResult value, 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) { /* @@ -2191,7 +2190,7 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic) LWLockAcquire(WALWriteLock, LW_EXCLUSIVE); - LogwrtResult = XLogCtl->LogwrtResult; + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); if (LogwrtResult.Write >= OldPageRqstPtr) { /* OK, someone wrote it already */ @@ -2438,7 +2437,7 @@ 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); /* * Since successive pages in the xlog cache are consecutively allocated, @@ -2636,7 +2635,6 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) */ if (LogwrtResult.Flush < WriteRqst.Flush && LogwrtResult.Flush < LogwrtResult.Write) - { /* * Could get here without iterating above loop, in which case we might @@ -2675,8 +2673,10 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) * code in a couple of places. */ { + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Write, LogwrtResult.Write); + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Flush, LogwrtResult.Flush); + pg_memory_barrier(); SpinLockAcquire(&XLogCtl->info_lck); - XLogCtl->LogwrtResult = LogwrtResult; if (XLogCtl->LogwrtRqst.Write < LogwrtResult.Write) XLogCtl->LogwrtRqst.Write = LogwrtResult.Write; if (XLogCtl->LogwrtRqst.Flush < LogwrtResult.Flush) @@ -2696,8 +2696,8 @@ XLogSetAsyncXactLSN(XLogRecPtr asyncXactLSN) XLogRecPtr WriteRqstPtr = asyncXactLSN; bool sleeping; + 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; @@ -2908,10 +2908,10 @@ XLogFlush(XLogRecPtr record) SpinLockAcquire(&XLogCtl->info_lck); if (WriteRqstPtr < XLogCtl->LogwrtRqst.Write) WriteRqstPtr = XLogCtl->LogwrtRqst.Write; - LogwrtResult = XLogCtl->LogwrtResult; SpinLockRelease(&XLogCtl->info_lck); /* done already? */ + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); if (record <= LogwrtResult.Flush) break; @@ -2939,7 +2939,7 @@ XLogFlush(XLogRecPtr record) } /* Got the lock; recheck whether request is satisfied */ - LogwrtResult = XLogCtl->LogwrtResult; + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); if (record <= LogwrtResult.Flush) { LWLockRelease(WALWriteLock); @@ -3056,7 +3056,6 @@ XLogBackgroundFlush(void) /* read LogwrtResult and update local state */ SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; WriteRqst = XLogCtl->LogwrtRqst; SpinLockRelease(&XLogCtl->info_lck); @@ -3064,8 +3063,10 @@ 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) { + pg_memory_barrier(); SpinLockAcquire(&XLogCtl->info_lck); WriteRqst.Write = XLogCtl->asyncXactLSN; SpinLockRelease(&XLogCtl->info_lck); @@ -3081,6 +3082,7 @@ XLogBackgroundFlush(void) { if (openLogFile >= 0) { + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); if (!XLByteInPrevSeg(LogwrtResult.Write, openLogSegNo, wal_segment_size)) { @@ -3140,7 +3142,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 +3231,7 @@ XLogNeedsFlush(XLogRecPtr record) return false; /* read LogwrtResult and update local state */ - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); /* check again */ if (record <= LogwrtResult.Flush) @@ -7779,7 +7780,8 @@ StartupXLOG(void) LogwrtResult.Write = LogwrtResult.Flush = EndOfLog; - XLogCtl->LogwrtResult = LogwrtResult; + pg_atomic_write_u64(&XLogCtl->LogwrtResult.Write, EndOfLog); + pg_atomic_write_u64(&XLogCtl->LogwrtResult.Flush, EndOfLog); XLogCtl->LogwrtRqst.Write = EndOfLog; XLogCtl->LogwrtRqst.Flush = EndOfLog; @@ -8498,9 +8500,7 @@ GetInsertRecPtr(void) XLogRecPtr GetFlushRecPtr(void) { - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); return LogwrtResult.Flush; } @@ -11649,9 +11649,7 @@ 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); return LogwrtResult.Write; } -- 2.20.1 --0OAP2g/MAC+5xKAE-- ^ permalink raw reply [nested|flat] 25+ messages in thread
* [PATCH 2/2] make LogwrtResult atomic @ 2021-02-02 17:03 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 25+ messages in thread From: Alvaro Herrera @ 2021-02-02 17:03 UTC (permalink / raw) --- src/backend/access/transam/xlog.c | 66 +++++++++++++++---------------- 1 file changed, 32 insertions(+), 34 deletions(-) diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index f03bd473e2..10802bd56f 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -405,12 +405,9 @@ static XLogRecPtr RedoStartLSN = InvalidXLogRecPtr; * These structs are identical but are declared separately to indicate their * slightly different functions. * - * To read XLogCtl->LogwrtResult, you must hold either info_lck or - * WALWriteLock. To update it, you need to hold both locks. The point of - * this arrangement is that the value can be examined by code that already - * holds WALWriteLock without needing to grab info_lck as well. In addition - * to the shared variable, each backend has a private copy of LogwrtResult, - * which is updated when convenient. + * XLogCtl->LogwrtResult is read and written 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. * * 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. @@ -433,6 +430,12 @@ static XLogRecPtr RedoStartLSN = InvalidXLogRecPtr; *---------- */ +typedef struct XLogwrtAtomic +{ + pg_atomic_uint64 Write; /* last byte + 1 of write position */ + pg_atomic_uint64 Flush; /* last byte + 1 of flush position */ +} XLogwrtAtomic; + typedef struct XLogwrtRqst { XLogRecPtr Write; /* last byte + 1 to write out */ @@ -596,6 +599,7 @@ typedef struct XLogCtlData { XLogCtlInsert Insert; + XLogwrtAtomic LogwrtResult; /* Protected by info_lck: */ XLogwrtRqst LogwrtRqst; XLogRecPtr RedoRecPtr; /* a recent copy of Insert->RedoRecPtr */ @@ -613,12 +617,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). * @@ -1172,9 +1170,10 @@ XLogInsertRecord(XLogRecData *rdata, /* 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); + /* update local result copy */ + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); } /* @@ -2170,13 +2169,13 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic) SpinLockAcquire(&XLogCtl->info_lck); if (XLogCtl->LogwrtRqst.Write < OldPageRqstPtr) XLogCtl->LogwrtRqst.Write = OldPageRqstPtr; - LogwrtResult = XLogCtl->LogwrtResult; SpinLockRelease(&XLogCtl->info_lck); /* * Now that we have an up-to-date LogwrtResult value, 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) { /* @@ -2191,7 +2190,7 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic) LWLockAcquire(WALWriteLock, LW_EXCLUSIVE); - LogwrtResult = XLogCtl->LogwrtResult; + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); if (LogwrtResult.Write >= OldPageRqstPtr) { /* OK, someone wrote it already */ @@ -2438,7 +2437,7 @@ 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); /* * Since successive pages in the xlog cache are consecutively allocated, @@ -2636,7 +2635,6 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) */ if (LogwrtResult.Flush < WriteRqst.Flush && LogwrtResult.Flush < LogwrtResult.Write) - { /* * Could get here without iterating above loop, in which case we might @@ -2675,8 +2673,10 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) * code in a couple of places. */ { + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Write, LogwrtResult.Write); + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Flush, LogwrtResult.Flush); + pg_memory_barrier(); SpinLockAcquire(&XLogCtl->info_lck); - XLogCtl->LogwrtResult = LogwrtResult; if (XLogCtl->LogwrtRqst.Write < LogwrtResult.Write) XLogCtl->LogwrtRqst.Write = LogwrtResult.Write; if (XLogCtl->LogwrtRqst.Flush < LogwrtResult.Flush) @@ -2696,8 +2696,8 @@ XLogSetAsyncXactLSN(XLogRecPtr asyncXactLSN) XLogRecPtr WriteRqstPtr = asyncXactLSN; bool sleeping; + 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; @@ -2908,10 +2908,10 @@ XLogFlush(XLogRecPtr record) SpinLockAcquire(&XLogCtl->info_lck); if (WriteRqstPtr < XLogCtl->LogwrtRqst.Write) WriteRqstPtr = XLogCtl->LogwrtRqst.Write; - LogwrtResult = XLogCtl->LogwrtResult; SpinLockRelease(&XLogCtl->info_lck); /* done already? */ + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); if (record <= LogwrtResult.Flush) break; @@ -2939,7 +2939,7 @@ XLogFlush(XLogRecPtr record) } /* Got the lock; recheck whether request is satisfied */ - LogwrtResult = XLogCtl->LogwrtResult; + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); if (record <= LogwrtResult.Flush) { LWLockRelease(WALWriteLock); @@ -3056,7 +3056,6 @@ XLogBackgroundFlush(void) /* read LogwrtResult and update local state */ SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; WriteRqst = XLogCtl->LogwrtRqst; SpinLockRelease(&XLogCtl->info_lck); @@ -3064,8 +3063,10 @@ 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) { + pg_memory_barrier(); SpinLockAcquire(&XLogCtl->info_lck); WriteRqst.Write = XLogCtl->asyncXactLSN; SpinLockRelease(&XLogCtl->info_lck); @@ -3081,6 +3082,7 @@ XLogBackgroundFlush(void) { if (openLogFile >= 0) { + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); if (!XLByteInPrevSeg(LogwrtResult.Write, openLogSegNo, wal_segment_size)) { @@ -3140,7 +3142,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 +3231,7 @@ XLogNeedsFlush(XLogRecPtr record) return false; /* read LogwrtResult and update local state */ - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); /* check again */ if (record <= LogwrtResult.Flush) @@ -7779,7 +7780,8 @@ StartupXLOG(void) LogwrtResult.Write = LogwrtResult.Flush = EndOfLog; - XLogCtl->LogwrtResult = LogwrtResult; + pg_atomic_write_u64(&XLogCtl->LogwrtResult.Write, EndOfLog); + pg_atomic_write_u64(&XLogCtl->LogwrtResult.Flush, EndOfLog); XLogCtl->LogwrtRqst.Write = EndOfLog; XLogCtl->LogwrtRqst.Flush = EndOfLog; @@ -8498,9 +8500,7 @@ GetInsertRecPtr(void) XLogRecPtr GetFlushRecPtr(void) { - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); return LogwrtResult.Flush; } @@ -11649,9 +11649,7 @@ 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); return LogwrtResult.Write; } -- 2.20.1 --0OAP2g/MAC+5xKAE-- ^ permalink raw reply [nested|flat] 25+ messages in thread
* [PATCH 2/2] make LogwrtResult atomic @ 2021-02-02 17:03 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 25+ messages in thread From: Alvaro Herrera @ 2021-02-02 17:03 UTC (permalink / raw) --- src/backend/access/transam/xlog.c | 66 +++++++++++++++---------------- 1 file changed, 32 insertions(+), 34 deletions(-) diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index f03bd473e2..10802bd56f 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -405,12 +405,9 @@ static XLogRecPtr RedoStartLSN = InvalidXLogRecPtr; * These structs are identical but are declared separately to indicate their * slightly different functions. * - * To read XLogCtl->LogwrtResult, you must hold either info_lck or - * WALWriteLock. To update it, you need to hold both locks. The point of - * this arrangement is that the value can be examined by code that already - * holds WALWriteLock without needing to grab info_lck as well. In addition - * to the shared variable, each backend has a private copy of LogwrtResult, - * which is updated when convenient. + * XLogCtl->LogwrtResult is read and written 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. * * 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. @@ -433,6 +430,12 @@ static XLogRecPtr RedoStartLSN = InvalidXLogRecPtr; *---------- */ +typedef struct XLogwrtAtomic +{ + pg_atomic_uint64 Write; /* last byte + 1 of write position */ + pg_atomic_uint64 Flush; /* last byte + 1 of flush position */ +} XLogwrtAtomic; + typedef struct XLogwrtRqst { XLogRecPtr Write; /* last byte + 1 to write out */ @@ -596,6 +599,7 @@ typedef struct XLogCtlData { XLogCtlInsert Insert; + XLogwrtAtomic LogwrtResult; /* Protected by info_lck: */ XLogwrtRqst LogwrtRqst; XLogRecPtr RedoRecPtr; /* a recent copy of Insert->RedoRecPtr */ @@ -613,12 +617,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). * @@ -1172,9 +1170,10 @@ XLogInsertRecord(XLogRecData *rdata, /* 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); + /* update local result copy */ + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); } /* @@ -2170,13 +2169,13 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic) SpinLockAcquire(&XLogCtl->info_lck); if (XLogCtl->LogwrtRqst.Write < OldPageRqstPtr) XLogCtl->LogwrtRqst.Write = OldPageRqstPtr; - LogwrtResult = XLogCtl->LogwrtResult; SpinLockRelease(&XLogCtl->info_lck); /* * Now that we have an up-to-date LogwrtResult value, 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) { /* @@ -2191,7 +2190,7 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic) LWLockAcquire(WALWriteLock, LW_EXCLUSIVE); - LogwrtResult = XLogCtl->LogwrtResult; + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); if (LogwrtResult.Write >= OldPageRqstPtr) { /* OK, someone wrote it already */ @@ -2438,7 +2437,7 @@ 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); /* * Since successive pages in the xlog cache are consecutively allocated, @@ -2636,7 +2635,6 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) */ if (LogwrtResult.Flush < WriteRqst.Flush && LogwrtResult.Flush < LogwrtResult.Write) - { /* * Could get here without iterating above loop, in which case we might @@ -2675,8 +2673,10 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) * code in a couple of places. */ { + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Write, LogwrtResult.Write); + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Flush, LogwrtResult.Flush); + pg_memory_barrier(); SpinLockAcquire(&XLogCtl->info_lck); - XLogCtl->LogwrtResult = LogwrtResult; if (XLogCtl->LogwrtRqst.Write < LogwrtResult.Write) XLogCtl->LogwrtRqst.Write = LogwrtResult.Write; if (XLogCtl->LogwrtRqst.Flush < LogwrtResult.Flush) @@ -2696,8 +2696,8 @@ XLogSetAsyncXactLSN(XLogRecPtr asyncXactLSN) XLogRecPtr WriteRqstPtr = asyncXactLSN; bool sleeping; + 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; @@ -2908,10 +2908,10 @@ XLogFlush(XLogRecPtr record) SpinLockAcquire(&XLogCtl->info_lck); if (WriteRqstPtr < XLogCtl->LogwrtRqst.Write) WriteRqstPtr = XLogCtl->LogwrtRqst.Write; - LogwrtResult = XLogCtl->LogwrtResult; SpinLockRelease(&XLogCtl->info_lck); /* done already? */ + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); if (record <= LogwrtResult.Flush) break; @@ -2939,7 +2939,7 @@ XLogFlush(XLogRecPtr record) } /* Got the lock; recheck whether request is satisfied */ - LogwrtResult = XLogCtl->LogwrtResult; + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); if (record <= LogwrtResult.Flush) { LWLockRelease(WALWriteLock); @@ -3056,7 +3056,6 @@ XLogBackgroundFlush(void) /* read LogwrtResult and update local state */ SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; WriteRqst = XLogCtl->LogwrtRqst; SpinLockRelease(&XLogCtl->info_lck); @@ -3064,8 +3063,10 @@ 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) { + pg_memory_barrier(); SpinLockAcquire(&XLogCtl->info_lck); WriteRqst.Write = XLogCtl->asyncXactLSN; SpinLockRelease(&XLogCtl->info_lck); @@ -3081,6 +3082,7 @@ XLogBackgroundFlush(void) { if (openLogFile >= 0) { + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); if (!XLByteInPrevSeg(LogwrtResult.Write, openLogSegNo, wal_segment_size)) { @@ -3140,7 +3142,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 +3231,7 @@ XLogNeedsFlush(XLogRecPtr record) return false; /* read LogwrtResult and update local state */ - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); /* check again */ if (record <= LogwrtResult.Flush) @@ -7779,7 +7780,8 @@ StartupXLOG(void) LogwrtResult.Write = LogwrtResult.Flush = EndOfLog; - XLogCtl->LogwrtResult = LogwrtResult; + pg_atomic_write_u64(&XLogCtl->LogwrtResult.Write, EndOfLog); + pg_atomic_write_u64(&XLogCtl->LogwrtResult.Flush, EndOfLog); XLogCtl->LogwrtRqst.Write = EndOfLog; XLogCtl->LogwrtRqst.Flush = EndOfLog; @@ -8498,9 +8500,7 @@ GetInsertRecPtr(void) XLogRecPtr GetFlushRecPtr(void) { - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); return LogwrtResult.Flush; } @@ -11649,9 +11649,7 @@ 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); return LogwrtResult.Write; } -- 2.20.1 --0OAP2g/MAC+5xKAE-- ^ permalink raw reply [nested|flat] 25+ messages in thread
* [PATCH 2/2] make LogwrtResult atomic @ 2021-02-02 17:03 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 25+ messages in thread From: Alvaro Herrera @ 2021-02-02 17:03 UTC (permalink / raw) --- src/backend/access/transam/xlog.c | 66 +++++++++++++++---------------- 1 file changed, 32 insertions(+), 34 deletions(-) diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index f03bd473e2..10802bd56f 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -405,12 +405,9 @@ static XLogRecPtr RedoStartLSN = InvalidXLogRecPtr; * These structs are identical but are declared separately to indicate their * slightly different functions. * - * To read XLogCtl->LogwrtResult, you must hold either info_lck or - * WALWriteLock. To update it, you need to hold both locks. The point of - * this arrangement is that the value can be examined by code that already - * holds WALWriteLock without needing to grab info_lck as well. In addition - * to the shared variable, each backend has a private copy of LogwrtResult, - * which is updated when convenient. + * XLogCtl->LogwrtResult is read and written 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. * * 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. @@ -433,6 +430,12 @@ static XLogRecPtr RedoStartLSN = InvalidXLogRecPtr; *---------- */ +typedef struct XLogwrtAtomic +{ + pg_atomic_uint64 Write; /* last byte + 1 of write position */ + pg_atomic_uint64 Flush; /* last byte + 1 of flush position */ +} XLogwrtAtomic; + typedef struct XLogwrtRqst { XLogRecPtr Write; /* last byte + 1 to write out */ @@ -596,6 +599,7 @@ typedef struct XLogCtlData { XLogCtlInsert Insert; + XLogwrtAtomic LogwrtResult; /* Protected by info_lck: */ XLogwrtRqst LogwrtRqst; XLogRecPtr RedoRecPtr; /* a recent copy of Insert->RedoRecPtr */ @@ -613,12 +617,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). * @@ -1172,9 +1170,10 @@ XLogInsertRecord(XLogRecData *rdata, /* 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); + /* update local result copy */ + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); } /* @@ -2170,13 +2169,13 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic) SpinLockAcquire(&XLogCtl->info_lck); if (XLogCtl->LogwrtRqst.Write < OldPageRqstPtr) XLogCtl->LogwrtRqst.Write = OldPageRqstPtr; - LogwrtResult = XLogCtl->LogwrtResult; SpinLockRelease(&XLogCtl->info_lck); /* * Now that we have an up-to-date LogwrtResult value, 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) { /* @@ -2191,7 +2190,7 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic) LWLockAcquire(WALWriteLock, LW_EXCLUSIVE); - LogwrtResult = XLogCtl->LogwrtResult; + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); if (LogwrtResult.Write >= OldPageRqstPtr) { /* OK, someone wrote it already */ @@ -2438,7 +2437,7 @@ 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); /* * Since successive pages in the xlog cache are consecutively allocated, @@ -2636,7 +2635,6 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) */ if (LogwrtResult.Flush < WriteRqst.Flush && LogwrtResult.Flush < LogwrtResult.Write) - { /* * Could get here without iterating above loop, in which case we might @@ -2675,8 +2673,10 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) * code in a couple of places. */ { + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Write, LogwrtResult.Write); + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Flush, LogwrtResult.Flush); + pg_memory_barrier(); SpinLockAcquire(&XLogCtl->info_lck); - XLogCtl->LogwrtResult = LogwrtResult; if (XLogCtl->LogwrtRqst.Write < LogwrtResult.Write) XLogCtl->LogwrtRqst.Write = LogwrtResult.Write; if (XLogCtl->LogwrtRqst.Flush < LogwrtResult.Flush) @@ -2696,8 +2696,8 @@ XLogSetAsyncXactLSN(XLogRecPtr asyncXactLSN) XLogRecPtr WriteRqstPtr = asyncXactLSN; bool sleeping; + 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; @@ -2908,10 +2908,10 @@ XLogFlush(XLogRecPtr record) SpinLockAcquire(&XLogCtl->info_lck); if (WriteRqstPtr < XLogCtl->LogwrtRqst.Write) WriteRqstPtr = XLogCtl->LogwrtRqst.Write; - LogwrtResult = XLogCtl->LogwrtResult; SpinLockRelease(&XLogCtl->info_lck); /* done already? */ + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); if (record <= LogwrtResult.Flush) break; @@ -2939,7 +2939,7 @@ XLogFlush(XLogRecPtr record) } /* Got the lock; recheck whether request is satisfied */ - LogwrtResult = XLogCtl->LogwrtResult; + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); if (record <= LogwrtResult.Flush) { LWLockRelease(WALWriteLock); @@ -3056,7 +3056,6 @@ XLogBackgroundFlush(void) /* read LogwrtResult and update local state */ SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; WriteRqst = XLogCtl->LogwrtRqst; SpinLockRelease(&XLogCtl->info_lck); @@ -3064,8 +3063,10 @@ 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) { + pg_memory_barrier(); SpinLockAcquire(&XLogCtl->info_lck); WriteRqst.Write = XLogCtl->asyncXactLSN; SpinLockRelease(&XLogCtl->info_lck); @@ -3081,6 +3082,7 @@ XLogBackgroundFlush(void) { if (openLogFile >= 0) { + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); if (!XLByteInPrevSeg(LogwrtResult.Write, openLogSegNo, wal_segment_size)) { @@ -3140,7 +3142,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 +3231,7 @@ XLogNeedsFlush(XLogRecPtr record) return false; /* read LogwrtResult and update local state */ - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); /* check again */ if (record <= LogwrtResult.Flush) @@ -7779,7 +7780,8 @@ StartupXLOG(void) LogwrtResult.Write = LogwrtResult.Flush = EndOfLog; - XLogCtl->LogwrtResult = LogwrtResult; + pg_atomic_write_u64(&XLogCtl->LogwrtResult.Write, EndOfLog); + pg_atomic_write_u64(&XLogCtl->LogwrtResult.Flush, EndOfLog); XLogCtl->LogwrtRqst.Write = EndOfLog; XLogCtl->LogwrtRqst.Flush = EndOfLog; @@ -8498,9 +8500,7 @@ GetInsertRecPtr(void) XLogRecPtr GetFlushRecPtr(void) { - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); return LogwrtResult.Flush; } @@ -11649,9 +11649,7 @@ 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); return LogwrtResult.Write; } -- 2.20.1 --0OAP2g/MAC+5xKAE-- ^ permalink raw reply [nested|flat] 25+ messages in thread
* [PATCH 2/2] make LogwrtResult atomic @ 2021-02-02 17:03 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 25+ messages in thread From: Alvaro Herrera @ 2021-02-02 17:03 UTC (permalink / raw) --- src/backend/access/transam/xlog.c | 66 +++++++++++++++---------------- 1 file changed, 32 insertions(+), 34 deletions(-) diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index f03bd473e2..10802bd56f 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -405,12 +405,9 @@ static XLogRecPtr RedoStartLSN = InvalidXLogRecPtr; * These structs are identical but are declared separately to indicate their * slightly different functions. * - * To read XLogCtl->LogwrtResult, you must hold either info_lck or - * WALWriteLock. To update it, you need to hold both locks. The point of - * this arrangement is that the value can be examined by code that already - * holds WALWriteLock without needing to grab info_lck as well. In addition - * to the shared variable, each backend has a private copy of LogwrtResult, - * which is updated when convenient. + * XLogCtl->LogwrtResult is read and written 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. * * 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. @@ -433,6 +430,12 @@ static XLogRecPtr RedoStartLSN = InvalidXLogRecPtr; *---------- */ +typedef struct XLogwrtAtomic +{ + pg_atomic_uint64 Write; /* last byte + 1 of write position */ + pg_atomic_uint64 Flush; /* last byte + 1 of flush position */ +} XLogwrtAtomic; + typedef struct XLogwrtRqst { XLogRecPtr Write; /* last byte + 1 to write out */ @@ -596,6 +599,7 @@ typedef struct XLogCtlData { XLogCtlInsert Insert; + XLogwrtAtomic LogwrtResult; /* Protected by info_lck: */ XLogwrtRqst LogwrtRqst; XLogRecPtr RedoRecPtr; /* a recent copy of Insert->RedoRecPtr */ @@ -613,12 +617,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). * @@ -1172,9 +1170,10 @@ XLogInsertRecord(XLogRecData *rdata, /* 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); + /* update local result copy */ + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); } /* @@ -2170,13 +2169,13 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic) SpinLockAcquire(&XLogCtl->info_lck); if (XLogCtl->LogwrtRqst.Write < OldPageRqstPtr) XLogCtl->LogwrtRqst.Write = OldPageRqstPtr; - LogwrtResult = XLogCtl->LogwrtResult; SpinLockRelease(&XLogCtl->info_lck); /* * Now that we have an up-to-date LogwrtResult value, 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) { /* @@ -2191,7 +2190,7 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic) LWLockAcquire(WALWriteLock, LW_EXCLUSIVE); - LogwrtResult = XLogCtl->LogwrtResult; + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); if (LogwrtResult.Write >= OldPageRqstPtr) { /* OK, someone wrote it already */ @@ -2438,7 +2437,7 @@ 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); /* * Since successive pages in the xlog cache are consecutively allocated, @@ -2636,7 +2635,6 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) */ if (LogwrtResult.Flush < WriteRqst.Flush && LogwrtResult.Flush < LogwrtResult.Write) - { /* * Could get here without iterating above loop, in which case we might @@ -2675,8 +2673,10 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) * code in a couple of places. */ { + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Write, LogwrtResult.Write); + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Flush, LogwrtResult.Flush); + pg_memory_barrier(); SpinLockAcquire(&XLogCtl->info_lck); - XLogCtl->LogwrtResult = LogwrtResult; if (XLogCtl->LogwrtRqst.Write < LogwrtResult.Write) XLogCtl->LogwrtRqst.Write = LogwrtResult.Write; if (XLogCtl->LogwrtRqst.Flush < LogwrtResult.Flush) @@ -2696,8 +2696,8 @@ XLogSetAsyncXactLSN(XLogRecPtr asyncXactLSN) XLogRecPtr WriteRqstPtr = asyncXactLSN; bool sleeping; + 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; @@ -2908,10 +2908,10 @@ XLogFlush(XLogRecPtr record) SpinLockAcquire(&XLogCtl->info_lck); if (WriteRqstPtr < XLogCtl->LogwrtRqst.Write) WriteRqstPtr = XLogCtl->LogwrtRqst.Write; - LogwrtResult = XLogCtl->LogwrtResult; SpinLockRelease(&XLogCtl->info_lck); /* done already? */ + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); if (record <= LogwrtResult.Flush) break; @@ -2939,7 +2939,7 @@ XLogFlush(XLogRecPtr record) } /* Got the lock; recheck whether request is satisfied */ - LogwrtResult = XLogCtl->LogwrtResult; + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); if (record <= LogwrtResult.Flush) { LWLockRelease(WALWriteLock); @@ -3056,7 +3056,6 @@ XLogBackgroundFlush(void) /* read LogwrtResult and update local state */ SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; WriteRqst = XLogCtl->LogwrtRqst; SpinLockRelease(&XLogCtl->info_lck); @@ -3064,8 +3063,10 @@ 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) { + pg_memory_barrier(); SpinLockAcquire(&XLogCtl->info_lck); WriteRqst.Write = XLogCtl->asyncXactLSN; SpinLockRelease(&XLogCtl->info_lck); @@ -3081,6 +3082,7 @@ XLogBackgroundFlush(void) { if (openLogFile >= 0) { + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); if (!XLByteInPrevSeg(LogwrtResult.Write, openLogSegNo, wal_segment_size)) { @@ -3140,7 +3142,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 +3231,7 @@ XLogNeedsFlush(XLogRecPtr record) return false; /* read LogwrtResult and update local state */ - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); /* check again */ if (record <= LogwrtResult.Flush) @@ -7779,7 +7780,8 @@ StartupXLOG(void) LogwrtResult.Write = LogwrtResult.Flush = EndOfLog; - XLogCtl->LogwrtResult = LogwrtResult; + pg_atomic_write_u64(&XLogCtl->LogwrtResult.Write, EndOfLog); + pg_atomic_write_u64(&XLogCtl->LogwrtResult.Flush, EndOfLog); XLogCtl->LogwrtRqst.Write = EndOfLog; XLogCtl->LogwrtRqst.Flush = EndOfLog; @@ -8498,9 +8500,7 @@ GetInsertRecPtr(void) XLogRecPtr GetFlushRecPtr(void) { - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); return LogwrtResult.Flush; } @@ -11649,9 +11649,7 @@ 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); return LogwrtResult.Write; } -- 2.20.1 --0OAP2g/MAC+5xKAE-- ^ permalink raw reply [nested|flat] 25+ messages in thread
* [PATCH 2/2] make LogwrtResult atomic @ 2021-02-02 17:03 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 25+ messages in thread From: Alvaro Herrera @ 2021-02-02 17:03 UTC (permalink / raw) --- src/backend/access/transam/xlog.c | 66 +++++++++++++++---------------- 1 file changed, 32 insertions(+), 34 deletions(-) diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index f03bd473e2..10802bd56f 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -405,12 +405,9 @@ static XLogRecPtr RedoStartLSN = InvalidXLogRecPtr; * These structs are identical but are declared separately to indicate their * slightly different functions. * - * To read XLogCtl->LogwrtResult, you must hold either info_lck or - * WALWriteLock. To update it, you need to hold both locks. The point of - * this arrangement is that the value can be examined by code that already - * holds WALWriteLock without needing to grab info_lck as well. In addition - * to the shared variable, each backend has a private copy of LogwrtResult, - * which is updated when convenient. + * XLogCtl->LogwrtResult is read and written 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. * * 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. @@ -433,6 +430,12 @@ static XLogRecPtr RedoStartLSN = InvalidXLogRecPtr; *---------- */ +typedef struct XLogwrtAtomic +{ + pg_atomic_uint64 Write; /* last byte + 1 of write position */ + pg_atomic_uint64 Flush; /* last byte + 1 of flush position */ +} XLogwrtAtomic; + typedef struct XLogwrtRqst { XLogRecPtr Write; /* last byte + 1 to write out */ @@ -596,6 +599,7 @@ typedef struct XLogCtlData { XLogCtlInsert Insert; + XLogwrtAtomic LogwrtResult; /* Protected by info_lck: */ XLogwrtRqst LogwrtRqst; XLogRecPtr RedoRecPtr; /* a recent copy of Insert->RedoRecPtr */ @@ -613,12 +617,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). * @@ -1172,9 +1170,10 @@ XLogInsertRecord(XLogRecData *rdata, /* 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); + /* update local result copy */ + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); } /* @@ -2170,13 +2169,13 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic) SpinLockAcquire(&XLogCtl->info_lck); if (XLogCtl->LogwrtRqst.Write < OldPageRqstPtr) XLogCtl->LogwrtRqst.Write = OldPageRqstPtr; - LogwrtResult = XLogCtl->LogwrtResult; SpinLockRelease(&XLogCtl->info_lck); /* * Now that we have an up-to-date LogwrtResult value, 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) { /* @@ -2191,7 +2190,7 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic) LWLockAcquire(WALWriteLock, LW_EXCLUSIVE); - LogwrtResult = XLogCtl->LogwrtResult; + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); if (LogwrtResult.Write >= OldPageRqstPtr) { /* OK, someone wrote it already */ @@ -2438,7 +2437,7 @@ 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); /* * Since successive pages in the xlog cache are consecutively allocated, @@ -2636,7 +2635,6 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) */ if (LogwrtResult.Flush < WriteRqst.Flush && LogwrtResult.Flush < LogwrtResult.Write) - { /* * Could get here without iterating above loop, in which case we might @@ -2675,8 +2673,10 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) * code in a couple of places. */ { + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Write, LogwrtResult.Write); + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Flush, LogwrtResult.Flush); + pg_memory_barrier(); SpinLockAcquire(&XLogCtl->info_lck); - XLogCtl->LogwrtResult = LogwrtResult; if (XLogCtl->LogwrtRqst.Write < LogwrtResult.Write) XLogCtl->LogwrtRqst.Write = LogwrtResult.Write; if (XLogCtl->LogwrtRqst.Flush < LogwrtResult.Flush) @@ -2696,8 +2696,8 @@ XLogSetAsyncXactLSN(XLogRecPtr asyncXactLSN) XLogRecPtr WriteRqstPtr = asyncXactLSN; bool sleeping; + 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; @@ -2908,10 +2908,10 @@ XLogFlush(XLogRecPtr record) SpinLockAcquire(&XLogCtl->info_lck); if (WriteRqstPtr < XLogCtl->LogwrtRqst.Write) WriteRqstPtr = XLogCtl->LogwrtRqst.Write; - LogwrtResult = XLogCtl->LogwrtResult; SpinLockRelease(&XLogCtl->info_lck); /* done already? */ + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); if (record <= LogwrtResult.Flush) break; @@ -2939,7 +2939,7 @@ XLogFlush(XLogRecPtr record) } /* Got the lock; recheck whether request is satisfied */ - LogwrtResult = XLogCtl->LogwrtResult; + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); if (record <= LogwrtResult.Flush) { LWLockRelease(WALWriteLock); @@ -3056,7 +3056,6 @@ XLogBackgroundFlush(void) /* read LogwrtResult and update local state */ SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; WriteRqst = XLogCtl->LogwrtRqst; SpinLockRelease(&XLogCtl->info_lck); @@ -3064,8 +3063,10 @@ 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) { + pg_memory_barrier(); SpinLockAcquire(&XLogCtl->info_lck); WriteRqst.Write = XLogCtl->asyncXactLSN; SpinLockRelease(&XLogCtl->info_lck); @@ -3081,6 +3082,7 @@ XLogBackgroundFlush(void) { if (openLogFile >= 0) { + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); if (!XLByteInPrevSeg(LogwrtResult.Write, openLogSegNo, wal_segment_size)) { @@ -3140,7 +3142,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 +3231,7 @@ XLogNeedsFlush(XLogRecPtr record) return false; /* read LogwrtResult and update local state */ - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); /* check again */ if (record <= LogwrtResult.Flush) @@ -7779,7 +7780,8 @@ StartupXLOG(void) LogwrtResult.Write = LogwrtResult.Flush = EndOfLog; - XLogCtl->LogwrtResult = LogwrtResult; + pg_atomic_write_u64(&XLogCtl->LogwrtResult.Write, EndOfLog); + pg_atomic_write_u64(&XLogCtl->LogwrtResult.Flush, EndOfLog); XLogCtl->LogwrtRqst.Write = EndOfLog; XLogCtl->LogwrtRqst.Flush = EndOfLog; @@ -8498,9 +8500,7 @@ GetInsertRecPtr(void) XLogRecPtr GetFlushRecPtr(void) { - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); return LogwrtResult.Flush; } @@ -11649,9 +11649,7 @@ 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); return LogwrtResult.Write; } -- 2.20.1 --0OAP2g/MAC+5xKAE-- ^ permalink raw reply [nested|flat] 25+ messages in thread
* [PATCH 2/2] make LogwrtResult atomic @ 2021-02-02 17:03 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 25+ messages in thread From: Alvaro Herrera @ 2021-02-02 17:03 UTC (permalink / raw) --- src/backend/access/transam/xlog.c | 66 +++++++++++++++---------------- 1 file changed, 32 insertions(+), 34 deletions(-) diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index f03bd473e2..10802bd56f 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -405,12 +405,9 @@ static XLogRecPtr RedoStartLSN = InvalidXLogRecPtr; * These structs are identical but are declared separately to indicate their * slightly different functions. * - * To read XLogCtl->LogwrtResult, you must hold either info_lck or - * WALWriteLock. To update it, you need to hold both locks. The point of - * this arrangement is that the value can be examined by code that already - * holds WALWriteLock without needing to grab info_lck as well. In addition - * to the shared variable, each backend has a private copy of LogwrtResult, - * which is updated when convenient. + * XLogCtl->LogwrtResult is read and written 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. * * 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. @@ -433,6 +430,12 @@ static XLogRecPtr RedoStartLSN = InvalidXLogRecPtr; *---------- */ +typedef struct XLogwrtAtomic +{ + pg_atomic_uint64 Write; /* last byte + 1 of write position */ + pg_atomic_uint64 Flush; /* last byte + 1 of flush position */ +} XLogwrtAtomic; + typedef struct XLogwrtRqst { XLogRecPtr Write; /* last byte + 1 to write out */ @@ -596,6 +599,7 @@ typedef struct XLogCtlData { XLogCtlInsert Insert; + XLogwrtAtomic LogwrtResult; /* Protected by info_lck: */ XLogwrtRqst LogwrtRqst; XLogRecPtr RedoRecPtr; /* a recent copy of Insert->RedoRecPtr */ @@ -613,12 +617,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). * @@ -1172,9 +1170,10 @@ XLogInsertRecord(XLogRecData *rdata, /* 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); + /* update local result copy */ + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); } /* @@ -2170,13 +2169,13 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic) SpinLockAcquire(&XLogCtl->info_lck); if (XLogCtl->LogwrtRqst.Write < OldPageRqstPtr) XLogCtl->LogwrtRqst.Write = OldPageRqstPtr; - LogwrtResult = XLogCtl->LogwrtResult; SpinLockRelease(&XLogCtl->info_lck); /* * Now that we have an up-to-date LogwrtResult value, 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) { /* @@ -2191,7 +2190,7 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic) LWLockAcquire(WALWriteLock, LW_EXCLUSIVE); - LogwrtResult = XLogCtl->LogwrtResult; + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); if (LogwrtResult.Write >= OldPageRqstPtr) { /* OK, someone wrote it already */ @@ -2438,7 +2437,7 @@ 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); /* * Since successive pages in the xlog cache are consecutively allocated, @@ -2636,7 +2635,6 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) */ if (LogwrtResult.Flush < WriteRqst.Flush && LogwrtResult.Flush < LogwrtResult.Write) - { /* * Could get here without iterating above loop, in which case we might @@ -2675,8 +2673,10 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) * code in a couple of places. */ { + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Write, LogwrtResult.Write); + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Flush, LogwrtResult.Flush); + pg_memory_barrier(); SpinLockAcquire(&XLogCtl->info_lck); - XLogCtl->LogwrtResult = LogwrtResult; if (XLogCtl->LogwrtRqst.Write < LogwrtResult.Write) XLogCtl->LogwrtRqst.Write = LogwrtResult.Write; if (XLogCtl->LogwrtRqst.Flush < LogwrtResult.Flush) @@ -2696,8 +2696,8 @@ XLogSetAsyncXactLSN(XLogRecPtr asyncXactLSN) XLogRecPtr WriteRqstPtr = asyncXactLSN; bool sleeping; + 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; @@ -2908,10 +2908,10 @@ XLogFlush(XLogRecPtr record) SpinLockAcquire(&XLogCtl->info_lck); if (WriteRqstPtr < XLogCtl->LogwrtRqst.Write) WriteRqstPtr = XLogCtl->LogwrtRqst.Write; - LogwrtResult = XLogCtl->LogwrtResult; SpinLockRelease(&XLogCtl->info_lck); /* done already? */ + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); if (record <= LogwrtResult.Flush) break; @@ -2939,7 +2939,7 @@ XLogFlush(XLogRecPtr record) } /* Got the lock; recheck whether request is satisfied */ - LogwrtResult = XLogCtl->LogwrtResult; + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); if (record <= LogwrtResult.Flush) { LWLockRelease(WALWriteLock); @@ -3056,7 +3056,6 @@ XLogBackgroundFlush(void) /* read LogwrtResult and update local state */ SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; WriteRqst = XLogCtl->LogwrtRqst; SpinLockRelease(&XLogCtl->info_lck); @@ -3064,8 +3063,10 @@ 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) { + pg_memory_barrier(); SpinLockAcquire(&XLogCtl->info_lck); WriteRqst.Write = XLogCtl->asyncXactLSN; SpinLockRelease(&XLogCtl->info_lck); @@ -3081,6 +3082,7 @@ XLogBackgroundFlush(void) { if (openLogFile >= 0) { + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); if (!XLByteInPrevSeg(LogwrtResult.Write, openLogSegNo, wal_segment_size)) { @@ -3140,7 +3142,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 +3231,7 @@ XLogNeedsFlush(XLogRecPtr record) return false; /* read LogwrtResult and update local state */ - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); /* check again */ if (record <= LogwrtResult.Flush) @@ -7779,7 +7780,8 @@ StartupXLOG(void) LogwrtResult.Write = LogwrtResult.Flush = EndOfLog; - XLogCtl->LogwrtResult = LogwrtResult; + pg_atomic_write_u64(&XLogCtl->LogwrtResult.Write, EndOfLog); + pg_atomic_write_u64(&XLogCtl->LogwrtResult.Flush, EndOfLog); XLogCtl->LogwrtRqst.Write = EndOfLog; XLogCtl->LogwrtRqst.Flush = EndOfLog; @@ -8498,9 +8500,7 @@ GetInsertRecPtr(void) XLogRecPtr GetFlushRecPtr(void) { - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); return LogwrtResult.Flush; } @@ -11649,9 +11649,7 @@ 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); return LogwrtResult.Write; } -- 2.20.1 --0OAP2g/MAC+5xKAE-- ^ permalink raw reply [nested|flat] 25+ messages in thread
* [PATCH 2/2] make LogwrtResult atomic @ 2021-02-02 17:03 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 25+ messages in thread From: Alvaro Herrera @ 2021-02-02 17:03 UTC (permalink / raw) --- src/backend/access/transam/xlog.c | 66 +++++++++++++++---------------- 1 file changed, 32 insertions(+), 34 deletions(-) diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index f03bd473e2..10802bd56f 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -405,12 +405,9 @@ static XLogRecPtr RedoStartLSN = InvalidXLogRecPtr; * These structs are identical but are declared separately to indicate their * slightly different functions. * - * To read XLogCtl->LogwrtResult, you must hold either info_lck or - * WALWriteLock. To update it, you need to hold both locks. The point of - * this arrangement is that the value can be examined by code that already - * holds WALWriteLock without needing to grab info_lck as well. In addition - * to the shared variable, each backend has a private copy of LogwrtResult, - * which is updated when convenient. + * XLogCtl->LogwrtResult is read and written 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. * * 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. @@ -433,6 +430,12 @@ static XLogRecPtr RedoStartLSN = InvalidXLogRecPtr; *---------- */ +typedef struct XLogwrtAtomic +{ + pg_atomic_uint64 Write; /* last byte + 1 of write position */ + pg_atomic_uint64 Flush; /* last byte + 1 of flush position */ +} XLogwrtAtomic; + typedef struct XLogwrtRqst { XLogRecPtr Write; /* last byte + 1 to write out */ @@ -596,6 +599,7 @@ typedef struct XLogCtlData { XLogCtlInsert Insert; + XLogwrtAtomic LogwrtResult; /* Protected by info_lck: */ XLogwrtRqst LogwrtRqst; XLogRecPtr RedoRecPtr; /* a recent copy of Insert->RedoRecPtr */ @@ -613,12 +617,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). * @@ -1172,9 +1170,10 @@ XLogInsertRecord(XLogRecData *rdata, /* 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); + /* update local result copy */ + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); } /* @@ -2170,13 +2169,13 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic) SpinLockAcquire(&XLogCtl->info_lck); if (XLogCtl->LogwrtRqst.Write < OldPageRqstPtr) XLogCtl->LogwrtRqst.Write = OldPageRqstPtr; - LogwrtResult = XLogCtl->LogwrtResult; SpinLockRelease(&XLogCtl->info_lck); /* * Now that we have an up-to-date LogwrtResult value, 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) { /* @@ -2191,7 +2190,7 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic) LWLockAcquire(WALWriteLock, LW_EXCLUSIVE); - LogwrtResult = XLogCtl->LogwrtResult; + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); if (LogwrtResult.Write >= OldPageRqstPtr) { /* OK, someone wrote it already */ @@ -2438,7 +2437,7 @@ 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); /* * Since successive pages in the xlog cache are consecutively allocated, @@ -2636,7 +2635,6 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) */ if (LogwrtResult.Flush < WriteRqst.Flush && LogwrtResult.Flush < LogwrtResult.Write) - { /* * Could get here without iterating above loop, in which case we might @@ -2675,8 +2673,10 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) * code in a couple of places. */ { + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Write, LogwrtResult.Write); + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Flush, LogwrtResult.Flush); + pg_memory_barrier(); SpinLockAcquire(&XLogCtl->info_lck); - XLogCtl->LogwrtResult = LogwrtResult; if (XLogCtl->LogwrtRqst.Write < LogwrtResult.Write) XLogCtl->LogwrtRqst.Write = LogwrtResult.Write; if (XLogCtl->LogwrtRqst.Flush < LogwrtResult.Flush) @@ -2696,8 +2696,8 @@ XLogSetAsyncXactLSN(XLogRecPtr asyncXactLSN) XLogRecPtr WriteRqstPtr = asyncXactLSN; bool sleeping; + 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; @@ -2908,10 +2908,10 @@ XLogFlush(XLogRecPtr record) SpinLockAcquire(&XLogCtl->info_lck); if (WriteRqstPtr < XLogCtl->LogwrtRqst.Write) WriteRqstPtr = XLogCtl->LogwrtRqst.Write; - LogwrtResult = XLogCtl->LogwrtResult; SpinLockRelease(&XLogCtl->info_lck); /* done already? */ + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); if (record <= LogwrtResult.Flush) break; @@ -2939,7 +2939,7 @@ XLogFlush(XLogRecPtr record) } /* Got the lock; recheck whether request is satisfied */ - LogwrtResult = XLogCtl->LogwrtResult; + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); if (record <= LogwrtResult.Flush) { LWLockRelease(WALWriteLock); @@ -3056,7 +3056,6 @@ XLogBackgroundFlush(void) /* read LogwrtResult and update local state */ SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; WriteRqst = XLogCtl->LogwrtRqst; SpinLockRelease(&XLogCtl->info_lck); @@ -3064,8 +3063,10 @@ 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) { + pg_memory_barrier(); SpinLockAcquire(&XLogCtl->info_lck); WriteRqst.Write = XLogCtl->asyncXactLSN; SpinLockRelease(&XLogCtl->info_lck); @@ -3081,6 +3082,7 @@ XLogBackgroundFlush(void) { if (openLogFile >= 0) { + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); if (!XLByteInPrevSeg(LogwrtResult.Write, openLogSegNo, wal_segment_size)) { @@ -3140,7 +3142,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 +3231,7 @@ XLogNeedsFlush(XLogRecPtr record) return false; /* read LogwrtResult and update local state */ - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); /* check again */ if (record <= LogwrtResult.Flush) @@ -7779,7 +7780,8 @@ StartupXLOG(void) LogwrtResult.Write = LogwrtResult.Flush = EndOfLog; - XLogCtl->LogwrtResult = LogwrtResult; + pg_atomic_write_u64(&XLogCtl->LogwrtResult.Write, EndOfLog); + pg_atomic_write_u64(&XLogCtl->LogwrtResult.Flush, EndOfLog); XLogCtl->LogwrtRqst.Write = EndOfLog; XLogCtl->LogwrtRqst.Flush = EndOfLog; @@ -8498,9 +8500,7 @@ GetInsertRecPtr(void) XLogRecPtr GetFlushRecPtr(void) { - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); return LogwrtResult.Flush; } @@ -11649,9 +11649,7 @@ 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); return LogwrtResult.Write; } -- 2.20.1 --0OAP2g/MAC+5xKAE-- ^ permalink raw reply [nested|flat] 25+ messages in thread
* [PATCH 2/2] make LogwrtResult atomic @ 2021-02-02 17:03 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 25+ messages in thread From: Alvaro Herrera @ 2021-02-02 17:03 UTC (permalink / raw) --- src/backend/access/transam/xlog.c | 66 +++++++++++++++---------------- 1 file changed, 32 insertions(+), 34 deletions(-) diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index f03bd473e2..10802bd56f 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -405,12 +405,9 @@ static XLogRecPtr RedoStartLSN = InvalidXLogRecPtr; * These structs are identical but are declared separately to indicate their * slightly different functions. * - * To read XLogCtl->LogwrtResult, you must hold either info_lck or - * WALWriteLock. To update it, you need to hold both locks. The point of - * this arrangement is that the value can be examined by code that already - * holds WALWriteLock without needing to grab info_lck as well. In addition - * to the shared variable, each backend has a private copy of LogwrtResult, - * which is updated when convenient. + * XLogCtl->LogwrtResult is read and written 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. * * 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. @@ -433,6 +430,12 @@ static XLogRecPtr RedoStartLSN = InvalidXLogRecPtr; *---------- */ +typedef struct XLogwrtAtomic +{ + pg_atomic_uint64 Write; /* last byte + 1 of write position */ + pg_atomic_uint64 Flush; /* last byte + 1 of flush position */ +} XLogwrtAtomic; + typedef struct XLogwrtRqst { XLogRecPtr Write; /* last byte + 1 to write out */ @@ -596,6 +599,7 @@ typedef struct XLogCtlData { XLogCtlInsert Insert; + XLogwrtAtomic LogwrtResult; /* Protected by info_lck: */ XLogwrtRqst LogwrtRqst; XLogRecPtr RedoRecPtr; /* a recent copy of Insert->RedoRecPtr */ @@ -613,12 +617,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). * @@ -1172,9 +1170,10 @@ XLogInsertRecord(XLogRecData *rdata, /* 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); + /* update local result copy */ + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); } /* @@ -2170,13 +2169,13 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic) SpinLockAcquire(&XLogCtl->info_lck); if (XLogCtl->LogwrtRqst.Write < OldPageRqstPtr) XLogCtl->LogwrtRqst.Write = OldPageRqstPtr; - LogwrtResult = XLogCtl->LogwrtResult; SpinLockRelease(&XLogCtl->info_lck); /* * Now that we have an up-to-date LogwrtResult value, 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) { /* @@ -2191,7 +2190,7 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic) LWLockAcquire(WALWriteLock, LW_EXCLUSIVE); - LogwrtResult = XLogCtl->LogwrtResult; + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); if (LogwrtResult.Write >= OldPageRqstPtr) { /* OK, someone wrote it already */ @@ -2438,7 +2437,7 @@ 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); /* * Since successive pages in the xlog cache are consecutively allocated, @@ -2636,7 +2635,6 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) */ if (LogwrtResult.Flush < WriteRqst.Flush && LogwrtResult.Flush < LogwrtResult.Write) - { /* * Could get here without iterating above loop, in which case we might @@ -2675,8 +2673,10 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) * code in a couple of places. */ { + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Write, LogwrtResult.Write); + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Flush, LogwrtResult.Flush); + pg_memory_barrier(); SpinLockAcquire(&XLogCtl->info_lck); - XLogCtl->LogwrtResult = LogwrtResult; if (XLogCtl->LogwrtRqst.Write < LogwrtResult.Write) XLogCtl->LogwrtRqst.Write = LogwrtResult.Write; if (XLogCtl->LogwrtRqst.Flush < LogwrtResult.Flush) @@ -2696,8 +2696,8 @@ XLogSetAsyncXactLSN(XLogRecPtr asyncXactLSN) XLogRecPtr WriteRqstPtr = asyncXactLSN; bool sleeping; + 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; @@ -2908,10 +2908,10 @@ XLogFlush(XLogRecPtr record) SpinLockAcquire(&XLogCtl->info_lck); if (WriteRqstPtr < XLogCtl->LogwrtRqst.Write) WriteRqstPtr = XLogCtl->LogwrtRqst.Write; - LogwrtResult = XLogCtl->LogwrtResult; SpinLockRelease(&XLogCtl->info_lck); /* done already? */ + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); if (record <= LogwrtResult.Flush) break; @@ -2939,7 +2939,7 @@ XLogFlush(XLogRecPtr record) } /* Got the lock; recheck whether request is satisfied */ - LogwrtResult = XLogCtl->LogwrtResult; + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); if (record <= LogwrtResult.Flush) { LWLockRelease(WALWriteLock); @@ -3056,7 +3056,6 @@ XLogBackgroundFlush(void) /* read LogwrtResult and update local state */ SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; WriteRqst = XLogCtl->LogwrtRqst; SpinLockRelease(&XLogCtl->info_lck); @@ -3064,8 +3063,10 @@ 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) { + pg_memory_barrier(); SpinLockAcquire(&XLogCtl->info_lck); WriteRqst.Write = XLogCtl->asyncXactLSN; SpinLockRelease(&XLogCtl->info_lck); @@ -3081,6 +3082,7 @@ XLogBackgroundFlush(void) { if (openLogFile >= 0) { + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); if (!XLByteInPrevSeg(LogwrtResult.Write, openLogSegNo, wal_segment_size)) { @@ -3140,7 +3142,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 +3231,7 @@ XLogNeedsFlush(XLogRecPtr record) return false; /* read LogwrtResult and update local state */ - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); /* check again */ if (record <= LogwrtResult.Flush) @@ -7779,7 +7780,8 @@ StartupXLOG(void) LogwrtResult.Write = LogwrtResult.Flush = EndOfLog; - XLogCtl->LogwrtResult = LogwrtResult; + pg_atomic_write_u64(&XLogCtl->LogwrtResult.Write, EndOfLog); + pg_atomic_write_u64(&XLogCtl->LogwrtResult.Flush, EndOfLog); XLogCtl->LogwrtRqst.Write = EndOfLog; XLogCtl->LogwrtRqst.Flush = EndOfLog; @@ -8498,9 +8500,7 @@ GetInsertRecPtr(void) XLogRecPtr GetFlushRecPtr(void) { - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); return LogwrtResult.Flush; } @@ -11649,9 +11649,7 @@ 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); return LogwrtResult.Write; } -- 2.20.1 --0OAP2g/MAC+5xKAE-- ^ permalink raw reply [nested|flat] 25+ messages in thread
* [PATCH 2/2] make LogwrtResult atomic @ 2021-02-02 17:03 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 25+ messages in thread From: Alvaro Herrera @ 2021-02-02 17:03 UTC (permalink / raw) --- src/backend/access/transam/xlog.c | 66 +++++++++++++++---------------- 1 file changed, 32 insertions(+), 34 deletions(-) diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index f03bd473e2..10802bd56f 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -405,12 +405,9 @@ static XLogRecPtr RedoStartLSN = InvalidXLogRecPtr; * These structs are identical but are declared separately to indicate their * slightly different functions. * - * To read XLogCtl->LogwrtResult, you must hold either info_lck or - * WALWriteLock. To update it, you need to hold both locks. The point of - * this arrangement is that the value can be examined by code that already - * holds WALWriteLock without needing to grab info_lck as well. In addition - * to the shared variable, each backend has a private copy of LogwrtResult, - * which is updated when convenient. + * XLogCtl->LogwrtResult is read and written 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. * * 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. @@ -433,6 +430,12 @@ static XLogRecPtr RedoStartLSN = InvalidXLogRecPtr; *---------- */ +typedef struct XLogwrtAtomic +{ + pg_atomic_uint64 Write; /* last byte + 1 of write position */ + pg_atomic_uint64 Flush; /* last byte + 1 of flush position */ +} XLogwrtAtomic; + typedef struct XLogwrtRqst { XLogRecPtr Write; /* last byte + 1 to write out */ @@ -596,6 +599,7 @@ typedef struct XLogCtlData { XLogCtlInsert Insert; + XLogwrtAtomic LogwrtResult; /* Protected by info_lck: */ XLogwrtRqst LogwrtRqst; XLogRecPtr RedoRecPtr; /* a recent copy of Insert->RedoRecPtr */ @@ -613,12 +617,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). * @@ -1172,9 +1170,10 @@ XLogInsertRecord(XLogRecData *rdata, /* 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); + /* update local result copy */ + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); } /* @@ -2170,13 +2169,13 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic) SpinLockAcquire(&XLogCtl->info_lck); if (XLogCtl->LogwrtRqst.Write < OldPageRqstPtr) XLogCtl->LogwrtRqst.Write = OldPageRqstPtr; - LogwrtResult = XLogCtl->LogwrtResult; SpinLockRelease(&XLogCtl->info_lck); /* * Now that we have an up-to-date LogwrtResult value, 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) { /* @@ -2191,7 +2190,7 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic) LWLockAcquire(WALWriteLock, LW_EXCLUSIVE); - LogwrtResult = XLogCtl->LogwrtResult; + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); if (LogwrtResult.Write >= OldPageRqstPtr) { /* OK, someone wrote it already */ @@ -2438,7 +2437,7 @@ 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); /* * Since successive pages in the xlog cache are consecutively allocated, @@ -2636,7 +2635,6 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) */ if (LogwrtResult.Flush < WriteRqst.Flush && LogwrtResult.Flush < LogwrtResult.Write) - { /* * Could get here without iterating above loop, in which case we might @@ -2675,8 +2673,10 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) * code in a couple of places. */ { + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Write, LogwrtResult.Write); + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Flush, LogwrtResult.Flush); + pg_memory_barrier(); SpinLockAcquire(&XLogCtl->info_lck); - XLogCtl->LogwrtResult = LogwrtResult; if (XLogCtl->LogwrtRqst.Write < LogwrtResult.Write) XLogCtl->LogwrtRqst.Write = LogwrtResult.Write; if (XLogCtl->LogwrtRqst.Flush < LogwrtResult.Flush) @@ -2696,8 +2696,8 @@ XLogSetAsyncXactLSN(XLogRecPtr asyncXactLSN) XLogRecPtr WriteRqstPtr = asyncXactLSN; bool sleeping; + 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; @@ -2908,10 +2908,10 @@ XLogFlush(XLogRecPtr record) SpinLockAcquire(&XLogCtl->info_lck); if (WriteRqstPtr < XLogCtl->LogwrtRqst.Write) WriteRqstPtr = XLogCtl->LogwrtRqst.Write; - LogwrtResult = XLogCtl->LogwrtResult; SpinLockRelease(&XLogCtl->info_lck); /* done already? */ + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); if (record <= LogwrtResult.Flush) break; @@ -2939,7 +2939,7 @@ XLogFlush(XLogRecPtr record) } /* Got the lock; recheck whether request is satisfied */ - LogwrtResult = XLogCtl->LogwrtResult; + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); if (record <= LogwrtResult.Flush) { LWLockRelease(WALWriteLock); @@ -3056,7 +3056,6 @@ XLogBackgroundFlush(void) /* read LogwrtResult and update local state */ SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; WriteRqst = XLogCtl->LogwrtRqst; SpinLockRelease(&XLogCtl->info_lck); @@ -3064,8 +3063,10 @@ 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) { + pg_memory_barrier(); SpinLockAcquire(&XLogCtl->info_lck); WriteRqst.Write = XLogCtl->asyncXactLSN; SpinLockRelease(&XLogCtl->info_lck); @@ -3081,6 +3082,7 @@ XLogBackgroundFlush(void) { if (openLogFile >= 0) { + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); if (!XLByteInPrevSeg(LogwrtResult.Write, openLogSegNo, wal_segment_size)) { @@ -3140,7 +3142,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 +3231,7 @@ XLogNeedsFlush(XLogRecPtr record) return false; /* read LogwrtResult and update local state */ - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); /* check again */ if (record <= LogwrtResult.Flush) @@ -7779,7 +7780,8 @@ StartupXLOG(void) LogwrtResult.Write = LogwrtResult.Flush = EndOfLog; - XLogCtl->LogwrtResult = LogwrtResult; + pg_atomic_write_u64(&XLogCtl->LogwrtResult.Write, EndOfLog); + pg_atomic_write_u64(&XLogCtl->LogwrtResult.Flush, EndOfLog); XLogCtl->LogwrtRqst.Write = EndOfLog; XLogCtl->LogwrtRqst.Flush = EndOfLog; @@ -8498,9 +8500,7 @@ GetInsertRecPtr(void) XLogRecPtr GetFlushRecPtr(void) { - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); return LogwrtResult.Flush; } @@ -11649,9 +11649,7 @@ 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); return LogwrtResult.Write; } -- 2.20.1 --0OAP2g/MAC+5xKAE-- ^ permalink raw reply [nested|flat] 25+ messages in thread
* [PATCH 2/2] make LogwrtResult atomic @ 2021-02-02 17:03 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 25+ messages in thread From: Alvaro Herrera @ 2021-02-02 17:03 UTC (permalink / raw) --- src/backend/access/transam/xlog.c | 66 +++++++++++++++---------------- 1 file changed, 32 insertions(+), 34 deletions(-) diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index f03bd473e2..10802bd56f 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -405,12 +405,9 @@ static XLogRecPtr RedoStartLSN = InvalidXLogRecPtr; * These structs are identical but are declared separately to indicate their * slightly different functions. * - * To read XLogCtl->LogwrtResult, you must hold either info_lck or - * WALWriteLock. To update it, you need to hold both locks. The point of - * this arrangement is that the value can be examined by code that already - * holds WALWriteLock without needing to grab info_lck as well. In addition - * to the shared variable, each backend has a private copy of LogwrtResult, - * which is updated when convenient. + * XLogCtl->LogwrtResult is read and written 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. * * 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. @@ -433,6 +430,12 @@ static XLogRecPtr RedoStartLSN = InvalidXLogRecPtr; *---------- */ +typedef struct XLogwrtAtomic +{ + pg_atomic_uint64 Write; /* last byte + 1 of write position */ + pg_atomic_uint64 Flush; /* last byte + 1 of flush position */ +} XLogwrtAtomic; + typedef struct XLogwrtRqst { XLogRecPtr Write; /* last byte + 1 to write out */ @@ -596,6 +599,7 @@ typedef struct XLogCtlData { XLogCtlInsert Insert; + XLogwrtAtomic LogwrtResult; /* Protected by info_lck: */ XLogwrtRqst LogwrtRqst; XLogRecPtr RedoRecPtr; /* a recent copy of Insert->RedoRecPtr */ @@ -613,12 +617,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). * @@ -1172,9 +1170,10 @@ XLogInsertRecord(XLogRecData *rdata, /* 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); + /* update local result copy */ + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); } /* @@ -2170,13 +2169,13 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic) SpinLockAcquire(&XLogCtl->info_lck); if (XLogCtl->LogwrtRqst.Write < OldPageRqstPtr) XLogCtl->LogwrtRqst.Write = OldPageRqstPtr; - LogwrtResult = XLogCtl->LogwrtResult; SpinLockRelease(&XLogCtl->info_lck); /* * Now that we have an up-to-date LogwrtResult value, 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) { /* @@ -2191,7 +2190,7 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic) LWLockAcquire(WALWriteLock, LW_EXCLUSIVE); - LogwrtResult = XLogCtl->LogwrtResult; + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); if (LogwrtResult.Write >= OldPageRqstPtr) { /* OK, someone wrote it already */ @@ -2438,7 +2437,7 @@ 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); /* * Since successive pages in the xlog cache are consecutively allocated, @@ -2636,7 +2635,6 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) */ if (LogwrtResult.Flush < WriteRqst.Flush && LogwrtResult.Flush < LogwrtResult.Write) - { /* * Could get here without iterating above loop, in which case we might @@ -2675,8 +2673,10 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) * code in a couple of places. */ { + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Write, LogwrtResult.Write); + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Flush, LogwrtResult.Flush); + pg_memory_barrier(); SpinLockAcquire(&XLogCtl->info_lck); - XLogCtl->LogwrtResult = LogwrtResult; if (XLogCtl->LogwrtRqst.Write < LogwrtResult.Write) XLogCtl->LogwrtRqst.Write = LogwrtResult.Write; if (XLogCtl->LogwrtRqst.Flush < LogwrtResult.Flush) @@ -2696,8 +2696,8 @@ XLogSetAsyncXactLSN(XLogRecPtr asyncXactLSN) XLogRecPtr WriteRqstPtr = asyncXactLSN; bool sleeping; + 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; @@ -2908,10 +2908,10 @@ XLogFlush(XLogRecPtr record) SpinLockAcquire(&XLogCtl->info_lck); if (WriteRqstPtr < XLogCtl->LogwrtRqst.Write) WriteRqstPtr = XLogCtl->LogwrtRqst.Write; - LogwrtResult = XLogCtl->LogwrtResult; SpinLockRelease(&XLogCtl->info_lck); /* done already? */ + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); if (record <= LogwrtResult.Flush) break; @@ -2939,7 +2939,7 @@ XLogFlush(XLogRecPtr record) } /* Got the lock; recheck whether request is satisfied */ - LogwrtResult = XLogCtl->LogwrtResult; + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); if (record <= LogwrtResult.Flush) { LWLockRelease(WALWriteLock); @@ -3056,7 +3056,6 @@ XLogBackgroundFlush(void) /* read LogwrtResult and update local state */ SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; WriteRqst = XLogCtl->LogwrtRqst; SpinLockRelease(&XLogCtl->info_lck); @@ -3064,8 +3063,10 @@ 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) { + pg_memory_barrier(); SpinLockAcquire(&XLogCtl->info_lck); WriteRqst.Write = XLogCtl->asyncXactLSN; SpinLockRelease(&XLogCtl->info_lck); @@ -3081,6 +3082,7 @@ XLogBackgroundFlush(void) { if (openLogFile >= 0) { + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); if (!XLByteInPrevSeg(LogwrtResult.Write, openLogSegNo, wal_segment_size)) { @@ -3140,7 +3142,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 +3231,7 @@ XLogNeedsFlush(XLogRecPtr record) return false; /* read LogwrtResult and update local state */ - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); /* check again */ if (record <= LogwrtResult.Flush) @@ -7779,7 +7780,8 @@ StartupXLOG(void) LogwrtResult.Write = LogwrtResult.Flush = EndOfLog; - XLogCtl->LogwrtResult = LogwrtResult; + pg_atomic_write_u64(&XLogCtl->LogwrtResult.Write, EndOfLog); + pg_atomic_write_u64(&XLogCtl->LogwrtResult.Flush, EndOfLog); XLogCtl->LogwrtRqst.Write = EndOfLog; XLogCtl->LogwrtRqst.Flush = EndOfLog; @@ -8498,9 +8500,7 @@ GetInsertRecPtr(void) XLogRecPtr GetFlushRecPtr(void) { - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); return LogwrtResult.Flush; } @@ -11649,9 +11649,7 @@ 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); return LogwrtResult.Write; } -- 2.20.1 --0OAP2g/MAC+5xKAE-- ^ permalink raw reply [nested|flat] 25+ messages in thread
* [PATCH 2/2] make LogwrtResult atomic @ 2021-02-02 17:03 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 25+ messages in thread From: Alvaro Herrera @ 2021-02-02 17:03 UTC (permalink / raw) --- src/backend/access/transam/xlog.c | 66 +++++++++++++++---------------- 1 file changed, 32 insertions(+), 34 deletions(-) diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index f03bd473e2..10802bd56f 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -405,12 +405,9 @@ static XLogRecPtr RedoStartLSN = InvalidXLogRecPtr; * These structs are identical but are declared separately to indicate their * slightly different functions. * - * To read XLogCtl->LogwrtResult, you must hold either info_lck or - * WALWriteLock. To update it, you need to hold both locks. The point of - * this arrangement is that the value can be examined by code that already - * holds WALWriteLock without needing to grab info_lck as well. In addition - * to the shared variable, each backend has a private copy of LogwrtResult, - * which is updated when convenient. + * XLogCtl->LogwrtResult is read and written 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. * * 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. @@ -433,6 +430,12 @@ static XLogRecPtr RedoStartLSN = InvalidXLogRecPtr; *---------- */ +typedef struct XLogwrtAtomic +{ + pg_atomic_uint64 Write; /* last byte + 1 of write position */ + pg_atomic_uint64 Flush; /* last byte + 1 of flush position */ +} XLogwrtAtomic; + typedef struct XLogwrtRqst { XLogRecPtr Write; /* last byte + 1 to write out */ @@ -596,6 +599,7 @@ typedef struct XLogCtlData { XLogCtlInsert Insert; + XLogwrtAtomic LogwrtResult; /* Protected by info_lck: */ XLogwrtRqst LogwrtRqst; XLogRecPtr RedoRecPtr; /* a recent copy of Insert->RedoRecPtr */ @@ -613,12 +617,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). * @@ -1172,9 +1170,10 @@ XLogInsertRecord(XLogRecData *rdata, /* 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); + /* update local result copy */ + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); } /* @@ -2170,13 +2169,13 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic) SpinLockAcquire(&XLogCtl->info_lck); if (XLogCtl->LogwrtRqst.Write < OldPageRqstPtr) XLogCtl->LogwrtRqst.Write = OldPageRqstPtr; - LogwrtResult = XLogCtl->LogwrtResult; SpinLockRelease(&XLogCtl->info_lck); /* * Now that we have an up-to-date LogwrtResult value, 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) { /* @@ -2191,7 +2190,7 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic) LWLockAcquire(WALWriteLock, LW_EXCLUSIVE); - LogwrtResult = XLogCtl->LogwrtResult; + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); if (LogwrtResult.Write >= OldPageRqstPtr) { /* OK, someone wrote it already */ @@ -2438,7 +2437,7 @@ 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); /* * Since successive pages in the xlog cache are consecutively allocated, @@ -2636,7 +2635,6 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) */ if (LogwrtResult.Flush < WriteRqst.Flush && LogwrtResult.Flush < LogwrtResult.Write) - { /* * Could get here without iterating above loop, in which case we might @@ -2675,8 +2673,10 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) * code in a couple of places. */ { + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Write, LogwrtResult.Write); + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Flush, LogwrtResult.Flush); + pg_memory_barrier(); SpinLockAcquire(&XLogCtl->info_lck); - XLogCtl->LogwrtResult = LogwrtResult; if (XLogCtl->LogwrtRqst.Write < LogwrtResult.Write) XLogCtl->LogwrtRqst.Write = LogwrtResult.Write; if (XLogCtl->LogwrtRqst.Flush < LogwrtResult.Flush) @@ -2696,8 +2696,8 @@ XLogSetAsyncXactLSN(XLogRecPtr asyncXactLSN) XLogRecPtr WriteRqstPtr = asyncXactLSN; bool sleeping; + 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; @@ -2908,10 +2908,10 @@ XLogFlush(XLogRecPtr record) SpinLockAcquire(&XLogCtl->info_lck); if (WriteRqstPtr < XLogCtl->LogwrtRqst.Write) WriteRqstPtr = XLogCtl->LogwrtRqst.Write; - LogwrtResult = XLogCtl->LogwrtResult; SpinLockRelease(&XLogCtl->info_lck); /* done already? */ + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); if (record <= LogwrtResult.Flush) break; @@ -2939,7 +2939,7 @@ XLogFlush(XLogRecPtr record) } /* Got the lock; recheck whether request is satisfied */ - LogwrtResult = XLogCtl->LogwrtResult; + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); if (record <= LogwrtResult.Flush) { LWLockRelease(WALWriteLock); @@ -3056,7 +3056,6 @@ XLogBackgroundFlush(void) /* read LogwrtResult and update local state */ SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; WriteRqst = XLogCtl->LogwrtRqst; SpinLockRelease(&XLogCtl->info_lck); @@ -3064,8 +3063,10 @@ 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) { + pg_memory_barrier(); SpinLockAcquire(&XLogCtl->info_lck); WriteRqst.Write = XLogCtl->asyncXactLSN; SpinLockRelease(&XLogCtl->info_lck); @@ -3081,6 +3082,7 @@ XLogBackgroundFlush(void) { if (openLogFile >= 0) { + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); if (!XLByteInPrevSeg(LogwrtResult.Write, openLogSegNo, wal_segment_size)) { @@ -3140,7 +3142,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 +3231,7 @@ XLogNeedsFlush(XLogRecPtr record) return false; /* read LogwrtResult and update local state */ - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); /* check again */ if (record <= LogwrtResult.Flush) @@ -7779,7 +7780,8 @@ StartupXLOG(void) LogwrtResult.Write = LogwrtResult.Flush = EndOfLog; - XLogCtl->LogwrtResult = LogwrtResult; + pg_atomic_write_u64(&XLogCtl->LogwrtResult.Write, EndOfLog); + pg_atomic_write_u64(&XLogCtl->LogwrtResult.Flush, EndOfLog); XLogCtl->LogwrtRqst.Write = EndOfLog; XLogCtl->LogwrtRqst.Flush = EndOfLog; @@ -8498,9 +8500,7 @@ GetInsertRecPtr(void) XLogRecPtr GetFlushRecPtr(void) { - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); return LogwrtResult.Flush; } @@ -11649,9 +11649,7 @@ 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); return LogwrtResult.Write; } -- 2.20.1 --0OAP2g/MAC+5xKAE-- ^ permalink raw reply [nested|flat] 25+ messages in thread
* [PATCH 2/2] make LogwrtResult atomic @ 2021-02-02 17:03 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 25+ messages in thread From: Alvaro Herrera @ 2021-02-02 17:03 UTC (permalink / raw) --- src/backend/access/transam/xlog.c | 66 +++++++++++++++---------------- 1 file changed, 32 insertions(+), 34 deletions(-) diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index f03bd473e2..10802bd56f 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -405,12 +405,9 @@ static XLogRecPtr RedoStartLSN = InvalidXLogRecPtr; * These structs are identical but are declared separately to indicate their * slightly different functions. * - * To read XLogCtl->LogwrtResult, you must hold either info_lck or - * WALWriteLock. To update it, you need to hold both locks. The point of - * this arrangement is that the value can be examined by code that already - * holds WALWriteLock without needing to grab info_lck as well. In addition - * to the shared variable, each backend has a private copy of LogwrtResult, - * which is updated when convenient. + * XLogCtl->LogwrtResult is read and written 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. * * 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. @@ -433,6 +430,12 @@ static XLogRecPtr RedoStartLSN = InvalidXLogRecPtr; *---------- */ +typedef struct XLogwrtAtomic +{ + pg_atomic_uint64 Write; /* last byte + 1 of write position */ + pg_atomic_uint64 Flush; /* last byte + 1 of flush position */ +} XLogwrtAtomic; + typedef struct XLogwrtRqst { XLogRecPtr Write; /* last byte + 1 to write out */ @@ -596,6 +599,7 @@ typedef struct XLogCtlData { XLogCtlInsert Insert; + XLogwrtAtomic LogwrtResult; /* Protected by info_lck: */ XLogwrtRqst LogwrtRqst; XLogRecPtr RedoRecPtr; /* a recent copy of Insert->RedoRecPtr */ @@ -613,12 +617,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). * @@ -1172,9 +1170,10 @@ XLogInsertRecord(XLogRecData *rdata, /* 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); + /* update local result copy */ + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); } /* @@ -2170,13 +2169,13 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic) SpinLockAcquire(&XLogCtl->info_lck); if (XLogCtl->LogwrtRqst.Write < OldPageRqstPtr) XLogCtl->LogwrtRqst.Write = OldPageRqstPtr; - LogwrtResult = XLogCtl->LogwrtResult; SpinLockRelease(&XLogCtl->info_lck); /* * Now that we have an up-to-date LogwrtResult value, 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) { /* @@ -2191,7 +2190,7 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic) LWLockAcquire(WALWriteLock, LW_EXCLUSIVE); - LogwrtResult = XLogCtl->LogwrtResult; + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); if (LogwrtResult.Write >= OldPageRqstPtr) { /* OK, someone wrote it already */ @@ -2438,7 +2437,7 @@ 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); /* * Since successive pages in the xlog cache are consecutively allocated, @@ -2636,7 +2635,6 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) */ if (LogwrtResult.Flush < WriteRqst.Flush && LogwrtResult.Flush < LogwrtResult.Write) - { /* * Could get here without iterating above loop, in which case we might @@ -2675,8 +2673,10 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) * code in a couple of places. */ { + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Write, LogwrtResult.Write); + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Flush, LogwrtResult.Flush); + pg_memory_barrier(); SpinLockAcquire(&XLogCtl->info_lck); - XLogCtl->LogwrtResult = LogwrtResult; if (XLogCtl->LogwrtRqst.Write < LogwrtResult.Write) XLogCtl->LogwrtRqst.Write = LogwrtResult.Write; if (XLogCtl->LogwrtRqst.Flush < LogwrtResult.Flush) @@ -2696,8 +2696,8 @@ XLogSetAsyncXactLSN(XLogRecPtr asyncXactLSN) XLogRecPtr WriteRqstPtr = asyncXactLSN; bool sleeping; + 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; @@ -2908,10 +2908,10 @@ XLogFlush(XLogRecPtr record) SpinLockAcquire(&XLogCtl->info_lck); if (WriteRqstPtr < XLogCtl->LogwrtRqst.Write) WriteRqstPtr = XLogCtl->LogwrtRqst.Write; - LogwrtResult = XLogCtl->LogwrtResult; SpinLockRelease(&XLogCtl->info_lck); /* done already? */ + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); if (record <= LogwrtResult.Flush) break; @@ -2939,7 +2939,7 @@ XLogFlush(XLogRecPtr record) } /* Got the lock; recheck whether request is satisfied */ - LogwrtResult = XLogCtl->LogwrtResult; + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); if (record <= LogwrtResult.Flush) { LWLockRelease(WALWriteLock); @@ -3056,7 +3056,6 @@ XLogBackgroundFlush(void) /* read LogwrtResult and update local state */ SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; WriteRqst = XLogCtl->LogwrtRqst; SpinLockRelease(&XLogCtl->info_lck); @@ -3064,8 +3063,10 @@ 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) { + pg_memory_barrier(); SpinLockAcquire(&XLogCtl->info_lck); WriteRqst.Write = XLogCtl->asyncXactLSN; SpinLockRelease(&XLogCtl->info_lck); @@ -3081,6 +3082,7 @@ XLogBackgroundFlush(void) { if (openLogFile >= 0) { + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); if (!XLByteInPrevSeg(LogwrtResult.Write, openLogSegNo, wal_segment_size)) { @@ -3140,7 +3142,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 +3231,7 @@ XLogNeedsFlush(XLogRecPtr record) return false; /* read LogwrtResult and update local state */ - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); /* check again */ if (record <= LogwrtResult.Flush) @@ -7779,7 +7780,8 @@ StartupXLOG(void) LogwrtResult.Write = LogwrtResult.Flush = EndOfLog; - XLogCtl->LogwrtResult = LogwrtResult; + pg_atomic_write_u64(&XLogCtl->LogwrtResult.Write, EndOfLog); + pg_atomic_write_u64(&XLogCtl->LogwrtResult.Flush, EndOfLog); XLogCtl->LogwrtRqst.Write = EndOfLog; XLogCtl->LogwrtRqst.Flush = EndOfLog; @@ -8498,9 +8500,7 @@ GetInsertRecPtr(void) XLogRecPtr GetFlushRecPtr(void) { - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); return LogwrtResult.Flush; } @@ -11649,9 +11649,7 @@ 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); return LogwrtResult.Write; } -- 2.20.1 --0OAP2g/MAC+5xKAE-- ^ permalink raw reply [nested|flat] 25+ messages in thread
* [PATCH 2/2] make LogwrtResult atomic @ 2021-02-02 17:03 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 25+ messages in thread From: Alvaro Herrera @ 2021-02-02 17:03 UTC (permalink / raw) --- src/backend/access/transam/xlog.c | 66 +++++++++++++++---------------- 1 file changed, 32 insertions(+), 34 deletions(-) diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index f03bd473e2..10802bd56f 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -405,12 +405,9 @@ static XLogRecPtr RedoStartLSN = InvalidXLogRecPtr; * These structs are identical but are declared separately to indicate their * slightly different functions. * - * To read XLogCtl->LogwrtResult, you must hold either info_lck or - * WALWriteLock. To update it, you need to hold both locks. The point of - * this arrangement is that the value can be examined by code that already - * holds WALWriteLock without needing to grab info_lck as well. In addition - * to the shared variable, each backend has a private copy of LogwrtResult, - * which is updated when convenient. + * XLogCtl->LogwrtResult is read and written 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. * * 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. @@ -433,6 +430,12 @@ static XLogRecPtr RedoStartLSN = InvalidXLogRecPtr; *---------- */ +typedef struct XLogwrtAtomic +{ + pg_atomic_uint64 Write; /* last byte + 1 of write position */ + pg_atomic_uint64 Flush; /* last byte + 1 of flush position */ +} XLogwrtAtomic; + typedef struct XLogwrtRqst { XLogRecPtr Write; /* last byte + 1 to write out */ @@ -596,6 +599,7 @@ typedef struct XLogCtlData { XLogCtlInsert Insert; + XLogwrtAtomic LogwrtResult; /* Protected by info_lck: */ XLogwrtRqst LogwrtRqst; XLogRecPtr RedoRecPtr; /* a recent copy of Insert->RedoRecPtr */ @@ -613,12 +617,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). * @@ -1172,9 +1170,10 @@ XLogInsertRecord(XLogRecData *rdata, /* 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); + /* update local result copy */ + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); } /* @@ -2170,13 +2169,13 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic) SpinLockAcquire(&XLogCtl->info_lck); if (XLogCtl->LogwrtRqst.Write < OldPageRqstPtr) XLogCtl->LogwrtRqst.Write = OldPageRqstPtr; - LogwrtResult = XLogCtl->LogwrtResult; SpinLockRelease(&XLogCtl->info_lck); /* * Now that we have an up-to-date LogwrtResult value, 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) { /* @@ -2191,7 +2190,7 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic) LWLockAcquire(WALWriteLock, LW_EXCLUSIVE); - LogwrtResult = XLogCtl->LogwrtResult; + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); if (LogwrtResult.Write >= OldPageRqstPtr) { /* OK, someone wrote it already */ @@ -2438,7 +2437,7 @@ 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); /* * Since successive pages in the xlog cache are consecutively allocated, @@ -2636,7 +2635,6 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) */ if (LogwrtResult.Flush < WriteRqst.Flush && LogwrtResult.Flush < LogwrtResult.Write) - { /* * Could get here without iterating above loop, in which case we might @@ -2675,8 +2673,10 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) * code in a couple of places. */ { + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Write, LogwrtResult.Write); + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Flush, LogwrtResult.Flush); + pg_memory_barrier(); SpinLockAcquire(&XLogCtl->info_lck); - XLogCtl->LogwrtResult = LogwrtResult; if (XLogCtl->LogwrtRqst.Write < LogwrtResult.Write) XLogCtl->LogwrtRqst.Write = LogwrtResult.Write; if (XLogCtl->LogwrtRqst.Flush < LogwrtResult.Flush) @@ -2696,8 +2696,8 @@ XLogSetAsyncXactLSN(XLogRecPtr asyncXactLSN) XLogRecPtr WriteRqstPtr = asyncXactLSN; bool sleeping; + 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; @@ -2908,10 +2908,10 @@ XLogFlush(XLogRecPtr record) SpinLockAcquire(&XLogCtl->info_lck); if (WriteRqstPtr < XLogCtl->LogwrtRqst.Write) WriteRqstPtr = XLogCtl->LogwrtRqst.Write; - LogwrtResult = XLogCtl->LogwrtResult; SpinLockRelease(&XLogCtl->info_lck); /* done already? */ + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); if (record <= LogwrtResult.Flush) break; @@ -2939,7 +2939,7 @@ XLogFlush(XLogRecPtr record) } /* Got the lock; recheck whether request is satisfied */ - LogwrtResult = XLogCtl->LogwrtResult; + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); if (record <= LogwrtResult.Flush) { LWLockRelease(WALWriteLock); @@ -3056,7 +3056,6 @@ XLogBackgroundFlush(void) /* read LogwrtResult and update local state */ SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; WriteRqst = XLogCtl->LogwrtRqst; SpinLockRelease(&XLogCtl->info_lck); @@ -3064,8 +3063,10 @@ 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) { + pg_memory_barrier(); SpinLockAcquire(&XLogCtl->info_lck); WriteRqst.Write = XLogCtl->asyncXactLSN; SpinLockRelease(&XLogCtl->info_lck); @@ -3081,6 +3082,7 @@ XLogBackgroundFlush(void) { if (openLogFile >= 0) { + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); if (!XLByteInPrevSeg(LogwrtResult.Write, openLogSegNo, wal_segment_size)) { @@ -3140,7 +3142,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 +3231,7 @@ XLogNeedsFlush(XLogRecPtr record) return false; /* read LogwrtResult and update local state */ - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); /* check again */ if (record <= LogwrtResult.Flush) @@ -7779,7 +7780,8 @@ StartupXLOG(void) LogwrtResult.Write = LogwrtResult.Flush = EndOfLog; - XLogCtl->LogwrtResult = LogwrtResult; + pg_atomic_write_u64(&XLogCtl->LogwrtResult.Write, EndOfLog); + pg_atomic_write_u64(&XLogCtl->LogwrtResult.Flush, EndOfLog); XLogCtl->LogwrtRqst.Write = EndOfLog; XLogCtl->LogwrtRqst.Flush = EndOfLog; @@ -8498,9 +8500,7 @@ GetInsertRecPtr(void) XLogRecPtr GetFlushRecPtr(void) { - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); return LogwrtResult.Flush; } @@ -11649,9 +11649,7 @@ 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); return LogwrtResult.Write; } -- 2.20.1 --0OAP2g/MAC+5xKAE-- ^ permalink raw reply [nested|flat] 25+ messages in thread
* [PATCH 2/2] make LogwrtResult atomic @ 2021-02-02 17:03 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 25+ messages in thread From: Alvaro Herrera @ 2021-02-02 17:03 UTC (permalink / raw) --- src/backend/access/transam/xlog.c | 66 +++++++++++++++---------------- 1 file changed, 32 insertions(+), 34 deletions(-) diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index f03bd473e2..10802bd56f 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -405,12 +405,9 @@ static XLogRecPtr RedoStartLSN = InvalidXLogRecPtr; * These structs are identical but are declared separately to indicate their * slightly different functions. * - * To read XLogCtl->LogwrtResult, you must hold either info_lck or - * WALWriteLock. To update it, you need to hold both locks. The point of - * this arrangement is that the value can be examined by code that already - * holds WALWriteLock without needing to grab info_lck as well. In addition - * to the shared variable, each backend has a private copy of LogwrtResult, - * which is updated when convenient. + * XLogCtl->LogwrtResult is read and written 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. * * 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. @@ -433,6 +430,12 @@ static XLogRecPtr RedoStartLSN = InvalidXLogRecPtr; *---------- */ +typedef struct XLogwrtAtomic +{ + pg_atomic_uint64 Write; /* last byte + 1 of write position */ + pg_atomic_uint64 Flush; /* last byte + 1 of flush position */ +} XLogwrtAtomic; + typedef struct XLogwrtRqst { XLogRecPtr Write; /* last byte + 1 to write out */ @@ -596,6 +599,7 @@ typedef struct XLogCtlData { XLogCtlInsert Insert; + XLogwrtAtomic LogwrtResult; /* Protected by info_lck: */ XLogwrtRqst LogwrtRqst; XLogRecPtr RedoRecPtr; /* a recent copy of Insert->RedoRecPtr */ @@ -613,12 +617,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). * @@ -1172,9 +1170,10 @@ XLogInsertRecord(XLogRecData *rdata, /* 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); + /* update local result copy */ + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); } /* @@ -2170,13 +2169,13 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic) SpinLockAcquire(&XLogCtl->info_lck); if (XLogCtl->LogwrtRqst.Write < OldPageRqstPtr) XLogCtl->LogwrtRqst.Write = OldPageRqstPtr; - LogwrtResult = XLogCtl->LogwrtResult; SpinLockRelease(&XLogCtl->info_lck); /* * Now that we have an up-to-date LogwrtResult value, 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) { /* @@ -2191,7 +2190,7 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic) LWLockAcquire(WALWriteLock, LW_EXCLUSIVE); - LogwrtResult = XLogCtl->LogwrtResult; + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); if (LogwrtResult.Write >= OldPageRqstPtr) { /* OK, someone wrote it already */ @@ -2438,7 +2437,7 @@ 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); /* * Since successive pages in the xlog cache are consecutively allocated, @@ -2636,7 +2635,6 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) */ if (LogwrtResult.Flush < WriteRqst.Flush && LogwrtResult.Flush < LogwrtResult.Write) - { /* * Could get here without iterating above loop, in which case we might @@ -2675,8 +2673,10 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) * code in a couple of places. */ { + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Write, LogwrtResult.Write); + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Flush, LogwrtResult.Flush); + pg_memory_barrier(); SpinLockAcquire(&XLogCtl->info_lck); - XLogCtl->LogwrtResult = LogwrtResult; if (XLogCtl->LogwrtRqst.Write < LogwrtResult.Write) XLogCtl->LogwrtRqst.Write = LogwrtResult.Write; if (XLogCtl->LogwrtRqst.Flush < LogwrtResult.Flush) @@ -2696,8 +2696,8 @@ XLogSetAsyncXactLSN(XLogRecPtr asyncXactLSN) XLogRecPtr WriteRqstPtr = asyncXactLSN; bool sleeping; + 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; @@ -2908,10 +2908,10 @@ XLogFlush(XLogRecPtr record) SpinLockAcquire(&XLogCtl->info_lck); if (WriteRqstPtr < XLogCtl->LogwrtRqst.Write) WriteRqstPtr = XLogCtl->LogwrtRqst.Write; - LogwrtResult = XLogCtl->LogwrtResult; SpinLockRelease(&XLogCtl->info_lck); /* done already? */ + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); if (record <= LogwrtResult.Flush) break; @@ -2939,7 +2939,7 @@ XLogFlush(XLogRecPtr record) } /* Got the lock; recheck whether request is satisfied */ - LogwrtResult = XLogCtl->LogwrtResult; + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); if (record <= LogwrtResult.Flush) { LWLockRelease(WALWriteLock); @@ -3056,7 +3056,6 @@ XLogBackgroundFlush(void) /* read LogwrtResult and update local state */ SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; WriteRqst = XLogCtl->LogwrtRqst; SpinLockRelease(&XLogCtl->info_lck); @@ -3064,8 +3063,10 @@ 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) { + pg_memory_barrier(); SpinLockAcquire(&XLogCtl->info_lck); WriteRqst.Write = XLogCtl->asyncXactLSN; SpinLockRelease(&XLogCtl->info_lck); @@ -3081,6 +3082,7 @@ XLogBackgroundFlush(void) { if (openLogFile >= 0) { + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); if (!XLByteInPrevSeg(LogwrtResult.Write, openLogSegNo, wal_segment_size)) { @@ -3140,7 +3142,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 +3231,7 @@ XLogNeedsFlush(XLogRecPtr record) return false; /* read LogwrtResult and update local state */ - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); /* check again */ if (record <= LogwrtResult.Flush) @@ -7779,7 +7780,8 @@ StartupXLOG(void) LogwrtResult.Write = LogwrtResult.Flush = EndOfLog; - XLogCtl->LogwrtResult = LogwrtResult; + pg_atomic_write_u64(&XLogCtl->LogwrtResult.Write, EndOfLog); + pg_atomic_write_u64(&XLogCtl->LogwrtResult.Flush, EndOfLog); XLogCtl->LogwrtRqst.Write = EndOfLog; XLogCtl->LogwrtRqst.Flush = EndOfLog; @@ -8498,9 +8500,7 @@ GetInsertRecPtr(void) XLogRecPtr GetFlushRecPtr(void) { - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); return LogwrtResult.Flush; } @@ -11649,9 +11649,7 @@ 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); return LogwrtResult.Write; } -- 2.20.1 --0OAP2g/MAC+5xKAE-- ^ permalink raw reply [nested|flat] 25+ messages in thread
* [PATCH 2/2] make LogwrtResult atomic @ 2021-02-02 17:03 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 25+ messages in thread From: Alvaro Herrera @ 2021-02-02 17:03 UTC (permalink / raw) --- src/backend/access/transam/xlog.c | 66 +++++++++++++++---------------- 1 file changed, 32 insertions(+), 34 deletions(-) diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index f03bd473e2..10802bd56f 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -405,12 +405,9 @@ static XLogRecPtr RedoStartLSN = InvalidXLogRecPtr; * These structs are identical but are declared separately to indicate their * slightly different functions. * - * To read XLogCtl->LogwrtResult, you must hold either info_lck or - * WALWriteLock. To update it, you need to hold both locks. The point of - * this arrangement is that the value can be examined by code that already - * holds WALWriteLock without needing to grab info_lck as well. In addition - * to the shared variable, each backend has a private copy of LogwrtResult, - * which is updated when convenient. + * XLogCtl->LogwrtResult is read and written 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. * * 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. @@ -433,6 +430,12 @@ static XLogRecPtr RedoStartLSN = InvalidXLogRecPtr; *---------- */ +typedef struct XLogwrtAtomic +{ + pg_atomic_uint64 Write; /* last byte + 1 of write position */ + pg_atomic_uint64 Flush; /* last byte + 1 of flush position */ +} XLogwrtAtomic; + typedef struct XLogwrtRqst { XLogRecPtr Write; /* last byte + 1 to write out */ @@ -596,6 +599,7 @@ typedef struct XLogCtlData { XLogCtlInsert Insert; + XLogwrtAtomic LogwrtResult; /* Protected by info_lck: */ XLogwrtRqst LogwrtRqst; XLogRecPtr RedoRecPtr; /* a recent copy of Insert->RedoRecPtr */ @@ -613,12 +617,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). * @@ -1172,9 +1170,10 @@ XLogInsertRecord(XLogRecData *rdata, /* 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); + /* update local result copy */ + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); } /* @@ -2170,13 +2169,13 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic) SpinLockAcquire(&XLogCtl->info_lck); if (XLogCtl->LogwrtRqst.Write < OldPageRqstPtr) XLogCtl->LogwrtRqst.Write = OldPageRqstPtr; - LogwrtResult = XLogCtl->LogwrtResult; SpinLockRelease(&XLogCtl->info_lck); /* * Now that we have an up-to-date LogwrtResult value, 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) { /* @@ -2191,7 +2190,7 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic) LWLockAcquire(WALWriteLock, LW_EXCLUSIVE); - LogwrtResult = XLogCtl->LogwrtResult; + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); if (LogwrtResult.Write >= OldPageRqstPtr) { /* OK, someone wrote it already */ @@ -2438,7 +2437,7 @@ 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); /* * Since successive pages in the xlog cache are consecutively allocated, @@ -2636,7 +2635,6 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) */ if (LogwrtResult.Flush < WriteRqst.Flush && LogwrtResult.Flush < LogwrtResult.Write) - { /* * Could get here without iterating above loop, in which case we might @@ -2675,8 +2673,10 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) * code in a couple of places. */ { + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Write, LogwrtResult.Write); + pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Flush, LogwrtResult.Flush); + pg_memory_barrier(); SpinLockAcquire(&XLogCtl->info_lck); - XLogCtl->LogwrtResult = LogwrtResult; if (XLogCtl->LogwrtRqst.Write < LogwrtResult.Write) XLogCtl->LogwrtRqst.Write = LogwrtResult.Write; if (XLogCtl->LogwrtRqst.Flush < LogwrtResult.Flush) @@ -2696,8 +2696,8 @@ XLogSetAsyncXactLSN(XLogRecPtr asyncXactLSN) XLogRecPtr WriteRqstPtr = asyncXactLSN; bool sleeping; + 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; @@ -2908,10 +2908,10 @@ XLogFlush(XLogRecPtr record) SpinLockAcquire(&XLogCtl->info_lck); if (WriteRqstPtr < XLogCtl->LogwrtRqst.Write) WriteRqstPtr = XLogCtl->LogwrtRqst.Write; - LogwrtResult = XLogCtl->LogwrtResult; SpinLockRelease(&XLogCtl->info_lck); /* done already? */ + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); if (record <= LogwrtResult.Flush) break; @@ -2939,7 +2939,7 @@ XLogFlush(XLogRecPtr record) } /* Got the lock; recheck whether request is satisfied */ - LogwrtResult = XLogCtl->LogwrtResult; + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); if (record <= LogwrtResult.Flush) { LWLockRelease(WALWriteLock); @@ -3056,7 +3056,6 @@ XLogBackgroundFlush(void) /* read LogwrtResult and update local state */ SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; WriteRqst = XLogCtl->LogwrtRqst; SpinLockRelease(&XLogCtl->info_lck); @@ -3064,8 +3063,10 @@ 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) { + pg_memory_barrier(); SpinLockAcquire(&XLogCtl->info_lck); WriteRqst.Write = XLogCtl->asyncXactLSN; SpinLockRelease(&XLogCtl->info_lck); @@ -3081,6 +3082,7 @@ XLogBackgroundFlush(void) { if (openLogFile >= 0) { + LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write); if (!XLByteInPrevSeg(LogwrtResult.Write, openLogSegNo, wal_segment_size)) { @@ -3140,7 +3142,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 +3231,7 @@ XLogNeedsFlush(XLogRecPtr record) return false; /* read LogwrtResult and update local state */ - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); /* check again */ if (record <= LogwrtResult.Flush) @@ -7779,7 +7780,8 @@ StartupXLOG(void) LogwrtResult.Write = LogwrtResult.Flush = EndOfLog; - XLogCtl->LogwrtResult = LogwrtResult; + pg_atomic_write_u64(&XLogCtl->LogwrtResult.Write, EndOfLog); + pg_atomic_write_u64(&XLogCtl->LogwrtResult.Flush, EndOfLog); XLogCtl->LogwrtRqst.Write = EndOfLog; XLogCtl->LogwrtRqst.Flush = EndOfLog; @@ -8498,9 +8500,7 @@ GetInsertRecPtr(void) XLogRecPtr GetFlushRecPtr(void) { - SpinLockAcquire(&XLogCtl->info_lck); - LogwrtResult = XLogCtl->LogwrtResult; - SpinLockRelease(&XLogCtl->info_lck); + LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush); return LogwrtResult.Flush; } @@ -11649,9 +11649,7 @@ 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); return LogwrtResult.Write; } -- 2.20.1 --0OAP2g/MAC+5xKAE-- ^ permalink raw reply [nested|flat] 25+ messages in thread
* POC: make mxidoff 64 bits @ 2024-04-23 08:23 Maxim Orlov <[email protected]> 0 siblings, 1 reply; 25+ messages in thread From: Maxim Orlov @ 2024-04-23 08:23 UTC (permalink / raw) To: Postgres hackers <[email protected]> Hi! I've been trying to introduce 64-bit transaction identifications to Postgres for quite a while [0]. All this implies, of course, an enormous amount of change that will have to take place in various modules. Consider this, the patch set become too big to be committed “at once”. The obvious solutions was to try to split the patch set into smaller ones. But here it comes a new challenge, not every one of these parts, make Postgres better at the moment. Actually, even switching to a FullTransactionId in PGPROC will have disadvantage in increasing of WAL size [1]. In fact, I believe, we're dealing with the chicken and the egg problem here. Not able to commit full patch set since it is too big to handle and not able to commit parts of it, since they make sense all together and do not help improve Postgres at the moment. But it's not that bad. Since the commit 4ed8f0913bfdb5f, added in [3], we are capable to use 64 bits to indexing SLRUs. PROPOSAL Make multixact offsets 64 bit. RATIONALE It is not very difficult to overflow 32-bit mxidoff. Since, it is created for every unique combination of the transaction for each tuple, including XIDs and respective flags. And when a transaction is added to a specific multixact, it is rewritten with a new offset. In other words, it is possible to overflow the offsets of multixacts without overflowing the multixacts themselves and/or ordinary transactions. I believe, there was something about in the hackers mail lists, but I could not find links now. PFA, patch. Here is a WIP version. Upgrade machinery should be added later. As always, any opinions on a subject a very welcome! [0] https://www.postgresql.org/message-id/flat/CACG%3DezZe1NQSCnfHOr78AtAZxJZeCvxrts0ygrxYwe%3DpyyjVWA%4... [1] https://www.postgresql.org/message-id/flat/CACG%3DezY7msw%2Bjip%3Drtfvnfz051dRqz4s-diuO46v3rAoAE0T0g... [3] https://postgr.es/m/CAJ7c6TPDOYBYrnCAeyndkBktO0WG2xSdYduTF0nxq%2BvfkmTF5Q%40mail.gmail.com -- Best regards, Maxim Orlov. Attachments: [application/octet-stream] 0001-WIP-mxidoff-to-64bit.patch (23.7K, ../../CACG=ezaWg7_nt-8ey4aKv2w9LcuLthHknwCawmBgEeTnJrJTcw@mail.gmail.com/3-0001-WIP-mxidoff-to-64bit.patch) download | inline diff: From 622a47c2fe7b10de398390358cee97151f10fbb9 Mon Sep 17 00:00:00 2001 From: Maxim Orlov <[email protected]> Date: Wed, 6 Mar 2024 11:11:33 +0300 Subject: [PATCH] WIP : mxidoff to 64bit --- src/backend/access/rmgrdesc/mxactdesc.c | 9 +- src/backend/access/rmgrdesc/xlogdesc.c | 4 +- src/backend/access/transam/multixact.c | 230 ++++------------------ src/backend/access/transam/xlogrecovery.c | 5 +- src/bin/pg_controldata/pg_controldata.c | 4 +- src/bin/pg_resetwal/pg_resetwal.c | 10 +- src/bin/pg_resetwal/t/001_basic.pl | 2 +- src/include/access/multixact.h | 2 +- src/include/c.h | 2 +- 9 files changed, 58 insertions(+), 210 deletions(-) diff --git a/src/backend/access/rmgrdesc/mxactdesc.c b/src/backend/access/rmgrdesc/mxactdesc.c index 3e8ad4d5ef..1b486de38c 100644 --- a/src/backend/access/rmgrdesc/mxactdesc.c +++ b/src/backend/access/rmgrdesc/mxactdesc.c @@ -65,8 +65,8 @@ multixact_desc(StringInfo buf, XLogReaderState *record) xl_multixact_create *xlrec = (xl_multixact_create *) rec; int i; - appendStringInfo(buf, "%u offset %u nmembers %d: ", xlrec->mid, - xlrec->moff, xlrec->nmembers); + appendStringInfo(buf, "%u offset %llu nmembers %d: ", xlrec->mid, + (unsigned long long) xlrec->moff, xlrec->nmembers); for (i = 0; i < xlrec->nmembers; i++) out_member(buf, &xlrec->members[i]); } @@ -74,9 +74,10 @@ multixact_desc(StringInfo buf, XLogReaderState *record) { xl_multixact_truncate *xlrec = (xl_multixact_truncate *) rec; - appendStringInfo(buf, "offsets [%u, %u), members [%u, %u)", + appendStringInfo(buf, "offsets [%u, %u), members [%llu, %llu)", xlrec->startTruncOff, xlrec->endTruncOff, - xlrec->startTruncMemb, xlrec->endTruncMemb); + (unsigned long long) xlrec->startTruncMemb, + (unsigned long long) xlrec->endTruncMemb); } } diff --git a/src/backend/access/rmgrdesc/xlogdesc.c b/src/backend/access/rmgrdesc/xlogdesc.c index e455400716..1e4abc7bcf 100644 --- a/src/backend/access/rmgrdesc/xlogdesc.c +++ b/src/backend/access/rmgrdesc/xlogdesc.c @@ -45,7 +45,7 @@ xlog_desc(StringInfo buf, XLogReaderState *record) CheckPoint *checkpoint = (CheckPoint *) rec; appendStringInfo(buf, "redo %X/%X; " - "tli %u; prev tli %u; fpw %s; xid %u:%u; oid %u; multi %u; offset %u; " + "tli %u; prev tli %u; fpw %s; xid %u:%u; oid %u; multi %u; offset %llu; " "oldest xid %u in DB %u; oldest multi %u in DB %u; " "oldest/newest commit timestamp xid: %u/%u; " "oldest running xid %u; %s", @@ -57,7 +57,7 @@ xlog_desc(StringInfo buf, XLogReaderState *record) XidFromFullTransactionId(checkpoint->nextXid), checkpoint->nextOid, checkpoint->nextMulti, - checkpoint->nextMultiOffset, + (unsigned long long) checkpoint->nextMultiOffset, checkpoint->oldestXid, checkpoint->oldestXidDB, checkpoint->oldestMulti, diff --git a/src/backend/access/transam/multixact.c b/src/backend/access/transam/multixact.c index 380c866d71..a547d6598f 100644 --- a/src/backend/access/transam/multixact.c +++ b/src/backend/access/transam/multixact.c @@ -95,14 +95,6 @@ /* * Defines for MultiXactOffset page sizes. A page is the same BLCKSZ as is * used everywhere else in Postgres. - * - * Note: because MultiXactOffsets are 32 bits and wrap around at 0xFFFFFFFF, - * MultiXact page numbering also wraps around at - * 0xFFFFFFFF/MULTIXACT_OFFSETS_PER_PAGE, and segment numbering at - * 0xFFFFFFFF/MULTIXACT_OFFSETS_PER_PAGE/SLRU_PAGES_PER_SEGMENT. We need - * take no explicit notice of that fact in this module, except when comparing - * segment and page numbers in TruncateMultiXact (see - * MultiXactOffsetPagePrecedes). */ /* We need four bytes per offset */ @@ -156,7 +148,7 @@ ((uint32) ((0xFFFFFFFF % MULTIXACT_MEMBERS_PER_PAGE) + 1)) /* page in which a member is to be found */ -#define MXOffsetToMemberPage(xid) ((xid) / (TransactionId) MULTIXACT_MEMBERS_PER_PAGE) +#define MXOffsetToMemberPage(xid) ((xid) / (MultiXactOffset) MULTIXACT_MEMBERS_PER_PAGE) #define MXOffsetToMemberSegment(xid) (MXOffsetToMemberPage(xid) / SLRU_PAGES_PER_SEGMENT) /* Location (byte offset within page) of flag word for a given member */ @@ -230,9 +222,6 @@ typedef struct MultiXactStateData MultiXactId multiStopLimit; MultiXactId multiWrapLimit; - /* support for members anti-wraparound measures */ - MultiXactOffset offsetStopLimit; /* known if oldestOffsetKnown */ - /* * This is used to sleep until a multixact offset is written when we want * to create the next one. @@ -367,8 +356,6 @@ static bool MultiXactOffsetPrecedes(MultiXactOffset offset1, MultiXactOffset offset2); static void ExtendMultiXactOffset(MultiXactId multi); static void ExtendMultiXactMember(MultiXactOffset offset, int nmembers); -static bool MultiXactOffsetWouldWrap(MultiXactOffset boundary, - MultiXactOffset start, uint32 distance); static bool SetOffsetVacuumLimit(bool is_startup); static bool find_multixact_start(MultiXactId multi, MultiXactOffset *result); static void WriteMZeroPageXlogRec(int64 pageno, uint8 info); @@ -1117,78 +1104,6 @@ GetNewMultiXactId(int nmembers, MultiXactOffset *offset) else *offset = nextOffset; - /*---------- - * Protect against overrun of the members space as well, with the - * following rules: - * - * If we're past offsetStopLimit, refuse to generate more multis. - * If we're close to offsetStopLimit, emit a warning. - * - * Arbitrarily, we start emitting warnings when we're 20 segments or less - * from offsetStopLimit. - * - * Note we haven't updated the shared state yet, so if we fail at this - * point, the multixact ID we grabbed can still be used by the next guy. - * - * Note that there is no point in forcing autovacuum runs here: the - * multixact freeze settings would have to be reduced for that to have any - * effect. - *---------- - */ -#define OFFSET_WARN_SEGMENTS 20 - if (MultiXactState->oldestOffsetKnown && - MultiXactOffsetWouldWrap(MultiXactState->offsetStopLimit, nextOffset, - nmembers)) - { - /* see comment in the corresponding offsets wraparound case */ - SendPostmasterSignal(PMSIGNAL_START_AUTOVAC_LAUNCHER); - - ereport(ERROR, - (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED), - errmsg("multixact \"members\" limit exceeded"), - errdetail_plural("This command would create a multixact with %u members, but the remaining space is only enough for %u member.", - "This command would create a multixact with %u members, but the remaining space is only enough for %u members.", - MultiXactState->offsetStopLimit - nextOffset - 1, - nmembers, - MultiXactState->offsetStopLimit - nextOffset - 1), - errhint("Execute a database-wide VACUUM in database with OID %u with reduced vacuum_multixact_freeze_min_age and vacuum_multixact_freeze_table_age settings.", - MultiXactState->oldestMultiXactDB))); - } - - /* - * Check whether we should kick autovacuum into action, to prevent members - * wraparound. NB we use a much larger window to trigger autovacuum than - * just the warning limit. The warning is just a measure of last resort - - * this is in line with GetNewTransactionId's behaviour. - */ - if (!MultiXactState->oldestOffsetKnown || - (MultiXactState->nextOffset - MultiXactState->oldestOffset - > MULTIXACT_MEMBER_SAFE_THRESHOLD)) - { - /* - * To avoid swamping the postmaster with signals, we issue the autovac - * request only when crossing a segment boundary. With default - * compilation settings that's roughly after 50k members. This still - * gives plenty of chances before we get into real trouble. - */ - if ((MXOffsetToMemberPage(nextOffset) / SLRU_PAGES_PER_SEGMENT) != - (MXOffsetToMemberPage(nextOffset + nmembers) / SLRU_PAGES_PER_SEGMENT)) - SendPostmasterSignal(PMSIGNAL_START_AUTOVAC_LAUNCHER); - } - - if (MultiXactState->oldestOffsetKnown && - MultiXactOffsetWouldWrap(MultiXactState->offsetStopLimit, - nextOffset, - nmembers + MULTIXACT_MEMBERS_PER_PAGE * SLRU_PAGES_PER_SEGMENT * OFFSET_WARN_SEGMENTS)) - ereport(WARNING, - (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED), - errmsg_plural("database with OID %u must be vacuumed before %d more multixact member is used", - "database with OID %u must be vacuumed before %d more multixact members are used", - MultiXactState->offsetStopLimit - nextOffset + nmembers, - MultiXactState->oldestMultiXactDB, - MultiXactState->offsetStopLimit - nextOffset + nmembers), - errhint("Execute a database-wide VACUUM in that database with reduced vacuum_multixact_freeze_min_age and vacuum_multixact_freeze_table_age settings."))); - ExtendMultiXactMember(nextOffset, nmembers); /* @@ -1217,7 +1132,8 @@ GetNewMultiXactId(int nmembers, MultiXactOffset *offset) LWLockRelease(MultiXactGenLock); - debug_elog4(DEBUG2, "GetNew: returning %u offset %u", result, *offset); + debug_elog4(DEBUG2, "GetNew: returning %u offset %llu", result, + (unsigned long long) *offset); return result; } @@ -1926,7 +1842,7 @@ MultiXactShmemInit(void) "pg_multixact/offsets", LWTRANCHE_MULTIXACTOFFSET_BUFFER, LWTRANCHE_MULTIXACTOFFSET_SLRU, SYNC_HANDLER_MULTIXACT_OFFSET, - false); + true); SlruPagePrecedesUnitTests(MultiXactOffsetCtl, MULTIXACT_OFFSETS_PER_PAGE); SimpleLruInit(MultiXactMemberCtl, "multixact_member", multixact_member_buffers, 0, @@ -2244,8 +2160,9 @@ MultiXactGetCheckptMulti(bool is_shutdown, LWLockRelease(MultiXactGenLock); debug_elog6(DEBUG2, - "MultiXact: checkpoint is nextMulti %u, nextOffset %u, oldestMulti %u in DB %u", - *nextMulti, *nextMultiOffset, *oldestMulti, *oldestMultiDB); + "MultiXact: checkpoint is nextMulti %u, nextOffset %llu, oldestMulti %u in DB %u", + *nextMulti, (unsigned long long) *nextMultiOffset, *oldestMulti, + *oldestMultiDB); } /* @@ -2279,8 +2196,8 @@ void MultiXactSetNextMXact(MultiXactId nextMulti, MultiXactOffset nextMultiOffset) { - debug_elog4(DEBUG2, "MultiXact: setting next multi to %u offset %u", - nextMulti, nextMultiOffset); + debug_elog4(DEBUG2, "MultiXact: setting next multi to %u offset %llu", + nextMulti, (unsigned long long) nextMultiOffset); LWLockAcquire(MultiXactGenLock, LW_EXCLUSIVE); MultiXactState->nextMXact = nextMulti; MultiXactState->nextOffset = nextMultiOffset; @@ -2470,8 +2387,8 @@ MultiXactAdvanceNextMXact(MultiXactId minMulti, } if (MultiXactOffsetPrecedes(MultiXactState->nextOffset, minMultiOffset)) { - debug_elog3(DEBUG2, "MultiXact: setting next offset to %u", - minMultiOffset); + debug_elog3(DEBUG2, "MultiXact: setting next offset to %llu", + (unsigned long long) minMultiOffset); MultiXactState->nextOffset = minMultiOffset; } LWLockRelease(MultiXactGenLock); @@ -2670,8 +2587,6 @@ SetOffsetVacuumLimit(bool is_startup) MultiXactOffset nextOffset; bool oldestOffsetKnown = false; bool prevOldestOffsetKnown; - MultiXactOffset offsetStopLimit = 0; - MultiXactOffset prevOffsetStopLimit; /* * NB: Have to prevent concurrent truncation, we might otherwise try to @@ -2686,7 +2601,6 @@ SetOffsetVacuumLimit(bool is_startup) nextOffset = MultiXactState->nextOffset; prevOldestOffsetKnown = MultiXactState->oldestOffsetKnown; prevOldestOffset = MultiXactState->oldestOffset; - prevOffsetStopLimit = MultiXactState->offsetStopLimit; Assert(MultiXactState->finishedStartup); LWLockRelease(MultiXactGenLock); @@ -2717,11 +2631,7 @@ SetOffsetVacuumLimit(bool is_startup) oldestOffsetKnown = find_multixact_start(oldestMultiXactId, &oldestOffset); - if (oldestOffsetKnown) - ereport(DEBUG1, - (errmsg_internal("oldest MultiXactId member is at offset %u", - oldestOffset))); - else + if (!oldestOffsetKnown) ereport(LOG, (errmsg("MultiXact member wraparound protections are disabled because oldest checkpointed MultiXact %u does not exist on disk", oldestMultiXactId))); @@ -2734,24 +2644,7 @@ SetOffsetVacuumLimit(bool is_startup) * overrun of old data in the members SLRU area. We can only do so if the * oldest offset is known though. */ - if (oldestOffsetKnown) - { - /* move back to start of the corresponding segment */ - offsetStopLimit = oldestOffset - (oldestOffset % - (MULTIXACT_MEMBERS_PER_PAGE * SLRU_PAGES_PER_SEGMENT)); - - /* always leave one segment before the wraparound point */ - offsetStopLimit -= (MULTIXACT_MEMBERS_PER_PAGE * SLRU_PAGES_PER_SEGMENT); - - if (!prevOldestOffsetKnown && !is_startup) - ereport(LOG, - (errmsg("MultiXact member wraparound protections are now enabled"))); - - ereport(DEBUG1, - (errmsg_internal("MultiXact member stop limit is now %u based on MultiXact %u", - offsetStopLimit, oldestMultiXactId))); - } - else if (prevOldestOffsetKnown) + if (prevOldestOffsetKnown) { /* * If we failed to get the oldest offset this time, but we have a @@ -2761,14 +2654,12 @@ SetOffsetVacuumLimit(bool is_startup) */ oldestOffset = prevOldestOffset; oldestOffsetKnown = true; - offsetStopLimit = prevOffsetStopLimit; } /* Install the computed values */ LWLockAcquire(MultiXactGenLock, LW_EXCLUSIVE); MultiXactState->oldestOffset = oldestOffset; MultiXactState->oldestOffsetKnown = oldestOffsetKnown; - MultiXactState->offsetStopLimit = offsetStopLimit; LWLockRelease(MultiXactGenLock); /* @@ -2778,54 +2669,6 @@ SetOffsetVacuumLimit(bool is_startup) (nextOffset - oldestOffset > MULTIXACT_MEMBER_SAFE_THRESHOLD); } -/* - * Return whether adding "distance" to "start" would move past "boundary". - * - * We use this to determine whether the addition is "wrapping around" the - * boundary point, hence the name. The reason we don't want to use the regular - * 2^31-modulo arithmetic here is that we want to be able to use the whole of - * the 2^32-1 space here, allowing for more multixacts than would fit - * otherwise. - */ -static bool -MultiXactOffsetWouldWrap(MultiXactOffset boundary, MultiXactOffset start, - uint32 distance) -{ - MultiXactOffset finish; - - /* - * Note that offset number 0 is not used (see GetMultiXactIdMembers), so - * if the addition wraps around the UINT_MAX boundary, skip that value. - */ - finish = start + distance; - if (finish < start) - finish++; - - /*----------------------------------------------------------------------- - * When the boundary is numerically greater than the starting point, any - * value numerically between the two is not wrapped: - * - * <----S----B----> - * [---) = F wrapped past B (and UINT_MAX) - * [---) = F not wrapped - * [----] = F wrapped past B - * - * When the boundary is numerically less than the starting point (i.e. the - * UINT_MAX wraparound occurs somewhere in between) then all values in - * between are wrapped: - * - * <----B----S----> - * [---) = F not wrapped past B (but wrapped past UINT_MAX) - * [---) = F wrapped past B (and UINT_MAX) - * [----] = F not wrapped - *----------------------------------------------------------------------- - */ - if (start < boundary) - return finish >= boundary || finish < start; - else - return finish >= boundary && finish < start; -} - /* * Find the starting offset of the given MultiXactId. * @@ -2946,8 +2789,9 @@ MultiXactMemberFreezeThreshold(void) * we try to eliminate from the system is based on how far we are past * MULTIXACT_MEMBER_SAFE_THRESHOLD. */ - fraction = (double) (members - MULTIXACT_MEMBER_SAFE_THRESHOLD) / - (MULTIXACT_MEMBER_DANGER_THRESHOLD - MULTIXACT_MEMBER_SAFE_THRESHOLD); + fraction = (double) (members - MULTIXACT_MEMBER_SAFE_THRESHOLD); + fraction /= (double) (MULTIXACT_MEMBER_DANGER_THRESHOLD - MULTIXACT_MEMBER_SAFE_THRESHOLD); + victim_multixacts = multixacts * fraction; /* fraction could be > 1.0, but lowest possible freeze age is zero */ @@ -2991,10 +2835,10 @@ SlruScanDirCbFindEarliest(SlruCtl ctl, char *filename, int64 segpage, void *data static void PerformMembersTruncation(MultiXactOffset oldestOffset, MultiXactOffset newOldestOffset) { - const int maxsegment = MXOffsetToMemberSegment(MaxMultiXactOffset); - int startsegment = MXOffsetToMemberSegment(oldestOffset); - int endsegment = MXOffsetToMemberSegment(newOldestOffset); - int segment = startsegment; + const int64 maxsegment = MXOffsetToMemberSegment(MaxMultiXactOffset); + int64 startsegment = MXOffsetToMemberSegment(oldestOffset); + int64 endsegment = MXOffsetToMemberSegment(newOldestOffset); + int64 segment = startsegment; /* * Delete all the segments but the last one. The last segment can still @@ -3002,7 +2846,8 @@ PerformMembersTruncation(MultiXactOffset oldestOffset, MultiXactOffset newOldest */ while (segment != endsegment) { - elog(DEBUG2, "truncating multixact members segment %x", segment); + elog(DEBUG2, "truncating multixact members segment %llx", + (unsigned long long) segment); SlruDeleteSegment(MultiXactMemberCtl, segment); /* move to next segment, handling wraparound correctly */ @@ -3153,14 +2998,14 @@ TruncateMultiXact(MultiXactId newOldestMulti, Oid newOldestMultiDB) } elog(DEBUG1, "performing multixact truncation: " - "offsets [%u, %u), offsets segments [%x, %x), " - "members [%u, %u), members segments [%x, %x)", + "offsets [%u, %u), offsets segments [%llx, %llx), " + "members [%llu, %llu), members segments [%llx, %llx)", oldestMulti, newOldestMulti, - MultiXactIdToOffsetSegment(oldestMulti), - MultiXactIdToOffsetSegment(newOldestMulti), - oldestOffset, newOldestOffset, - MXOffsetToMemberSegment(oldestOffset), - MXOffsetToMemberSegment(newOldestOffset)); + (unsigned long long) MultiXactIdToOffsetSegment(oldestMulti), + (unsigned long long) MultiXactIdToOffsetSegment(newOldestMulti), + (unsigned long long) oldestOffset, (unsigned long long) newOldestOffset, + (unsigned long long) MXOffsetToMemberSegment(oldestOffset), + (unsigned long long) MXOffsetToMemberSegment(newOldestOffset)); /* * Do truncation, and the WAL logging of the truncation, in a critical @@ -3285,7 +3130,7 @@ MultiXactIdPrecedesOrEquals(MultiXactId multi1, MultiXactId multi2) static bool MultiXactOffsetPrecedes(MultiXactOffset offset1, MultiXactOffset offset2) { - int32 diff = (int32) (offset1 - offset2); + int64 diff = (int64) (offset1 - offset2); return (diff < 0); } @@ -3413,14 +3258,15 @@ multixact_redo(XLogReaderState *record) SizeOfMultiXactTruncate); elog(DEBUG1, "replaying multixact truncation: " - "offsets [%u, %u), offsets segments [%x, %x), " - "members [%u, %u), members segments [%x, %x)", + "offsets [%u, %u), offsets segments [%llx, %llx), " + "members [%llu, %llu), members segments [%llx, %llx)", xlrec.startTruncOff, xlrec.endTruncOff, - MultiXactIdToOffsetSegment(xlrec.startTruncOff), - MultiXactIdToOffsetSegment(xlrec.endTruncOff), - xlrec.startTruncMemb, xlrec.endTruncMemb, - MXOffsetToMemberSegment(xlrec.startTruncMemb), - MXOffsetToMemberSegment(xlrec.endTruncMemb)); + (unsigned long long) MultiXactIdToOffsetSegment(xlrec.startTruncOff), + (unsigned long long) MultiXactIdToOffsetSegment(xlrec.endTruncOff), + (unsigned long long) xlrec.startTruncMemb, + (unsigned long long) xlrec.endTruncMemb, + (unsigned long long) MXOffsetToMemberSegment(xlrec.startTruncMemb), + (unsigned long long) MXOffsetToMemberSegment(xlrec.endTruncMemb)); /* should not be required, but more than cheap enough */ LWLockAcquire(MultiXactTruncationLock, LW_EXCLUSIVE); diff --git a/src/backend/access/transam/xlogrecovery.c b/src/backend/access/transam/xlogrecovery.c index 29c5bec084..aa9566b01c 100644 --- a/src/backend/access/transam/xlogrecovery.c +++ b/src/backend/access/transam/xlogrecovery.c @@ -876,8 +876,9 @@ InitWalRecovery(ControlFileData *ControlFile, bool *wasShutdown_ptr, U64FromFullTransactionId(checkPoint.nextXid), checkPoint.nextOid))); ereport(DEBUG1, - (errmsg_internal("next MultiXactId: %u; next MultiXactOffset: %u", - checkPoint.nextMulti, checkPoint.nextMultiOffset))); + (errmsg_internal("next MultiXactId: %u; next MultiXactOffset: %llu", + checkPoint.nextMulti, + (unsigned long long) checkPoint.nextMultiOffset))); ereport(DEBUG1, (errmsg_internal("oldest unfrozen transaction ID: %u, in database %u", checkPoint.oldestXid, checkPoint.oldestXidDB))); diff --git a/src/bin/pg_controldata/pg_controldata.c b/src/bin/pg_controldata/pg_controldata.c index 93e0837947..d38c752860 100644 --- a/src/bin/pg_controldata/pg_controldata.c +++ b/src/bin/pg_controldata/pg_controldata.c @@ -253,8 +253,8 @@ main(int argc, char *argv[]) ControlFile->checkPointCopy.nextOid); printf(_("Latest checkpoint's NextMultiXactId: %u\n"), ControlFile->checkPointCopy.nextMulti); - printf(_("Latest checkpoint's NextMultiOffset: %u\n"), - ControlFile->checkPointCopy.nextMultiOffset); + printf(_("Latest checkpoint's NextMultiOffset: %llu\n"), + (unsigned long long) ControlFile->checkPointCopy.nextMultiOffset); printf(_("Latest checkpoint's oldestXID: %u\n"), ControlFile->checkPointCopy.oldestXid); printf(_("Latest checkpoint's oldestXID's DB: %u\n"), diff --git a/src/bin/pg_resetwal/pg_resetwal.c b/src/bin/pg_resetwal/pg_resetwal.c index e9dcb5a6d8..1af2ce4b93 100644 --- a/src/bin/pg_resetwal/pg_resetwal.c +++ b/src/bin/pg_resetwal/pg_resetwal.c @@ -264,7 +264,7 @@ main(int argc, char *argv[]) case 'O': errno = 0; - set_mxoff = strtoul(optarg, &endptr, 0); + set_mxoff = strtou64(optarg, &endptr, 0); if (endptr == optarg || *endptr != '\0' || errno != 0) { pg_log_error("invalid argument for option %s", "-O"); @@ -737,8 +737,8 @@ PrintControlValues(bool guessed) ControlFile.checkPointCopy.nextOid); printf(_("Latest checkpoint's NextMultiXactId: %u\n"), ControlFile.checkPointCopy.nextMulti); - printf(_("Latest checkpoint's NextMultiOffset: %u\n"), - ControlFile.checkPointCopy.nextMultiOffset); + printf(_("Latest checkpoint's NextMultiOffset: %llu\n"), + (unsigned long long) ControlFile.checkPointCopy.nextMultiOffset); printf(_("Latest checkpoint's oldestXID: %u\n"), ControlFile.checkPointCopy.oldestXid); printf(_("Latest checkpoint's oldestXID's DB: %u\n"), @@ -809,8 +809,8 @@ PrintNewControlValues(void) if (set_mxoff != -1) { - printf(_("NextMultiOffset: %u\n"), - ControlFile.checkPointCopy.nextMultiOffset); + printf(_("NextMultiOffset: %llu\n"), + (unsigned long long) ControlFile.checkPointCopy.nextMultiOffset); } if (set_oid != 0) diff --git a/src/bin/pg_resetwal/t/001_basic.pl b/src/bin/pg_resetwal/t/001_basic.pl index 9829e48106..f8a8eef44d 100644 --- a/src/bin/pg_resetwal/t/001_basic.pl +++ b/src/bin/pg_resetwal/t/001_basic.pl @@ -206,7 +206,7 @@ push @cmd, sprintf("%d,%d", hex($files[0]) == 0 ? 3 : hex($files[0]), hex($files[-1])); @files = get_slru_files('pg_multixact/offsets'); -$mult = 32 * $blcksz / 4; +$mult = 32 * $blcksz / 8; # -m argument is "new,old" push @cmd, '-m', sprintf("%d,%d", diff --git a/src/include/access/multixact.h b/src/include/access/multixact.h index 7ffd256c74..90583634ec 100644 --- a/src/include/access/multixact.h +++ b/src/include/access/multixact.h @@ -27,7 +27,7 @@ #define MultiXactIdIsValid(multi) ((multi) != InvalidMultiXactId) -#define MaxMultiXactOffset ((MultiXactOffset) 0xFFFFFFFF) +#define MaxMultiXactOffset UINT64CONST(0xFFFFFFFFFFFFFFFF) /* * Possible multixact lock modes ("status"). The first four modes are for diff --git a/src/include/c.h b/src/include/c.h index dc1841346c..ccfb82b478 100644 --- a/src/include/c.h +++ b/src/include/c.h @@ -661,7 +661,7 @@ typedef uint32 SubTransactionId; /* MultiXactId must be equivalent to TransactionId, to fit in t_xmax */ typedef TransactionId MultiXactId; -typedef uint32 MultiXactOffset; +typedef uint64 MultiXactOffset; typedef uint32 CommandId; -- 2.34.1 ^ permalink raw reply [nested|flat] 25+ messages in thread
* Re: POC: make mxidoff 64 bits @ 2024-04-23 13:02 Andrey M. Borodin <[email protected]> parent: Maxim Orlov <[email protected]> 0 siblings, 0 replies; 25+ messages in thread From: Andrey M. Borodin @ 2024-04-23 13:02 UTC (permalink / raw) To: Maxim Orlov <[email protected]>; +Cc: Postgres hackers <[email protected]> > On 23 Apr 2024, at 11:23, Maxim Orlov <[email protected]> wrote: > > Make multixact offsets 64 bit. - ereport(ERROR, - (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED), - errmsg("multixact \"members\" limit exceeded"), Personally, I'd be happy with this! We had some incidents where the only mitigation was vacuum settings tweaking. BTW as a side note... I see lot's of casts to (unsigned long long), can't we just cast to MultiXactOffset? Best regards, Andrey Borodin. ^ permalink raw reply [nested|flat] 25+ messages in thread
end of thread, other threads:[~2024-04-23 13:02 UTC | newest] Thread overview: 25+ messages (download: mbox mbox.gz follow: Atom feed) -- links below jump to the message on this page -- 2021-02-02 17:03 [PATCH 2/2] make LogwrtResult atomic Alvaro Herrera <[email protected]> 2021-02-02 17:03 [PATCH 2/2] make LogwrtResult atomic Alvaro Herrera <[email protected]> 2021-02-02 17:03 [PATCH 2/2] make LogwrtResult atomic Alvaro Herrera <[email protected]> 2021-02-02 17:03 [PATCH 2/2] make LogwrtResult atomic Alvaro Herrera <[email protected]> 2021-02-02 17:03 [PATCH 2/2] make LogwrtResult atomic Alvaro Herrera <[email protected]> 2021-02-02 17:03 [PATCH 2/2] make LogwrtResult atomic Alvaro Herrera <[email protected]> 2021-02-02 17:03 [PATCH 2/2] make LogwrtResult atomic Alvaro Herrera <[email protected]> 2021-02-02 17:03 [PATCH 2/2] make LogwrtResult atomic Alvaro Herrera <[email protected]> 2021-02-02 17:03 [PATCH 2/2] make LogwrtResult atomic Alvaro Herrera <[email protected]> 2021-02-02 17:03 [PATCH 2/2] make LogwrtResult atomic Alvaro Herrera <[email protected]> 2021-02-02 17:03 [PATCH 2/2] make LogwrtResult atomic Alvaro Herrera <[email protected]> 2021-02-02 17:03 [PATCH 2/2] make LogwrtResult atomic Alvaro Herrera <[email protected]> 2021-02-02 17:03 [PATCH 2/2] make LogwrtResult atomic Alvaro Herrera <[email protected]> 2021-02-02 17:03 [PATCH 2/2] make LogwrtResult atomic Alvaro Herrera <[email protected]> 2021-02-02 17:03 [PATCH 2/2] make LogwrtResult atomic Alvaro Herrera <[email protected]> 2021-02-02 17:03 [PATCH 2/2] make LogwrtResult atomic Alvaro Herrera <[email protected]> 2021-02-02 17:03 [PATCH 2/2] make LogwrtResult atomic Alvaro Herrera <[email protected]> 2021-02-02 17:03 [PATCH 2/2] make LogwrtResult atomic Alvaro Herrera <[email protected]> 2021-02-02 17:03 [PATCH 2/2] make LogwrtResult atomic Alvaro Herrera <[email protected]> 2021-02-02 17:03 [PATCH v5 1/2] make LogwrtResult atomic Alvaro Herrera <[email protected]> 2021-02-02 17:03 [PATCH 2/2] make LogwrtResult atomic Alvaro Herrera <[email protected]> 2021-02-02 17:03 [PATCH 2/2] make LogwrtResult atomic Alvaro Herrera <[email protected]> 2021-02-02 17:03 [PATCH 2/2] make LogwrtResult atomic Alvaro Herrera <[email protected]> 2024-04-23 08:23 POC: make mxidoff 64 bits Maxim Orlov <[email protected]> 2024-04-23 13:02 ` Re: POC: make mxidoff 64 bits Andrey M. Borodin <[email protected]>
This inbox is served by agora; see mirroring instructions for how to clone and mirror all data and code used for this inbox