xlog: Avoid repeated scans of all xlog insert locks when replacing contents.

Author:
Reviewed-By:
Discussion: https://postgr.es/m/
Backpatch:

diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 02a1c10c03424338b4e9d31255e761db08c48f14..0050773c70c7338b3e9683beb752e33dec6cc428 100644 (file)
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -587,6 +587,8 @@ typedef struct XLogCtlInsert
    int         nonExclusiveBackups;
    XLogRecPtr  lastBackupStart;
 
+   pg_atomic_uint64 knownCompletedUpto;
+
    /*
     * WAL insertion locks.
     */
@@ -1780,6 +1782,22 @@ WALInsertLockUpdateInsertingAt(XLogRecPtr insertingAt)
                        insertingAt);
 }
 
+static bool
+XLogInsertionsKnownFinished(XLogRecPtr upto)
+{
+   XLogRecPtr  knownFinishedUpto;
+
+   knownFinishedUpto = (XLogRecPtr) pg_atomic_read_u64(&XLogCtl->Insert.knownCompletedUpto);
+
+   if (upto <= knownFinishedUpto)
+   {
+       pg_read_barrier();
+       return true;
+   }
+   else
+       return false;
+}
+
 /*
  * Wait for any WAL insertions < upto to finish.
  *
@@ -1875,6 +1893,20 @@ WaitXLogInsertionsToFinish(XLogRecPtr upto)
        if (insertingat != InvalidXLogRecPtr && insertingat < finishedUpto)
            finishedUpto = insertingat;
    }
+
+   {
+       XLogRecPtr knownCompletedUpto =
+           (XLogRecPtr) pg_atomic_read_u64(&XLogCtl->Insert.knownCompletedUpto);
+
+       while (knownCompletedUpto < finishedUpto)
+       {
+           if (pg_atomic_compare_exchange_u64(&XLogCtl->Insert.knownCompletedUpto,
+                                              &knownCompletedUpto,
+                                              finishedUpto))
+               break;
+       }
+   }
+
    return finishedUpto;
 }
 
@@ -2190,7 +2222,11 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic)
                 */
                LWLockRelease(WALBufMappingLock);
 
-               WaitXLogInsertionsToFinish(OldPageRqstPtr);
+               if (!XLogInsertionsKnownFinished(OldPageRqstPtr))
+               {
+                   WaitXLogInsertionsToFinish(OldPageRqstPtr);
+                   Assert(XLogInsertionsKnownFinished(OldPageRqstPtr));
+               }
 
                LWLockAcquire(WALWriteLock, LW_EXCLUSIVE);
 
@@ -2438,6 +2474,8 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible)
    /* We should always be inside a critical section here */
    Assert(CritSectionCount > 0);
 
+   Assert(XLogInsertionsKnownFinished(WriteRqst.Write));
+
    /*
     * Update local LogwrtResult (caller probably did this already, but...)
     */
@@ -5213,6 +5251,8 @@ XLOGShmemInit(void)
    SpinLockInit(&XLogCtl->info_lck);
    SpinLockInit(&XLogCtl->ulsn_lck);
    InitSharedLatch(&XLogCtl->recoveryWakeupLatch);
+
+   pg_atomic_init_u64(&XLogCtl->Insert.knownCompletedUpto, 0);
 }
 
 /*