Revise things so that the group leader fields are NULL, rather than

MyProc, when no group locking is in use.

diff --git a/src/backend/storage/lmgr/lock.c b/src/backend/storage/lmgr/lock.c
index e3a2401b2841ce97a60d73fde5c9c7e760a95ba4..c754dd0af645120f04bb21c0ec786f0ddd153fcb 100644 (file)
--- a/src/backend/storage/lmgr/lock.c
+++ b/src/backend/storage/lmgr/lock.c
@@ -3098,7 +3098,7 @@ PostPrepare_Locks(TransactionId xid)
        int                     partition;
 
        /* Can't prepare a lock group follower. */
-       Assert(MyProc == LockGroupLeader);
+       Assert(LockGroupLeader == MyProc || LockGroupLeader == NULL);
 
        /* This is a critical section: any error means big trouble */
        START_CRIT_SECTION();
@@ -3201,6 +3201,19 @@ PostPrepare_Locks(TransactionId xid)
 
                        Assert(proclock->tag.myProc == MyProc);
 
+                       /*
+                        * We shouldn't be in a lock group, except for a single-entry
+                        * group for which we're the leader, which is OK.  We need to
+                        * clear the groupLeader pointer in that case, so that the dummy
+                        * PGPROC doesn't end up pointing back to our PGPROC.
+                        */
+                       if (proclock->groupLeader != NULL)
+                       {
+                               Assert(proclock->groupLeader == MyProc);
+                               Assert(MyProc->lockGroupMembers == 1);
+                               proclock->groupLeader = NULL;
+                       }
+
                        lock = proclock->tag.myLock;
 
                        /* Ignore VXID locks */
@@ -3213,7 +3226,6 @@ PostPrepare_Locks(TransactionId xid)
                        Assert(lock->nGranted >= 0);
                        Assert(lock->nGranted <= lock->nRequested);
                        Assert((proclock->holdMask & ~lock->grantMask) == 0);
-                       Assert(proclock->tag.myProc == proclock->groupLeader);
 
                        /* Ignore it if nothing to release (must be a session lock) */
                        if (proclock->releaseMask == 0)
@@ -3258,9 +3270,6 @@ PostPrepare_Locks(TransactionId xid)
                        SHMQueueInsertBefore(&(newproc->myProcLocks[partition]),
                                                                 &proclock->procLink);
 
-                       /* Update lock group leader. */
-                       proclock->groupLeader = newproc;
-
                        PROCLOCK_PRINT("PostPrepare_Locks: updated", proclock);
                }                                               /* loop over PROCLOCKs within this partition */
 
@@ -3794,7 +3803,7 @@ lock_twophase_recover(TransactionId xid, uint16 info,
         */
        if (!found)
        {
-               proclock->groupLeader = proc;
+               proclock->groupLeader = NULL;
                proclock->holdMask = 0;
                proclock->releaseMask = 0;
                /* Add proclock to appropriate lists */

diff --git a/src/backend/storage/lmgr/proc.c b/src/backend/storage/lmgr/proc.c
index 778d05ade0613cb2e7ce126aff8de9af32c264d8..4706c2207830032beb43e662985cd615f576003e 100644 (file)
--- a/src/backend/storage/lmgr/proc.c
+++ b/src/backend/storage/lmgr/proc.c
@@ -64,10 +64,9 @@ PGPROC          *MyProc = NULL;
 PGXACT    *MyPgXact = NULL;
 
 /*
- * If we're not in a lock group, LockGroupLeader should be set to the
- * same value as MyProc, if this backend intends to do any heavyweight locking.
- * This also occurs when we are a lock group leader.  When we are a lock
- * group following, LockGroupLeader points to the group leader's PGPROC.
+ * If we're not in a lock group, LockGroupLeader will be NULL.  Otherwise,
+ * it should be the set to the leader of the lock group of which we're a
+ * member.  This will be the same as MyProc iff we're the group leader.
  */
 PGPROC    *LockGroupLeader = NULL;
 
@@ -238,7 +237,6 @@ InitProcGlobal(void)
                }
 
                procs[i].pgprocno = i;
-               procs[i].lockGroupLeader = &procs[i];
 
                /*
                 * Newly created PGPROCs for normal backends, autovacuum and bgworkers
@@ -354,7 +352,6 @@ InitProcess(void)
                                 errmsg("sorry, too many clients already")));
        }
        MyPgXact = &ProcGlobal->allPgXact[MyProc->pgprocno];
-       LockGroupLeader = MyProc;
 
        /*
         * Cross-check that the PGPROC is of the type we expect; if this were
@@ -397,7 +394,7 @@ InitProcess(void)
        MyProc->lwWaitLink = NULL;
        MyProc->waitLock = NULL;
        MyProc->waitProcLock = NULL;
-       Assert(MyProc->lockGroupLeader == MyProc);
+       Assert(MyProc->lockGroupLeader == NULL);
 #ifdef USE_ASSERT_CHECKING
        {
                int                     i;
@@ -561,7 +558,7 @@ InitAuxiliaryProcess(void)
        MyProc->lwWaitLink = NULL;
        MyProc->waitLock = NULL;
        MyProc->waitProcLock = NULL;
-       Assert(MyProc->lockGroupLeader == MyProc);
+       Assert(MyProc->lockGroupLeader == NULL);
 #ifdef USE_ASSERT_CHECKING
        {
                int                     i;
@@ -816,7 +813,7 @@ ProcKill(int code, Datum arg)
                ReplicationSlotRelease();
 
        /* If we're a lock group member, detach from the lock group. */
-       if (LockGroupLeader != MyProc)
+       if (LockGroupLeader != NULL && LockGroupLeader != MyProc)
        {
                int     members;
 
@@ -825,7 +822,7 @@ ProcKill(int code, Datum arg)
                LWLockRelease(LockGroupLeader->backendLock);
 
                LWLockAcquire(MyProc->backendLock, LW_EXCLUSIVE);
-               MyProc->lockGroupLeader = MyProc;
+               MyProc->lockGroupLeader = NULL;
                LWLockRelease(MyProc->backendLock);
 
                /* If we're the last member of the lock group, detach the PGPROC. */
@@ -874,7 +871,7 @@ ProcKill(int code, Datum arg)
         * be a lock group leader, allowing our PGPROC to be recycled
         * immediately.
         */
-       if (IsLockGroupLeader)
+       if (LockGroupLeader == MyProc)
        {
                int     members;
 
@@ -882,8 +879,7 @@ ProcKill(int code, Datum arg)
                members = --proc->lockGroupMembers;
                LWLockRelease(proc->backendLock);
 
-               if (members == 0)
-                       IsLockGroupLeader = false;
+               LockGroupLeader = NULL;
        }
 
        /*
@@ -891,7 +887,7 @@ ProcKill(int code, Datum arg)
         * designation, then we can immediately release our PGPROC.  If not, then
         * the last group member will do that on exit.
         */
-       if (!IsLockGroupLeader)
+       if (LockGroupLeader == NULL)
        {
                PGPROC * volatile * procgloballist;
 
@@ -1709,18 +1705,18 @@ void
 BecomeLockGroupLeader(void)
 {
        /* Can't be leader and follower. */
-       Assert(LockGroupLeader == MyProc);
+       Assert(LockGroupLeader == NULL || LockGroupLeader == MyProc);
 
        /* This can be called more than once; but we must not redo the work. */
-       if (!IsLockGroupLeader)
+       if (LockGroupLeader == NULL)
        {
                LWLockAcquire(MyProc->backendLock, LW_EXCLUSIVE);
                Assert(MyProc->lockGroupMembers == 0);
-               Assert(MyProc->lockGroupLeader == MyProc);
+               Assert(MyProc->lockGroupLeader == NULL);
+               MyProc->lockGroupLeader = MyProc;
                MyProc->lockGroupLeaderIdentifier = MyProcPid;
                MyProc->lockGroupMembers = 1;
                LWLockRelease(MyProc->backendLock);
-               IsLockGroupLeader = true;
        }
 }
 
@@ -1739,11 +1735,8 @@ BecomeLockGroupFollower(PGPROC *leader, int pid)
 {
        bool    ok = false;
 
-       /* Can't be leader and follower. */
-       Assert(!IsLockGroupLeader);
-
-       /* Can't become a follower if we already are one. */
-       Assert(MyProc == LockGroupLeader);
+       /* Can't become a follower if we already in a lock group. */
+       Assert(LockGroupLeader == NULL);
 
        /* Can't follow ourselves. */
        Assert(MyProc != leader);

diff --git a/src/include/storage/lock.h b/src/include/storage/lock.h
index 8f8fed9b684a5153803a9a5ac31d00014077f0f8..33aece3927fbfb295cf767d734d1e1612af91b7e 100644 (file)
--- a/src/include/storage/lock.h
+++ b/src/include/storage/lock.h
@@ -362,7 +362,7 @@ typedef struct PROCLOCK
        PROCLOCKTAG tag;                        /* unique identifier of proclock object */
 
        /* data */
-       PGPROC     *groupLeader;        /* group leader, or MyProc if no lock group */
+       PGPROC     *groupLeader;        /* group leader, or NULL if no lock group */
        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 */