* looked up again. Now we use specialized access code so that the commit
* log can be broken into relatively small, independent segments.
*
- * Portions Copyright (c) 1996-2001, PostgreSQL Global Development Group
+ * XLOG interactions: this module generates an XLOG record whenever a new
+ * CLOG page is initialized to zeroes. Other writes of CLOG come from
+ * recording of transaction commit or abort in xact.c, which generates its
+ * own XLOG records for these events and will re-perform the status update
+ * on redo; so we need make no additional XLOG entry here. For synchronous
+ * transaction commits, the XLOG is guaranteed flushed through the XLOG commit
+ * record before we are called to log a commit, so the WAL rule "write xlog
+ * before data" is satisfied automatically. However, for async commits we
+ * must track the latest LSN affecting each CLOG page, so that we can flush
+ * XLOG that far and satisfy the WAL rule. We don't have to worry about this
+ * for aborts (whether sync or async), since the post-crash assumption would
+ * be that such transactions failed anyway.
+ *
+ * Portions Copyright (c) 1996-2019, PostgreSQL Global Development Group
* Portions Copyright (c) 1994, Regents of the University of California
*
- * $Header: /cvsroot/pgsql/src/backend/access/transam/clog.c,v 1.5 2001/10/25 05:49:22 momjian Exp $
+ * src/backend/access/transam/clog.c
*
*-------------------------------------------------------------------------
*/
#include "postgres.h"
-#include <fcntl.h>
-#include <dirent.h>
-#include <errno.h>
-#include <sys/stat.h>
-#include <sys/types.h>
-#include <unistd.h>
-
#include "access/clog.h"
-#include "storage/lwlock.h"
+#include "access/slru.h"
+#include "access/transam.h"
+#include "access/xlog.h"
+#include "access/xloginsert.h"
+#include "access/xlogutils.h"
#include "miscadmin.h"
-
+#include "pgstat.h"
+#include "pg_trace.h"
+#include "storage/proc.h"
/*
- * Defines for CLOG page and segment sizes. A page is the same BLCKSZ
- * as is used everywhere else in Postgres. The CLOG segment size can be
- * chosen somewhat arbitrarily; we make it 1 million transactions by default,
- * or 256Kb.
+ * Defines for CLOG page sizes. A page is the same BLCKSZ as is used
+ * everywhere else in Postgres.
*
* Note: because TransactionIds are 32 bits and wrap around at 0xFFFFFFFF,
* CLOG page numbering also wraps around at 0xFFFFFFFF/CLOG_XACTS_PER_PAGE,
- * and CLOG segment numbering at 0xFFFFFFFF/CLOG_XACTS_PER_SEGMENT. We need
- * take no explicit notice of that fact in this module, except when comparing
- * segment and page numbers in TruncateCLOG (see CLOGPagePrecedes).
+ * and CLOG segment numbering at
+ * 0xFFFFFFFF/CLOG_XACTS_PER_PAGE/SLRU_PAGES_PER_SEGMENT. We need take no
+ * explicit notice of that fact in this module, except when comparing segment
+ * and page numbers in TruncateCLOG (see CLOGPagePrecedes).
*/
-#define CLOG_BLCKSZ BLCKSZ
-
/* We need two bits per xact, so four xacts fit in a byte */
#define CLOG_BITS_PER_XACT 2
#define CLOG_XACTS_PER_BYTE 4
-#define CLOG_XACTS_PER_PAGE (CLOG_BLCKSZ * CLOG_XACTS_PER_BYTE)
+#define CLOG_XACTS_PER_PAGE (BLCKSZ * CLOG_XACTS_PER_BYTE)
#define CLOG_XACT_BITMASK ((1 << CLOG_BITS_PER_XACT) - 1)
-#define CLOG_XACTS_PER_SEGMENT 0x100000
-#define CLOG_PAGES_PER_SEGMENT (CLOG_XACTS_PER_SEGMENT / CLOG_XACTS_PER_PAGE)
-
#define TransactionIdToPage(xid) ((xid) / (TransactionId) CLOG_XACTS_PER_PAGE)
#define TransactionIdToPgIndex(xid) ((xid) % (TransactionId) CLOG_XACTS_PER_PAGE)
#define TransactionIdToByte(xid) (TransactionIdToPgIndex(xid) / CLOG_XACTS_PER_BYTE)
#define TransactionIdToBIndex(xid) ((xid) % (TransactionId) CLOG_XACTS_PER_BYTE)
+/* We store the latest async LSN for each group of transactions */
+#define CLOG_XACTS_PER_LSN_GROUP 32 /* keep this a power of 2 */
+#define CLOG_LSNS_PER_PAGE (CLOG_XACTS_PER_PAGE / CLOG_XACTS_PER_LSN_GROUP)
-/*----------
- * Shared-memory data structures for CLOG control
- *
- * We use a simple least-recently-used scheme to manage a pool of page
- * buffers for the CLOG. Under ordinary circumstances we expect that write
- * traffic will occur mostly to the latest CLOG page (and to the just-prior
- * page, soon after a page transition). Read traffic will probably touch
- * a larger span of pages, but in any case a fairly small number of page
- * buffers should be sufficient. So, we just search the buffers using plain
- * linear search; there's no need for a hashtable or anything fancy.
- * The management algorithm is straight LRU except that we will never swap
- * out the latest page (since we know it's going to be hit again eventually).
- *
- * We use an overall LWLock to protect the shared data structures, plus
- * per-buffer LWLocks that synchronize I/O for each buffer. A process
- * that is reading in or writing out a page buffer does not hold the control
- * lock, only the per-buffer lock for the buffer it is working on.
- *
- * To change the page number or state of a buffer, one must normally hold
- * the control lock. (The sole exception to this rule is that a writer
- * process changes the state from DIRTY to WRITE_IN_PROGRESS while holding
- * only the per-buffer lock.) If the buffer's state is neither EMPTY nor
- * CLEAN, then there may be processes doing (or waiting to do) I/O on the
- * buffer, so the page number may not be changed, and the only allowed state
- * transition is to change WRITE_IN_PROGRESS to DIRTY after dirtying the page.
- * To do any other state transition involving a buffer with potential I/O
- * processes, one must hold both the per-buffer lock and the control lock.
- * (Note the control lock must be acquired second; do not wait on a buffer
- * lock while holding the control lock.) A process wishing to read a page
- * marks the buffer state as READ_IN_PROGRESS, then drops the control lock,
- * acquires the per-buffer lock, and rechecks the state before proceeding.
- * This recheck takes care of the possibility that someone else already did
- * the read, while the early marking prevents someone else from trying to
- * read the same page into a different buffer.
- *
- * Note we are assuming that read and write of the state value is atomic,
- * since I/O processes may examine and change the state while not holding
- * the control lock.
- *
- * As with the regular buffer manager, it is possible for another process
- * to re-dirty a page that is currently being written out. This is handled
- * by setting the page's state from WRITE_IN_PROGRESS to DIRTY. The writing
- * process must notice this and not mark the page CLEAN when it's done.
- *
- * XLOG interactions: this module generates an XLOG record whenever a new
- * CLOG page is initialized to zeroes. Other writes of CLOG come from
- * recording of transaction commit or abort in xact.c, which generates its
- * own XLOG records for these events and will re-perform the status update
- * on redo; so we need make no additional XLOG entry here. Also, the XLOG
- * is guaranteed flushed through the XLOG commit record before we are called
- * to log a commit, so the WAL rule "write xlog before data" is satisfied
- * automatically for commits, and we don't really care for aborts. Therefore,
- * we don't need to mark XLOG pages with LSN information; we have enough
- * synchronization already.
- *----------
- */
-
-typedef enum
-{
- CLOG_PAGE_EMPTY,/* CLOG buffer is not in use */
- CLOG_PAGE_READ_IN_PROGRESS, /* CLOG page is being read
- * in */
- CLOG_PAGE_CLEAN,/* CLOG page is valid and not dirty */
- CLOG_PAGE_DIRTY,/* CLOG page is valid but needs write */
- CLOG_PAGE_WRITE_IN_PROGRESS /* CLOG page is being
- * written out in */
-} ClogPageStatus;
-
-/*
- * Shared-memory state for CLOG.
- */
-typedef struct ClogCtlData
-{
- /*
- * Info for each buffer slot. Page number is undefined when status is
- * EMPTY. lru_count is essentially the number of operations since
- * last use of this page; the page with highest lru_count is the best
- * candidate to replace.
- */
- char *page_buffer[NUM_CLOG_BUFFERS];
- ClogPageStatus page_status[NUM_CLOG_BUFFERS];
- int page_number[NUM_CLOG_BUFFERS];
- unsigned int page_lru_count[NUM_CLOG_BUFFERS];
-
- /*
- * latest_page_number is the page number of the current end of the
- * CLOG; this is not critical data, since we use it only to avoid
- * swapping out the latest page.
- */
- int latest_page_number;
-} ClogCtlData;
-
-static ClogCtlData *ClogCtl = NULL;
+#define GetLSNIndex(slotno, xid) ((slotno) * CLOG_LSNS_PER_PAGE + \
+ ((xid) % (TransactionId) CLOG_XACTS_PER_PAGE) / CLOG_XACTS_PER_LSN_GROUP)
/*
- * ClogBufferLocks is set during CLOGShmemInit and does not change thereafter.
- * The value is automatically inherited by backends via fork, and
- * doesn't need to be in shared memory.
+ * The number of subtransactions below which we consider to apply clog group
+ * update optimization. Testing reveals that the number higher than this can
+ * hurt performance.
*/
-static LWLockId ClogBufferLocks[NUM_CLOG_BUFFERS]; /* Per-buffer I/O locks */
+#define THRESHOLD_SUBTRANS_CLOG_OPT 5
/*
- * ClogDir is set during CLOGShmemInit and does not change thereafter.
- * The value is automatically inherited by backends via fork, and
- * doesn't need to be in shared memory.
+ * Link to shared-memory data structures for CLOG control
*/
-static char ClogDir[MAXPGPATH];
+static SlruCtlData ClogCtlData;
-#define ClogFileName(path, seg) \
- snprintf(path, MAXPGPATH, "%s/%04X", ClogDir, seg)
-
-/*
- * Macro to mark a buffer slot "most recently used".
- */
-#define ClogRecentlyUsed(slotno) \
- do { \
- int iilru; \
- for (iilru = 0; iilru < NUM_CLOG_BUFFERS; iilru++) \
- ClogCtl->page_lru_count[iilru]++; \
- ClogCtl->page_lru_count[slotno] = 0; \
- } while (0)
+#define ClogCtl (&ClogCtlData)
static int ZeroCLOGPage(int pageno, bool writeXlog);
-static int ReadCLOGPage(int pageno);
-static void WriteCLOGPage(int slotno);
-static void CLOGPhysicalReadPage(int pageno, int slotno);
-static void CLOGPhysicalWritePage(int pageno, int slotno);
-static int SelectLRUCLOGPage(int pageno);
-static bool ScanCLOGDirectory(int cutoffPage, bool doDeletions);
static bool CLOGPagePrecedes(int page1, int page2);
static void WriteZeroPageXlogRec(int pageno);
+static void WriteTruncateXlogRec(int pageno, TransactionId oldestXact,
+ Oid oldestXactDb);
+static void TransactionIdSetPageStatus(TransactionId xid, int nsubxids,
+ TransactionId *subxids, XidStatus status,
+ XLogRecPtr lsn, int pageno,
+ bool all_xact_same_page);
+static void TransactionIdSetStatusBit(TransactionId xid, XidStatus status,
+ XLogRecPtr lsn, int slotno);
+static void set_status_by_pages(int nsubxids, TransactionId *subxids,
+ XidStatus status, XLogRecPtr lsn);
+static bool TransactionGroupUpdateXidStatus(TransactionId xid,
+ XidStatus status, XLogRecPtr lsn, int pageno);
+static void TransactionIdSetPageStatusInternal(TransactionId xid, int nsubxids,
+ TransactionId *subxids, XidStatus status,
+ XLogRecPtr lsn, int pageno);
/*
- * Record the final state of a transaction in the commit log.
+ * TransactionIdSetTreeStatus
+ *
+ * Record the final state of transaction entries in the commit log for
+ * a transaction and its subtransaction tree. Take care to ensure this is
+ * efficient, and as atomic as possible.
+ *
+ * xid is a single xid to set status for. This will typically be
+ * the top level transactionid for a top level commit or abort. It can
+ * also be a subtransaction when we record transaction aborts.
+ *
+ * subxids is an array of xids of length nsubxids, representing subtransactions
+ * in the tree of xid. In various cases nsubxids may be zero.
+ *
+ * lsn must be the WAL location of the commit record when recording an async
+ * commit. For a synchronous commit it can be InvalidXLogRecPtr, since the
+ * caller guarantees the commit record is already flushed in that case. It
+ * should be InvalidXLogRecPtr for abort cases, too.
+ *
+ * In the commit case, atomicity is limited by whether all the subxids are in
+ * the same CLOG page as xid. If they all are, then the lock will be grabbed
+ * only once, and the status will be set to committed directly. Otherwise
+ * we must
+ * 1. set sub-committed all subxids that are not on the same page as the
+ * main xid
+ * 2. atomically set committed the main xid and the subxids on the same page
+ * 3. go over the first bunch again and set them committed
+ * Note that as far as concurrent checkers are concerned, main transaction
+ * commit as a whole is still atomic.
+ *
+ * Example:
+ * TransactionId t commits and has subxids t1, t2, t3, t4
+ * t is on page p1, t1 is also on p1, t2 and t3 are on p2, t4 is on p3
+ * 1. update pages2-3:
+ * page2: set t2,t3 as sub-committed
+ * page3: set t4 as sub-committed
+ * 2. update page1:
+ * set t1 as sub-committed,
+ * then set t as committed,
+ then set t1 as committed
+ * 3. update pages2-3:
+ * page2: set t2,t3 as committed
+ * page3: set t4 as committed
*
* NB: this is a low-level routine and is NOT the preferred entry point
- * for most uses; TransactionLogUpdate() in transam.c is the intended caller.
+ * for most uses; functions in transam.c are the intended callers.
+ *
+ * XXX Think about issuing POSIX_FADV_WILLNEED on pages that we will need,
+ * but aren't yet in cache, as well as hinting pages not to fall out of
+ * cache yet.
*/
void
-TransactionIdSetStatus(TransactionId xid, XidStatus status)
+TransactionIdSetTreeStatus(TransactionId xid, int nsubxids,
+ TransactionId *subxids, XidStatus status, XLogRecPtr lsn)
{
- int pageno = TransactionIdToPage(xid);
- int byteno = TransactionIdToByte(xid);
- int bshift = TransactionIdToBIndex(xid) * CLOG_BITS_PER_XACT;
- int slotno;
- char *byteptr;
+ int pageno = TransactionIdToPage(xid); /* get page of parent */
+ int i;
Assert(status == TRANSACTION_STATUS_COMMITTED ||
status == TRANSACTION_STATUS_ABORTED);
- LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
-
- slotno = ReadCLOGPage(pageno);
- byteptr = ClogCtl->page_buffer[slotno] + byteno;
+ /*
+ * See how many subxids, if any, are on the same page as the parent, if
+ * any.
+ */
+ for (i = 0; i < nsubxids; i++)
+ {
+ if (TransactionIdToPage(subxids[i]) != pageno)
+ break;
+ }
- /* Current state should be 0 or target state */
- Assert(((*byteptr >> bshift) & CLOG_XACT_BITMASK) == 0 ||
- ((*byteptr >> bshift) & CLOG_XACT_BITMASK) == status);
+ /*
+ * Do all items fit on a single page?
+ */
+ if (i == nsubxids)
+ {
+ /*
+ * Set the parent and all subtransactions in a single call
+ */
+ TransactionIdSetPageStatus(xid, nsubxids, subxids, status, lsn,
+ pageno, true);
+ }
+ else
+ {
+ int nsubxids_on_first_page = i;
- *byteptr |= (status << bshift);
+ /*
+ * If this is a commit then we care about doing this correctly (i.e.
+ * using the subcommitted intermediate status). By here, we know
+ * we're updating more than one page of clog, so we must mark entries
+ * that are *not* on the first page so that they show as subcommitted
+ * before we then return to update the status to fully committed.
+ *
+ * To avoid touching the first page twice, skip marking subcommitted
+ * for the subxids on that first page.
+ */
+ if (status == TRANSACTION_STATUS_COMMITTED)
+ set_status_by_pages(nsubxids - nsubxids_on_first_page,
+ subxids + nsubxids_on_first_page,
+ TRANSACTION_STATUS_SUB_COMMITTED, lsn);
- ClogCtl->page_status[slotno] = CLOG_PAGE_DIRTY;
+ /*
+ * Now set the parent and subtransactions on same page as the parent,
+ * if any
+ */
+ pageno = TransactionIdToPage(xid);
+ TransactionIdSetPageStatus(xid, nsubxids_on_first_page, subxids, status,
+ lsn, pageno, false);
- LWLockRelease(CLogControlLock);
+ /*
+ * Now work through the rest of the subxids one clog page at a time,
+ * starting from the second page onwards, like we did above.
+ */
+ set_status_by_pages(nsubxids - nsubxids_on_first_page,
+ subxids + nsubxids_on_first_page,
+ status, lsn);
+ }
}
/*
- * Interrogate the state of a transaction in the commit log.
- *
- * NB: this is a low-level routine and is NOT the preferred entry point
- * for most uses; TransactionLogTest() in transam.c is the intended caller.
+ * Helper for TransactionIdSetTreeStatus: set the status for a bunch of
+ * transactions, chunking in the separate CLOG pages involved. We never
+ * pass the whole transaction tree to this function, only subtransactions
+ * that are on different pages to the top level transaction id.
*/
-XidStatus
-TransactionIdGetStatus(TransactionId xid)
+static void
+set_status_by_pages(int nsubxids, TransactionId *subxids,
+ XidStatus status, XLogRecPtr lsn)
{
- int pageno = TransactionIdToPage(xid);
- int byteno = TransactionIdToByte(xid);
- int bshift = TransactionIdToBIndex(xid) * CLOG_BITS_PER_XACT;
- int slotno;
- char *byteptr;
- XidStatus status;
+ int pageno = TransactionIdToPage(subxids[0]);
+ int offset = 0;
+ int i = 0;
- LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
-
- slotno = ReadCLOGPage(pageno);
- byteptr = ClogCtl->page_buffer[slotno] + byteno;
-
- status = (*byteptr >> bshift) & CLOG_XACT_BITMASK;
+ Assert(nsubxids > 0); /* else the pageno fetch above is unsafe */
- LWLockRelease(CLogControlLock);
+ while (i < nsubxids)
+ {
+ int num_on_page = 0;
+ int nextpageno;
- return status;
+ do
+ {
+ nextpageno = TransactionIdToPage(subxids[i]);
+ if (nextpageno != pageno)
+ break;
+ num_on_page++;
+ i++;
+ } while (i < nsubxids);
+
+ TransactionIdSetPageStatus(InvalidTransactionId,
+ num_on_page, subxids + offset,
+ status, lsn, pageno, false);
+ offset = i;
+ pageno = nextpageno;
+ }
}
-
/*
- * Initialization of shared memory for CLOG
+ * Record the final state of transaction entries in the commit log for all
+ * entries on a single page. Atomic only on this page.
*/
-
-int
-CLOGShmemSize(void)
-{
- return MAXALIGN(sizeof(ClogCtlData) + CLOG_BLCKSZ * NUM_CLOG_BUFFERS);
-}
-
-void
-CLOGShmemInit(void)
+static void
+TransactionIdSetPageStatus(TransactionId xid, int nsubxids,
+ TransactionId *subxids, XidStatus status,
+ XLogRecPtr lsn, int pageno,
+ bool all_xact_same_page)
{
- bool found;
- char *bufptr;
- int slotno;
-
- /* this must agree with space requested by CLOGShmemSize() */
- ClogCtl = (ClogCtlData *)
- ShmemInitStruct("CLOG Ctl",
- MAXALIGN(sizeof(ClogCtlData) +
- CLOG_BLCKSZ * NUM_CLOG_BUFFERS),
- &found);
- Assert(!found);
-
- memset(ClogCtl, 0, sizeof(ClogCtlData));
+ /* Can't use group update when PGPROC overflows. */
+ StaticAssertStmt(THRESHOLD_SUBTRANS_CLOG_OPT <= PGPROC_MAX_CACHED_SUBXIDS,
+ "group clog threshold less than PGPROC cached subxids");
- bufptr = ((char *) ClogCtl) + sizeof(ClogCtlData);
-
- for (slotno = 0; slotno < NUM_CLOG_BUFFERS; slotno++)
+ /*
+ * When there is contention on CLogControlLock, we try to group multiple
+ * updates; a single leader process will perform transaction status
+ * updates for multiple backends so that the number of times
+ * CLogControlLock needs to be acquired is reduced.
+ *
+ * For this optimization to be safe, the XID in MyPgXact and the subxids
+ * in MyProc must be the same as the ones for which we're setting the
+ * status. Check that this is the case.
+ *
+ * For this optimization to be efficient, we shouldn't have too many
+ * sub-XIDs and all of the XIDs for which we're adjusting clog should be
+ * on the same page. Check those conditions, too.
+ */
+ if (all_xact_same_page && xid == MyPgXact->xid &&
+ nsubxids <= THRESHOLD_SUBTRANS_CLOG_OPT &&
+ nsubxids == MyPgXact->nxids &&
+ memcmp(subxids, MyProc->subxids.xids,
+ nsubxids * sizeof(TransactionId)) == 0)
{
- ClogCtl->page_buffer[slotno] = bufptr;
- ClogCtl->page_status[slotno] = CLOG_PAGE_EMPTY;
- ClogBufferLocks[slotno] = LWLockAssign();
- bufptr += CLOG_BLCKSZ;
- }
-
- /* ClogCtl->latest_page_number will be set later */
+ /*
+ * We don't try to do group update optimization if a process has
+ * overflowed the subxids array in its PGPROC, since in that case we
+ * don't have a complete list of XIDs for it.
+ */
+ Assert(THRESHOLD_SUBTRANS_CLOG_OPT <= PGPROC_MAX_CACHED_SUBXIDS);
- /* Init CLOG directory path */
- snprintf(ClogDir, MAXPGPATH, "%s/pg_clog", DataDir);
-}
+ /*
+ * If we can immediately acquire CLogControlLock, we update the status
+ * of our own XID and release the lock. If not, try use group XID
+ * update. If that doesn't work out, fall back to waiting for the
+ * lock to perform an update for this transaction only.
+ */
+ if (LWLockConditionalAcquire(CLogControlLock, LW_EXCLUSIVE))
+ {
+ /* Got the lock without waiting! Do the update. */
+ TransactionIdSetPageStatusInternal(xid, nsubxids, subxids, status,
+ lsn, pageno);
+ LWLockRelease(CLogControlLock);
+ return;
+ }
+ else if (TransactionGroupUpdateXidStatus(xid, status, lsn, pageno))
+ {
+ /* Group update mechanism has done the work. */
+ return;
+ }
-/*
- * This func must be called ONCE on system install. It creates
- * the initial CLOG segment. (The CLOG directory is assumed to
- * have been created by the initdb shell script, and CLOGShmemInit
- * must have been called already.)
- */
-void
-BootStrapCLOG(void)
-{
- int slotno;
+ /* Fall through only if update isn't done yet. */
+ }
+ /* Group update not applicable, or couldn't accept this page number. */
LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
-
- /* Create and zero the first page of the commit log */
- slotno = ZeroCLOGPage(0, false);
-
- /* Make sure it's written out */
- WriteCLOGPage(slotno);
- Assert(ClogCtl->page_status[slotno] == CLOG_PAGE_CLEAN);
-
+ TransactionIdSetPageStatusInternal(xid, nsubxids, subxids, status,
+ lsn, pageno);
LWLockRelease(CLogControlLock);
}
/*
- * Initialize (or reinitialize) a page of CLOG to zeroes.
- * If writeXlog is TRUE, also emit an XLOG record saying we did this.
+ * Record the final state of transaction entry in the commit log
*
- * The page is not actually written, just set up in shared memory.
- * The slot number of the new page is returned.
- *
- * Control lock must be held at entry, and will be held at exit.
+ * We don't do any locking here; caller must handle that.
*/
-static int
-ZeroCLOGPage(int pageno, bool writeXlog)
+static void
+TransactionIdSetPageStatusInternal(TransactionId xid, int nsubxids,
+ TransactionId *subxids, XidStatus status,
+ XLogRecPtr lsn, int pageno)
{
int slotno;
+ int i;
- /* Find a suitable buffer slot for the page */
- slotno = SelectLRUCLOGPage(pageno);
- Assert(ClogCtl->page_status[slotno] == CLOG_PAGE_EMPTY ||
- ClogCtl->page_status[slotno] == CLOG_PAGE_CLEAN ||
- ClogCtl->page_number[slotno] == pageno);
+ Assert(status == TRANSACTION_STATUS_COMMITTED ||
+ status == TRANSACTION_STATUS_ABORTED ||
+ (status == TRANSACTION_STATUS_SUB_COMMITTED && !TransactionIdIsValid(xid)));
+ Assert(LWLockHeldByMeInMode(CLogControlLock, LW_EXCLUSIVE));
- /* Mark the slot as containing this page */
- ClogCtl->page_number[slotno] = pageno;
- ClogCtl->page_status[slotno] = CLOG_PAGE_DIRTY;
- ClogRecentlyUsed(slotno);
+ /*
+ * If we're doing an async commit (ie, lsn is valid), then we must wait
+ * for any active write on the page slot to complete. Otherwise our
+ * update could reach disk in that write, which will not do since we
+ * mustn't let it reach disk until we've done the appropriate WAL flush.
+ * But when lsn is invalid, it's OK to scribble on a page while it is
+ * write-busy, since we don't care if the update reaches disk sooner than
+ * we think.
+ */
+ slotno = SimpleLruReadPage(ClogCtl, pageno, XLogRecPtrIsInvalid(lsn), xid);
- /* Set the buffer to zeroes */
- MemSet(ClogCtl->page_buffer[slotno], 0, CLOG_BLCKSZ);
+ /*
+ * Set the main transaction id, if any.
+ *
+ * If we update more than one xid on this page while it is being written
+ * out, we might find that some of the bits go to disk and others don't.
+ * If we are updating commits on the page with the top-level xid that
+ * could break atomicity, so we subcommit the subxids first before we mark
+ * the top-level commit.
+ */
+ if (TransactionIdIsValid(xid))
+ {
+ /* Subtransactions first, if needed ... */
+ if (status == TRANSACTION_STATUS_COMMITTED)
+ {
+ for (i = 0; i < nsubxids; i++)
+ {
+ Assert(ClogCtl->shared->page_number[slotno] == TransactionIdToPage(subxids[i]));
+ TransactionIdSetStatusBit(subxids[i],
+ TRANSACTION_STATUS_SUB_COMMITTED,
+ lsn, slotno);
+ }
+ }
- /* Assume this page is now the latest active page */
- ClogCtl->latest_page_number = pageno;
+ /* ... then the main transaction */
+ TransactionIdSetStatusBit(xid, status, lsn, slotno);
+ }
- if (writeXlog)
- WriteZeroPageXlogRec(pageno);
+ /* Set the subtransactions */
+ for (i = 0; i < nsubxids; i++)
+ {
+ Assert(ClogCtl->shared->page_number[slotno] == TransactionIdToPage(subxids[i]));
+ TransactionIdSetStatusBit(subxids[i], status, lsn, slotno);
+ }
- return slotno;
+ ClogCtl->shared->page_dirty[slotno] = true;
}
/*
- * Find a CLOG page in a shared buffer, reading it in if necessary.
- * The page number must correspond to an already-initialized page.
- *
- * Return value is the shared-buffer slot number now holding the page.
- * The buffer's LRU access info is updated.
+ * When we cannot immediately acquire CLogControlLock in exclusive mode at
+ * commit time, add ourselves to a list of processes that need their XIDs
+ * status update. The first process to add itself to the list will acquire
+ * CLogControlLock in exclusive mode and set transaction status as required
+ * on behalf of all group members. This avoids a great deal of contention
+ * around CLogControlLock when many processes are trying to commit at once,
+ * since the lock need not be repeatedly handed off from one committing
+ * process to the next.
*
- * Control lock must be held at entry, and will be held at exit.
+ * Returns true when transaction status has been updated in clog; returns
+ * false if we decided against applying the optimization because the page
+ * number we need to update differs from those processes already waiting.
*/
-static int
-ReadCLOGPage(int pageno)
+static bool
+TransactionGroupUpdateXidStatus(TransactionId xid, XidStatus status,
+ XLogRecPtr lsn, int pageno)
{
- /* Outer loop handles restart if we lose the buffer to someone else */
- for (;;)
- {
- int slotno;
+ volatile PROC_HDR *procglobal = ProcGlobal;
+ PGPROC *proc = MyProc;
+ uint32 nextidx;
+ uint32 wakeidx;
+
+ /* We should definitely have an XID whose status needs to be updated. */
+ Assert(TransactionIdIsValid(xid));
+
+ /*
+ * Add ourselves to the list of processes needing a group XID status
+ * update.
+ */
+ proc->clogGroupMember = true;
+ proc->clogGroupMemberXid = xid;
+ proc->clogGroupMemberXidStatus = status;
+ proc->clogGroupMemberPage = pageno;
+ proc->clogGroupMemberLsn = lsn;
- /* See if page already is in memory; if not, pick victim slot */
- slotno = SelectLRUCLOGPage(pageno);
+ nextidx = pg_atomic_read_u32(&procglobal->clogGroupFirst);
- /* Did we find the page in memory? */
- if (ClogCtl->page_number[slotno] == pageno &&
- ClogCtl->page_status[slotno] != CLOG_PAGE_EMPTY)
+ while (true)
+ {
+ /*
+ * Add the proc to list, if the clog page where we need to update the
+ * current transaction status is same as group leader's clog page.
+ *
+ * There is a race condition here, which is that after doing the below
+ * check and before adding this proc's clog update to a group, the
+ * group leader might have already finished the group update for this
+ * page and becomes group leader of another group. This will lead to a
+ * situation where a single group can have different clog page
+ * updates. This isn't likely and will still work, just maybe a bit
+ * less efficiently.
+ */
+ if (nextidx != INVALID_PGPROCNO &&
+ ProcGlobal->allProcs[nextidx].clogGroupMemberPage != proc->clogGroupMemberPage)
{
- /* If page is still being read in, we cannot use it yet */
- if (ClogCtl->page_status[slotno] != CLOG_PAGE_READ_IN_PROGRESS)
- {
- /* otherwise, it's ready to use */
- ClogRecentlyUsed(slotno);
- return slotno;
- }
+ proc->clogGroupMember = false;
+ return false;
}
- else
+
+ pg_atomic_write_u32(&proc->clogGroupNext, nextidx);
+
+ if (pg_atomic_compare_exchange_u32(&procglobal->clogGroupFirst,
+ &nextidx,
+ (uint32) proc->pgprocno))
+ break;
+ }
+
+ /*
+ * If the list was not empty, the leader will update the status of our
+ * XID. It is impossible to have followers without a leader because the
+ * first process that has added itself to the list will always have
+ * nextidx as INVALID_PGPROCNO.
+ */
+ if (nextidx != INVALID_PGPROCNO)
+ {
+ int extraWaits = 0;
+
+ /* Sleep until the leader updates our XID status. */
+ pgstat_report_wait_start(WAIT_EVENT_CLOG_GROUP_UPDATE);
+ for (;;)
{
- /* We found no match; assert we selected a freeable slot */
- Assert(ClogCtl->page_status[slotno] == CLOG_PAGE_EMPTY ||
- ClogCtl->page_status[slotno] == CLOG_PAGE_CLEAN);
+ /* acts as a read barrier */
+ PGSemaphoreLock(proc->sem);
+ if (!proc->clogGroupMember)
+ break;
+ extraWaits++;
}
+ pgstat_report_wait_end();
- /* Mark the slot read-busy (no-op if it already was) */
- ClogCtl->page_number[slotno] = pageno;
- ClogCtl->page_status[slotno] = CLOG_PAGE_READ_IN_PROGRESS;
+ Assert(pg_atomic_read_u32(&proc->clogGroupNext) == INVALID_PGPROCNO);
- /*
- * Temporarily mark page as recently-used to discourage
- * SelectLRUCLOGPage from selecting it again for someone else.
- */
- ClogCtl->page_lru_count[slotno] = 0;
+ /* Fix semaphore count for any absorbed wakeups */
+ while (extraWaits-- > 0)
+ PGSemaphoreUnlock(proc->sem);
+ return true;
+ }
- /* Release shared lock, grab per-buffer lock instead */
- LWLockRelease(CLogControlLock);
- LWLockAcquire(ClogBufferLocks[slotno], LW_EXCLUSIVE);
+ /* We are the leader. Acquire the lock on behalf of everyone. */
+ LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
+
+ /*
+ * Now that we've got the lock, clear the list of processes waiting for
+ * group XID status update, saving a pointer to the head of the list.
+ * Trying to pop elements one at a time could lead to an ABA problem.
+ */
+ nextidx = pg_atomic_exchange_u32(&procglobal->clogGroupFirst,
+ INVALID_PGPROCNO);
+
+ /* Remember head of list so we can perform wakeups after dropping lock. */
+ wakeidx = nextidx;
+
+ /* Walk the list and update the status of all XIDs. */
+ while (nextidx != INVALID_PGPROCNO)
+ {
+ PGPROC *proc = &ProcGlobal->allProcs[nextidx];
+ PGXACT *pgxact = &ProcGlobal->allPgXact[nextidx];
/*
- * Check to see if someone else already did the read, or took the
- * buffer away from us. If so, restart from the top.
+ * Overflowed transactions should not use group XID status update
+ * mechanism.
*/
- if (ClogCtl->page_number[slotno] != pageno ||
- ClogCtl->page_status[slotno] != CLOG_PAGE_READ_IN_PROGRESS)
- {
- LWLockRelease(ClogBufferLocks[slotno]);
- LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
- continue;
- }
+ Assert(!pgxact->overflowed);
- /* Okay, do the read */
- CLOGPhysicalReadPage(pageno, slotno);
+ TransactionIdSetPageStatusInternal(proc->clogGroupMemberXid,
+ pgxact->nxids,
+ proc->subxids.xids,
+ proc->clogGroupMemberXidStatus,
+ proc->clogGroupMemberLsn,
+ proc->clogGroupMemberPage);
- /* Re-acquire shared control lock and update page state */
- LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
+ /* Move to next proc in list. */
+ nextidx = pg_atomic_read_u32(&proc->clogGroupNext);
+ }
+
+ /* We're done with the lock now. */
+ LWLockRelease(CLogControlLock);
+
+ /*
+ * Now that we've released the lock, go back and wake everybody up. We
+ * don't do this under the lock so as to keep lock hold times to a
+ * minimum.
+ */
+ while (wakeidx != INVALID_PGPROCNO)
+ {
+ PGPROC *proc = &ProcGlobal->allProcs[wakeidx];
- Assert(ClogCtl->page_number[slotno] == pageno &&
- ClogCtl->page_status[slotno] == CLOG_PAGE_READ_IN_PROGRESS);
+ wakeidx = pg_atomic_read_u32(&proc->clogGroupNext);
+ pg_atomic_write_u32(&proc->clogGroupNext, INVALID_PGPROCNO);
- ClogCtl->page_status[slotno] = CLOG_PAGE_CLEAN;
+ /* ensure all previous writes are visible before follower continues. */
+ pg_write_barrier();
- LWLockRelease(ClogBufferLocks[slotno]);
+ proc->clogGroupMember = false;
- ClogRecentlyUsed(slotno);
- return slotno;
+ if (proc != MyProc)
+ PGSemaphoreUnlock(proc->sem);
}
+
+ return true;
}
/*
- * Write a CLOG page from a shared buffer, if necessary.
- * Does nothing if the specified slot is not dirty.
- *
- * NOTE: only one write attempt is made here. Hence, it is possible that
- * the page is still dirty at exit (if someone else re-dirtied it during
- * the write). However, we *do* attempt a fresh write even if the page
- * is already being written; this is for checkpoints.
+ * Sets the commit status of a single transaction.
*
- * Control lock must be held at entry, and will be held at exit.
+ * Must be called with CLogControlLock held
*/
static void
-WriteCLOGPage(int slotno)
+TransactionIdSetStatusBit(TransactionId xid, XidStatus status, XLogRecPtr lsn, int slotno)
{
- int pageno;
-
- /* Do nothing if page does not need writing */
- if (ClogCtl->page_status[slotno] != CLOG_PAGE_DIRTY &&
- ClogCtl->page_status[slotno] != CLOG_PAGE_WRITE_IN_PROGRESS)
- return;
+ int byteno = TransactionIdToByte(xid);
+ int bshift = TransactionIdToBIndex(xid) * CLOG_BITS_PER_XACT;
+ char *byteptr;
+ char byteval;
+ char curval;
- pageno = ClogCtl->page_number[slotno];
-
- /* Release shared lock, grab per-buffer lock instead */
- LWLockRelease(CLogControlLock);
- LWLockAcquire(ClogBufferLocks[slotno], LW_EXCLUSIVE);
+ byteptr = ClogCtl->shared->page_buffer[slotno] + byteno;
+ curval = (*byteptr >> bshift) & CLOG_XACT_BITMASK;
/*
- * Check to see if someone else already did the write, or took the
- * buffer away from us. If so, do nothing. NOTE: we really should
- * never see WRITE_IN_PROGRESS here, since that state should only
- * occur while the writer is holding the buffer lock. But accept it
- * so that we have a recovery path if a writer aborts.
+ * When replaying transactions during recovery we still need to perform
+ * the two phases of subcommit and then commit. However, some transactions
+ * are already correctly marked, so we just treat those as a no-op which
+ * allows us to keep the following Assert as restrictive as possible.
*/
- if (ClogCtl->page_number[slotno] != pageno ||
- (ClogCtl->page_status[slotno] != CLOG_PAGE_DIRTY &&
- ClogCtl->page_status[slotno] != CLOG_PAGE_WRITE_IN_PROGRESS))
- {
- LWLockRelease(ClogBufferLocks[slotno]);
- LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
+ if (InRecovery && status == TRANSACTION_STATUS_SUB_COMMITTED &&
+ curval == TRANSACTION_STATUS_COMMITTED)
return;
- }
/*
- * Mark the slot write-busy. After this point, a transaction status
- * update on this page will mark it dirty again. NB: we are assuming
- * that read/write of the page status field is atomic, since we change
- * the state while not holding control lock. However, we cannot set
- * this state any sooner, or we'd possibly fool a previous writer into
- * thinking he's successfully dumped the page when he hasn't.
- * (Scenario: other writer starts, page is redirtied, we come along
- * and set WRITE_IN_PROGRESS again, other writer completes and sets
- * CLEAN because redirty info has been lost, then we think it's clean
- * too.)
+ * Current state change should be from 0 or subcommitted to target state
+ * or we should already be there when replaying changes during recovery.
*/
- ClogCtl->page_status[slotno] = CLOG_PAGE_WRITE_IN_PROGRESS;
-
- /* Okay, do the write */
- CLOGPhysicalWritePage(pageno, slotno);
+ Assert(curval == 0 ||
+ (curval == TRANSACTION_STATUS_SUB_COMMITTED &&
+ status != TRANSACTION_STATUS_IN_PROGRESS) ||
+ curval == status);
- /* Re-acquire shared control lock and update page state */
- LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
+ /* note this assumes exclusive access to the clog page */
+ byteval = *byteptr;
+ byteval &= ~(((1 << CLOG_BITS_PER_XACT) - 1) << bshift);
+ byteval |= (status << bshift);
+ *byteptr = byteval;
- Assert(ClogCtl->page_number[slotno] == pageno &&
- (ClogCtl->page_status[slotno] == CLOG_PAGE_WRITE_IN_PROGRESS ||
- ClogCtl->page_status[slotno] == CLOG_PAGE_DIRTY));
-
- /* Cannot set CLEAN if someone re-dirtied page since write started */
- if (ClogCtl->page_status[slotno] == CLOG_PAGE_WRITE_IN_PROGRESS)
- ClogCtl->page_status[slotno] = CLOG_PAGE_CLEAN;
+ /*
+ * Update the group LSN if the transaction completion LSN is higher.
+ *
+ * Note: lsn will be invalid when supplied during InRecovery processing,
+ * so we don't need to do anything special to avoid LSN updates during
+ * recovery. After recovery completes the next clog change will set the
+ * LSN correctly.
+ */
+ if (!XLogRecPtrIsInvalid(lsn))
+ {
+ int lsnindex = GetLSNIndex(slotno, xid);
- LWLockRelease(ClogBufferLocks[slotno]);
+ if (ClogCtl->shared->group_lsn[lsnindex] < lsn)
+ ClogCtl->shared->group_lsn[lsnindex] = lsn;
+ }
}
/*
- * Physical read of a (previously existing) page into a buffer slot
+ * Interrogate the state of a transaction in the commit log.
*
- * For now, assume it's not worth keeping a file pointer open across
- * read/write operations. We could cache one virtual file pointer ...
+ * Aside from the actual commit status, this function returns (into *lsn)
+ * an LSN that is late enough to be able to guarantee that if we flush up to
+ * that LSN then we will have flushed the transaction's commit record to disk.
+ * The result is not necessarily the exact LSN of the transaction's commit
+ * record! For example, for long-past transactions (those whose clog pages
+ * already migrated to disk), we'll return InvalidXLogRecPtr. Also, because
+ * we group transactions on the same clog page to conserve storage, we might
+ * return the LSN of a later transaction that falls into the same group.
+ *
+ * NB: this is a low-level routine and is NOT the preferred entry point
+ * for most uses; TransactionLogFetch() in transam.c is the intended caller.
*/
-static void
-CLOGPhysicalReadPage(int pageno, int slotno)
+XidStatus
+TransactionIdGetStatus(TransactionId xid, XLogRecPtr *lsn)
{
- int segno = pageno / CLOG_PAGES_PER_SEGMENT;
- int rpageno = pageno % CLOG_PAGES_PER_SEGMENT;
- int offset = rpageno * CLOG_BLCKSZ;
- char path[MAXPGPATH];
- int fd;
+ int pageno = TransactionIdToPage(xid);
+ int byteno = TransactionIdToByte(xid);
+ int bshift = TransactionIdToBIndex(xid) * CLOG_BITS_PER_XACT;
+ int slotno;
+ int lsnindex;
+ char *byteptr;
+ XidStatus status;
- ClogFileName(path, segno);
+ /* lock is acquired by SimpleLruReadPage_ReadOnly */
- /*
- * In a crash-and-restart situation, it's possible for us to receive
- * commands to set the commit status of transactions whose bits are in
- * already-truncated segments of the commit log (see notes in
- * CLOGPhysicalWritePage). Hence, if we are InRecovery, allow the
- * case where the file doesn't exist, and return zeroes instead.
- */
- fd = BasicOpenFile(path, O_RDWR | PG_BINARY, S_IRUSR | S_IWUSR);
- if (fd < 0)
- {
- if (errno != ENOENT || !InRecovery)
- elog(STOP, "open of %s failed: %m", path);
- elog(DEBUG, "clog file %s doesn't exist, reading as zeroes", path);
- MemSet(ClogCtl->page_buffer[slotno], 0, CLOG_BLCKSZ);
- return;
- }
+ slotno = SimpleLruReadPage_ReadOnly(ClogCtl, pageno, xid);
+ byteptr = ClogCtl->shared->page_buffer[slotno] + byteno;
+
+ status = (*byteptr >> bshift) & CLOG_XACT_BITMASK;
- if (lseek(fd, (off_t) offset, SEEK_SET) < 0)
- elog(STOP, "lseek of clog file %u, offset %u failed: %m",
- segno, offset);
+ lsnindex = GetLSNIndex(slotno, xid);
+ *lsn = ClogCtl->shared->group_lsn[lsnindex];
- errno = 0;
- if (read(fd, ClogCtl->page_buffer[slotno], CLOG_BLCKSZ) != CLOG_BLCKSZ)
- elog(STOP, "read of clog file %u, offset %u failed: %m",
- segno, offset);
+ LWLockRelease(CLogControlLock);
- close(fd);
+ return status;
}
/*
- * Physical write of a page from a buffer slot
+ * Number of shared CLOG buffers.
*
- * For now, assume it's not worth keeping a file pointer open across
- * read/write operations. We could cache one virtual file pointer ...
+ * On larger multi-processor systems, it is possible to have many CLOG page
+ * requests in flight at one time which could lead to disk access for CLOG
+ * page if the required page is not found in memory. Testing revealed that we
+ * can get the best performance by having 128 CLOG buffers, more than that it
+ * doesn't improve performance.
+ *
+ * Unconditionally keeping the number of CLOG buffers to 128 did not seem like
+ * a good idea, because it would increase the minimum amount of shared memory
+ * required to start, which could be a problem for people running very small
+ * configurations. The following formula seems to represent a reasonable
+ * compromise: people with very low values for shared_buffers will get fewer
+ * CLOG buffers as well, and everyone else will get 128.
*/
-static void
-CLOGPhysicalWritePage(int pageno, int slotno)
+Size
+CLOGShmemBuffers(void)
{
- int segno = pageno / CLOG_PAGES_PER_SEGMENT;
- int rpageno = pageno % CLOG_PAGES_PER_SEGMENT;
- int offset = rpageno * CLOG_BLCKSZ;
- char path[MAXPGPATH];
- int fd;
+ return Min(128, Max(4, NBuffers / 512));
+}
- ClogFileName(path, segno);
+/*
+ * Initialization of shared memory for CLOG
+ */
+Size
+CLOGShmemSize(void)
+{
+ return SimpleLruShmemSize(CLOGShmemBuffers(), CLOG_LSNS_PER_PAGE);
+}
- /*
- * If the file doesn't already exist, we should create it. It is
- * possible for this to need to happen when writing a page that's not
- * first in its segment; we assume the OS can cope with that. (Note:
- * it might seem that it'd be okay to create files only when
- * ZeroCLOGPage is called for the first page of a segment. However,
- * if after a crash and restart the REDO logic elects to replay the
- * log from a checkpoint before the latest one, then it's possible
- * that we will get commands to set transaction status of transactions
- * that have already been truncated from the commit log. Easiest way
- * to deal with that is to accept references to nonexistent files here
- * and in CLOGPhysicalReadPage.)
- */
- fd = BasicOpenFile(path, O_RDWR | PG_BINARY, S_IRUSR | S_IWUSR);
- if (fd < 0)
- {
- if (errno != ENOENT)
- elog(STOP, "open of %s failed: %m", path);
- fd = BasicOpenFile(path, O_RDWR | O_CREAT | O_EXCL | PG_BINARY,
- S_IRUSR | S_IWUSR);
- if (fd < 0)
- elog(STOP, "creation of file %s failed: %m", path);
- }
+void
+CLOGShmemInit(void)
+{
+ ClogCtl->PagePrecedes = CLOGPagePrecedes;
+ SimpleLruInit(ClogCtl, "clog", CLOGShmemBuffers(), CLOG_LSNS_PER_PAGE,
+ CLogControlLock, "pg_xact", LWTRANCHE_CLOG_BUFFERS);
+}
- if (lseek(fd, (off_t) offset, SEEK_SET) < 0)
- elog(STOP, "lseek of clog file %u, offset %u failed: %m",
- segno, offset);
+/*
+ * This func must be called ONCE on system install. It creates
+ * the initial CLOG segment. (The CLOG directory is assumed to
+ * have been created by initdb, and CLOGShmemInit must have been
+ * called already.)
+ */
+void
+BootStrapCLOG(void)
+{
+ int slotno;
- errno = 0;
- if (write(fd, ClogCtl->page_buffer[slotno], CLOG_BLCKSZ) != CLOG_BLCKSZ)
- {
- /* if write didn't set errno, assume problem is no disk space */
- if (errno == 0)
- errno = ENOSPC;
- elog(STOP, "write of clog file %u, offset %u failed: %m",
- segno, offset);
- }
+ LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
+
+ /* Create and zero the first page of the commit log */
+ slotno = ZeroCLOGPage(0, false);
- close(fd);
+ /* Make sure it's written out */
+ SimpleLruWritePage(ClogCtl, slotno);
+ Assert(!ClogCtl->shared->page_dirty[slotno]);
+
+ LWLockRelease(CLogControlLock);
}
/*
- * Select the slot to re-use when we need a free slot.
+ * Initialize (or reinitialize) a page of CLOG to zeroes.
+ * If writeXlog is true, also emit an XLOG record saying we did this.
*
- * The target page number is passed because we need to consider the
- * possibility that some other process reads in the target page while
- * we are doing I/O to free a slot. Hence, check or recheck to see if
- * any slot already holds the target page, and return that slot if so.
- * Thus, the returned slot is *either* a slot already holding the pageno
- * (could be any state except EMPTY), *or* a freeable slot (state EMPTY
- * or CLEAN).
+ * The page is not actually written, just set up in shared memory.
+ * The slot number of the new page is returned.
*
* Control lock must be held at entry, and will be held at exit.
*/
static int
-SelectLRUCLOGPage(int pageno)
+ZeroCLOGPage(int pageno, bool writeXlog)
{
- /* Outer loop handles restart after I/O */
- for (;;)
- {
- int slotno;
- int bestslot = 0;
- unsigned int bestcount = 0;
-
- /* See if page already has a buffer assigned */
- for (slotno = 0; slotno < NUM_CLOG_BUFFERS; slotno++)
- {
- if (ClogCtl->page_number[slotno] == pageno &&
- ClogCtl->page_status[slotno] != CLOG_PAGE_EMPTY)
- return slotno;
- }
-
- /*
- * If we find any EMPTY slot, just select that one. Else locate
- * the least-recently-used slot that isn't the latest CLOG page.
- */
- for (slotno = 0; slotno < NUM_CLOG_BUFFERS; slotno++)
- {
- if (ClogCtl->page_status[slotno] == CLOG_PAGE_EMPTY)
- return slotno;
- if (ClogCtl->page_lru_count[slotno] > bestcount &&
- ClogCtl->page_number[slotno] != ClogCtl->latest_page_number)
- {
- bestslot = slotno;
- bestcount = ClogCtl->page_lru_count[slotno];
- }
- }
+ int slotno;
- /*
- * If the selected page is clean, we're set.
- */
- if (ClogCtl->page_status[bestslot] == CLOG_PAGE_CLEAN)
- return bestslot;
+ slotno = SimpleLruZeroPage(ClogCtl, pageno);
- /*
- * We need to do I/O. Normal case is that we have to write it
- * out, but it's possible in the worst case to have selected a
- * read-busy page. In that case we use ReadCLOGPage to wait for
- * the read to complete.
- */
- if (ClogCtl->page_status[bestslot] == CLOG_PAGE_READ_IN_PROGRESS)
- (void) ReadCLOGPage(ClogCtl->page_number[bestslot]);
- else
- WriteCLOGPage(bestslot);
+ if (writeXlog)
+ WriteZeroPageXlogRec(pageno);
- /*
- * Now loop back and try again. This is the easiest way of
- * dealing with corner cases such as the victim page being
- * re-dirtied while we wrote it.
- */
- }
+ return slotno;
}
/*
* This must be called ONCE during postmaster or standalone-backend startup,
- * after StartupXLOG has initialized ShmemVariableCache->nextXid.
+ * after StartupXLOG has initialized ShmemVariableCache->nextFullXid.
*/
void
StartupCLOG(void)
{
+ TransactionId xid = XidFromFullTransactionId(ShmemVariableCache->nextFullXid);
+ int pageno = TransactionIdToPage(xid);
+
+ LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
+
/*
* Initialize our idea of the latest page number.
*/
- ClogCtl->latest_page_number = TransactionIdToPage(ShmemVariableCache->nextXid);
+ ClogCtl->shared->latest_page_number = pageno;
+
+ LWLockRelease(CLogControlLock);
}
/*
- * This must be called ONCE during postmaster or standalone-backend shutdown
+ * This must be called ONCE at the end of startup/recovery.
*/
void
-ShutdownCLOG(void)
+TrimCLOG(void)
{
- int slotno;
+ TransactionId xid = XidFromFullTransactionId(ShmemVariableCache->nextFullXid);
+ int pageno = TransactionIdToPage(xid);
LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
- for (slotno = 0; slotno < NUM_CLOG_BUFFERS; slotno++)
+ /*
+ * Re-Initialize our idea of the latest page number.
+ */
+ ClogCtl->shared->latest_page_number = pageno;
+
+ /*
+ * Zero out the remainder of the current clog page. Under normal
+ * circumstances it should be zeroes already, but it seems at least
+ * theoretically possible that XLOG replay will have settled on a nextXID
+ * value that is less than the last XID actually used and marked by the
+ * previous database lifecycle (since subtransaction commit writes clog
+ * but makes no WAL entry). Let's just be safe. (We need not worry about
+ * pages beyond the current one, since those will be zeroed when first
+ * used. For the same reason, there is no need to do anything when
+ * nextFullXid is exactly at a page boundary; and it's likely that the
+ * "current" page doesn't exist yet in that case.)
+ */
+ if (TransactionIdToPgIndex(xid) != 0)
{
- WriteCLOGPage(slotno);
- Assert(ClogCtl->page_status[slotno] == CLOG_PAGE_EMPTY ||
- ClogCtl->page_status[slotno] == CLOG_PAGE_CLEAN);
+ int byteno = TransactionIdToByte(xid);
+ int bshift = TransactionIdToBIndex(xid) * CLOG_BITS_PER_XACT;
+ int slotno;
+ char *byteptr;
+
+ slotno = SimpleLruReadPage(ClogCtl, pageno, false, xid);
+ byteptr = ClogCtl->shared->page_buffer[slotno] + byteno;
+
+ /* Zero so-far-unused positions in the current byte */
+ *byteptr &= (1 << bshift) - 1;
+ /* Zero the rest of the page */
+ MemSet(byteptr + 1, 0, BLCKSZ - byteno - 1);
+
+ ClogCtl->shared->page_dirty[slotno] = true;
}
LWLockRelease(CLogControlLock);
}
/*
- * Perform a checkpoint --- either during shutdown, or on-the-fly
+ * This must be called ONCE during postmaster or standalone-backend shutdown
*/
void
-CheckPointCLOG(void)
+ShutdownCLOG(void)
{
- int slotno;
+ /* Flush dirty CLOG pages to disk */
+ TRACE_POSTGRESQL_CLOG_CHECKPOINT_START(false);
+ SimpleLruFlush(ClogCtl, false);
- LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
+ /*
+ * fsync pg_xact to ensure that any files flushed previously are durably
+ * on disk.
+ */
+ fsync_fname("pg_xact", true);
- for (slotno = 0; slotno < NUM_CLOG_BUFFERS; slotno++)
- {
- WriteCLOGPage(slotno);
+ TRACE_POSTGRESQL_CLOG_CHECKPOINT_DONE(false);
+}
- /*
- * We cannot assert that the slot is clean now, since another
- * process might have re-dirtied it already. That's okay.
- */
- }
+/*
+ * Perform a checkpoint --- either during shutdown, or on-the-fly
+ */
+void
+CheckPointCLOG(void)
+{
+ /* Flush dirty CLOG pages to disk */
+ TRACE_POSTGRESQL_CLOG_CHECKPOINT_START(true);
+ SimpleLruFlush(ClogCtl, true);
- LWLockRelease(CLogControlLock);
+ /*
+ * fsync pg_xact to ensure that any files flushed previously are durably
+ * on disk.
+ */
+ fsync_fname("pg_xact", true);
+
+ TRACE_POSTGRESQL_CLOG_CHECKPOINT_DONE(true);
}
/*
* Remove all CLOG segments before the one holding the passed transaction ID
*
- * When this is called, we know that the database logically contains no
- * reference to transaction IDs older than oldestXact. However, we must
- * not truncate the CLOG until we have performed a checkpoint, to ensure
- * that no such references remain on disk either; else a crash just after
- * the truncation might leave us with a problem. Since CLOG segments hold
- * a large number of transactions, the opportunity to actually remove a
- * segment is fairly rare, and so it seems best not to do the checkpoint
- * unless we have confirmed that there is a removable segment. Therefore
- * we issue the checkpoint command here, not in higher-level code as might
- * seem cleaner.
+ * Before removing any CLOG data, we must flush XLOG to disk, to ensure
+ * that any recently-emitted FREEZE_PAGE records have reached disk; otherwise
+ * a crash and restart might leave us with some unfrozen tuples referencing
+ * removed CLOG data. We choose to emit a special TRUNCATE XLOG record too.
+ * Replaying the deletion from XLOG is not critical, since the files could
+ * just as well be removed later, but doing so prevents a long-running hot
+ * standby server from acquiring an unreasonably bloated CLOG directory.
+ *
+ * Since CLOG segments hold a large number of transactions, the opportunity to
+ * actually remove a segment is fairly rare, and so it seems best not to do
+ * the XLOG flush unless we have confirmed that there is a removable segment.
*/
void
-TruncateCLOG(TransactionId oldestXact)
+TruncateCLOG(TransactionId oldestXact, Oid oldestxid_datoid)
{
int cutoffPage;
- int slotno;
/*
- * The cutoff point is the start of the segment containing oldestXact.
+ * The cutoff point is the start of the segment containing oldestXact. We
+ * pass the *page* containing oldestXact to SimpleLruTruncate.
*/
- oldestXact -= oldestXact % CLOG_XACTS_PER_SEGMENT;
cutoffPage = TransactionIdToPage(oldestXact);
- if (!ScanCLOGDirectory(cutoffPage, false))
+ /* Check to see if there's any files that could be removed */
+ if (!SlruScanDirectory(ClogCtl, SlruScanDirCbReportPresence, &cutoffPage))
return; /* nothing to remove */
- /* Perform a CHECKPOINT */
- CreateCheckPoint(false);
-
/*
- * Scan CLOG shared memory and remove any pages preceding the cutoff
- * page, to ensure we won't rewrite them later. (Any dirty pages
- * should have been flushed already during the checkpoint, we're just
- * being extra careful here.)
+ * Advance oldestClogXid before truncating clog, so concurrent xact status
+ * lookups can ensure they don't attempt to access truncated-away clog.
+ *
+ * It's only necessary to do this if we will actually truncate away clog
+ * pages.
*/
- LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
-
-restart:;
+ AdvanceOldestClogXid(oldestXact);
/*
- * While we are holding the lock, make an important safety check: the
- * planned cutoff point must be <= the current CLOG endpoint page.
- * Otherwise we have already wrapped around, and proceeding with the
- * truncation would risk removing the current CLOG segment.
+ * Write XLOG record and flush XLOG to disk. We record the oldest xid
+ * we're keeping information about here so we can ensure that it's always
+ * ahead of clog truncation in case we crash, and so a standby finds out
+ * the new valid xid before the next checkpoint.
*/
- if (CLOGPagePrecedes(ClogCtl->latest_page_number, cutoffPage))
- {
- LWLockRelease(CLogControlLock);
- elog(LOG, "unable to truncate commit log: apparent wraparound");
- return;
- }
-
- for (slotno = 0; slotno < NUM_CLOG_BUFFERS; slotno++)
- {
- if (ClogCtl->page_status[slotno] == CLOG_PAGE_EMPTY)
- continue;
- if (!CLOGPagePrecedes(ClogCtl->page_number[slotno], cutoffPage))
- continue;
-
- /*
- * If page is CLEAN, just change state to EMPTY (expected case).
- */
- if (ClogCtl->page_status[slotno] == CLOG_PAGE_CLEAN)
- {
- ClogCtl->page_status[slotno] = CLOG_PAGE_EMPTY;
- continue;
- }
-
- /*
- * Hmm, we have (or may have) I/O operations acting on the page,
- * so we've got to wait for them to finish and then start again.
- * This is the same logic as in SelectLRUCLOGPage.
- */
- if (ClogCtl->page_status[slotno] == CLOG_PAGE_READ_IN_PROGRESS)
- (void) ReadCLOGPage(ClogCtl->page_number[slotno]);
- else
- WriteCLOGPage(slotno);
- goto restart;
- }
-
- LWLockRelease(CLogControlLock);
+ WriteTruncateXlogRec(cutoffPage, oldestXact, oldestxid_datoid);
/* Now we can remove the old CLOG segment(s) */
- (void) ScanCLOGDirectory(cutoffPage, true);
+ SimpleLruTruncate(ClogCtl, cutoffPage);
}
-/*
- * TruncateCLOG subroutine: scan CLOG directory for removable segments.
- * Actually remove them iff doDeletions is true. Return TRUE iff any
- * removable segments were found. Note: no locking is needed.
- */
-static bool
-ScanCLOGDirectory(int cutoffPage, bool doDeletions)
-{
- bool found = false;
- DIR *cldir;
- struct dirent *clde;
- int segno;
- int segpage;
- char path[MAXPGPATH];
-
- cldir = opendir(ClogDir);
- if (cldir == NULL)
- elog(STOP, "could not open transaction-commit log directory (%s): %m",
- ClogDir);
-
- errno = 0;
- while ((clde = readdir(cldir)) != NULL)
- {
- if (strlen(clde->d_name) == 4 &&
- strspn(clde->d_name, "0123456789ABCDEF") == 4)
- {
- segno = (int) strtol(clde->d_name, NULL, 16);
- segpage = segno * CLOG_PAGES_PER_SEGMENT;
- if (CLOGPagePrecedes(segpage, cutoffPage))
- {
- found = true;
- if (doDeletions)
- {
- elog(LOG, "removing commit log file %s", clde->d_name);
- snprintf(path, MAXPGPATH, "%s/%s", ClogDir, clde->d_name);
- unlink(path);
- }
- }
- }
- errno = 0;
- }
- if (errno)
- elog(STOP, "could not read transaction-commit log directory (%s): %m",
- ClogDir);
- closedir(cldir);
-
- return found;
-}
/*
* Decide which of two CLOG page numbers is "older" for truncation purposes.
TransactionId xid1;
TransactionId xid2;
- xid1 = (TransactionId) page1 *CLOG_XACTS_PER_PAGE;
-
+ xid1 = ((TransactionId) page1) * CLOG_XACTS_PER_PAGE;
xid1 += FirstNormalTransactionId;
- xid2 = (TransactionId) page2 *CLOG_XACTS_PER_PAGE;
-
+ xid2 = ((TransactionId) page2) * CLOG_XACTS_PER_PAGE;
xid2 += FirstNormalTransactionId;
return TransactionIdPrecedes(xid1, xid2);
/*
* Write a ZEROPAGE xlog record
- *
- * Note: xlog record is marked as outside transaction control, since we
- * want it to be redone whether the invoking transaction commits or not.
- * (Besides which, this is normally done just before entering a transaction.)
*/
static void
WriteZeroPageXlogRec(int pageno)
{
- XLogRecData rdata;
+ XLogBeginInsert();
+ XLogRegisterData((char *) (&pageno), sizeof(int));
+ (void) XLogInsert(RM_CLOG_ID, CLOG_ZEROPAGE);
+}
+
+/*
+ * Write a TRUNCATE xlog record
+ *
+ * We must flush the xlog record to disk before returning --- see notes
+ * in TruncateCLOG().
+ */
+static void
+WriteTruncateXlogRec(int pageno, TransactionId oldestXact, Oid oldestXactDb)
+{
+ XLogRecPtr recptr;
+ xl_clog_truncate xlrec;
+
+ xlrec.pageno = pageno;
+ xlrec.oldestXact = oldestXact;
+ xlrec.oldestXactDb = oldestXactDb;
- rdata.buffer = InvalidBuffer;
- rdata.data = (char *) (&pageno);
- rdata.len = sizeof(int);
- rdata.next = NULL;
- (void) XLogInsert(RM_CLOG_ID, CLOG_ZEROPAGE | XLOG_NO_TRAN, &rdata);
+ XLogBeginInsert();
+ XLogRegisterData((char *) (&xlrec), sizeof(xl_clog_truncate));
+ recptr = XLogInsert(RM_CLOG_ID, CLOG_TRUNCATE);
+ XLogFlush(recptr);
}
/*
* CLOG resource manager's routines
*/
void
-clog_redo(XLogRecPtr lsn, XLogRecord *record)
+clog_redo(XLogReaderState *record)
{
- uint8 info = record->xl_info & ~XLR_INFO_MASK;
+ uint8 info = XLogRecGetInfo(record) & ~XLR_INFO_MASK;
+
+ /* Backup blocks are not used in clog records */
+ Assert(!XLogRecHasAnyBlockRefs(record));
if (info == CLOG_ZEROPAGE)
{
LWLockAcquire(CLogControlLock, LW_EXCLUSIVE);
slotno = ZeroCLOGPage(pageno, false);
- WriteCLOGPage(slotno);
- Assert(ClogCtl->page_status[slotno] == CLOG_PAGE_CLEAN);
+ SimpleLruWritePage(ClogCtl, slotno);
+ Assert(!ClogCtl->shared->page_dirty[slotno]);
LWLockRelease(CLogControlLock);
}
-}
+ else if (info == CLOG_TRUNCATE)
+ {
+ xl_clog_truncate xlrec;
-void
-clog_undo(XLogRecPtr lsn, XLogRecord *record)
-{
-}
+ memcpy(&xlrec, XLogRecGetData(record), sizeof(xl_clog_truncate));
-void
-clog_desc(char *buf, uint8 xl_info, char *rec)
-{
- uint8 info = xl_info & ~XLR_INFO_MASK;
+ /*
+ * During XLOG replay, latest_page_number isn't set up yet; insert a
+ * suitable value to bypass the sanity test in SimpleLruTruncate.
+ */
+ ClogCtl->shared->latest_page_number = xlrec.pageno;
- if (info == CLOG_ZEROPAGE)
- {
- int pageno;
+ AdvanceOldestClogXid(xlrec.oldestXact);
- memcpy(&pageno, rec, sizeof(int));
- sprintf(buf + strlen(buf), "zeropage: %d", pageno);
+ SimpleLruTruncate(ClogCtl, xlrec.pageno);
}
else
- strcat(buf, "UNKNOWN");
+ elog(PANIC, "clog_redo: unknown op code %u", info);
}
projects / postgresql / blobdiff