Split PGPROC 'links' field into two, for clarity

The field was mainly used for the position in a LOCK's wait queue, but
also as the position in a the freelist when the PGPROC entry was not
in use. The reuse saves some memory at the expense of readability,
which seems like a bad tradeoff. If we wanted to make the struct
smaller there's other things we could do, but we're actually just
discussing adding padding to the struct for performance reasons.

Reviewed-by: Bertrand Drouvot <bertranddrouvot.pg@gmail.com>
Discussion: https://www.postgresql.org/message-id/3dd6f70c-b94d-4428-8e75-74a7136396be@iki.fi

diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c
index eabc4d48208f320eba4b9c08e817df0610233e5d..e4340b596409d905a8417c7f676c902667f67678 100644 (file)
--- a/src/backend/access/transam/twophase.c
+++ b/src/backend/access/transam/twophase.c
@@ -447,7 +447,6 @@ MarkAsPreparingGuts(GlobalTransaction gxact, FullTransactionId fxid,
 
        /* Initialize the PGPROC entry */
        MemSet(proc, 0, sizeof(PGPROC));
-       dlist_node_init(&proc->links);
        proc->waitStatus = PROC_WAIT_STATUS_OK;
        if (LocalTransactionIdIsValid(MyProc->vxid.lxid))
        {
@@ -474,6 +473,7 @@ MarkAsPreparingGuts(GlobalTransaction gxact, FullTransactionId fxid,
        proc->lwWaiting = LW_WS_NOT_WAITING;
        proc->lwWaitMode = 0;
        proc->waitLock = NULL;
+       dlist_node_init(&proc->waitLink);
        proc->waitProcLock = NULL;
        pg_atomic_init_u64(&proc->waitStart, 0);
        for (i = 0; i < NUM_LOCK_PARTITIONS; i++)

diff --git a/src/backend/storage/lmgr/deadlock.c b/src/backend/storage/lmgr/deadlock.c
index c0c4ed57d9ec3eb2f427bf5c494d29c75388fab6..b8962d875b6570c58b8b5a02d68ac4aac9684655 100644 (file)
--- a/src/backend/storage/lmgr/deadlock.c
+++ b/src/backend/storage/lmgr/deadlock.c
@@ -262,7 +262,7 @@ DeadLockCheck(PGPROC *proc)
                /* Reset the queue and re-add procs in the desired order */
                dclist_init(waitQueue);
                for (int j = 0; j < nProcs; j++)
-                       dclist_push_tail(waitQueue, &procs[j]->links);
+                       dclist_push_tail(waitQueue, &procs[j]->waitLink);
 
 #ifdef DEBUG_DEADLOCK
                PrintLockQueue(lock, "rearranged to:");
@@ -504,7 +504,7 @@ FindLockCycleRecurse(PGPROC *checkProc,
         * If the process is waiting, there is an outgoing waits-for edge to each
         * process that blocks it.
         */
-       if (checkProc->links.next != NULL && checkProc->waitLock != NULL &&
+       if (!dlist_node_is_detached(&checkProc->waitLink) &&
                FindLockCycleRecurseMember(checkProc, checkProc, depth, softEdges,
                                                                   nSoftEdges))
                return true;
@@ -522,7 +522,7 @@ FindLockCycleRecurse(PGPROC *checkProc,
 
                memberProc = dlist_container(PGPROC, lockGroupLink, iter.cur);
 
-               if (memberProc->links.next != NULL && memberProc->waitLock != NULL &&
+               if (!dlist_node_is_detached(&memberProc->waitLink) && memberProc->waitLock != NULL &&
                        memberProc != checkProc &&
                        FindLockCycleRecurseMember(memberProc, checkProc, depth, softEdges,
                                                                           nSoftEdges))
@@ -715,7 +715,7 @@ FindLockCycleRecurseMember(PGPROC *checkProc,
                {
                        dclist_foreach(proc_iter, waitQueue)
                        {
-                               proc = dlist_container(PGPROC, links, proc_iter.cur);
+                               proc = dlist_container(PGPROC, waitLink, proc_iter.cur);
 
                                if (proc->lockGroupLeader == checkProcLeader)
                                        lastGroupMember = proc;
@@ -730,7 +730,7 @@ FindLockCycleRecurseMember(PGPROC *checkProc,
                {
                        PGPROC     *leader;
 
-                       proc = dlist_container(PGPROC, links, proc_iter.cur);
+                       proc = dlist_container(PGPROC, waitLink, proc_iter.cur);
 
                        leader = proc->lockGroupLeader == NULL ? proc :
                                proc->lockGroupLeader;
@@ -879,7 +879,7 @@ TopoSort(LOCK *lock,
        i = 0;
        dclist_foreach(proc_iter, waitQueue)
        {
-               proc = dlist_container(PGPROC, links, proc_iter.cur);
+               proc = dlist_container(PGPROC, waitLink, proc_iter.cur);
                topoProcs[i++] = proc;
        }
        Assert(i == queue_size);
@@ -1059,7 +1059,7 @@ PrintLockQueue(LOCK *lock, const char *info)
 
        dclist_foreach(proc_iter, waitQueue)
        {
-               PGPROC     *proc = dlist_container(PGPROC, links, proc_iter.cur);
+               PGPROC     *proc = dlist_container(PGPROC, waitLink, proc_iter.cur);
 
                printf(" %d", proc->pid);
        }

diff --git a/src/backend/storage/lmgr/lock.c b/src/backend/storage/lmgr/lock.c
index e1168ad3837a10b0195a1bb8b891e04047c9a127..d930c66cdbd7c14c658155b289086cd0d089ac99 100644 (file)
--- a/src/backend/storage/lmgr/lock.c
+++ b/src/backend/storage/lmgr/lock.c
@@ -2052,13 +2052,13 @@ RemoveFromWaitQueue(PGPROC *proc, uint32 hashcode)
 
        /* Make sure proc is waiting */
        Assert(proc->waitStatus == PROC_WAIT_STATUS_WAITING);
-       Assert(proc->links.next != NULL);
+       Assert(!dlist_node_is_detached(&proc->waitLink));
        Assert(waitLock);
        Assert(!dclist_is_empty(&waitLock->waitProcs));
        Assert(0 < lockmethodid && lockmethodid < lengthof(LockMethods));
 
        /* Remove proc from lock's wait queue */
-       dclist_delete_from_thoroughly(&waitLock->waitProcs, &proc->links);
+       dclist_delete_from_thoroughly(&waitLock->waitProcs, &proc->waitLink);
 
        /* Undo increments of request counts by waiting process */
        Assert(waitLock->nRequested > 0);
@@ -4143,7 +4143,7 @@ GetSingleProcBlockerStatusData(PGPROC *blocked_proc, BlockedProcsData *data)
        /* Collect PIDs from the lock's wait queue, stopping at blocked_proc */
        dclist_foreach(proc_iter, waitQueue)
        {
-               PGPROC     *queued_proc = dlist_container(PGPROC, links, proc_iter.cur);
+               PGPROC     *queued_proc = dlist_container(PGPROC, waitLink, proc_iter.cur);
 
                if (queued_proc == blocked_proc)
                        break;

diff --git a/src/backend/storage/lmgr/proc.c b/src/backend/storage/lmgr/proc.c
index fd8318bdf3d63efa7a22ccbbd5b9189ec92dce78..e2f34075d39e961ae6caf43a6d49abfc11c0389e 100644 (file)
--- a/src/backend/storage/lmgr/proc.c
+++ b/src/backend/storage/lmgr/proc.c
@@ -323,25 +323,25 @@ InitProcGlobal(void)
                if (i < MaxConnections)
                {
                        /* PGPROC for normal backend, add to freeProcs list */
-                       dlist_push_tail(&ProcGlobal->freeProcs, &proc->links);
+                       dlist_push_tail(&ProcGlobal->freeProcs, &proc->freeProcsLink);
                        proc->procgloballist = &ProcGlobal->freeProcs;
                }
                else if (i < MaxConnections + autovacuum_worker_slots + NUM_SPECIAL_WORKER_PROCS)
                {
                        /* PGPROC for AV or special worker, add to autovacFreeProcs list */
-                       dlist_push_tail(&ProcGlobal->autovacFreeProcs, &proc->links);
+                       dlist_push_tail(&ProcGlobal->autovacFreeProcs, &proc->freeProcsLink);
                        proc->procgloballist = &ProcGlobal->autovacFreeProcs;
                }
                else if (i < MaxConnections + autovacuum_worker_slots + NUM_SPECIAL_WORKER_PROCS + max_worker_processes)
                {
                        /* PGPROC for bgworker, add to bgworkerFreeProcs list */
-                       dlist_push_tail(&ProcGlobal->bgworkerFreeProcs, &proc->links);
+                       dlist_push_tail(&ProcGlobal->bgworkerFreeProcs, &proc->freeProcsLink);
                        proc->procgloballist = &ProcGlobal->bgworkerFreeProcs;
                }
                else if (i < MaxBackends)
                {
                        /* PGPROC for walsender, add to walsenderFreeProcs list */
-                       dlist_push_tail(&ProcGlobal->walsenderFreeProcs, &proc->links);
+                       dlist_push_tail(&ProcGlobal->walsenderFreeProcs, &proc->freeProcsLink);
                        proc->procgloballist = &ProcGlobal->walsenderFreeProcs;
                }
 
@@ -424,7 +424,7 @@ InitProcess(void)
 
        if (!dlist_is_empty(procgloballist))
        {
-               MyProc = dlist_container(PGPROC, links, dlist_pop_head_node(procgloballist));
+               MyProc = dlist_container(PGPROC, freeProcsLink, dlist_pop_head_node(procgloballist));
                SpinLockRelease(&ProcGlobal->freeProcsLock);
        }
        else
@@ -457,7 +457,7 @@ InitProcess(void)
         * Initialize all fields of MyProc, except for those previously
         * initialized by InitProcGlobal.
         */
-       dlist_node_init(&MyProc->links);
+       dlist_node_init(&MyProc->freeProcsLink);
        MyProc->waitStatus = PROC_WAIT_STATUS_OK;
        MyProc->fpVXIDLock = false;
        MyProc->fpLocalTransactionId = InvalidLocalTransactionId;
@@ -479,6 +479,7 @@ InitProcess(void)
        MyProc->lwWaiting = LW_WS_NOT_WAITING;
        MyProc->lwWaitMode = 0;
        MyProc->waitLock = NULL;
+       dlist_node_init(&MyProc->waitLink);
        MyProc->waitProcLock = NULL;
        pg_atomic_write_u64(&MyProc->waitStart, 0);
 #ifdef USE_ASSERT_CHECKING
@@ -658,7 +659,7 @@ InitAuxiliaryProcess(void)
         * Initialize all fields of MyProc, except for those previously
         * initialized by InitProcGlobal.
         */
-       dlist_node_init(&MyProc->links);
+       dlist_node_init(&MyProc->freeProcsLink);
        MyProc->waitStatus = PROC_WAIT_STATUS_OK;
        MyProc->fpVXIDLock = false;
        MyProc->fpLocalTransactionId = InvalidLocalTransactionId;
@@ -675,6 +676,7 @@ InitAuxiliaryProcess(void)
        MyProc->lwWaiting = LW_WS_NOT_WAITING;
        MyProc->lwWaitMode = 0;
        MyProc->waitLock = NULL;
+       dlist_node_init(&MyProc->waitLink);
        MyProc->waitProcLock = NULL;
        pg_atomic_write_u64(&MyProc->waitStart, 0);
 #ifdef USE_ASSERT_CHECKING
@@ -835,7 +837,7 @@ LockErrorCleanup(void)
        partitionLock = LockHashPartitionLock(lockAwaited->hashcode);
        LWLockAcquire(partitionLock, LW_EXCLUSIVE);
 
-       if (!dlist_node_is_detached(&MyProc->links))
+       if (!dlist_node_is_detached(&MyProc->waitLink))
        {
                /* We could not have been granted the lock yet */
                RemoveFromWaitQueue(MyProc, lockAwaited->hashcode);
@@ -967,7 +969,7 @@ ProcKill(int code, Datum arg)
 
                                /* Leader exited first; return its PGPROC. */
                                SpinLockAcquire(&ProcGlobal->freeProcsLock);
-                               dlist_push_head(procgloballist, &leader->links);
+                               dlist_push_head(procgloballist, &leader->freeProcsLink);
                                SpinLockRelease(&ProcGlobal->freeProcsLock);
                        }
                }
@@ -1012,7 +1014,7 @@ ProcKill(int code, Datum arg)
                Assert(dlist_is_empty(&proc->lockGroupMembers));
 
                /* Return PGPROC structure (and semaphore) to appropriate freelist */
-               dlist_push_tail(procgloballist, &proc->links);
+               dlist_push_tail(procgloballist, &proc->freeProcsLink);
        }
 
        /* Update shared estimate of spins_per_delay */
@@ -1201,7 +1203,7 @@ JoinWaitQueue(LOCALLOCK *locallock, LockMethod lockMethodTable, bool dontWait)
 
                dclist_foreach(iter, waitQueue)
                {
-                       PGPROC     *proc = dlist_container(PGPROC, links, iter.cur);
+                       PGPROC     *proc = dlist_container(PGPROC, waitLink, iter.cur);
 
                        /*
                         * If we're part of the same locking group as this waiter, its
@@ -1265,9 +1267,9 @@ JoinWaitQueue(LOCALLOCK *locallock, LockMethod lockMethodTable, bool dontWait)
         * Insert self into queue, at the position determined above.
         */
        if (insert_before)
-               dclist_insert_before(waitQueue, &insert_before->links, &MyProc->links);
+               dclist_insert_before(waitQueue, &insert_before->waitLink, &MyProc->waitLink);
        else
-               dclist_push_tail(waitQueue, &MyProc->links);
+               dclist_push_tail(waitQueue, &MyProc->waitLink);
 
        lock->waitMask |= LOCKBIT_ON(lockmode);
 
@@ -1689,7 +1691,7 @@ ProcSleep(LOCALLOCK *locallock)
 /*
  * ProcWakeup -- wake up a process by setting its latch.
  *
- *      Also remove the process from the wait queue and set its links invalid.
+ *      Also remove the process from the wait queue and set its waitLink invalid.
  *
  * The appropriate lock partition lock must be held by caller.
  *
@@ -1701,13 +1703,13 @@ ProcSleep(LOCALLOCK *locallock)
 void
 ProcWakeup(PGPROC *proc, ProcWaitStatus waitStatus)
 {
-       if (dlist_node_is_detached(&proc->links))
+       if (dlist_node_is_detached(&proc->waitLink))
                return;
 
        Assert(proc->waitStatus == PROC_WAIT_STATUS_WAITING);
 
        /* Remove process from wait queue */
-       dclist_delete_from_thoroughly(&proc->waitLock->waitProcs, &proc->links);
+       dclist_delete_from_thoroughly(&proc->waitLock->waitProcs, &proc->waitLink);
 
        /* Clean up process' state and pass it the ok/fail signal */
        proc->waitLock = NULL;
@@ -1738,7 +1740,7 @@ ProcLockWakeup(LockMethod lockMethodTable, LOCK *lock)
 
        dclist_foreach_modify(miter, waitQueue)
        {
-               PGPROC     *proc = dlist_container(PGPROC, links, miter.cur);
+               PGPROC     *proc = dlist_container(PGPROC, waitLink, miter.cur);
                LOCKMODE        lockmode = proc->waitLockMode;
 
                /*
@@ -1802,8 +1804,7 @@ CheckDeadLock(void)
         * We check by looking to see if we've been unlinked from the wait queue.
         * This is safe because we hold the lock partition lock.
         */
-       if (MyProc->links.prev == NULL ||
-               MyProc->links.next == NULL)
+       if (dlist_node_is_detached(&MyProc->waitLink))
        {
                result = DS_NO_DEADLOCK;
                goto check_done;

diff --git a/src/include/storage/proc.h b/src/include/storage/proc.h
index 23e5cd98161c0bd7f4fb11aa2a2ffd9398412725..e165b414241fd2e37c4f7b6f1520b2758556a092 100644 (file)
--- a/src/include/storage/proc.h
+++ b/src/include/storage/proc.h
@@ -154,10 +154,6 @@ typedef enum
  * Each backend has a PGPROC struct in shared memory.  There is also a list of
  * currently-unused PGPROC structs that will be reallocated to new backends.
  *
- * links: list link for any list the PGPROC is in.  When waiting for a lock,
- * the PGPROC is linked into that lock's waitProcs queue.  A recycled PGPROC
- * is linked into ProcGlobal's freeProcs list.
- *
  * Note: twophase.c also sets up a dummy PGPROC struct for each currently
  * prepared transaction.  These PGPROCs appear in the ProcArray data structure
  * so that the prepared transactions appear to be still running and are
@@ -184,8 +180,9 @@ typedef enum
  */
 struct PGPROC
 {
-       dlist_node      links;                  /* list link if process is in a list */
        dlist_head *procgloballist; /* procglobal list that owns this PGPROC */
+       dlist_node      freeProcsLink;  /* link in procgloballist, when in recycled
+                                                                * state */
 
        PGSemaphore sem;                        /* ONE semaphore to sleep on */
        ProcWaitStatus waitStatus;
@@ -263,6 +260,7 @@ struct PGPROC
        /* Info about lock the process is currently waiting for, if any. */
        /* waitLock and waitProcLock are NULL if not currently waiting. */
        LOCK       *waitLock;           /* Lock object we're sleeping on ... */
+       dlist_node      waitLink;               /* position in waitLock->waitProcs queue */
        PROCLOCK   *waitProcLock;       /* Per-holder info for awaited lock */
        LOCKMODE        waitLockMode;   /* type of lock we're waiting for */
        LOCKMASK        heldLocks;              /* bitmask for lock types already held on this