From f4ac105fbb654d625ab93e50021cf6b577cd23ee Mon Sep 17 00:00:00 2001
From: Nikhil Sontakke <nikhils@2ndQuadrant.com>
Date: Wed, 28 Mar 2018 13:59:26 +0530
Subject: [PATCH 2/7] Introduce LogicalLockTransaction/LogicalUnlockTransaction
  APIs

When a transaction aborts, it's changes are considered unnecessary
for other transactions. That means the changes may be either cleaned
up by vacuum or removed from HOT chains (thus made inaccessible
through indexes), and there may be other such consequences.

When decoding committed transactions this is not an issue, and we
never decode transactions that abort before the decoding starts
(where decoding means passing it to ReorderBufferCommit).

But for in-progress transactions - for example when decoding prepared
transactions on PREPARE (and not COMMIT PREPARED as before), this
may cause failures when the output plugin consults catalogs (both
system and user-defined).

To prevent aborts concurrent with plugins accessing catalogs, we
introduce an API the output plugins are required to use (when
decoding in-progress transactions only).

Before accessing any catalogs, output plugins are required to call
LogicalLockTransaction and then release it using
LogicalUnlockTransaction. Implementation is via adding support for
decoding groups. Use LockHashPartitionLockByProc on the group leader
to get the LWLock protecting these fields. For prepared and uncommitted
transactions, decoding backends working on the same XID will link
themselves up to the corresponding PGPROC entry (decodeGroupLeader).

They will remove themselves when they are done decoding.

If the prepared or uncommitted transaction decides to abort, then
the decodeGroupLeader will set the decodeAbortPending flag allowing
the decodeGroupMembers to abort their decoding appropriately
---
 src/backend/replication/logical/logical.c | 161 +++++++++++++++++
 src/backend/storage/ipc/procarray.c       |  39 +++++
 src/backend/storage/lmgr/README           |  39 +++++
 src/backend/storage/lmgr/proc.c           | 282 ++++++++++++++++++++++++++++++
 src/include/replication/logical.h         |   2 +
 src/include/replication/reorderbuffer.h   |  15 ++
 src/include/storage/proc.h                |  25 +++
 src/include/storage/procarray.h           |   1 +
 8 files changed, 564 insertions(+)

diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c
index 3d8ad7ddf8..a9b043be88 100644
--- a/src/backend/replication/logical/logical.c
+++ b/src/backend/replication/logical/logical.c
@@ -1017,3 +1017,164 @@ LogicalConfirmReceivedLocation(XLogRecPtr lsn)
 		SpinLockRelease(&MyReplicationSlot->mutex);
 	}
 }
+
+bool
+LogicalLockTransaction(ReorderBufferTXN *txn)
+{
+	bool	ok = false;
+
+	/*
+	 * Prepared transactions and uncommitted transactions
+	 * that have modified catalogs need to interlock with
+	 * concurrent rollback to ensure that there are no
+	 * issues while decoding
+	 */
+
+	if (!rbtxn_has_catalog_changes(txn))
+		return true;
+
+	/*
+	 * Is it a prepared txn? Similar checks for uncommitted
+	 * transactions when we start supporting them
+	 */
+	if (!rbtxn_prepared(txn))
+		return true;
+
+	/* check cached status */
+	if (rbtxn_commit(txn))
+		return true;
+	if (rbtxn_rollback(txn))
+		return false;
+
+	/*
+	 * Find the PROC that is handling this XID and add ourself as a
+	 * decodeGroupMember
+	 */
+	if (MyProc->decodeGroupLeader == NULL)
+	{
+		PGPROC *proc = BecomeDecodeGroupLeader(txn->xid, rbtxn_prepared(txn));
+
+		/*
+		 * If decodeGroupLeader is NULL, then the only possibility
+		 * is that the transaction completed and went away
+		 */
+		if (proc == NULL)
+		{
+			Assert(!TransactionIdIsInProgress(txn->xid));
+			if (TransactionIdDidCommit(txn->xid))
+			{
+				txn->txn_flags |= RBTXN_COMMIT;
+				return true;
+			}
+			else
+			{
+				txn->txn_flags |= RBTXN_ROLLBACK;
+				return false;
+			}
+		}
+
+		/* Add ourself as a decodeGroupMember */
+		if (!BecomeDecodeGroupMember(proc, proc->pid, rbtxn_prepared(txn)))
+		{
+			Assert(!TransactionIdIsInProgress(txn->xid));
+			if (TransactionIdDidCommit(txn->xid))
+			{
+				txn->txn_flags |= RBTXN_COMMIT;
+				return true;
+			}
+			else
+			{
+				txn->txn_flags |= RBTXN_ROLLBACK;
+				return false;
+			}
+		}
+	}
+
+	/*
+	 * If we were able to add ourself, then Abort processing will
+	 * interlock with us. Check if the transaction is still around
+	 */
+	Assert(MyProc->decodeGroupLeader);
+
+	if (MyProc->decodeGroupLeader)
+	{
+		LWLock	   *leader_lwlock;
+
+		leader_lwlock = LockHashPartitionLockByProc(MyProc->decodeGroupLeader);
+		LWLockAcquire(leader_lwlock, LW_SHARED);
+		if (MyProc->decodeAbortPending)
+		{
+			/*
+			 * Remove ourself from the decodeGroupMembership and return
+			 * false so that the decoding plugin also initiates abort
+			 * processing
+			 */
+			LWLockRelease(leader_lwlock);
+			LWLockAcquire(leader_lwlock, LW_EXCLUSIVE);
+			RemoveDecodeGroupMemberLocked(MyProc->decodeGroupLeader);
+			/* reset the bool to let the leader know that we are going away */
+			MyProc->decodeAbortPending = false;
+			MyProc->decodeLocked = false;
+			txn->txn_flags |= RBTXN_ROLLBACK;
+			ok = false;
+		}
+		else
+		{
+			MyProc->decodeLocked = true;
+			ok = true;
+		}
+		LWLockRelease(leader_lwlock);
+	}
+	else
+		return false;
+
+	return ok;
+}
+
+void
+LogicalUnlockTransaction(ReorderBufferTXN *txn)
+{
+	LWLock	   *leader_lwlock;
+
+	/*
+	 * Prepared transactions and uncommitted transactions
+	 * that have modified catalogs need to interlock with
+	 * concurrent rollback to ensure that there are no
+	 * issues while decoding
+	 */
+
+	if (!rbtxn_has_catalog_changes(txn))
+		return;
+
+	/*
+	 * Is it a prepared txn? Similar checks for uncommitted
+	 * transactions when we start supporting them
+	 */
+	if (!rbtxn_prepared(txn))
+		return;
+
+	/* check cached status */
+	if (rbtxn_commit(txn))
+		return;
+	if (rbtxn_rollback(txn))
+		return;
+
+	Assert(MyProc->decodeGroupLeader);
+	leader_lwlock = LockHashPartitionLockByProc(MyProc->decodeGroupLeader);
+	LWLockAcquire(leader_lwlock, LW_SHARED);
+	if (MyProc->decodeAbortPending)
+	{
+		/*
+		 * Remove ourself from the decodeGroupMembership
+		 */
+		LWLockRelease(leader_lwlock);
+		LWLockAcquire(leader_lwlock, LW_EXCLUSIVE);
+		RemoveDecodeGroupMemberLocked(MyProc->decodeGroupLeader);
+		/* reset the bool to let the leader know that we are going away */
+		MyProc->decodeAbortPending = false;
+		txn->txn_flags |= RBTXN_ROLLBACK;
+	}
+	MyProc->decodeLocked = false;
+	LWLockRelease(leader_lwlock);
+	return;
+}
diff --git a/src/backend/storage/ipc/procarray.c b/src/backend/storage/ipc/procarray.c
index afe1c03aa3..2be2910207 100644
--- a/src/backend/storage/ipc/procarray.c
+++ b/src/backend/storage/ipc/procarray.c
@@ -2432,6 +2432,45 @@ BackendXidGetPid(TransactionId xid)
 	return result;
 }
 
+/*
+ * BackendXidGetProc -- get a backend's PGPROC given its XID
+ *
+ * Note that it is up to the caller to be sure that the question
+ * remains meaningful for long enough for the answer to be used ...
+ *
+ * Only main transaction Ids are considered.
+ *
+ */
+PGPROC *
+BackendXidGetProc(TransactionId xid)
+{
+	PGPROC			*result = NULL;
+	ProcArrayStruct *arrayP = procArray;
+	int			index;
+
+	if (xid == InvalidTransactionId)	/* never match invalid xid */
+		return 0;
+
+	LWLockAcquire(ProcArrayLock, LW_SHARED);
+
+	for (index = 0; index < arrayP->numProcs; index++)
+	{
+		int			pgprocno = arrayP->pgprocnos[index];
+		PGPROC *proc = &allProcs[pgprocno];
+		volatile PGXACT *pgxact = &allPgXact[pgprocno];
+
+		if (pgxact->xid == xid)
+		{
+			result = proc;
+			break;
+		}
+	}
+
+	LWLockRelease(ProcArrayLock);
+
+	return result;
+}
+
 /*
  * IsBackendPid -- is a given pid a running backend
  *
diff --git a/src/backend/storage/lmgr/README b/src/backend/storage/lmgr/README
index 56b0a12a3e..9742a348cf 100644
--- a/src/backend/storage/lmgr/README
+++ b/src/backend/storage/lmgr/README
@@ -679,6 +679,45 @@ worker, and the worker fails to join the lock group unless the given PGPROC
 still has the same PID and is still a lock group leader.  We assume that
 PIDs are not recycled quickly enough for this interlock to fail.
 
+Decode Group Locking
+--------------------
+
+We use an infrastructure which is very similar to the above group locking
+of parallel processes to create a group of backends that are performing
+logical decoding of an uncommitted or a prepared transaction.
+
+Decode Group locking adds five new members to each PGPROC:
+decodeGroupLeader, decodeGroupMembers, decodeGroupLink, decodeLocked and
+decodeAbortPending.  A PGPROC's decodeGroupLeader is NULL for processes
+not involved in logical decoding.  When a process wants to decode an
+uncommitted or prepared transaction then it finds out the PGPROC
+structure which is associated with that transaction id and makes that
+PGPROC structure as its decodeGroupLeader.  The decodeGroupMembers field
+is only used in the leader; it is a list of the member PGPROCs of the
+decode group (the leader and all backends decoding this transaction id).
+The decodeGroupLink field is the list link for this list. The decoding
+backend marks itself as decodeLocked while it is accessing catalog
+metadata for its decoding requirements via the
+LogicalLockTransaction API.  It resets the same via the
+LogicalUnlockTransaction API. Meanwhile, if the transaction id of this
+uncommitted or prepared transaction decides to abort then the PGPROC
+structure corresponding to it sets decodeAbortPending on itself and also
+on all the decodeGroupMembers entries.  The decodeGroupMembers entries
+stop decoding of this aborted transaction and exit. When all the
+decoding backends have exited then the aborting transaction goes ahead
+with its regular processing.
+
+All five of these fields are considered to be protected by a lock manager
+partition lock.  The partition lock that protects these fields within a given
+lock group is chosen by taking the leader's pgprocno modulo the number of lock
+manager partitions.  Holding this single lock allows safe manipulation of the
+decodeGroupMembers list for the lock group.
+
+The decodeGroupLeader's PGPROC and also its PID is accessible to each
+decoding backend.  And the decoding backend fails to join the decode
+lock group unless the given PGPROC still has the same PID and is still
+a decode group leader.  We assume that PIDs are not recycled quickly
+enough for this interlock to fail.
 
 User Locks (Advisory Locks)
 ---------------------------
diff --git a/src/backend/storage/lmgr/proc.c b/src/backend/storage/lmgr/proc.c
index 6f30e082b2..26d35c7807 100644
--- a/src/backend/storage/lmgr/proc.c
+++ b/src/backend/storage/lmgr/proc.c
@@ -267,6 +267,11 @@ InitProcGlobal(void)
 
 		/* Initialize lockGroupMembers list. */
 		dlist_init(&procs[i].lockGroupMembers);
+
+		/* Initialize decodeGroupMembers list. */
+		dlist_init(&procs[i].decodeGroupMembers);
+		procs[i].decodeAbortPending = false;
+		procs[i].decodeLocked = false;
 	}
 
 	/*
@@ -406,6 +411,12 @@ InitProcess(void)
 	Assert(MyProc->lockGroupLeader == NULL);
 	Assert(dlist_is_empty(&MyProc->lockGroupMembers));
 
+	/* Check that group decode fields are in a proper initial state. */
+	Assert(MyProc->decodeGroupLeader == NULL);
+	Assert(dlist_is_empty(&MyProc->decodeGroupMembers));
+	MyProc->decodeAbortPending = false;
+	MyProc->decodeLocked = false;
+
 	/* Initialize wait event information. */
 	MyProc->wait_event_info = 0;
 
@@ -581,6 +592,12 @@ InitAuxiliaryProcess(void)
 	Assert(MyProc->lockGroupLeader == NULL);
 	Assert(dlist_is_empty(&MyProc->lockGroupMembers));
 
+	/* Check that group decode fields are in a proper initial state. */
+	Assert(MyProc->decodeGroupLeader == NULL);
+	Assert(dlist_is_empty(&MyProc->decodeGroupMembers));
+	MyProc->decodeAbortPending = false;
+	MyProc->decodeLocked = false;
+
 	/*
 	 * We might be reusing a semaphore that belonged to a failed process. So
 	 * be careful and reinitialize its value here.  (This is not strictly
@@ -1887,3 +1904,268 @@ BecomeLockGroupMember(PGPROC *leader, int pid)
 
 	return ok;
 }
+
+/*
+ * BecomeDecodeGroupLeader - designate process as decode group leader
+ *
+ * Once this function has returned, other processes can join the decode group
+ * by calling BecomeDecodeGroupMember.
+ */
+PGPROC *
+BecomeDecodeGroupLeader(TransactionId xid, bool is_prepared)
+{
+	PGPROC *proc = NULL;
+	int			pid;
+	LWLock	   *leader_lwlock;
+
+	Assert(xid != InvalidTransactionId);
+
+
+	proc = BackendXidGetProc(xid);
+	if (proc)
+		pid = proc->pid;
+
+	/*
+	 * This proc will become decodeGroupLeader if it's
+	 * not already
+	 */
+	if (proc && proc->decodeGroupLeader != proc)
+	{
+		volatile PGXACT *pgxact;
+		/* Create single-member group, containing this proc. */
+		leader_lwlock = LockHashPartitionLockByProc(proc);
+		LWLockAcquire(leader_lwlock, LW_EXCLUSIVE);
+		/* recheck we are still the same */
+		pgxact = &ProcGlobal->allPgXact[proc->pgprocno];
+		if (proc->pid == pid && pgxact->xid == xid)
+		{
+			if (is_prepared)
+				Assert(pid == 0);
+			/* recheck if someone else did not already assign us */
+			if (proc->decodeGroupLeader != proc)
+			{
+				/* We had better not be a follower. */
+				Assert(proc->decodeGroupLeader == NULL);
+				proc->decodeGroupLeader = proc;
+				dlist_push_head(&proc->decodeGroupMembers,
+								&proc->decodeGroupLink);
+			}
+		}
+		else
+		{
+			/* proc entry is gone */
+			proc = NULL;
+		}
+		LWLockRelease(leader_lwlock);
+	}
+
+	elog(DEBUG1, "became group leader (%p)", proc);
+	return proc;
+}
+
+/*
+ * BecomeDecodeGroupMember - designate process as decode group member
+ *
+ * This is pretty straightforward except for the possibility that the leader
+ * whose group we're trying to join might exit before we manage to do so;
+ * and the PGPROC might get recycled for an unrelated process.  To avoid
+ * that, we require the caller to pass the PID of the intended PGPROC as
+ * an interlock.  Returns true if we successfully join the intended lock
+ * group, and false if not.
+ */
+bool
+BecomeDecodeGroupMember(PGPROC *leader, int pid, bool is_prepared)
+{
+	LWLock	   *leader_lwlock;
+	bool		ok = false;
+
+	/* Group leader can't become member of group */
+	Assert(MyProc != leader);
+
+	/* Can't already be a member of a group */
+	Assert(MyProc->decodeGroupLeader == NULL);
+
+	/* PID must be valid OR this is a prepared transaction. */
+	Assert(pid != 0 || is_prepared);
+
+	/*
+	 * Get lock protecting the group fields.  Note LockHashPartitionLockByProc
+	 * accesses leader->pgprocno in a PGPROC that might be free.  This is safe
+	 * because all PGPROCs' pgprocno fields are set during shared memory
+	 * initialization and never change thereafter; so we will acquire the
+	 * correct lock even if the leader PGPROC is in process of being recycled.
+	 */
+	leader_lwlock = LockHashPartitionLockByProc(leader);
+	LWLockAcquire(leader_lwlock, LW_EXCLUSIVE);
+
+	/* Is this the leader we're looking for? */
+	if (leader->pid == pid && leader->decodeGroupLeader == leader)
+	{
+		if (is_prepared)
+			Assert(pid == 0);
+		/* is the leader going away? */
+		if (leader->decodeAbortPending)
+			ok = false;
+		else
+		{
+			/* OK, join the group */
+			ok = true;
+			MyProc->decodeGroupLeader = leader;
+			dlist_push_tail(&leader->decodeGroupMembers, &MyProc->decodeGroupLink);
+		}
+	}
+	else
+		MyProc->decodeGroupLeader = NULL;
+	LWLockRelease(leader_lwlock);
+
+	elog(DEBUG1, "became group member (%p) to (%p)", MyProc, leader);
+	return ok;
+}
+
+/*
+ * Remove a decodeGroupMember from the decodeGroupMembership of
+ * decodeGroupLeader
+ * Acquire lock
+ */
+void
+RemoveDecodeGroupMember(PGPROC *leader)
+{
+	LWLock	   *leader_lwlock;
+
+	leader_lwlock = LockHashPartitionLockByProc(leader);
+	LWLockAcquire(leader_lwlock, LW_EXCLUSIVE);
+	RemoveDecodeGroupMemberLocked(leader);
+	LWLockRelease(leader_lwlock);
+
+	return;
+}
+
+/*
+ * Remove a decodeGroupMember from the decodeGroupMembership of
+ * decodeGroupLeader
+ * Assumes that the caller is holding appropriate lock
+ */
+void
+RemoveDecodeGroupMemberLocked(PGPROC *leader)
+{
+	Assert(!dlist_is_empty(&leader->decodeGroupMembers));
+	dlist_delete(&MyProc->decodeGroupLink);
+	/* leader links to itself, so never empty */
+	Assert(!dlist_is_empty(&leader->decodeGroupMembers));
+	MyProc->decodeGroupLeader = NULL;
+	elog(DEBUG1, "removed group member (%p) from (%p)", MyProc, leader);
+
+	return;
+}
+
+/*
+ * Indicate to all decodeGroupMembers that this transaction is
+ * going away.
+ *
+ * Wait for all decodeGroupMembers to ack back before returning
+ * from here but only in case of aborts.
+ *
+ * This function should be called *after* the proc has been
+ * removed from the procArray.
+ *
+ * If the transaction is committing, it's ok for the
+ * decoders to continue merrily. When it tries to lock this
+ * proc, it won't find it and check for transaction status
+ * and cache the commit status for future calls in
+ * LogicalLockTransaction
+ */
+void
+LogicalDecodeRemoveTransaction(PGPROC *leader, bool isCommit)
+{
+	LWLock	   *leader_lwlock;
+	dlist_mutable_iter change_i;
+	dlist_iter  iter;
+	PGPROC	   *proc;
+	bool		do_wait;
+
+	leader_lwlock = LockHashPartitionLockByProc(leader);
+	LWLockAcquire(leader_lwlock, LW_EXCLUSIVE);
+	/* mark ourself as aborting */
+	if (!isCommit)
+		leader->decodeAbortPending = true;
+
+	if (leader->decodeGroupLeader == NULL)
+	{
+		Assert(dlist_is_empty(&leader->decodeGroupMembers));
+		LWLockRelease(leader_lwlock);
+		return;
+	}
+
+recheck:
+	do_wait = false;
+	Assert(leader->decodeGroupLeader == leader);
+	Assert(!dlist_is_empty(&leader->decodeGroupMembers));
+	if (!isCommit)
+	{
+		dlist_foreach(iter, &leader->decodeGroupMembers)
+		{
+			proc = dlist_container(PGPROC, decodeGroupLink, iter.cur);
+			/* mark the proc to indicate abort is pending */
+			if (proc == leader)
+				continue;
+			if (!proc->decodeAbortPending)
+			{
+				proc->decodeAbortPending = true;
+				elog(DEBUG1, "marking group member (%p) from (%p) for abort",
+					 proc, leader);
+			}
+			/* if the proc is currently locked, wait */
+			if (proc->decodeLocked)
+				do_wait = true;
+		}
+
+		if (do_wait)
+		{
+			int			rc;
+			LWLockRelease(leader_lwlock);
+
+			elog(LOG, "Waiting for backends to abort decoding");
+			/*
+			 * Wait on our latch to allow decodeGroupMembers to
+			 * go away soon
+			 */
+			rc = WaitLatch(MyLatch,
+						   WL_LATCH_SET | WL_TIMEOUT | WL_POSTMASTER_DEATH,
+						   100L,
+						   WAIT_EVENT_PG_SLEEP);
+			ResetLatch(MyLatch);
+
+			/* emergency bailout if postmaster has died */
+			if (rc & WL_POSTMASTER_DEATH)
+				proc_exit(1);
+
+			CHECK_FOR_INTERRUPTS();
+
+			/* Recheck decodeGroupMembers */
+			LWLockAcquire(leader_lwlock, LW_EXCLUSIVE);
+			goto recheck;
+		}
+	}
+
+	/*
+	 * All backends exited cleanly in case of aborts above,
+	 * remove decodeGroupMembers now for both commit/abort cases
+	 */
+	Assert(leader->decodeGroupLeader == leader);
+	Assert(!dlist_is_empty(&leader->decodeGroupMembers));
+	dlist_foreach_modify(change_i, &leader->decodeGroupMembers)
+	{
+		proc = dlist_container(PGPROC, decodeGroupLink, change_i.cur);
+		Assert(!proc->decodeLocked);
+		dlist_delete(&proc->decodeGroupLink);
+		elog(DEBUG1, "deleting group member (%p) from (%p)",
+			 proc, leader);
+		proc->decodeGroupLeader = NULL;
+	}
+	Assert(dlist_is_empty(&leader->decodeGroupMembers));
+	leader->decodeGroupLeader = NULL;
+	leader->decodeAbortPending = false;
+	LWLockRelease(leader_lwlock);
+
+	return;
+}
diff --git a/src/include/replication/logical.h b/src/include/replication/logical.h
index 619c5f4d73..63b14367f0 100644
--- a/src/include/replication/logical.h
+++ b/src/include/replication/logical.h
@@ -117,6 +117,8 @@ extern void LogicalIncreaseXminForSlot(XLogRecPtr lsn, TransactionId xmin);
 extern void LogicalIncreaseRestartDecodingForSlot(XLogRecPtr current_lsn,
 									  XLogRecPtr restart_lsn);
 extern void LogicalConfirmReceivedLocation(XLogRecPtr lsn);
+extern bool LogicalLockTransaction(ReorderBufferTXN *txn);
+extern void LogicalUnlockTransaction(ReorderBufferTXN *txn);
 
 extern bool filter_by_origin_cb_wrapper(LogicalDecodingContext *ctx, RepOriginId origin_id);
 
diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h
index 177ef98e43..385bb486bb 100644
--- a/src/include/replication/reorderbuffer.h
+++ b/src/include/replication/reorderbuffer.h
@@ -141,6 +141,11 @@ typedef struct ReorderBufferChange
 #define RBTXN_HAS_CATALOG_CHANGES 0x0001
 #define RBTXN_IS_SUBXACT          0x0002
 #define RBTXN_IS_SERIALIZED       0x0004
+#define RBTXN_PREPARE             0x0008
+#define RBTXN_COMMIT_PREPARED     0x0010
+#define RBTXN_ROLLBACK_PREPARED   0x0020
+#define RBTXN_COMMIT              0x0040
+#define RBTXN_ROLLBACK            0x0080
 
 /* does the txn have catalog changes */
 #define rbtxn_has_catalog_changes(txn) (txn->txn_flags & RBTXN_HAS_CATALOG_CHANGES)
@@ -154,6 +159,16 @@ typedef struct ReorderBufferChange
  * nentries_mem == nentries.
  */
 #define rbtxn_is_serialized(txn)       (txn->txn_flags & RBTXN_IS_SERIALIZED)
+/* is this txn prepared? */
+#define rbtxn_prepared(txn)            (txn->txn_flags & RBTXN_PREPARE)
+/* was this prepared txn committed in the meanwhile? */
+#define rbtxn_commit_prepared(txn)     (txn->txn_flags & RBTXN_COMMIT_PREPARED)
+/* was this prepared txn aborted in the meanwhile? */
+#define rbtxn_rollback_prepared(txn)   (txn->txn_flags & RBTXN_ROLLBACK_PREPARED)
+/* was this txn committed in the meanwhile? */
+#define rbtxn_commit(txn)              (txn->txn_flags & RBTXN_COMMIT)
+/* was this prepared txn aborted in the meanwhile? */
+#define rbtxn_rollback(txn)            (txn->txn_flags & RBTXN_ROLLBACK)
 
 typedef struct ReorderBufferTXN
 {
diff --git a/src/include/storage/proc.h b/src/include/storage/proc.h
index 5c19a61dcf..fdfc582874 100644
--- a/src/include/storage/proc.h
+++ b/src/include/storage/proc.h
@@ -200,6 +200,26 @@ struct PGPROC
 	PGPROC	   *lockGroupLeader;	/* lock group leader, if I'm a member */
 	dlist_head	lockGroupMembers;	/* list of members, if I'm a leader */
 	dlist_node	lockGroupLink;	/* my member link, if I'm a member */
+
+	/*
+	 * Support for decoding groups.  Use LockHashPartitionLockByProc on the group
+	 * leader to get the LWLock protecting these fields.
+	 *
+	 * For prepared and uncommitted transactions, decoding backends working on
+	 * the same XID will link themselves up to the corresponding PGPROC
+	 * entry (decodeGroupLeader).
+	 *
+	 * They will remove themselves when they are done decoding.
+	 *
+	 * If the prepared or uncommitted transaction decides to abort, then
+	 * the decodeGroupLeader will set the decodeAbortPending flag allowing
+	 * the decodeGroupMembers to abort their decoding appropriately
+	 */
+	PGPROC	   *decodeGroupLeader;	/* decode group leader, if I'm a member */
+	dlist_head	decodeGroupMembers;	/* list of members, if I'm a leader */
+	dlist_node	decodeGroupLink;	/* my member link, if I'm a member */
+	bool		decodeLocked;		/* is it currently locked by this proc? */
+	bool		decodeAbortPending; /* is the decode group leader aborting? */
 };
 
 /* NOTE: "typedef struct PGPROC PGPROC" appears in storage/lock.h. */
@@ -326,5 +346,10 @@ extern PGPROC *AuxiliaryPidGetProc(int pid);
 
 extern void BecomeLockGroupLeader(void);
 extern bool BecomeLockGroupMember(PGPROC *leader, int pid);
+extern PGPROC *BecomeDecodeGroupLeader(TransactionId xid, bool is_prepared);
+extern bool BecomeDecodeGroupMember(PGPROC *leader, int pid, bool is_prepared);
+extern void RemoveDecodeGroupMember(PGPROC *leader);
+extern void RemoveDecodeGroupMemberLocked(PGPROC *leader);
+extern void LogicalDecodeRemoveTransaction(PGPROC *leader, bool isCommit);
 
 #endif							/* PROC_H */
diff --git a/src/include/storage/procarray.h b/src/include/storage/procarray.h
index 75bab2985f..776de2470e 100644
--- a/src/include/storage/procarray.h
+++ b/src/include/storage/procarray.h
@@ -97,6 +97,7 @@ extern bool HaveVirtualXIDsDelayingChkpt(VirtualTransactionId *vxids, int nvxids
 
 extern PGPROC *BackendPidGetProc(int pid);
 extern PGPROC *BackendPidGetProcWithLock(int pid);
+extern PGPROC *BackendXidGetProc(TransactionId xid);
 extern int	BackendXidGetPid(TransactionId xid);
 extern bool IsBackendPid(int pid);
 
-- 
2.14.3 (Apple Git-98)

