commit 83b1e4fcd74b4dd6c6992395f21e4fe606c8e80d Author: Heikki Linnakangas Date: Thu Jun 14 23:53:17 2012 +0300 Rebase code from xloginsert-noslots branch. This is based on xloginsert-scale18.patch, but instead of slots, use the xl_rem_len to indicate that a record has been fully written. diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index 3f5e0b2..0d0e799 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -42,6 +42,7 @@ #include "postmaster/startup.h" #include "replication/walreceiver.h" #include "replication/walsender.h" +#include "storage/barrier.h" #include "storage/bufmgr.h" #include "storage/fd.h" #include "storage/ipc.h" @@ -261,14 +262,26 @@ XLogRecPtr XactLastRecEnd = {0, 0}; * (which is almost but not quite the same as a pointer to the most recent * CHECKPOINT record). We update this from the shared-memory copy, * XLogCtl->Insert.RedoRecPtr, whenever we can safely do so (ie, when we - * hold the Insert lock). See XLogInsert for details. We are also allowed - * to update from XLogCtl->Insert.RedoRecPtr if we hold the info_lck; + * hold the insertpos lock). See XLogInsert for details. We are also allowed + * to update from XLogCtl->RedoRecPtr if we hold the info_lck; * see GetRedoRecPtr. A freshly spawned backend obtains the value during * InitXLOGAccess. */ static XLogRecPtr RedoRecPtr; /* + * doPageWrites is this backend's local copy of the Insert->fullPageWrites || + * Insert->forcePageWrites. It is refreshed at every insertion. + */ +static bool doPageWrites; + +/* + * FinalizedUpto is this backend's local copy of XLogCtl->Insert.FinalizedUpto. + * Everything before this is CRC'd and ready for writing out. + */ +static XLogRecPtr FinalizedUpto = { 0, 0 }; + +/* * RedoStartLSN points to the checkpoint's REDO location which is specified * in a backup label file, backup history file or control file. In standby * mode, XLOG streaming usually starts from the position where an invalid @@ -300,10 +313,15 @@ static XLogRecPtr RedoStartLSN = {0, 0}; * (protected by info_lck), but we don't need to cache any copies of it. * * info_lck is only held long enough to read/update the protected variables, - * so it's a plain spinlock. The other locks are held longer (potentially - * over I/O operations), so we use LWLocks for them. These locks are: + * so it's a plain spinlock. insertpos_lck protects the current logical + * insert location, ie. the head of reserved WAL space. The other locks are + * held longer (potentially over I/O operations), so we use LWLocks for them. + * These locks are: * - * WALInsertLock: must be held to insert a record into the WAL buffers. + * WALBufMappingLock: must be held to replace a page in the WAL buffer cache. + * This is only held while initializing and changing the mapping. If the + * contents of the buffer being replaced haven't been written yet, the mapping + * lock is released while the write is done, and reacquired afterwards. * * WALWriteLock: must be held to write WAL buffers to disk (XLogWrite or * XLogFlush). @@ -315,6 +333,93 @@ static XLogRecPtr RedoStartLSN = {0, 0}; * only one checkpointer at a time; currently, with all checkpoints done by * the checkpointer, this is just pro forma). * + * WALInsertShareLocks: This lock is partitioned into multiple lwlocks. To + * hold it in share mode, it's enough to hold any of the lwlocks in share mode, + * but to hold it in exclusive mode, you must grab all the lwlocks. It must + * be held in share-mode while inserting a new XLOG record, and in exclusive + * mode when changing RedoRecPtr or fullPageWrites. Those fields determine + * whether full-page images are included in a record, and they change very + * seldom, so we prefer to be fast and non-contended when they need to be + * read, and slow when they're changed. + * + * + * Inserting a new WAL record is a three-step process: + * + * 1. Reserve the right amount of space from the WAL. The current head of + * reserved space is kept in Insert->CurrBytePos, and is protected by + * insertpos_lck. Try to keep this section as short as possible, + * insertpos_lck can be heavily contended on a busy system. + * + * 2. Copy the record to the reserved WAL space. This involves finding the + * correct WAL buffer containing the reserved space, and copying the + * record in place. This can be done concurrently in multiple processes. + * + * 3. Finalize the record by filling in xl_prev, and updating the CRC with it. + * This can be done by another process, long after step 2. This only needs + * to be done just before the record is flushed to disk, so it's done in + * bulk at that point. + * + * To allow as much parallelism as possible, the conteneded portion of step 1 + * is performed while only holding a spinlock. The duration the spinlock + * needs to be held is minimized by minimizing the calculations that have to + * be done while holding the lock. The current tip of reserved WAL is kept + * in CurrBytePos, as a byte position that only counts "usable" bytes in WAL, + * that is, it excludes all WAL page headers. The mapping between "usable" byte + * positions and physical positions (XLogRecPtrs) can be done outside the + * locked region, and because the usable byte position doesn't include any + * headers, reserving X bytes from WAL is simply "CurrBytePos += X". On + * platforms that have an atomic 64-bit fetch-and-add instruction, we don't + * even need a spinlock (XXX: not implemented yet - ATM spinlock is always + * used). + * + * Step 2 can usually be done completely in parallel. If the required WAL + * page is not initialized yet, you have to grab WALBufMappingLock to + * initialize it, but we pre-initialize WAL buffers in the WAL writer to + * avoid that from happening in the critical path. + * + * In step 2, the xl_prev field is left at 0/0, because even though we've + * reserved a slice of WAL space for the record, we don't know where the + * previous record began. We could keep track of that along with CurrBytePos, + * in step 1, but then it would no longer be possible to implement it with + * an atomic fetch-and-add instruction. So at step 3, we finalize all the + * records by filling in xl_prev, and calculating the final CRC that includes + * xl_prev as well. Finalization starts from the end of the last finalized + * records, and walks the chain of WAL records until it hits a record with + * xl_tot_len == 0. Setting xl_tot_len is a sign that the record is fully + * written - a memory barrier ensures that xl_tot_len is not seen by other + * processes before the rest of the record. If the record doesn't fit on the + * page, setting xl_tot_len indicates that the record is fully written up to + * the page boundary, and on the next page, setting XLP_FIRST_IS_CONTRECORD + * acts as a signal that the continued part is fully written to the page. + * + * XXX: There is currently no good mechanism to wait for step 2 of an + * insertion to finish. Step 3 busy-loops. In the previous version of this + * patch, which used "insertion slots", the slot included a linked list of + * PGPROCs waiting for the slot to finish inserting, similar to LWLocks. + * We'll probably need to add something like that, busy-waiting is not good. + * + * + * Deadlock analysis + * ----------------- + * + * It's important to call WaitXLogInsertionsToFinish() *before* acquiring + * WALWriteLock. Otherwise you might get stuck waiting for an insertion to + * finish (or at least advance to next uninitialized page), while you're + * holding WALWriteLock. That would be bad, because the backend you're waiting + * for might need to acquire WALWriteLock, too, to evict an old buffer, so + * you'd get deadlock. + * + * WaitXLogInsertionsToFinish() will not get stuck indefinitely, as long as + * it's called with a location that's known to be already allocated in the WAL + * buffers. Calling it with the position of a record you've already inserted + * satisfies that condition, so the common pattern: + * + * recptr = XLogInsert(...) + * XLogFlush(recptr) + * + * is safe. It can't get stuck, because an insertion to a WAL page that's + * already initialized in cache can always proceed without waiting on a lock. + * *---------- */ @@ -335,12 +440,26 @@ typedef struct XLogwrtResult */ typedef struct XLogCtlInsert { - XLogRecPtr PrevRecord; /* start of previously-inserted record */ - int curridx; /* current block index in cache */ - XLogPageHeader currpage; /* points to header of block in cache */ - char *currpos; /* current insertion point in cache */ - XLogRecPtr RedoRecPtr; /* current redo point for insertions */ - bool forcePageWrites; /* forcing full-page writes for PITR? */ + slock_t insertpos_lck; /* protects CurrBytePos */ + + /* + * CurrBytePos is the very tip of the reserved WAL space at the moment. + * The next record will be inserted there. + */ + uint64 CurrBytePos; + + /* + * These fields track the progress of record finalization. FinalizedUpto + * points to the end of fully finalized portion - everything before it + * is ready to be written to disk. LastFinalizedRecord points to the + * beginning of the last finalized record. When the next record is + * finalized, it is written to the xl_prev of the next record. If + * ExpectingContRecord is true, we are stopped at a page boundary, in the + * middle of a WAL record. These fields are protected by WALInsertTailLock. + */ + XLogRecPtr FinalizedUpto; + XLogRecPtr LastFinalizedRecord; + bool ExpectingContRecord; /* * fullPageWrites is the master copy used by all backends to determine @@ -348,7 +467,11 @@ typedef struct XLogCtlInsert * one. This is required because, when full_page_writes is changed * by SIGHUP, we must WAL-log it before it actually affects * WAL-logging by backends. Checkpointer sets at startup or after SIGHUP. + * + * These fields are protected by WALInsertShareLocks. */ + XLogRecPtr RedoRecPtr; /* current redo point for insertions */ + bool forcePageWrites; /* forcing full-page writes for PITR? */ bool fullPageWrites; /* @@ -372,16 +495,21 @@ typedef struct XLogCtlWrite pg_time_t lastSegSwitchTime; /* time of last xlog segment switch */ } XLogCtlWrite; + /* * Total shared-memory state for XLOG. */ typedef struct XLogCtlData { - /* Protected by WALInsertLock: */ + /* + * Note: Insert must be the first field in the struct or it won't be + * aligned to a cache-line boundary like we want it to be. + */ XLogCtlInsert Insert; /* Protected by info_lck: */ XLogwrtRqst LogwrtRqst; + XLogRecPtr RedoRecPtr; /* a recent copy of Insert->RedoRecPtr */ uint32 ckptXidEpoch; /* nextXID & epoch of latest checkpoint */ TransactionId ckptXid; XLogRecPtr asyncXactLSN; /* LSN of newest async commit/abort */ @@ -397,9 +525,18 @@ typedef struct XLogCtlData XLogwrtResult LogwrtResult; /* + * To change curridx and the identity of a buffer, you need to hold + * WALBufMappingLock. To change the identity of a buffer that's still + * dirty, the old page needs to be written out first, and for that you + * need WALWriteLock, and you need to ensure that there's no in-progress + * insertions to the page by calling WaitXLogInsertionsToFinish(). + */ + int curridx; /* latest initialized block index in cache */ + + /* * These values do not change after startup, although the pointed-to pages * and xlblocks values certainly do. Permission to read/write the pages - * and xlblocks values depends on WALInsertLock and WALWriteLock. + * and xlblocks values depends on WALBufMappingLock and WALWriteLock. */ char *pages; /* buffers for unwritten XLOG pages */ XLogRecPtr *xlblocks; /* 1st byte ptr-s + XLOG_BLCKSZ */ @@ -479,30 +616,37 @@ static XLogCtlData *XLogCtl = NULL; static ControlFileData *ControlFile = NULL; /* - * Macros for managing XLogInsert state. In most cases, the calling routine - * has local copies of XLogCtl->Insert and/or XLogCtl->Insert->curridx, - * so these are passed as parameters instead of being fetched via XLogCtl. + * Calculate the amount of space left on the page after 'endptr'. + * Beware multiple evaluation! */ +#define INSERT_FREESPACE(endptr) \ + (((endptr).xrecoff % XLOG_BLCKSZ == 0) ? 0 : (XLOG_BLCKSZ - (endptr).xrecoff % XLOG_BLCKSZ)) -/* Free space remaining in the current xlog page buffer */ -#define INSERT_FREESPACE(Insert) \ - (XLOG_BLCKSZ - ((Insert)->currpos - (char *) (Insert)->currpage)) +/* + * Macros to advance to next buffer index and insertion slot. + */ +#define NextBufIdx(idx) \ + (((idx) == XLogCtl->XLogCacheBlck) ? 0 : ((idx) + 1)) -/* Construct XLogRecPtr value for current insertion point */ -#define INSERT_RECPTR(recptr,Insert,curridx) \ - do { \ - (recptr).xlogid = XLogCtl->xlblocks[curridx].xlogid; \ - (recptr).xrecoff = \ - XLogCtl->xlblocks[curridx].xrecoff - INSERT_FREESPACE(Insert); \ - if (XLogCtl->xlblocks[curridx].xrecoff == 0) \ - (recptr).xlogid = XLogCtl->xlblocks[curridx].xlogid - 1; \ - } while(0) +/* + * XLogRecPtrToBufIdx returns the index of the WAL buffer that holds, or + * would hold if it was in cache, the page containing 'recptr'. + * + * XLogRecEndPtrToBufIdx is the same, but a pointer to the first byte of a + * page is taken to mean the previous page. + */ +#define XLogRecPtrToBufIdx(recptr) \ + (((((((uint64) (recptr).xlogid) << 32) + (recptr).xrecoff)) / XLOG_BLCKSZ) % (XLogCtl->XLogCacheBlck + 1)) -#define PrevBufIdx(idx) \ - (((idx) == 0) ? XLogCtl->XLogCacheBlck : ((idx) - 1)) +#define XLogRecEndPtrToBufIdx(recptr) \ + (((((((uint64) (recptr).xlogid) << 32) + (recptr).xrecoff - 1)) / XLOG_BLCKSZ) % (XLogCtl->XLogCacheBlck + 1)) -#define NextBufIdx(idx) \ - (((idx) == XLogCtl->XLogCacheBlck) ? 0 : ((idx) + 1)) +/* + * These are the number of bytes usable in a WAL page and segment, excluding + * page headers. + */ +#define UsableBytesInPage (XLOG_BLCKSZ - SizeOfXLogShortPHD) +#define UsableBytesInSegment ((XLOG_SEG_SIZE / XLOG_BLCKSZ) * UsableBytesInPage - (SizeOfXLogLongPHD - SizeOfXLogShortPHD)) /* * Private, possibly out-of-date copy of shared LogwrtResult. @@ -625,9 +769,9 @@ static void KeepLogSeg(XLogRecPtr recptr, XLogSegNo *logSegNo); static bool XLogCheckBuffer(XLogRecData *rdata, bool doPageWrites, XLogRecPtr *lsn, BkpBlock *bkpb); -static bool AdvanceXLInsertBuffer(bool new_segment); +static void AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic); static bool XLogCheckpointNeeded(XLogSegNo new_segno); -static void XLogWrite(XLogwrtRqst WriteRqst, bool flexible, bool xlog_switch); +static void XLogWrite(XLogwrtRqst WriteRqst, bool flexible); static bool InstallXLogFileSegment(XLogSegNo *segno, char *tmppath, bool find_free, int *max_advance, bool use_lock); @@ -674,6 +818,75 @@ static bool read_backup_label(XLogRecPtr *checkPointLoc, static void rm_redo_error_callback(void *arg); static int get_sync_bit(int method); +static void CopyXLogRecordToWAL(int write_len, bool isLogSwitch, + XLogRecData *rdata, + XLogRecPtr StartPos, XLogRecPtr EndPos); +static void ReserveXLogInsertLocation(int size, XLogRecPtr *StartPos, + XLogRecPtr *EndPos); +static bool ReserveXLogSwitch(XLogRecPtr *StartPos, XLogRecPtr *EndPos); +static XLogRecPtr WaitXLogInsertionsToFinish(XLogRecPtr upto); +static char *GetXLogBuffer(XLogRecPtr ptr, bool failok); +static XLogRecPtr XLogBytePosToRecPtr(uint64 bytepos); +static XLogRecPtr XLogBytePosToEndRecPtr(uint64 bytepos); +static uint64 XLogRecPtrToBytePos(XLogRecPtr ptr); + +/* + * Equivalent of LWLockAcquire() for the partitioned WALInsertShareLock. + */ +static void +WALInsertLockAcquire(LWLockMode mode) +{ + int lockid; + + if (mode == LW_EXCLUSIVE) + { + /* + * To acquire the lock in exclusive mode, need to hold all the + * partition locks. + */ + for (lockid = FirstWALInsertShareLock; lockid <= LastWALInsertShareLock; lockid++) + { + LWLockAcquire(lockid, LW_EXCLUSIVE); + } + } + else + { + /* + * Grab one of the partitioned locks. It doesn't matter which one, + * but to avoid contention, it's good if different processes choose + * different locks. + */ + lockid = FirstWALInsertShareLock + + (MyProc->pgprocno % (LastWALInsertShareLock - FirstWALInsertShareLock + 1)); + LWLockAcquire(lockid, LW_SHARED); + } +} + +/* + * Equivalent of LWLockRelease() for the partitioned WALInsertShareLock. + */ +static void +WALInsertLockRelease(LWLockMode mode) +{ + int lockid; + + if (mode == LW_EXCLUSIVE) + { + for (lockid = FirstWALInsertShareLock; lockid <= LastWALInsertShareLock; lockid++) + { + LWLockRelease(lockid); + } + } + else + { + /* + * this calculation better match the one used when the lock was + * acquired. + */ + lockid = FirstWALInsertShareLock + (MyProc->pgprocno % (LastWALInsertShareLock - FirstWALInsertShareLock + 1)); + LWLockRelease(lockid); + } +} /* * Insert an XLOG record having the specified RMID and info bytes, @@ -693,11 +906,7 @@ static int get_sync_bit(int method); XLogRecPtr XLogInsert(RmgrId rmid, uint8 info, XLogRecData *rdata) { - XLogCtlInsert *Insert = &XLogCtl->Insert; - XLogRecPtr RecPtr; - XLogRecPtr WriteRqst; - uint32 freespace; - int curridx; + volatile XLogCtlInsert *Insert = &XLogCtl->Insert; XLogRecData *rdt; XLogRecData *rdt_lastnormal; Buffer dtbuf[XLR_MAX_BKP_BLOCKS]; @@ -712,12 +921,14 @@ XLogInsert(RmgrId rmid, uint8 info, XLogRecData *rdata) uint32 len, write_len; unsigned i; - bool updrqst; - bool doPageWrites; bool isLogSwitch = (rmid == RM_XLOG_ID && info == XLOG_SWITCH); - uint8 info_orig = info; static XLogRecord *rechdr; + XLogRecPtr StartPos; + XLogRecPtr EndPos; + /* + * On the first call, allocate a buffer to hold the xlog record. + */ if (rechdr == NULL) { rechdr = malloc(SizeOfXLogRecord); @@ -742,40 +953,33 @@ XLogInsert(RmgrId rmid, uint8 info, XLogRecData *rdata) */ if (IsBootstrapProcessingMode() && rmid != RM_XLOG_ID) { - RecPtr.xlogid = 0; - RecPtr.xrecoff = SizeOfXLogLongPHD; /* start of 1st chkpt record */ - return RecPtr; + EndPos.xlogid = 0; + EndPos.xrecoff = SizeOfXLogLongPHD; /* start of 1st chkpt record */ + return EndPos; } /* * Here we scan the rdata chain, to determine which buffers must be backed * up. * - * We may have to loop back to here if a race condition is detected below. - * We could prevent the race by doing all this work while holding the - * insert lock, but it seems better to avoid doing CRC calculations while - * holding the lock. - * * We add entries for backup blocks to the chain, so that they don't * need any special treatment in the critical section where the chunks are - * copied into the WAL buffers. Those entries have to be unlinked from the - * chain if we have to loop back here. + * copied into the WAL buffers. + * + * First acquire WALInsertShareLock, to prevent RedoRecPtr and + * force/fullPageWrites flags from changing. */ -begin:; + WALInsertLockAcquire(isLogSwitch ? LW_EXCLUSIVE : LW_SHARED); + + doPageWrites = Insert->forcePageWrites || Insert->fullPageWrites; + RedoRecPtr = Insert->RedoRecPtr; + for (i = 0; i < XLR_MAX_BKP_BLOCKS; i++) { dtbuf[i] = InvalidBuffer; dtbuf_bkp[i] = false; } - /* - * Decide if we need to do full-page writes in this XLOG record: true if - * full_page_writes is on or we have a PITR request for it. Since we - * don't yet have the insert lock, fullPageWrites and forcePageWrites - * could change under us, but we'll recheck them once we have the lock. - */ - doPageWrites = Insert->fullPageWrites || Insert->forcePageWrites; - len = 0; for (rdt = rdata;;) { @@ -831,8 +1035,7 @@ begin:; * NOTE: We disallow len == 0 because it provides a useful bit of extra * error checking in ReadRecord. This means that all callers of * XLogInsert must supply at least some not-in-a-buffer data. However, we - * make an exception for XLOG SWITCH records because we don't want them to - * ever cross a segment boundary. + * make an exception for XLOG SWITCH records. */ if (len == 0 && !isLogSwitch) elog(PANIC, "invalid xlog record length %u", len); @@ -840,9 +1043,7 @@ begin:; /* * Make additional rdata chain entries for the backup blocks, so that we * don't need to special-case them in the write loop. This modifies the - * original rdata chain, but we keep a pointer to the last regular entry, - * rdt_lastnormal, so that we can undo this if we have to loop back to the - * beginning. + * original rdata chain. * * At the exit of this loop, write_len includes the backup block data. * @@ -912,15 +1113,23 @@ begin:; COMP_CRC32(rdata_crc, rdt->data, rdt->len); /* - * Construct record header (prev-link and CRC are filled in later), and - * make that the first chunk in the chain. + * Construct record header (prev-link is filled in later, in record + * finalization), and make that the first chunk in the chain. */ rechdr->xl_xid = GetCurrentTransactionIdIfAny(); rechdr->xl_tot_len = SizeOfXLogRecord + write_len; rechdr->xl_len = len; /* doesn't include backup blocks */ rechdr->xl_info = info; rechdr->xl_rmid = rmid; + rechdr->xl_prev = InvalidXLogRecPtr; + /* + * The CRC calculated here doesn't include the correct prev-link yet. + * It will be updated in record finalization. + */ + COMP_CRC32(rdata_crc, ((char *) rechdr), offsetof(XLogRecord, xl_prev)); + rechdr->xl_crc = rdata_crc; + /* Make the record header the first chunk in the chain */ hdr_rdt.next = rdata; hdr_rdt.data = (char *) rechdr; hdr_rdt.len = SizeOfXLogRecord; @@ -929,118 +1138,82 @@ begin:; START_CRIT_SECTION(); - /* Now wait to get insert lock */ - LWLockAcquire(WALInsertLock, LW_EXCLUSIVE); - /* - * Check to see if my RedoRecPtr is out of date. If so, may have to go - * back and recompute everything. This can only happen just after a - * checkpoint, so it's better to be slow in this case and fast otherwise. - * - * If we aren't doing full-page writes then RedoRecPtr doesn't actually - * affect the contents of the XLOG record, so we'll update our local copy - * but not force a recomputation. + * Reserve space for the record from the WAL, and copy the record there. */ - if (!XLByteEQ(RedoRecPtr, Insert->RedoRecPtr)) + if (isLogSwitch) { - Assert(XLByteLT(RedoRecPtr, Insert->RedoRecPtr)); - RedoRecPtr = Insert->RedoRecPtr; + if (ReserveXLogSwitch(&StartPos, &EndPos)) + { + WaitXLogInsertionsToFinish(StartPos); - if (doPageWrites) + CopyXLogRecordToWAL(write_len, isLogSwitch, &hdr_rdt, + StartPos, EndPos); + } + else { - for (i = 0; i < XLR_MAX_BKP_BLOCKS; i++) - { - if (dtbuf[i] == InvalidBuffer) - continue; - if (dtbuf_bkp[i] == false && - XLByteLE(dtbuf_lsn[i], RedoRecPtr)) - { - /* - * Oops, this buffer now needs to be backed up, but we - * didn't think so above. Start over. - */ - LWLockRelease(WALInsertLock); - END_CRIT_SECTION(); - rdt_lastnormal->next = NULL; - info = info_orig; - goto begin; - } - } + /* + * The current insert location was already exactly at the beginning + * of a segment, so there's no need to switch. + */ } } - - /* - * Also check to see if fullPageWrites or forcePageWrites was just turned on; - * if we weren't already doing full-page writes then go back and recompute. - * (If it was just turned off, we could recompute the record without full pages, - * but we choose not to bother.) - */ - if ((Insert->fullPageWrites || Insert->forcePageWrites) && !doPageWrites) + else { - /* Oops, must redo it with full-page data. */ - LWLockRelease(WALInsertLock); - END_CRIT_SECTION(); - rdt_lastnormal->next = NULL; - info = info_orig; - goto begin; + ReserveXLogInsertLocation(write_len, &StartPos, &EndPos); + + /* And copy the record there. */ + CopyXLogRecordToWAL(write_len, isLogSwitch, &hdr_rdt, StartPos, EndPos); } + END_CRIT_SECTION(); + + WALInsertLockRelease(isLogSwitch ? LW_EXCLUSIVE : LW_SHARED); /* - * If the current page is completely full, the record goes to the next - * page, right after the page header. + * Update shared LogwrtRqst.Write, if we crossed page boundary. */ - updrqst = false; - freespace = INSERT_FREESPACE(Insert); - if (freespace == 0) + if (StartPos.xrecoff / XLOG_BLCKSZ != EndPos.xrecoff / XLOG_BLCKSZ) { - updrqst = AdvanceXLInsertBuffer(false); - freespace = INSERT_FREESPACE(Insert); - } + /* use volatile pointer to prevent code rearrangement */ + volatile XLogCtlData *xlogctl = XLogCtl; - /* Compute record's XLOG location */ - curridx = Insert->curridx; - INSERT_RECPTR(RecPtr, Insert, curridx); + SpinLockAcquire(&xlogctl->info_lck); + /* advance global request to include new block(s) */ + if (XLByteLT(xlogctl->LogwrtRqst.Write, EndPos)) + xlogctl->LogwrtRqst.Write = EndPos; + /* update local result copy while I have the chance */ + LogwrtResult = xlogctl->LogwrtResult; + SpinLockRelease(&xlogctl->info_lck); + } /* - * If the record is an XLOG_SWITCH, and we are exactly at the start of a - * segment, we need not insert it (and don't want to because we'd like - * consecutive switch requests to be no-ops). Instead, make sure - * everything is written and flushed through the end of the prior segment, - * and return the prior segment's end address. + * If this was an XLOG_SWITCH record, flush the record and the empty + * padding space that fills the rest of the segment, and perform + * end-of-segment actions (eg, notifying archiver). */ - if (isLogSwitch && - (RecPtr.xrecoff % XLogSegSize) == SizeOfXLogLongPHD) + if (isLogSwitch) { - /* We can release insert lock immediately */ - LWLockRelease(WALInsertLock); - - RecPtr.xrecoff -= SizeOfXLogLongPHD; - - LWLockAcquire(WALWriteLock, LW_EXCLUSIVE); - LogwrtResult = XLogCtl->LogwrtResult; - if (!XLByteLE(RecPtr, LogwrtResult.Flush)) + TRACE_POSTGRESQL_XLOG_SWITCH(); + XLogFlush(EndPos); + /* + * Even though we reserved the rest of the segment for us, which is + * reflected in EndPos, we return a pointer to just the end of the + * xlog-switch record. + */ + if (StartPos.xrecoff % XLOG_SEG_SIZE != 0) { - XLogwrtRqst FlushRqst; - - FlushRqst.Write = RecPtr; - FlushRqst.Flush = RecPtr; - XLogWrite(FlushRqst, false, false); + EndPos = StartPos; + XLByteAdvance(EndPos, SizeOfXLogRecord); + if (StartPos.xrecoff / XLOG_BLCKSZ != EndPos.xrecoff / XLOG_BLCKSZ) + { + if (EndPos.xrecoff % XLOG_SEG_SIZE == EndPos.xrecoff % XLOG_BLCKSZ) + EndPos.xrecoff += SizeOfXLogLongPHD; + else + EndPos.xrecoff += SizeOfXLogShortPHD; + } } - LWLockRelease(WALWriteLock); - - END_CRIT_SECTION(); - - return RecPtr; } - /* Finish the record header */ - rechdr->xl_prev = Insert->PrevRecord; - - /* Now we can finish computing the record's CRC */ - COMP_CRC32(rdata_crc, (char *) rechdr, offsetof(XLogRecord, xl_crc)); - FIN_CRC32(rdata_crc); - rechdr->xl_crc = rdata_crc; - #ifdef WAL_DEBUG if (XLOG_DEBUG) { @@ -1048,7 +1221,7 @@ begin:; initStringInfo(&buf); appendStringInfo(&buf, "INSERT @ %X/%X: ", - RecPtr.xlogid, RecPtr.xrecoff); + EndPos.xlogid, EndPos.xrecoff); xlog_outrec(&buf, rechdr); if (rdata->data != NULL) { @@ -1060,165 +1233,741 @@ begin:; } #endif - /* Record begin of record in appropriate places */ - ProcLastRecPtr = RecPtr; - Insert->PrevRecord = RecPtr; + /* + * Update our global variables + */ + ProcLastRecPtr = StartPos; + XactLastRecEnd = EndPos; + + return EndPos; +} + +/* + * Subroutine of XLogInsert. Copies a WAL record to an already-reserved + * area in the WAL. + */ +static void +CopyXLogRecordToWAL(int write_len, bool isLogSwitch, + XLogRecData *rdata, + XLogRecPtr StartPos, XLogRecPtr EndPos) +{ + char *currpos; + int freespace; + int written; + XLogRecPtr CurrPos; + XLogRecord *rechdr; + uint32 *tot_len_p; + bool firstpage = true; + XLogPageHeader pagehdr = NULL; + + /* The first chunk should be the record header */ + rechdr = (XLogRecord *) rdata->data; + Assert(rdata->len == SizeOfXLogRecord); /* - * Append the data, including backup blocks if any + * When we write the record, we initially leave xl_tot_len at zero, + * and set it to the correct value only after copying the rest of the + * record in place. That way when a process sees that xl_tot_len is set, + * it knows that the record is fully copied in place (or the part that + * fits on this page, anyway). */ - rdata = &hdr_rdt; - while (write_len) + Assert(rechdr->xl_tot_len == write_len); + rechdr->xl_tot_len = 0; + + /* Get the right WAL page to start inserting to */ + CurrPos = StartPos; + currpos = GetXLogBuffer(CurrPos, false); + freespace = INSERT_FREESPACE(CurrPos); + + /* + * there should be enough space for at least the first field (xl_tot_len) + * on this page. + */ + Assert(freespace >= sizeof(uint32)); + tot_len_p = (uint32 *) currpos; + + /* Copy record data */ + written = 0; + while (rdata != NULL) { - while (rdata->data == NULL) - rdata = rdata->next; + char *rdata_data = rdata->data; + int rdata_len = rdata->len; - if (freespace > 0) + while (rdata_len > freespace) { - if (rdata->len > freespace) + /* + * Write what fits on this page, and continue on the next page. + */ + Assert (((uint64) currpos) % XLOG_BLCKSZ >= SizeOfXLogShortPHD || freespace == 0); + memcpy(currpos, rdata_data, freespace); + rdata_data += freespace; + rdata_len -= freespace; + written += freespace; + XLByteAdvance(CurrPos, freespace); + + /* + * Before we step to the next page, let others know that we're done + * copying to this page, by setting xl_tot_len (or + * XLP_FIRST_IS_CONT_RECORD, if we're continuing from previous + * page). + */ + pg_write_barrier(); + if (firstpage) { - memcpy(Insert->currpos, rdata->data, freespace); - rdata->data += freespace; - rdata->len -= freespace; - write_len -= freespace; + *tot_len_p = write_len; + firstpage = false; } else + pagehdr->xlp_info |= XLP_FIRST_IS_CONTRECORD; + + /* + * Get pointer to beginning of next page, and set the xlp_rem_len + * in the page header. We don't set XLP_FIRST_IS_CONTRECORD yet, + * that is used to signal that we're done copying, so it's done + * last. + * + * It's safe to set the contrecord flag and xlp_rem_len without a + * lock on the page. All the other flags were already set when the + * page was initialized, in AdvanceXLInsertBuffer, and we're the + * only backend that needs to set the contrecord flag. + */ + currpos = GetXLogBuffer(CurrPos, false); + pagehdr = (XLogPageHeader) currpos; + pagehdr->xlp_rem_len = write_len - written; + + /* skip over the page header */ + if (CurrPos.xrecoff % XLogSegSize == 0) { - memcpy(Insert->currpos, rdata->data, rdata->len); - freespace -= rdata->len; - write_len -= rdata->len; - Insert->currpos += rdata->len; - rdata = rdata->next; - continue; + CurrPos.xrecoff += SizeOfXLogLongPHD; + currpos += SizeOfXLogLongPHD; } + else + { + CurrPos.xrecoff += SizeOfXLogShortPHD; + currpos += SizeOfXLogShortPHD; + } + freespace = INSERT_FREESPACE(CurrPos); } - /* Use next buffer */ - updrqst = AdvanceXLInsertBuffer(false); - curridx = Insert->curridx; - /* Insert cont-record header */ - Insert->currpage->xlp_info |= XLP_FIRST_IS_CONTRECORD; - Insert->currpage->xlp_rem_len = write_len; - freespace = INSERT_FREESPACE(Insert); + Assert (((uint64) currpos) % XLOG_BLCKSZ >= SizeOfXLogShortPHD || rdata_len == 0); + memcpy(currpos, rdata_data, rdata_len); + currpos += rdata_len; + XLByteAdvance(CurrPos, rdata_len); + freespace -= rdata_len; + written += rdata_len; + + rdata = rdata->next; } + Assert(written == write_len); - /* Ensure next record will be properly aligned */ - Insert->currpos = (char *) Insert->currpage + - MAXALIGN(Insert->currpos - (char *) Insert->currpage); - freespace = INSERT_FREESPACE(Insert); + /* Align the end position, so that the next record starts aligned */ + if (CurrPos.xrecoff % MAXIMUM_ALIGNOF != 0) + { + CurrPos.xrecoff = MAXALIGN(CurrPos.xrecoff); + if (CurrPos.xrecoff == 0) + { + /* crossed a logid boundary */ + CurrPos.xlogid += 1; + } + } /* - * The recptr I return is the beginning of the *next* record. This will be - * stored as LSN for changed data pages... + * Done! Let others know that we're finished. */ - INSERT_RECPTR(RecPtr, Insert, curridx); + pg_write_barrier(); + if (firstpage) + *tot_len_p = write_len; + else + pagehdr->xlp_info |= XLP_FIRST_IS_CONTRECORD; /* - * If the record is an XLOG_SWITCH, we must now write and flush all the - * existing data, and then forcibly advance to the start of the next - * segment. It's not good to do this I/O while holding the insert lock, - * but there seems too much risk of confusion if we try to release the - * lock sooner. Fortunately xlog switch needn't be a high-performance - * operation anyway... + * If this was an xlog-switch, it's not enough to write the switch record, + * we also have to consume all the remaining space in the WAL segment. + * We have already reserved it for us, but we still need to make sure it's + * allocated and zeroed in the WAL buffers so that when the caller (or + * someone else) does XLogWrite(), it can really write out all the zeros. */ - if (isLogSwitch) + if (isLogSwitch && CurrPos.xrecoff % XLOG_SEG_SIZE != 0) { - XLogwrtRqst FlushRqst; - XLogRecPtr OldSegEnd; + volatile XLogCtlInsert *Insert = &XLogCtl->Insert; - TRACE_POSTGRESQL_XLOG_SWITCH(); + WaitXLogInsertionsToFinish(CurrPos); - LWLockAcquire(WALWriteLock, LW_EXCLUSIVE); + /* An xlog-switch record doesn't contain any data besides the header */ + Assert(write_len == SizeOfXLogRecord); /* - * Flush through the end of the page containing XLOG_SWITCH, and - * perform end-of-segment actions (eg, notifying archiver). + * We do this one page at a time, to make sure we don't deadlock + * against ourselves if wal_buffers < XLOG_SEG_SIZE. */ - WriteRqst = XLogCtl->xlblocks[curridx]; - FlushRqst.Write = WriteRqst; - FlushRqst.Flush = WriteRqst; - XLogWrite(FlushRqst, false, true); - - /* Set up the next buffer as first page of next segment */ - /* Note: AdvanceXLInsertBuffer cannot need to do I/O here */ - (void) AdvanceXLInsertBuffer(true); + Assert(EndPos.xrecoff % XLogSegSize == 0); - /* There should be no unwritten data */ - curridx = Insert->curridx; - Assert(curridx == XLogCtl->Write.curridx); + /* Use up all the remaining space on the first page */ + XLByteAdvance(CurrPos, freespace); - /* Compute end address of old segment */ - OldSegEnd = XLogCtl->xlblocks[curridx]; - if (OldSegEnd.xrecoff == 0) + while (XLByteLT(CurrPos, EndPos)) { - /* crossing a logid boundary */ - OldSegEnd.xlogid -= 1; + /* initialize the next page (if not initialized already) */ + AdvanceXLInsertBuffer(CurrPos, false); + XLByteAdvance(CurrPos, XLOG_BLCKSZ); + + /* + * Update FinalizedUpto immediately. FinalizeRecord() doesn't know + * that an xlog-switch record consumes the rest of the segment, + * so we have to do this ourselves. + */ + LWLockAcquire(WALInsertTailLock, LW_EXCLUSIVE); + FinalizedUpto = Insert->FinalizedUpto = CurrPos; + Insert->ExpectingContRecord = false; + Assert(XLByteEQ(Insert->LastFinalizedRecord, StartPos)); + LWLockRelease(WALInsertTailLock); } - OldSegEnd.xrecoff -= XLOG_BLCKSZ; + } + if (!XLByteEQ(CurrPos, EndPos)) + elog(PANIC, "space reserved for WAL record does not match what was written"); +} + +/* + * Reserves the right amount of space for a record of given size from the WAL. + * *StartPos_p is set to the beginning of the reserved section, *EndPos_p to + * its end+1. + * + * This is the performance critical part of XLogInsert that must be serialized + * across backends. The rest can happen mostly in parallel. + * + * NB: The space calculation here must match the code in CopyXLogRecordToWAL, + * where we actually copy the record to the reserved space. + */ +static void +ReserveXLogInsertLocation(int size, XLogRecPtr *StartPos, XLogRecPtr *EndPos) +{ + volatile XLogCtlInsert *Insert = &XLogCtl->Insert; + uint64 startbytepos; + uint64 endbytepos; - /* Make it look like we've written and synced all of old segment */ - LogwrtResult.Write = OldSegEnd; - LogwrtResult.Flush = OldSegEnd; + size = MAXALIGN(size); - /* - * Update shared-memory status --- this code should match XLogWrite - */ - { - /* use volatile pointer to prevent code rearrangement */ - volatile XLogCtlData *xlogctl = XLogCtl; + /* All (non xlog-switch) records should contain data. */ + Assert(size > SizeOfXLogRecord); - SpinLockAcquire(&xlogctl->info_lck); - xlogctl->LogwrtResult = LogwrtResult; - if (XLByteLT(xlogctl->LogwrtRqst.Write, LogwrtResult.Write)) - xlogctl->LogwrtRqst.Write = LogwrtResult.Write; - if (XLByteLT(xlogctl->LogwrtRqst.Flush, LogwrtResult.Flush)) - xlogctl->LogwrtRqst.Flush = LogwrtResult.Flush; - SpinLockRelease(&xlogctl->info_lck); - } + SpinLockAcquire(&Insert->insertpos_lck); - LWLockRelease(WALWriteLock); + startbytepos = Insert->CurrBytePos; + endbytepos = startbytepos + size; + Insert->CurrBytePos = endbytepos; + + SpinLockRelease(&Insert->insertpos_lck); + + *StartPos = XLogBytePosToRecPtr(startbytepos); + Assert(XLogRecPtrToBytePos(*StartPos) == startbytepos); + *EndPos = XLogBytePosToEndRecPtr(endbytepos); + Assert(XLogRecPtrToBytePos(*EndPos) == endbytepos); +} + +/* + * Like ReserveXLogInsertLocation(), but for an xlog-switch record. + * + * A log-switch record is handled slightly differently. The rest of the + * segment will be reserved for this insertion, as indicated by the returned + * *EndPos_p value. However, if we are already at the beginning of the current + * segment, the *EndPos_p is set to the current location without reserving + * any space, and the function returns false. +*/ +static bool +ReserveXLogSwitch(XLogRecPtr *StartPos, XLogRecPtr *EndPos) +{ + volatile XLogCtlInsert *Insert = &XLogCtl->Insert; + uint64 startbytepos; + uint64 endbytepos; + uint32 size = SizeOfXLogRecord; + XLogRecPtr ptr; + uint32 segleft; + + SpinLockAcquire(&Insert->insertpos_lck); + + startbytepos = Insert->CurrBytePos; + + ptr = XLogBytePosToEndRecPtr(startbytepos); + if (ptr.xrecoff % XLOG_SEG_SIZE == 0) + { + SpinLockRelease(&Insert->insertpos_lck); + *EndPos = *StartPos = ptr; + return false; + } + + *StartPos = XLogBytePosToRecPtr(startbytepos); + + endbytepos = startbytepos + size; - updrqst = false; /* done already */ + *EndPos = XLogBytePosToEndRecPtr(endbytepos); + Assert(XLogRecPtrToBytePos(*EndPos) == endbytepos); + + Assert(XLogRecPtrToBytePos(*StartPos) == startbytepos); + + segleft = XLOG_SEG_SIZE - (EndPos->xrecoff % XLOG_SEG_SIZE); + if (segleft != XLOG_SEG_SIZE) + { + /* consume the rest of the segment */ + EndPos->xrecoff += segleft; + endbytepos = XLogRecPtrToBytePos(*EndPos); + } + Insert->CurrBytePos = endbytepos; + + SpinLockRelease(&Insert->insertpos_lck); + + Assert(EndPos->xrecoff % XLOG_BLCKSZ == 0); + + return true; +} + +/* + * Get a pointer to the right location in the WAL buffer containing the + * given XLogRecPtr. + * + * If the page is not initialized yet, it is initialized. That might require + * evicting an old dirty buffer from the buffer cache, which means I/O. + * Unless failok == true, in which case the function returns NULL instead. + * + * The caller must ensure that the page containing the requested location + * isn't evicted yet, and won't be evicted. If you have reserved some WAL + * space, and not yet marked that you're done inserting it (by not having + * set xl_tot_len yet), that is enough. You should not be holding onto + * anything < ptr, though, because that might lead to deadlock if we would + * need to evict an old buffer to make room for the new one. + */ +static char * +GetXLogBuffer(XLogRecPtr ptr, bool failok) +{ + int idx; + XLogRecPtr endptr; + static uint32 cachedXlogid = 0; + static uint32 cachedPage = 0; + static char *cachedPos = NULL; + XLogRecPtr expectedEndPtr; + + /* + * Fast path for the common case that we need to access again the same + * page as last time. + */ + if (ptr.xlogid == cachedXlogid && ptr.xrecoff / XLOG_BLCKSZ == cachedPage) + { + Assert(((XLogPageHeader) cachedPos)->xlp_magic == XLOG_PAGE_MAGIC); + Assert(((XLogPageHeader) cachedPos)->xlp_pageaddr.xlogid == cachedXlogid); + Assert(((XLogPageHeader) cachedPos)->xlp_pageaddr.xrecoff == cachedPage * XLOG_BLCKSZ); + return cachedPos + ptr.xrecoff % XLOG_BLCKSZ; + } + + /* + * The XLog buffer cache is organized so that a page must always be loaded + * to a particular buffer. That way we can easily calculate the buffer + * a given page must be loaded into, from the XLogRecPtr alone. + */ + idx = XLogRecPtrToBufIdx(ptr); + + /* + * See what page is loaded in the buffer at the moment. It could be the + * page we're looking for, or something older. It can't be anything newer + * - that would imply the page we're looking for has already been written + * out to disk and evicted, and the caller is responsible for making sure + * that doesn't happen. + * + * However, we don't hold a lock while we read the value. If someone has + * just initialized the page, it's possible that we get a "torn read" of + * the XLogRecPtr, and see a bogus value. That's ok, we'll grab the + * mapping lock (in AdvanceXLInsertBuffer) and retry if we see anything + * else than the page we're looking for. But it means that when we do this + * unlocked read, we might see a value that appears to be ahead of the + * page we're looking for. Don't PANIC on that, until we've verified the + * value while holding the lock. + */ + expectedEndPtr = ptr; + XLByteAdvance(expectedEndPtr, XLOG_BLCKSZ - ptr.xrecoff % XLOG_BLCKSZ); + + endptr = XLogCtl->xlblocks[idx]; + if (!XLByteEQ(expectedEndPtr, endptr)) + { + if (failok) + return NULL; + + AdvanceXLInsertBuffer(ptr, false); + endptr = XLogCtl->xlblocks[idx]; + + if (!XLByteEQ(expectedEndPtr, endptr)) + elog(PANIC, "could not find WAL buffer for %X/%X", + ptr.xlogid, ptr.xrecoff); + } + + /* + * Found the buffer holding this page. Return a pointer to the right + * offset within the page. + */ + cachedXlogid = ptr.xlogid; + cachedPage = ptr.xrecoff / XLOG_BLCKSZ; + cachedPos = XLogCtl->pages + idx * (Size) XLOG_BLCKSZ; + + Assert(((XLogPageHeader) cachedPos)->xlp_magic == XLOG_PAGE_MAGIC); + Assert(((XLogPageHeader) cachedPos)->xlp_pageaddr.xlogid == cachedXlogid); + Assert(((XLogPageHeader) cachedPos)->xlp_pageaddr.xrecoff == cachedPage * XLOG_BLCKSZ); + Assert(((XLogPageHeader) cachedPos)->xlp_pageaddr.xrecoff == ptr.xrecoff - (ptr.xrecoff % XLOG_BLCKSZ)); + + return cachedPos + ptr.xrecoff % XLOG_BLCKSZ; +} + +/* + * Converts a "usable byte position" to XLogRecPtr. A usable byte position + * is the position starting from the beginning of WAL, excluding all WAL + * page headers. + */ +static XLogRecPtr +XLogBytePosToRecPtr(uint64 bytepos) +{ + uint64 fullsegs; + uint64 fullpages; + uint64 bytesleft; + uint32 seg_offset; + XLogRecPtr result; + + fullsegs = bytepos / UsableBytesInSegment; + bytesleft = bytepos % UsableBytesInSegment; + + if (bytesleft < XLOG_BLCKSZ - SizeOfXLogLongPHD) + { + /* fits on first page of segment */ + seg_offset = bytesleft + SizeOfXLogLongPHD; + } + else + { + /* account for the first page on segment with long header */ + seg_offset = XLOG_BLCKSZ; + bytesleft -= XLOG_BLCKSZ - SizeOfXLogLongPHD; + + fullpages = bytesleft / UsableBytesInPage; + bytesleft = bytesleft % UsableBytesInPage; + + seg_offset += fullpages * XLOG_BLCKSZ + bytesleft + SizeOfXLogShortPHD; + } + + XLogSegNoOffsetToRecPtr(fullsegs, seg_offset, result); + + return result; +} + +/* + * Like XLogBytePosToEndRecPtr, but a page boundary is represented by pointer + * to beginning of page, not to where the first xlog record goes to. + */ +static XLogRecPtr +XLogBytePosToEndRecPtr(uint64 bytepos) +{ + uint64 fullsegs; + uint64 fullpages; + uint64 bytesleft; + uint32 seg_offset; + XLogRecPtr result; + + fullsegs = bytepos / UsableBytesInSegment; + bytesleft = bytepos % UsableBytesInSegment; + + if (bytesleft < XLOG_BLCKSZ - SizeOfXLogLongPHD) + { + /* fits on first page of segment */ + if (bytesleft == 0) + seg_offset = 0; + else + seg_offset = bytesleft + SizeOfXLogLongPHD; } else { - /* normal case, ie not xlog switch */ + /* account for the first page on segment with long header */ + seg_offset = XLOG_BLCKSZ; + bytesleft -= XLOG_BLCKSZ - SizeOfXLogLongPHD; + + fullpages = bytesleft / UsableBytesInPage; + bytesleft = bytesleft % UsableBytesInPage; + + if (bytesleft == 0) + seg_offset += fullpages * XLOG_BLCKSZ + bytesleft; + else + seg_offset += fullpages * XLOG_BLCKSZ + bytesleft + SizeOfXLogShortPHD; + } + + XLogSegNoOffsetToRecPtr(fullsegs, seg_offset, result); + + return result; +} + +/* + * Convert an XLogRecPtr to a "usable byte position". + */ +static uint64 +XLogRecPtrToBytePos(XLogRecPtr ptr) +{ + uint64 fullsegs; + uint32 fullpages; + uint32 offset; + uint64 result; + + XLByteToSeg(ptr, fullsegs); - /* Need to update shared LogwrtRqst if some block was filled up */ - if (freespace == 0) + fullpages = (ptr.xrecoff % XLOG_SEG_SIZE) / XLOG_BLCKSZ; + offset = ptr.xrecoff % XLOG_BLCKSZ; + + if (fullpages == 0) + { + result = fullsegs * UsableBytesInSegment; + if (offset > 0) { - /* curridx is filled and available for writing out */ - updrqst = true; + Assert(offset >= SizeOfXLogLongPHD); + result += offset - SizeOfXLogLongPHD; } - else + } + else + { + result = fullsegs * UsableBytesInSegment + + (XLOG_BLCKSZ - SizeOfXLogLongPHD) + /* account for first page */ + (fullpages - 1) * UsableBytesInPage; /* full pages */ + if (offset > 0) { - /* if updrqst already set, write through end of previous buf */ - curridx = PrevBufIdx(curridx); + Assert(offset >= SizeOfXLogShortPHD); + result += offset - SizeOfXLogShortPHD; } - WriteRqst = XLogCtl->xlblocks[curridx]; } - LWLockRelease(WALInsertLock); + return result; +} + +/* + * Attempt to finalize next record, if it's been copied in place. + */ +static bool +FinalizeRecord(void) +{ + XLogCtlInsert *Insert = &XLogCtl->Insert; + XLogRecPtr ptr; + uint32 len; + char *p; + int freespace; + XLogRecPtr StartPos, EndPos; + + ptr = XLogCtl->Insert.FinalizedUpto; + p = GetXLogBuffer(ptr, true); + if (p == NULL) + return false; + + StartPos = ptr; - if (updrqst) + /* + * If LastFinalizedRecord points to beginning of page, assume it's + * a continuation record. + */ + if (XLogCtl->Insert.ExpectingContRecord) { - /* use volatile pointer to prevent code rearrangement */ - volatile XLogCtlData *xlogctl = XLogCtl; + XLogPageHeader pagehdr = (XLogPageHeader) p; + int pagehdrsize; - SpinLockAcquire(&xlogctl->info_lck); - /* advance global request to include new block(s) */ - if (XLByteLT(xlogctl->LogwrtRqst.Write, WriteRqst)) - xlogctl->LogwrtRqst.Write = WriteRqst; - /* update local result copy while I have the chance */ - LogwrtResult = xlogctl->LogwrtResult; - SpinLockRelease(&xlogctl->info_lck); + Assert(pagehdr->xlp_magic == XLOG_PAGE_MAGIC); + pagehdrsize = (ptr.xrecoff % XLOG_SEG_SIZE == 0) ? SizeOfXLogLongPHD : SizeOfXLogShortPHD; + Assert(pagehdrsize == XLogPageHeaderSize(pagehdr)); + + if ((pagehdr->xlp_info & XLP_FIRST_IS_CONTRECORD) == 0) + return false; + + pg_memory_barrier(); + + /* Cool, the part of this continued record on this page is done */ + len = MAXALIGN(pagehdr->xlp_rem_len); + Assert(len > 0 && len < 1000000); + if (len < XLOG_BLCKSZ - pagehdrsize) + { + ptr.xrecoff += pagehdrsize + len; + Insert->ExpectingContRecord = false; + } + else if (len == XLOG_BLCKSZ - pagehdrsize) + { + XLByteAdvance(ptr, XLOG_BLCKSZ); + Insert->ExpectingContRecord = false; + } + else + { + XLByteAdvance(ptr, XLOG_BLCKSZ); + Insert->ExpectingContRecord = true; + } + FinalizedUpto = XLogCtl->Insert.FinalizedUpto = ptr; + EndPos = ptr; } + else + { + XLogRecPtr LastFinalized = Insert->LastFinalizedRecord; + pg_crc32 rdata_crc; + XLogRecPtr recstart; + char *recstartp; + + /* if we're located at page boundary, skip page header */ + if (ptr.xrecoff % XLOG_BLCKSZ == 0) + { + if (ptr.xrecoff % XLOG_SEG_SIZE == 0) + { + ptr.xrecoff += SizeOfXLogLongPHD; + p += SizeOfXLogLongPHD; + } + else + { + ptr.xrecoff += SizeOfXLogShortPHD; + p += SizeOfXLogShortPHD; + } + } - XactLastRecEnd = RecPtr; + recstart = ptr; + recstartp = p; - END_CRIT_SECTION(); + /* NB: we might not have the full header on this page! */ + /* fetch record->xl_tot_len */ + len = MAXALIGN(*((uint32 *) p)); + if (len == 0) + return false; - return RecPtr; + pg_memory_barrier(); + + /* Cool, this record is done. Set xl_prev, and finish CRC calculation. */ + /* xl_prev might be on next page */ + freespace = INSERT_FREESPACE(ptr); + if (freespace < offsetof(XLogRecord, xl_prev) + sizeof(XLogRecPtr)) + { + XLogPageHeader pagehdr; + int pagehdrsize; + int off = offsetof(XLogRecord, xl_prev) - freespace; + + XLByteAdvance(ptr, freespace); + p = GetXLogBuffer(ptr, true); + if (p == NULL) + return false; + + pagehdr = (XLogPageHeader) p; + pagehdrsize = (ptr.xrecoff % XLOG_SEG_SIZE == 0) ? SizeOfXLogLongPHD : SizeOfXLogShortPHD; + + Assert(pagehdr->xlp_magic == XLOG_PAGE_MAGIC); + Assert(pagehdrsize == XLogPageHeaderSize(pagehdr)); + + /* + * If the rest of the record header has not been copied in place + * yet, bail out. + */ + if ((pagehdr->xlp_info & XLP_FIRST_IS_CONTRECORD) == 0) + return false; + p += pagehdrsize + off; + } + else + { + p += offsetof(XLogRecord, xl_prev); + } + Assert (((uint64) p) % XLOG_BLCKSZ >= SizeOfXLogShortPHD); + *((XLogRecPtr *) p) = LastFinalized; + + /* xl_crc might be on next page, if xl_prev was not */ + ptr = recstart; + p = recstartp; + + freespace = INSERT_FREESPACE(ptr); + if (freespace < offsetof(XLogRecord, xl_crc) + sizeof(pg_crc32)) + { + XLogPageHeader pagehdr; + int pagehdrsize; + int off = offsetof(XLogRecord, xl_crc) - freespace; + + XLByteAdvance(ptr, freespace); + p = GetXLogBuffer(ptr, true); + if (p == NULL) + return false; + + pagehdr = (XLogPageHeader) p; + pagehdrsize = (ptr.xrecoff % XLOG_SEG_SIZE == 0) ? SizeOfXLogLongPHD : SizeOfXLogShortPHD; + + Assert(pagehdr->xlp_magic == XLOG_PAGE_MAGIC); + Assert(pagehdrsize == XLogPageHeaderSize(pagehdr)); + + if ((pagehdr->xlp_info & XLP_FIRST_IS_CONTRECORD) == 0) + return false; + p += pagehdrsize + off; + } + else + { + p += offsetof(XLogRecord, xl_crc); + } + Assert (((uint64) p) % XLOG_BLCKSZ >= SizeOfXLogShortPHD); + + /* Update CRC with xl_prev, finish it with FIN_CRC32, and write back */ + rdata_crc = *((pg_crc32 *) p); + COMP_CRC32(rdata_crc, ((char *) &LastFinalized), sizeof(XLogRecPtr)); + FIN_CRC32(rdata_crc); + *((pg_crc32 *) p) = rdata_crc; + + /* + * Update FinalizedUpto to end of record, or end of page where this + * record began, if it didn't fit page. + */ + ptr = recstart; + freespace = INSERT_FREESPACE(ptr); + if (len <= freespace) + { + XLByteAdvance(ptr, len); + Insert->ExpectingContRecord = false; + } + else + { + XLByteAdvance(ptr, freespace); + Insert->ExpectingContRecord = true; + } + FinalizedUpto = XLogCtl->Insert.FinalizedUpto = ptr; + EndPos = ptr; + /* + * Update LastFinalizedRecord, so that we can set xl_prev link on + * the next record correctly. + */ + XLogCtl->Insert.LastFinalizedRecord = recstart; + } + +#ifdef NOT_USED + elog(LOG, "FINALIZE @ %X/%X - %X/%X", + StartPos.xlogid, StartPos.xrecoff, EndPos.xlogid, EndPos.xrecoff); +#endif + + return true; +} + +/* + * Wait for any insertions < upto to finish. + * + * Returns a value >= upto, which indicates the oldest in-progress insertion + * that we saw (or if there are non in-progress, the next insert position). + */ +static XLogRecPtr +WaitXLogInsertionsToFinish(XLogRecPtr upto) +{ + if (MyProc == NULL) + elog(PANIC, "cannot wait without a PGPROC structure"); + + if (XLByteLE(upto, FinalizedUpto)) + return FinalizedUpto; + + /* + * XXX: Busy-loop until we succeed to finalize up to the requested + * point + */ + for (;;) + { + /* Only allow one process to finalize at a time */ + LWLockAcquire(WALInsertTailLock, LW_EXCLUSIVE); + + /* While we're at it, finalize as far as we can. */ + while (FinalizeRecord()); + FinalizedUpto = XLogCtl->Insert.FinalizedUpto; + + LWLockRelease(WALInsertTailLock); + + /* Is this enough? */ + if (XLogRecPtrIsInvalid(upto) || XLByteLE(upto, FinalizedUpto)) + return FinalizedUpto; + } } /* @@ -1445,31 +2194,34 @@ XLogArchiveCleanup(const char *xlog) } /* - * Advance the Insert state to the next buffer page, writing out the next - * buffer if it still contains unwritten data. - * - * If new_segment is TRUE then we set up the next buffer page as the first - * page of the next xlog segment file, possibly but not usually the next - * consecutive file page. - * - * The global LogwrtRqst.Write pointer needs to be advanced to include the - * just-filled page. If we can do this for free (without an extra lock), - * we do so here. Otherwise the caller must do it. We return TRUE if the - * request update still needs to be done, FALSE if we did it internally. - * - * Must be called with WALInsertLock held. + * Initialize XLOG buffers, writing out old buffers if they still contain + * unwritten data, upto the page containing 'upto'. Or if 'opportunistic' is + * true, initialize as many pages as we can without having to write out + * unwritten data. Any new pages are initialized to zeros, with pages headers + * initialized properly. */ -static bool -AdvanceXLInsertBuffer(bool new_segment) +static void +AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic) { XLogCtlInsert *Insert = &XLogCtl->Insert; - int nextidx = NextBufIdx(Insert->curridx); - bool update_needed = true; + int nextidx; XLogRecPtr OldPageRqstPtr; XLogwrtRqst WriteRqst; - XLogRecPtr NewPageEndPtr; + XLogRecPtr NewPageEndPtr = InvalidXLogRecPtr; XLogRecPtr NewPageBeginPtr; XLogPageHeader NewPage; + int npages = 0; + + LWLockAcquire(WALBufMappingLock, LW_EXCLUSIVE); + + /* + * Now that we have the lock, check if someone initialized the page + * already. + */ +/* XXX: fix indentation before commit */ +while (!XLByteLT(upto, XLogCtl->xlblocks[XLogCtl->curridx]) || opportunistic) +{ + nextidx = NextBufIdx(XLogCtl->curridx); /* * Get ending-offset of the buffer page we need to replace (this may be @@ -1479,10 +2231,12 @@ AdvanceXLInsertBuffer(bool new_segment) OldPageRqstPtr = XLogCtl->xlblocks[nextidx]; if (!XLByteLE(OldPageRqstPtr, LogwrtResult.Write)) { - /* nope, got work to do... */ - XLogRecPtr FinishedPageRqstPtr; - - FinishedPageRqstPtr = XLogCtl->xlblocks[Insert->curridx]; + /* + * Nope, got work to do. If we just want to pre-initialize as much as + * we can without flushing, give up now. + */ + if (opportunistic) + break; /* Before waiting, get info_lck and update LogwrtResult */ { @@ -1490,21 +2244,27 @@ AdvanceXLInsertBuffer(bool new_segment) volatile XLogCtlData *xlogctl = XLogCtl; SpinLockAcquire(&xlogctl->info_lck); - if (XLByteLT(xlogctl->LogwrtRqst.Write, FinishedPageRqstPtr)) - xlogctl->LogwrtRqst.Write = FinishedPageRqstPtr; + if (XLByteLT(xlogctl->LogwrtRqst.Write, OldPageRqstPtr)) + xlogctl->LogwrtRqst.Write = OldPageRqstPtr; LogwrtResult = xlogctl->LogwrtResult; SpinLockRelease(&xlogctl->info_lck); } - update_needed = false; /* Did the shared-request update */ - /* * Now that we have an up-to-date LogwrtResult value, see if we still * need to write it or if someone else already did. */ if (!XLByteLE(OldPageRqstPtr, LogwrtResult.Write)) { - /* Must acquire write lock */ + /* + * Must acquire write lock. Release WALBufMappingLock first, to + * make sure that all insertions that we need to wait for can + * finish (up to this same position). Otherwise we risk deadlock. + */ + LWLockRelease(WALBufMappingLock); + + WaitXLogInsertionsToFinish(OldPageRqstPtr); + LWLockAcquire(WALWriteLock, LW_EXCLUSIVE); LogwrtResult = XLogCtl->LogwrtResult; if (XLByteLE(OldPageRqstPtr, LogwrtResult.Write)) @@ -1514,18 +2274,18 @@ AdvanceXLInsertBuffer(bool new_segment) } else { - /* - * Have to write buffers while holding insert lock. This is - * not good, so only write as much as we absolutely must. - */ + /* Have to write it ourselves */ TRACE_POSTGRESQL_WAL_BUFFER_WRITE_DIRTY_START(); WriteRqst.Write = OldPageRqstPtr; WriteRqst.Flush.xlogid = 0; WriteRqst.Flush.xrecoff = 0; - XLogWrite(WriteRqst, false, false); + XLogWrite(WriteRqst, false); LWLockRelease(WALWriteLock); TRACE_POSTGRESQL_WAL_BUFFER_WRITE_DIRTY_DONE(); } + /* Re-acquire WALBufMappingLock and retry */ + LWLockAcquire(WALBufMappingLock, LW_EXCLUSIVE); + continue; } } @@ -1533,25 +2293,16 @@ AdvanceXLInsertBuffer(bool new_segment) * Now the next buffer slot is free and we can set it up to be the next * output page. */ - NewPageBeginPtr = XLogCtl->xlblocks[Insert->curridx]; - - if (new_segment) - { - /* force it to a segment start point */ - if (NewPageBeginPtr.xrecoff % XLogSegSize != 0) - XLByteAdvance(NewPageBeginPtr, - XLogSegSize - NewPageBeginPtr.xrecoff % XLogSegSize); - } + NewPageBeginPtr = XLogCtl->xlblocks[XLogCtl->curridx]; NewPageEndPtr = NewPageBeginPtr; XLByteAdvance(NewPageEndPtr, XLOG_BLCKSZ); - XLogCtl->xlblocks[nextidx] = NewPageEndPtr; - NewPage = (XLogPageHeader) (XLogCtl->pages + nextidx * (Size) XLOG_BLCKSZ); - Insert->curridx = nextidx; - Insert->currpage = NewPage; + Assert(NewPageEndPtr.xrecoff % XLOG_BLCKSZ == 0); + Assert(XLogRecEndPtrToBufIdx(NewPageEndPtr) == nextidx); + Assert(XLogRecPtrToBufIdx(NewPageBeginPtr) == nextidx); - Insert->currpos = ((char *) NewPage) +SizeOfXLogShortPHD; + NewPage = (XLogPageHeader) (XLogCtl->pages + nextidx * (Size) XLOG_BLCKSZ); /* * Be sure to re-zero the buffer so that bytes beyond what we've written @@ -1567,6 +2318,7 @@ AdvanceXLInsertBuffer(bool new_segment) /* NewPage->xlp_info = 0; */ /* done by memset */ NewPage ->xlp_tli = ThisTimeLineID; NewPage ->xlp_pageaddr = NewPageBeginPtr; + /* NewPage ->xlp_rem_len = InvalidXLogRecPtr; */ /* done by memset */ /* * If online backup is not in progress, mark the header to indicate that @@ -1594,11 +2346,28 @@ AdvanceXLInsertBuffer(bool new_segment) NewLongPage->xlp_seg_size = XLogSegSize; NewLongPage->xlp_xlog_blcksz = XLOG_BLCKSZ; NewPage ->xlp_info |= XLP_LONG_HEADER; - - Insert->currpos = ((char *) NewPage) +SizeOfXLogLongPHD; } - return update_needed; + /* + * Make sure the initialization of the page becomes visible to others + * before the xlblocks update. GetXLogBuffer() reads xlblocks without + * holding a lock. + */ + pg_write_barrier(); + + *((volatile XLogRecPtr *) &XLogCtl->xlblocks[nextidx]) = NewPageEndPtr; + + XLogCtl->curridx = nextidx; + + npages++; +} + LWLockRelease(WALBufMappingLock); + +#ifdef WAL_DEBUG + if (npages > 0) + elog(DEBUG1, "initialized %d pages, upto %X/%X", + npages, NewPageEndPtr.xlogid, NewPageEndPtr.xrecoff); +#endif } /* @@ -1630,16 +2399,12 @@ XLogCheckpointNeeded(XLogSegNo new_segno) * This option allows us to avoid uselessly issuing multiple writes when a * single one would do. * - * If xlog_switch == TRUE, we are intending an xlog segment switch, so - * perform end-of-segment actions after writing the last page, even if - * it's not physically the end of its segment. (NB: this will work properly - * only if caller specifies WriteRqst == page-end and flexible == false, - * and there is some data to write.) - * - * Must be called with WALWriteLock held. + * Must be called with WALWriteLock held. And you must've called + * WaitXLogInsertionsToFinish(WriteRqst) before grabbing the lock to make sure + * the data is ready to write. */ static void -XLogWrite(XLogwrtRqst WriteRqst, bool flexible, bool xlog_switch) +XLogWrite(XLogwrtRqst WriteRqst, bool flexible) { XLogCtlWrite *Write = &XLogCtl->Write; bool ispartialpage; @@ -1688,14 +2453,14 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible, bool xlog_switch) * if we're passed a bogus WriteRqst.Write that is past the end of the * last page that's been initialized by AdvanceXLInsertBuffer. */ - if (!XLByteLT(LogwrtResult.Write, XLogCtl->xlblocks[curridx])) + XLogRecPtr EndPtr = XLogCtl->xlblocks[curridx]; + if (!XLByteLT(LogwrtResult.Write, EndPtr)) elog(PANIC, "xlog write request %X/%X is past end of log %X/%X", LogwrtResult.Write.xlogid, LogwrtResult.Write.xrecoff, - XLogCtl->xlblocks[curridx].xlogid, - XLogCtl->xlblocks[curridx].xrecoff); + EndPtr.xlogid, EndPtr.xrecoff); /* Advance LogwrtResult.Write to end of current buffer page */ - LogwrtResult.Write = XLogCtl->xlblocks[curridx]; + LogwrtResult.Write = EndPtr; ispartialpage = XLByteLT(WriteRqst.Write, LogwrtResult.Write); if (!XLByteInPrevSeg(LogwrtResult.Write, openLogSegNo)) @@ -1778,6 +2543,12 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible, bool xlog_switch) XLogFileNameP(ThisTimeLineID, openLogSegNo), openLogOff, (unsigned long) nbytes))); } +#ifdef CLOBBER_FREED_MEMORY + if (!ispartialpage) + memset(from, 0x7E, nbytes); + else if (npages > 1) + memset(from, 0x7E, nbytes - XLOG_BLCKSZ); +#endif /* Update state for write */ openLogOff += nbytes; @@ -1791,16 +2562,13 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible, bool xlog_switch) * later. Doing it here ensures that one and only one backend will * perform this fsync. * - * We also do this if this is the last page written for an xlog - * switch. - * * This is also the right place to notify the Archiver that the * segment is ready to copy to archival storage, and to update the * timer for archive_timeout, and to signal for a checkpoint if * too many logfile segments have been used since the last * checkpoint. */ - if (finishing_seg || (xlog_switch && last_iteration)) + if (finishing_seg) { issue_xlog_fsync(openLogFile, openLogSegNo); LogwrtResult.Flush = LogwrtResult.Write; /* end of page */ @@ -1865,7 +2633,9 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible, bool xlog_switch) openLogFile = XLogFileOpen(openLogSegNo); openLogOff = 0; } + elog(LOG, "flushing seg %ld (explicit)", openLogSegNo); issue_xlog_fsync(openLogFile, openLogSegNo); + elog(LOG, "done flushing seg %ld (explicit)", openLogSegNo); } LogwrtResult.Flush = LogwrtResult.Write; } @@ -2066,6 +2836,7 @@ XLogFlush(XLogRecPtr record) { /* use volatile pointer to prevent code rearrangement */ volatile XLogCtlData *xlogctl = XLogCtl; + XLogRecPtr insertpos; /* read LogwrtResult and update local state */ SpinLockAcquire(&xlogctl->info_lck); @@ -2079,6 +2850,12 @@ XLogFlush(XLogRecPtr record) break; /* + * Before actually performing the write, wait for all in-flight + * insertions to the pages we're about to write to finish. + */ + insertpos = WaitXLogInsertionsToFinish(WriteRqstPtr); + + /* * Try to get the write lock. If we can't get it immediately, wait * until it's released, and recheck if we still need to do the flush * or if the backend that held the lock did it for us already. This @@ -2098,31 +2875,10 @@ XLogFlush(XLogRecPtr record) LogwrtResult = XLogCtl->LogwrtResult; if (!XLByteLE(record, LogwrtResult.Flush)) { - /* try to write/flush later additions to XLOG as well */ - if (LWLockConditionalAcquire(WALInsertLock, LW_EXCLUSIVE)) - { - XLogCtlInsert *Insert = &XLogCtl->Insert; - uint32 freespace = INSERT_FREESPACE(Insert); + WriteRqst.Write = insertpos; + WriteRqst.Flush = insertpos; - if (freespace == 0) /* buffer is full */ - WriteRqstPtr = XLogCtl->xlblocks[Insert->curridx]; - else - { - WriteRqstPtr = XLogCtl->xlblocks[Insert->curridx]; - if (WriteRqstPtr.xrecoff == 0) - WriteRqstPtr.xlogid--; - WriteRqstPtr.xrecoff -= freespace; - } - LWLockRelease(WALInsertLock); - WriteRqst.Write = WriteRqstPtr; - WriteRqst.Flush = WriteRqstPtr; - } - else - { - WriteRqst.Write = WriteRqstPtr; - WriteRqst.Flush = record; - } - XLogWrite(WriteRqst, false, false); + XLogWrite(WriteRqst, false); } LWLockRelease(WALWriteLock); /* done */ @@ -2240,7 +2996,8 @@ XLogBackgroundFlush(void) START_CRIT_SECTION(); - /* now wait for the write lock */ + /* now wait for any in-progress insertions to finish and get write lock */ + WaitXLogInsertionsToFinish(WriteRqstPtr); LWLockAcquire(WALWriteLock, LW_EXCLUSIVE); LogwrtResult = XLogCtl->LogwrtResult; if (!XLByteLE(WriteRqstPtr, LogwrtResult.Flush)) @@ -2249,13 +3006,19 @@ XLogBackgroundFlush(void) WriteRqst.Write = WriteRqstPtr; WriteRqst.Flush = WriteRqstPtr; - XLogWrite(WriteRqst, flexible, false); + XLogWrite(WriteRqst, flexible); wrote_something = true; } LWLockRelease(WALWriteLock); END_CRIT_SECTION(); + /* + * Great, done. To take some work off the critical path, try to initialize + * as many of the no-longer-needed WAL buffers for future use as we can. + */ + AdvanceXLInsertBuffer(InvalidXLogRecPtr, true); + return wrote_something; } @@ -5066,6 +5829,7 @@ XLOGShmemSize(void) /* XLogCtl */ size = sizeof(XLogCtlData); + /* xlblocks array */ size = add_size(size, mul_size(sizeof(XLogRecPtr), XLOGbuffers)); /* extra alignment padding for XLOG I/O buffers */ @@ -5091,8 +5855,7 @@ XLOGShmemInit(void) ControlFile = (ControlFileData *) ShmemInitStruct("Control File", sizeof(ControlFileData), &foundCFile); - XLogCtl = (XLogCtlData *) - ShmemInitStruct("XLOG Ctl", XLOGShmemSize(), &foundXLog); + allocptr = ShmemInitStruct("XLOG Ctl", XLOGShmemSize(), &foundXLog); if (foundCFile || foundXLog) { @@ -5100,7 +5863,7 @@ XLOGShmemInit(void) Assert(foundCFile && foundXLog); return; } - + XLogCtl = (XLogCtlData *) allocptr; memset(XLogCtl, 0, sizeof(XLogCtlData)); /* @@ -5108,7 +5871,7 @@ XLOGShmemInit(void) * multiple of the alignment for same, so no extra alignment padding is * needed here. */ - allocptr = ((char *) XLogCtl) + sizeof(XLogCtlData); + allocptr += sizeof(XLogCtlData); XLogCtl->xlblocks = (XLogRecPtr *) allocptr; memset(XLogCtl->xlblocks, 0, sizeof(XLogRecPtr) * XLOGbuffers); allocptr += sizeof(XLogRecPtr) * XLOGbuffers; @@ -5128,7 +5891,12 @@ XLOGShmemInit(void) XLogCtl->SharedRecoveryInProgress = true; XLogCtl->SharedHotStandbyActive = false; XLogCtl->WalWriterSleeping = false; - XLogCtl->Insert.currpage = (XLogPageHeader) (XLogCtl->pages); + + XLogCtl->Insert.LastFinalizedRecord = InvalidXLogRecPtr; + XLogCtl->Insert.FinalizedUpto = InvalidXLogRecPtr; + XLogCtl->Insert.ExpectingContRecord = false; + + SpinLockInit(&XLogCtl->Insert.insertpos_lck); SpinLockInit(&XLogCtl->info_lck); InitSharedLatch(&XLogCtl->recoveryWakeupLatch); @@ -6006,6 +6774,7 @@ StartupXLOG(void) bool backupEndRequired = false; bool backupFromStandby = false; DBState dbstate_at_startup; + int firstIdx; /* * Read control file and check XLOG status looks valid. @@ -6258,7 +7027,7 @@ StartupXLOG(void) lastFullPageWrites = checkPoint.fullPageWrites; - RedoRecPtr = XLogCtl->Insert.RedoRecPtr = checkPoint.redo; + RedoRecPtr = XLogCtl->RedoRecPtr = XLogCtl->Insert.RedoRecPtr = checkPoint.redo; if (XLByteLT(RecPtr, checkPoint.redo)) ereport(PANIC, @@ -6814,9 +7583,13 @@ StartupXLOG(void) openLogFile = XLogFileOpen(openLogSegNo); openLogOff = 0; Insert = &XLogCtl->Insert; - Insert->PrevRecord = LastRec; - XLogCtl->xlblocks[0].xlogid = (openLogSegNo * XLOG_SEG_SIZE) >> 32; - XLogCtl->xlblocks[0].xrecoff = + Insert->LastFinalizedRecord = LastRec; + + firstIdx = XLogRecEndPtrToBufIdx(EndOfLog); + XLogCtl->curridx = firstIdx; + + XLogCtl->xlblocks[firstIdx].xlogid = (openLogSegNo * XLOG_SEG_SIZE) >> 32; + XLogCtl->xlblocks[firstIdx].xrecoff = ((EndOfLog.xrecoff - 1) / XLOG_BLCKSZ + 1) * XLOG_BLCKSZ; /* @@ -6824,10 +7597,11 @@ StartupXLOG(void) * record spans, not the one it starts in. The last block is indeed the * one we want to use. */ - Assert(readOff == (XLogCtl->xlblocks[0].xrecoff - XLOG_BLCKSZ) % XLogSegSize); - memcpy((char *) Insert->currpage, readBuf, XLOG_BLCKSZ); - Insert->currpos = (char *) Insert->currpage + - (EndOfLog.xrecoff + XLOG_BLCKSZ - XLogCtl->xlblocks[0].xrecoff); + Assert(readOff == (XLogCtl->xlblocks[firstIdx].xrecoff - XLOG_BLCKSZ) % XLogSegSize); + memcpy((char *) &XLogCtl->pages[firstIdx * XLOG_BLCKSZ], readBuf, XLOG_BLCKSZ); + Insert->FinalizedUpto = EndOfLog; + Insert->ExpectingContRecord = false; + Insert->CurrBytePos = XLogRecPtrToBytePos(EndOfLog); LogwrtResult.Write = LogwrtResult.Flush = EndOfLog; @@ -6836,12 +7610,12 @@ StartupXLOG(void) XLogCtl->LogwrtRqst.Write = EndOfLog; XLogCtl->LogwrtRqst.Flush = EndOfLog; - freespace = INSERT_FREESPACE(Insert); + freespace = INSERT_FREESPACE(EndOfLog); if (freespace > 0) { /* Make sure rest of page is zero */ - MemSet(Insert->currpos, 0, freespace); - XLogCtl->Write.curridx = 0; + MemSet(&XLogCtl->pages[firstIdx * XLOG_BLCKSZ] + EndOfLog.xrecoff % XLOG_BLCKSZ, 0, freespace); + XLogCtl->Write.curridx = firstIdx; } else { @@ -6853,7 +7627,7 @@ StartupXLOG(void) * this is sufficient. The first actual attempt to insert a log * record will advance the insert state. */ - XLogCtl->Write.curridx = NextBufIdx(0); + XLogCtl->Write.curridx = NextBufIdx(firstIdx); } /* Pre-scan prepared transactions to find out the range of XIDs present */ @@ -6864,7 +7638,7 @@ StartupXLOG(void) * XLOG_FPW_CHANGE record before resource manager writes cleanup * WAL records or checkpoint record is written. */ - Insert->fullPageWrites = lastFullPageWrites; + Insert->fullPageWrites = doPageWrites = lastFullPageWrites; LocalSetXLogInsertAllowed(); UpdateFullPageWrites(); LocalXLogInsertAllowed = -1; @@ -7332,21 +8106,29 @@ InitXLOGAccess(void) } /* - * Once spawned, a backend may update its local RedoRecPtr from - * XLogCtl->Insert.RedoRecPtr; it must hold the insert lock or info_lck - * to do so. This is done in XLogInsert() or GetRedoRecPtr(). + * Return the current Redo pointer from shared memory. + * + * As a side-effect, the local RedoRecPtr copy is updated. */ XLogRecPtr GetRedoRecPtr(void) { /* use volatile pointer to prevent code rearrangement */ volatile XLogCtlData *xlogctl = XLogCtl; + XLogRecPtr ptr; + /* + * The possibly not up-to-date copy in XlogCtl is enough. Even if we + * grabbed WALInsertShareLock to read the master copy, someone might update + * it just after we've released the lock. + */ SpinLockAcquire(&xlogctl->info_lck); - Assert(XLByteLE(RedoRecPtr, xlogctl->Insert.RedoRecPtr)); - RedoRecPtr = xlogctl->Insert.RedoRecPtr; + ptr = xlogctl->RedoRecPtr; SpinLockRelease(&xlogctl->info_lck); + if (XLByteLT(RedoRecPtr, ptr)) + RedoRecPtr = xlogctl->RedoRecPtr; + return RedoRecPtr; } @@ -7355,7 +8137,7 @@ GetRedoRecPtr(void) * * NOTE: The value *actually* returned is the position of the last full * xlog page. It lags behind the real insert position by at most 1 page. - * For that, we don't need to acquire WALInsertLock which can be quite + * For that, we don't need to acquire WALInsertShareLock which is * heavily contended, and an approximation is enough for the current * usage of this function. */ @@ -7630,6 +8412,8 @@ LogCheckpointEnd(bool restartpoint) void CreateCheckPoint(int flags) { + /* use volatile pointer to prevent code rearrangement */ + volatile XLogCtlData *xlogctl = XLogCtl; bool shutdown; CheckPoint checkPoint; XLogRecPtr recptr; @@ -7641,6 +8425,7 @@ CreateCheckPoint(int flags) XLogSegNo insert_logSegNo; TransactionId *inCommitXids; int nInCommit; + XLogRecPtr curInsert; /* * An end-of-recovery checkpoint is really a shutdown checkpoint, just @@ -7709,10 +8494,11 @@ CreateCheckPoint(int flags) checkPoint.oldestActiveXid = InvalidTransactionId; /* - * We must hold WALInsertLock while examining insert state to determine + * We must hold insertpos_lck while examining insert state to determine * the checkpoint REDO pointer. */ - LWLockAcquire(WALInsertLock, LW_EXCLUSIVE); + WALInsertLockAcquire(LW_EXCLUSIVE); + curInsert = XLogBytePosToRecPtr(Insert->CurrBytePos); /* * If this isn't a shutdown or forced checkpoint, and we have not switched @@ -7724,7 +8510,7 @@ CreateCheckPoint(int flags) * (Perhaps it'd make even more sense to checkpoint only when the previous * checkpoint record is in a different xlog page?) * - * While holding the WALInsertLock we find the current WAL insertion point + * While holding insertpos_lck we find the current WAL insertion point * and compare that with the starting point of the last checkpoint, which * is the redo pointer. We use the redo pointer because the start and end * points of a checkpoint can be hundreds of files apart on large systems @@ -7733,14 +8519,11 @@ CreateCheckPoint(int flags) if ((flags & (CHECKPOINT_IS_SHUTDOWN | CHECKPOINT_END_OF_RECOVERY | CHECKPOINT_FORCE)) == 0) { - XLogRecPtr curInsert; - - INSERT_RECPTR(curInsert, Insert, Insert->curridx); XLByteToSeg(curInsert, insert_logSegNo); XLByteToSeg(ControlFile->checkPointCopy.redo, redo_logSegNo); if (insert_logSegNo == redo_logSegNo) { - LWLockRelease(WALInsertLock); + WALInsertLockRelease(LW_EXCLUSIVE); LWLockRelease(CheckpointLock); END_CRIT_SECTION(); return; @@ -7767,18 +8550,19 @@ CreateCheckPoint(int flags) * the buffer flush work. Those XLOG records are logically after the * checkpoint, even though physically before it. Got that? */ - freespace = INSERT_FREESPACE(Insert); + freespace = INSERT_FREESPACE(curInsert); if (freespace == 0) { - (void) AdvanceXLInsertBuffer(false); - /* OK to ignore update return flag, since we will do flush anyway */ - freespace = INSERT_FREESPACE(Insert); + if (curInsert.xrecoff % XLogSegSize == 0) + curInsert.xrecoff += SizeOfXLogLongPHD; + else + curInsert.xrecoff += SizeOfXLogShortPHD; } - INSERT_RECPTR(checkPoint.redo, Insert, Insert->curridx); + checkPoint.redo = curInsert; /* * Here we update the shared RedoRecPtr for future XLogInsert calls; this - * must be done while holding the insert lock AND the info_lck. + * must be done while holding the insert lock. * * Note: if we fail to complete the checkpoint, RedoRecPtr will be left * pointing past where it really needs to point. This is okay; the only @@ -7787,20 +8571,18 @@ CreateCheckPoint(int flags) * XLogInserts that happen while we are dumping buffers must assume that * their buffer changes are not included in the checkpoint. */ - { - /* use volatile pointer to prevent code rearrangement */ - volatile XLogCtlData *xlogctl = XLogCtl; - - SpinLockAcquire(&xlogctl->info_lck); - RedoRecPtr = xlogctl->Insert.RedoRecPtr = checkPoint.redo; - SpinLockRelease(&xlogctl->info_lck); - } + RedoRecPtr = xlogctl->Insert.RedoRecPtr = checkPoint.redo; /* * Now we can release WAL insert lock, allowing other xacts to proceed * while we are flushing disk buffers. */ - LWLockRelease(WALInsertLock); + WALInsertLockRelease(LW_EXCLUSIVE); + + /* Update the info_lck-protected copy of RedoRecPtr as well */ + SpinLockAcquire(&xlogctl->info_lck); + xlogctl->RedoRecPtr = checkPoint.redo; + SpinLockRelease(&xlogctl->info_lck); /* * If enabled, log checkpoint start. We postpone this until now so as not @@ -7932,7 +8714,9 @@ CreateCheckPoint(int flags) */ if (shutdown && !XLByteEQ(checkPoint.redo, ProcLastRecPtr)) ereport(PANIC, - (errmsg("concurrent transaction log activity while database system is shutting down"))); + (errmsg("concurrent transaction log activity while database system is shutting down (%X/%X vs %X/%X", + checkPoint.redo.xlogid, checkPoint.redo.xrecoff, + ProcLastRecPtr.xlogid, ProcLastRecPtr.xrecoff))); /* * Select point at which we can truncate the log, which we base on the @@ -8185,15 +8969,18 @@ CreateRestartPoint(int flags) * the number of segments replayed since last restartpoint, and request a * restartpoint if it exceeds checkpoint_segments. * - * You need to hold WALInsertLock and info_lck to update it, although - * during recovery acquiring WALInsertLock is just pro forma, because - * there is no other processes updating Insert.RedoRecPtr. + * Like in CreatecheckPoint(), hold WALInsertLock to update it, although + * during recovery acquiring insertpos_lck is just pro forma, because no + * WAL insertions are happening. */ - LWLockAcquire(WALInsertLock, LW_EXCLUSIVE); - SpinLockAcquire(&xlogctl->info_lck); + WALInsertLockAcquire(LW_EXCLUSIVE); xlogctl->Insert.RedoRecPtr = lastCheckPoint.redo; + WALInsertLockRelease(LW_EXCLUSIVE); + + /* Also update the info_lck-protected copy */ + SpinLockAcquire(&xlogctl->info_lck); + xlogctl->RedoRecPtr = lastCheckPoint.redo; SpinLockRelease(&xlogctl->info_lck); - LWLockRelease(WALInsertLock); /* * Prepare to accumulate statistics. @@ -8461,7 +9248,7 @@ XLogReportParameters(void) void UpdateFullPageWrites(void) { - XLogCtlInsert *Insert = &XLogCtl->Insert; + volatile XLogCtlInsert *Insert = &XLogCtl->Insert; /* * Do nothing if full_page_writes has not been changed. @@ -8484,9 +9271,9 @@ UpdateFullPageWrites(void) */ if (fullPageWrites) { - LWLockAcquire(WALInsertLock, LW_EXCLUSIVE); + WALInsertLockAcquire(LW_EXCLUSIVE); Insert->fullPageWrites = true; - LWLockRelease(WALInsertLock); + WALInsertLockRelease(LW_EXCLUSIVE); } /* @@ -8507,9 +9294,9 @@ UpdateFullPageWrites(void) if (!fullPageWrites) { - LWLockAcquire(WALInsertLock, LW_EXCLUSIVE); + WALInsertLockAcquire(LW_EXCLUSIVE); Insert->fullPageWrites = false; - LWLockRelease(WALInsertLock); + WALInsertLockRelease(LW_EXCLUSIVE); } END_CRIT_SECTION(); } @@ -9070,6 +9857,7 @@ XLogFileNameP(TimeLineID tli, XLogSegNo segno) XLogRecPtr do_pg_start_backup(const char *backupidstr, bool fast, char **labelfile) { + volatile XLogCtlInsert *Insert = &XLogCtl->Insert; bool exclusive = (labelfile == NULL); bool backup_started_in_recovery = false; XLogRecPtr checkpointloc; @@ -9131,26 +9919,26 @@ do_pg_start_backup(const char *backupidstr, bool fast, char **labelfile) * Note that forcePageWrites has no effect during an online backup from * the standby. * - * We must hold WALInsertLock to change the value of forcePageWrites, to - * ensure adequate interlocking against XLogInsert(). + * We must hold WALInsertLock to change the value of forcePageWrites, + * to ensure adequate interlocking against XLogInsert(). */ - LWLockAcquire(WALInsertLock, LW_EXCLUSIVE); + WALInsertLockAcquire(LW_EXCLUSIVE); if (exclusive) { - if (XLogCtl->Insert.exclusiveBackup) + if (Insert->exclusiveBackup) { - LWLockRelease(WALInsertLock); + WALInsertLockRelease(LW_EXCLUSIVE); ereport(ERROR, (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), errmsg("a backup is already in progress"), errhint("Run pg_stop_backup() and try again."))); } - XLogCtl->Insert.exclusiveBackup = true; + Insert->exclusiveBackup = true; } else - XLogCtl->Insert.nonExclusiveBackups++; - XLogCtl->Insert.forcePageWrites = true; - LWLockRelease(WALInsertLock); + Insert->nonExclusiveBackups++; + Insert->forcePageWrites = true; + WALInsertLockRelease(LW_EXCLUSIVE); /* Ensure we release forcePageWrites if fail below */ PG_ENSURE_ERROR_CLEANUP(pg_start_backup_callback, (Datum) BoolGetDatum(exclusive)); @@ -9263,13 +10051,13 @@ do_pg_start_backup(const char *backupidstr, bool fast, char **labelfile) * taking a checkpoint right after another is not that expensive * either because only few buffers have been dirtied yet. */ - LWLockAcquire(WALInsertLock, LW_SHARED); - if (XLByteLT(XLogCtl->Insert.lastBackupStart, startpoint)) + WALInsertLockAcquire(LW_EXCLUSIVE); + if (XLByteLT(Insert->lastBackupStart, startpoint)) { - XLogCtl->Insert.lastBackupStart = startpoint; + Insert->lastBackupStart = startpoint; gotUniqueStartpoint = true; } - LWLockRelease(WALInsertLock); + WALInsertLockRelease(LW_EXCLUSIVE); } while (!gotUniqueStartpoint); XLByteToSeg(startpoint, _logSegNo); @@ -9353,27 +10141,28 @@ do_pg_start_backup(const char *backupidstr, bool fast, char **labelfile) static void pg_start_backup_callback(int code, Datum arg) { + volatile XLogCtlInsert *Insert = &XLogCtl->Insert; bool exclusive = DatumGetBool(arg); /* Update backup counters and forcePageWrites on failure */ - LWLockAcquire(WALInsertLock, LW_EXCLUSIVE); + WALInsertLockAcquire(LW_EXCLUSIVE); if (exclusive) { - Assert(XLogCtl->Insert.exclusiveBackup); - XLogCtl->Insert.exclusiveBackup = false; + Assert(Insert->exclusiveBackup); + Insert->exclusiveBackup = false; } else { - Assert(XLogCtl->Insert.nonExclusiveBackups > 0); - XLogCtl->Insert.nonExclusiveBackups--; + Assert(Insert->nonExclusiveBackups > 0); + Insert->nonExclusiveBackups--; } - if (!XLogCtl->Insert.exclusiveBackup && - XLogCtl->Insert.nonExclusiveBackups == 0) + if (!Insert->exclusiveBackup && + Insert->nonExclusiveBackups == 0) { - XLogCtl->Insert.forcePageWrites = false; + Insert->forcePageWrites = false; } - LWLockRelease(WALInsertLock); + WALInsertLockRelease(LW_EXCLUSIVE); } /* @@ -9386,6 +10175,7 @@ pg_start_backup_callback(int code, Datum arg) XLogRecPtr do_pg_stop_backup(char *labelfile, bool waitforarchive) { + volatile XLogCtlInsert *Insert = &XLogCtl->Insert; bool exclusive = (labelfile == NULL); bool backup_started_in_recovery = false; XLogRecPtr startpoint; @@ -9438,9 +10228,9 @@ do_pg_stop_backup(char *labelfile, bool waitforarchive) /* * OK to update backup counters and forcePageWrites */ - LWLockAcquire(WALInsertLock, LW_EXCLUSIVE); + WALInsertLockAcquire(LW_EXCLUSIVE); if (exclusive) - XLogCtl->Insert.exclusiveBackup = false; + Insert->exclusiveBackup = false; else { /* @@ -9449,16 +10239,16 @@ do_pg_stop_backup(char *labelfile, bool waitforarchive) * backups, it is expected that each do_pg_start_backup() call is * matched by exactly one do_pg_stop_backup() call. */ - Assert(XLogCtl->Insert.nonExclusiveBackups > 0); - XLogCtl->Insert.nonExclusiveBackups--; + Assert(Insert->nonExclusiveBackups > 0); + Insert->nonExclusiveBackups--; } - if (!XLogCtl->Insert.exclusiveBackup && - XLogCtl->Insert.nonExclusiveBackups == 0) + if (!Insert->exclusiveBackup && + Insert->nonExclusiveBackups == 0) { - XLogCtl->Insert.forcePageWrites = false; + Insert->forcePageWrites = false; } - LWLockRelease(WALInsertLock); + WALInsertLockRelease(LW_EXCLUSIVE); if (exclusive) { @@ -9736,16 +10526,18 @@ do_pg_stop_backup(char *labelfile, bool waitforarchive) void do_pg_abort_backup(void) { - LWLockAcquire(WALInsertLock, LW_EXCLUSIVE); - Assert(XLogCtl->Insert.nonExclusiveBackups > 0); - XLogCtl->Insert.nonExclusiveBackups--; + volatile XLogCtlInsert *Insert = &XLogCtl->Insert; + + WALInsertLockAcquire(LW_EXCLUSIVE); + Assert(Insert->nonExclusiveBackups > 0); + Insert->nonExclusiveBackups--; - if (!XLogCtl->Insert.exclusiveBackup && - XLogCtl->Insert.nonExclusiveBackups == 0) + if (!Insert->exclusiveBackup && + Insert->nonExclusiveBackups == 0) { - XLogCtl->Insert.forcePageWrites = false; + Insert->forcePageWrites = false; } - LWLockRelease(WALInsertLock); + WALInsertLockRelease(LW_EXCLUSIVE); } /* @@ -9799,14 +10591,14 @@ GetStandbyFlushRecPtr(void) XLogRecPtr GetXLogInsertRecPtr(void) { - XLogCtlInsert *Insert = &XLogCtl->Insert; - XLogRecPtr current_recptr; + volatile XLogCtlInsert *Insert = &XLogCtl->Insert; + uint64 current_bytepos; - LWLockAcquire(WALInsertLock, LW_SHARED); - INSERT_RECPTR(current_recptr, Insert, Insert->curridx); - LWLockRelease(WALInsertLock); + SpinLockAcquire(&Insert->insertpos_lck); + current_bytepos = Insert->CurrBytePos; + SpinLockRelease(&Insert->insertpos_lck); - return current_recptr; + return XLogBytePosToRecPtr(current_bytepos); } /* diff --git a/src/backend/storage/ipc/procarray.c b/src/backend/storage/ipc/procarray.c index 26469c4..8d6567f 100644 --- a/src/backend/storage/ipc/procarray.c +++ b/src/backend/storage/ipc/procarray.c @@ -1753,9 +1753,10 @@ GetOldestActiveTransactionId(void) * the result is somewhat indeterminate, but we don't really care. Even in * a multiprocessor with delayed writes to shared memory, it should be certain * that setting of inCommit will propagate to shared memory when the backend - * takes the WALInsertLock, so we cannot fail to see an xact as inCommit if - * it's already inserted its commit record. Whether it takes a little while - * for clearing of inCommit to propagate is unimportant for correctness. + * takes a lock to write the WAL record, so we cannot fail to see an xact as + * inCommit if it's already inserted its commit record. Whether it takes a + * little while for clearing of inCommit to propagate is unimportant for + * correctness. */ int GetTransactionsInCommit(TransactionId **xids_p) diff --git a/src/backend/storage/lmgr/spin.c b/src/backend/storage/lmgr/spin.c index d262efa..479ef9a 100644 --- a/src/backend/storage/lmgr/spin.c +++ b/src/backend/storage/lmgr/spin.c @@ -56,6 +56,9 @@ SpinlockSemas(void) * * For now, though, we just need a few spinlocks (10 should be plenty) * plus one for each LWLock and one for each buffer header. + * + * XXX: remember to adjust this for the number of spinlocks needed by the + * xlog.c changes before committing! */ return NumLWLocks() + NBuffers + 10; } diff --git a/src/include/access/xlog_internal.h b/src/include/access/xlog_internal.h index a958856..03f854e 100644 --- a/src/include/access/xlog_internal.h +++ b/src/include/access/xlog_internal.h @@ -163,8 +163,7 @@ typedef XLogLongPageHeaderData *XLogLongPageHeader; /* Check if an xrecoff value is in a plausible range */ #define XRecOffIsValid(xrecoff) \ - ((xrecoff) % XLOG_BLCKSZ >= SizeOfXLogShortPHD && \ - (XLOG_BLCKSZ - (xrecoff) % XLOG_BLCKSZ) >= SizeOfXLogRecord) + ((xrecoff) % XLOG_BLCKSZ >= SizeOfXLogShortPHD) /* * The XLog directory and control file (relative to $PGDATA) diff --git a/src/include/pg_config_manual.h b/src/include/pg_config_manual.h index ac45ee6..2883549 100644 --- a/src/include/pg_config_manual.h +++ b/src/include/pg_config_manual.h @@ -247,7 +247,7 @@ * Enable debugging print statements for WAL-related operations; see * also the wal_debug GUC var. */ -/* #define WAL_DEBUG */ +#define WAL_DEBUG /* * Enable tracing of resource consumption during sort operations; diff --git a/src/include/storage/lwlock.h b/src/include/storage/lwlock.h index 6b59efc..3d9d5d9 100644 --- a/src/include/storage/lwlock.h +++ b/src/include/storage/lwlock.h @@ -53,7 +53,7 @@ typedef enum LWLockId ProcArrayLock, SInvalReadLock, SInvalWriteLock, - WALInsertLock, + WALBufMappingLock, WALWriteLock, ControlFileLock, CheckpointLock, @@ -79,6 +79,15 @@ typedef enum LWLockId SerializablePredicateLockListLock, OldSerXidLock, SyncRepLock, + WALInsertTailLock, + FirstWALInsertShareLock, + WALInsertShareLock2, + WALInsertShareLock3, + WALInsertShareLock4, + WALInsertShareLock5, + WALInsertShareLock6, + WALInsertShareLock7, + LastWALInsertShareLock, /* Individual lock IDs end here */ FirstBufMappingLock, FirstLockMgrLock = FirstBufMappingLock + NUM_BUFFER_PARTITIONS,