v11-0004-Fix-CREATE-INDEX-CONCURRENTLY-in-precence-of-vxi.patch

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

---
 src/backend/access/transam/twophase.c |  47 ++++++++++-
 src/backend/storage/lmgr/lock.c       | 115 ++++++++++++++++++++++----
 src/include/access/twophase.h         |  13 +++
 3 files changed, 154 insertions(+), 21 deletions(-)

diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c
index f67d813c56..ef551dd0d3 100644
--- a/src/backend/access/transam/twophase.c
+++ b/src/backend/access/transam/twophase.c
@@ -459,14 +459,15 @@ 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 */
+	proc->lxid = MyProc->lxid;
 	proc->xid = xid;
 	Assert(proc->xmin == InvalidTransactionId);
 	proc->delayChkpt = false;
 	proc->statusFlags = 0;
 	proc->pid = 0;
-	proc->backendId = InvalidBackendId;
+	/* May be backendId of startup process */
+	proc->backendId = MyBackendId;
 	proc->databaseId = databaseid;
 	proc->roleId = owner;
 	proc->tempNamespaceId = InvalidOid;
@@ -846,6 +847,46 @@ TwoPhaseGetGXact(TransactionId xid, bool lock_held)
 	return result;
 }
 
+/*
+ * TwoPhaseGetXidByVXid
+ *		Try to lookup for vxid among prepared xacts
+ */
+XidListEntry
+TwoPhaseGetXidByVXid(VirtualTransactionId vxid)
+{
+	int				i;
+	XidListEntry	result;
+	result.next = NULL;
+	result.xid = 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(vxid, *proc);
+
+		if (VirtualTransactionIdEquals(vxid, proc_vxid))
+		{
+			if (result.xid != InvalidTransactionId)
+			{
+				/* Already has candidate - need to alloc some space */
+				XidListEntry *copy = palloc(sizeof(XidListEntry));
+				copy->next = result.next;
+				copy->xid = result.xid;
+				result.next = copy;
+			}
+			result.xid = 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 108b4d9023..137eeedb93 100644
--- a/src/backend/storage/lmgr/lock.c
+++ b/src/backend/storage/lmgr/lock.c
@@ -386,6 +386,7 @@ static void LockRefindAndRelease(LockMethod lockMethodTable, PGPROC *proc,
 								 bool decrement_strong_lock_count);
 static void GetSingleProcBlockerStatusData(PGPROC *blocked_proc,
 										   BlockedProcsData *data);
+static bool WaitXact(VirtualTransactionId vxid, TransactionId xid, bool wait);
 
 
 /*
@@ -3017,6 +3018,12 @@ GetLockConflicts(const LOCKTAG *locktag, LOCKMODE lockmode, int *countp)
 				/* Conflict! */
 				GET_VXID_FROM_PGPROC(vxid, *proc);
 
+				/* Prefer real Xid over local VXid */
+				if (TransactionIdIsValid(proc->xid))
+				{
+					vxid.backendId = InvalidBackendId;
+					vxid.localTransactionId = proc->xid;
+				}
 				if (VirtualTransactionIdIsValid(vxid))
 					vxids[count++] = vxid;
 				/* else, xact already committed or aborted */
@@ -3078,6 +3085,12 @@ GetLockConflicts(const LOCKTAG *locktag, LOCKMODE lockmode, int *countp)
 
 				GET_VXID_FROM_PGPROC(vxid, *proc);
 
+				/* Prefer real Xid over local VXid */
+				if (TransactionIdIsValid(proc->xid))
+				{
+					vxid.backendId = InvalidBackendId;
+					vxid.localTransactionId = proc->xid;
+				}
 				if (VirtualTransactionIdIsValid(vxid))
 				{
 					int			i;
@@ -4447,6 +4460,76 @@ VirtualXactLockTableCleanup(void)
 	}
 }
 
+/*
+ *		WaitXact
+ *
+ *		Wait for xid completition if have xid. Otherwise try to find xid among
+ *		two-phase procarray entries.
+ */
+static bool WaitXact(VirtualTransactionId vxid, TransactionId xid, bool wait)
+{
+	LockAcquireResult	lar;
+	LOCKTAG				tag;
+	XidListEntry		xidlist;
+	XidListEntry	   *xidlist_ptr = NULL; /* pointer to TwoPhaseGetXidByVXid()s pallocs */
+	bool				result;
+
+	if (TransactionIdIsValid(xid))
+	{
+		/* We know exact xid - no need to search in 2PC state */
+		xidlist.xid = xid;
+		xidlist.next = NULL;
+	}
+	else
+	{
+		/* You cant have vxid among 2PCs if you have no 2PCs */
+		if (max_prepared_xacts == 0)
+			return true;
+
+		/*
+		 * We must search for vxids in 2pc state
+		 * XXX: O(N*N) complexity where N is number of prepared xacts
+		 */
+		xidlist = TwoPhaseGetXidByVXid(vxid);
+		/* Return if transaction is gone entirely */
+		if (!TransactionIdIsValid(xidlist.xid))
+			return true;
+		xidlist_ptr = xidlist.next;
+	}
+
+
+	while (true)
+	{
+		/* Iterate over possible xids */
+		Assert(TransactionIdIsValid(xidlist.xid));
+		SET_LOCKTAG_TRANSACTION(tag, xidlist.xid);
+		lar = LockAcquire(&tag, ShareLock, false, !wait);
+		if (lar == LOCKACQUIRE_NOT_AVAIL)
+		{
+			result = false;
+			break;
+		}
+
+		LockRelease(&tag, ShareLock, false);
+
+		if (xidlist.next == NULL)
+		{
+			result = true;
+			break;
+		}
+		xidlist = *xidlist.next;
+	}
+
+	/* Free all pallocs made by TwoPhaseGetXidByVXid() */
+	while (xidlist_ptr)
+	{
+		void *chunk = xidlist_ptr;
+		xidlist_ptr = xidlist_ptr->next;
+		pfree(chunk);
+	}
+	return result;
+}
+
 /*
  *		VirtualXactLock
  *
@@ -4459,25 +4542,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 WaitXact(vxid, vxid.localTransactionId, wait);
 
 	SET_LOCKTAG_VIRTUALTRANSACTION(tag, vxid);
 
@@ -4491,7 +4567,7 @@ VirtualXactLock(VirtualTransactionId vxid, bool wait)
 	 */
 	proc = BackendIdGetProc(vxid.backendId);
 	if (proc == NULL)
-		return true;
+		return WaitXact(vxid, InvalidTransactionId, wait);
 
 	/*
 	 * We must acquire this lock before checking the backendId and lxid
@@ -4505,9 +4581,12 @@ VirtualXactLock(VirtualTransactionId vxid, bool wait)
 		|| proc->fpLocalTransactionId != vxid.localTransactionId)
 	{
 		LWLockRelease(&proc->fpInfoLock);
-		return true;
+		return WaitXact(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.
@@ -4559,7 +4638,7 @@ VirtualXactLock(VirtualTransactionId vxid, bool wait)
 	(void) LockAcquire(&tag, ShareLock, false, false);
 
 	LockRelease(&tag, ShareLock, false);
-	return true;
+	return WaitXact(vxid, xid, wait);
 }
 
 /*
diff --git a/src/include/access/twophase.h b/src/include/access/twophase.h
index 91786da784..ba0ac17359 100644
--- a/src/include/access/twophase.h
+++ b/src/include/access/twophase.h
@@ -25,6 +25,17 @@
  */
 typedef struct GlobalTransactionData *GlobalTransaction;
 
+/* 
+ * XidListEntry is expected to be used as list very rarely. Under normal
+ * circumstances TwoPhaseGetXidByVXid() returns only one xid.
+ * But under certain conditions can return many xids or nothing.
+ */
+typedef struct XidListEntry
+{
+	TransactionId xid;
+	struct XidListEntry* next;
+} XidListEntry;
+
 /* GUC variable */
 extern PGDLLIMPORT int max_prepared_xacts;
 
@@ -58,4 +69,6 @@ extern void PrepareRedoAdd(char *buf, XLogRecPtr start_lsn,
 						   XLogRecPtr end_lsn, RepOriginId origin_id);
 extern void PrepareRedoRemove(TransactionId xid, bool giveWarning);
 extern void restoreTwoPhaseData(void);
+
+extern XidListEntry TwoPhaseGetXidByVXid(VirtualTransactionId vxid);
 #endif							/* TWOPHASE_H */
-- 
2.24.3 (Apple Git-128)