diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c index df4853b..5f92627 100644 --- a/src/backend/access/heap/heapam.c +++ b/src/backend/access/heap/heapam.c @@ -93,6 +93,9 @@ static void HeapSatisfiesHOTandKeyUpdate(Relation relation, bool *satisfies_hot, bool *satisfies_key, bool *satisfies_id, HeapTuple oldtup, HeapTuple newtup); +static bool heap_acquire_tuplock(Relation relation, ItemPointer tid, + LockTupleMode mode, LockWaitPolicy wait_policy, + bool *have_tuple_lock); static void compute_new_xmax_infomask(TransactionId xmax, uint16 old_infomask, uint16 old_infomask2, TransactionId add_to_xmax, LockTupleMode mode, bool is_update, @@ -110,6 +113,8 @@ static void MultiXactIdWait(MultiXactId multi, MultiXactStatus status, uint16 in int *remaining); static bool ConditionalMultiXactIdWait(MultiXactId multi, MultiXactStatus status, uint16 infomask, Relation rel, int *remaining); +static bool Would_MultiXactIdWait_Block(MultiXactId multi, MultiXactStatus status, + uint16 infomask); static XLogRecPtr log_heap_new_cid(Relation relation, HeapTuple tup); static HeapTuple ExtractReplicaIdentity(Relation rel, HeapTuple tup, bool key_modified, bool *copy); @@ -2700,11 +2705,8 @@ l1: * this arranges that we stay at the head of the line while rechecking * tuple state. */ - if (!have_tuple_lock) - { - LockTupleTuplock(relation, &(tp.t_self), LockTupleExclusive); - have_tuple_lock = true; - } + heap_acquire_tuplock(relation, &(tp.t_self), LockTupleExclusive, + LockWaitBlock, &have_tuple_lock); /* * Sleep until concurrent transaction ends. Note that we don't care @@ -3224,21 +3226,6 @@ l2: LockBuffer(buffer, BUFFER_LOCK_UNLOCK); /* - * Acquire tuple lock to establish our priority for the tuple (see - * heap_lock_tuple). LockTuple will release us when we are - * next-in-line for the tuple. - * - * If we are forced to "start over" below, we keep the tuple lock; - * this arranges that we stay at the head of the line while rechecking - * tuple state. - */ - if (!have_tuple_lock) - { - LockTupleTuplock(relation, &(oldtup.t_self), *lockmode); - have_tuple_lock = true; - } - - /* * Now we have to do something about the existing locker. If it's a * multi, sleep on it; we might be awakened before it is completely * gone (or even not sleep at all in some cases); we need to preserve @@ -3248,12 +3235,30 @@ l2: * before actually going to sleep. If the update doesn't conflict * with the locks, we just continue without sleeping (but making sure * it is preserved). + * + * Before sleeping, we need to acquire tuple lock to establish our + * priority for the tuple (see heap_lock_tuple). LockTuple will + * release us when we are next-in-line for the tuple. Note we must not + * acquire the tuple lock until we're sure we're going to sleep; + * otherwise we're open for race conditions with other transactions + * holding the tuple lock which sleep on us. + * + * If we are forced to "start over" below, we keep the tuple lock; + * this arranges that we stay at the head of the line while rechecking + * tuple state. */ if (infomask & HEAP_XMAX_IS_MULTI) { TransactionId update_xact; int remain; + /* acquire tuple lock, if necessary */ + if (Would_MultiXactIdWait_Block((MultiXactId) xwait, mxact_status, infomask)) + { + heap_acquire_tuplock(relation, &(oldtup.t_self), *lockmode, + LockWaitBlock, &have_tuple_lock); + } + /* wait for multixact */ MultiXactIdWait((MultiXactId) xwait, mxact_status, infomask, relation, &oldtup.t_data->t_ctid, XLTW_Update, @@ -3330,7 +3335,12 @@ l2: } else { - /* wait for regular transaction to end */ + /* + * Wait for regular transaction to end; but first, acquire + * tuple lock. + */ + heap_acquire_tuplock(relation, &(oldtup.t_self), *lockmode, + LockWaitBlock, &have_tuple_lock); XactLockTableWait(xwait, relation, &oldtup.t_data->t_ctid, XLTW_Update); LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE); @@ -4038,6 +4048,48 @@ get_mxact_status_for_lock(LockTupleMode mode, bool is_update) return (MultiXactStatus) retval; } +/* + * Acquire heavyweight lock on the given tuple, in preparation for acquiring + * its normal, Xmax-based tuple lock. + * + * have_tuple_lock is an input and output parameter: on input, it indicates + * whether the lock has previously been acquired (and this function does + * nothing in that case). If this function returns success, have_tuple_lock + * has been flipped to true. + * + * Returns false if it was unable to obtain the lock; this can only happen if + * wait_policy is Skip. + */ +static bool +heap_acquire_tuplock(Relation relation, ItemPointer tid, LockTupleMode mode, + LockWaitPolicy wait_policy, bool *have_tuple_lock) +{ + if (*have_tuple_lock) + return true; + + switch (wait_policy) + { + case LockWaitBlock: + LockTupleTuplock(relation, tid, mode); + break; + + case LockWaitSkip: + if (!ConditionalLockTupleTuplock(relation, tid, mode)) + return false; + break; + + case LockWaitError: + if (!ConditionalLockTupleTuplock(relation, tid, mode)) + ereport(ERROR, + (errcode(ERRCODE_LOCK_NOT_AVAILABLE), + errmsg("could not obtain lock on row in relation \"%s\"", + RelationGetRelationName(relation)))); + break; + } + *have_tuple_lock = true; + + return true; +} /* * heap_lock_tuple - lock a tuple in shared or exclusive mode @@ -4171,42 +4223,6 @@ l3: } /* - * Acquire tuple lock to establish our priority for the tuple. - * LockTuple will release us when we are next-in-line for the tuple. - * We must do this even if we are share-locking. - * - * If we are forced to "start over" below, we keep the tuple lock; - * this arranges that we stay at the head of the line while rechecking - * tuple state. - */ - if (!have_tuple_lock) - { - switch (wait_policy) - { - case LockWaitBlock: - LockTupleTuplock(relation, tid, mode); - break; - case LockWaitSkip: - if (!ConditionalLockTupleTuplock(relation, tid, mode)) - { - result = HeapTupleWouldBlock; - /* recovery code expects to have buffer lock held */ - LockBuffer(*buffer, BUFFER_LOCK_EXCLUSIVE); - goto failed; - } - break; - case LockWaitError: - if (!ConditionalLockTupleTuplock(relation, tid, mode)) - ereport(ERROR, - (errcode(ERRCODE_LOCK_NOT_AVAILABLE), - errmsg("could not obtain lock on row in relation \"%s\"", - RelationGetRelationName(relation)))); - break; - } - have_tuple_lock = true; - } - - /* * Initially assume that we will have to wait for the locking * transaction(s) to finish. We check various cases below in which * this can be turned off. @@ -4327,6 +4343,9 @@ l3: * We don't need to allow old multixacts here; if that had * been the case, HeapTupleSatisfiesUpdate would have returned * MayBeUpdated and we wouldn't be here. + * + * XXX looks like this can be replaced with a call to + * Would_MultiXactIdWait_Block */ nmembers = GetMultiXactIdMembers(xwait, &members, false, @@ -4397,6 +4416,28 @@ l3: if (require_sleep) { + /* + * Acquire tuple lock to establish our priority for the tuple. + * LockTuple will release us when we are next-in-line for the tuple. + * We must do this even if we are share-locking. + * + * If we are forced to "start over" below, we keep the tuple lock; + * this arranges that we stay at the head of the line while rechecking + * tuple state. + */ + if (!heap_acquire_tuplock(relation, tid, mode, wait_policy, + &have_tuple_lock)) + { + /* + * This can only happen if wait_policy is Skip and the lock + * couldn't be obtained. + */ + result = HeapTupleWouldBlock; + /* recovery code expects to have buffer lock held */ + LockBuffer(*buffer, BUFFER_LOCK_EXCLUSIVE); + goto failed; + } + if (infomask & HEAP_XMAX_IS_MULTI) { MultiXactStatus status = get_mxact_status_for_lock(mode, false); @@ -6188,6 +6229,48 @@ Do_MultiXactIdWait(MultiXactId multi, MultiXactStatus status, } /* + * return true if sleeping on the given multixact with the given status would + * cause to sleep until some xact is done. + */ +static bool +Would_MultiXactIdWait_Block(MultiXactId multi, MultiXactStatus status, + uint16 infomask) +{ + bool allow_old; + int nmembers; + MultiXactMember *members; + bool result = false; + + allow_old = !(infomask & HEAP_LOCK_MASK) && HEAP_XMAX_IS_LOCKED_ONLY(infomask); + nmembers = GetMultiXactIdMembers(multi, &members, allow_old, + HEAP_XMAX_IS_LOCKED_ONLY(infomask)); + + if (nmembers >= 0) + { + int i; + + for (i = 0; i < nmembers; i++) + { + TransactionId memxid = members[i].xid; + MultiXactStatus memstatus = members[i].status; + + if (TransactionIdIsCurrentTransactionId(memxid)) + continue; + + if (!DoLockModesConflict(LOCKMODE_from_mxstatus(memstatus), + LOCKMODE_from_mxstatus(status))) + continue; + + result = true; + break; + } + pfree(members); + } + + return result; +} + +/* * MultiXactIdWait * Sleep on a MultiXactId. *