From: Alvaro Herrera <alvherre@alvh.no-ip.org>
To: Tom Lane <tgl@sss.pgh.pa.us>
Cc: Andres Freund <andres@anarazel.de>
Cc: pgsql-hackers@lists.postgresql.org, Jaime Casanova <jcasanov@systemguards.com.ec>
Subject: Re: LogwrtResult contended spinlock
Date: Tue, 5 Apr 2022 20:16:28 +0200
Message-ID: <202204051816.c2u5wz6ru3ni@alvherre.pgsql> (raw)
In-Reply-To: <202204051547.jikyi32qqvfc@alvherre.pgsql>
Apologies -- I selected the wrong commit to extract the commit message
from. Here it is again. I also removed an obsolete /* XXX */ comment.
--
Álvaro Herrera Breisgau, Deutschland — https://www.EnterpriseDB.com/Attachments:
[text/x-diff] v9-0001-Make-XLogCtl-LogwrtResult-accessible-with-atomics.patch (12.9K, ../202204051816.c2u5wz6ru3ni@alvherre.pgsql/2-v9-0001-Make-XLogCtl-LogwrtResult-accessible-with-atomics.patch)
download | inline diff:
From 0b26f90f1b95f8e9b932eb34bbf9c2a50729cf60 Mon Sep 17 00:00:00 2001
From: Alvaro Herrera <alvherre@alvh.no-ip.org>
Date: Tue, 2 Feb 2021 14:03:43 -0300
Subject: [PATCH v9] Make XLogCtl->LogwrtResult accessible with atomics
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
Currently, access to LogwrtResult is protected by a spinlock. This
becomes severely contended in some scenarios, such as with a largish
replication flock: walsenders all calling GetFlushRecPtr repeatedly
cause the processor heat up to the point where eggs can be fried on top.
This can be reduced to a non-problem by replacing XLogCtl->LogwrtResult
with a struct containing a pair of atomically accessed variables. Do so.
In a few places, we can adjust the exact location where the locals are
updated to account for the fact that we no longer need the spinlock.
Author: Álvaro Herrera <alvherre@alvh.no-ip.org>
Discussion: https://postgr.es/m/20200831182156.GA3983@alvherre.pgsql
---
src/backend/access/transam/xlog.c | 106 ++++++++++++++----------------
src/include/port/atomics.h | 29 ++++++++
src/tools/pgindent/typedefs.list | 1 +
3 files changed, 78 insertions(+), 58 deletions(-)
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.cindex 17a56152f1..50fdfcbc90 100644--- a/src/backend/access/transam/xlog.c+++ b/src/backend/access/transam/xlog.c@@ -285,16 +285,13 @@ static bool doPageWrites;
*
* 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.- * These structs are identical but are declared separately to indicate their- * slightly different functions.+ * LogWrtResult indicates the byte positions we have already written/fsynced.+ * These structs are similar but are declared separately to indicate their+ * slightly different functions; in addition, the latter is read and written+ * using atomic operations.
*
- * 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.+ * 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.
@@ -317,6 +314,12 @@ static bool doPageWrites;
*----------
*/
+typedef struct XLogwrtAtomic+{+ pg_atomic_uint64 Write; /* last byte + 1 written out */+ pg_atomic_uint64 Flush; /* last byte + 1 flushed */+} XLogwrtAtomic;+
typedef struct XLogwrtRqst
{
XLogRecPtr Write; /* last byte + 1 to write out */
@@ -480,6 +483,7 @@ typedef struct XLogCtlData
{
XLogCtlInsert Insert;
+ XLogwrtAtomic LogwrtResult; /* uses atomics */
/* Protected by info_lck: */
XLogwrtRqst LogwrtRqst;
XLogRecPtr RedoRecPtr; /* a recent copy of Insert->RedoRecPtr */
@@ -497,12 +501,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).
*
@@ -622,7 +620,7 @@ 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};
@@ -932,8 +930,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);
}
@@ -1811,6 +1807,7 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, TimeLineID tli, bool opportunistic)
* Now that we have the lock, check if someone initialized the page
* already.
*/
+ LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write);
while (upto >= XLogCtl->InitializedUpTo || opportunistic)
{
nextidx = XLogRecPtrToBufIdx(XLogCtl->InitializedUpTo);
@@ -1830,17 +1827,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 LogwrtResult.Write and see if we still need+ * to write it or if someone else already did.
*/
+ LogwrtResult.Write =+ pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write);
if (LogwrtResult.Write < OldPageRqstPtr)
{
/*
@@ -1855,7 +1853,8 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, TimeLineID tli, 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 */
@@ -2101,7 +2100,7 @@ XLogWrite(XLogwrtRqst WriteRqst, TimeLineID tli, 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,
@@ -2316,6 +2315,10 @@ XLogWrite(XLogwrtRqst WriteRqst, TimeLineID tli, bool flexible)
Assert(npages == 0);
+ /* Publish current write result position */+ pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Write,+ LogwrtResult.Write);+
/*
* If asked to flush, do so
*/
@@ -2352,22 +2355,9 @@ XLogWrite(XLogwrtRqst WriteRqst, TimeLineID tli, bool flexible)
LogwrtResult.Flush = LogwrtResult.Write;
}
- /*- * Update shared-memory status- *- * We 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;- SpinLockRelease(&XLogCtl->info_lck);- }+ /* Publish current flush result position */+ pg_atomic_monotonic_advance_u64(&XLogCtl->LogwrtResult.Flush,+ LogwrtResult.Flush);
}
/*
@@ -2381,8 +2371,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;
@@ -2583,14 +2573,14 @@ XLogFlush(XLogRecPtr record)
{
XLogRecPtr insertpos;
- /* read LogwrtResult and update local state */+ /* Skip ahead write request pointer to latest written globally */
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;
@@ -2618,7 +2608,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);
@@ -2742,7 +2732,6 @@ XLogBackgroundFlush(void)
/* read LogwrtResult and update local state */
SpinLockAcquire(&XLogCtl->info_lck);
- LogwrtResult = XLogCtl->LogwrtResult;
WriteRqst = XLogCtl->LogwrtRqst;
SpinLockRelease(&XLogCtl->info_lck);
@@ -2750,8 +2739,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);
@@ -2767,6 +2758,7 @@ XLogBackgroundFlush(void)
{
if (openLogFile >= 0)
{
+ LogwrtResult.Write = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Write);
if (!XLByteInPrevSeg(LogwrtResult.Write, openLogSegNo,
wal_segment_size))
{
@@ -2826,7 +2818,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)
{
@@ -2914,9 +2907,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)
@@ -5487,10 +5478,13 @@ StartupXLOG(void)
XLogCtl->InitializedUpTo = EndOfLog;
}
+ /*+ * Update local and shared status. This is OK to do without any locks+ * because no other process can be reading or writing WAL yet.+ */
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;
@@ -5925,9 +5919,7 @@ GetFlushRecPtr(TimeLineID *insertTLI)
{
Assert(XLogCtl->SharedRecoveryState == RECOVERY_STATE_DONE);
- SpinLockAcquire(&XLogCtl->info_lck);- LogwrtResult = XLogCtl->LogwrtResult;- SpinLockRelease(&XLogCtl->info_lck);+ LogwrtResult.Flush = pg_atomic_read_u64(&XLogCtl->LogwrtResult.Flush);
/*
* If we're writing and flushing WAL, the time line can't be changing, so
@@ -9095,9 +9087,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;
}
diff --git a/src/include/port/atomics.h b/src/include/port/atomics.hindex 9550e04aaa..7abed1785a 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.listindex 4b12c575ab..74e1070433 100644--- a/src/tools/pgindent/typedefs.list+++ b/src/tools/pgindent/typedefs.list@@ -2975,6 +2975,7 @@ XLogRecoveryCtlData
XLogRedoAction
XLogSegNo
XLogSource
+XLogwrtAtomic
XLogwrtResult
XLogwrtRqst
XPVIV
--
2.30.2
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: pgsql-hackers@postgresql.org
Cc: alvherre@alvh.no-ip.org, tgl@sss.pgh.pa.us, andres@anarazel.de, jcasanov@systemguards.com.ec
Subject: Re: LogwrtResult contended spinlock
In-Reply-To: <202204051816.c2u5wz6ru3ni@alvherre.pgsql>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by DDX for PostgreSQL; see mirroring instructions
for how to clone and mirror all data and code used for this inbox