From 5a6b6553dfd91d1d42e6fb42a43d172915a393e5 Mon Sep 17 00:00:00 2001
From: Andrey Borodin <amborodin@acm.org>
Date: Mon, 19 Jul 2021 11:50:02 +0500
Subject: [PATCH v14 4/6] Fix CREATE INDEX CONCURRENTLY in precence of vxids
 converted to 2PC

Transaction having vxid can be converted to 2PC after GetLockConflicts()
call. To be sure we await it's completition we setup lxid\backend id
if 2PC PGPROCs so that we can find what to wait for if we only have vxid.
---
 src/backend/access/transam/twophase.c | 58 ++++++++++++++++++-
 src/backend/storage/lmgr/lock.c       | 83 +++++++++++++++++++++------
 src/include/access/twophase.h         |  2 +
 3 files changed, 122 insertions(+), 21 deletions(-)

diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c
index 2156de187c..1c5dd0342d 100644
--- a/src/backend/access/transam/twophase.c
+++ b/src/backend/access/transam/twophase.c
@@ -459,14 +459,27 @@ MarkAsPreparingGuts(GlobalTransaction gxact, TransactionId xid, const char *gid,
 	proc->pgprocno = gxact->pgprocno;
 	SHMQueueElemInit(&(proc->links));
 	proc->waitStatus = PROC_WAIT_STATUS_OK;
-	/* We set up the gxact's VXID as InvalidBackendId/XID */
-	proc->lxid = (LocalTransactionId) xid;
+	/* We set up the gxact's VXID as real for CIC purposes */
+	if (LocalTransactionIdIsValid(MyProc->lxid))
+	{
+		proc->lxid = MyProc->lxid;
+		proc->backendId = MyBackendId;
+	}
+	else
+	{
+		/*
+		 * Probably we are in startup process.
+		 * Set any VXID that will be recognized as VirtualTransactionIdIsValid
+		 * in GetLockConflicts().
+		 */
+		proc->lxid = xid;
+		proc->backendId = InvalidBackendId;
+	}
 	proc->xid = xid;
 	Assert(proc->xmin == InvalidTransactionId);
 	proc->delayChkpt = false;
 	proc->statusFlags = 0;
 	proc->pid = 0;
-	proc->backendId = InvalidBackendId;
 	proc->databaseId = databaseid;
 	proc->roleId = owner;
 	proc->tempNamespaceId = InvalidOid;
@@ -846,6 +859,45 @@ TwoPhaseGetGXact(TransactionId xid, bool lock_held)
 	return result;
 }
 
+/*
+ * TwoPhaseGetXidByVXid
+ *		Try to lookup for vxid among prepared xacts.
+ *		If more then one 2PC have same vxid - return any and set overflown to
+ *		true.
+ */
+TransactionId
+TwoPhaseGetXidByVXid(VirtualTransactionId vxid, bool *overflown)
+{
+	int				i;
+	TransactionId	result = InvalidTransactionId;
+
+	LWLockAcquire(TwoPhaseStateLock, LW_SHARED);
+
+	for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
+	{
+		GlobalTransaction	gxact = TwoPhaseState->prepXacts[i];
+		PGPROC			   *proc = &ProcGlobal->allProcs[gxact->pgprocno];
+		VirtualTransactionId proc_vxid;
+		GET_VXID_FROM_PGPROC(proc_vxid, *proc);
+
+		if (VirtualTransactionIdEquals(vxid, proc_vxid))
+		{
+			if (result != InvalidTransactionId)
+			{
+				/* Already has candidate - need to set overflown */
+				*overflown = true;
+				return result;
+			}
+			/* Found one, but need to check that there is no others */
+			result = gxact->xid;
+		}
+	}
+
+	LWLockRelease(TwoPhaseStateLock);
+
+	return result;
+}
+
 /*
  * TwoPhaseGetDummyBackendId
  *		Get the dummy backend ID for prepared transaction specified by XID
diff --git a/src/backend/storage/lmgr/lock.c b/src/backend/storage/lmgr/lock.c
index 364654e106..f55324d0da 100644
--- a/src/backend/storage/lmgr/lock.c
+++ b/src/backend/storage/lmgr/lock.c
@@ -387,6 +387,7 @@ static void LockRefindAndRelease(LockMethod lockMethodTable, PGPROC *proc,
 								 bool decrement_strong_lock_count);
 static void GetSingleProcBlockerStatusData(PGPROC *blocked_proc,
 										   BlockedProcsData *data);
+static bool WaitPreparedXact(VirtualTransactionId vxid, TransactionId xid, bool wait);
 
 
 /*
@@ -4529,6 +4530,56 @@ VirtualXactLockTableCleanup(void)
 	}
 }
 
+/*
+ *		WaitPreparedXact
+ *
+ *		Wait for xid completition if have xid. Otherwise try to find xid among
+ *		two-phase procarray entries.
+ */
+static bool WaitPreparedXact(VirtualTransactionId vxid, TransactionId xid, bool wait)
+{
+	LockAcquireResult	lar;
+	LOCKTAG				tag;
+	bool				overflown = false;
+
+	/* There is no point to wait for 2PCs if you have no 2PCs. */
+	if (max_prepared_xacts == 0)
+		return true;
+
+	do
+	{
+		/* Clear from restart attempts. */
+		if(overflown)
+		{
+			xid = InvalidTransactionId;
+			overflown = false;
+		}
+
+		/* If we have no xid - try to find one. */
+		if (!TransactionIdIsValid(xid))
+			xid = TwoPhaseGetXidByVXid(vxid, &overflown);
+
+		/* There is no 2PC with this vxid - exit */
+		if (!TransactionIdIsValid(xid))
+		{
+			Assert(!overflown);
+			return true;
+		}
+
+		/* Wait for xid completition */
+		SET_LOCKTAG_TRANSACTION(tag, xid);
+		lar = LockAcquire(&tag, ShareLock, false, !wait);
+		if (lar == LOCKACQUIRE_NOT_AVAIL)
+			return false;
+		LockRelease(&tag, ShareLock, false);
+
+		/* if there was multiple 2PCs with same vxid - start over. */
+		/* This never happens if we had a known xid as an argument. */
+	} while (overflown);
+
+	return true;
+}
+
 /*
  *		VirtualXactLock
  *
@@ -4541,25 +4592,18 @@ VirtualXactLockTableCleanup(void)
 bool
 VirtualXactLock(VirtualTransactionId vxid, bool wait)
 {
-	LOCKTAG		tag;
-	PGPROC	   *proc;
+	LOCKTAG			tag;
+	PGPROC		   *proc;
+	TransactionId	xid = InvalidTransactionId;
 
 	Assert(VirtualTransactionIdIsValid(vxid));
 
+	/*
+	 * Already prepared transactions don't hold vxid locks.  The
+	 * LocalTransactionId is always a normal, locked XID.
+	 */
 	if (VirtualTransactionIdIsPreparedXact(vxid))
-	{
-		LockAcquireResult lar;
-
-		/*
-		 * Prepared transactions don't hold vxid locks.  The
-		 * LocalTransactionId is always a normal, locked XID.
-		 */
-		SET_LOCKTAG_TRANSACTION(tag, vxid.localTransactionId);
-		lar = LockAcquire(&tag, ShareLock, false, !wait);
-		if (lar != LOCKACQUIRE_NOT_AVAIL)
-			LockRelease(&tag, ShareLock, false);
-		return lar != LOCKACQUIRE_NOT_AVAIL;
-	}
+		return WaitPreparedXact(vxid, vxid.localTransactionId, wait);
 
 	SET_LOCKTAG_VIRTUALTRANSACTION(tag, vxid);
 
@@ -4573,7 +4617,7 @@ VirtualXactLock(VirtualTransactionId vxid, bool wait)
 	 */
 	proc = BackendIdGetProc(vxid.backendId);
 	if (proc == NULL)
-		return true;
+		return WaitPreparedXact(vxid, InvalidTransactionId, wait);
 
 	/*
 	 * We must acquire this lock before checking the backendId and lxid
@@ -4587,9 +4631,12 @@ VirtualXactLock(VirtualTransactionId vxid, bool wait)
 		|| proc->fpLocalTransactionId != vxid.localTransactionId)
 	{
 		LWLockRelease(&proc->fpInfoLock);
-		return true;
+		return WaitPreparedXact(vxid, InvalidTransactionId, wait);
 	}
 
+	/* Save the xid to test if transaction coverted to 2pc later */
+	xid = proc->xid;
+
 	/*
 	 * 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.
@@ -4641,7 +4688,7 @@ VirtualXactLock(VirtualTransactionId vxid, bool wait)
 	(void) LockAcquire(&tag, ShareLock, false, false);
 
 	LockRelease(&tag, ShareLock, false);
-	return true;
+	return WaitPreparedXact(vxid, xid, wait);
 }
 
 /*
diff --git a/src/include/access/twophase.h b/src/include/access/twophase.h
index e27e1a8fe8..03d4f73868 100644
--- a/src/include/access/twophase.h
+++ b/src/include/access/twophase.h
@@ -60,4 +60,6 @@ extern void PrepareRedoRemove(TransactionId xid, bool giveWarning);
 extern void restoreTwoPhaseData(void);
 extern bool LookupGXact(const char *gid, XLogRecPtr prepare_at_lsn,
 						TimestampTz origin_prepare_timestamp);
+
+extern TransactionId TwoPhaseGetXidByVXid(VirtualTransactionId vxid, bool *overflown);
 #endif							/* TWOPHASE_H */
-- 
2.24.3 (Apple Git-128)

