Improve and cleanup ProcArrayAdd(), ProcArrayRemove().

941697c3c1a changed ProcArrayAdd()/Remove() substantially. As reported by
zhanyi, I missed that due to the introduction of the PGPROC->pgxactoff
ProcArrayRemove() does not need to search for the position in
procArray->pgprocnos anymore - that's pgxactoff. Remove the search loop.

The removal of the search loop reduces assertion coverage a bit - but we can
easily do better than before by adding assertions to other loops over
pgprocnos elements.

Also do a bit of cleanup, mostly by reducing line lengths by introducing local
helper variables and adding newlines.

Author: zhanyi <w@hidva.com>
Author: Andres Freund <andres@anarazel.de>
Discussion: https://postgr.es/m/tencent_5624AA3B116B3D1C31CA9744@qq.com

src/backend/storage/ipc/procarray.c

@@ -446,6 +446,7 @@ ProcArrayAdd(PGPROC *proc)
 {
 	ProcArrayStruct *arrayP = procArray;
 	int			index;
+	int			movecount;
 
 	/* See ProcGlobal comment explaining why both locks are held */
 	LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
@@ -474,33 +475,48 @@ ProcArrayAdd(PGPROC *proc)
 	 */
 	for (index = 0; index < arrayP->numProcs; index++)
 	{
-		/*
+		int			procno PG_USED_FOR_ASSERTS_ONLY = arrayP->pgprocnos[index];
-		 * If we are the first PGPROC or if we have found our right position
+
-		 * in the array, break
+		Assert(procno >= 0 && procno < (arrayP->maxProcs + NUM_AUXILIARY_PROCS));
-		 */
+		Assert(allProcs[procno].pgxactoff == index);
-		if ((arrayP->pgprocnos[index] == -1) || (arrayP->pgprocnos[index] > proc->pgprocno))
+
+		/* If we have found our right position in the array, break */
+		if (arrayP->pgprocnos[index] > proc->pgprocno)
 			break;
 	}
 
-	memmove(&arrayP->pgprocnos[index + 1], &arrayP->pgprocnos[index],
+	movecount = arrayP->numProcs - index;
-			(arrayP->numProcs - index) * sizeof(*arrayP->pgprocnos));
+	memmove(&arrayP->pgprocnos[index + 1],
-	memmove(&ProcGlobal->xids[index + 1], &ProcGlobal->xids[index],
+			&arrayP->pgprocnos[index],
-			(arrayP->numProcs - index) * sizeof(*ProcGlobal->xids));
+			movecount * sizeof(*arrayP->pgprocnos));
-	memmove(&ProcGlobal->subxidStates[index + 1], &ProcGlobal->subxidStates[index],
+	memmove(&ProcGlobal->xids[index + 1],
-			(arrayP->numProcs - index) * sizeof(*ProcGlobal->subxidStates));
+			&ProcGlobal->xids[index],
-	memmove(&ProcGlobal->statusFlags[index + 1], &ProcGlobal->statusFlags[index],
+			movecount * sizeof(*ProcGlobal->xids));
-			(arrayP->numProcs - index) * sizeof(*ProcGlobal->statusFlags));
+	memmove(&ProcGlobal->subxidStates[index + 1],
+			&ProcGlobal->subxidStates[index],
+			movecount * sizeof(*ProcGlobal->subxidStates));
+	memmove(&ProcGlobal->statusFlags[index + 1],
+			&ProcGlobal->statusFlags[index],
+			movecount * sizeof(*ProcGlobal->statusFlags));
 
 	arrayP->pgprocnos[index] = proc->pgprocno;
+	proc->pgxactoff = index;
 	ProcGlobal->xids[index] = proc->xid;
 	ProcGlobal->subxidStates[index] = proc->subxidStatus;
 	ProcGlobal->statusFlags[index] = proc->statusFlags;
 
 	arrayP->numProcs++;
 
+	/* adjust pgxactoff for all following PGPROCs */
+	index++;
 	for (; index < arrayP->numProcs; index++)
 	{
-		allProcs[arrayP->pgprocnos[index]].pgxactoff = index;
+		int			procno = arrayP->pgprocnos[index];
+
+		Assert(procno >= 0 && procno < (arrayP->maxProcs + NUM_AUXILIARY_PROCS));
+		Assert(allProcs[procno].pgxactoff == index - 1);
+
+		allProcs[procno].pgxactoff = index;
 	}
 
 	/*
@@ -525,7 +541,8 @@ void
 ProcArrayRemove(PGPROC *proc, TransactionId latestXid)
 {
 	ProcArrayStruct *arrayP = procArray;
-	int			index;
+	int			myoff;
+	int			movecount;
 
 #ifdef XIDCACHE_DEBUG
 	/* dump stats at backend shutdown, but not prepared-xact end */
@@ -537,11 +554,14 @@ ProcArrayRemove(PGPROC *proc, TransactionId latestXid)
 	LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
 	LWLockAcquire(XidGenLock, LW_EXCLUSIVE);
 
-	Assert(ProcGlobal->allProcs[arrayP->pgprocnos[proc->pgxactoff]].pgxactoff == proc->pgxactoff);
+	myoff = proc->pgxactoff;
+
+	Assert(myoff >= 0 && myoff < arrayP->numProcs);
+	Assert(ProcGlobal->allProcs[arrayP->pgprocnos[myoff]].pgxactoff == myoff);
 
 	if (TransactionIdIsValid(latestXid))
 	{
-		Assert(TransactionIdIsValid(ProcGlobal->xids[proc->pgxactoff]));
+		Assert(TransactionIdIsValid(ProcGlobal->xids[myoff]));
 
 		/* Advance global latestCompletedXid while holding the lock */
 		MaintainLatestCompletedXid(latestXid);
@@ -549,57 +569,60 @@ ProcArrayRemove(PGPROC *proc, TransactionId latestXid)
 		/* Same with xactCompletionCount  */
 		ShmemVariableCache->xactCompletionCount++;
 
-		ProcGlobal->xids[proc->pgxactoff] = 0;
+		ProcGlobal->xids[myoff] = InvalidTransactionId;
-		ProcGlobal->subxidStates[proc->pgxactoff].overflowed = false;
+		ProcGlobal->subxidStates[myoff].overflowed = false;
-		ProcGlobal->subxidStates[proc->pgxactoff].count = 0;
+		ProcGlobal->subxidStates[myoff].count = 0;
 	}
 	else
 	{
 		/* Shouldn't be trying to remove a live transaction here */
-		Assert(!TransactionIdIsValid(ProcGlobal->xids[proc->pgxactoff]));
+		Assert(!TransactionIdIsValid(ProcGlobal->xids[myoff]));
 	}
 
-	Assert(TransactionIdIsValid(ProcGlobal->xids[proc->pgxactoff] == 0));
+	Assert(!TransactionIdIsValid(ProcGlobal->xids[myoff]));
-	Assert(TransactionIdIsValid(ProcGlobal->subxidStates[proc->pgxactoff].count == 0));
+	Assert(ProcGlobal->subxidStates[myoff].count == 0);
-	Assert(TransactionIdIsValid(ProcGlobal->subxidStates[proc->pgxactoff].overflowed == false));
+	Assert(ProcGlobal->subxidStates[myoff].overflowed == false);
-	ProcGlobal->statusFlags[proc->pgxactoff] = 0;
+
+	ProcGlobal->statusFlags[myoff] = 0;
 
-	for (index = 0; index < arrayP->numProcs; index++)
-	{
-		if (arrayP->pgprocnos[index] == proc->pgprocno)
-		{
 	/* Keep the PGPROC array sorted. See notes above */
-			memmove(&arrayP->pgprocnos[index], &arrayP->pgprocnos[index + 1],
+	movecount = arrayP->numProcs - myoff - 1;
-					(arrayP->numProcs - index - 1) * sizeof(*arrayP->pgprocnos));
+	memmove(&arrayP->pgprocnos[myoff],
-			memmove(&ProcGlobal->xids[index], &ProcGlobal->xids[index + 1],
+			&arrayP->pgprocnos[myoff + 1],
-					(arrayP->numProcs - index - 1) * sizeof(*ProcGlobal->xids));
+			movecount * sizeof(*arrayP->pgprocnos));
-			memmove(&ProcGlobal->subxidStates[index], &ProcGlobal->subxidStates[index + 1],
+	memmove(&ProcGlobal->xids[myoff],
-					(arrayP->numProcs - index - 1) * sizeof(*ProcGlobal->subxidStates));
+			&ProcGlobal->xids[myoff + 1],
-			memmove(&ProcGlobal->statusFlags[index], &ProcGlobal->statusFlags[index + 1],
+			movecount * sizeof(*ProcGlobal->xids));
-					(arrayP->numProcs - index - 1) * sizeof(*ProcGlobal->statusFlags));
+	memmove(&ProcGlobal->subxidStates[myoff],
+			&ProcGlobal->subxidStates[myoff + 1],
+			movecount * sizeof(*ProcGlobal->subxidStates));
+	memmove(&ProcGlobal->statusFlags[myoff],
+			&ProcGlobal->statusFlags[myoff + 1],
+			movecount * sizeof(*ProcGlobal->statusFlags));
 
 	arrayP->pgprocnos[arrayP->numProcs - 1] = -1;	/* for debugging */
 	arrayP->numProcs--;
 
-			/* adjust for removed PGPROC */
+	/*
-			for (; index < arrayP->numProcs; index++)
+	 * Adjust pgxactoff of following procs for removed PGPROC (note that
-				allProcs[arrayP->pgprocnos[index]].pgxactoff--;
+	 * numProcs already has been decremented).
+	 */
+	for (int index = myoff; index < arrayP->numProcs; index++)
+	{
+		int			procno = arrayP->pgprocnos[index];
+
+		Assert(procno >= 0 && procno < (arrayP->maxProcs + NUM_AUXILIARY_PROCS));
+		Assert(allProcs[procno].pgxactoff - 1 == index);
+
+		allProcs[procno].pgxactoff = index;
+	}
 
 	/*
-			 * Release in reversed acquisition order, to reduce frequency of
+	 * Release in reversed acquisition order, to reduce frequency of having to
-			 * having to wait for XidGenLock while holding ProcArrayLock.
+	 * wait for XidGenLock while holding ProcArrayLock.
 	 */
 	LWLockRelease(XidGenLock);
 	LWLockRelease(ProcArrayLock);
-			return;
-		}
-	}
-
-	/* Oops */
-	LWLockRelease(XidGenLock);
-	LWLockRelease(ProcArrayLock);
-
-	elog(LOG, "failed to find proc %p in ProcArray", proc);
 }