group_update_clog_v3.patch

diff --git a/src/backend/access/transam/clog.c b/src/backend/access/transam/clog.c
index ea83655..486abb2 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,164 @@ 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 thransaction 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;
+
+	/* 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;
+
+		nextidx = pg_atomic_read_u32(&procglobal->firstupdateXidStatusElem);
+
+		/*
+		 * 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.
+		 */
+		if (nextidx != INVALID_PGPROCNO)
+		{
+			initial_proc = &ProcGlobal->allProcs[nextidx];
+
+			if (initial_proc->clogPage != proc->clogPage)
+				return false;
+		}
+		pg_atomic_write_u32(&proc->nextupdateXidStatusElem, nextidx);
+
+		if (pg_atomic_compare_exchange_u32(&procglobal->firstupdateXidStatusElem,
+										   &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);
+		if (pg_atomic_compare_exchange_u32(&procglobal->firstupdateXidStatusElem,
+										   &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..a2ecbff 100644
--- a/src/backend/storage/lmgr/proc.c
+++ b/src/backend/storage/lmgr/proc.c
@@ -182,6 +182,7 @@ InitProcGlobal(void)
 	ProcGlobal->walwriterLatch = NULL;
 	ProcGlobal->checkpointerLatch = NULL;
 	pg_atomic_init_u32(&ProcGlobal->firstClearXidElem, INVALID_PGPROCNO);
+	pg_atomic_init_u32(&ProcGlobal->firstupdateXidStatusElem, INVALID_PGPROCNO);
 
 	/*
 	 * Create and initialize all the PGPROC structures we'll need.  There are
@@ -397,6 +398,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..f71b261 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"
@@ -146,6 +147,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 +218,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;
 	/* WALWriter process's latch */
 	Latch	   *walwriterLatch;
 	/* Checkpointer process's latch */