Reduce ProcArrayLock contention by removing backends in batches.

When a write transaction commits, it must clear its XID advertised via
the ProcArray, which requires that we hold ProcArrayLock in exclusive
mode in order to prevent concurrent processes running GetSnapshotData
from seeing inconsistent results.  When many processes try to commit
at once, ProcArrayLock must change hands repeatedly, with each
concurrent process trying to commit waking up to acquire the lock in
turn.  To make things more efficient, when more than one backend is
trying to commit a write transaction at the same time, have just one
of them acquire ProcArrayLock in exclusive mode and clear the XIDs of
all processes in the group.  Benchmarking reveals that this is much
more efficient at very high client counts.

Amit Kapila, heavily revised by me, with some review also from Pavan
Deolasee.

src/backend/access/transam/README

@@ -252,6 +252,9 @@ implementation of this is that GetSnapshotData takes the ProcArrayLock in
 shared mode (so that multiple backends can take snapshots in parallel),
 but ProcArrayEndTransaction must take the ProcArrayLock in exclusive mode
 while clearing MyPgXact->xid at transaction end (either commit or abort).
+(To reduce context switching, when multiple transactions commit nearly
+simultaneously, we have one backend take ProcArrayLock and clear the XIDs
+of multiple processes at once.)
 
 ProcArrayEndTransaction also holds the lock while advancing the shared
 latestCompletedXid variable.  This allows GetSnapshotData to use

src/backend/storage/ipc/procarray.c

@@ -167,6 +167,9 @@ static int KnownAssignedXidsGetAndSetXmin(TransactionId *xarray,
 static TransactionId KnownAssignedXidsGetOldestXmin(void);
 static void KnownAssignedXidsDisplay(int trace_level);
 static void KnownAssignedXidsReset(void);
+static inline void ProcArrayEndTransactionInternal(PGPROC *proc,
+								PGXACT *pgxact, TransactionId latestXid);
+static void ProcArrayGroupClearXid(PGPROC *proc, TransactionId latestXid);
 
 /*
  * Report shared-memory space needed by CreateSharedProcArray.
@@ -399,8 +402,49 @@ ProcArrayEndTransaction(PGPROC *proc, TransactionId latestXid)
 		 */
 		Assert(TransactionIdIsValid(allPgXact[proc->pgprocno].xid));
 
-		LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
+		/*
+		 * If we can immediately acquire ProcArrayLock, we clear our own XID
+		 * and release the lock.  If not, use group XID clearing to improve
+		 * efficiency.
+		 */
+		if (LWLockConditionalAcquire(ProcArrayLock, LW_EXCLUSIVE))
+		{
+			ProcArrayEndTransactionInternal(proc, pgxact, latestXid);
+			LWLockRelease(ProcArrayLock);
+		}
+		else
+			ProcArrayGroupClearXid(proc, latestXid);
+	}
+	else
+	{
+		/*
+		 * If we have no XID, we don't need to lock, since we won't affect
+		 * anyone else's calculation of a snapshot.  We might change their
+		 * estimate of global xmin, but that's OK.
+		 */
+		Assert(!TransactionIdIsValid(allPgXact[proc->pgprocno].xid));
+
+		proc->lxid = InvalidLocalTransactionId;
+		pgxact->xmin = InvalidTransactionId;
+		/* must be cleared with xid/xmin: */
+		pgxact->vacuumFlags &= ~PROC_VACUUM_STATE_MASK;
+		pgxact->delayChkpt = false;		/* be sure this is cleared in abort */
+		proc->recoveryConflictPending = false;
 
+		Assert(pgxact->nxids == 0);
+		Assert(pgxact->overflowed == false);
+	}
+}
+
+/*
+ * Mark a write transaction as no longer running.
+ *
+ * We don't do any locking here; caller must handle that.
+ */
+static inline void
+ProcArrayEndTransactionInternal(PGPROC *proc, PGXACT *pgxact,
+								TransactionId latestXid)
+{
 	pgxact->xid = InvalidTransactionId;
 	proc->lxid = InvalidLocalTransactionId;
 	pgxact->xmin = InvalidTransactionId;
@@ -417,30 +461,112 @@ ProcArrayEndTransaction(PGPROC *proc, TransactionId latestXid)
 	if (TransactionIdPrecedes(ShmemVariableCache->latestCompletedXid,
 							  latestXid))
 		ShmemVariableCache->latestCompletedXid = latestXid;
+}
 
-		LWLockRelease(ProcArrayLock);
+/*
+ * ProcArrayGroupClearXid -- group XID clearing
+ *
+ * When we cannot immediately acquire ProcArrayLock in exclusive mode at
+ * commit time, add ourselves to a list of processes that need their XIDs
+ * cleared.  The first process to add itself to the list will acquire
+ * ProcArrayLock in exclusive mode and perform ProcArrayEndTransactionInternal
+ * on behalf of all group members.  This avoids a great deal of context
+ * switching when many processes are trying to commit at once, since the lock
+ * only needs to be handed from the last share-locker to one process waiting
+ * for the exclusive lock, rather than to each one in turn.
+ */
+static void
+ProcArrayGroupClearXid(PGPROC *proc, TransactionId latestXid)
+{
+	volatile PROC_HDR *procglobal = ProcGlobal;
+	uint32		nextidx;
+	uint32		wakeidx;
+	int			extraWaits = -1;
+
+	/* We should definitely have an XID to clear. */
+	Assert(TransactionIdIsValid(allPgXact[proc->pgprocno].xid));
+
+	/* Add ourselves to the list of processes needing a group XID clear. */
+	proc->backendLatestXid = latestXid;
+	while (true)
+	{
+		nextidx = pg_atomic_read_u32(&procglobal->nextClearXidElem);
+		pg_atomic_write_u32(&proc->nextClearXidElem, nextidx);
+
+		if (pg_atomic_compare_exchange_u32(&procglobal->nextClearXidElem,
+										   &nextidx,
+										   (uint32) proc->pgprocno))
+			break;
 	}
-	else
+
+	/* If the list was not empty, the leader will clear our XID. */
+	if (nextidx != INVALID_PGPROCNO)
 	{
+		/* Sleep until the leader clears our XID. */
+		while (pg_atomic_read_u32(&proc->nextClearXidElem) != INVALID_PGPROCNO)
+		{
+			extraWaits++;
+			PGSemaphoreLock(&proc->sem);
+		}
+
+		/* Fix semaphore count for any absorbed wakeups */
+		while (extraWaits-- > 0)
+			PGSemaphoreUnlock(&proc->sem);
+		return;
+	}
+
+	/* We are the leader.  Acquire the lock on behalf of everyone. */
+	LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
+
 	/*
-		 * If we have no XID, we don't need to lock, since we won't affect
-		 * anyone else's calculation of a snapshot.  We might change their
-		 * estimate of global xmin, but that's OK.
+	 * Now that we've got the lock, clear the list of processes waiting for
+	 * group XID clearing, saving a pointer to the head of the list.
 	 */
-		Assert(!TransactionIdIsValid(allPgXact[proc->pgprocno].xid));
+	while (true)
+	{
+		nextidx = pg_atomic_read_u32(&procglobal->nextClearXidElem);
+		if (pg_atomic_compare_exchange_u32(&procglobal->nextClearXidElem,
+										   &nextidx,
+										   INVALID_PGPROCNO))
+			break;
+	}
 
-		proc->lxid = InvalidLocalTransactionId;
-		pgxact->xmin = InvalidTransactionId;
-		/* must be cleared with xid/xmin: */
-		pgxact->vacuumFlags &= ~PROC_VACUUM_STATE_MASK;
-		pgxact->delayChkpt = false;		/* be sure this is cleared in abort */
-		proc->recoveryConflictPending = false;
+	/* Remember head of list so we can perform wakeups after dropping lock. */
+	wakeidx = nextidx;
 
-		Assert(pgxact->nxids == 0);
-		Assert(pgxact->overflowed == false);
+	/* Walk the list and clear all XIDs. */
+	while (nextidx != INVALID_PGPROCNO)
+	{
+		PGPROC	*proc = &allProcs[nextidx];
+		PGXACT	*pgxact = &allPgXact[nextidx];
+
+		ProcArrayEndTransactionInternal(proc, pgxact, proc->backendLatestXid);
+
+		/* Move to next proc in list. */
+		nextidx = pg_atomic_read_u32(&proc->nextClearXidElem);
 	}
-}
 
+	/* We're done with the lock now. */
+	LWLockRelease(ProcArrayLock);
+
+	/*
+	 * 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 = &allProcs[wakeidx];
+
+		wakeidx = pg_atomic_read_u32(&proc->nextClearXidElem);
+		pg_atomic_write_u32(&proc->nextClearXidElem, INVALID_PGPROCNO);
+
+		if (proc != MyProc)
+			PGSemaphoreUnlock(&proc->sem);
+	}
+}
 
 /*
  * ProcArrayClearTransaction -- clear the transaction fields

src/backend/storage/lmgr/proc.c

@@ -181,6 +181,7 @@ InitProcGlobal(void)
 	ProcGlobal->startupBufferPinWaitBufId = -1;
 	ProcGlobal->walwriterLatch = NULL;
 	ProcGlobal->checkpointerLatch = NULL;
+	pg_atomic_init_u32(&ProcGlobal->nextClearXidElem, INVALID_PGPROCNO);
 
 	/*
 	 * Create and initialize all the PGPROC structures we'll need.  There are
@@ -393,6 +394,10 @@ InitProcess(void)
 	MyProc->syncRepState = SYNC_REP_NOT_WAITING;
 	SHMQueueElemInit(&(MyProc->syncRepLinks));
 
+	/* Initialize fields for group XID clearing. */
+	MyProc->backendLatestXid = InvalidTransactionId;
+	pg_atomic_init_u32(&MyProc->nextClearXidElem, 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

src/include/storage/proc.h

@@ -58,6 +58,12 @@ struct XidCache
  */
 #define		FP_LOCK_SLOTS_PER_BACKEND 16
 
+/*
+ * An invalid pgprocno.  Must be larger than the maximum number of PGPROC
+ * structures we could possibly have.  See comments for MAX_BACKENDS.
+ */
+#define INVALID_PGPROCNO		PG_INT32_MAX
+
 /*
  * 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.
@@ -135,6 +141,10 @@ struct PGPROC
 
 	struct XidCache subxids;	/* cache for subtransaction XIDs */
 
+	/* Support for group XID clearing. */
+	volatile pg_atomic_uint32	nextClearXidElem;
+	TransactionId	backendLatestXid;
+
 	/* Per-backend LWLock.  Protects fields below. */
 	LWLock	   *backendLock;	/* protects the fields below */
 
@@ -196,6 +206,8 @@ typedef struct PROC_HDR
 	PGPROC	   *autovacFreeProcs;
 	/* Head of list of bgworker free PGPROC structures */
 	PGPROC	   *bgworkerFreeProcs;
+	/* First pgproc waiting for group XID clear */
+	volatile pg_atomic_uint32 nextClearXidElem;
 	/* WALWriter process's latch */
 	Latch	   *walwriterLatch;
 	/* Checkpointer process's latch */