lazy vxid locks, v2

diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c
index 7f01a83..f0c80f1 100644
--- a/src/backend/access/transam/xact.c
+++ b/src/backend/access/transam/xact.c
@@ -1725,14 +1725,7 @@ StartTransaction(void)
 	/*
 	 * Lock the virtual transaction id before we announce it in the proc array
 	 */
-	VirtualXactLockTableInsert(vxid);
-
-	/*
-	 * Advertise it in the proc array.	We assume assignment of
-	 * LocalTransactionID is atomic, and the backendId should be set already.
-	 */
-	Assert(MyProc->backendId == vxid.backendId);
-	MyProc->lxid = vxid.localTransactionId;
+	VirtualXactLockInitialize(vxid);
 
 	TRACE_POSTGRESQL_TRANSACTION_START(vxid.localTransactionId);
 
@@ -1878,6 +1871,7 @@ CommitTransaction(void)
 	 * must be done _before_ releasing locks we hold and _after_
 	 * RecordTransactionCommit.
 	 */
+	VirtualXactLockCleanup();
 	ProcArrayEndTransaction(MyProc, latestXid);
 
 	/*
@@ -2140,6 +2134,7 @@ PrepareTransaction(void)
 	 * done *after* the prepared transaction has been marked valid, else
 	 * someone may think it is unlocked and recyclable.
 	 */
+	VirtualXactLockCleanup();
 	ProcArrayClearTransaction(MyProc);
 
 	/*
@@ -2306,6 +2301,7 @@ AbortTransaction(void)
 	 * must be done _before_ releasing locks we hold and _after_
 	 * RecordTransactionAbort.
 	 */
+	VirtualXactLockCleanup();
 	ProcArrayEndTransaction(MyProc, latestXid);
 
 	/*
diff --git a/src/backend/commands/indexcmds.c b/src/backend/commands/indexcmds.c
index b7c021d..a583399 100644
--- a/src/backend/commands/indexcmds.c
+++ b/src/backend/commands/indexcmds.c
@@ -482,7 +482,7 @@ DefineIndex(RangeVar *heapRelation,
 
 	while (VirtualTransactionIdIsValid(*old_lockholders))
 	{
-		VirtualXactLockTableWait(*old_lockholders);
+		VirtualXactLock(*old_lockholders, true);
 		old_lockholders++;
 	}
 
@@ -568,7 +568,7 @@ DefineIndex(RangeVar *heapRelation,
 
 	while (VirtualTransactionIdIsValid(*old_lockholders))
 	{
-		VirtualXactLockTableWait(*old_lockholders);
+		VirtualXactLock(*old_lockholders, true);
 		old_lockholders++;
 	}
 
@@ -665,7 +665,7 @@ DefineIndex(RangeVar *heapRelation,
 		}
 
 		if (VirtualTransactionIdIsValid(old_snapshots[i]))
-			VirtualXactLockTableWait(old_snapshots[i]);
+			VirtualXactLock(old_snapshots[i], true);
 	}
 
 	/*
diff --git a/src/backend/postmaster/autovacuum.c b/src/backend/postmaster/autovacuum.c
index 483a829..8b233ef 100644
--- a/src/backend/postmaster/autovacuum.c
+++ b/src/backend/postmaster/autovacuum.c
@@ -1992,7 +1992,7 @@ do_autovacuum(void)
 			backendID = GetTempNamespaceBackendId(classForm->relnamespace);
 
 			/* We just ignore it if the owning backend is still active */
-			if (backendID == MyBackendId || !BackendIdIsActive(backendID))
+			if (backendID == MyBackendId || BackendIdGetProc(backendID) == NULL)
 			{
 				/*
 				 * We found an orphan temp table (which was probably left
diff --git a/src/backend/storage/ipc/procarray.c b/src/backend/storage/ipc/procarray.c
index e7593fa..2174061 100644
--- a/src/backend/storage/ipc/procarray.c
+++ b/src/backend/storage/ipc/procarray.c
@@ -363,7 +363,6 @@ ProcArrayEndTransaction(PGPROC *proc, TransactionId latestXid)
 		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;
@@ -390,7 +389,6 @@ ProcArrayEndTransaction(PGPROC *proc, TransactionId latestXid)
 		 */
 		Assert(!TransactionIdIsValid(proc->xid));
 
-		proc->lxid = InvalidLocalTransactionId;
 		proc->xmin = InvalidTransactionId;
 		/* must be cleared with xid/xmin: */
 		proc->vacuumFlags &= ~PROC_VACUUM_STATE_MASK;
@@ -421,7 +419,6 @@ ProcArrayClearTransaction(PGPROC *proc)
 	 * ProcArray.
 	 */
 	proc->xid = InvalidTransactionId;
-	proc->lxid = InvalidLocalTransactionId;
 	proc->xmin = InvalidTransactionId;
 	proc->recoveryConflictPending = false;
 
diff --git a/src/backend/storage/ipc/sinvaladt.c b/src/backend/storage/ipc/sinvaladt.c
index 4f446aa..807b7be 100644
--- a/src/backend/storage/ipc/sinvaladt.c
+++ b/src/backend/storage/ipc/sinvaladt.c
@@ -139,6 +139,7 @@ typedef struct ProcState
 {
 	/* procPid is zero in an inactive ProcState array entry. */
 	pid_t		procPid;		/* PID of backend, for signaling */
+	PGPROC	   *proc;			/* PGPROC of backend */
 	/* nextMsgNum is meaningless if procPid == 0 or resetState is true. */
 	int			nextMsgNum;		/* next message number to read */
 	bool		resetState;		/* backend needs to reset its state */
@@ -245,6 +246,7 @@ CreateSharedInvalidationState(void)
 	for (i = 0; i < shmInvalBuffer->maxBackends; i++)
 	{
 		shmInvalBuffer->procState[i].procPid = 0;		/* inactive */
+		shmInvalBuffer->procState[i].proc = NULL;
 		shmInvalBuffer->procState[i].nextMsgNum = 0;	/* meaningless */
 		shmInvalBuffer->procState[i].resetState = false;
 		shmInvalBuffer->procState[i].signaled = false;
@@ -313,6 +315,7 @@ SharedInvalBackendInit(bool sendOnly)
 
 	/* mark myself active, with all extant messages already read */
 	stateP->procPid = MyProcPid;
+	stateP->proc = MyProc;
 	stateP->nextMsgNum = segP->maxMsgNum;
 	stateP->resetState = false;
 	stateP->signaled = false;
@@ -352,6 +355,7 @@ CleanupInvalidationState(int status, Datum arg)
 
 	/* Mark myself inactive */
 	stateP->procPid = 0;
+	stateP->proc = NULL;
 	stateP->nextMsgNum = 0;
 	stateP->resetState = false;
 	stateP->signaled = false;
@@ -368,13 +372,16 @@ CleanupInvalidationState(int status, Datum arg)
 }
 
 /*
- * BackendIdIsActive
- *		Test if the given backend ID is currently assigned to a process.
+ * BackendIdGetProc
+ *		Get the PGPROC structure for a backend, given the backend ID.
+ *		The result may be out of date arbitrarily quickly, so the caller
+ *		must be careful about how this information is used.  NULL is
+ *		returned if the backend is not active.
  */
-bool
-BackendIdIsActive(int backendID)
+PGPROC *
+BackendIdGetProc(int backendID)
 {
-	bool		result;
+	PGPROC	   *result = NULL;
 	SISeg	   *segP = shmInvalBuffer;
 
 	/* Need to lock out additions/removals of backends */
@@ -384,10 +391,8 @@ BackendIdIsActive(int backendID)
 	{
 		ProcState  *stateP = &segP->procState[backendID - 1];
 
-		result = (stateP->procPid != 0);
+		result = stateP->proc;
 	}
-	else
-		result = false;
 
 	LWLockRelease(SInvalWriteLock);
 
diff --git a/src/backend/storage/ipc/standby.c b/src/backend/storage/ipc/standby.c
index 75b5ab4..3456e4a 100644
--- a/src/backend/storage/ipc/standby.c
+++ b/src/backend/storage/ipc/standby.c
@@ -82,7 +82,7 @@ InitRecoveryTransactionEnvironment(void)
 	 */
 	vxid.backendId = MyBackendId;
 	vxid.localTransactionId = GetNextLocalTransactionId();
-	VirtualXactLockTableInsert(vxid);
+	VirtualXactLockInitialize(vxid);
 
 	standbyState = STANDBY_INITIALIZED;
 }
@@ -201,7 +201,7 @@ ResolveRecoveryConflictWithVirtualXIDs(VirtualTransactionId *waitlist,
 		standbyWait_us = STANDBY_INITIAL_WAIT_US;
 
 		/* wait until the virtual xid is gone */
-		while (!ConditionalVirtualXactLockTableWait(*waitlist))
+		while (!VirtualXactLock(*waitlist, false))
 		{
 			/*
 			 * Report via ps if we have been waiting for more than 500 msec
diff --git a/src/backend/storage/lmgr/lmgr.c b/src/backend/storage/lmgr/lmgr.c
index 859b385..9d0994e 100644
--- a/src/backend/storage/lmgr/lmgr.c
+++ b/src/backend/storage/lmgr/lmgr.c
@@ -514,70 +514,6 @@ ConditionalXactLockTableWait(TransactionId xid)
 	return true;
 }
 
-
-/*
- *		VirtualXactLockTableInsert
- *
- * Insert a lock showing that the given virtual transaction ID is running ---
- * this is done at main transaction start when its VXID is assigned.
- * The lock can then be used to wait for the transaction to finish.
- */
-void
-VirtualXactLockTableInsert(VirtualTransactionId vxid)
-{
-	LOCKTAG		tag;
-
-	Assert(VirtualTransactionIdIsValid(vxid));
-
-	SET_LOCKTAG_VIRTUALTRANSACTION(tag, vxid);
-
-	(void) LockAcquire(&tag, ExclusiveLock, false, false);
-}
-
-/*
- *		VirtualXactLockTableWait
- *
- * Waits until the lock on the given VXID is released, which shows that
- * the top-level transaction owning the VXID has ended.
- */
-void
-VirtualXactLockTableWait(VirtualTransactionId vxid)
-{
-	LOCKTAG		tag;
-
-	Assert(VirtualTransactionIdIsValid(vxid));
-
-	SET_LOCKTAG_VIRTUALTRANSACTION(tag, vxid);
-
-	(void) LockAcquire(&tag, ShareLock, false, false);
-
-	LockRelease(&tag, ShareLock, false);
-}
-
-/*
- *		ConditionalVirtualXactLockTableWait
- *
- * As above, but only lock if we can get the lock without blocking.
- * Returns TRUE if the lock was acquired.
- */
-bool
-ConditionalVirtualXactLockTableWait(VirtualTransactionId vxid)
-{
-	LOCKTAG		tag;
-
-	Assert(VirtualTransactionIdIsValid(vxid));
-
-	SET_LOCKTAG_VIRTUALTRANSACTION(tag, vxid);
-
-	if (LockAcquire(&tag, ShareLock, false, true) == LOCKACQUIRE_NOT_AVAIL)
-		return false;
-
-	LockRelease(&tag, ShareLock, false);
-
-	return true;
-}
-
-
 /*
  *		LockDatabaseObject
  *
diff --git a/src/backend/storage/lmgr/lock.c b/src/backend/storage/lmgr/lock.c
index e809a58..665f898 100644
--- a/src/backend/storage/lmgr/lock.c
+++ b/src/backend/storage/lmgr/lock.c
@@ -38,6 +38,7 @@
 #include "miscadmin.h"
 #include "pg_trace.h"
 #include "pgstat.h"
+#include "storage/sinvaladt.h"
 #include "storage/standby.h"
 #include "utils/memutils.h"
 #include "utils/ps_status.h"
@@ -138,6 +139,9 @@ static int			FastPathLocalUseCount = 0;
 #define FAST_PATH_CHECK_LOCKMODE(proc, n, l) \
 	 ((proc)->fpLockBits & (UINT64CONST(1) << FAST_PATH_BIT_POSITION(n, l)))
 
+#define FAST_PATH_DEFER_VXID_LOCK		\
+	(UINT64CONST(1) << (FAST_PATH_BITS_PER_SLOT * FP_LOCK_SLOTS_PER_BACKEND))
+
 /*
  * The fast-path lock mechanism is concerned only with relation locks on
  * unshared relations by backends bound to a database.  The fast-path
@@ -3531,3 +3535,154 @@ lock_twophase_postabort(TransactionId xid, uint16 info,
 {
 	lock_twophase_postcommit(xid, info, recdata, len);
 }
+
+/*
+ *		VirtualXactLockInitialize
+ *
+ *		We set a flag in MyProc->fpLockState indicating that we have a
+ *		"deferred" VXID lock.  That is, we have an active VXID, but we
+ *		haven't actually taken an exclusive lock on it.  VXID locks are
+ *		rarely waited for, so it makes sense to defer the actual lock
+ *		acquisition to the point when it's needed.  Another backend wishing
+ *		to wait on the lock can acquire the lock on our behalf and then
+ *		wait on it.  We'll figure it all out in VirtualXactLockCleanup().
+ *
+ *		We set lxid while holding the lock to guarantee that anyone who
+ *		sees the lxid set and subsequently takes our backendLock will also
+ *		see the FAST_PATH_DEFER_VXID lock bit set.
+ */
+void
+VirtualXactLockInitialize(VirtualTransactionId vxid)
+{
+	Assert(VirtualTransactionIdIsValid(vxid));
+
+	LWLockAcquire(MyProc->backendLock, LW_EXCLUSIVE);
+
+	Assert(MyProc->backendId == vxid.backendId);
+	MyProc->lxid = vxid.localTransactionId;
+	MyProc->fpLockBits |= FAST_PATH_DEFER_VXID_LOCK;
+
+	LWLockRelease(MyProc->backendLock);
+}
+
+/*
+ *		VirtualXactLockCleanup
+ *
+ *		Check whether a VXID lock has been materialized; if so, release it,
+ *		unblocking waiters.
+ */
+void
+VirtualXactLockCleanup()
+{
+	VirtualTransactionId	vxid;
+	bool	cleanup = false;
+
+	Assert(MyProc->backendId != InvalidBackendId);
+	Assert(MyProc->lxid != InvalidLocalTransactionId);
+
+	GET_VXID_FROM_PGPROC(vxid, *MyProc);
+
+	LWLockAcquire(MyProc->backendLock, LW_EXCLUSIVE);
+
+	if ((MyProc->fpLockBits & FAST_PATH_DEFER_VXID_LOCK) != 0)
+		MyProc->fpLockBits &= ~FAST_PATH_DEFER_VXID_LOCK;
+	else
+		cleanup = true;
+	MyProc->lxid = InvalidLocalTransactionId;
+
+	LWLockRelease(MyProc->backendLock);
+
+	/* If someone materialized the lock on our behalf, we must release it. */
+	if (cleanup)
+	{
+		LOCKTAG	locktag;
+
+		SET_LOCKTAG_VIRTUALTRANSACTION(locktag, vxid);
+		LockRefindAndRelease(LockMethods[DEFAULT_LOCKMETHOD], MyProc,
+							 &locktag, ExclusiveLock, false);
+	}	
+}
+
+/*
+ *		VirtualXactLock
+ *
+ * If wait = true, wait until the given VXID has been released, and then
+ * return true.
+ *
+ * If wait = false, just check whether the VXID is still running, and return
+ * true or false.
+ */
+bool
+VirtualXactLock(VirtualTransactionId vxid, bool wait)
+{
+	LOCKTAG		tag;
+	PGPROC	   *proc;
+
+	Assert(VirtualTransactionIdIsValid(vxid));
+
+	SET_LOCKTAG_VIRTUALTRANSACTION(tag, vxid);
+
+	/*
+	 * If a lock table entry must be made, this is the PGPROC on whose behalf
+	 * it must be done.  Note that the transaction might end or the PGPROC
+	 * might be reassigned to a new backend before we get around to examining
+	 * it, but it doesn't matter.  If we find upon examination that the
+	 * relevant lxid is no longer running here, that's enough to prove that
+	 * it's no longer running anywhere.
+	 */
+	proc = BackendIdGetProc(vxid.backendId);
+
+	/*
+	 * We must acquire this lock before checking the backendId and lxid
+	 * against the ones we're waiting for.  The target backend will only
+	 * set or clear lxid while holding this lock.
+	 */
+	LWLockAcquire(proc->backendLock, LW_EXCLUSIVE);
+
+	/* If the transaction has ended, our work here is done. */
+	if (proc->backendId != vxid.backendId || proc->lxid != vxid.localTransactionId)
+	{
+		LWLockRelease(proc->backendLock);
+		return true;
+	}
+
+	/*
+	 * If we aren't asked to wait, there's no need to set up a lock table
+	 * entry.  The transaction is still in progress, so just return false.
+	 */
+	if (!wait)
+	{
+		LWLockRelease(proc->backendLock);
+		return false;
+	}
+
+	/*
+	 * OK, we're going to need to sleep on the VXID.  But first, we must set
+	 * up the primary lock table entry, if needed.
+	 */
+	if ((proc->fpLockBits & FAST_PATH_DEFER_VXID_LOCK) != 0)
+	{
+		PROCLOCK   *proclock;
+		uint32		hashcode;
+
+		hashcode = LockTagHashCode(&tag);
+		proclock = SetupLockInTable(LockMethods[DEFAULT_LOCKMETHOD], proc,
+									&tag, hashcode, ExclusiveLock);
+		if (!proclock)
+			ereport(ERROR,
+					(errcode(ERRCODE_OUT_OF_MEMORY),
+					 errmsg("out of shared memory"),
+		  errhint("You might need to increase max_locks_per_transaction.")));
+		GrantLock(proclock->tag.myLock, proclock, ExclusiveLock);
+		proc->fpLockBits &= ~FAST_PATH_DEFER_VXID_LOCK;
+	}
+
+	/* Done with proc->fpLockBits */
+	LWLockRelease(proc->backendLock);
+
+	/* Time to wait. */
+	(void) LockAcquire(&tag, ShareLock, false, false);
+
+	LockRelease(&tag, ShareLock, false);
+	return true;
+}
diff --git a/src/include/storage/lmgr.h b/src/include/storage/lmgr.h
index bd44d92..340f6a3 100644
--- a/src/include/storage/lmgr.h
+++ b/src/include/storage/lmgr.h
@@ -56,11 +56,6 @@ extern void XactLockTableDelete(TransactionId xid);
 extern void XactLockTableWait(TransactionId xid);
 extern bool ConditionalXactLockTableWait(TransactionId xid);
 
-/* Lock a VXID (used to wait for a transaction to finish) */
-extern void VirtualXactLockTableInsert(VirtualTransactionId vxid);
-extern void VirtualXactLockTableWait(VirtualTransactionId vxid);
-extern bool ConditionalVirtualXactLockTableWait(VirtualTransactionId vxid);
-
 /* Lock a general object (other than a relation) of the current database */
 extern void LockDatabaseObject(Oid classid, Oid objid, uint16 objsubid,
 				   LOCKMODE lockmode);
diff --git a/src/include/storage/lock.h b/src/include/storage/lock.h
index b3aeef9..eb6b953 100644
--- a/src/include/storage/lock.h
+++ b/src/include/storage/lock.h
@@ -543,4 +543,9 @@ extern void DumpLocks(PGPROC *proc);
 extern void DumpAllLocks(void);
 #endif
 
+/* Lock a VXID (used to wait for a transaction to finish) */
+extern void VirtualXactLockInitialize(VirtualTransactionId vxid);
+extern void VirtualXactLockCleanup(void);
+extern bool VirtualXactLock(VirtualTransactionId vxid, bool wait);
+
 #endif   /* LOCK_H */
diff --git a/src/include/storage/sinvaladt.h b/src/include/storage/sinvaladt.h
index c703558..a61d696 100644
--- a/src/include/storage/sinvaladt.h
+++ b/src/include/storage/sinvaladt.h
@@ -22,6 +22,7 @@
 #ifndef SINVALADT_H
 #define SINVALADT_H
 
+#include "storage/proc.h"
 #include "storage/sinval.h"
 
 /*
@@ -30,7 +31,7 @@
 extern Size SInvalShmemSize(void);
 extern void CreateSharedInvalidationState(void);
 extern void SharedInvalBackendInit(bool sendOnly);
-extern bool BackendIdIsActive(int backendID);
+extern PGPROC *BackendIdGetProc(int backendID);
 
 extern void SIInsertDataEntries(const SharedInvalidationMessage *data, int n);
 extern int	SIGetDataEntries(SharedInvalidationMessage *data, int datasize);