From 53f0dc9b79aa072e2b49d41be1dc63f3e838aa2b Mon Sep 17 00:00:00 2001
From: Andres Freund <andres@anarazel.de>
Date: Wed, 19 Feb 2020 12:23:28 -0800
Subject: [PATCH v1 2/6] Use dlists instead of SHM_QUEUE for heavyweight locks.

Todo:
- Consider using singly linked list?
- Consider removing PROC_QUEUE - the size doesn't seem important.
---
 src/include/storage/lock.h            |   9 +-
 src/include/storage/proc.h            |  16 +--
 src/backend/access/transam/twophase.c |   4 +-
 src/backend/storage/lmgr/deadlock.c   |  51 +++----
 src/backend/storage/lmgr/lock.c       | 121 ++++++----------
 src/backend/storage/lmgr/proc.c       | 192 +++++++++++---------------
 6 files changed, 161 insertions(+), 232 deletions(-)

diff --git a/src/include/storage/lock.h b/src/include/storage/lock.h
index bb8e4e6e5b7..3569f145092 100644
--- a/src/include/storage/lock.h
+++ b/src/include/storage/lock.h
@@ -18,6 +18,7 @@
 #error "lock.h may not be included from frontend code"
 #endif
 
+#include "lib/ilist.h"
 #include "storage/backendid.h"
 #include "storage/lockdefs.h"
 #include "storage/lwlock.h"
@@ -28,7 +29,7 @@ typedef struct PGPROC PGPROC;
 
 typedef struct PROC_QUEUE
 {
-	SHM_QUEUE	links;			/* head of list of PGPROC objects */
+	dlist_head	links;			/* list of PGPROC objects */
 	int			size;			/* number of entries in list */
 } PROC_QUEUE;
 
@@ -292,7 +293,7 @@ typedef struct LOCK
 	/* data */
 	LOCKMASK	grantMask;		/* bitmask for lock types already granted */
 	LOCKMASK	waitMask;		/* bitmask for lock types awaited */
-	SHM_QUEUE	procLocks;		/* list of PROCLOCK objects assoc. with lock */
+	dlist_head	procLocks;		/* list of PROCLOCK objects assoc. with lock */
 	PROC_QUEUE	waitProcs;		/* list of PGPROC objects waiting on lock */
 	int			requested[MAX_LOCKMODES];	/* counts of requested locks */
 	int			nRequested;		/* total of requested[] array */
@@ -353,8 +354,8 @@ typedef struct PROCLOCK
 	PGPROC	   *groupLeader;	/* proc's lock group leader, or proc itself */
 	LOCKMASK	holdMask;		/* bitmask for lock types currently held */
 	LOCKMASK	releaseMask;	/* bitmask for lock types to be released */
-	SHM_QUEUE	lockLink;		/* list link in LOCK's list of proclocks */
-	SHM_QUEUE	procLink;		/* list link in PGPROC's list of proclocks */
+	dlist_node	lockLink;		/* list link in LOCK's list of proclocks */
+	dlist_node	procLink;		/* list link in PGPROC's list of proclocks */
 } PROCLOCK;
 
 #define PROCLOCK_LOCKMETHOD(proclock) \
diff --git a/src/include/storage/proc.h b/src/include/storage/proc.h
index d21780108bb..2ba37f250de 100644
--- a/src/include/storage/proc.h
+++ b/src/include/storage/proc.h
@@ -95,8 +95,8 @@ struct XidCache
 struct PGPROC
 {
 	/* proc->links MUST BE FIRST IN STRUCT (see ProcSleep,ProcWakeup,etc) */
-	SHM_QUEUE	links;			/* list link if process is in a list */
-	PGPROC	  **procgloballist; /* procglobal list that owns this PGPROC */
+	dlist_node	links;			/* list link if process is in a list */
+	dlist_head *procgloballist; /* procglobal list that owns this PGPROC */
 
 	PGSemaphore sem;			/* ONE semaphore to sleep on */
 	int			waitStatus;		/* STATUS_WAITING, STATUS_OK or STATUS_ERROR */
@@ -157,7 +157,7 @@ struct PGPROC
 	 * linked into one of these lists, according to the partition number of
 	 * their lock.
 	 */
-	SHM_QUEUE	myProcLocks[NUM_LOCK_PARTITIONS];
+	dlist_head	myProcLocks[NUM_LOCK_PARTITIONS];
 
 	struct XidCache subxids;	/* cache for subtransaction XIDs */
 
@@ -250,13 +250,13 @@ typedef struct PROC_HDR
 	/* Length of allProcs array */
 	uint32		allProcCount;
 	/* Head of list of free PGPROC structures */
-	PGPROC	   *freeProcs;
+	dlist_head	freeProcs;
 	/* Head of list of autovacuum's free PGPROC structures */
-	PGPROC	   *autovacFreeProcs;
+	dlist_head autovacFreeProcs;
 	/* Head of list of bgworker free PGPROC structures */
-	PGPROC	   *bgworkerFreeProcs;
+	dlist_head bgworkerFreeProcs;
 	/* Head of list of walsender free PGPROC structures */
-	PGPROC	   *walsenderFreeProcs;
+	dlist_head walsenderFreeProcs;
 	/* First pgproc waiting for group XID clear */
 	pg_atomic_uint32 procArrayGroupFirst;
 	/* First pgproc waiting for group transaction status update */
@@ -318,7 +318,7 @@ extern void ProcReleaseLocks(bool isCommit);
 
 extern void ProcQueueInit(PROC_QUEUE *queue);
 extern int	ProcSleep(LOCALLOCK *locallock, LockMethod lockMethodTable);
-extern PGPROC *ProcWakeup(PGPROC *proc, int waitStatus);
+extern void ProcWakeup(PGPROC *proc, int waitStatus);
 extern void ProcLockWakeup(LockMethod lockMethodTable, LOCK *lock);
 extern void CheckDeadLockAlert(void);
 extern bool IsWaitingForLock(void);
diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c
index 5adf956f413..39d7e5463c1 100644
--- a/src/backend/access/transam/twophase.c
+++ b/src/backend/access/transam/twophase.c
@@ -459,7 +459,7 @@ MarkAsPreparingGuts(GlobalTransaction gxact, TransactionId xid, const char *gid,
 	/* Initialize the PGPROC entry */
 	MemSet(proc, 0, sizeof(PGPROC));
 	proc->pgprocno = gxact->pgprocno;
-	SHMQueueElemInit(&(proc->links));
+	dlist_node_init(&proc->links);
 	proc->waitStatus = STATUS_OK;
 	/* We set up the gxact's VXID as InvalidBackendId/XID */
 	proc->lxid = (LocalTransactionId) xid;
@@ -478,7 +478,7 @@ MarkAsPreparingGuts(GlobalTransaction gxact, TransactionId xid, const char *gid,
 	proc->waitLock = NULL;
 	proc->waitProcLock = NULL;
 	for (i = 0; i < NUM_LOCK_PARTITIONS; i++)
-		SHMQueueInit(&(proc->myProcLocks[i]));
+		dlist_init(&proc->myProcLocks[i]);
 	/* subxid data must be filled later by GXactLoadSubxactData */
 	pgxact->overflowed = false;
 	pgxact->nxids = 0;
diff --git a/src/backend/storage/lmgr/deadlock.c b/src/backend/storage/lmgr/deadlock.c
index f8c5df08e69..ca2abea07f1 100644
--- a/src/backend/storage/lmgr/deadlock.c
+++ b/src/backend/storage/lmgr/deadlock.c
@@ -216,9 +216,6 @@ InitDeadLockChecking(void)
 DeadLockState
 DeadLockCheck(PGPROC *proc)
 {
-	int			i,
-				j;
-
 	/* Initialize to "no constraints" */
 	nCurConstraints = 0;
 	nPossibleConstraints = 0;
@@ -246,7 +243,7 @@ DeadLockCheck(PGPROC *proc)
 	}
 
 	/* Apply any needed rearrangements of wait queues */
-	for (i = 0; i < nWaitOrders; i++)
+	for (int i = 0; i < nWaitOrders; i++)
 	{
 		LOCK	   *lock = waitOrders[i].lock;
 		PGPROC	  **procs = waitOrders[i].procs;
@@ -261,9 +258,9 @@ DeadLockCheck(PGPROC *proc)
 
 		/* Reset the queue and re-add procs in the desired order */
 		ProcQueueInit(waitQueue);
-		for (j = 0; j < nProcs; j++)
+		for (int j = 0; j < nProcs; j++)
 		{
-			SHMQueueInsertBefore(&(waitQueue->links), &(procs[j]->links));
+			dlist_push_tail(&waitQueue->links, &(procs[j]->links));
 			waitQueue->size++;
 		}
 
@@ -545,8 +542,7 @@ FindLockCycleRecurseMember(PGPROC *checkProc,
 	PGPROC	   *proc;
 	LOCK	   *lock = checkProc->waitLock;
 	PGXACT	   *pgxact;
-	PROCLOCK   *proclock;
-	SHM_QUEUE  *procLocks;
+	dlist_iter	proclock_iter;
 	LockMethod	lockMethodTable;
 	PROC_QUEUE *waitQueue;
 	int			queue_size;
@@ -563,13 +559,9 @@ FindLockCycleRecurseMember(PGPROC *checkProc,
 	 * Scan for procs that already hold conflicting locks.  These are "hard"
 	 * edges in the waits-for graph.
 	 */
-	procLocks = &(lock->procLocks);
-
-	proclock = (PROCLOCK *) SHMQueueNext(procLocks, procLocks,
-										 offsetof(PROCLOCK, lockLink));
-
-	while (proclock)
+	dlist_foreach(proclock_iter, &lock->procLocks)
 	{
+		PROCLOCK   *proclock = dlist_container(PROCLOCK, lockLink, proclock_iter.cur);
 		PGPROC	   *leader;
 
 		proc = proclock->tag.myProc;
@@ -629,9 +621,6 @@ FindLockCycleRecurseMember(PGPROC *checkProc,
 				}
 			}
 		}
-
-		proclock = (PROCLOCK *) SHMQueueNext(procLocks, &proclock->lockLink,
-											 offsetof(PROCLOCK, lockLink));
 	}
 
 	/*
@@ -704,6 +693,7 @@ FindLockCycleRecurseMember(PGPROC *checkProc,
 	else
 	{
 		PGPROC	   *lastGroupMember = NULL;
+		dlist_iter	proc_iter;
 
 		/* Use the true lock wait queue order */
 		waitQueue = &(lock->waitProcs);
@@ -719,13 +709,14 @@ FindLockCycleRecurseMember(PGPROC *checkProc,
 			lastGroupMember = checkProc;
 		else
 		{
-			proc = (PGPROC *) waitQueue->links.next;
-			queue_size = waitQueue->size;
-			while (queue_size-- > 0)
+			dlist_iter iter;
+
+			dlist_foreach(iter, &waitQueue->links)
 			{
+				proc = dlist_container(PGPROC, links, iter.cur);
+
 				if (proc->lockGroupLeader == checkProcLeader)
 					lastGroupMember = proc;
-				proc = (PGPROC *) proc->links.next;
 			}
 			Assert(lastGroupMember != NULL);
 		}
@@ -733,12 +724,12 @@ FindLockCycleRecurseMember(PGPROC *checkProc,
 		/*
 		 * OK, now rescan (or scan) the queue to identify the soft conflicts.
 		 */
-		queue_size = waitQueue->size;
-		proc = (PGPROC *) waitQueue->links.next;
-		while (queue_size-- > 0)
+		dlist_foreach(proc_iter, &waitQueue->links)
 		{
 			PGPROC	   *leader;
 
+			proc = dlist_container(PGPROC, links, proc_iter.cur);
+
 			leader = proc->lockGroupLeader == NULL ? proc :
 				proc->lockGroupLeader;
 
@@ -772,8 +763,6 @@ FindLockCycleRecurseMember(PGPROC *checkProc,
 					return true;
 				}
 			}
-
-			proc = (PGPROC *) proc->links.next;
 		}
 	}
 
@@ -882,14 +871,16 @@ TopoSort(LOCK *lock,
 				k,
 				kk,
 				last;
+	dlist_iter	proc_iter;
 
 	/* First, fill topoProcs[] array with the procs in their current order */
-	proc = (PGPROC *) waitQueue->links.next;
-	for (i = 0; i < queue_size; i++)
+	i = 0;
+	dlist_foreach(proc_iter, &waitQueue->links)
 	{
-		topoProcs[i] = proc;
-		proc = (PGPROC *) proc->links.next;
+		proc = dlist_container(PGPROC, links, proc_iter.cur);
+		topoProcs[i++] = proc;
 	}
+	Assert(i == queue_size);
 
 	/*
 	 * Scan the constraints, and for each proc in the array, generate a count
diff --git a/src/backend/storage/lmgr/lock.c b/src/backend/storage/lmgr/lock.c
index 56dba09299d..01ac3c06c5e 100644
--- a/src/backend/storage/lmgr/lock.c
+++ b/src/backend/storage/lmgr/lock.c
@@ -1005,8 +1005,8 @@ LockAcquireExtended(const LOCKTAG *locktag,
 				uint32		proclock_hashcode;
 
 				proclock_hashcode = ProcLockHashCode(&proclock->tag, hashcode);
-				SHMQueueDelete(&proclock->lockLink);
-				SHMQueueDelete(&proclock->procLink);
+				dlist_delete(&proclock->lockLink);
+				dlist_delete(&proclock->procLink);
 				if (!hash_search_with_hash_value(LockMethodProcLockHash,
 												 (void *) &(proclock->tag),
 												 proclock_hashcode,
@@ -1141,7 +1141,7 @@ SetupLockInTable(LockMethod lockMethodTable, PGPROC *proc,
 	{
 		lock->grantMask = 0;
 		lock->waitMask = 0;
-		SHMQueueInit(&(lock->procLocks));
+		dlist_init(&lock->procLocks);
 		ProcQueueInit(&(lock->waitProcs));
 		lock->nRequested = 0;
 		lock->nGranted = 0;
@@ -1184,7 +1184,7 @@ SetupLockInTable(LockMethod lockMethodTable, PGPROC *proc,
 			 * of shared memory, because there won't be anything to cause
 			 * anyone to release the lock object later.
 			 */
-			Assert(SHMQueueEmpty(&(lock->procLocks)));
+			Assert(dlist_is_empty(&(lock->procLocks)));
 			if (!hash_search_with_hash_value(LockMethodLockHash,
 											 (void *) &(lock->tag),
 											 hashcode,
@@ -1217,9 +1217,8 @@ SetupLockInTable(LockMethod lockMethodTable, PGPROC *proc,
 		proclock->holdMask = 0;
 		proclock->releaseMask = 0;
 		/* Add proclock to appropriate lists */
-		SHMQueueInsertBefore(&lock->procLocks, &proclock->lockLink);
-		SHMQueueInsertBefore(&(proc->myProcLocks[partition]),
-							 &proclock->procLink);
+		dlist_push_tail(&lock->procLocks, &proclock->lockLink);
+		dlist_push_tail(&proc->myProcLocks[partition], &proclock->procLink);
 		PROCLOCK_PRINT("LockAcquire: new", proclock);
 	}
 	else
@@ -1349,9 +1348,8 @@ LockCheckConflicts(LockMethod lockMethodTable,
 	int			conflictMask = lockMethodTable->conflictTab[lockmode];
 	int			conflictsRemaining[MAX_LOCKMODES];
 	int			totalConflictsRemaining = 0;
+	dlist_iter	proclock_iter;
 	int			i;
-	SHM_QUEUE  *procLocks;
-	PROCLOCK   *otherproclock;
 
 	/*
 	 * first check for global conflicts: If no locks conflict with my request,
@@ -1411,11 +1409,11 @@ LockCheckConflicts(LockMethod lockMethodTable,
 	 * shared memory state more complex (and larger) but it doesn't seem worth
 	 * it.
 	 */
-	procLocks = &(lock->procLocks);
-	otherproclock = (PROCLOCK *)
-		SHMQueueNext(procLocks, procLocks, offsetof(PROCLOCK, lockLink));
-	while (otherproclock != NULL)
+	dlist_foreach(proclock_iter, &lock->procLocks)
 	{
+		PROCLOCK   *otherproclock =
+			dlist_container(PROCLOCK, lockLink, proclock_iter.cur);
+
 		if (proclock != otherproclock &&
 			proclock->groupLeader == otherproclock->groupLeader &&
 			(otherproclock->holdMask & conflictMask) != 0)
@@ -1440,9 +1438,6 @@ LockCheckConflicts(LockMethod lockMethodTable,
 				return false;
 			}
 		}
-		otherproclock = (PROCLOCK *)
-			SHMQueueNext(procLocks, &otherproclock->lockLink,
-						 offsetof(PROCLOCK, lockLink));
 	}
 
 	/* Nope, it's a real conflict. */
@@ -1555,8 +1550,8 @@ CleanUpLock(LOCK *lock, PROCLOCK *proclock,
 		uint32		proclock_hashcode;
 
 		PROCLOCK_PRINT("CleanUpLock: deleting", proclock);
-		SHMQueueDelete(&proclock->lockLink);
-		SHMQueueDelete(&proclock->procLink);
+		dlist_delete(&proclock->lockLink);
+		dlist_delete(&proclock->procLink);
 		proclock_hashcode = ProcLockHashCode(&proclock->tag, hashcode);
 		if (!hash_search_with_hash_value(LockMethodProcLockHash,
 										 (void *) &(proclock->tag),
@@ -1573,7 +1568,7 @@ CleanUpLock(LOCK *lock, PROCLOCK *proclock,
 		 * object.
 		 */
 		LOCK_PRINT("CleanUpLock: deleting", lock, 0);
-		Assert(SHMQueueEmpty(&(lock->procLocks)));
+		Assert(dlist_is_empty(&lock->procLocks));
 		if (!hash_search_with_hash_value(LockMethodLockHash,
 										 (void *) &(lock->tag),
 										 hashcode,
@@ -1837,7 +1832,7 @@ RemoveFromWaitQueue(PGPROC *proc, uint32 hashcode)
 	Assert(0 < lockmethodid && lockmethodid < lengthof(LockMethods));
 
 	/* Remove proc from lock's wait queue */
-	SHMQueueDelete(&(proc->links));
+	dlist_delete(&proc->links);
 	waitLock->waitProcs.size--;
 
 	/* Undo increments of request counts by waiting process */
@@ -2092,7 +2087,6 @@ LockReleaseAll(LOCKMETHODID lockmethodid, bool allLocks)
 				numLockModes;
 	LOCALLOCK  *locallock;
 	LOCK	   *lock;
-	PROCLOCK   *proclock;
 	int			partition;
 	bool		have_fast_path_lwlock = false;
 
@@ -2249,8 +2243,8 @@ LockReleaseAll(LOCKMETHODID lockmethodid, bool allLocks)
 	for (partition = 0; partition < NUM_LOCK_PARTITIONS; partition++)
 	{
 		LWLock	   *partitionLock;
-		SHM_QUEUE  *procLocks = &(MyProc->myProcLocks[partition]);
-		PROCLOCK   *nextplock;
+		dlist_head *procLocks = &MyProc->myProcLocks[partition];
+		dlist_mutable_iter proclock_iter;
 
 		partitionLock = LockHashPartitionLockByIndex(partition);
 
@@ -2273,24 +2267,16 @@ LockReleaseAll(LOCKMETHODID lockmethodid, bool allLocks)
 		 * locallock situation, we lose that guarantee for fast-path locks.
 		 * This is not ideal.
 		 */
-		if (SHMQueueNext(procLocks, procLocks,
-						 offsetof(PROCLOCK, procLink)) == NULL)
+		if (dlist_is_empty(procLocks))
 			continue;			/* needn't examine this partition */
 
 		LWLockAcquire(partitionLock, LW_EXCLUSIVE);
 
-		for (proclock = (PROCLOCK *) SHMQueueNext(procLocks, procLocks,
-												  offsetof(PROCLOCK, procLink));
-			 proclock;
-			 proclock = nextplock)
+		dlist_foreach_modify(proclock_iter, procLocks)
 		{
+			PROCLOCK   *proclock = dlist_container(PROCLOCK, procLink, proclock_iter.cur);
 			bool		wakeupNeeded = false;
 
-			/* Get link first, since we may unlink/delete this proclock */
-			nextplock = (PROCLOCK *)
-				SHMQueueNext(procLocks, &proclock->procLink,
-							 offsetof(PROCLOCK, procLink));
-
 			Assert(proclock->tag.myProc == MyProc);
 
 			lock = proclock->tag.myLock;
@@ -2823,7 +2809,7 @@ GetLockConflicts(const LOCKTAG *locktag, LOCKMODE lockmode, int *countp)
 	LockMethod	lockMethodTable;
 	LOCK	   *lock;
 	LOCKMASK	conflictMask;
-	SHM_QUEUE  *procLocks;
+	dlist_iter	proclock_iter;
 	PROCLOCK   *proclock;
 	uint32		hashcode;
 	LWLock	   *partitionLock;
@@ -2971,14 +2957,10 @@ GetLockConflicts(const LOCKTAG *locktag, LOCKMODE lockmode, int *countp)
 	/*
 	 * Examine each existing holder (or awaiter) of the lock.
 	 */
-
-	procLocks = &(lock->procLocks);
-
-	proclock = (PROCLOCK *) SHMQueueNext(procLocks, procLocks,
-										 offsetof(PROCLOCK, lockLink));
-
-	while (proclock)
+	dlist_foreach(proclock_iter, &lock->procLocks)
 	{
+		proclock = dlist_container(PROCLOCK, lockLink, proclock_iter.cur);
+
 		if (conflictMask & proclock->holdMask)
 		{
 			PGPROC	   *proc = proclock->tag.myProc;
@@ -3008,9 +2990,6 @@ GetLockConflicts(const LOCKTAG *locktag, LOCKMODE lockmode, int *countp)
 				}
 			}
 		}
-
-		proclock = (PROCLOCK *) SHMQueueNext(procLocks, &proclock->lockLink,
-											 offsetof(PROCLOCK, lockLink));
 	}
 
 	LWLockRelease(partitionLock);
@@ -3328,8 +3307,8 @@ PostPrepare_Locks(TransactionId xid)
 	for (partition = 0; partition < NUM_LOCK_PARTITIONS; partition++)
 	{
 		LWLock	   *partitionLock;
-		SHM_QUEUE  *procLocks = &(MyProc->myProcLocks[partition]);
-		PROCLOCK   *nextplock;
+		dlist_head *procLocks = &(MyProc->myProcLocks[partition]);
+		dlist_mutable_iter proclock_iter;
 
 		partitionLock = LockHashPartitionLockByIndex(partition);
 
@@ -3341,21 +3320,14 @@ PostPrepare_Locks(TransactionId xid)
 		 * another backend is adding something to our lists now.  For safety,
 		 * though, we code this the same way as in LockReleaseAll.
 		 */
-		if (SHMQueueNext(procLocks, procLocks,
-						 offsetof(PROCLOCK, procLink)) == NULL)
+		if (dlist_is_empty(procLocks))
 			continue;			/* needn't examine this partition */
 
 		LWLockAcquire(partitionLock, LW_EXCLUSIVE);
 
-		for (proclock = (PROCLOCK *) SHMQueueNext(procLocks, procLocks,
-												  offsetof(PROCLOCK, procLink));
-			 proclock;
-			 proclock = nextplock)
+		dlist_foreach_modify(proclock_iter, procLocks)
 		{
-			/* Get link first, since we may unlink/relink this proclock */
-			nextplock = (PROCLOCK *)
-				SHMQueueNext(procLocks, &proclock->procLink,
-							 offsetof(PROCLOCK, procLink));
+			proclock = dlist_container(PROCLOCK, procLink, proclock_iter.cur);
 
 			Assert(proclock->tag.myProc == MyProc);
 
@@ -3393,7 +3365,7 @@ PostPrepare_Locks(TransactionId xid)
 			 * same hash partition, cf proclock_hash().  So the partition lock
 			 * we already hold is sufficient for this.
 			 */
-			SHMQueueDelete(&proclock->procLink);
+			dlist_delete(&proclock->procLink);
 
 			/*
 			 * Create the new hash key for the proclock.
@@ -3419,8 +3391,7 @@ PostPrepare_Locks(TransactionId xid)
 				elog(PANIC, "duplicate entry found while reassigning a prepared transaction's locks");
 
 			/* Re-link into the new proc's proclock list */
-			SHMQueueInsertBefore(&(newproc->myProcLocks[partition]),
-								 &proclock->procLink);
+			dlist_push_tail(&newproc->myProcLocks[partition], &proclock->procLink);
 
 			PROCLOCK_PRINT("PostPrepare_Locks: updated", proclock);
 		}						/* loop over PROCLOCKs within this partition */
@@ -3741,12 +3712,10 @@ GetSingleProcBlockerStatusData(PGPROC *blocked_proc, BlockedProcsData *data)
 {
 	LOCK	   *theLock = blocked_proc->waitLock;
 	BlockedProcData *bproc;
-	SHM_QUEUE  *procLocks;
-	PROCLOCK   *proclock;
+	dlist_iter	proclock_iter;
+	dlist_iter	proc_iter;
 	PROC_QUEUE *waitQueue;
-	PGPROC	   *proc;
 	int			queue_size;
-	int			i;
 
 	/* Nothing to do if this proc is not blocked */
 	if (theLock == NULL)
@@ -3764,11 +3733,10 @@ GetSingleProcBlockerStatusData(PGPROC *blocked_proc, BlockedProcsData *data)
 	 */
 
 	/* Collect all PROCLOCKs associated with theLock */
-	procLocks = &(theLock->procLocks);
-	proclock = (PROCLOCK *) SHMQueueNext(procLocks, procLocks,
-										 offsetof(PROCLOCK, lockLink));
-	while (proclock)
+	dlist_foreach(proclock_iter, &theLock->procLocks)
 	{
+		PROCLOCK   *proclock =
+			dlist_container(PROCLOCK, lockLink, proclock_iter.cur);
 		PGPROC	   *proc = proclock->tag.myProc;
 		LOCK	   *lock = proclock->tag.myLock;
 		LockInstanceData *instance;
@@ -3793,9 +3761,6 @@ GetSingleProcBlockerStatusData(PGPROC *blocked_proc, BlockedProcsData *data)
 		instance->leaderPid = proclock->groupLeader->pid;
 		instance->fastpath = false;
 		data->nlocks++;
-
-		proclock = (PROCLOCK *) SHMQueueNext(procLocks, &proclock->lockLink,
-											 offsetof(PROCLOCK, lockLink));
 	}
 
 	/* Enlarge waiter_pids[] if it's too small to hold all wait queue PIDs */
@@ -3811,9 +3776,9 @@ GetSingleProcBlockerStatusData(PGPROC *blocked_proc, BlockedProcsData *data)
 	}
 
 	/* Collect PIDs from the lock's wait queue, stopping at blocked_proc */
-	proc = (PGPROC *) waitQueue->links.next;
-	for (i = 0; i < queue_size; i++)
+	dlist_foreach(proc_iter, &waitQueue->links)
 	{
+		PGPROC	   *proc = dlist_container(PGPROC, links, proc_iter.cur);
 		if (proc == blocked_proc)
 			break;
 		data->waiter_pids[data->npids++] = proc->pid;
@@ -4090,7 +4055,7 @@ lock_twophase_recover(TransactionId xid, uint16 info,
 	{
 		lock->grantMask = 0;
 		lock->waitMask = 0;
-		SHMQueueInit(&(lock->procLocks));
+		dlist_init(&lock->procLocks);
 		ProcQueueInit(&(lock->waitProcs));
 		lock->nRequested = 0;
 		lock->nGranted = 0;
@@ -4133,7 +4098,7 @@ lock_twophase_recover(TransactionId xid, uint16 info,
 			 * of shared memory, because there won't be anything to cause
 			 * anyone to release the lock object later.
 			 */
-			Assert(SHMQueueEmpty(&(lock->procLocks)));
+			Assert(dlist_is_empty(&lock->procLocks));
 			if (!hash_search_with_hash_value(LockMethodLockHash,
 											 (void *) &(lock->tag),
 											 hashcode,
@@ -4158,9 +4123,9 @@ lock_twophase_recover(TransactionId xid, uint16 info,
 		proclock->holdMask = 0;
 		proclock->releaseMask = 0;
 		/* Add proclock to appropriate lists */
-		SHMQueueInsertBefore(&lock->procLocks, &proclock->lockLink);
-		SHMQueueInsertBefore(&(proc->myProcLocks[partition]),
-							 &proclock->procLink);
+		dlist_push_tail(&lock->procLocks, &proclock->lockLink);
+		dlist_push_tail(&proc->myProcLocks[partition],
+						&proclock->procLink);
 		PROCLOCK_PRINT("lock_twophase_recover: new", proclock);
 	}
 	else
diff --git a/src/backend/storage/lmgr/proc.c b/src/backend/storage/lmgr/proc.c
index eb321f72ea4..5a157fc07d8 100644
--- a/src/backend/storage/lmgr/proc.c
+++ b/src/backend/storage/lmgr/proc.c
@@ -178,10 +178,10 @@ InitProcGlobal(void)
 	 * Initialize the data structures.
 	 */
 	ProcGlobal->spins_per_delay = DEFAULT_SPINS_PER_DELAY;
-	ProcGlobal->freeProcs = NULL;
-	ProcGlobal->autovacFreeProcs = NULL;
-	ProcGlobal->bgworkerFreeProcs = NULL;
-	ProcGlobal->walsenderFreeProcs = NULL;
+	dlist_init(&ProcGlobal->freeProcs);
+	dlist_init(&ProcGlobal->autovacFreeProcs);
+	dlist_init(&ProcGlobal->bgworkerFreeProcs);
+	dlist_init(&ProcGlobal->walsenderFreeProcs);
 	ProcGlobal->startupProc = NULL;
 	ProcGlobal->startupProcPid = 0;
 	ProcGlobal->startupBufferPinWaitBufId = -1;
@@ -218,6 +218,8 @@ InitProcGlobal(void)
 
 	for (i = 0; i < TotalProcs; i++)
 	{
+		PGPROC	   *proc = &procs[i];
+
 		/* Common initialization for all PGPROCs, regardless of type. */
 
 		/*
@@ -227,11 +229,11 @@ InitProcGlobal(void)
 		 */
 		if (i < MaxBackends + NUM_AUXILIARY_PROCS)
 		{
-			procs[i].sem = PGSemaphoreCreate();
-			InitSharedLatch(&(procs[i].procLatch));
-			LWLockInitialize(&(procs[i].backendLock), LWTRANCHE_PROC);
+			proc->sem = PGSemaphoreCreate();
+			InitSharedLatch(&(proc->procLatch));
+			LWLockInitialize(&(proc->backendLock), LWTRANCHE_PROC);
 		}
-		procs[i].pgprocno = i;
+		proc->pgprocno = i;
 
 		/*
 		 * Newly created PGPROCs for normal backends, autovacuum and bgworkers
@@ -244,45 +246,41 @@ InitProcGlobal(void)
 		if (i < MaxConnections)
 		{
 			/* PGPROC for normal backend, add to freeProcs list */
-			procs[i].links.next = (SHM_QUEUE *) ProcGlobal->freeProcs;
-			ProcGlobal->freeProcs = &procs[i];
-			procs[i].procgloballist = &ProcGlobal->freeProcs;
+			dlist_push_head(&ProcGlobal->freeProcs, &proc->links);
+			proc->procgloballist = &ProcGlobal->freeProcs;
 		}
 		else if (i < MaxConnections + autovacuum_max_workers + 1)
 		{
 			/* PGPROC for AV launcher/worker, add to autovacFreeProcs list */
-			procs[i].links.next = (SHM_QUEUE *) ProcGlobal->autovacFreeProcs;
-			ProcGlobal->autovacFreeProcs = &procs[i];
-			procs[i].procgloballist = &ProcGlobal->autovacFreeProcs;
+			dlist_push_head(&ProcGlobal->autovacFreeProcs, &proc->links);
+			proc->procgloballist = &ProcGlobal->autovacFreeProcs;
 		}
 		else if (i < MaxConnections + autovacuum_max_workers + 1 + max_worker_processes)
 		{
 			/* PGPROC for bgworker, add to bgworkerFreeProcs list */
-			procs[i].links.next = (SHM_QUEUE *) ProcGlobal->bgworkerFreeProcs;
-			ProcGlobal->bgworkerFreeProcs = &procs[i];
-			procs[i].procgloballist = &ProcGlobal->bgworkerFreeProcs;
+			dlist_push_head(&ProcGlobal->bgworkerFreeProcs, &proc->links);
+			proc->procgloballist = &ProcGlobal->bgworkerFreeProcs;
 		}
 		else if (i < MaxBackends)
 		{
 			/* PGPROC for walsender, add to walsenderFreeProcs list */
-			procs[i].links.next = (SHM_QUEUE *) ProcGlobal->walsenderFreeProcs;
-			ProcGlobal->walsenderFreeProcs = &procs[i];
-			procs[i].procgloballist = &ProcGlobal->walsenderFreeProcs;
+			dlist_push_head(&ProcGlobal->walsenderFreeProcs, &proc->links);
+			proc->procgloballist = &ProcGlobal->walsenderFreeProcs;
 		}
 
 		/* Initialize myProcLocks[] shared memory queues. */
 		for (j = 0; j < NUM_LOCK_PARTITIONS; j++)
-			SHMQueueInit(&(procs[i].myProcLocks[j]));
+			dlist_init(&(proc->myProcLocks[j]));
 
 		/* Initialize lockGroupMembers list. */
-		dlist_init(&procs[i].lockGroupMembers);
+		dlist_init(&proc->lockGroupMembers);
 
 		/*
 		 * Initialize the atomic variables, otherwise, it won't be safe to
 		 * access them for backends that aren't currently in use.
 		 */
-		pg_atomic_init_u32(&(procs[i].procArrayGroupNext), INVALID_PGPROCNO);
-		pg_atomic_init_u32(&(procs[i].clogGroupNext), INVALID_PGPROCNO);
+		pg_atomic_init_u32(&(proc->procArrayGroupNext), INVALID_PGPROCNO);
+		pg_atomic_init_u32(&(proc->clogGroupNext), INVALID_PGPROCNO);
 	}
 
 	/*
@@ -303,7 +301,7 @@ InitProcGlobal(void)
 void
 InitProcess(void)
 {
-	PGPROC	   *volatile *procgloballist;
+	dlist_head *procgloballist;
 
 	/*
 	 * ProcGlobal should be set up already (if we are a backend, we inherit
@@ -336,11 +334,9 @@ InitProcess(void)
 
 	set_spins_per_delay(ProcGlobal->spins_per_delay);
 
-	MyProc = *procgloballist;
-
-	if (MyProc != NULL)
+	if (!dlist_is_empty(procgloballist))
 	{
-		*procgloballist = (PGPROC *) MyProc->links.next;
+		MyProc = (PGPROC*) dlist_pop_head_node(procgloballist);
 		SpinLockRelease(ProcStructLock);
 	}
 	else
@@ -382,7 +378,7 @@ InitProcess(void)
 	 * Initialize all fields of MyProc, except for those previously
 	 * initialized by InitProcGlobal.
 	 */
-	SHMQueueElemInit(&(MyProc->links));
+	dlist_node_init(&MyProc->links);
 	MyProc->waitStatus = STATUS_OK;
 	MyProc->lxid = InvalidLocalTransactionId;
 	MyProc->fpVXIDLock = false;
@@ -411,7 +407,7 @@ InitProcess(void)
 
 		/* Last process should have released all locks. */
 		for (i = 0; i < NUM_LOCK_PARTITIONS; i++)
-			Assert(SHMQueueEmpty(&(MyProc->myProcLocks[i])));
+			Assert(dlist_is_empty(&(MyProc->myProcLocks[i])));
 	}
 #endif
 	MyProc->recoveryConflictPending = false;
@@ -566,7 +562,7 @@ InitAuxiliaryProcess(void)
 	 * Initialize all fields of MyProc, except for those previously
 	 * initialized by InitProcGlobal.
 	 */
-	SHMQueueElemInit(&(MyProc->links));
+	dlist_node_init(&MyProc->links);
 	MyProc->waitStatus = STATUS_OK;
 	MyProc->lxid = InvalidLocalTransactionId;
 	MyProc->fpVXIDLock = false;
@@ -590,7 +586,7 @@ InitAuxiliaryProcess(void)
 
 		/* Last process should have released all locks. */
 		for (i = 0; i < NUM_LOCK_PARTITIONS; i++)
-			Assert(SHMQueueEmpty(&(MyProc->myProcLocks[i])));
+			Assert(dlist_is_empty(&(MyProc->myProcLocks[i])));
 	}
 #endif
 
@@ -670,16 +666,15 @@ GetStartupBufferPinWaitBufId(void)
 bool
 HaveNFreeProcs(int n)
 {
-	PGPROC	   *proc;
+	dlist_iter	iter;
 
 	SpinLockAcquire(ProcStructLock);
 
-	proc = ProcGlobal->freeProcs;
-
-	while (n > 0 && proc != NULL)
+	dlist_foreach(iter, &ProcGlobal->freeProcs)
 	{
-		proc = (PGPROC *) proc->links.next;
 		n--;
+		if (n == 0)
+			break;
 	}
 
 	SpinLockRelease(ProcStructLock);
@@ -742,7 +737,7 @@ LockErrorCleanup(void)
 	partitionLock = LockHashPartitionLock(lockAwaited->hashcode);
 	LWLockAcquire(partitionLock, LW_EXCLUSIVE);
 
-	if (MyProc->links.next != NULL)
+	if (!dlist_node_is_detached(&MyProc->links))
 	{
 		/* We could not have been granted the lock yet */
 		RemoveFromWaitQueue(MyProc, lockAwaited->hashcode);
@@ -815,7 +810,7 @@ static void
 ProcKill(int code, Datum arg)
 {
 	PGPROC	   *proc;
-	PGPROC	   *volatile *procgloballist;
+	dlist_head *procgloballist;
 
 	Assert(MyProc != NULL);
 
@@ -828,7 +823,7 @@ ProcKill(int code, Datum arg)
 
 		/* Last process should have released all locks. */
 		for (i = 0; i < NUM_LOCK_PARTITIONS; i++)
-			Assert(SHMQueueEmpty(&(MyProc->myProcLocks[i])));
+			Assert(dlist_is_empty(&(MyProc->myProcLocks[i])));
 	}
 #endif
 
@@ -851,7 +846,7 @@ ProcKill(int code, Datum arg)
 
 	/*
 	 * Detach from any lock group of which we are a member.  If the leader
-	 * exist before all other group members, its PGPROC will remain allocated
+	 * exits before all other group members, its PGPROC will remain allocated
 	 * until the last group process exits; that process must return the
 	 * leader's PGPROC to the appropriate list.
 	 */
@@ -872,8 +867,7 @@ ProcKill(int code, Datum arg)
 
 				/* Leader exited first; return its PGPROC. */
 				SpinLockAcquire(ProcStructLock);
-				leader->links.next = (SHM_QUEUE *) *procgloballist;
-				*procgloballist = leader;
+				dlist_push_head(procgloballist, &leader->links);
 				SpinLockRelease(ProcStructLock);
 			}
 		}
@@ -907,8 +901,7 @@ ProcKill(int code, Datum arg)
 		Assert(dlist_is_empty(&proc->lockGroupMembers));
 
 		/* Return PGPROC structure (and semaphore) to appropriate freelist */
-		proc->links.next = (SHM_QUEUE *) *procgloballist;
-		*procgloballist = proc;
+		dlist_push_tail(procgloballist, &proc->links);
 	}
 
 	/* Update shared estimate of spins_per_delay */
@@ -1037,7 +1030,7 @@ ProcQueueAlloc(const char *name)
 void
 ProcQueueInit(PROC_QUEUE *queue)
 {
-	SHMQueueInit(&(queue->links));
+	dlist_init(&queue->links);
 	queue->size = 0;
 }
 
@@ -1068,12 +1061,11 @@ ProcSleep(LOCALLOCK *locallock, LockMethod lockMethodTable)
 	LWLock	   *partitionLock = LockHashPartitionLock(hashcode);
 	PROC_QUEUE *waitQueue = &(lock->waitProcs);
 	LOCKMASK	myHeldLocks = MyProc->heldLocks;
+	PGPROC	   *insert_before = NULL;
 	bool		early_deadlock = false;
 	bool		allow_autovacuum_cancel = true;
 	int			myWaitStatus;
-	PGPROC	   *proc;
 	PGPROC	   *leader = MyProc->lockGroupLeader;
-	int			i;
 
 	/*
 	 * If group locking is in use, locks held by members of my locking group
@@ -1081,18 +1073,16 @@ ProcSleep(LOCALLOCK *locallock, LockMethod lockMethodTable)
 	 */
 	if (leader != NULL)
 	{
-		SHM_QUEUE  *procLocks = &(lock->procLocks);
-		PROCLOCK   *otherproclock;
+		dlist_iter	iter;
 
-		otherproclock = (PROCLOCK *)
-			SHMQueueNext(procLocks, procLocks, offsetof(PROCLOCK, lockLink));
-		while (otherproclock != NULL)
+		dlist_foreach(iter, &lock->procLocks)
 		{
+			PROCLOCK   *otherproclock;
+
+			otherproclock = dlist_container(PROCLOCK, lockLink, iter.cur);
+
 			if (otherproclock->groupLeader == leader)
 				myHeldLocks |= otherproclock->holdMask;
-			otherproclock = (PROCLOCK *)
-				SHMQueueNext(procLocks, &otherproclock->lockLink,
-							 offsetof(PROCLOCK, lockLink));
 		}
 	}
 
@@ -1116,20 +1106,23 @@ ProcSleep(LOCALLOCK *locallock, LockMethod lockMethodTable)
 	if (myHeldLocks != 0)
 	{
 		LOCKMASK	aheadRequests = 0;
+		dlist_iter	iter;
 
-		proc = (PGPROC *) waitQueue->links.next;
-		for (i = 0; i < waitQueue->size; i++)
+		// FIXME: Shouldn't we just use the correct offset math?
+		StaticAssertStmt(offsetof(PGPROC, links) == 0, "odd casting");
+
+		dlist_foreach(iter, &waitQueue->links)
 		{
+			PGPROC	   *proc = dlist_container(PGPROC, links, iter.cur);
+
 			/*
 			 * If we're part of the same locking group as this waiter, its
 			 * locks neither conflict with ours nor contribute to
 			 * aheadRequests.
 			 */
 			if (leader != NULL && leader == proc->lockGroupLeader)
-			{
-				proc = (PGPROC *) proc->links.next;
 				continue;
-			}
+
 			/* Must he wait for me? */
 			if (lockMethodTable->conflictTab[proc->waitLockMode] & myHeldLocks)
 			{
@@ -1157,29 +1150,25 @@ ProcSleep(LOCALLOCK *locallock, LockMethod lockMethodTable)
 					GrantAwaitedLock();
 					return STATUS_OK;
 				}
-				/* Break out of loop to put myself before him */
+
+				/* Put myself into wait queue before conflicting process */
+				insert_before = proc;
 				break;
 			}
 			/* Nope, so advance to next waiter */
 			aheadRequests |= LOCKBIT_ON(proc->waitLockMode);
-			proc = (PGPROC *) proc->links.next;
 		}
-
-		/*
-		 * If we fall out of loop normally, proc points to waitQueue head, so
-		 * we will insert at tail of queue as desired.
-		 */
-	}
-	else
-	{
-		/* I hold no locks, so I can't push in front of anyone. */
-		proc = (PGPROC *) &(waitQueue->links);
 	}
 
 	/*
-	 * Insert self into queue, ahead of the given proc (or at tail of queue).
+	 * Insert self into queue, ahead of the determined proc or at the tail.
 	 */
-	SHMQueueInsertBefore(&(proc->links), &(MyProc->links));
+	if (insert_before)
+		dlist_insert_before(&insert_before->links, &MyProc->links);
+	else
+		dlist_push_tail(&waitQueue->links, &MyProc->links);
+
+
 	waitQueue->size++;
 
 	lock->waitMask |= LOCKBIT_ON(lockmode);
@@ -1384,7 +1373,7 @@ ProcSleep(LOCALLOCK *locallock, LockMethod lockMethodTable)
 			long		secs;
 			int			usecs;
 			long		msecs;
-			SHM_QUEUE  *procLocks;
+			dlist_iter	proc_iter;
 			PROCLOCK   *proclock;
 			bool		first_holder = true,
 						first_waiter = true;
@@ -1414,12 +1403,11 @@ ProcSleep(LOCALLOCK *locallock, LockMethod lockMethodTable)
 
 			LWLockAcquire(partitionLock, LW_SHARED);
 
-			procLocks = &(lock->procLocks);
-			proclock = (PROCLOCK *) SHMQueueNext(procLocks, procLocks,
-												 offsetof(PROCLOCK, lockLink));
-
-			while (proclock)
+			dlist_foreach(proc_iter, &lock->procLocks)
 			{
+				proclock =
+					dlist_container(PROCLOCK, lockLink, proc_iter.cur);
+
 				/*
 				 * we are a waiter if myProc->waitProcLock == proclock; we are
 				 * a holder if it is NULL or something different
@@ -1450,9 +1438,6 @@ ProcSleep(LOCALLOCK *locallock, LockMethod lockMethodTable)
 
 					lockHoldersNum++;
 				}
-
-				proclock = (PROCLOCK *) SHMQueueNext(procLocks, &proclock->lockLink,
-													 offsetof(PROCLOCK, lockLink));
 			}
 
 			LWLockRelease(partitionLock);
@@ -1577,7 +1562,6 @@ ProcSleep(LOCALLOCK *locallock, LockMethod lockMethodTable)
  * ProcWakeup -- wake up a process by setting its latch.
  *
  *	 Also remove the process from the wait queue and set its links invalid.
- *	 RETURN: the next process in the wait queue.
  *
  * The appropriate lock partition lock must be held by caller.
  *
@@ -1586,22 +1570,17 @@ ProcSleep(LOCALLOCK *locallock, LockMethod lockMethodTable)
  * to twiddle the lock's request counts too --- see RemoveFromWaitQueue.
  * Hence, in practice the waitStatus parameter must be STATUS_OK.
  */
-PGPROC *
+void
 ProcWakeup(PGPROC *proc, int waitStatus)
 {
-	PGPROC	   *retProc;
-
 	/* Proc should be sleeping ... */
-	if (proc->links.prev == NULL ||
-		proc->links.next == NULL)
-		return NULL;
+	if (dlist_node_is_detached(&proc->links))
+		return;
+
 	Assert(proc->waitStatus == STATUS_WAITING);
 
-	/* Save next process before we zap the list link */
-	retProc = (PGPROC *) proc->links.next;
-
 	/* Remove process from wait queue */
-	SHMQueueDelete(&(proc->links));
+	dlist_delete_thoroughly(&proc->links);
 	(proc->waitLock->waitProcs.size)--;
 
 	/* Clean up process' state and pass it the ok/fail signal */
@@ -1611,8 +1590,6 @@ ProcWakeup(PGPROC *proc, int waitStatus)
 
 	/* And awaken it */
 	SetLatch(&proc->procLatch);
-
-	return retProc;
 }
 
 /*
@@ -1626,19 +1603,17 @@ void
 ProcLockWakeup(LockMethod lockMethodTable, LOCK *lock)
 {
 	PROC_QUEUE *waitQueue = &(lock->waitProcs);
-	int			queue_size = waitQueue->size;
-	PGPROC	   *proc;
 	LOCKMASK	aheadRequests = 0;
+	dlist_mutable_iter miter;
 
-	Assert(queue_size >= 0);
+	Assert(waitQueue->size >= 0);
 
-	if (queue_size == 0)
+	if (waitQueue->size == 0)
 		return;
 
-	proc = (PGPROC *) waitQueue->links.next;
-
-	while (queue_size-- > 0)
+	dlist_foreach_modify(miter, &waitQueue->links)
 	{
+		PGPROC	   *proc = dlist_container(PGPROC, links, miter.cur);
 		LOCKMODE	lockmode = proc->waitLockMode;
 
 		/*
@@ -1651,7 +1626,7 @@ ProcLockWakeup(LockMethod lockMethodTable, LOCK *lock)
 		{
 			/* OK to waken */
 			GrantLock(lock, proc->waitProcLock, lockmode);
-			proc = ProcWakeup(proc, STATUS_OK);
+			ProcWakeup(proc, STATUS_OK);
 
 			/*
 			 * ProcWakeup removes proc from the lock's waiting process queue
@@ -1661,11 +1636,8 @@ ProcLockWakeup(LockMethod lockMethodTable, LOCK *lock)
 		}
 		else
 		{
-			/*
-			 * Cannot wake this guy. Remember his request for later checks.
-			 */
+			/* Lock conflicts: don't wake, but remember for later checks. */
 			aheadRequests |= LOCKBIT_ON(lockmode);
-			proc = (PGPROC *) proc->links.next;
 		}
 	}
 
-- 
2.25.0.114.g5b0ca878e0

