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

diff --git a/src/backend/storage/ipc/procarray.c b/src/backend/storage/ipc/procarray.c
index 42a89fc5dc9ff255c7a114a395d819390add9b3e..085bd1e40778ed4a56b812e3838e3ddf193130dd 100644 (file)
--- a/src/backend/storage/ipc/procarray.c
+++ b/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++)
        {
-               /*
-                * If we are the first PGPROC or if we have found our right position
-                * in the array, break
-                */
-               if ((arrayP->pgprocnos[index] == -1) || (arrayP->pgprocnos[index] > proc->pgprocno))
+               int                     procno PG_USED_FOR_ASSERTS_ONLY = arrayP->pgprocnos[index];
+
+               Assert(procno >= 0 && procno < (arrayP->maxProcs + NUM_AUXILIARY_PROCS));
+               Assert(allProcs[procno].pgxactoff == index);
+
+               /* 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],
-                       (arrayP->numProcs - index) * sizeof(*arrayP->pgprocnos));
-       memmove(&ProcGlobal->xids[index + 1], &ProcGlobal->xids[index],
-                       (arrayP->numProcs - index) * sizeof(*ProcGlobal->xids));
-       memmove(&ProcGlobal->subxidStates[index + 1], &ProcGlobal->subxidStates[index],
-                       (arrayP->numProcs - index) * sizeof(*ProcGlobal->subxidStates));
-       memmove(&ProcGlobal->statusFlags[index + 1], &ProcGlobal->statusFlags[index],
-                       (arrayP->numProcs - index) * sizeof(*ProcGlobal->statusFlags));
+       movecount = arrayP->numProcs - index;
+       memmove(&arrayP->pgprocnos[index + 1],
+                       &arrayP->pgprocnos[index],
+                       movecount * sizeof(*arrayP->pgprocnos));
+       memmove(&ProcGlobal->xids[index + 1],
+                       &ProcGlobal->xids[index],
+                       movecount * sizeof(*ProcGlobal->xids));
+       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->subxidStates[proc->pgxactoff].overflowed = false;
-               ProcGlobal->subxidStates[proc->pgxactoff].count = 0;
+               ProcGlobal->xids[myoff] = InvalidTransactionId;
+               ProcGlobal->subxidStates[myoff].overflowed = false;
+               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->subxidStates[proc->pgxactoff].count == 0));
-       Assert(TransactionIdIsValid(ProcGlobal->subxidStates[proc->pgxactoff].overflowed == false));
-       ProcGlobal->statusFlags[proc->pgxactoff] = 0;
+       Assert(!TransactionIdIsValid(ProcGlobal->xids[myoff]));
+       Assert(ProcGlobal->subxidStates[myoff].count == 0);
+       Assert(ProcGlobal->subxidStates[myoff].overflowed == false);
+
+       ProcGlobal->statusFlags[myoff] = 0;
+
+       /* Keep the PGPROC array sorted. See notes above */
+       movecount = arrayP->numProcs - myoff - 1;
+       memmove(&arrayP->pgprocnos[myoff],
+                       &arrayP->pgprocnos[myoff + 1],
+                       movecount * sizeof(*arrayP->pgprocnos));
+       memmove(&ProcGlobal->xids[myoff],
+                       &ProcGlobal->xids[myoff + 1],
+                       movecount * sizeof(*ProcGlobal->xids));
+       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--;
 
-       for (index = 0; index < arrayP->numProcs; index++)
+       /*
+        * Adjust pgxactoff of following procs for removed PGPROC (note that
+        * numProcs already has been decremented).
+        */
+       for (int index = myoff; 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],
-                                       (arrayP->numProcs - index - 1) * sizeof(*arrayP->pgprocnos));
-                       memmove(&ProcGlobal->xids[index], &ProcGlobal->xids[index + 1],
-                                       (arrayP->numProcs - index - 1) * sizeof(*ProcGlobal->xids));
-                       memmove(&ProcGlobal->subxidStates[index], &ProcGlobal->subxidStates[index + 1],
-                                       (arrayP->numProcs - index - 1) * sizeof(*ProcGlobal->subxidStates));
-                       memmove(&ProcGlobal->statusFlags[index], &ProcGlobal->statusFlags[index + 1],
-                                       (arrayP->numProcs - index - 1) * sizeof(*ProcGlobal->statusFlags));
-
-                       arrayP->pgprocnos[arrayP->numProcs - 1] = -1;   /* for debugging */
-                       arrayP->numProcs--;
-
-                       /* adjust for removed PGPROC */
-                       for (; index < arrayP->numProcs; index++)
-                               allProcs[arrayP->pgprocnos[index]].pgxactoff--;
+               int                     procno = arrayP->pgprocnos[index];
 
-                       /*
-                        * Release in reversed acquisition order, to reduce frequency of
-                        * having to wait for XidGenLock while holding ProcArrayLock.
-                        */
-                       LWLockRelease(XidGenLock);
-                       LWLockRelease(ProcArrayLock);
-                       return;
-               }
+               Assert(procno >= 0 && procno < (arrayP->maxProcs + NUM_AUXILIARY_PROCS));
+               Assert(allProcs[procno].pgxactoff - 1 == index);
+
+               allProcs[procno].pgxactoff = index;
        }
 
-       /* Oops */
+       /*
+        * Release in reversed acquisition order, to reduce frequency of having to
+        * wait for XidGenLock while holding ProcArrayLock.
+        */
        LWLockRelease(XidGenLock);
        LWLockRelease(ProcArrayLock);
-
-       elog(LOG, "failed to find proc %p in ProcArray", proc);
 }