Trial group locking code for ProcSleep.group_locking

1 files changed, 61 insertions, 21 deletions

diff --git a/src/backend/storage/lmgr/proc.c b/src/backend/storage/lmgr/proc.c
index b671800bd4..1af985115a 100644
--- a/src/backend/storage/lmgr/proc.c
+++ b/src/backend/storage/lmgr/proc.c
@@ -1037,18 +1037,53 @@ ProcSleep(LOCALLOCK *locallock, LockMethod lockMethodTable)
 	bool		early_deadlock = false;
 	bool		allow_autovacuum_cancel = true;
 	int			myWaitStatus;
-	PGPROC	   *proc;
+	PGPROC	   *proc = NULL;
 	int			i;
+	PGPROC	   *groupLeader = LockGroupLeader;
+
+	/*
+	 * Ignore trivial lock groups.
+	 *
+	 * We read MyProc->lockGroupMembers here without a lock.  The read itself
+	 * is atomic; while the value could be changing under us, it can't change
+	 * from a value < 2 to a value >= 2 while any group locks are actually
+	 * present.  Similarly, when iterating over the wait queue, we needn't
+	 * worry that the lock group membership of a process will change under us:
+	 * that's not allowed while a process holds any locks.
+	 */
+	if (MyProc == groupLeader && MyProc->lockGroupMembers >= 2)
+		groupLeader = NULL;
 
 	/*
 	 * Determine where to add myself in the wait queue.
 	 *
-	 * Normally I should go at the end of the queue.  However, if I already
-	 * hold locks that conflict with the request of any previous waiter, put
-	 * myself in the queue just in front of the first such waiter. This is not
-	 * a necessary step, since deadlock detection would move me to before that
-	 * waiter anyway; but it's relatively cheap to detect such a conflict
-	 * immediately, and avoid delaying till deadlock timeout.
+	 * Normally I should go at the end of the queue.  However, if I'm a
+	 * member of a lock group, and some other member of the lock group is
+	 * already waiting for a lock, then add add myself just after the
+	 * existing waiters.  This is necessary for correctness; any code that
+	 * scans the wait queue is entitled to assume that lockers from the same
+	 * group are in consecutive positions in the queue.
+	 */
+	if (groupLeader != NULL)
+	{
+		PGPROC *cproc = (PGPROC *) waitQueue->links.next;
+
+		for (i = 0; i < waitQueue->size; i++)
+		{
+			if (cproc->lockGroupLeader == LockGroupLeader)
+				proc = cproc;
+			else if (proc != NULL)
+				break;
+			cproc = (PGPROC *) cproc->links.next;
+		}
+	}
+
+	/*
+	 * If I already hold locks that conflict with the request of any previous
+	 * waiter, put myself in the queue just in front of the first such waiter.
+	 * This is not a necessary step, since deadlock detection would move me
+	 * to before that waiter anyway; but it's relatively cheap to detect such
+	 * a conflict immediately, and avoid delaying till deadlock timeout.
 	 *
 	 * Special case: if I find I should go in front of some waiter, check to
 	 * see if I conflict with already-held locks or the requests before that
@@ -1059,16 +1094,24 @@ ProcSleep(LOCALLOCK *locallock, LockMethod lockMethodTable)
 	 */
 	if (myHeldLocks != 0)
 	{
+		PGPROC *cproc = (PGPROC *) waitQueue->links.next;
 		LOCKMASK	aheadRequests = 0;
 
-		proc = (PGPROC *) waitQueue->links.next;
 		for (i = 0; i < waitQueue->size; i++)
 		{
+			/*
+			 * If we reached our own lock group in the wait queue without
+			 * finding a conflict, we aren't going to find one at all prior
+			 * to the insertion point, so bail out.
+			 */
+			if (groupLeader != NULL && cproc->lockGroupLeader == groupLeader)
+				break;
+
 			/* Must he wait for me? */
-			if (lockMethodTable->conflictTab[proc->waitLockMode] & myHeldLocks)
+			if (lockMethodTable->conflictTab[cproc->waitLockMode] & myHeldLocks)
 			{
 				/* Must I wait for him ? */
-				if (lockMethodTable->conflictTab[lockmode] & proc->heldLocks)
+				if (lockMethodTable->conflictTab[lockmode] & cproc->heldLocks)
 				{
 					/*
 					 * Yes, so we have a deadlock.  Easiest way to clean up
@@ -1077,7 +1120,7 @@ ProcSleep(LOCALLOCK *locallock, LockMethod lockMethodTable)
 					 * a flag to check below, and break out of loop.  Also,
 					 * record deadlock info for later message.
 					 */
-					RememberSimpleDeadLock(MyProc, lockmode, lock, proc);
+					RememberSimpleDeadLock(MyProc, lockmode, lock, cproc);
 					early_deadlock = true;
 					break;
 				}
@@ -1093,22 +1136,19 @@ ProcSleep(LOCALLOCK *locallock, LockMethod lockMethodTable)
 					GrantAwaitedLock();
 					return STATUS_OK;
 				}
-				/* Break out of loop to put myself before him */
+				/* Break out of loop and put myself before him */
+				proc = cproc;
 				break;
 			}
 			/* Nope, so advance to next waiter */
-			aheadRequests |= LOCKBIT_ON(proc->waitLockMode);
-			proc = (PGPROC *) proc->links.next;
+			aheadRequests |= LOCKBIT_ON(cproc->waitLockMode);
+			cproc = (PGPROC *) cproc->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
+
+	if (proc == NULL)
 	{
-		/* I hold no locks, so I can't push in front of anyone. */
+		/* No special case applies, so I can't push in front of anyone. */
 		proc = (PGPROC *) &(waitQueue->links);
 	}