procarraylock-v1.patch

diff --git a/src/backend/commands/analyze.c b/src/backend/commands/analyze.c
index 32985a4..d6bba6f 100644
--- a/src/backend/commands/analyze.c
+++ b/src/backend/commands/analyze.c
@@ -40,6 +40,7 @@
 #include "storage/lmgr.h"
 #include "storage/proc.h"
 #include "storage/procarray.h"
+#include "storage/procarraylock.h"
 #include "utils/acl.h"
 #include "utils/attoptcache.h"
 #include "utils/datum.h"
@@ -222,9 +223,9 @@ analyze_rel(Oid relid, VacuumStmt *vacstmt, BufferAccessStrategy bstrategy)
 	/*
 	 * OK, let's do it.  First let other backends know I'm in ANALYZE.
 	 */
-	LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
+	ProcArrayLockAcquire(PAL_EXCLUSIVE);
 	MyProc->vacuumFlags |= PROC_IN_ANALYZE;
-	LWLockRelease(ProcArrayLock);
+	ProcArrayLockRelease();
 
 	/*
 	 * Do the normal non-recursive ANALYZE.
@@ -249,9 +250,9 @@ analyze_rel(Oid relid, VacuumStmt *vacstmt, BufferAccessStrategy bstrategy)
 	 * Reset my PGPROC flag.  Note: we need this here, and not in vacuum_rel,
 	 * because the vacuum flag is cleared by the end-of-xact code.
 	 */
-	LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
+	ProcArrayLockAcquire(PAL_EXCLUSIVE);
 	MyProc->vacuumFlags &= ~PROC_IN_ANALYZE;
-	LWLockRelease(ProcArrayLock);
+	ProcArrayLockRelease();
 }
 
 /*
diff --git a/src/backend/commands/vacuum.c b/src/backend/commands/vacuum.c
index f42504c..823dab9 100644
--- a/src/backend/commands/vacuum.c
+++ b/src/backend/commands/vacuum.c
@@ -39,6 +39,7 @@
 #include "storage/lmgr.h"
 #include "storage/proc.h"
 #include "storage/procarray.h"
+#include "storage/procarraylock.h"
 #include "utils/acl.h"
 #include "utils/fmgroids.h"
 #include "utils/guc.h"
@@ -892,11 +893,11 @@ vacuum_rel(Oid relid, VacuumStmt *vacstmt, bool do_toast, bool for_wraparound)
 		 * MyProc->xid/xmin, else OldestXmin might appear to go backwards,
 		 * which is probably Not Good.
 		 */
-		LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
+		ProcArrayLockAcquire(PAL_EXCLUSIVE);
 		MyProc->vacuumFlags |= PROC_IN_VACUUM;
 		if (for_wraparound)
 			MyProc->vacuumFlags |= PROC_VACUUM_FOR_WRAPAROUND;
-		LWLockRelease(ProcArrayLock);
+		ProcArrayLockRelease();
 	}
 
 	/*
diff --git a/src/backend/storage/ipc/procarray.c b/src/backend/storage/ipc/procarray.c
index 1a48485..39c5080 100644
--- a/src/backend/storage/ipc/procarray.c
+++ b/src/backend/storage/ipc/procarray.c
@@ -52,6 +52,7 @@
 #include "access/twophase.h"
 #include "miscadmin.h"
 #include "storage/procarray.h"
+#include "storage/procarraylock.h"
 #include "storage/spin.h"
 #include "utils/builtins.h"
 #include "utils/snapmgr.h"
@@ -254,7 +255,7 @@ ProcArrayAdd(PGPROC *proc)
 {
 	ProcArrayStruct *arrayP = procArray;
 
-	LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
+	ProcArrayLockAcquire(PAL_EXCLUSIVE);
 
 	if (arrayP->numProcs >= arrayP->maxProcs)
 	{
@@ -263,7 +264,7 @@ ProcArrayAdd(PGPROC *proc)
 		 * fixed supply of PGPROC structs too, and so we should have failed
 		 * earlier.)
 		 */
-		LWLockRelease(ProcArrayLock);
+		ProcArrayLockRelease();
 		ereport(FATAL,
 				(errcode(ERRCODE_TOO_MANY_CONNECTIONS),
 				 errmsg("sorry, too many clients already")));
@@ -272,7 +273,7 @@ ProcArrayAdd(PGPROC *proc)
 	arrayP->procs[arrayP->numProcs] = proc;
 	arrayP->numProcs++;
 
-	LWLockRelease(ProcArrayLock);
+	ProcArrayLockRelease();
 }
 
 /*
@@ -297,7 +298,7 @@ ProcArrayRemove(PGPROC *proc, TransactionId latestXid)
 		DisplayXidCache();
 #endif
 
-	LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
+	ProcArrayLockAcquire(PAL_EXCLUSIVE);
 
 	if (TransactionIdIsValid(latestXid))
 	{
@@ -321,13 +322,13 @@ ProcArrayRemove(PGPROC *proc, TransactionId latestXid)
 			arrayP->procs[index] = arrayP->procs[arrayP->numProcs - 1];
 			arrayP->procs[arrayP->numProcs - 1] = NULL; /* for debugging */
 			arrayP->numProcs--;
-			LWLockRelease(ProcArrayLock);
+			ProcArrayLockRelease();
 			return;
 		}
 	}
 
 	/* Ooops */
-	LWLockRelease(ProcArrayLock);
+	ProcArrayLockRelease();
 
 	elog(LOG, "failed to find proc %p in ProcArray", proc);
 }
@@ -351,54 +352,15 @@ ProcArrayEndTransaction(PGPROC *proc, TransactionId latestXid)
 {
 	if (TransactionIdIsValid(latestXid))
 	{
-		/*
-		 * We must lock ProcArrayLock while clearing proc->xid, so that we do
-		 * not exit the set of "running" transactions while someone else is
-		 * taking a snapshot.  See discussion in
-		 * src/backend/access/transam/README.
-		 */
-		Assert(TransactionIdIsValid(proc->xid));
-
-		LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
-
-		proc->xid = InvalidTransactionId;
-		proc->lxid = InvalidLocalTransactionId;
-		proc->xmin = InvalidTransactionId;
-		/* must be cleared with xid/xmin: */
-		proc->vacuumFlags &= ~PROC_VACUUM_STATE_MASK;
-		proc->inCommit = false; /* be sure this is cleared in abort */
-		proc->recoveryConflictPending = false;
-
-		/* Clear the subtransaction-XID cache too while holding the lock */
-		proc->subxids.nxids = 0;
-		proc->subxids.overflowed = false;
-
-		/* Also advance global latestCompletedXid while holding the lock */
-		if (TransactionIdPrecedes(ShmemVariableCache->latestCompletedXid,
-								  latestXid))
-			ShmemVariableCache->latestCompletedXid = latestXid;
-
-		LWLockRelease(ProcArrayLock);
+		Assert(proc == MyProc);
+		ProcArrayLockClearTransaction(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(proc->xid));
-
-		proc->lxid = InvalidLocalTransactionId;
 		proc->xmin = InvalidTransactionId;
-		/* must be cleared with xid/xmin: */
-		proc->vacuumFlags &= ~PROC_VACUUM_STATE_MASK;
-		proc->inCommit = false; /* be sure this is cleared in abort */
-		proc->recoveryConflictPending = false;
 
-		Assert(proc->subxids.nxids == 0);
-		Assert(proc->subxids.overflowed == false);
-	}
+	proc->lxid = InvalidLocalTransactionId;
+	proc->inCommit = false; /* be sure this is cleared in abort */
+	proc->recoveryConflictPending = false;
 }
 
 
@@ -528,7 +490,7 @@ ProcArrayApplyRecoveryInfo(RunningTransactions running)
 	/*
 	 * Nobody else is running yet, but take locks anyhow
 	 */
-	LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
+	ProcArrayLockAcquire(PAL_EXCLUSIVE);
 
 	/*
 	 * KnownAssignedXids is sorted so we cannot just add the xids, we have to
@@ -635,7 +597,7 @@ ProcArrayApplyRecoveryInfo(RunningTransactions running)
 	Assert(TransactionIdIsNormal(ShmemVariableCache->latestCompletedXid));
 	Assert(TransactionIdIsValid(ShmemVariableCache->nextXid));
 
-	LWLockRelease(ProcArrayLock);
+	ProcArrayLockRelease();
 
 	KnownAssignedXidsDisplay(trace_recovery(DEBUG3));
 	if (standbyState == STANDBY_SNAPSHOT_READY)
@@ -690,7 +652,7 @@ ProcArrayApplyXidAssignment(TransactionId topxid,
 	/*
 	 * Uses same locking as transaction commit
 	 */
-	LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
+	ProcArrayLockAcquire(PAL_EXCLUSIVE);
 
 	/*
 	 * Remove subxids from known-assigned-xacts.
@@ -703,7 +665,7 @@ ProcArrayApplyXidAssignment(TransactionId topxid,
 	if (TransactionIdPrecedes(procArray->lastOverflowedXid, max_xid))
 		procArray->lastOverflowedXid = max_xid;
 
-	LWLockRelease(ProcArrayLock);
+	ProcArrayLockRelease();
 }
 
 /*
@@ -795,7 +757,7 @@ TransactionIdIsInProgress(TransactionId xid)
 					 errmsg("out of memory")));
 	}
 
-	LWLockAcquire(ProcArrayLock, LW_SHARED);
+	ProcArrayLockAcquire(PAL_SHARED);
 
 	/*
 	 * Now that we have the lock, we can check latestCompletedXid; if the
@@ -803,7 +765,7 @@ TransactionIdIsInProgress(TransactionId xid)
 	 */
 	if (TransactionIdPrecedes(ShmemVariableCache->latestCompletedXid, xid))
 	{
-		LWLockRelease(ProcArrayLock);
+		ProcArrayLockRelease();
 		xc_by_latest_xid_inc();
 		return true;
 	}
@@ -829,7 +791,7 @@ TransactionIdIsInProgress(TransactionId xid)
 		 */
 		if (TransactionIdEquals(pxid, xid))
 		{
-			LWLockRelease(ProcArrayLock);
+			ProcArrayLockRelease();
 			xc_by_main_xid_inc();
 			return true;
 		}
@@ -851,7 +813,7 @@ TransactionIdIsInProgress(TransactionId xid)
 
 			if (TransactionIdEquals(cxid, xid))
 			{
-				LWLockRelease(ProcArrayLock);
+				ProcArrayLockRelease();
 				xc_by_child_xid_inc();
 				return true;
 			}
@@ -879,7 +841,7 @@ TransactionIdIsInProgress(TransactionId xid)
 
 		if (KnownAssignedXidExists(xid))
 		{
-			LWLockRelease(ProcArrayLock);
+			ProcArrayLockRelease();
 			xc_by_known_assigned_inc();
 			return true;
 		}
@@ -895,7 +857,7 @@ TransactionIdIsInProgress(TransactionId xid)
 			nxids = KnownAssignedXidsGet(xids, xid);
 	}
 
-	LWLockRelease(ProcArrayLock);
+	ProcArrayLockRelease();
 
 	/*
 	 * If none of the relevant caches overflowed, we know the Xid is not
@@ -961,7 +923,7 @@ TransactionIdIsActive(TransactionId xid)
 	if (TransactionIdPrecedes(xid, RecentXmin))
 		return false;
 
-	LWLockAcquire(ProcArrayLock, LW_SHARED);
+	ProcArrayLockAcquire(PAL_SHARED);
 
 	for (i = 0; i < arrayP->numProcs; i++)
 	{
@@ -983,7 +945,7 @@ TransactionIdIsActive(TransactionId xid)
 		}
 	}
 
-	LWLockRelease(ProcArrayLock);
+	ProcArrayLockRelease();
 
 	return result;
 }
@@ -1046,7 +1008,7 @@ GetOldestXmin(bool allDbs, bool ignoreVacuum)
 	/* Cannot look for individual databases during recovery */
 	Assert(allDbs || !RecoveryInProgress());
 
-	LWLockAcquire(ProcArrayLock, LW_SHARED);
+	ProcArrayLockAcquire(PAL_SHARED);
 
 	/*
 	 * We initialize the MIN() calculation with latestCompletedXid + 1. This
@@ -1099,7 +1061,7 @@ GetOldestXmin(bool allDbs, bool ignoreVacuum)
 		 */
 		TransactionId kaxmin = KnownAssignedXidsGetOldestXmin();
 
-		LWLockRelease(ProcArrayLock);
+		ProcArrayLockRelease();
 
 		if (TransactionIdIsNormal(kaxmin) &&
 			TransactionIdPrecedes(kaxmin, result))
@@ -1110,7 +1072,7 @@ GetOldestXmin(bool allDbs, bool ignoreVacuum)
 		/*
 		 * No other information needed, so release the lock immediately.
 		 */
-		LWLockRelease(ProcArrayLock);
+		ProcArrayLockRelease();
 
 		/*
 		 * Compute the cutoff XID by subtracting vacuum_defer_cleanup_age,
@@ -1239,7 +1201,7 @@ GetSnapshotData(Snapshot snapshot)
 	 * It is sufficient to get shared lock on ProcArrayLock, even if we are
 	 * going to set MyProc->xmin.
 	 */
-	LWLockAcquire(ProcArrayLock, LW_SHARED);
+	ProcArrayLockAcquire(PAL_SHARED);
 
 	/* xmax is always latestCompletedXid + 1 */
 	xmax = ShmemVariableCache->latestCompletedXid;
@@ -1375,7 +1337,7 @@ GetSnapshotData(Snapshot snapshot)
 	if (!TransactionIdIsValid(MyProc->xmin))
 		MyProc->xmin = TransactionXmin = xmin;
 
-	LWLockRelease(ProcArrayLock);
+	ProcArrayLockRelease();
 
 	/*
 	 * Update globalxmin to include actual process xids.  This is a slightly
@@ -1432,7 +1394,7 @@ ProcArrayInstallImportedXmin(TransactionId xmin, TransactionId sourcexid)
 		return false;
 
 	/* Get lock so source xact can't end while we're doing this */
-	LWLockAcquire(ProcArrayLock, LW_SHARED);
+	ProcArrayLockAcquire(PAL_SHARED);
 
 	for (index = 0; index < arrayP->numProcs; index++)
 	{
@@ -1476,7 +1438,7 @@ ProcArrayInstallImportedXmin(TransactionId xmin, TransactionId sourcexid)
 		break;
 	}
 
-	LWLockRelease(ProcArrayLock);
+	ProcArrayLockRelease();
 
 	return result;
 }
@@ -1550,7 +1512,7 @@ GetRunningTransactionData(void)
 	 * Ensure that no xids enter or leave the procarray while we obtain
 	 * snapshot.
 	 */
-	LWLockAcquire(ProcArrayLock, LW_SHARED);
+	ProcArrayLockAcquire(PAL_SHARED);
 	LWLockAcquire(XidGenLock, LW_SHARED);
 
 	latestCompletedXid = ShmemVariableCache->latestCompletedXid;
@@ -1611,7 +1573,7 @@ GetRunningTransactionData(void)
 	CurrentRunningXacts->latestCompletedXid = latestCompletedXid;
 
 	/* We don't release XidGenLock here, the caller is responsible for that */
-	LWLockRelease(ProcArrayLock);
+	ProcArrayLockRelease();
 
 	Assert(TransactionIdIsValid(CurrentRunningXacts->nextXid));
 	Assert(TransactionIdIsValid(CurrentRunningXacts->oldestRunningXid));
@@ -1644,7 +1606,7 @@ GetOldestActiveTransactionId(void)
 
 	Assert(!RecoveryInProgress());
 
-	LWLockAcquire(ProcArrayLock, LW_SHARED);
+	ProcArrayLockAcquire(PAL_SHARED);
 
 	oldestRunningXid = ShmemVariableCache->nextXid;
 
@@ -1672,7 +1634,7 @@ GetOldestActiveTransactionId(void)
 		 */
 	}
 
-	LWLockRelease(ProcArrayLock);
+	ProcArrayLockRelease();
 
 	return oldestRunningXid;
 }
@@ -1705,7 +1667,7 @@ GetTransactionsInCommit(TransactionId **xids_p)
 	xids = (TransactionId *) palloc(arrayP->maxProcs * sizeof(TransactionId));
 	nxids = 0;
 
-	LWLockAcquire(ProcArrayLock, LW_SHARED);
+	ProcArrayLockAcquire(PAL_SHARED);
 
 	for (index = 0; index < arrayP->numProcs; index++)
 	{
@@ -1718,7 +1680,7 @@ GetTransactionsInCommit(TransactionId **xids_p)
 			xids[nxids++] = pxid;
 	}
 
-	LWLockRelease(ProcArrayLock);
+	ProcArrayLockRelease();
 
 	*xids_p = xids;
 	return nxids;
@@ -1740,7 +1702,7 @@ HaveTransactionsInCommit(TransactionId *xids, int nxids)
 	ProcArrayStruct *arrayP = procArray;
 	int			index;
 
-	LWLockAcquire(ProcArrayLock, LW_SHARED);
+	ProcArrayLockAcquire(PAL_SHARED);
 
 	for (index = 0; index < arrayP->numProcs; index++)
 	{
@@ -1766,7 +1728,7 @@ HaveTransactionsInCommit(TransactionId *xids, int nxids)
 		}
 	}
 
-	LWLockRelease(ProcArrayLock);
+	ProcArrayLockRelease();
 
 	return result;
 }
@@ -1788,7 +1750,7 @@ BackendPidGetProc(int pid)
 	if (pid == 0)				/* never match dummy PGPROCs */
 		return NULL;
 
-	LWLockAcquire(ProcArrayLock, LW_SHARED);
+	ProcArrayLockAcquire(PAL_SHARED);
 
 	for (index = 0; index < arrayP->numProcs; index++)
 	{
@@ -1801,7 +1763,7 @@ BackendPidGetProc(int pid)
 		}
 	}
 
-	LWLockRelease(ProcArrayLock);
+	ProcArrayLockRelease();
 
 	return result;
 }
@@ -1829,7 +1791,7 @@ BackendXidGetPid(TransactionId xid)
 	if (xid == InvalidTransactionId)	/* never match invalid xid */
 		return 0;
 
-	LWLockAcquire(ProcArrayLock, LW_SHARED);
+	ProcArrayLockAcquire(PAL_SHARED);
 
 	for (index = 0; index < arrayP->numProcs; index++)
 	{
@@ -1842,7 +1804,7 @@ BackendXidGetPid(TransactionId xid)
 		}
 	}
 
-	LWLockRelease(ProcArrayLock);
+	ProcArrayLockRelease();
 
 	return result;
 }
@@ -1897,7 +1859,7 @@ GetCurrentVirtualXIDs(TransactionId limitXmin, bool excludeXmin0,
 	vxids = (VirtualTransactionId *)
 		palloc(sizeof(VirtualTransactionId) * arrayP->maxProcs);
 
-	LWLockAcquire(ProcArrayLock, LW_SHARED);
+	ProcArrayLockAcquire(PAL_SHARED);
 
 	for (index = 0; index < arrayP->numProcs; index++)
 	{
@@ -1933,7 +1895,7 @@ GetCurrentVirtualXIDs(TransactionId limitXmin, bool excludeXmin0,
 		}
 	}
 
-	LWLockRelease(ProcArrayLock);
+	ProcArrayLockRelease();
 
 	*nvxids = count;
 	return vxids;
@@ -1992,7 +1954,7 @@ GetConflictingVirtualXIDs(TransactionId limitXmin, Oid dbOid)
 					 errmsg("out of memory")));
 	}
 
-	LWLockAcquire(ProcArrayLock, LW_SHARED);
+	ProcArrayLockAcquire(PAL_SHARED);
 
 	for (index = 0; index < arrayP->numProcs; index++)
 	{
@@ -2025,7 +1987,7 @@ GetConflictingVirtualXIDs(TransactionId limitXmin, Oid dbOid)
 		}
 	}
 
-	LWLockRelease(ProcArrayLock);
+	ProcArrayLockRelease();
 
 	/* add the terminator */
 	vxids[count].backendId = InvalidBackendId;
@@ -2046,7 +2008,7 @@ CancelVirtualTransaction(VirtualTransactionId vxid, ProcSignalReason sigmode)
 	int			index;
 	pid_t		pid = 0;
 
-	LWLockAcquire(ProcArrayLock, LW_SHARED);
+	ProcArrayLockAcquire(PAL_SHARED);
 
 	for (index = 0; index < arrayP->numProcs; index++)
 	{
@@ -2072,7 +2034,7 @@ CancelVirtualTransaction(VirtualTransactionId vxid, ProcSignalReason sigmode)
 		}
 	}
 
-	LWLockRelease(ProcArrayLock);
+	ProcArrayLockRelease();
 
 	return pid;
 }
@@ -2146,7 +2108,7 @@ CountDBBackends(Oid databaseid)
 	int			count = 0;
 	int			index;
 
-	LWLockAcquire(ProcArrayLock, LW_SHARED);
+	ProcArrayLockAcquire(PAL_SHARED);
 
 	for (index = 0; index < arrayP->numProcs; index++)
 	{
@@ -2159,7 +2121,7 @@ CountDBBackends(Oid databaseid)
 			count++;
 	}
 
-	LWLockRelease(ProcArrayLock);
+	ProcArrayLockRelease();
 
 	return count;
 }
@@ -2175,7 +2137,7 @@ CancelDBBackends(Oid databaseid, ProcSignalReason sigmode, bool conflictPending)
 	pid_t		pid = 0;
 
 	/* tell all backends to die */
-	LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
+	ProcArrayLockAcquire(PAL_EXCLUSIVE);
 
 	for (index = 0; index < arrayP->numProcs; index++)
 	{
@@ -2200,7 +2162,7 @@ CancelDBBackends(Oid databaseid, ProcSignalReason sigmode, bool conflictPending)
 		}
 	}
 
-	LWLockRelease(ProcArrayLock);
+	ProcArrayLockRelease();
 }
 
 /*
@@ -2213,7 +2175,7 @@ CountUserBackends(Oid roleid)
 	int			count = 0;
 	int			index;
 
-	LWLockAcquire(ProcArrayLock, LW_SHARED);
+	ProcArrayLockAcquire(PAL_SHARED);
 
 	for (index = 0; index < arrayP->numProcs; index++)
 	{
@@ -2225,7 +2187,7 @@ CountUserBackends(Oid roleid)
 			count++;
 	}
 
-	LWLockRelease(ProcArrayLock);
+	ProcArrayLockRelease();
 
 	return count;
 }
@@ -2273,7 +2235,7 @@ CountOtherDBBackends(Oid databaseId, int *nbackends, int *nprepared)
 
 		*nbackends = *nprepared = 0;
 
-		LWLockAcquire(ProcArrayLock, LW_SHARED);
+		ProcArrayLockAcquire(PAL_SHARED);
 
 		for (index = 0; index < arrayP->numProcs; index++)
 		{
@@ -2297,7 +2259,7 @@ CountOtherDBBackends(Oid databaseId, int *nbackends, int *nprepared)
 			}
 		}
 
-		LWLockRelease(ProcArrayLock);
+		ProcArrayLockRelease();
 
 		if (!found)
 			return false;		/* no conflicting backends, so done */
@@ -2350,7 +2312,7 @@ XidCacheRemoveRunningXids(TransactionId xid,
 	 * to abort subtransactions, but pending closer analysis we'd best be
 	 * conservative.
 	 */
-	LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
+	ProcArrayLockAcquire(PAL_EXCLUSIVE);
 
 	/*
 	 * Under normal circumstances xid and xids[] will be in increasing order,
@@ -2398,7 +2360,7 @@ XidCacheRemoveRunningXids(TransactionId xid,
 							  latestXid))
 		ShmemVariableCache->latestCompletedXid = latestXid;
 
-	LWLockRelease(ProcArrayLock);
+	ProcArrayLockRelease();
 }
 
 #ifdef XIDCACHE_DEBUG
@@ -2565,7 +2527,7 @@ ExpireTreeKnownAssignedTransactionIds(TransactionId xid, int nsubxids,
 	/*
 	 * Uses same locking as transaction commit
 	 */
-	LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
+	ProcArrayLockAcquire(PAL_EXCLUSIVE);
 
 	KnownAssignedXidsRemoveTree(xid, nsubxids, subxids);
 
@@ -2574,7 +2536,7 @@ ExpireTreeKnownAssignedTransactionIds(TransactionId xid, int nsubxids,
 							  max_xid))
 		ShmemVariableCache->latestCompletedXid = max_xid;
 
-	LWLockRelease(ProcArrayLock);
+	ProcArrayLockRelease();
 }
 
 /*
@@ -2584,9 +2546,9 @@ ExpireTreeKnownAssignedTransactionIds(TransactionId xid, int nsubxids,
 void
 ExpireAllKnownAssignedTransactionIds(void)
 {
-	LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
+	ProcArrayLockAcquire(PAL_EXCLUSIVE);
 	KnownAssignedXidsRemovePreceding(InvalidTransactionId);
-	LWLockRelease(ProcArrayLock);
+	ProcArrayLockRelease();
 }
 
 /*
@@ -2596,9 +2558,9 @@ ExpireAllKnownAssignedTransactionIds(void)
 void
 ExpireOldKnownAssignedTransactionIds(TransactionId xid)
 {
-	LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
+	ProcArrayLockAcquire(PAL_EXCLUSIVE);
 	KnownAssignedXidsRemovePreceding(xid);
-	LWLockRelease(ProcArrayLock);
+	ProcArrayLockRelease();
 }
 
 
@@ -2820,7 +2782,7 @@ KnownAssignedXidsAdd(TransactionId from_xid, TransactionId to_xid,
 	{
 		/* must hold lock to compress */
 		if (!exclusive_lock)
-			LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
+			ProcArrayLockAcquire(PAL_EXCLUSIVE);
 
 		KnownAssignedXidsCompress(true);
 
@@ -2828,7 +2790,7 @@ KnownAssignedXidsAdd(TransactionId from_xid, TransactionId to_xid,
 		/* note: we no longer care about the tail pointer */
 
 		if (!exclusive_lock)
-			LWLockRelease(ProcArrayLock);
+			ProcArrayLockRelease();
 
 		/*
 		 * If it still won't fit then we're out of memory
diff --git a/src/backend/storage/lmgr/Makefile b/src/backend/storage/lmgr/Makefile
index 3730e51..27eaa97 100644
--- a/src/backend/storage/lmgr/Makefile
+++ b/src/backend/storage/lmgr/Makefile
@@ -13,7 +13,7 @@ top_builddir = ../../../..
 include $(top_builddir)/src/Makefile.global
 
 OBJS = flexlock.o lmgr.o lock.o proc.o deadlock.o lwlock.o spin.o s_lock.o \
-	predicate.o
+	procarraylock.o predicate.o
 
 include $(top_srcdir)/src/backend/common.mk
 
diff --git a/src/backend/storage/lmgr/flexlock.c b/src/backend/storage/lmgr/flexlock.c
index 7f657b3..c88bd24 100644
--- a/src/backend/storage/lmgr/flexlock.c
+++ b/src/backend/storage/lmgr/flexlock.c
@@ -30,6 +30,7 @@
 #include "storage/lwlock.h"
 #include "storage/predicate.h"
 #include "storage/proc.h"
+#include "storage/procarraylock.h"
 #include "storage/spin.h"
 #include "utils/elog.h"
 
@@ -177,9 +178,14 @@ CreateFlexLocks(void)
 
 	FlexLockArray = (FlexLockPadded *) ptr;
 
-	/* All of the "fixed" FlexLocks are LWLocks. */
+	/* All of the "fixed" FlexLocks are LWLocks - except ProcArrayLock. */
 	for (id = 0, lock = FlexLockArray; id < NumFixedFlexLocks; id++, lock++)
-		FlexLockInit(&lock->flex, FLEXLOCK_TYPE_LWLOCK);
+	{
+		if (id == ProcArrayLock)
+			FlexLockInit(&lock->flex, FLEXLOCK_TYPE_PROCARRAYLOCK);
+		else
+			FlexLockInit(&lock->flex, FLEXLOCK_TYPE_LWLOCK);
+	}
 
 	/*
 	 * Initialize the dynamic-allocation counter, which is stored just before
@@ -324,13 +330,20 @@ FlexLockReleaseAll(void)
 {
 	while (num_held_flexlocks > 0)
 	{
+		FlexLockId	id;
+		FlexLock   *flex;
+
 		HOLD_INTERRUPTS();		/* match the upcoming RESUME_INTERRUPTS */
 
-		/*
-		 * FLEXTODO: When we have multiple types of flex locks, this will
-		 * need to call the appropriate release function for each lock type.
-		 */
-		LWLockRelease(held_flexlocks[num_held_flexlocks - 1]);
+		id = held_flexlocks[num_held_flexlocks - 1];
+		flex = &FlexLockArray[id].flex;
+		if (flex->locktype == FLEXLOCK_TYPE_LWLOCK)
+			LWLockRelease(id);
+		else
+		{
+			Assert(id == ProcArrayLock);
+			ProcArrayLockRelease();
+		}
 	}
 }
 
diff --git a/src/backend/storage/lmgr/proc.c b/src/backend/storage/lmgr/proc.c
index 57da345..510a4c2 100644
--- a/src/backend/storage/lmgr/proc.c
+++ b/src/backend/storage/lmgr/proc.c
@@ -45,6 +45,7 @@
 #include "storage/pmsignal.h"
 #include "storage/proc.h"
 #include "storage/procarray.h"
+#include "storage/procarraylock.h"
 #include "storage/procsignal.h"
 #include "storage/spin.h"
 #include "utils/timestamp.h"
@@ -1046,7 +1047,7 @@ ProcSleep(LOCALLOCK *locallock, LockMethod lockMethodTable)
 		{
 			PGPROC	   *autovac = GetBlockingAutoVacuumPgproc();
 
-			LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
+			ProcArrayLockAcquire(PAL_EXCLUSIVE);
 
 			/*
 			 * Only do it if the worker is not working to protect against Xid
@@ -1062,7 +1063,7 @@ ProcSleep(LOCALLOCK *locallock, LockMethod lockMethodTable)
 					 pid);
 
 				/* don't hold the lock across the kill() syscall */
-				LWLockRelease(ProcArrayLock);
+				ProcArrayLockRelease();
 
 				/* send the autovacuum worker Back to Old Kent Road */
 				if (kill(pid, SIGINT) < 0)
@@ -1074,7 +1075,7 @@ ProcSleep(LOCALLOCK *locallock, LockMethod lockMethodTable)
 				}
 			}
 			else
-				LWLockRelease(ProcArrayLock);
+				ProcArrayLockRelease();
 
 			/* prevent signal from being resent more than once */
 			allow_autovacuum_cancel = false;
diff --git a/src/backend/storage/lmgr/procarraylock.c b/src/backend/storage/lmgr/procarraylock.c
new file mode 100644
index 0000000..6838ed6
--- /dev/null
+++ b/src/backend/storage/lmgr/procarraylock.c
@@ -0,0 +1,341 @@
+/*-------------------------------------------------------------------------
+ *
+ * procarraylock.c
+ *	  Lock management for the ProcArray
+ *
+ * Because the ProcArray data structure is highly trafficked, it is
+ * critical that mutual exclusion for ProcArray options be as efficient
+ * as possible.  A particular problem is transaction end (commit or abort)
+ * which cannot be done in parallel with snapshot acquisition.  We
+ * therefore include some special hacks to deal with this case efficiently.
+ *
+ * Portions Copyright (c) 1996-2011, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * IDENTIFICATION
+ *	  src/backend/storage/lmgr/procarraylock.c
+ *
+ *-------------------------------------------------------------------------
+ */
+#include "postgres.h"
+
+#include "miscadmin.h"
+#include "pg_trace.h"
+#include "access/transam.h"
+#include "storage/flexlock_internals.h"
+#include "storage/ipc.h"
+#include "storage/procarraylock.h"
+#include "storage/proc.h"
+#include "storage/spin.h"
+
+typedef struct ProcArrayLockStruct
+{
+	FlexLock	flex;			/* common FlexLock infrastructure */
+	char		exclusive;		/* # of exclusive holders (0 or 1) */
+	int			shared;			/* # of shared holders (0..MaxBackends) */
+	PGPROC	   *ending;			/* transactions wishing to clear state */
+	TransactionId	latest_ending_xid;	/* latest ending XID */
+} ProcArrayLockStruct;
+
+/* There is only one ProcArrayLock. */
+#define	ProcArrayLockPointer() \
+	(AssertMacro(FlexLockArray[ProcArrayLock].flex.locktype == \
+		FLEXLOCK_TYPE_PROCARRAYLOCK), \
+	 (volatile ProcArrayLockStruct *) &FlexLockArray[ProcArrayLock])
+
+/*
+ * ProcArrayLockAcquire - acquire a lightweight lock in the specified mode
+ *
+ * If the lock is not available, sleep until it is.
+ *
+ * Side effect: cancel/die interrupts are held off until lock release.
+ */
+void
+ProcArrayLockAcquire(ProcArrayLockMode mode)
+{
+	volatile ProcArrayLockStruct *lock = ProcArrayLockPointer();
+	PGPROC	   *proc = MyProc;
+	bool		retry = false;
+	int			extraWaits = 0;
+
+	/*
+	 * We can't wait if we haven't got a PGPROC.  This should only occur
+	 * during bootstrap or shared memory initialization.  Put an Assert here
+	 * to catch unsafe coding practices.
+	 */
+	Assert(!(proc == NULL && IsUnderPostmaster));
+
+	/*
+	 * Lock out cancel/die interrupts until we exit the code section protected
+	 * by the ProcArrayLock.  This ensures that interrupts will not interfere
+     * with manipulations of data structures in shared memory.
+	 */
+	HOLD_INTERRUPTS();
+
+	/*
+	 * Loop here to try to acquire lock after each time we are signaled by
+	 * ProcArrayLockRelease.  See comments in LWLockAcquire for an explanation
+	 * of why do we not attempt to hand off the lock directly.
+	 */
+	for (;;)
+	{
+		bool		mustwait;
+
+		/* Acquire mutex.  Time spent holding mutex should be short! */
+		SpinLockAcquire(&lock->flex.mutex);
+
+		/* If retrying, allow LWLockRelease to release waiters again */
+		if (retry)
+			lock->flex.releaseOK = true;
+
+		/* If I can get the lock, do so quickly. */
+		if (mode == PAL_EXCLUSIVE)
+		{
+			if (lock->exclusive == 0 && lock->shared == 0)
+			{
+				lock->exclusive++;
+				mustwait = false;
+			}
+			else
+				mustwait = true;
+		}
+		else
+		{
+			if (lock->exclusive == 0)
+			{
+				lock->shared++;
+				mustwait = false;
+			}
+			else
+				mustwait = true;
+		}
+
+		if (!mustwait)
+			break;				/* got the lock */
+
+		/* Add myself to wait queue. */
+		FlexLockJoinWaitQueue(lock, (int) mode);
+
+		/* Can release the mutex now */
+		SpinLockRelease(&lock->flex.mutex);
+
+		/* Wait until awakened. */
+		extraWaits += FlexLockWait(ProcArrayLock, mode);
+
+		/* Now loop back and try to acquire lock again. */
+		retry = true;
+	}
+
+	/* We are done updating shared state of the lock itself. */
+	SpinLockRelease(&lock->flex.mutex);
+
+	TRACE_POSTGRESQL_FLEXLOCK_ACQUIRE(lockid, mode);
+
+	/* Add lock to list of locks held by this backend */
+	FlexLockRemember(ProcArrayLock);
+
+	/*
+	 * Fix the process wait semaphore's count for any absorbed wakeups.
+	 */
+	while (extraWaits-- > 0)
+		PGSemaphoreUnlock(&proc->sem);
+}
+
+/*
+ * ProcArrayLockClearTransaction - safely clear transaction details
+ *
+ * This can't be done while ProcArrayLock is held, but it's so fast that
+ * we can afford to do it while holding the spinlock, rather than acquiring
+ * and releasing the lock.
+ */
+void
+ProcArrayLockClearTransaction(TransactionId latestXid)
+{
+	volatile ProcArrayLockStruct *lock = ProcArrayLockPointer();
+	PGPROC	   *proc = MyProc;
+	int			extraWaits = 0;
+	bool		mustwait;
+
+	HOLD_INTERRUPTS();
+
+	/* Acquire mutex.  Time spent holding mutex should be short! */
+	SpinLockAcquire(&lock->flex.mutex);
+
+	if (lock->exclusive == 0 && lock->shared == 0)
+	{
+		{
+			volatile PGPROC *vproc = proc;
+			/* If there are no lockers, clar the critical PGPROC fields. */
+			vproc->xid = InvalidTransactionId;
+	        vproc->xmin = InvalidTransactionId;
+	        /* must be cleared with xid/xmin: */
+	        vproc->vacuumFlags &= ~PROC_VACUUM_STATE_MASK;
+			vproc->subxids.nxids = 0;
+			vproc->subxids.overflowed = false;
+		}
+		mustwait = false;
+
+        /* Also advance global latestCompletedXid while holding the lock */
+        if (TransactionIdPrecedes(ShmemVariableCache->latestCompletedXid,
+                                  latestXid))
+            ShmemVariableCache->latestCompletedXid = latestXid;
+	}
+	else
+	{
+		/* Rats, must wait. */
+		proc->flWaitLink = lock->ending;
+		lock->ending = proc;
+		if (!TransactionIdIsValid(lock->latest_ending_xid) ||
+				TransactionIdPrecedes(lock->latest_ending_xid, latestXid)) 
+			lock->latest_ending_xid = latestXid;
+		mustwait = true;
+	}
+
+	/* Can release the mutex now */
+	SpinLockRelease(&lock->flex.mutex);
+
+	/*
+	 * If we were not able to perfom the operation immediately, we must wait.
+	 * But we need not retry after being awoken, because the last lock holder
+	 * to release the lock will do the work first, on our behalf.
+	 */
+	if (mustwait)
+	{
+		extraWaits += FlexLockWait(ProcArrayLock, 2);
+		while (extraWaits-- > 0)
+			PGSemaphoreUnlock(&proc->sem);
+	}
+
+	RESUME_INTERRUPTS();
+}
+
+/*
+ * ProcArrayLockRelease - release a previously acquired lock
+ */
+void
+ProcArrayLockRelease(void)
+{
+	volatile ProcArrayLockStruct *lock = ProcArrayLockPointer();
+	PGPROC	   *head;
+	PGPROC	   *ending = NULL;
+	PGPROC	   *proc;
+
+	FlexLockForget(ProcArrayLock);
+
+	/* Acquire mutex.  Time spent holding mutex should be short! */
+	SpinLockAcquire(&lock->flex.mutex);
+
+	/* Release my hold on lock */
+	if (lock->exclusive > 0)
+		lock->exclusive--;
+	else
+	{
+		Assert(lock->shared > 0);
+		lock->shared--;
+	}
+
+	/*
+	 * If the lock is now free, but there are some transactions trying to
+	 * end, we must clear the critical PGPROC fields for them, and save a
+	 * list of them so we can wake them up.
+	 */
+	if (lock->exclusive == 0 && lock->shared == 0 && lock->ending != NULL)
+	{
+		volatile PGPROC *vproc;
+
+		ending = lock->ending;
+		vproc = ending;
+
+		while (vproc != NULL)
+		{
+        	vproc->xid = InvalidTransactionId;
+	        vproc->xmin = InvalidTransactionId;
+	        /* must be cleared with xid/xmin: */
+	        vproc->vacuumFlags &= ~PROC_VACUUM_STATE_MASK;
+			vproc->subxids.nxids = 0;
+			vproc->subxids.overflowed = false;
+			vproc = vproc->flWaitLink;
+		}
+
+		/* Also advance global latestCompletedXid */
+		if (TransactionIdPrecedes(ShmemVariableCache->latestCompletedXid,
+								  lock->latest_ending_xid))
+			ShmemVariableCache->latestCompletedXid = lock->latest_ending_xid;
+
+		/* Reset lock state. */
+		lock->ending = NULL;
+		lock->latest_ending_xid = InvalidTransactionId;
+	}
+
+	/*
+	 * See if I need to awaken any waiters.  If I released a non-last shared
+	 * hold, there cannot be anything to do.  Also, do not awaken any waiters
+	 * if someone has already awakened waiters that haven't yet acquired the
+	 * lock.
+	 */
+	head = lock->flex.head;
+	if (head != NULL)
+	{
+		if (lock->exclusive == 0 && lock->shared == 0 && lock->flex.releaseOK)
+		{
+			/*
+			 * Remove the to-be-awakened PGPROCs from the queue.  If the front
+			 * waiter wants exclusive lock, awaken him only. Otherwise awaken
+			 * as many waiters as want shared access.
+			 */
+			proc = head;
+			if (proc->flWaitMode != LW_EXCLUSIVE)
+			{
+				while (proc->flWaitLink != NULL &&
+					   proc->flWaitLink->flWaitMode != LW_EXCLUSIVE)
+					proc = proc->flWaitLink;
+			}
+			/* proc is now the last PGPROC to be released */
+			lock->flex.head = proc->flWaitLink;
+			proc->flWaitLink = NULL;
+			/* prevent additional wakeups until retryer gets to run */
+			lock->flex.releaseOK = false;
+		}
+		else
+		{
+			/* lock is still held, can't awaken anything */
+			head = NULL;
+		}
+	}
+
+	/* We are done updating shared state of the lock itself. */
+	SpinLockRelease(&lock->flex.mutex);
+
+	TRACE_POSTGRESQL_FLEXLOCK_RELEASE(lockid);
+
+	/*
+	 * Awaken any waiters I removed from the queue.
+	 */
+	while (head != NULL)
+	{
+		FlexLockDebug("LWLockRelease", lockid, "release waiter");
+		proc = head;
+		head = proc->flWaitLink;
+		proc->flWaitLink = NULL;
+		proc->flWaitResult = 1;		/* any non-zero value will do */
+		PGSemaphoreUnlock(&proc->sem);
+	}
+
+	/*
+	 * Also awaken any processes whose critical PGPROC fields I cleared
+	 */
+	while (ending != NULL)
+	{
+		FlexLockDebug("LWLockRelease", lockid, "release ending");
+		proc = ending;
+		ending = proc->flWaitLink;
+		proc->flWaitLink = NULL;
+		proc->flWaitResult = 1;		/* any non-zero value will do */
+		PGSemaphoreUnlock(&proc->sem);
+	}
+
+	/*
+	 * Now okay to allow cancel/die interrupts.
+	 */
+	RESUME_INTERRUPTS();
+}
diff --git a/src/include/storage/flexlock_internals.h b/src/include/storage/flexlock_internals.h
index 5f78da7..d1bca45 100644
--- a/src/include/storage/flexlock_internals.h
+++ b/src/include/storage/flexlock_internals.h
@@ -43,6 +43,7 @@ typedef struct FlexLock
 } FlexLock;
 
 #define FLEXLOCK_TYPE_LWLOCK			'l'
+#define FLEXLOCK_TYPE_PROCARRAYLOCK		'p'
 
 typedef union FlexLockPadded
 {
diff --git a/src/include/storage/procarraylock.h b/src/include/storage/procarraylock.h
new file mode 100644
index 0000000..678ca6f
--- /dev/null
+++ b/src/include/storage/procarraylock.h
@@ -0,0 +1,28 @@
+/*-------------------------------------------------------------------------
+ *
+ * procarraylock.h
+ *	  Lock management for the ProcArray
+ *
+ * Portions Copyright (c) 1996-2011, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/include/storage/lwlock.h
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef PROCARRAYLOCK_H
+#define PROCARRAYLOCK_H
+
+#include "storage/flexlock.h"
+
+typedef enum ProcArrayLockMode
+{
+	PAL_EXCLUSIVE,
+	PAL_SHARED
+} ProcArrayLockMode;
+
+extern void ProcArrayLockAcquire(ProcArrayLockMode mode);
+extern void ProcArrayLockClearTransaction(TransactionId latestXid);
+extern void ProcArrayLockRelease(void);
+
+#endif   /* PROCARRAYLOCK_H */