From 2f4a32b1d09419167fe5040465c8a3464010e012 Mon Sep 17 00:00:00 2001 From: Zhiguo Zhou Date: Thu, 2 Jan 2025 12:04:02 +0800 Subject: [PATCH] Lock-free XLog Reservation from WAL Removed PrevBytePos to eliminate lock contention, allowing atomic updates to CurrBytePos. Adjusted XLog reservation to exceed current XLog memory slightly, enabling the next XLog's prev-link update without waiting for the current XLog. Backends now update next XLog's prev-link before inserting current log, breaking sequential write convention. Updated GetXLogBuffer to handle separate access and update LSNs. Increased NUM_XLOGINSERT_LOCKS. --- src/backend/access/transam/xlog.c | 186 +++++++++++++++++++++--------- 1 file changed, 131 insertions(+), 55 deletions(-) diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index 6f58412bca..6215ea6977 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -147,7 +147,7 @@ int wal_segment_size = DEFAULT_XLOG_SEG_SIZE; * to happen concurrently, but adds some CPU overhead to flushing the WAL, * which needs to iterate all the locks. */ -#define NUM_XLOGINSERT_LOCKS 8 +#define NUM_XLOGINSERT_LOCKS 128 /* * Max distance from last checkpoint, before triggering a new xlog-based @@ -404,8 +404,7 @@ typedef struct XLogCtlInsert * prev-link of the next record. These are stored as "usable byte * positions" rather than XLogRecPtrs (see XLogBytePosToRecPtr()). */ - uint64 CurrBytePos; - uint64 PrevBytePos; + pg_atomic_uint64 CurrBytePos; /* * Make sure the above heavily-contended spinlock and byte positions are @@ -700,12 +699,14 @@ static void CopyXLogRecordToWAL(int write_len, bool isLogSwitch, XLogRecData *rdata, XLogRecPtr StartPos, XLogRecPtr EndPos, TimeLineID tli); +static XLogRecPtr GetPrevRecPtr(XLogRecPtr hdr, TimeLineID tli); +static void SetPrevRecPtr(XLogRecPtr hdr, XLogRecPtr PrevPos, + TimeLineID tli, bool in_order); static void ReserveXLogInsertLocation(int size, XLogRecPtr *StartPos, - XLogRecPtr *EndPos, XLogRecPtr *PrevPtr); -static bool ReserveXLogSwitch(XLogRecPtr *StartPos, XLogRecPtr *EndPos, - XLogRecPtr *PrevPtr); + XLogRecPtr *EndPos); +static bool ReserveXLogSwitch(XLogRecPtr *StartPos, XLogRecPtr *EndPos); static XLogRecPtr WaitXLogInsertionsToFinish(XLogRecPtr upto); -static char *GetXLogBuffer(XLogRecPtr ptr, TimeLineID tli); +static char *GetXLogBuffer(XLogRecPtr ptr, XLogRecPtr upto, TimeLineID tli); static XLogRecPtr XLogBytePosToRecPtr(uint64 bytepos); static XLogRecPtr XLogBytePosToEndRecPtr(uint64 bytepos); static uint64 XLogRecPtrToBytePos(XLogRecPtr ptr); @@ -862,8 +863,7 @@ XLogInsertRecord(XLogRecData *rdata, * Reserve space for the record in the WAL. This also sets the xl_prev * pointer. */ - ReserveXLogInsertLocation(rechdr->xl_tot_len, &StartPos, &EndPos, - &rechdr->xl_prev); + ReserveXLogInsertLocation(rechdr->xl_tot_len, &StartPos, &EndPos); /* Normal records are always inserted. */ inserted = true; @@ -883,7 +883,7 @@ XLogInsertRecord(XLogRecData *rdata, */ Assert(fpw_lsn == InvalidXLogRecPtr); WALInsertLockAcquireExclusive(); - inserted = ReserveXLogSwitch(&StartPos, &EndPos, &rechdr->xl_prev); + inserted = ReserveXLogSwitch(&StartPos, &EndPos); } else { @@ -898,14 +898,22 @@ XLogInsertRecord(XLogRecData *rdata, */ Assert(fpw_lsn == InvalidXLogRecPtr); WALInsertLockAcquireExclusive(); - ReserveXLogInsertLocation(rechdr->xl_tot_len, &StartPos, &EndPos, - &rechdr->xl_prev); + ReserveXLogInsertLocation(rechdr->xl_tot_len, &StartPos, &EndPos); RedoRecPtr = Insert->RedoRecPtr = StartPos; inserted = true; } if (inserted) { + bool islargerecord; + + islargerecord = EndPos - StartPos >= (Size) XLOGbuffers * XLOG_BLCKSZ; + if (!islargerecord) + { + SetPrevRecPtr(EndPos, StartPos, insertTLI, false); + } + while (!(rechdr->xl_prev = GetPrevRecPtr(StartPos, insertTLI))); + /* * Now that xl_prev has been filled in, calculate CRC of the record * header. @@ -923,6 +931,11 @@ XLogInsertRecord(XLogRecData *rdata, class == WALINSERT_SPECIAL_SWITCH, rdata, StartPos, EndPos, insertTLI); + if (islargerecord) + { + SetPrevRecPtr(EndPos, StartPos, insertTLI, true); + } + /* * Unless record is flagged as not important, update LSN of last * important record in the current slot. When holding all locks, just @@ -1105,13 +1118,11 @@ XLogInsertRecord(XLogRecData *rdata, * inline. We use pg_attribute_always_inline here to try to convince it. */ static pg_attribute_always_inline void -ReserveXLogInsertLocation(int size, XLogRecPtr *StartPos, XLogRecPtr *EndPos, - XLogRecPtr *PrevPtr) +ReserveXLogInsertLocation(int size, XLogRecPtr *StartPos, XLogRecPtr *EndPos) { XLogCtlInsert *Insert = &XLogCtl->Insert; uint64 startbytepos; uint64 endbytepos; - uint64 prevbytepos; size = MAXALIGN(size); @@ -1128,19 +1139,11 @@ ReserveXLogInsertLocation(int size, XLogRecPtr *StartPos, XLogRecPtr *EndPos, * because the usable byte position doesn't include any headers, reserving * X bytes from WAL is almost as simple as "CurrBytePos += X". */ - SpinLockAcquire(&Insert->insertpos_lck); - - startbytepos = Insert->CurrBytePos; + startbytepos = pg_atomic_fetch_add_u64(&Insert->CurrBytePos, size); endbytepos = startbytepos + size; - prevbytepos = Insert->PrevBytePos; - Insert->CurrBytePos = endbytepos; - Insert->PrevBytePos = startbytepos; - - SpinLockRelease(&Insert->insertpos_lck); *StartPos = XLogBytePosToRecPtr(startbytepos); *EndPos = XLogBytePosToEndRecPtr(endbytepos); - *PrevPtr = XLogBytePosToRecPtr(prevbytepos); /* * Check that the conversions between "usable byte positions" and @@ -1148,7 +1151,6 @@ ReserveXLogInsertLocation(int size, XLogRecPtr *StartPos, XLogRecPtr *EndPos, */ Assert(XLogRecPtrToBytePos(*StartPos) == startbytepos); Assert(XLogRecPtrToBytePos(*EndPos) == endbytepos); - Assert(XLogRecPtrToBytePos(*PrevPtr) == prevbytepos); } /* @@ -1161,12 +1163,11 @@ ReserveXLogInsertLocation(int size, XLogRecPtr *StartPos, XLogRecPtr *EndPos, * reserving any space, and the function returns false. */ static bool -ReserveXLogSwitch(XLogRecPtr *StartPos, XLogRecPtr *EndPos, XLogRecPtr *PrevPtr) +ReserveXLogSwitch(XLogRecPtr *StartPos, XLogRecPtr *EndPos) { XLogCtlInsert *Insert = &XLogCtl->Insert; uint64 startbytepos; uint64 endbytepos; - uint64 prevbytepos; uint32 size = MAXALIGN(SizeOfXLogRecord); XLogRecPtr ptr; uint32 segleft; @@ -1179,7 +1180,7 @@ ReserveXLogSwitch(XLogRecPtr *StartPos, XLogRecPtr *EndPos, XLogRecPtr *PrevPtr) */ SpinLockAcquire(&Insert->insertpos_lck); - startbytepos = Insert->CurrBytePos; + startbytepos = pg_atomic_read_u64(&Insert->CurrBytePos); ptr = XLogBytePosToEndRecPtr(startbytepos); if (XLogSegmentOffset(ptr, wal_segment_size) == 0) @@ -1190,7 +1191,6 @@ ReserveXLogSwitch(XLogRecPtr *StartPos, XLogRecPtr *EndPos, XLogRecPtr *PrevPtr) } endbytepos = startbytepos + size; - prevbytepos = Insert->PrevBytePos; *StartPos = XLogBytePosToRecPtr(startbytepos); *EndPos = XLogBytePosToEndRecPtr(endbytepos); @@ -1202,17 +1202,13 @@ ReserveXLogSwitch(XLogRecPtr *StartPos, XLogRecPtr *EndPos, XLogRecPtr *PrevPtr) *EndPos += segleft; endbytepos = XLogRecPtrToBytePos(*EndPos); } - Insert->CurrBytePos = endbytepos; - Insert->PrevBytePos = startbytepos; + pg_atomic_write_u64(&Insert->CurrBytePos, endbytepos); SpinLockRelease(&Insert->insertpos_lck); - *PrevPtr = XLogBytePosToRecPtr(prevbytepos); - Assert(XLogSegmentOffset(*EndPos, wal_segment_size) == 0); Assert(XLogRecPtrToBytePos(*EndPos) == endbytepos); Assert(XLogRecPtrToBytePos(*StartPos) == startbytepos); - Assert(XLogRecPtrToBytePos(*PrevPtr) == prevbytepos); return true; } @@ -1236,7 +1232,7 @@ CopyXLogRecordToWAL(int write_len, bool isLogSwitch, XLogRecData *rdata, * inserting to. */ CurrPos = StartPos; - currpos = GetXLogBuffer(CurrPos, tli); + currpos = GetXLogBuffer(CurrPos, CurrPos, tli); freespace = INSERT_FREESPACE(CurrPos); /* @@ -1273,7 +1269,7 @@ CopyXLogRecordToWAL(int write_len, bool isLogSwitch, XLogRecData *rdata, * page was initialized, in AdvanceXLInsertBuffer, and we're the * only backend that needs to set the contrecord flag. */ - currpos = GetXLogBuffer(CurrPos, tli); + currpos = GetXLogBuffer(CurrPos, CurrPos, tli); pagehdr = (XLogPageHeader) currpos; pagehdr->xlp_rem_len = write_len - written; pagehdr->xlp_info |= XLP_FIRST_IS_CONTRECORD; @@ -1346,7 +1342,7 @@ CopyXLogRecordToWAL(int write_len, bool isLogSwitch, XLogRecData *rdata, * (which itself calls the two methods we need) to get the pointer * and zero most of the page. Then we just zero the page header. */ - currpos = GetXLogBuffer(CurrPos, tli); + currpos = GetXLogBuffer(CurrPos, CurrPos, tli); MemSet(currpos, 0, SizeOfXLogShortPHD); CurrPos += XLOG_BLCKSZ; @@ -1364,6 +1360,90 @@ CopyXLogRecordToWAL(int write_len, bool isLogSwitch, XLogRecData *rdata, errmsg_internal("space reserved for WAL record does not match what was written")); } +static XLogRecPtr +GetPrevRecPtr(XLogRecPtr hdr, TimeLineID tli) +{ + uint64 xlprevbytepos; + char *xl_prev_pos; + int freespace; + + XLogRecPtr res; + XLogRecPtr xl_prev_ptr; + + xlprevbytepos = XLogRecPtrToBytePos(hdr) + offsetof(XLogRecord, xl_prev); + xl_prev_ptr = XLogBytePosToRecPtr(xlprevbytepos); + xl_prev_pos = GetXLogBuffer(xl_prev_ptr, hdr, tli); + freespace = INSERT_FREESPACE(xl_prev_ptr); + + if (freespace >= sizeof(XLogRecPtr)) + { + res = *((XLogRecPtr *) xl_prev_pos); + } + else + { + char *res_data; + + res_data = (char *) &res; + memcpy(res_data, xl_prev_pos, freespace); + + if (XLogSegmentOffset(xl_prev_ptr + freespace, wal_segment_size) == 0) + { + xl_prev_pos += SizeOfXLogLongPHD; + } + else + { + xl_prev_pos += SizeOfXLogShortPHD; + } + + memcpy(res_data + freespace, xl_prev_pos, sizeof(XLogRecPtr) - freespace); + } + + return res; +} + +static void +SetPrevRecPtr(XLogRecPtr hdr, XLogRecPtr xl_prev_to_insert, TimeLineID tli, bool in_order) +{ + uint64 xlprevbytepos; + char *xl_prev_pos; + int freespace; + + XLogRecPtr xl_prev_ptr; + + xlprevbytepos = XLogRecPtrToBytePos(hdr) + offsetof(XLogRecord, xl_prev); + xl_prev_ptr = XLogBytePosToRecPtr(xlprevbytepos); + + if (in_order) + xl_prev_pos = GetXLogBuffer(xl_prev_ptr, xl_prev_ptr, tli); + else + xl_prev_pos = GetXLogBuffer(xl_prev_ptr, xl_prev_to_insert, tli); + + freespace = INSERT_FREESPACE(xl_prev_ptr); + + if (freespace >= sizeof(XLogRecPtr)) + { + *((XLogRecPtr *) xl_prev_pos) = xl_prev_to_insert; + } + else + { + char *xl_prev_to_insert_data; + + xl_prev_to_insert_data = (char *) &xl_prev_to_insert; + memcpy(xl_prev_pos, xl_prev_to_insert_data, freespace); + + if (XLogSegmentOffset(xl_prev_ptr + freespace, wal_segment_size) == 0) + { + xl_prev_pos += SizeOfXLogLongPHD; + } + else + { + xl_prev_pos += SizeOfXLogShortPHD; + } + + memcpy(xl_prev_pos, xl_prev_to_insert_data + freespace, sizeof(XLogRecPtr) - freespace); + } +} + /* * Acquire a WAL insertion lock, for inserting to WAL. */ @@ -1522,9 +1602,7 @@ WaitXLogInsertionsToFinish(XLogRecPtr upto) return inserted; /* Read the current insert position */ - SpinLockAcquire(&Insert->insertpos_lck); - bytepos = Insert->CurrBytePos; - SpinLockRelease(&Insert->insertpos_lck); + bytepos = pg_atomic_read_u64(&Insert->CurrBytePos); reservedUpto = XLogBytePosToEndRecPtr(bytepos); /* @@ -1629,7 +1707,7 @@ WaitXLogInsertionsToFinish(XLogRecPtr upto) * later, because older buffers might be recycled already) */ static char * -GetXLogBuffer(XLogRecPtr ptr, TimeLineID tli) +GetXLogBuffer(XLogRecPtr ptr, XLogRecPtr upto, TimeLineID tli) { int idx; XLogRecPtr endptr; @@ -1690,14 +1768,14 @@ GetXLogBuffer(XLogRecPtr ptr, TimeLineID tli) * sure that it's initialized, before we let insertingAt to move past * the page header. */ - if (ptr % XLOG_BLCKSZ == SizeOfXLogShortPHD && - XLogSegmentOffset(ptr, wal_segment_size) > XLOG_BLCKSZ) - initializedUpto = ptr - SizeOfXLogShortPHD; - else if (ptr % XLOG_BLCKSZ == SizeOfXLogLongPHD && - XLogSegmentOffset(ptr, wal_segment_size) < XLOG_BLCKSZ) - initializedUpto = ptr - SizeOfXLogLongPHD; + if (upto % XLOG_BLCKSZ == SizeOfXLogShortPHD && + XLogSegmentOffset(upto, wal_segment_size) > XLOG_BLCKSZ) + initializedUpto = upto - SizeOfXLogShortPHD; + else if (upto % XLOG_BLCKSZ == SizeOfXLogLongPHD && + XLogSegmentOffset(upto, wal_segment_size) < XLOG_BLCKSZ) + initializedUpto = upto - SizeOfXLogLongPHD; else - initializedUpto = ptr; + initializedUpto = upto; WALInsertLockUpdateInsertingAt(initializedUpto); @@ -6018,8 +6096,7 @@ StartupXLOG(void) * previous incarnation. */ Insert = &XLogCtl->Insert; - Insert->PrevBytePos = XLogRecPtrToBytePos(endOfRecoveryInfo->lastRec); - Insert->CurrBytePos = XLogRecPtrToBytePos(EndOfLog); + pg_atomic_write_u64(&Insert->CurrBytePos, XLogRecPtrToBytePos(EndOfLog)); /* * Tricky point here: lastPage contains the *last* block that the LastRec @@ -6053,6 +6130,7 @@ StartupXLOG(void) */ XLogCtl->InitializedUpTo = EndOfLog; } + SetPrevRecPtr(EndOfLog, endOfRecoveryInfo->lastRec, newTLI, true); /* * Update local and shared status. This is OK to do without any locks @@ -7005,7 +7083,7 @@ CreateCheckPoint(int flags) if (shutdown) { - XLogRecPtr curInsert = XLogBytePosToRecPtr(Insert->CurrBytePos); + XLogRecPtr curInsert = XLogBytePosToRecPtr(pg_atomic_read_u64(&Insert->CurrBytePos)); /* * Compute new REDO record ptr = location of next XLOG record. @@ -7473,7 +7551,7 @@ CreateOverwriteContrecordRecord(XLogRecPtr aborted_lsn, XLogRecPtr pagePtr, * insertion lock is just pro forma. */ WALInsertLockAcquire(); - pagehdr = (XLogPageHeader) GetXLogBuffer(pagePtr, newTLI); + pagehdr = (XLogPageHeader) GetXLogBuffer(pagePtr, pagePtr, newTLI); pagehdr->xlp_info |= XLP_FIRST_IS_OVERWRITE_CONTRECORD; WALInsertLockRelease(); @@ -9437,9 +9515,7 @@ GetXLogInsertRecPtr(void) XLogCtlInsert *Insert = &XLogCtl->Insert; uint64 current_bytepos; - SpinLockAcquire(&Insert->insertpos_lck); - current_bytepos = Insert->CurrBytePos; - SpinLockRelease(&Insert->insertpos_lck); + current_bytepos = pg_atomic_read_u64(&Insert->CurrBytePos); return XLogBytePosToRecPtr(current_bytepos); } -- 2.43.0