Stash group leader information in PROCLOCK entries.

diff --git a/src/backend/storage/lmgr/lock.c b/src/backend/storage/lmgr/lock.c
index 723051efb50d7f764b10f4e0a4a9406e3de00bc2..e3a2401b2841ce97a60d73fde5c9c7e760a95ba4 100644 (file)
--- a/src/backend/storage/lmgr/lock.c
+++ b/src/backend/storage/lmgr/lock.c
@@ -339,7 +339,8 @@ PROCLOCK_PRINT(const char *where, const PROCLOCK *proclockP)
 static uint32 proclock_hash(const void *key, Size keysize);
 static void RemoveLocalLock(LOCALLOCK *locallock);
 static PROCLOCK *SetupLockInTable(LockMethod lockMethodTable, PGPROC *proc,
-                                const LOCKTAG *locktag, uint32 hashcode, LOCKMODE lockmode);
+                                PGPROC *leader, const LOCKTAG *locktag, uint32 hashcode,
+                                LOCKMODE lockmode);
 static void GrantLockLocal(LOCALLOCK *locallock, ResourceOwner owner);
 static void BeginStrongLockAcquire(LOCALLOCK *locallock, uint32 fasthashcode);
 static void FinishStrongLockAcquire(void);
@@ -894,8 +895,8 @@ LockAcquireExtended(const LOCKTAG *locktag,
         * away anytime.  So we have to use SetupLockInTable() to recompute the
         * lock and proclock pointers, even if they're already set.
         */
-       proclock = SetupLockInTable(lockMethodTable, MyProc, locktag,
-                                                               hashcode, lockmode);
+       proclock = SetupLockInTable(lockMethodTable, MyProc, LockGroupLeader,
+                                                               locktag, hashcode, lockmode);
        if (!proclock)
        {
                AbortStrongLockAcquire();
@@ -1053,7 +1054,7 @@ LockAcquireExtended(const LOCKTAG *locktag,
  * held at exit.
  */
 static PROCLOCK *
-SetupLockInTable(LockMethod lockMethodTable, PGPROC *proc,
+SetupLockInTable(LockMethod lockMethodTable, PGPROC *proc, PGPROC *leader,
                                 const LOCKTAG *locktag, uint32 hashcode, LOCKMODE lockmode)
 {
        LOCK       *lock;
@@ -1141,6 +1142,7 @@ SetupLockInTable(LockMethod lockMethodTable, PGPROC *proc,
        {
                uint32          partition = LockHashPartition(hashcode);
 
+               proclock->groupLeader = leader;
                proclock->holdMask = 0;
                proclock->releaseMask = 0;
                /* Add proclock to appropriate lists */
@@ -2534,7 +2536,8 @@ FastPathTransferRelationLocks(LockMethod lockMethodTable, const LOCKTAG *locktag
 
                                if (!FAST_PATH_CHECK_LOCKMODE(proc, f, lockmode))
                                        continue;
-                               proclock = SetupLockInTable(lockMethodTable, proc, locktag,
+                               proclock = SetupLockInTable(lockMethodTable, proc,
+                                                                                       proc->lockGroupLeader, locktag,
                                                                                        hashcode, lockmode);
                                if (!proclock)
                                {
@@ -2590,8 +2593,8 @@ FastPathGetRelationLockEntry(LOCALLOCK *locallock)
                /* Find or create lock object. */
                LWLockAcquire(partitionLock, LW_EXCLUSIVE);
 
-               proclock = SetupLockInTable(lockMethodTable, MyProc, locktag,
-                                                                       locallock->hashcode, lockmode);
+               proclock = SetupLockInTable(lockMethodTable, MyProc, LockGroupLeader,
+                                                                       locktag, locallock->hashcode, lockmode);
                if (!proclock)
                {
                        LWLockRelease(partitionLock);
@@ -3094,6 +3097,9 @@ PostPrepare_Locks(TransactionId xid)
        PROCLOCKTAG proclocktag;
        int                     partition;
 
+       /* Can't prepare a lock group follower. */
+       Assert(MyProc == LockGroupLeader);
+
        /* This is a critical section: any error means big trouble */
        START_CRIT_SECTION();
 
@@ -3207,6 +3213,7 @@ 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)
@@ -3251,6 +3258,9 @@ 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 */
 
@@ -3784,6 +3794,7 @@ lock_twophase_recover(TransactionId xid, uint16 info,
         */
        if (!found)
        {
+               proclock->groupLeader = proc;
                proclock->holdMask = 0;
                proclock->releaseMask = 0;
                /* Add proclock to appropriate lists */
@@ -4058,7 +4069,8 @@ VirtualXactLock(VirtualTransactionId vxid, bool wait)
                LWLockAcquire(partitionLock, LW_EXCLUSIVE);
 
                proclock = SetupLockInTable(LockMethods[DEFAULT_LOCKMETHOD], proc,
-                                                                       &tag, hashcode, ExclusiveLock);
+                                                                       proc->lockGroupLeader, &tag, hashcode,
+                                                                       ExclusiveLock);
                if (!proclock)
                {
                        LWLockRelease(partitionLock);

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