diff --git a/src/backend/access/transam/clog.c b/src/backend/access/transam/clog.c index ea83655..7ac7ff5 100644 --- a/src/backend/access/transam/clog.c +++ b/src/backend/access/transam/clog.c @@ -35,11 +35,13 @@ #include "access/clog.h" #include "access/slru.h" #include "access/transam.h" +#include "access/twophase.h" #include "access/xlog.h" #include "access/xloginsert.h" #include "access/xlogutils.h" #include "miscadmin.h" #include "pg_trace.h" +#include "storage/proc.h" /* * Defines for CLOG page sizes. A page is the same BLCKSZ as is used @@ -86,11 +88,17 @@ static void WriteZeroPageXlogRec(int pageno); static void WriteTruncateXlogRec(int pageno); static void TransactionIdSetPageStatus(TransactionId xid, int nsubxids, TransactionId *subxids, XidStatus status, - XLogRecPtr lsn, int pageno); + XLogRecPtr lsn, int pageno, + bool all_trans_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); /* @@ -173,7 +181,7 @@ TransactionIdSetTreeStatus(TransactionId xid, int nsubxids, * Set the parent and all subtransactions in a single call */ TransactionIdSetPageStatus(xid, nsubxids, subxids, status, lsn, - pageno); + pageno, true); } else { @@ -200,7 +208,7 @@ TransactionIdSetTreeStatus(TransactionId xid, int nsubxids, */ pageno = TransactionIdToPage(xid); TransactionIdSetPageStatus(xid, nsubxids_on_first_page, subxids, status, - lsn, pageno); + lsn, pageno, false); /* * Now work through the rest of the subxids one clog page at a time, @@ -238,7 +246,7 @@ set_status_by_pages(int nsubxids, TransactionId *subxids, TransactionIdSetPageStatus(InvalidTransactionId, num_on_page, subxids + offset, - status, lsn, pageno); + status, lsn, pageno, false); offset = i; pageno = TransactionIdToPage(subxids[offset]); } @@ -248,12 +256,78 @@ set_status_by_pages(int nsubxids, TransactionId *subxids, * Record the final state of transaction entries in the commit log for * all entries on a single page. Atomic only on this page. * + * Group the status update for transactions. This improves the efficiency + * of the transaction status update by reducing the number of lock + * acquisitions required for it. To achieve the group transaction status + * update, we need to populate the transaction status related information + * in shared memory and doing it for overflowed sub-transactions would need + * a big chunk of shared memory, so we are not doing this optimization for + * such cases. This optimization is only applicable if the transaction and + * all child sub-transactions belong to same page which we presume to be the + * most common case, we might be able to apply this when they are not on same + * page, but that needs us to map sub-transactions in proc's XidCache based + * on pageno for which each time Group leader needs to set the transaction + * status and that can lead to some performance penalty as well because it + * needs to be done after acquiring CLogControlLock, so let's leave that + * case for now. We don't do this optimization for prepared transactions + * as the dummy proc associated with such transactions doesn't have a + * semaphore associated with it and the same is required for group status + * update. We choose not to create a semaphore for dummy procs for this + * purpose as the advantage of using this optimization for prepared transactions + * is not clear. + * * Otherwise API is same as TransactionIdSetTreeStatus() */ static void TransactionIdSetPageStatus(TransactionId xid, int nsubxids, TransactionId *subxids, XidStatus status, - XLogRecPtr lsn, int pageno) + XLogRecPtr lsn, int pageno, + bool all_trans_same_page) +{ + if (!InRecovery && + all_trans_same_page && + nsubxids < PGPROC_MAX_CACHED_SUBXIDS && + !IsGXactActive()) + { + /* + * If we can immediately acquire CLogControlLock, we update the status + * of our own XID and release the lock. If not, use group XID status + * update to improve efficiency and if still not able to update, then + * acquire CLogControlLock and update it. + */ + if (LWLockConditionalAcquire(CLogControlLock, LW_EXCLUSIVE)) + { + TransactionIdSetPageStatusInternal(xid, nsubxids, subxids, status, lsn, pageno); + LWLockRelease(CLogControlLock); + } + else if (!TransactionGroupUpdateXidStatus(xid, status, lsn, pageno)) + { + LWLockAcquire(CLogControlLock, LW_EXCLUSIVE); + + TransactionIdSetPageStatusInternal(xid, nsubxids, subxids, status, lsn, pageno); + + LWLockRelease(CLogControlLock); + } + } + else + { + LWLockAcquire(CLogControlLock, LW_EXCLUSIVE); + + TransactionIdSetPageStatusInternal(xid, nsubxids, subxids, status, lsn, pageno); + + LWLockRelease(CLogControlLock); + } +} + +/* + * Record the final state of transaction entry in the commit log + * + * We don't do any locking here; caller must handle that. + */ +static void +TransactionIdSetPageStatusInternal(TransactionId xid, int nsubxids, + TransactionId *subxids, XidStatus status, + XLogRecPtr lsn, int pageno) { int slotno; int i; @@ -262,8 +336,6 @@ TransactionIdSetPageStatus(TransactionId xid, int nsubxids, status == TRANSACTION_STATUS_ABORTED || (status == TRANSACTION_STATUS_SUB_COMMITTED && !TransactionIdIsValid(xid))); - LWLockAcquire(CLogControlLock, LW_EXCLUSIVE); - /* * 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 @@ -310,8 +382,188 @@ TransactionIdSetPageStatus(TransactionId xid, int nsubxids, } ClogCtl->shared->page_dirty[slotno] = true; +} + +/* + * 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 perform TransactionIdSetPageStatusInternal + * 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. + * + * Returns true, if transaction status is updated in clog page, else return + * false. + */ +static bool +TransactionGroupUpdateXidStatus(TransactionId xid, XidStatus status, + XLogRecPtr lsn, int pageno) +{ + volatile PROC_HDR *procglobal = ProcGlobal; + PGPROC *proc = MyProc; + uint32 nextidx; + uint32 wakeidx; + int extraWaits = -1; + int group_no = -1; + + /* 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->updateXidStatus = true; + proc->memberXid = xid; + proc->memberXidstatus = status; + proc->clogPage = pageno; + proc->asyncCommitLsn = lsn; + while (true) + { + PGPROC *initial_proc; + bool add_proc_to_group = false; + + /* + * Add the proc to any existing list of groups if the clog page where we + * need to update the current transaction status matches with the lists + * group leaders' clog page or create a new group if any of the slot for + * groups is available. We need multiple groups to ensure that we can + * use group tansaction status update mechanism when transactions are + * operating on different pages. It is not realistic to assume that + * transactions that are 32768 transactionids apart will arrive at the + * same time, however it is quite possible that transactions that + * belong to clog page boundaries could arrive at same time. + */ + for (group_no = 0; group_no < NUM_GROUPS; group_no++) + { + nextidx = pg_atomic_read_u32(&procglobal->firstupdateXidStatusElem[group_no]); + + if (nextidx != INVALID_PGPROCNO) + { + initial_proc = &ProcGlobal->allProcs[nextidx]; + + if (initial_proc->clogPage == proc->clogPage) + { + add_proc_to_group = true; + break; + } + } + else + { + add_proc_to_group = true; + break; + } + } + + if (!add_proc_to_group) + return false; + + pg_atomic_write_u32(&proc->nextupdateXidStatusElem, nextidx); + + if (pg_atomic_compare_exchange_u32(&procglobal->firstupdateXidStatusElem[group_no], + &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) + { + /* Sleep until the leader updates our XID status. */ + for (;;) + { + /* acts as a read barrier */ + PGSemaphoreLock(&proc->sem); + if (!proc->updateXidStatus) + break; + extraWaits++; + } + + Assert(pg_atomic_read_u32(&proc->nextupdateXidStatusElem) == INVALID_PGPROCNO); + + /* Fix semaphore count for any absorbed wakeups */ + while (extraWaits-- > 0) + PGSemaphoreUnlock(&proc->sem); + return true; + } + + /* 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. + */ + while (true) + { + nextidx = pg_atomic_read_u32(&procglobal->firstupdateXidStatusElem[group_no]); + if (pg_atomic_compare_exchange_u32(&procglobal->firstupdateXidStatusElem[group_no], + &nextidx, + INVALID_PGPROCNO)) + break; + } + + /* 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]; + + /* + * Overflowed transactions should not use group XID status update + * mechanism. + */ + Assert(!pgxact->overflowed); + + TransactionIdSetPageStatusInternal(proc->memberXid, + pgxact->nxids, + proc->subxids.xids, + proc->memberXidstatus, + proc->asyncCommitLsn, + proc->clogPage); + + /* Move to next proc in list. */ + nextidx = pg_atomic_read_u32(&proc->nextupdateXidStatusElem); + } + + /* 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. The system calls we need to perform to wake other processes + * up are probably much slower than the simple memory writes we did while + * holding the lock. + */ + while (wakeidx != INVALID_PGPROCNO) + { + PGPROC *proc = &ProcGlobal->allProcs[wakeidx]; + + wakeidx = pg_atomic_read_u32(&proc->nextupdateXidStatusElem); + pg_atomic_write_u32(&proc->nextupdateXidStatusElem, INVALID_PGPROCNO); + + /* ensure all previous writes are visible before follower continues. */ + pg_write_barrier(); + + proc->updateXidStatus = false; + + if (proc != MyProc) + PGSemaphoreUnlock(&proc->sem); + } + + return true; } /* diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c index 8c47e0f..471280f 100644 --- a/src/backend/access/transam/twophase.c +++ b/src/backend/access/transam/twophase.c @@ -828,6 +828,17 @@ TwoPhaseGetDummyProc(TransactionId xid) return &ProcGlobal->allProcs[gxact->pgprocno]; } +/* + * IsGXactActive + * Return true if there is a Global transaction entry currently + * locked by us. + */ +bool +IsGXactActive(void) +{ + return MyLockedGxact ? true : false; +} + /************************************************************************/ /* State file support */ /************************************************************************/ diff --git a/src/backend/storage/lmgr/proc.c b/src/backend/storage/lmgr/proc.c index bb10c1b..69fabb8 100644 --- a/src/backend/storage/lmgr/proc.c +++ b/src/backend/storage/lmgr/proc.c @@ -183,6 +183,9 @@ InitProcGlobal(void) ProcGlobal->checkpointerLatch = NULL; pg_atomic_init_u32(&ProcGlobal->firstClearXidElem, INVALID_PGPROCNO); + for (i = 0; i < NUM_GROUPS; i++) + pg_atomic_init_u32(&ProcGlobal->firstupdateXidStatusElem[i], INVALID_PGPROCNO); + /* * Create and initialize all the PGPROC structures we'll need. There are * five separate consumers: (1) normal backends, (2) autovacuum workers @@ -397,6 +400,14 @@ InitProcess(void) MyProc->backendLatestXid = InvalidTransactionId; pg_atomic_init_u32(&MyProc->nextClearXidElem, INVALID_PGPROCNO); + /* Initialize fields for group transaction status update. */ + MyProc->updateXidStatus = false; + MyProc->memberXid = InvalidTransactionId; + MyProc->memberXidstatus = TRANSACTION_STATUS_IN_PROGRESS; + MyProc->clogPage = -1; + MyProc->asyncCommitLsn = InvalidXLogRecPtr; + pg_atomic_init_u32(&MyProc->nextupdateXidStatusElem, INVALID_PGPROCNO); + /* * Acquire ownership of the PGPROC's latch, so that we can use WaitLatch * on it. That allows us to repoint the process latch, which so far diff --git a/src/include/access/twophase.h b/src/include/access/twophase.h index 71ddff7..feec01d 100644 --- a/src/include/access/twophase.h +++ b/src/include/access/twophase.h @@ -35,6 +35,7 @@ extern void PostPrepare_Twophase(void); extern PGPROC *TwoPhaseGetDummyProc(TransactionId xid); extern BackendId TwoPhaseGetDummyBackendId(TransactionId xid); +extern bool IsGXactActive(void); extern GlobalTransaction MarkAsPreparing(TransactionId xid, const char *gid, TimestampTz prepared_at, diff --git a/src/include/storage/proc.h b/src/include/storage/proc.h index 3d68017..bab0a59 100644 --- a/src/include/storage/proc.h +++ b/src/include/storage/proc.h @@ -14,6 +14,7 @@ #ifndef _PROC_H_ #define _PROC_H_ +#include "access/clog.h" #include "access/xlogdefs.h" #include "lib/ilist.h" #include "storage/latch.h" @@ -65,6 +66,12 @@ struct XidCache #define INVALID_PGPROCNO PG_INT32_MAX /* + * Allow small number of groups for group transaction status update. See + * comments in function TransactionGroupUpdateXidStatus. + */ +#define NUM_GROUPS 2 + +/* * Each backend has a PGPROC struct in shared memory. There is also a list of * currently-unused PGPROC structs that will be reallocated to new backends. * @@ -146,6 +153,14 @@ struct PGPROC pg_atomic_uint32 nextClearXidElem; TransactionId backendLatestXid; + /* Support for group transaction status update. */ + bool updateXidStatus; + pg_atomic_uint32 nextupdateXidStatusElem; + TransactionId memberXid; + XidStatus memberXidstatus; + int clogPage; + XLogRecPtr asyncCommitLsn; + /* Per-backend LWLock. Protects fields below. */ LWLock *backendLock; /* protects the fields below */ @@ -209,6 +224,8 @@ typedef struct PROC_HDR PGPROC *bgworkerFreeProcs; /* First pgproc waiting for group XID clear */ pg_atomic_uint32 firstClearXidElem; + /* First pgproc waiting for group transaction status update */ + pg_atomic_uint32 firstupdateXidStatusElem[NUM_GROUPS]; /* WALWriter process's latch */ Latch *walwriterLatch; /* Checkpointer process's latch */