xloginsert-scale-18.patch
text/x-diff
Filename: xloginsert-scale-18.patch
Type: text/x-diff
Part: 0
Patch
Same data as JSON:
GET /api/v1/attachments/:id/patch
the parsed metadata as JSON — format, series position, per-file stats; never the diff bytes.
API reference →
Format: context
| File | + | − |
|---|---|---|
| src/backend/access/transam/xlog.c | 1246 | 0 |
| src/backend/storage/ipc/procarray.c | 4 | 0 |
| src/backend/storage/lmgr/spin.c | 3 | 0 |
| src/include/storage/lwlock.h | 2 | 0 |
*** a/src/backend/access/transam/xlog.c
--- b/src/backend/access/transam/xlog.c
***************
*** 42,47 ****
--- 42,48 ----
#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,274 **** 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;
* see GetRedoRecPtr. A freshly spawned backend obtains the value during
* InitXLOGAccess.
*/
static XLogRecPtr RedoRecPtr;
/*
* 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
--- 262,281 ----
* (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 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;
+
+ /*
* 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,309 **** 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:
*
! * WALInsertLock: must be held to insert a record into the WAL buffers.
*
* WALWriteLock: must be held to write WAL buffers to disk (XLogWrite or
* XLogFlush).
--- 307,321 ----
* (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. 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:
*
! * 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,320 **** static XLogRecPtr RedoStartLSN = {0, 0};
--- 327,399 ----
* only one checkpointer at a time; currently, with all checkpoints done by
* the checkpointer, this is just pro forma).
*
+ *
+ * Inserting a new WAL record is a two-step process:
+ *
+ * 1. Reserve the right amount of space from the WAL, and the next insertion
+ * slot to advertise that the insertion is in progress. The current head
+ * of reserved space is kept in Insert->CurrPos, 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.
+ *
+ * To allow as much parallelism as possible for step 2, we try hard to avoid
+ * lock contention in that code path. Each insertion is asssigned its own
+ * "XLog insertion slot", which is used to advertise the position the backend
+ * is writing to. The slot is marked as in-use in step 1, while holding
+ * insertpos_lck, by setting the position field in the slot. When the backend
+ * is finished with the insertion, it clears its slot. Each slot is protected
+ * by a separate spinlock, to keep contention minimal.
+ *
+ * The insertion slots also provide a mechanism to wait for an insertion to
+ * finish. This is important when an XLOG page is written out - any
+ * in-progress insertions must finish copying data to the page first, or the
+ * on-disk copy will be incomplete. Waiting is done by the
+ * WaitXLogInsertionsToFinish() function. It adds the current process to the
+ * waiting queue in the slot it needs to wait for, and when that insertion
+ * finishes (or proceeds to the next page, at least), the inserter wakes up
+ * the process.
+ *
+ * The insertion slots form a ring. Insert->nextslot points to the next free
+ * slot, and Insert->lastslot points to the last slot that's still in use.
+ * lastslot can lag behind reality by any number of slots, as long as nextslot
+ * doesn't catch up with it. lastslot is advanced by
+ * WaitXLogInsertionsToFinish(), and is protected by WALInsertTailLock.
+ * nextslot is advanced in ReserveXLogInsertLocation() and is protected by
+ * insertpos_lck. Both slot variables are 32-bit integers, so that they can
+ * be read atomically without holding a lock. nextslot == lastslot means that
+ * all the slots are empty.
+ *
+ * Whenever the ring fills up, ie. when nextslot wraps around and catches up
+ * with lastslot, ReserveXLogInsertLocation() has to wait for the oldest
+ * insertion to finish and advance lastslot, to make room for the new
+ * insertion. This is also handled by WaitXLogInsertionsToFinish().
+ *
+ *
+ * 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,344 **** 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? */
--- 414,435 ----
*/
typedef struct XLogCtlInsert
{
! slock_t insertpos_lck; /* protects all the fields in this struct
! * (except lastslot). */
!
! int32 nextslot; /* next insertion slot to use */
! int32 lastslot; /* last in-use insertion slot (protected by
! * WALInsertTailLock) */
!
! /*
! * CurrPos is the very tip of the reserved WAL space at the moment.
! * The next record will be inserted there (or somewhere after it if
! * there's not enough space on the current page). PrevRecord points to
! * the beginning of the last record already reserved. It might not be
! * fully copied into place yet, but we know its exact location already.
! */
! XLogRecPtr CurrPos;
! XLogRecPtr PrevRecord;
XLogRecPtr RedoRecPtr; /* current redo point for insertions */
bool forcePageWrites; /* forcing full-page writes for PITR? */
***************
*** 364,369 **** typedef struct XLogCtlInsert
--- 455,471 ----
} XLogCtlInsert;
/*
+ * We force XLogCtlInsert to be aligned at a 64-byte boundary, to ensure
+ * that the inserpos_lck spinlock and the fields it protects are all on the
+ * same cache line (assuming the cache line size is at least 64 bytes).
+ * Testing shows that that makes a big difference in performance. If the
+ * struct grows larger than 64 bytes, this needs to be enlarged, too, but
+ * then it won't fit on a single cache line on systems with smaller cache
+ * line size anyway.
+ */
+ #define XLOGCTLINSERT_ALIGNMENT (64)
+
+ /*
* Shared state data for XLogWrite/XLogFlush.
*/
typedef struct XLogCtlWrite
***************
*** 372,387 **** 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: */
XLogCtlInsert Insert;
/* Protected by info_lck: */
XLogwrtRqst LogwrtRqst;
uint32 ckptXidEpoch; /* nextXID & epoch of latest checkpoint */
TransactionId ckptXid;
XLogRecPtr asyncXactLSN; /* LSN of newest async commit/abort */
--- 474,512 ----
pg_time_t lastSegSwitchTime; /* time of last xlog segment switch */
} XLogCtlWrite;
+
+ /*
+ * Slots for in-progress WAL insertions.
+ */
+ typedef struct
+ {
+ slock_t lck;
+ bool finished;
+ XLogRecPtr CurrPos; /* current position this process is inserting to */
+ PGPROC *head; /* head of list of waiting PGPROCs */
+ PGPROC *tail; /* tail of list of waiting PGPROCs */
+ } XLogInsertSlot;
+
+ #define NumXLogInsertSlots 512
+
/*
* Total shared-memory state for XLOG.
*/
typedef struct XLogCtlData
{
! /*
! * 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.
! *
! * Protected by insertpos_lck.
! */
XLogCtlInsert Insert;
+ XLogInsertSlot XLogInsertSlots[NumXLogInsertSlots];
+
/* 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 */
***************
*** 398,406 **** typedef struct XLogCtlData
XLogwrtResult LogwrtResult;
/*
* 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.
*/
char *pages; /* buffers for unwritten XLOG pages */
XLogRecPtr *xlblocks; /* 1st byte ptr-s + XLOG_BLCKSZ */
--- 523,540 ----
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 WALBufMappingLock and WALWriteLock.
*/
char *pages; /* buffers for unwritten XLOG pages */
XLogRecPtr *xlblocks; /* 1st byte ptr-s + XLOG_BLCKSZ */
***************
*** 478,505 **** 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.
*/
! /* Free space remaining in the current xlog page buffer */
! #define INSERT_FREESPACE(Insert) \
! (XLOG_BLCKSZ - ((Insert)->currpos - (char *) (Insert)->currpage))
! /* Construct XLogRecPtr value for current insertion point */
! #define INSERT_RECPTR(recptr,Insert,curridx) \
! ( \
! (recptr).xlogid = XLogCtl->xlblocks[curridx].xlogid, \
! (recptr).xrecoff = \
! XLogCtl->xlblocks[curridx].xrecoff - INSERT_FREESPACE(Insert) \
! )
! #define PrevBufIdx(idx) \
! (((idx) == 0) ? XLogCtl->XLogCacheBlck : ((idx) - 1))
! #define NextBufIdx(idx) \
! (((idx) == XLogCtl->XLogCacheBlck) ? 0 : ((idx) + 1))
/*
* Private, possibly out-of-date copy of shared LogwrtResult.
--- 612,660 ----
static ControlFileData *ControlFile = NULL;
/*
! * 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))
! /*
! * Macros to advance to next buffer index and insertion slot.
! */
! #define NextBufIdx(idx) \
! (((idx) == XLogCtl->XLogCacheBlck) ? 0 : ((idx) + 1))
! #define NextSlotNo(idx) (((idx) + 1) % NumXLogInsertSlots)
! /*
! * The mapping from a slot number to an element in the XLogInsertSlots array
! * is not a simple linear mapping as you would expect. We want to spread out
! * the accesses to the slots, to reduce the contention on the cache lines of
! * logically adjacent slots.
! *
! * To do that, we swap the four low-order bits with high order bits:
! *
! * 123456789 -> 1567892345
! */
! static inline int
! SlotNoToIdx(int slotno)
! {
! return (slotno & 0xffffff00) | (slotno & 0xf0) >> 4 | (slotno & 0x0f) << 4;
! }
!
! /*
! * 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 * (uint64) XLogFileSize) + (recptr).xrecoff)) / XLOG_BLCKSZ) % (XLogCtl->XLogCacheBlck + 1))
!
! #define XLogRecEndPtrToBufIdx(recptr) \
! ((((((uint64) (recptr).xlogid * (uint64) XLogFileSize) + (recptr).xrecoff - 1)) / XLOG_BLCKSZ) % (XLogCtl->XLogCacheBlck + 1))
/*
* Private, possibly out-of-date copy of shared LogwrtResult.
***************
*** 625,633 **** static void KeepLogSeg(XLogRecPtr recptr, uint32 *logId, uint32 *logSeg);
static bool XLogCheckBuffer(XLogRecData *rdata, bool doPageWrites,
XLogRecPtr *lsn, BkpBlock *bkpb);
! static bool AdvanceXLInsertBuffer(bool new_segment);
static bool XLogCheckpointNeeded(uint32 logid, uint32 logseg);
! static void XLogWrite(XLogwrtRqst WriteRqst, bool flexible, bool xlog_switch);
static bool InstallXLogFileSegment(uint32 *log, uint32 *seg, char *tmppath,
bool find_free, int *max_advance,
bool use_lock);
--- 780,788 ----
static bool XLogCheckBuffer(XLogRecData *rdata, bool doPageWrites,
XLogRecPtr *lsn, BkpBlock *bkpb);
! static void AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic);
static bool XLogCheckpointNeeded(uint32 logid, uint32 logseg);
! static void XLogWrite(XLogwrtRqst WriteRqst, bool flexible);
static bool InstallXLogFileSegment(uint32 *log, uint32 *seg, char *tmppath,
bool find_free, int *max_advance,
bool use_lock);
***************
*** 674,679 **** static bool read_backup_label(XLogRecPtr *checkPointLoc,
--- 829,849 ----
static void rm_redo_error_callback(void *arg);
static int get_sync_bit(int method);
+ static void CopyXLogRecordToWAL(int write_len, bool isLogSwitch,
+ XLogRecord *rechdr,
+ XLogRecData *rdata, pg_crc32 rdata_crc,
+ int slotno,
+ XLogRecPtr StartPos, XLogRecPtr EndPos);
+ static bool ReserveXLogInsertLocation(int size,
+ bool isLogSwitch,
+ XLogRecPtr *PrevRecord_p, XLogRecPtr *StartPos_p,
+ XLogRecPtr *EndPos_p,
+ int *myslotno_p, bool *updrqst_p);
+ static void UpdateSlotCurrPos(int myslotno, XLogRecPtr CurrPos, bool finished);
+ static void ReuseOldSlots(void);
+ static XLogRecPtr WaitXLogInsertionsToFinish(XLogRecPtr upto);
+ static char *GetXLogBuffer(int slotno, XLogRecPtr ptr);
+
/*
* Insert an XLOG record having the specified RMID and info bytes,
***************
*** 693,705 **** static int get_sync_bit(int method);
XLogRecPtr
XLogInsert(RmgrId rmid, uint8 info, XLogRecData *rdata)
{
- XLogCtlInsert *Insert = &XLogCtl->Insert;
- XLogRecord *record;
- XLogContRecord *contrecord;
- XLogRecPtr RecPtr;
- XLogRecPtr WriteRqst;
- uint32 freespace;
- int curridx;
XLogRecData *rdt;
XLogRecData *rdt_lastnormal;
Buffer dtbuf[XLR_MAX_BKP_BLOCKS];
--- 863,868 ----
***************
*** 714,722 **** XLogInsert(RmgrId rmid, uint8 info, XLogRecData *rdata)
write_len;
unsigned i;
bool updrqst;
- bool doPageWrites;
bool isLogSwitch = (rmid == RM_XLOG_ID && info == XLOG_SWITCH);
uint8 info_orig = info;
/* cross-check on whether we should be here or not */
if (!XLogInsertAllowed())
--- 877,889 ----
write_len;
unsigned i;
bool updrqst;
bool isLogSwitch = (rmid == RM_XLOG_ID && info == XLOG_SWITCH);
uint8 info_orig = info;
+ XLogRecord rechdr;
+ XLogRecPtr PrevRecord;
+ XLogRecPtr StartPos;
+ XLogRecPtr EndPos;
+ int myslotno;
/* cross-check on whether we should be here or not */
if (!XLogInsertAllowed())
***************
*** 734,742 **** 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;
}
/*
--- 901,909 ----
*/
if (IsBootstrapProcessingMode() && rmid != RM_XLOG_ID)
{
! EndPos.xlogid = 0;
! EndPos.xrecoff = SizeOfXLogLongPHD; /* start of 1st chkpt record */
! return EndPos;
}
/*
***************
*** 760,773 **** begin:;
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;;)
{
--- 927,932 ----
***************
*** 903,1035 **** begin:;
for (rdt = rdata; rdt != NULL; rdt = rdt->next)
COMP_CRC32(rdata_crc, rdt->data, rdt->len);
! 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.
*/
! if (!XLByteEQ(RedoRecPtr, Insert->RedoRecPtr))
{
! Assert(XLByteLT(RedoRecPtr, Insert->RedoRecPtr));
! RedoRecPtr = Insert->RedoRecPtr;
! if (doPageWrites)
{
! 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;
! }
! }
}
}
!
! /*
! * 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)
{
! /* Oops, must redo it with full-page data. */
! LWLockRelease(WALInsertLock);
! END_CRIT_SECTION();
! rdt_lastnormal->next = NULL;
! info = info_orig;
! goto begin;
}
/*
! * If there isn't enough space on the current XLOG page for a record
! * header, advance to the next page (leaving the unused space as zeroes).
*/
! updrqst = false;
! freespace = INSERT_FREESPACE(Insert);
! if (freespace < SizeOfXLogRecord)
{
! updrqst = AdvanceXLInsertBuffer(false);
! freespace = INSERT_FREESPACE(Insert);
! }
! /* Compute record's XLOG location */
! curridx = Insert->curridx;
! INSERT_RECPTR(RecPtr, Insert, curridx);
/*
! * 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 (isLogSwitch &&
! (RecPtr.xrecoff % XLogSegSize) == SizeOfXLogLongPHD)
{
! /* We can release insert lock immediately */
! LWLockRelease(WALInsertLock);
!
! RecPtr.xrecoff -= SizeOfXLogLongPHD;
! if (RecPtr.xrecoff == 0)
! {
! /* crossing a logid boundary */
! RecPtr.xlogid -= 1;
! RecPtr.xrecoff = XLogFileSize;
! }
!
! LWLockAcquire(WALWriteLock, LW_EXCLUSIVE);
! LogwrtResult = XLogCtl->LogwrtResult;
! if (!XLByteLE(RecPtr, LogwrtResult.Flush))
! {
! XLogwrtRqst FlushRqst;
!
! FlushRqst.Write = RecPtr;
! FlushRqst.Flush = RecPtr;
! XLogWrite(FlushRqst, false, false);
! }
! LWLockRelease(WALWriteLock);
! END_CRIT_SECTION();
! return RecPtr;
}
! /* Insert record header */
!
! record = (XLogRecord *) Insert->currpos;
! record->xl_prev = Insert->PrevRecord;
! record->xl_xid = GetCurrentTransactionIdIfAny();
! record->xl_tot_len = SizeOfXLogRecord + write_len;
! record->xl_len = len; /* doesn't include backup blocks */
! record->xl_info = info;
! record->xl_rmid = rmid;
!
! /* Now we can finish computing the record's CRC */
! COMP_CRC32(rdata_crc, (char *) record + sizeof(pg_crc32),
! SizeOfXLogRecord - sizeof(pg_crc32));
! FIN_CRC32(rdata_crc);
! record->xl_crc = rdata_crc;
#ifdef WAL_DEBUG
if (XLOG_DEBUG)
--- 1062,1167 ----
for (rdt = rdata; rdt != NULL; rdt = rdt->next)
COMP_CRC32(rdata_crc, rdt->data, rdt->len);
! /* Construct record header. */
! MemSet(&rechdr, 0, sizeof(rechdr));
! /* rechdr.xl_prev is set later */
! 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;
! START_CRIT_SECTION();
/*
! * Try to reserve space for the record from the WAL.
*/
! if (!ReserveXLogInsertLocation(write_len, isLogSwitch,
! &PrevRecord, &StartPos, &EndPos,
! &myslotno, &updrqst))
{
! /*
! * Reservation failed. This could be because the record was an
! * XLOG_SWITCH, and we're exactly at the start of a segment. In that
! * case 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.
! *
! * The other reason for failure is that someone changed RedoRecPtr
! * or forcePageWrites after we had constructed our WAL record. In
! * that case we need to redo it with full-page data.
! */
! END_CRIT_SECTION();
! if (isLogSwitch && !XLogRecPtrIsInvalid(EndPos))
{
! XLogFlush(EndPos);
! return EndPos;
! }
! else
! {
! rdt_lastnormal->next = NULL;
! info = info_orig;
! goto begin;
}
}
! else
{
! /*
! * Reservation succeeded. Finish the record header by setting
! * prev-link (now that we know it), and finish computing the record's
! * CRC (in CopyXLogRecordToWAL). Then copy the record to the space
! * we reserved.
! */
! rechdr.xl_prev = PrevRecord;
! CopyXLogRecordToWAL(write_len, isLogSwitch, &rechdr,
! rdata, rdata_crc, myslotno, StartPos, EndPos);
}
+ END_CRIT_SECTION();
/*
! * Update shared LogwrtRqst.Write, if we crossed page boundary.
*/
! if (updrqst)
{
! /* use volatile pointer to prevent code rearrangement */
! volatile XLogCtlData *xlogctl = XLogCtl;
! 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 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)
{
! 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.
! */
! EndPos.xlogid = StartPos.xlogid;
! EndPos.xrecoff = StartPos.xrecoff + SizeOfXLogRecord;
}
! /*
! * Update our global variables
! */
! ProcLastRecPtr = StartPos;
! XactLastRecEnd = EndPos;
#ifdef WAL_DEBUG
if (XLOG_DEBUG)
***************
*** 1038,1219 **** begin:;
initStringInfo(&buf);
appendStringInfo(&buf, "INSERT @ %X/%X: ",
! RecPtr.xlogid, RecPtr.xrecoff);
! xlog_outrec(&buf, record);
if (rdata->data != NULL)
{
appendStringInfo(&buf, " - ");
! RmgrTable[record->xl_rmid].rm_desc(&buf, record->xl_info, rdata->data);
}
elog(LOG, "%s", buf.data);
pfree(buf.data);
}
#endif
- /* Record begin of record in appropriate places */
- ProcLastRecPtr = RecPtr;
- Insert->PrevRecord = RecPtr;
-
- Insert->currpos += SizeOfXLogRecord;
- freespace -= SizeOfXLogRecord;
-
/*
! * Append the data, including backup blocks if any
*/
! while (write_len)
! {
! while (rdata->data == NULL)
! rdata = rdata->next;
! if (freespace > 0)
{
! if (rdata->len > freespace)
{
! memcpy(Insert->currpos, rdata->data, freespace);
rdata->data += freespace;
rdata->len -= freespace;
! write_len -= freespace;
! }
! else
! {
! memcpy(Insert->currpos, rdata->data, rdata->len);
! freespace -= rdata->len;
! write_len -= rdata->len;
! Insert->currpos += rdata->len;
! rdata = rdata->next;
! continue;
}
}
! /* Use next buffer */
! updrqst = AdvanceXLInsertBuffer(false);
! curridx = Insert->curridx;
! /* Insert cont-record header */
! Insert->currpage->xlp_info |= XLP_FIRST_IS_CONTRECORD;
! contrecord = (XLogContRecord *) Insert->currpos;
! contrecord->xl_rem_len = write_len;
! Insert->currpos += SizeOfXLogContRecord;
! freespace = INSERT_FREESPACE(Insert);
}
! /* Ensure next record will be properly aligned */
! Insert->currpos = (char *) Insert->currpage +
! MAXALIGN(Insert->currpos - (char *) Insert->currpage);
! freespace = INSERT_FREESPACE(Insert);
/*
! * The recptr I return is the beginning of the *next* record. This will be
! * stored as LSN for changed data pages...
*/
! INSERT_RECPTR(RecPtr, Insert, curridx);
/*
! * 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 (isLogSwitch)
! {
! XLogwrtRqst FlushRqst;
! XLogRecPtr OldSegEnd;
! TRACE_POSTGRESQL_XLOG_SWITCH();
! LWLockAcquire(WALWriteLock, LW_EXCLUSIVE);
/*
! * Flush through the end of the page containing XLOG_SWITCH, and
! * perform end-of-segment actions (eg, notifying archiver).
*/
! 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);
!
! /* There should be no unwritten data */
! curridx = Insert->curridx;
! Assert(curridx == XLogCtl->Write.curridx);
!
! /* Compute end address of old segment */
! OldSegEnd = XLogCtl->xlblocks[curridx];
! OldSegEnd.xrecoff -= XLOG_BLCKSZ;
! if (OldSegEnd.xrecoff == 0)
! {
! /* crossing a logid boundary */
! OldSegEnd.xlogid -= 1;
! OldSegEnd.xrecoff = XLogFileSize;
! }
! /* Make it look like we've written and synced all of old segment */
! LogwrtResult.Write = OldSegEnd;
! LogwrtResult.Flush = OldSegEnd;
/*
! * Update shared-memory status --- this code should match XLogWrite
*/
{
! /* use volatile pointer to prevent code rearrangement */
! volatile XLogCtlData *xlogctl = XLogCtl;
! 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);
! }
! LWLockRelease(WALWriteLock);
! updrqst = false; /* done already */
}
else
{
! /* normal case, ie not xlog switch */
! /* Need to update shared LogwrtRqst if some block was filled up */
! if (freespace < SizeOfXLogRecord)
{
! /* curridx is filled and available for writing out */
updrqst = true;
}
! else
! {
! /* if updrqst already set, write through end of previous buf */
! curridx = PrevBufIdx(curridx);
! }
! WriteRqst = XLogCtl->xlblocks[curridx];
}
! LWLockRelease(WALInsertLock);
! if (updrqst)
{
! /* use volatile pointer to prevent code rearrangement */
! volatile XLogCtlData *xlogctl = XLogCtl;
! 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);
}
! XactLastRecEnd = RecPtr;
! END_CRIT_SECTION();
! return RecPtr;
}
/*
--- 1170,1954 ----
initStringInfo(&buf);
appendStringInfo(&buf, "INSERT @ %X/%X: ",
! EndPos.xlogid, EndPos.xrecoff);
! xlog_outrec(&buf, &rechdr);
if (rdata->data != NULL)
{
appendStringInfo(&buf, " - ");
! RmgrTable[rmid].rm_desc(&buf, rechdr.xl_info, rdata->data);
}
elog(LOG, "%s", buf.data);
pfree(buf.data);
}
#endif
/*
! * The recptr I return is the beginning of the *next* record. This will
! * be stored as LSN for changed data pages...
*/
! 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, XLogRecord *rechdr,
! XLogRecData *rdata, pg_crc32 rdata_crc,
! int myslotno,
! XLogRecPtr StartPos, XLogRecPtr EndPos)
! {
! char *currpos;
! XLogRecord *record;
! int freespace;
! int written;
! XLogRecPtr CurrPos;
!
! /* Get the right WAL page to start inserting to */
! CurrPos = StartPos;
! currpos = GetXLogBuffer(myslotno, CurrPos);
!
! /* Copy the record header in place, and finish calculating CRC */
! record = (XLogRecord *) currpos;
! memcpy(record, rechdr, sizeof(XLogRecord));
! COMP_CRC32(rdata_crc, currpos + sizeof(pg_crc32),
! SizeOfXLogRecord - sizeof(pg_crc32));
! FIN_CRC32(rdata_crc);
! record->xl_crc = rdata_crc;
!
! currpos += SizeOfXLogRecord;
! XLByteAdvance(CurrPos, SizeOfXLogRecord);
! freespace = INSERT_FREESPACE(CurrPos);
!
! if (!isLogSwitch)
! {
! /* Copy record data */
! written = 0;
! while (rdata != NULL)
{
! while (rdata->len > freespace)
{
! /*
! * Write what fits on this page, then write the continuation
! * record, and continue on the next page.
! */
! XLogContRecord *contrecord;
!
! memcpy(currpos, rdata->data, freespace);
rdata->data += freespace;
rdata->len -= freespace;
! written += freespace;
! XLByteAdvance(CurrPos, freespace);
!
! /*
! * Get pointer to beginning of next page, and set the
! * XLP_FIRST_IS_CONTRECORD flag in the page header.
! *
! * It's safe to set the contrecord flag without a lock on the
! * page. All the other flags are set in AdvanceXLInsertBuffer,
! * and we're the only backend that needs to set the contrecord
! * flag.
! */
! currpos = GetXLogBuffer(myslotno, CurrPos);
! ((XLogPageHeader) currpos)->xlp_info |= XLP_FIRST_IS_CONTRECORD;
!
! /* skip over the page header, and write continuation record */
! if (CurrPos.xrecoff % XLogSegSize == 0)
! {
! CurrPos.xrecoff += SizeOfXLogLongPHD;
! currpos += SizeOfXLogLongPHD;
! }
! else
! {
! CurrPos.xrecoff += SizeOfXLogShortPHD;
! currpos += SizeOfXLogShortPHD;
! }
! contrecord = (XLogContRecord *) currpos;
! contrecord->xl_rem_len = write_len - written;
!
! currpos += SizeOfXLogContRecord;
! CurrPos.xrecoff += SizeOfXLogContRecord;
!
! freespace = INSERT_FREESPACE(CurrPos);
}
+
+ 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);
! /* Align the end position, so that the next record starts aligned */
! CurrPos.xrecoff = MAXALIGN(CurrPos.xrecoff);
! if (CurrPos.xrecoff >= XLogFileSize)
! {
! /* crossed a logid boundary */
! CurrPos.xlogid += 1;
! CurrPos.xrecoff = 0;
! }
!
! if (!XLByteEQ(CurrPos, EndPos))
! elog(PANIC, "space reserved for WAL record does not match what was written");
}
+ else
+ {
+ /* An xlog-switch record doesn't contain any data besides the header */
+ Assert(write_len == 0);
+
+ /*
+ * An xlog-switch record consumes all the remaining space on the
+ * WAL segment. We have already reserved it for us, but we still need
+ * to make sure it's been 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.
+ *
+ * We do this one page at a time, to make sure we don't deadlock
+ * against ourselves if wal_buffers < XLOG_SEG_SIZE.
+ */
+ Assert(EndPos.xrecoff % XLogSegSize == 0);
+
+ /* Use up all the remaining space on the first page */
+ XLByteAdvance(CurrPos, freespace);
! while (XLByteLT(CurrPos, EndPos))
! {
! /*
! * like in the non-xlog-switch codepath, let others know that
! * we're done writing up to the end of this page
! */
! UpdateSlotCurrPos(myslotno, CurrPos, false);
! /* initialize the next page (if not initialized already */
! AdvanceXLInsertBuffer(CurrPos, false);
! XLByteAdvance(CurrPos, XLOG_BLCKSZ);
! }
! }
/*
! * Done! Clear CurrPos in our slot to let others know that we're
! * finished.
*/
! UpdateSlotCurrPos(myslotno, InvalidXLogRecPtr, true);
/*
! * When we run out of insertion slots, the next inserter has to grab the
! * WALInsertTailLock to clean up some old slots. That stalls all new
! * insertions. The WAL writer process cleans up old slots periodically,
! * but on a busy system that might not be enough. So we try to clean up
! * old ones every time we've gone through 1/4 of all the slots.
*/
! if (myslotno % (NumXLogInsertSlots / 4) == 0)
! ReuseOldSlots();
! }
! /*
! * 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, and *PrevRecord_p to the beginning of the previous record to set
! * to the prev-link of the record header.
! *
! * 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.
! *
! * *updrqst_p is set to true, if this record ends on different page than
! * the previous one - the caller should update the shared LogwrtRqst value
! * after it's done inserting the record in that case, so that the WAL page
! * that filled up gets written out at the next convenient moment.
! *
! * While holding insertpos_lck, sets myslot->CurrPos to the starting position,
! * (or the end of previous record, to be exact) to let others know that we're
! * busy inserting to the reserved area. The caller must clear it when the
! * insertion is finished.
! *
! * Returns true on success, or false if RedoRecPtr or forcePageWrites was
! * changed. On failure, the shared state is not modified.
! *
! * 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 bool
! ReserveXLogInsertLocation(int size,
! bool isLogSwitch,
! XLogRecPtr *PrevRecord_p, XLogRecPtr *StartPos_p,
! XLogRecPtr *EndPos_p,
! int *myslotno_p, bool *updrqst_p)
! {
! volatile XLogCtlInsert *Insert = &XLogCtl->Insert;
! int freespace;
! XLogRecPtr ptr;
! XLogRecPtr StartPos;
! int32 nextslot;
! int32 lastslot;
! bool updrqst = false;
! bool didPageWrites = doPageWrites;
!
! /* log-switch records should contain no data */
! Assert(!isLogSwitch || size == 0);
! size = SizeOfXLogRecord + size;
!
! retry:
! SpinLockAcquire(&Insert->insertpos_lck);
+ doPageWrites = Insert->forcePageWrites || Insert->fullPageWrites;
+ if (!XLByteEQ(RedoRecPtr, Insert->RedoRecPtr) ||
+ (!didPageWrites && doPageWrites))
+ {
/*
! * Oops, a checkpoint just happened, or forcePageWrites was just
! * turned on. Start XLogInsert() all over, because we might have to
! * include more full-page images in the record.
*/
! RedoRecPtr = Insert->RedoRecPtr;
! SpinLockRelease(&Insert->insertpos_lck);
! *EndPos_p = InvalidXLogRecPtr;
! return false;
! }
!
! /*
! * Reserve the next insertion slot for us.
! *
! * First check that the slot is not still in use. Modifications to
! * lastslot are protected by WALInsertTailLock, but here we assume that
! * reading an int32 is atomic. Another process might advance lastslot at
! * the same time, but not past nextslot.
! */
! lastslot = Insert->lastslot;
! nextslot = Insert->nextslot;
! if (NextSlotNo(nextslot) == lastslot)
! {
! /*
! * Oops, we've "caught our tail" and the oldest slot is still in use.
! * Have to wait for it to become vacant, and retry.
! */
! SpinLockRelease(&Insert->insertpos_lck);
! WaitXLogInsertionsToFinish(InvalidXLogRecPtr);
! goto retry;
! }
!
! /*
! * Got the slot. Now reserve the right amount of space from the WAL for
! * our record.
! */
! ptr = Insert->CurrPos;
! *PrevRecord_p = Insert->PrevRecord;
!
! /*
! * If there isn't enough space on the current XLOG page for a record
! * header, advance to the next page (leaving the unused space as zeroes).
! */
! freespace = INSERT_FREESPACE(ptr);
! if (freespace < SizeOfXLogRecord)
! {
! XLByteAdvance(ptr, freespace);
! if (ptr.xrecoff % XLogSegSize == 0)
! ptr.xrecoff += SizeOfXLogLongPHD;
! else
! ptr.xrecoff += SizeOfXLogShortPHD;
! freespace = INSERT_FREESPACE(ptr);
! updrqst = true;
! }
+ /*
+ * We are now at the starting position of our record. Now figure out how
+ * the data will be split across the WAL pages, to calculate where the
+ * record ends.
+ */
+ StartPos = ptr;
+
+ if (isLogSwitch)
+ {
/*
! * 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). Otherwise the
! * XLOG_SWITCH record should consume all the remaining space on the
! * current segment.
*/
+ if ((ptr.xrecoff % XLogSegSize) == SizeOfXLogLongPHD)
{
! /* We can release insert lock immediately */
! SpinLockRelease(&Insert->insertpos_lck);
! ptr.xrecoff -= SizeOfXLogLongPHD;
! if (ptr.xrecoff == 0)
! {
! /* crossing a logid boundary */
! ptr.xlogid -= 1;
! ptr.xrecoff = XLogFileSize;
! }
! *EndPos_p = ptr;
! *StartPos_p = ptr;
! *myslotno_p = 0;
! return false;
! }
! else
! {
! if (ptr.xrecoff % XLOG_SEG_SIZE != 0)
! {
! int segleft = XLOG_SEG_SIZE - (ptr.xrecoff % XLOG_SEG_SIZE);
! ptr.xrecoff += segleft;
! }
! updrqst = true;
! }
}
else
{
! /*
! * A normal record, ie. not xlog-switch.
! *
! * Calculate how the record will be laid out across WAL pages. The
! * straightforward way to do this would be a loop that fills in the
! * WAL pages one at a time, tracking how much of the size is still
! * left. That's how the CopyXLogRecordToWAL() works when actually
! * copying the data. However, we want to avoid looping to keep this
! * spinlock-protected block as short as possible, in case the record
! * spans pages.
! */
! int sizeleft = size;
! if (sizeleft > freespace)
{
! int pagesneeded;
! int pagesleftonseg;
! int fullpages;
!
! /* First fill the first page with as much data as fits. */
! sizeleft -= freespace;
! ptr.xrecoff += freespace;
!
! /* We're now positioned at the beginning of the next page */
! Assert(ptr.xrecoff % XLOG_BLCKSZ == 0);
! do
! {
! if (ptr.xrecoff >= XLogFileSize)
! {
! /* crossing a logid boundary */
! ptr.xlogid++;
! ptr.xrecoff = 0;
! }
!
! /*
! * If we're positioned at the beginning of a segment, take
! * into account that the first page needs a long header.
! */
! if (ptr.xrecoff % XLOG_SEG_SIZE == 0)
! sizeleft += (SizeOfXLogLongPHD - SizeOfXLogShortPHD);
!
! /*
! * Calculate the number of extra pages we need. Each page
! * will have a continuation record at the beginning.
! *
! * We do the calculation assuming that all the pages have a
! * short header. We don't know whether we have to cross to
! * the next segment until we've calculated how many pages we
! * need. If it turns out that we do, we'll fill up the current
! * segment, and loop back to add the long page header to
! * sizeleft, and continue calculation from there.
! */
! #define SpaceOnXLogPage (XLOG_BLCKSZ - SizeOfXLogShortPHD - SizeOfXLogContRecord)
! pagesneeded = (sizeleft + SpaceOnXLogPage - 1) / SpaceOnXLogPage;
!
! pagesleftonseg = (XLOG_SEG_SIZE - (ptr.xrecoff % XLOG_SEG_SIZE)) / XLOG_BLCKSZ;
!
! if (pagesneeded <= pagesleftonseg)
! {
! /*
! * Fits in this segment. Skip over all the full pages, to
! * the last page that will (possibly) be only partially
! * filled.
! */
! fullpages = pagesneeded - 1;
! }
! else
! {
! /*
! * Doesn't fit in this segment. Fit as much as does, and
! * continue from next segment.
! */
! fullpages = pagesleftonseg;
! }
!
! sizeleft -= fullpages * SpaceOnXLogPage;
! ptr.xrecoff += fullpages * XLOG_BLCKSZ;
! } while (pagesneeded > pagesleftonseg);
!
! /*
! * We're now positioned at the beginning of the last page this
! * record spans. The rest should fit on this page.
! *
! * Note: We already took into account the long header above.
! */
! ptr.xrecoff += SizeOfXLogShortPHD;
! ptr.xrecoff += SizeOfXLogContRecord;
!
! Assert(sizeleft <= INSERT_FREESPACE(ptr) && sizeleft > 0);
!
updrqst = true;
}
!
! /*
! * The rest fits on this page. Note that we mustn't use XLByteAdvance
! * here, because if this record just fills up a logical log file,
! * we want ptr.xlogid to point to log file that was filled, not the
! * next one. XLogWrite gets upset otherwise.
! */
! ptr.xrecoff += sizeleft;
!
! /* Align the end position, so that the next record starts aligned */
! ptr.xrecoff = MAXALIGN(ptr.xrecoff);
}
! /* Update the shared state before releasing the lock */
! Insert->CurrPos = ptr;
! Insert->PrevRecord = StartPos;
! Insert->nextslot = NextSlotNo(nextslot);
! SpinLockRelease(&Insert->insertpos_lck);
!
! #ifdef RESERVE_XLOGINSERT_LOCATION_DEBUG
! elog(LOG, "reserved xlog: prev %X/%X, start %X/%X, end %X/%X (len %d)",
! PrevRecord_p->xlogid, PrevRecord_p->xrecoff,
! StartPos.xlogid, StartPos.xrecoff,
! ptr.xlogid, ptr.xrecoff,
! size);
! #endif
!
! /*
! * XLogWrite gets upset if given a pointer to the beginning of a logical
! * log file. If the record ends exactly at the end of one, we must return
! * the last byte on previous log file + 1 as our ending position, not
! * the first byte on the next log file.
! */
! Assert(ptr.xrecoff != 0);
!
! *EndPos_p = ptr;
! *StartPos_p = StartPos;
! *myslotno_p = nextslot;
! *updrqst_p = updrqst;
!
! return true;
! }
!
! /*
! * Update slot's CurrPos variable, and wake up anyone waiting on it.
! */
! static void
! UpdateSlotCurrPos(int myslotno, XLogRecPtr CurrPos, bool finished)
! {
! volatile XLogInsertSlot *myslot = &XLogCtl->XLogInsertSlots[SlotNoToIdx(myslotno)];
! PGPROC *head;
!
! /* Must not move backwards. */
! Assert(finished || XLByteLE(myslot->CurrPos, CurrPos));
!
! /*
! * The write-barrier ensures that the changes we made to the WAL pages
! * are visible to everyone before the update of CurrPos.
! *
! * XXX: I'm not sure if this is necessary. Doesn't the spinlock
! * acquire/release act as an implicit barrier?
! */
! pg_write_barrier();
!
! SpinLockAcquire(&myslot->lck);
! myslot->finished = finished;
! myslot->CurrPos = CurrPos;
! head = myslot->head;
! myslot->head = myslot->tail = NULL;
! SpinLockRelease(&myslot->lck);
! while (head != NULL)
{
! PGPROC *proc = head;
! head = proc->lwWaitLink;
! proc->lwWaitLink = NULL;
! proc->lwWaiting = false;
! PGSemaphoreUnlock(&proc->sem);
! }
! }
! /*
! * 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.
! *
! * The caller must ensure that the page containing the requested location
! * isn't evicted yet, and won't be evicted, by holding onto an XLOG insertion
! * slot with CurrPos set to something <= ptr. This function can update the
! * advertised CurrPos up to 'ptr', so you are not allowed to access anything
! * older than 'ptr' after calling this function.
! */
! static char *
! GetXLogBuffer(int myslotno, XLogRecPtr ptr)
! {
! 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)
! return cachedPos + ptr.xrecoff % XLOG_BLCKSZ;
!
! cachedXlogid = ptr.xlogid;
! cachedPage = ptr.xrecoff / XLOG_BLCKSZ;
!
! /*
! * The XLog buffer cache is organized so 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, which shouldn't happen as long as the caller
! * has set its slot's CurrPos correctly.
! *
! * 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.xlogid = ptr.xlogid;
! expectedEndPtr.xrecoff = ptr.xrecoff - ptr.xrecoff % XLOG_BLCKSZ + XLOG_BLCKSZ;
!
! endptr = XLogCtl->xlblocks[idx];
! if (!XLByteEQ(expectedEndPtr, endptr))
! {
! /*
! * Before we try to initialize the buffer for this page, let others
! * know how far we've inserted, by updating the CurrPos field in our
! * slot. This is important because AdvanceXLInsertBuffer() might need
! * to wait for some insertions to finish so that it can write out the
! * old page from the buffer. Updating our slot before waiting for a
! * new buffer ensures that we don't deadlock with ourselves if we run
! * out of clean buffers.
! *
! * Note that we must not advance CurrPos past the page header yet.
! * Otherwise someone might try to flush up to that point, which would
! * fail if the next page was not initialized yet.
! */
! XLogRecPtr pagebeginptr = ptr;
! pagebeginptr.xrecoff -= ptr.xrecoff % XLOG_BLCKSZ;
! UpdateSlotCurrPos(myslotno, pagebeginptr, false);
!
! 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.
! */
! cachedPos = XLogCtl->pages + idx * (Size) XLOG_BLCKSZ;
! return XLogCtl->pages + idx * (Size) XLOG_BLCKSZ +
! ptr.xrecoff % XLOG_BLCKSZ;
! }
! /*
! * Try to mark old insertion slots as free for reuse.
! */
! static void
! ReuseOldSlots(void)
! {
! volatile XLogCtlInsert *Insert = &XLogCtl->Insert;
! int lastslot;
! int nextslot;
! /* Give up if someone else is already doing this */
! if (!LWLockConditionalAcquire(WALInsertTailLock, LW_EXCLUSIVE))
! return;
!
! SpinLockAcquire(&Insert->insertpos_lck);
! lastslot = Insert->lastslot;
! nextslot = Insert->nextslot;
! SpinLockRelease(&Insert->insertpos_lck);
!
! while (lastslot != nextslot)
! {
! /*
! * Check if the oldest slot is still in use. We don't do any locking
! * here, we just give up as soon as we find a slot that's still in
! * use.
! */
! volatile XLogInsertSlot *slot;
! slot = &XLogCtl->XLogInsertSlots[SlotNoToIdx(lastslot)];
!
! if (!slot->finished)
! break;
!
! /* Reinitialize the slot for reuse */
! slot->finished = false;
!
! lastslot = NextSlotNo(lastslot);
! }
!
! /*
! * Update lastslot before we release the lock. (We don't need to grab
! * insertpos_lck here, on the assumption that writing an int32 is atomic)
! */
! Insert->lastslot = lastslot;
! LWLockRelease(WALInsertTailLock);
! }
!
! /*
! * Wait for any insertions < upto to finish. If upto is invalid, we wait until
! * at least one slot is available for insertion.
! *
! * Returns a value >= upto, which indicates the oldest in-progress insertion
! * that we saw in the array (or if there are non in-progress, the next insert
! * position).
! */
! static XLogRecPtr
! WaitXLogInsertionsToFinish(XLogRecPtr upto)
! {
! volatile XLogCtlInsert *Insert = &XLogCtl->Insert;
! int lastslot;
! int nextslot;
! XLogRecPtr LastPos;
! int extraWaits = 0;
!
! if (MyProc == NULL)
! elog(PANIC, "cannot wait without a PGPROC structure");
!
! retry:
! /* Only allow one backend to advance lastslot at a time */
! LWLockAcquire(WALInsertTailLock, LW_EXCLUSIVE);
!
! /*
! * Read lastslot, nextslot, and current end of reserved XLOG.
! */
! SpinLockAcquire(&Insert->insertpos_lck);
! LastPos = Insert->CurrPos;
! lastslot = Insert->lastslot;
! nextslot = Insert->nextslot;
! SpinLockRelease(&Insert->insertpos_lck);
!
! while (lastslot != nextslot)
! {
! /*
! * Examine the oldest slot still in use.
! */
! volatile XLogInsertSlot *slot;
! XLogRecPtr slotptr;
!
! slot = &XLogCtl->XLogInsertSlots[SlotNoToIdx(lastslot)];
!
! /* First, a quick check without the lock. */
! if (slot->finished)
! {
! lastslot = NextSlotNo(lastslot);
! /* Reinitialize the slot for reuse */
! slot->finished = false;
! continue;
! }
!
! SpinLockAcquire(&slot->lck);
! slotptr = slot->CurrPos;
!
! if (slot->finished)
! {
! /*
! * The insertion has already finished, we just need to advance
! * lastslot to make the slot available for reuse.
! */
! /* Reinitialize the slot for reuse */
! slot->finished = false;
! SpinLockRelease(&slot->lck);
! lastslot = NextSlotNo(lastslot);
! continue;
! }
! else
! {
! /*
! * The insertion is still in-progress. If we just needed for
! * any slot to become available and there is at least one slot
! * free now, or if this slot's CurrPos >= upto, we can
! * stop here. Otherwise we have to wait for it to finish.
! */
! if ((XLogRecPtrIsInvalid(upto) && NextSlotNo(nextslot) != lastslot)
! || (!XLogRecPtrIsInvalid(upto) && XLByteLE(upto, slotptr)))
! {
! SpinLockRelease(&slot->lck);
! LastPos = slotptr;
! break;
! }
! else
! {
! /* Wait for this insertion to finish. */
! MyProc->lwWaiting = true;
! MyProc->lwWaitMode = 0; /* doesn't matter */
! MyProc->lwWaitLink = NULL;
! if (slot->head == NULL)
! slot->head = MyProc;
! else
! slot->tail->lwWaitLink = MyProc;
! slot->tail = MyProc;
! SpinLockRelease(&slot->lck);
!
! Insert->lastslot = lastslot;
! LWLockRelease(WALInsertTailLock);
! for (;;)
! {
! PGSemaphoreLock(&MyProc->sem, false);
! if (!MyProc->lwWaiting)
! break;
! extraWaits++;
! }
!
! /*
! * The insertion has now finished. Start all over. While we
! * were not holding the tail-lock, someone might've filled up
! * all slots again.
! */
! goto retry;
! }
! }
! }
!
! /*
! * Update lastslot before we release the lock. (We don't need to grab
! * insertpos_lck here, on the assumption that writing an int32 is atomic)
! */
! Insert->lastslot = lastslot;
! LWLockRelease(WALInsertTailLock);
!
! while (extraWaits-- > 0)
! PGSemaphoreUnlock(&MyProc->sem);
!
! return LastPos;
}
/*
***************
*** 1440,1469 **** 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.
*/
! static bool
! AdvanceXLInsertBuffer(bool new_segment)
{
XLogCtlInsert *Insert = &XLogCtl->Insert;
! int nextidx = NextBufIdx(Insert->curridx);
! bool update_needed = true;
XLogRecPtr OldPageRqstPtr;
XLogwrtRqst WriteRqst;
! XLogRecPtr NewPageEndPtr;
XLogPageHeader NewPage;
/*
* Get ending-offset of the buffer page we need to replace (this may be
--- 2175,2207 ----
}
/*
! * 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 void
! AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic)
{
XLogCtlInsert *Insert = &XLogCtl->Insert;
! int nextidx;
XLogRecPtr OldPageRqstPtr;
XLogwrtRqst WriteRqst;
! XLogRecPtr NewPageEndPtr = InvalidXLogRecPtr;
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
***************
*** 1473,1482 **** 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];
/* Before waiting, get info_lck and update LogwrtResult */
{
--- 2211,2222 ----
OldPageRqstPtr = XLogCtl->xlblocks[nextidx];
if (!XLByteLE(OldPageRqstPtr, LogwrtResult.Write))
{
! /*
! * 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 */
{
***************
*** 1484,1504 **** AdvanceXLInsertBuffer(bool new_segment)
volatile XLogCtlData *xlogctl = XLogCtl;
SpinLockAcquire(&xlogctl->info_lck);
! if (XLByteLT(xlogctl->LogwrtRqst.Write, FinishedPageRqstPtr))
! xlogctl->LogwrtRqst.Write = FinishedPageRqstPtr;
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 */
LWLockAcquire(WALWriteLock, LW_EXCLUSIVE);
LogwrtResult = XLogCtl->LogwrtResult;
if (XLByteLE(OldPageRqstPtr, LogwrtResult.Write))
--- 2224,2250 ----
volatile XLogCtlData *xlogctl = XLogCtl;
SpinLockAcquire(&xlogctl->info_lck);
! if (XLByteLT(xlogctl->LogwrtRqst.Write, OldPageRqstPtr))
! xlogctl->LogwrtRqst.Write = OldPageRqstPtr;
LogwrtResult = xlogctl->LogwrtResult;
SpinLockRelease(&xlogctl->info_lck);
}
/*
* Now that we have an up-to-date LogwrtResult value, see if we still
* need to write it or if someone else already did.
*/
if (!XLByteLE(OldPageRqstPtr, LogwrtResult.Write))
{
! /*
! * 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))
***************
*** 1508,1525 **** 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.
! */
TRACE_POSTGRESQL_WAL_BUFFER_WRITE_DIRTY_START();
WriteRqst.Write = OldPageRqstPtr;
WriteRqst.Flush.xlogid = 0;
WriteRqst.Flush.xrecoff = 0;
! XLogWrite(WriteRqst, false, false);
LWLockRelease(WALWriteLock);
TRACE_POSTGRESQL_WAL_BUFFER_WRITE_DIRTY_DONE();
}
}
}
--- 2254,2271 ----
}
else
{
! /* 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);
LWLockRelease(WALWriteLock);
TRACE_POSTGRESQL_WAL_BUFFER_WRITE_DIRTY_DONE();
}
+ /* Re-acquire WALBufMappingLock and retry */
+ LWLockAcquire(WALBufMappingLock, LW_EXCLUSIVE);
+ continue;
}
}
***************
*** 1527,1540 **** AdvanceXLInsertBuffer(bool new_segment)
* Now the next buffer slot is free and we can set it up to be the next
* output page.
*/
! NewPageEndPtr = XLogCtl->xlblocks[Insert->curridx];
!
! if (new_segment)
! {
! /* force it to a segment start point */
! NewPageEndPtr.xrecoff += XLogSegSize - 1;
! NewPageEndPtr.xrecoff -= NewPageEndPtr.xrecoff % XLogSegSize;
! }
if (NewPageEndPtr.xrecoff >= XLogFileSize)
{
--- 2273,2279 ----
* Now the next buffer slot is free and we can set it up to be the next
* output page.
*/
! NewPageEndPtr = XLogCtl->xlblocks[XLogCtl->curridx];
if (NewPageEndPtr.xrecoff >= XLogFileSize)
{
***************
*** 1544,1556 **** AdvanceXLInsertBuffer(bool new_segment)
}
else
NewPageEndPtr.xrecoff += XLOG_BLCKSZ;
! XLogCtl->xlblocks[nextidx] = NewPageEndPtr;
! NewPage = (XLogPageHeader) (XLogCtl->pages + nextidx * (Size) XLOG_BLCKSZ);
!
! Insert->curridx = nextidx;
! Insert->currpage = NewPage;
! Insert->currpos = ((char *) NewPage) +SizeOfXLogShortPHD;
/*
* Be sure to re-zero the buffer so that bytes beyond what we've written
--- 2283,2292 ----
}
else
NewPageEndPtr.xrecoff += XLOG_BLCKSZ;
! Assert(NewPageEndPtr.xrecoff % XLOG_BLCKSZ == 0);
! Assert(XLogRecEndPtrToBufIdx(NewPageEndPtr) == nextidx);
! NewPage = (XLogPageHeader) (XLogCtl->pages + nextidx * (Size) XLOG_BLCKSZ);
/*
* Be sure to re-zero the buffer so that bytes beyond what we've written
***************
*** 1594,1604 **** 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;
}
/*
--- 2330,2357 ----
NewLongPage->xlp_seg_size = XLogSegSize;
NewLongPage->xlp_xlog_blcksz = XLOG_BLCKSZ;
NewPage ->xlp_info |= XLP_LONG_HEADER;
}
! /*
! * 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
}
/*
***************
*** 1643,1658 **** XLogCheckpointNeeded(uint32 logid, uint32 logseg)
* 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.
*/
static void
! XLogWrite(XLogwrtRqst WriteRqst, bool flexible, bool xlog_switch)
{
XLogCtlWrite *Write = &XLogCtl->Write;
bool ispartialpage;
--- 2396,2407 ----
* This option allows us to avoid uselessly issuing multiple writes when a
* single one would do.
*
! * 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)
{
XLogCtlWrite *Write = &XLogCtl->Write;
bool ispartialpage;
***************
*** 1701,1714 **** 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]))
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);
/* Advance LogwrtResult.Write to end of current buffer page */
! LogwrtResult.Write = XLogCtl->xlblocks[curridx];
ispartialpage = XLByteLT(WriteRqst.Write, LogwrtResult.Write);
if (!XLByteInPrevSeg(LogwrtResult.Write, openLogId, openLogSeg))
--- 2450,2463 ----
* if we're passed a bogus WriteRqst.Write that is past the end of the
* last page that's been initialized by AdvanceXLInsertBuffer.
*/
! 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,
! EndPtr.xlogid, EndPtr.xrecoff);
/* Advance LogwrtResult.Write to end of current buffer page */
! LogwrtResult.Write = EndPtr;
ispartialpage = XLByteLT(WriteRqst.Write, LogwrtResult.Write);
if (!XLByteInPrevSeg(LogwrtResult.Write, openLogId, openLogSeg))
***************
*** 1805,1820 **** 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))
{
issue_xlog_fsync(openLogFile, openLogId, openLogSeg);
LogwrtResult.Flush = LogwrtResult.Write; /* end of page */
--- 2554,2566 ----
* later. Doing it here ensures that one and only one backend will
* perform this fsync.
*
* 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)
{
issue_xlog_fsync(openLogFile, openLogId, openLogSeg);
LogwrtResult.Flush = LogwrtResult.Write; /* end of page */
***************
*** 2068,2073 **** XLogFlush(XLogRecPtr record)
--- 2814,2820 ----
{
/* use volatile pointer to prevent code rearrangement */
volatile XLogCtlData *xlogctl = XLogCtl;
+ XLogRecPtr insertpos;
/* read LogwrtResult and update local state */
SpinLockAcquire(&xlogctl->info_lck);
***************
*** 2081,2086 **** XLogFlush(XLogRecPtr record)
--- 2828,2839 ----
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
***************
*** 2100,2128 **** 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);
! if (freespace < SizeOfXLogRecord) /* buffer is full */
! WriteRqstPtr = XLogCtl->xlblocks[Insert->curridx];
! else
! {
! WriteRqstPtr = XLogCtl->xlblocks[Insert->curridx];
! WriteRqstPtr.xrecoff -= freespace;
! }
! LWLockRelease(WALInsertLock);
! WriteRqst.Write = WriteRqstPtr;
! WriteRqst.Flush = WriteRqstPtr;
! }
! else
! {
! WriteRqst.Write = WriteRqstPtr;
! WriteRqst.Flush = record;
! }
! XLogWrite(WriteRqst, false, false);
}
LWLockRelease(WALWriteLock);
/* done */
--- 2853,2862 ----
LogwrtResult = XLogCtl->LogwrtResult;
if (!XLByteLE(record, LogwrtResult.Flush))
{
! WriteRqst.Write = insertpos;
! WriteRqst.Flush = insertpos;
! XLogWrite(WriteRqst, false);
}
LWLockRelease(WALWriteLock);
/* done */
***************
*** 2237,2243 **** XLogBackgroundFlush(void)
START_CRIT_SECTION();
! /* now wait for the write lock */
LWLockAcquire(WALWriteLock, LW_EXCLUSIVE);
LogwrtResult = XLogCtl->LogwrtResult;
if (!XLByteLE(WriteRqstPtr, LogwrtResult.Flush))
--- 2971,2978 ----
START_CRIT_SECTION();
! /* 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))
***************
*** 2246,2256 **** XLogBackgroundFlush(void)
WriteRqst.Write = WriteRqstPtr;
WriteRqst.Flush = WriteRqstPtr;
! XLogWrite(WriteRqst, flexible, false);
}
- LWLockRelease(WALWriteLock);
END_CRIT_SECTION();
}
/*
--- 2981,2998 ----
WriteRqst.Write = WriteRqstPtr;
WriteRqst.Flush = WriteRqstPtr;
! XLogWrite(WriteRqst, flexible);
}
END_CRIT_SECTION();
+
+ LWLockRelease(WALWriteLock);
+
+ /*
+ * 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);
}
/*
***************
*** 5037,5042 **** XLOGShmemSize(void)
--- 5779,5791 ----
/* XLogCtl */
size = sizeof(XLogCtlData);
+
+ /*
+ * We force XLogCtl to be suitably aligned to put insertpos_lck and
+ * the fields it protects on the same cache line
+ */
+ size += XLOGCTLINSERT_ALIGNMENT;
+
/* xlblocks array */
size = add_size(size, mul_size(sizeof(XLogRecPtr), XLOGbuffers));
/* extra alignment padding for XLOG I/O buffers */
***************
*** 5059,5069 **** XLOGShmemInit(void)
bool foundCFile,
foundXLog;
char *allocptr;
ControlFile = (ControlFileData *)
ShmemInitStruct("Control File", sizeof(ControlFileData), &foundCFile);
! XLogCtl = (XLogCtlData *)
! ShmemInitStruct("XLOG Ctl", XLOGShmemSize(), &foundXLog);
if (foundCFile || foundXLog)
{
--- 5808,5818 ----
bool foundCFile,
foundXLog;
char *allocptr;
+ int i;
ControlFile = (ControlFileData *)
ShmemInitStruct("Control File", sizeof(ControlFileData), &foundCFile);
! allocptr = ShmemInitStruct("XLOG Ctl", XLOGShmemSize(), &foundXLog);
if (foundCFile || foundXLog)
{
***************
*** 5072,5077 **** XLOGShmemInit(void)
--- 5821,5833 ----
return;
}
+ /*
+ * Ensure desired alignment for XLogCtlInsert, so that insertpos_lck and
+ * the fields it protects fall on the same cache line.
+ */
+ allocptr += XLOGCTLINSERT_ALIGNMENT - ((uintptr_t) allocptr) % XLOGCTLINSERT_ALIGNMENT;
+ XLogCtl = (XLogCtlData *) allocptr;
+
memset(XLogCtl, 0, sizeof(XLogCtlData));
/*
***************
*** 5084,5089 **** XLOGShmemInit(void)
--- 5840,5857 ----
memset(XLogCtl->xlblocks, 0, sizeof(XLogRecPtr) * XLOGbuffers);
allocptr += sizeof(XLogRecPtr) * XLOGbuffers;
+ /* Initialize insertion slots */
+ for (i = 0; i < NumXLogInsertSlots; i++)
+ {
+ XLogInsertSlot *slot = &XLogCtl->XLogInsertSlots[i];
+ slot->CurrPos = InvalidXLogRecPtr;
+ slot->finished = false;
+ slot->head = slot->tail = NULL;
+ SpinLockInit(&slot->lck);
+ }
+ XLogCtl->Insert.nextslot = 0;
+ XLogCtl->Insert.lastslot = 0;
+
/*
* Align the start of the page buffers to an ALIGNOF_XLOG_BUFFER boundary.
*/
***************
*** 5098,5104 **** XLOGShmemInit(void)
XLogCtl->XLogCacheBlck = XLOGbuffers - 1;
XLogCtl->SharedRecoveryInProgress = true;
XLogCtl->SharedHotStandbyActive = false;
! XLogCtl->Insert.currpage = (XLogPageHeader) (XLogCtl->pages);
SpinLockInit(&XLogCtl->info_lck);
InitSharedLatch(&XLogCtl->recoveryWakeupLatch);
InitSharedLatch(&XLogCtl->WALWriterLatch);
--- 5866,5872 ----
XLogCtl->XLogCacheBlck = XLOGbuffers - 1;
XLogCtl->SharedRecoveryInProgress = true;
XLogCtl->SharedHotStandbyActive = false;
! SpinLockInit(&XLogCtl->Insert.insertpos_lck);
SpinLockInit(&XLogCtl->info_lck);
InitSharedLatch(&XLogCtl->recoveryWakeupLatch);
InitSharedLatch(&XLogCtl->WALWriterLatch);
***************
*** 5980,5985 **** StartupXLOG(void)
--- 6748,6754 ----
bool backupEndRequired = false;
bool backupFromStandby = false;
DBState dbstate_at_startup;
+ int firstIdx;
/*
* Read control file and check XLOG status looks valid.
***************
*** 6232,6238 **** StartupXLOG(void)
lastFullPageWrites = checkPoint.fullPageWrites;
! RedoRecPtr = XLogCtl->Insert.RedoRecPtr = checkPoint.redo;
if (XLByteLT(RecPtr, checkPoint.redo))
ereport(PANIC,
--- 7001,7007 ----
lastFullPageWrites = checkPoint.fullPageWrites;
! RedoRecPtr = XLogCtl->RedoRecPtr = XLogCtl->Insert.RedoRecPtr = checkPoint.redo;
if (XLByteLT(RecPtr, checkPoint.redo))
ereport(PANIC,
***************
*** 6786,6793 **** StartupXLOG(void)
openLogOff = 0;
Insert = &XLogCtl->Insert;
Insert->PrevRecord = LastRec;
! XLogCtl->xlblocks[0].xlogid = openLogId;
! XLogCtl->xlblocks[0].xrecoff =
((EndOfLog.xrecoff - 1) / XLOG_BLCKSZ + 1) * XLOG_BLCKSZ;
/*
--- 7555,7566 ----
openLogOff = 0;
Insert = &XLogCtl->Insert;
Insert->PrevRecord = LastRec;
!
! firstIdx = XLogRecEndPtrToBufIdx(EndOfLog);
! XLogCtl->curridx = firstIdx;
!
! XLogCtl->xlblocks[firstIdx].xlogid = openLogId;
! XLogCtl->xlblocks[firstIdx].xrecoff =
((EndOfLog.xrecoff - 1) / XLOG_BLCKSZ + 1) * XLOG_BLCKSZ;
/*
***************
*** 6795,6804 **** 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);
LogwrtResult.Write = LogwrtResult.Flush = EndOfLog;
--- 7568,7576 ----
* record spans, not the one it starts in. The last block is indeed the
* one we want to use.
*/
! Assert(readOff == (XLogCtl->xlblocks[firstIdx].xrecoff - XLOG_BLCKSZ) % XLogSegSize);
! memcpy((char *) &XLogCtl->pages[firstIdx * XLOG_BLCKSZ], readBuf, XLOG_BLCKSZ);
! Insert->CurrPos = EndOfLog;
LogwrtResult.Write = LogwrtResult.Flush = EndOfLog;
***************
*** 6807,6818 **** StartupXLOG(void)
XLogCtl->LogwrtRqst.Write = EndOfLog;
XLogCtl->LogwrtRqst.Flush = EndOfLog;
! freespace = INSERT_FREESPACE(Insert);
if (freespace > 0)
{
/* Make sure rest of page is zero */
! MemSet(Insert->currpos, 0, freespace);
! XLogCtl->Write.curridx = 0;
}
else
{
--- 7579,7590 ----
XLogCtl->LogwrtRqst.Write = EndOfLog;
XLogCtl->LogwrtRqst.Flush = EndOfLog;
! freespace = INSERT_FREESPACE(EndOfLog);
if (freespace > 0)
{
/* Make sure rest of page is zero */
! MemSet(&XLogCtl->pages[firstIdx * XLOG_BLCKSZ] + EndOfLog.xrecoff % XLOG_BLCKSZ, 0, freespace);
! XLogCtl->Write.curridx = firstIdx;
}
else
{
***************
*** 6824,6830 **** StartupXLOG(void)
* this is sufficient. The first actual attempt to insert a log
* record will advance the insert state.
*/
! XLogCtl->Write.curridx = NextBufIdx(0);
}
/* Pre-scan prepared transactions to find out the range of XIDs present */
--- 7596,7602 ----
* this is sufficient. The first actual attempt to insert a log
* record will advance the insert state.
*/
! XLogCtl->Write.curridx = NextBufIdx(firstIdx);
}
/* Pre-scan prepared transactions to find out the range of XIDs present */
***************
*** 6835,6841 **** StartupXLOG(void)
* XLOG_FPW_CHANGE record before resource manager writes cleanup
* WAL records or checkpoint record is written.
*/
! Insert->fullPageWrites = lastFullPageWrites;
LocalSetXLogInsertAllowed();
UpdateFullPageWrites();
LocalXLogInsertAllowed = -1;
--- 7607,7613 ----
* XLOG_FPW_CHANGE record before resource manager writes cleanup
* WAL records or checkpoint record is written.
*/
! Insert->fullPageWrites = doPageWrites = lastFullPageWrites;
LocalSetXLogInsertAllowed();
UpdateFullPageWrites();
LocalXLogInsertAllowed = -1;
***************
*** 7307,7327 **** 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().
*/
XLogRecPtr
GetRedoRecPtr(void)
{
/* use volatile pointer to prevent code rearrangement */
volatile XLogCtlData *xlogctl = XLogCtl;
SpinLockAcquire(&xlogctl->info_lck);
! Assert(XLByteLE(RedoRecPtr, xlogctl->Insert.RedoRecPtr));
! RedoRecPtr = xlogctl->Insert.RedoRecPtr;
SpinLockRelease(&xlogctl->info_lck);
return RedoRecPtr;
}
--- 8079,8107 ----
}
/*
! * 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 insertpos_lck to read the master copy, someone might update
+ * it just after we've released the lock.
+ */
SpinLockAcquire(&xlogctl->info_lck);
! ptr = xlogctl->RedoRecPtr;
SpinLockRelease(&xlogctl->info_lck);
+ if (XLByteLT(RedoRecPtr, ptr))
+ RedoRecPtr = xlogctl->RedoRecPtr;
+
return RedoRecPtr;
}
***************
*** 7330,7336 **** 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
* heavily contended, and an approximation is enough for the current
* usage of this function.
*/
--- 8110,8116 ----
*
* 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 insertpos_lck which can be quite
* heavily contended, and an approximation is enough for the current
* usage of this function.
*/
***************
*** 7592,7597 **** LogCheckpointEnd(bool restartpoint)
--- 8372,8379 ----
void
CreateCheckPoint(int flags)
{
+ /* use volatile pointer to prevent code rearrangement */
+ volatile XLogCtlData *xlogctl = XLogCtl;
bool shutdown;
CheckPoint checkPoint;
XLogRecPtr recptr;
***************
*** 7606,7611 **** CreateCheckPoint(int flags)
--- 8388,8394 ----
uint32 insert_logSeg;
TransactionId *inCommitXids;
int nInCommit;
+ XLogRecPtr curInsert;
/*
* An end-of-recovery checkpoint is really a shutdown checkpoint, just
***************
*** 7674,7683 **** CreateCheckPoint(int flags)
checkPoint.oldestActiveXid = InvalidTransactionId;
/*
! * We must hold WALInsertLock while examining insert state to determine
* the checkpoint REDO pointer.
*/
! LWLockAcquire(WALInsertLock, LW_EXCLUSIVE);
/*
* If this isn't a shutdown or forced checkpoint, and we have not switched
--- 8457,8467 ----
checkPoint.oldestActiveXid = InvalidTransactionId;
/*
! * We must hold insertpos_lck while examining insert state to determine
* the checkpoint REDO pointer.
*/
! SpinLockAcquire(&Insert->insertpos_lck);
! curInsert = Insert->CurrPos;
/*
* If this isn't a shutdown or forced checkpoint, and we have not switched
***************
*** 7689,7695 **** 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
* 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
--- 8473,8479 ----
* (Perhaps it'd make even more sense to checkpoint only when the previous
* checkpoint record is in a different xlog page?)
*
! * 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
***************
*** 7698,7712 **** 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_logId, insert_logSeg);
XLByteToSeg(ControlFile->checkPointCopy.redo, redo_logId, redo_logSeg);
if (insert_logId == redo_logId &&
insert_logSeg == redo_logSeg)
{
! LWLockRelease(WALInsertLock);
LWLockRelease(CheckpointLock);
END_CRIT_SECTION();
return;
--- 8482,8493 ----
if ((flags & (CHECKPOINT_IS_SHUTDOWN | CHECKPOINT_END_OF_RECOVERY |
CHECKPOINT_FORCE)) == 0)
{
XLByteToSeg(curInsert, insert_logId, insert_logSeg);
XLByteToSeg(ControlFile->checkPointCopy.redo, redo_logId, redo_logSeg);
if (insert_logId == redo_logId &&
insert_logSeg == redo_logSeg)
{
! SpinLockRelease(&Insert->insertpos_lck);
LWLockRelease(CheckpointLock);
END_CRIT_SECTION();
return;
***************
*** 7733,7750 **** 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);
if (freespace < SizeOfXLogRecord)
{
! (void) AdvanceXLInsertBuffer(false);
! /* OK to ignore update return flag, since we will do flush anyway */
! freespace = INSERT_FREESPACE(Insert);
}
! INSERT_RECPTR(checkPoint.redo, Insert, Insert->curridx);
/*
* Here we update the shared RedoRecPtr for future XLogInsert calls; this
! * must be done while holding the insert lock AND the info_lck.
*
* 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
--- 8514,8533 ----
* the buffer flush work. Those XLOG records are logically after the
* checkpoint, even though physically before it. Got that?
*/
! freespace = INSERT_FREESPACE(curInsert);
if (freespace < SizeOfXLogRecord)
{
! XLByteAdvance(curInsert, freespace);
! if (curInsert.xrecoff % XLogSegSize == 0)
! curInsert.xrecoff += SizeOfXLogLongPHD;
! else
! curInsert.xrecoff += SizeOfXLogShortPHD;
}
! checkPoint.redo = curInsert;
/*
* Here we update the shared RedoRecPtr for future XLogInsert calls; this
! * 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
***************
*** 7753,7772 **** 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);
! }
/*
* Now we can release WAL insert lock, allowing other xacts to proceed
* while we are flushing disk buffers.
*/
! LWLockRelease(WALInsertLock);
/*
* If enabled, log checkpoint start. We postpone this until now so as not
--- 8536,8553 ----
* XLogInserts that happen while we are dumping buffers must assume that
* their buffer changes are not included in the checkpoint.
*/
! RedoRecPtr = xlogctl->Insert.RedoRecPtr = checkPoint.redo;
/*
* Now we can release WAL insert lock, allowing other xacts to proceed
* while we are flushing disk buffers.
*/
! SpinLockRelease(&Insert->insertpos_lck);
!
! /* 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
***************
*** 7786,7792 **** CreateCheckPoint(int flags)
* we wait till he's out of his commit critical section before proceeding.
* See notes in RecordTransactionCommit().
*
! * Because we've already released WALInsertLock, this test is a bit fuzzy:
* it is possible that we will wait for xacts we didn't really need to
* wait for. But the delay should be short and it seems better to make
* checkpoint take a bit longer than to hold locks longer than necessary.
--- 8567,8573 ----
* we wait till he's out of his commit critical section before proceeding.
* See notes in RecordTransactionCommit().
*
! * Because we've already released insertpos_lck, this test is a bit fuzzy:
* it is possible that we will wait for xacts we didn't really need to
* wait for. But the delay should be short and it seems better to make
* checkpoint take a bit longer than to hold locks longer than necessary.
***************
*** 8153,8167 **** 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.
*/
! LWLockAcquire(WALInsertLock, LW_EXCLUSIVE);
! SpinLockAcquire(&xlogctl->info_lck);
xlogctl->Insert.RedoRecPtr = lastCheckPoint.redo;
SpinLockRelease(&xlogctl->info_lck);
- LWLockRelease(WALInsertLock);
/*
* Prepare to accumulate statistics.
--- 8934,8951 ----
* the number of segments replayed since last restartpoint, and request a
* restartpoint if it exceeds checkpoint_segments.
*
! * Like in CreatecheckPoint(), hold insertpos_lck to update it, although
! * during recovery acquiring insertpos_lck is just pro forma, because no
! * WAL insertions are happening.
*/
! SpinLockAcquire(&xlogctl->Insert.insertpos_lck);
xlogctl->Insert.RedoRecPtr = lastCheckPoint.redo;
+ SpinLockRelease(&xlogctl->Insert.insertpos_lck);
+
+ /* Also update the info_lck-protected copy */
+ SpinLockAcquire(&xlogctl->info_lck);
+ xlogctl->RedoRecPtr = lastCheckPoint.redo;
SpinLockRelease(&xlogctl->info_lck);
/*
* Prepare to accumulate statistics.
***************
*** 8448,8454 **** XLogReportParameters(void)
void
UpdateFullPageWrites(void)
{
! XLogCtlInsert *Insert = &XLogCtl->Insert;
/*
* Do nothing if full_page_writes has not been changed.
--- 9232,9238 ----
void
UpdateFullPageWrites(void)
{
! volatile XLogCtlInsert *Insert = &XLogCtl->Insert;
/*
* Do nothing if full_page_writes has not been changed.
***************
*** 8471,8479 **** UpdateFullPageWrites(void)
*/
if (fullPageWrites)
{
! LWLockAcquire(WALInsertLock, LW_EXCLUSIVE);
Insert->fullPageWrites = true;
! LWLockRelease(WALInsertLock);
}
/*
--- 9255,9263 ----
*/
if (fullPageWrites)
{
! SpinLockAcquire(&Insert->insertpos_lck);
Insert->fullPageWrites = true;
! SpinLockRelease(&Insert->insertpos_lck);
}
/*
***************
*** 8494,8502 **** UpdateFullPageWrites(void)
if (!fullPageWrites)
{
! LWLockAcquire(WALInsertLock, LW_EXCLUSIVE);
Insert->fullPageWrites = false;
! LWLockRelease(WALInsertLock);
}
END_CRIT_SECTION();
}
--- 9278,9286 ----
if (!fullPageWrites)
{
! SpinLockAcquire(&Insert->insertpos_lck);
Insert->fullPageWrites = false;
! SpinLockRelease(&Insert->insertpos_lck);
}
END_CRIT_SECTION();
}
***************
*** 9024,9029 **** issue_xlog_fsync(int fd, uint32 log, uint32 seg)
--- 9808,9814 ----
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;
***************
*** 9086,9111 **** 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().
*/
! LWLockAcquire(WALInsertLock, LW_EXCLUSIVE);
if (exclusive)
{
! if (XLogCtl->Insert.exclusiveBackup)
{
! LWLockRelease(WALInsertLock);
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;
}
else
! XLogCtl->Insert.nonExclusiveBackups++;
! XLogCtl->Insert.forcePageWrites = true;
! LWLockRelease(WALInsertLock);
/* Ensure we release forcePageWrites if fail below */
PG_ENSURE_ERROR_CLEANUP(pg_start_backup_callback, (Datum) BoolGetDatum(exclusive));
--- 9871,9896 ----
* Note that forcePageWrites has no effect during an online backup from
* the standby.
*
! * We must hold insertpos_lck to change the value of forcePageWrites, to
* ensure adequate interlocking against XLogInsert().
*/
! SpinLockAcquire(&Insert->insertpos_lck);
if (exclusive)
{
! if (Insert->exclusiveBackup)
{
! SpinLockRelease(&Insert->insertpos_lck);
ereport(ERROR,
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
errmsg("a backup is already in progress"),
errhint("Run pg_stop_backup() and try again.")));
}
! Insert->exclusiveBackup = true;
}
else
! Insert->nonExclusiveBackups++;
! Insert->forcePageWrites = true;
! SpinLockRelease(&Insert->insertpos_lck);
/* Ensure we release forcePageWrites if fail below */
PG_ENSURE_ERROR_CLEANUP(pg_start_backup_callback, (Datum) BoolGetDatum(exclusive));
***************
*** 9218,9230 **** 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))
{
! XLogCtl->Insert.lastBackupStart = startpoint;
gotUniqueStartpoint = true;
}
! LWLockRelease(WALInsertLock);
} while (!gotUniqueStartpoint);
XLByteToSeg(startpoint, _logId, _logSeg);
--- 10003,10015 ----
* taking a checkpoint right after another is not that expensive
* either because only few buffers have been dirtied yet.
*/
! SpinLockAcquire(&Insert->insertpos_lck);
! if (XLByteLT(Insert->lastBackupStart, startpoint))
{
! Insert->lastBackupStart = startpoint;
gotUniqueStartpoint = true;
}
! SpinLockRelease(&Insert->insertpos_lck);
} while (!gotUniqueStartpoint);
XLByteToSeg(startpoint, _logId, _logSeg);
***************
*** 9308,9334 **** do_pg_start_backup(const char *backupidstr, bool fast, char **labelfile)
static void
pg_start_backup_callback(int code, Datum arg)
{
bool exclusive = DatumGetBool(arg);
/* Update backup counters and forcePageWrites on failure */
! LWLockAcquire(WALInsertLock, LW_EXCLUSIVE);
if (exclusive)
{
! Assert(XLogCtl->Insert.exclusiveBackup);
! XLogCtl->Insert.exclusiveBackup = false;
}
else
{
! Assert(XLogCtl->Insert.nonExclusiveBackups > 0);
! XLogCtl->Insert.nonExclusiveBackups--;
}
! if (!XLogCtl->Insert.exclusiveBackup &&
! XLogCtl->Insert.nonExclusiveBackups == 0)
{
! XLogCtl->Insert.forcePageWrites = false;
}
! LWLockRelease(WALInsertLock);
}
/*
--- 10093,10120 ----
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 */
! SpinLockAcquire(&Insert->insertpos_lck);
if (exclusive)
{
! Assert(Insert->exclusiveBackup);
! Insert->exclusiveBackup = false;
}
else
{
! Assert(Insert->nonExclusiveBackups > 0);
! Insert->nonExclusiveBackups--;
}
! if (!Insert->exclusiveBackup &&
! Insert->nonExclusiveBackups == 0)
{
! Insert->forcePageWrites = false;
}
! SpinLockRelease(&Insert->insertpos_lck);
}
/*
***************
*** 9341,9346 **** pg_start_backup_callback(int code, Datum arg)
--- 10127,10133 ----
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;
***************
*** 9394,9402 **** do_pg_stop_backup(char *labelfile, bool waitforarchive)
/*
* OK to update backup counters and forcePageWrites
*/
! LWLockAcquire(WALInsertLock, LW_EXCLUSIVE);
if (exclusive)
! XLogCtl->Insert.exclusiveBackup = false;
else
{
/*
--- 10181,10189 ----
/*
* OK to update backup counters and forcePageWrites
*/
! SpinLockAcquire(&Insert->insertpos_lck);
if (exclusive)
! Insert->exclusiveBackup = false;
else
{
/*
***************
*** 9405,9420 **** 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--;
}
! if (!XLogCtl->Insert.exclusiveBackup &&
! XLogCtl->Insert.nonExclusiveBackups == 0)
{
! XLogCtl->Insert.forcePageWrites = false;
}
! LWLockRelease(WALInsertLock);
if (exclusive)
{
--- 10192,10207 ----
* backups, it is expected that each do_pg_start_backup() call is
* matched by exactly one do_pg_stop_backup() call.
*/
! Assert(Insert->nonExclusiveBackups > 0);
! Insert->nonExclusiveBackups--;
}
! if (!Insert->exclusiveBackup &&
! Insert->nonExclusiveBackups == 0)
{
! Insert->forcePageWrites = false;
}
! SpinLockRelease(&Insert->insertpos_lck);
if (exclusive)
{
***************
*** 9692,9707 **** 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--;
! if (!XLogCtl->Insert.exclusiveBackup &&
! XLogCtl->Insert.nonExclusiveBackups == 0)
{
! XLogCtl->Insert.forcePageWrites = false;
}
! LWLockRelease(WALInsertLock);
}
/*
--- 10479,10496 ----
void
do_pg_abort_backup(void)
{
! volatile XLogCtlInsert *Insert = &XLogCtl->Insert;
!
! SpinLockAcquire(&Insert->insertpos_lck);
! Assert(Insert->nonExclusiveBackups > 0);
! Insert->nonExclusiveBackups--;
! if (!Insert->exclusiveBackup &&
! Insert->nonExclusiveBackups == 0)
{
! Insert->forcePageWrites = false;
}
! SpinLockRelease(&Insert->insertpos_lck);
}
/*
***************
*** 9755,9766 **** GetStandbyFlushRecPtr(void)
XLogRecPtr
GetXLogInsertRecPtr(void)
{
! XLogCtlInsert *Insert = &XLogCtl->Insert;
XLogRecPtr current_recptr;
! LWLockAcquire(WALInsertLock, LW_SHARED);
! INSERT_RECPTR(current_recptr, Insert, Insert->curridx);
! LWLockRelease(WALInsertLock);
return current_recptr;
}
--- 10544,10555 ----
XLogRecPtr
GetXLogInsertRecPtr(void)
{
! volatile XLogCtlInsert *Insert = &XLogCtl->Insert;
XLogRecPtr current_recptr;
! SpinLockAcquire(&Insert->insertpos_lck);
! current_recptr = Insert->CurrPos;
! SpinLockRelease(&Insert->insertpos_lck);
return current_recptr;
}
*** a/src/backend/storage/ipc/procarray.c
--- b/src/backend/storage/ipc/procarray.c
***************
*** 1753,1761 **** 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.
*/
int
GetTransactionsInCommit(TransactionId **xids_p)
--- 1753,1762 ----
* 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 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)
*** a/src/backend/storage/lmgr/spin.c
--- b/src/backend/storage/lmgr/spin.c
***************
*** 56,61 **** SpinlockSemas(void)
--- 56,64 ----
*
* 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;
}
*** a/src/include/storage/lwlock.h
--- b/src/include/storage/lwlock.h
***************
*** 53,59 **** typedef enum LWLockId
ProcArrayLock,
SInvalReadLock,
SInvalWriteLock,
! WALInsertLock,
WALWriteLock,
ControlFileLock,
CheckpointLock,
--- 53,59 ----
ProcArrayLock,
SInvalReadLock,
SInvalWriteLock,
! WALBufMappingLock,
WALWriteLock,
ControlFileLock,
CheckpointLock,
***************
*** 79,84 **** typedef enum LWLockId
--- 79,85 ----
SerializablePredicateLockListLock,
OldSerXidLock,
SyncRepLock,
+ WALInsertTailLock,
/* Individual lock IDs end here */
FirstBufMappingLock,
FirstLockMgrLock = FirstBufMappingLock + NUM_BUFFER_PARTITIONS,