Re: FOR SHARE vs FOR UPDATE locks - Mailing list pgsql-hackers

"Heikki Linnakangas" <heikki@enterprisedb.com> writes:
> That way, the lock won't be downgraded back to a shared lock on 
> "rollback to savepoint", right? Though it's still better than throwing 
> an error, I think.

Correct, a rollback would leave the tuple still exclusive-locked.
So not perfect, but it's hard to see how to do better without a whole
lot more infrastructure, which the case is probably not worth.

I've just finished coding up the patch --- untested as yet, but anyone
see any problems?
        regards, tom lane

*** src/backend/access/heap/heapam.c.orig    Fri Nov 17 13:00:14 2006
--- src/backend/access/heap/heapam.c    Fri Dec  1 12:18:04 2006
***************
*** 2360,2365 ****
--- 2360,2366 ----     PageHeader    dp;     TransactionId xid;     TransactionId xmax;
+     TransactionId existing_subxact = InvalidTransactionId;     uint16        old_infomask;     uint16
new_infomask;    LOCKMODE    tuple_lock_type;
 
***************
*** 2398,2419 ****         LockBuffer(*buffer, BUFFER_LOCK_UNLOCK);          /*
!          * If we wish to acquire share lock, and the tuple is already
!          * share-locked by a multixact that includes any subtransaction of the
!          * current top transaction, then we effectively hold the desired lock
!          * already.  We *must* succeed without trying to take the tuple lock,
!          * else we will deadlock against anyone waiting to acquire exclusive
!          * lock.  We don't need to make any state changes in this case.          */
!         if (mode == LockTupleShared &&
!             (infomask & HEAP_XMAX_IS_MULTI) &&
!             MultiXactIdIsCurrent((MultiXactId) xwait))         {             Assert(infomask &
HEAP_XMAX_SHARED_LOCK);
!             /* Probably can't hold tuple lock here, but may as well check */
!             if (have_tuple_lock)
!                 UnlockTuple(relation, tid, tuple_lock_type);
!             return HeapTupleMayBeUpdated;         }          /*
--- 2399,2430 ----         LockBuffer(*buffer, BUFFER_LOCK_UNLOCK);          /*
!          * If the tuple is currently share-locked by a multixact, we have to
!          * check whether the multixact includes any live subtransaction of the
!          * current top transaction.  If so, then we effectively already hold
!          * share-lock, even if that XID isn't the current subxact.  That's
!          * because no such subtransaction could be aborted without also
!          * aborting the current subtransaction, and so its locks are as good
!          * as ours.          */
!         if (infomask & HEAP_XMAX_IS_MULTI)         {             Assert(infomask & HEAP_XMAX_SHARED_LOCK);
!             existing_subxact = MultiXactIdGetCurrent((MultiXactId) xwait);
!             /*
!              * Done if we have share lock and that's what the caller wants.
!              * We *must* check this before trying to take the tuple lock, else
!              * we will deadlock against anyone waiting to acquire exclusive
!              * lock.  We don't need to make any state changes in this case.
!              */
!             if (mode == LockTupleShared &&
!                 TransactionIdIsValid(existing_subxact))
!             {
!                 /* Probably can't hold tuple lock here, but check anyway */
!                 if (have_tuple_lock)
!                     UnlockTuple(relation, tid, tuple_lock_type);
!                 return HeapTupleMayBeUpdated;
!             }         }          /*
***************
*** 2570,2593 ****     if (!(old_infomask & (HEAP_XMAX_INVALID |                           HEAP_XMAX_COMMITTED |
                  HEAP_XMAX_IS_MULTI)) &&
 
-         (mode == LockTupleShared ?
-          (old_infomask & HEAP_IS_LOCKED) :
-          (old_infomask & HEAP_XMAX_EXCL_LOCK)) &&         TransactionIdIsCurrentTransactionId(xmax))     {
!         LockBuffer(*buffer, BUFFER_LOCK_UNLOCK);
!         /* Probably can't hold tuple lock here, but may as well check */
!         if (have_tuple_lock)
!             UnlockTuple(relation, tid, tuple_lock_type);
!         return HeapTupleMayBeUpdated;     }      /*      * Compute the new xmax and infomask to store into the tuple.
Note we do      * not modify the tuple just yet, because that would leave it in the wrong      * state if multixact.c
elogs.     */
 
!     xid = GetCurrentTransactionId();      new_infomask = old_infomask & ~(HEAP_XMAX_COMMITTED |
             HEAP_XMAX_INVALID |
 
--- 2581,2621 ----     if (!(old_infomask & (HEAP_XMAX_INVALID |                           HEAP_XMAX_COMMITTED |
                  HEAP_XMAX_IS_MULTI)) &&         TransactionIdIsCurrentTransactionId(xmax))     {
 
!         /* The tuple is locked by some existing subxact ... */
!         Assert(old_infomask & HEAP_IS_LOCKED);
!         existing_subxact = xmax;
!         /* ... but is it the desired lock type or stronger? */
!         if (mode == LockTupleShared ||
!             (old_infomask & HEAP_XMAX_EXCL_LOCK))
!         {
!             LockBuffer(*buffer, BUFFER_LOCK_UNLOCK);
!             /* Probably can't hold tuple lock here, but check anyway */
!             if (have_tuple_lock)
!                 UnlockTuple(relation, tid, tuple_lock_type);
!             return HeapTupleMayBeUpdated;
!         }     }      /*      * Compute the new xmax and infomask to store into the tuple.  Note we do      * not
modifythe tuple just yet, because that would leave it in the wrong      * state if multixact.c elogs.
 
+      *
+      * If we are upgrading a shared lock held by another subxact to exclusive,
+      * we need to mark the tuple as exclusively locked by the other subxact
+      * not this one.  Otherwise, a rollback of this subxact would leave the
+      * tuple apparently not locked at all.  We don't have enough
+      * infrastructure to keep track of both types of tuple lock, so we
+      * compromise by making the tuple appear to be exclusive-locked by the
+      * other, possibly longer-lived subxact.  (Again, there are no cases where
+      * a live subxact could be shorter-lived than the current one.)      */
!     if (TransactionIdIsValid(existing_subxact))
!         xid = existing_subxact;
!     else
!         xid = GetCurrentTransactionId();      new_infomask = old_infomask & ~(HEAP_XMAX_COMMITTED |
                 HEAP_XMAX_INVALID |
 
*** src/backend/access/transam/multixact.c.orig    Fri Nov 17 13:00:15 2006
--- src/backend/access/transam/multixact.c    Fri Dec  1 12:17:57 2006
***************
*** 381,388 ****      /*      * Checking for myself is cheap compared to looking in shared memory,
!      * so first do the equivalent of MultiXactIdIsCurrent().  This is not
!      * needed for correctness, it's just a fast path.      */     for (i = 0; i < nmembers; i++)     {
--- 381,388 ----      /*      * Checking for myself is cheap compared to looking in shared memory,
!      * so try that first.  This is not needed for correctness, it's just a
!      * fast path.      */     for (i = 0; i < nmembers; i++)     {
***************
*** 418,437 **** }  /*
!  * MultiXactIdIsCurrent
!  *        Returns true if the current transaction is a member of the MultiXactId.  *
!  * We return true if any live subtransaction of the current top-level
!  * transaction is a member.  This is appropriate for the same reason that a
!  * lock held by any such subtransaction is globally equivalent to a lock
!  * held by the current subtransaction: no such lock could be released without
!  * aborting this subtransaction, and hence releasing its locks.  So it's not
!  * necessary to add the current subxact to the MultiXact separately.  */
! bool
! MultiXactIdIsCurrent(MultiXactId multi) {
!     bool        result = false;     TransactionId *members;     int            nmembers;     int            i;
--- 418,437 ---- }  /*
!  * MultiXactIdGetCurrent
!  *        If any live subtransaction of the current backend is a member of
!  *        the MultiXactId, return its XID; else return InvalidTransactionId.  *
!  * If the MXACT contains multiple such subtransactions, it is unspecified
!  * which one is returned.  This doesn't matter in current usage because
!  * heap_lock_tuple takes care not to insert multiple subtransactions of the
!  * same backend into any MXACT.  If need be, we could modify this code to
!  * return the oldest such subxact, or some such rule.  */
! TransactionId
! MultiXactIdGetCurrent(MultiXactId multi) {
!     TransactionId result = InvalidTransactionId;     TransactionId *members;     int            nmembers;     int
      i;
 
***************
*** 439,451 ****     nmembers = GetMultiXactIdMembers(multi, &members);      if (nmembers < 0)
!         return false;      for (i = 0; i < nmembers; i++)     {         if
(TransactionIdIsCurrentTransactionId(members[i]))        {
 
!             result = true;             break;         }     }
--- 439,451 ----     nmembers = GetMultiXactIdMembers(multi, &members);      if (nmembers < 0)
!         return result;      for (i = 0; i < nmembers; i++)     {         if
(TransactionIdIsCurrentTransactionId(members[i]))        {
 
!             result = members[i];             break;         }     }
*** src/include/access/multixact.h.orig    Fri Nov 17 13:00:15 2006
--- src/include/access/multixact.h    Fri Dec  1 12:17:49 2006
***************
*** 45,51 **** extern MultiXactId MultiXactIdCreate(TransactionId xid1, TransactionId xid2); extern MultiXactId
MultiXactIdExpand(MultiXactIdmulti, TransactionId xid); extern bool MultiXactIdIsRunning(MultiXactId multi);
 
! extern bool MultiXactIdIsCurrent(MultiXactId multi); extern void MultiXactIdWait(MultiXactId multi); extern bool
ConditionalMultiXactIdWait(MultiXactIdmulti); extern void MultiXactIdSetOldestMember(void);
 
--- 45,51 ---- extern MultiXactId MultiXactIdCreate(TransactionId xid1, TransactionId xid2); extern MultiXactId
MultiXactIdExpand(MultiXactIdmulti, TransactionId xid); extern bool MultiXactIdIsRunning(MultiXactId multi);
 
! extern TransactionId MultiXactIdGetCurrent(MultiXactId multi); extern void MultiXactIdWait(MultiXactId multi); extern
boolConditionalMultiXactIdWait(MultiXactId multi); extern void MultiXactIdSetOldestMember(void);