greenplumn lmgr 源码
greenplumn lmgr 代码
文件路径：/src/backend/storage/lmgr/lmgr.c
/*-------------------------------------------------------------------------
*
* lmgr.c
* POSTGRES lock manager code
*
* Portions Copyright (c) 2006-2008, Greenplum inc
* Portions Copyright (c) 2012-Present VMware, Inc. or its affiliates.
* Portions Copyright (c) 1996-2019, PostgreSQL Global Development Group
* Portions Copyright (c) 1994, Regents of the University of California
*
*
* IDENTIFICATION
* src/backend/storage/lmgr/lmgr.c
*
*-------------------------------------------------------------------------
*/
#include "postgres.h"
#include "access/subtrans.h"
#include "access/transam.h"
#include "access/xact.h"
#include "catalog/catalog.h"
#include "commands/progress.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "storage/lmgr.h"
#include "storage/procarray.h"
#include "storage/sinvaladt.h"
#include "utils/inval.h"
#include "utils/memutils.h"
#include "access/heapam.h"
#include "catalog/namespace.h"
#include "cdb/cdbvars.h"
#include "storage/proc.h"
#include "utils/lsyscache.h" /* CDB: get_rel_namespace() */
/*
* Per-backend counter for generating speculative insertion tokens.
*
* This may wrap around, but that's OK as it's only used for the short
* duration between inserting a tuple and checking that there are no (unique)
* constraint violations. It's theoretically possible that a backend sees a
* tuple that was speculatively inserted by another backend, but before it has
* started waiting on the token, the other backend completes its insertion,
* and then performs 2^32 unrelated insertions. And after all that, the
* first backend finally calls SpeculativeInsertionLockAcquire(), with the
* intention of waiting for the first insertion to complete, but ends up
* waiting for the latest unrelated insertion instead. Even then, nothing
* particularly bad happens: in the worst case they deadlock, causing one of
* the transactions to abort.
*/
static uint32 speculativeInsertionToken = 0;
/*
* Struct to hold context info for transaction lock waits.
*
* 'oper' is the operation that needs to wait for the other transaction; 'rel'
* and 'ctid' specify the address of the tuple being waited for.
*/
typedef struct XactLockTableWaitInfo
{
XLTW_Oper oper;
Relation rel;
ItemPointer ctid;
} XactLockTableWaitInfo;
static void XactLockTableWaitErrorCb(void *arg);
/*
* RelationInitLockInfo
* Initializes the lock information in a relation descriptor.
*
* relcache.c must call this during creation of any reldesc.
*/
void
RelationInitLockInfo(Relation relation)
{
Assert(RelationIsValid(relation));
Assert(OidIsValid(RelationGetRelid(relation)));
relation->rd_lockInfo.lockRelId.relId = RelationGetRelid(relation);
if (relation->rd_rel->relisshared)
relation->rd_lockInfo.lockRelId.dbId = InvalidOid;
else
relation->rd_lockInfo.lockRelId.dbId = MyDatabaseId;
}
/*
* SetLocktagRelationOid
* Set up a locktag for a relation, given only relation OID
*/
static inline void
SetLocktagRelationOid(LOCKTAG *tag, Oid relid)
{
Oid dbid;
if (IsSharedRelation(relid))
dbid = InvalidOid;
else
dbid = MyDatabaseId;
SET_LOCKTAG_RELATION(*tag, dbid, relid);
}
/*
* LockRelationOid
*
* Lock a relation given only its OID. This should generally be used
* before attempting to open the relation's relcache entry.
*/
void
LockRelationOid(Oid relid, LOCKMODE lockmode)
{
LOCKTAG tag;
LOCALLOCK *locallock;
LockAcquireResult res;
SetLocktagRelationOid(&tag, relid);
res = LockAcquireExtended(&tag, lockmode, false, false, true, &locallock);
/*
* Now that we have the lock, check for invalidation messages, so that we
* will update or flush any stale relcache entry before we try to use it.
* RangeVarGetRelid() specifically relies on us for this. We can skip
* this in the not-uncommon case that we already had the same type of lock
* being requested, since then no one else could have modified the
* relcache entry in an undesirable way. (In the case where our own xact
* modifies the rel, the relcache update happens via
* CommandCounterIncrement, not here.)
*
* However, in corner cases where code acts on tables (usually catalogs)
* recursively, we might get here while still processing invalidation
* messages in some outer execution of this function or a sibling. The
* "cleared" status of the lock tells us whether we really are done
* absorbing relevant inval messages.
*/
if (res != LOCKACQUIRE_ALREADY_CLEAR)
{
AcceptInvalidationMessages();
MarkLockClear(locallock);
}
}
/*
* ConditionalLockRelationOid
*
* As above, but only lock if we can get the lock without blocking.
* Returns true iff the lock was acquired.
*
* NOTE: we do not currently need conditional versions of all the
* LockXXX routines in this file, but they could easily be added if needed.
*/
bool
ConditionalLockRelationOid(Oid relid, LOCKMODE lockmode)
{
LOCKTAG tag;
LOCALLOCK *locallock;
LockAcquireResult res;
SetLocktagRelationOid(&tag, relid);
res = LockAcquireExtended(&tag, lockmode, false, true, true, &locallock);
if (res == LOCKACQUIRE_NOT_AVAIL)
return false;
/*
* Now that we have the lock, check for invalidation messages; see notes
* in LockRelationOid.
*/
if (res != LOCKACQUIRE_ALREADY_CLEAR)
{
AcceptInvalidationMessages();
MarkLockClear(locallock);
}
return true;
}
/*
* UnlockRelationId
*
* Unlock, given a LockRelId. This is preferred over UnlockRelationOid
* for speed reasons.
*/
void
UnlockRelationId(LockRelId *relid, LOCKMODE lockmode)
{
LOCKTAG tag;
SET_LOCKTAG_RELATION(tag, relid->dbId, relid->relId);
LockRelease(&tag, lockmode, false);
}
/*
* UnlockRelationOid
*
* Unlock, given only a relation Oid. Use UnlockRelationId if you can.
*/
void
UnlockRelationOid(Oid relid, LOCKMODE lockmode)
{
LOCKTAG tag;
SetLocktagRelationOid(&tag, relid);
LockRelease(&tag, lockmode, false);
}
/*
* LockRelation
*
* This is a convenience routine for acquiring an additional lock on an
* already-open relation. Never try to do "relation_open(foo, NoLock)"
* and then lock with this.
*/
void
LockRelation(Relation relation, LOCKMODE lockmode)
{
LOCKTAG tag;
LOCALLOCK *locallock;
LockAcquireResult res;
SET_LOCKTAG_RELATION(tag,
relation->rd_lockInfo.lockRelId.dbId,
relation->rd_lockInfo.lockRelId.relId);
res = LockAcquireExtended(&tag, lockmode, false, false, true, &locallock);
/*
* Now that we have the lock, check for invalidation messages; see notes
* in LockRelationOid.
*/
if (res != LOCKACQUIRE_ALREADY_CLEAR)
{
AcceptInvalidationMessages();
MarkLockClear(locallock);
}
}
/*
* LockRelationNoWait
*
* Similar to LockReation, except that it is not waiting
* for the lock if it is not available.
*
*/
LockAcquireResult
LockRelationNoWait(Relation relation, LOCKMODE lockmode)
{
LOCKTAG tag;
LockAcquireResult res;
SET_LOCKTAG_RELATION(tag,
relation->rd_lockInfo.lockRelId.dbId,
relation->rd_lockInfo.lockRelId.relId);
res = LockAcquire(&tag, lockmode, false, true);
/*
* Now that we have the lock, check for invalidation messages; see notes
* in LockRelationOid.
*/
if (res != LOCKACQUIRE_ALREADY_HELD)
AcceptInvalidationMessages();
return res;
}
/*
* ConditionalLockRelation
*
* This is a convenience routine for acquiring an additional lock on an
* already-open relation. Never try to do "relation_open(foo, NoLock)"
* and then lock with this.
*/
bool
ConditionalLockRelation(Relation relation, LOCKMODE lockmode)
{
LOCKTAG tag;
LOCALLOCK *locallock;
LockAcquireResult res;
SET_LOCKTAG_RELATION(tag,
relation->rd_lockInfo.lockRelId.dbId,
relation->rd_lockInfo.lockRelId.relId);
res = LockAcquireExtended(&tag, lockmode, false, true, true, &locallock);
if (res == LOCKACQUIRE_NOT_AVAIL)
return false;
/*
* Now that we have the lock, check for invalidation messages; see notes
* in LockRelationOid.
*/
if (res != LOCKACQUIRE_ALREADY_CLEAR)
{
AcceptInvalidationMessages();
MarkLockClear(locallock);
}
return true;
}
/*
* UnlockRelation
*
* This is a convenience routine for unlocking a relation without also
* closing it.
*/
void
UnlockRelation(Relation relation, LOCKMODE lockmode)
{
LOCKTAG tag;
SET_LOCKTAG_RELATION(tag,
relation->rd_lockInfo.lockRelId.dbId,
relation->rd_lockInfo.lockRelId.relId);
LockRelease(&tag, lockmode, false);
}
/*
* CheckRelationLockedByMe
*
* Returns true if current transaction holds a lock on 'relation' of mode
* 'lockmode'. If 'orstronger' is true, a stronger lockmode is also OK.
* ("Stronger" is defined as "numerically higher", which is a bit
* semantically dubious but is OK for the purposes we use this for.)
*/
bool
CheckRelationLockedByMe(Relation relation, LOCKMODE lockmode, bool orstronger)
{
LOCKTAG tag;
SET_LOCKTAG_RELATION(tag,
relation->rd_lockInfo.lockRelId.dbId,
relation->rd_lockInfo.lockRelId.relId);
if (LockHeldByMe(&tag, lockmode))
return true;
if (orstronger)
{
LOCKMODE slockmode;
for (slockmode = lockmode + 1;
slockmode <= MaxLockMode;
slockmode++)
{
if (LockHeldByMe(&tag, slockmode))
{
#ifdef NOT_USED
/* Sometimes this might be useful for debugging purposes */
elog(WARNING, "lock mode %s substituted for %s on relation %s",
GetLockmodeName(tag.locktag_lockmethodid, slockmode),
GetLockmodeName(tag.locktag_lockmethodid, lockmode),
RelationGetRelationName(relation));
#endif
return true;
}
}
}
return false;
}
/*
* LockHasWaitersRelation
*
* This is a function to check whether someone else is waiting for a
* lock which we are currently holding.
*/
bool
LockHasWaitersRelation(Relation relation, LOCKMODE lockmode)
{
LOCKTAG tag;
SET_LOCKTAG_RELATION(tag,
relation->rd_lockInfo.lockRelId.dbId,
relation->rd_lockInfo.lockRelId.relId);
return LockHasWaiters(&tag, lockmode, false);
}
/*
* LockRelationIdForSession
*
* This routine grabs a session-level lock on the target relation. The
* session lock persists across transaction boundaries. It will be removed
* when UnlockRelationIdForSession() is called, or if an ereport(ERROR) occurs,
* or if the backend exits.
*
* Note that one should also grab a transaction-level lock on the rel
* in any transaction that actually uses the rel, to ensure that the
* relcache entry is up to date.
*/
void
LockRelationIdForSession(LockRelId *relid, LOCKMODE lockmode)
{
LOCKTAG tag;
SET_LOCKTAG_RELATION(tag, relid->dbId, relid->relId);
(void) LockAcquire(&tag, lockmode, true, false);
}
/*
* UnlockRelationIdForSession
*/
void
UnlockRelationIdForSession(LockRelId *relid, LOCKMODE lockmode)
{
LOCKTAG tag;
SET_LOCKTAG_RELATION(tag, relid->dbId, relid->relId);
LockRelease(&tag, lockmode, true);
}
/*
* LockRelationForExtension
*
* This lock tag is used to interlock addition of pages to relations.
* We need such locking because bufmgr/smgr definition of P_NEW is not
* race-condition-proof.
*
* We assume the caller is already holding some type of regular lock on
* the relation, so no AcceptInvalidationMessages call is needed here.
*/
void
LockRelationForExtension(Relation relation, LOCKMODE lockmode)
{
LOCKTAG tag;
SET_LOCKTAG_RELATION_EXTEND(tag,
relation->rd_lockInfo.lockRelId.dbId,
relation->rd_lockInfo.lockRelId.relId);
(void) LockAcquire(&tag, lockmode, false, false);
}
/*
* ConditionalLockRelationForExtension
*
* As above, but only lock if we can get the lock without blocking.
* Returns true iff the lock was acquired.
*/
bool
ConditionalLockRelationForExtension(Relation relation, LOCKMODE lockmode)
{
LOCKTAG tag;
SET_LOCKTAG_RELATION_EXTEND(tag,
relation->rd_lockInfo.lockRelId.dbId,
relation->rd_lockInfo.lockRelId.relId);
return (LockAcquire(&tag, lockmode, false, true) != LOCKACQUIRE_NOT_AVAIL);
}
/*
* RelationExtensionLockWaiterCount
*
* Count the number of processes waiting for the given relation extension lock.
*/
int
RelationExtensionLockWaiterCount(Relation relation)
{
LOCKTAG tag;
SET_LOCKTAG_RELATION_EXTEND(tag,
relation->rd_lockInfo.lockRelId.dbId,
relation->rd_lockInfo.lockRelId.relId);
return LockWaiterCount(&tag);
}
/*
* UnlockRelationForExtension
*/
void
UnlockRelationForExtension(Relation relation, LOCKMODE lockmode)
{
LOCKTAG tag;
SET_LOCKTAG_RELATION_EXTEND(tag,
relation->rd_lockInfo.lockRelId.dbId,
relation->rd_lockInfo.lockRelId.relId);
LockRelease(&tag, lockmode, false);
}
/*
* LockPage
*
* Obtain a page-level lock. This is currently used by some index access
* methods to lock individual index pages.
*/
void
LockPage(Relation relation, BlockNumber blkno, LOCKMODE lockmode)
{
LOCKTAG tag;
SET_LOCKTAG_PAGE(tag,
relation->rd_lockInfo.lockRelId.dbId,
relation->rd_lockInfo.lockRelId.relId,
blkno);
(void) LockAcquire(&tag, lockmode, false, false);
}
/*
* ConditionalLockPage
*
* As above, but only lock if we can get the lock without blocking.
* Returns true iff the lock was acquired.
*/
bool
ConditionalLockPage(Relation relation, BlockNumber blkno, LOCKMODE lockmode)
{
LOCKTAG tag;
SET_LOCKTAG_PAGE(tag,
relation->rd_lockInfo.lockRelId.dbId,
relation->rd_lockInfo.lockRelId.relId,
blkno);
return (LockAcquire(&tag, lockmode, false, true) != LOCKACQUIRE_NOT_AVAIL);
}
/*
* UnlockPage
*/
void
UnlockPage(Relation relation, BlockNumber blkno, LOCKMODE lockmode)
{
LOCKTAG tag;
SET_LOCKTAG_PAGE(tag,
relation->rd_lockInfo.lockRelId.dbId,
relation->rd_lockInfo.lockRelId.relId,
blkno);
LockRelease(&tag, lockmode, false);
}
/*
* LockTuple
*
* Obtain a tuple-level lock. This is used in a less-than-intuitive fashion
* because we can't afford to keep a separate lock in shared memory for every
* tuple. See heap_lock_tuple before using this!
*/
void
LockTuple(Relation relation, ItemPointer tid, LOCKMODE lockmode)
{
LOCKTAG tag;
SET_LOCKTAG_TUPLE(tag,
relation->rd_lockInfo.lockRelId.dbId,
relation->rd_lockInfo.lockRelId.relId,
ItemPointerGetBlockNumber(tid),
ItemPointerGetOffsetNumber(tid));
(void) LockAcquire(&tag, lockmode, false, false);
}
/*
* ConditionalLockTuple
*
* As above, but only lock if we can get the lock without blocking.
* Returns true iff the lock was acquired.
*/
bool
ConditionalLockTuple(Relation relation, ItemPointer tid, LOCKMODE lockmode)
{
LOCKTAG tag;
SET_LOCKTAG_TUPLE(tag,
relation->rd_lockInfo.lockRelId.dbId,
relation->rd_lockInfo.lockRelId.relId,
ItemPointerGetBlockNumber(tid),
ItemPointerGetOffsetNumber(tid));
return (LockAcquire(&tag, lockmode, false, true) != LOCKACQUIRE_NOT_AVAIL);
}
/*
* UnlockTuple
*/
void
UnlockTuple(Relation relation, ItemPointer tid, LOCKMODE lockmode)
{
LOCKTAG tag;
SET_LOCKTAG_TUPLE(tag,
relation->rd_lockInfo.lockRelId.dbId,
relation->rd_lockInfo.lockRelId.relId,
ItemPointerGetBlockNumber(tid),
ItemPointerGetOffsetNumber(tid));
LockRelease(&tag, lockmode, false);
}
/*
* XactLockTableInsert
*
* Insert a lock showing that the given transaction ID is running ---
* this is done when an XID is acquired by a transaction or subtransaction.
* The lock can then be used to wait for the transaction to finish.
*/
void
XactLockTableInsert(TransactionId xid)
{
LOCKTAG tag;
SET_LOCKTAG_TRANSACTION(tag, xid);
(void) LockAcquire(&tag, ExclusiveLock, false, false);
}
/*
* XactLockTableDelete
*
* Delete the lock showing that the given transaction ID is running.
* (This is never used for main transaction IDs; those locks are only
* released implicitly at transaction end. But we do use it for subtrans IDs.)
*/
void
XactLockTableDelete(TransactionId xid)
{
LOCKTAG tag;
SET_LOCKTAG_TRANSACTION(tag, xid);
LockRelease(&tag, ExclusiveLock, false);
}
/*
* XactLockTableWait
*
* Wait for the specified transaction to commit or abort. If an operation
* is specified, an error context callback is set up. If 'oper' is passed as
* None, no error context callback is set up.
*
* Note that this does the right thing for subtransactions: if we wait on a
* subtransaction, we will exit as soon as it aborts or its top parent commits.
* It takes some extra work to ensure this, because to save on shared memory
* the XID lock of a subtransaction is released when it ends, whether
* successfully or unsuccessfully. So we have to check if it's "still running"
* and if so wait for its parent.
*/
void
XactLockTableWait(TransactionId xid, Relation rel, ItemPointer ctid,
XLTW_Oper oper)
{
LOCKTAG tag;
XactLockTableWaitInfo info;
ErrorContextCallback callback;
bool first = true;
/*
* Concurrent update and delete will wait on segment when GDD is enabled (or
* the corner case that utility mode delete|update in segment which does not hold
* gxid), need to report the waited transactions to QD to make sure the they have
* the same transaction order on the master.
*/
if (Gp_role == GP_ROLE_EXECUTE)
{
MemoryContext oldContext;
DistributedTransactionId gxid = LocalXidGetDistributedXid(xid);
if (gxid != InvalidDistributedTransactionId)
{
/*
* allocate waitGxids in TopMemoryContext since it's used in
* 'commit prepared' and the TopTransactionContext has been delete
* after 'prepare'
*/
oldContext = MemoryContextSwitchTo(TopMemoryContext);
MyTmGxactLocal->waitGxids = lappend_int(MyTmGxactLocal->waitGxids, gxid);
MemoryContextSwitchTo(oldContext);
}
}
/*
* If an operation is specified, set up our verbose error context
* callback.
*/
if (oper != XLTW_None)
{
Assert(RelationIsValid(rel));
Assert(ItemPointerIsValid(ctid));
info.rel = rel;
info.ctid = ctid;
info.oper = oper;
callback.callback = XactLockTableWaitErrorCb;
callback.arg = &info;
callback.previous = error_context_stack;
error_context_stack = &callback;
}
for (;;)
{
Assert(TransactionIdIsValid(xid));
Assert(!TransactionIdEquals(xid, GetTopTransactionIdIfAny()));
SET_LOCKTAG_TRANSACTION(tag, xid);
(void) LockAcquire(&tag, ShareLock, false, false);
LockRelease(&tag, ShareLock, false);
if (!TransactionIdIsInProgress(xid))
break;
/*
* If the Xid belonged to a subtransaction, then the lock would have
* gone away as soon as it was finished; for correct tuple visibility,
* the right action is to wait on its parent transaction to go away.
* But instead of going levels up one by one, we can just wait for the
* topmost transaction to finish with the same end result, which also
* incurs less locktable traffic.
*
* Some uses of this function don't involve tuple visibility -- such
* as when building snapshots for logical decoding. It is possible to
* see a transaction in ProcArray before it registers itself in the
* locktable. The topmost transaction in that case is the same xid,
* so we try again after a short sleep. (Don't sleep the first time
* through, to avoid slowing down the normal case.)
*/
if (!first)
pg_usleep(1000L);
first = false;
xid = SubTransGetTopmostTransaction(xid);
}
if (oper != XLTW_None)
error_context_stack = callback.previous;
}
/*
* GxactLockTableInsert
*
* CDB: a copy of XactLockTableInsert
* Insert a lock showing that the given distributed transaction ID is running
*/
void
GxactLockTableInsert(DistributedTransactionId gxid)
{
LOCKTAG tag;
Assert(Gp_role == GP_ROLE_DISPATCH);
SET_LOCKTAG_DISTRIB_TRANSACTION(tag, gxid);
(void) LockAcquire(&tag, ExclusiveLock, false, false);
}
/*
* GxactLockTableWait
*
* CDB: a copy of XactLockTableWait, no need to take care of sub-transaction.
* Wait for the specified distributed transaction to commit or abort.
*/
void
GxactLockTableWait(DistributedTransactionId gxid)
{
LOCKTAG tag;
Assert(gxid != InvalidDistributedTransactionId);
Assert(Gp_role == GP_ROLE_DISPATCH);
SET_LOCKTAG_DISTRIB_TRANSACTION(tag, gxid);
(void) LockAcquire(&tag, ShareLock, false, false);
LockRelease(&tag, ShareLock, false);
}
/*
* ConditionalXactLockTableWait
*
* As above, but only lock if we can get the lock without blocking.
* Returns true if the lock was acquired.
*/
bool
ConditionalXactLockTableWait(TransactionId xid)
{
LOCKTAG tag;
bool first = true;
for (;;)
{
Assert(TransactionIdIsValid(xid));
Assert(!TransactionIdEquals(xid, GetTopTransactionIdIfAny()));
SET_LOCKTAG_TRANSACTION(tag, xid);
if (LockAcquire(&tag, ShareLock, false, true) == LOCKACQUIRE_NOT_AVAIL)
return false;
LockRelease(&tag, ShareLock, false);
if (!TransactionIdIsInProgress(xid))
break;
/* See XactLockTableWait about this case */
if (!first)
pg_usleep(1000L);
first = false;
xid = SubTransGetTopmostTransaction(xid);
}
return true;
}
/*
* SpeculativeInsertionLockAcquire
*
* Insert a lock showing that the given transaction ID is inserting a tuple,
* but hasn't yet decided whether it's going to keep it. The lock can then be
* used to wait for the decision to go ahead with the insertion, or aborting
* it.
*
* The token is used to distinguish multiple insertions by the same
* transaction. It is returned to caller.
*/
uint32
SpeculativeInsertionLockAcquire(TransactionId xid)
{
LOCKTAG tag;
speculativeInsertionToken++;
/*
* Check for wrap-around. Zero means no token is held, so don't use that.
*/
if (speculativeInsertionToken == 0)
speculativeInsertionToken = 1;
SET_LOCKTAG_SPECULATIVE_INSERTION(tag, xid, speculativeInsertionToken);
(void) LockAcquire(&tag, ExclusiveLock, false, false);
return speculativeInsertionToken;
}
/*
* SpeculativeInsertionLockRelease
*
* Delete the lock showing that the given transaction is speculatively
* inserting a tuple.
*/
void
SpeculativeInsertionLockRelease(TransactionId xid)
{
LOCKTAG tag;
SET_LOCKTAG_SPECULATIVE_INSERTION(tag, xid, speculativeInsertionToken);
LockRelease(&tag, ExclusiveLock, false);
}
/*
* SpeculativeInsertionWait
*
* Wait for the specified transaction to finish or abort the insertion of a
* tuple.
*/
void
SpeculativeInsertionWait(TransactionId xid, uint32 token)
{
LOCKTAG tag;
SET_LOCKTAG_SPECULATIVE_INSERTION(tag, xid, token);
Assert(TransactionIdIsValid(xid));
Assert(token != 0);
(void) LockAcquire(&tag, ShareLock, false, false);
LockRelease(&tag, ShareLock, false);
}
/*
* XactLockTableWaitErrorContextCb
* Error context callback for transaction lock waits.
*/
static void
XactLockTableWaitErrorCb(void *arg)
{
XactLockTableWaitInfo *info = (XactLockTableWaitInfo *) arg;
/*
* We would like to print schema name too, but that would require a
* syscache lookup.
*/
if (info->oper != XLTW_None &&
ItemPointerIsValid(info->ctid) && RelationIsValid(info->rel))
{
const char *cxt;
switch (info->oper)
{
case XLTW_Update:
cxt = gettext_noop("while updating tuple (%u,%u) in relation \"%s\"");
break;
case XLTW_Delete:
cxt = gettext_noop("while deleting tuple (%u,%u) in relation \"%s\"");
break;
case XLTW_Lock:
cxt = gettext_noop("while locking tuple (%u,%u) in relation \"%s\"");
break;
case XLTW_LockUpdated:
cxt = gettext_noop("while locking updated version (%u,%u) of tuple in relation \"%s\"");
break;
case XLTW_InsertIndex:
cxt = gettext_noop("while inserting index tuple (%u,%u) in relation \"%s\"");
break;
case XLTW_InsertIndexUnique:
cxt = gettext_noop("while checking uniqueness of tuple (%u,%u) in relation \"%s\"");
break;
case XLTW_FetchUpdated:
cxt = gettext_noop("while rechecking updated tuple (%u,%u) in relation \"%s\"");
break;
case XLTW_RecheckExclusionConstr:
cxt = gettext_noop("while checking exclusion constraint on tuple (%u,%u) in relation \"%s\"");
break;
default:
return;
}
errcontext(cxt,
ItemPointerGetBlockNumber(info->ctid),
ItemPointerGetOffsetNumber(info->ctid),
RelationGetRelationName(info->rel));
}
}
/*
* WaitForLockersMultiple
* Wait until no transaction holds locks that conflict with the given
* locktags at the given lockmode.
*
* To do this, obtain the current list of lockers, and wait on their VXIDs
* until they are finished.
*
* Note we don't try to acquire the locks on the given locktags, only the VXIDs
* of its lock holders; if somebody grabs a conflicting lock on the objects
* after we obtained our initial list of lockers, we will not wait for them.
*/
void
WaitForLockersMultiple(List *locktags, LOCKMODE lockmode, bool progress)
{
List *holders = NIL;
ListCell *lc;
int total = 0;
int done = 0;
/* Done if no locks to wait for */
if (list_length(locktags) == 0)
return;
/* Collect the transactions we need to wait on */
foreach(lc, locktags)
{
LOCKTAG *locktag = lfirst(lc);
int count;
holders = lappend(holders,
GetLockConflicts(locktag, lockmode,
progress ? &count : NULL));
if (progress)
total += count;
}
if (progress)
pgstat_progress_update_param(PROGRESS_WAITFOR_TOTAL, total);
/*
* Note: GetLockConflicts() never reports our own xid, hence we need not
* check for that. Also, prepared xacts are not reported, which is fine
* since they certainly aren't going to do anything anymore.
*/
/* Finally wait for each such transaction to complete */
foreach(lc, holders)
{
VirtualTransactionId *lockholders = lfirst(lc);
while (VirtualTransactionIdIsValid(*lockholders))
{
/*
* If requested, publish who we're going to wait for. This is not
* 100% accurate if they're already gone, but we don't care.
*/
if (progress)
{
PGPROC *holder = BackendIdGetProc(lockholders->backendId);
pgstat_progress_update_param(PROGRESS_WAITFOR_CURRENT_PID,
holder->pid);
}
VirtualXactLock(*lockholders, true);
lockholders++;
if (progress)
pgstat_progress_update_param(PROGRESS_WAITFOR_DONE, ++done);
}
}
if (progress)
{
const int index[] = {
PROGRESS_WAITFOR_TOTAL,
PROGRESS_WAITFOR_DONE,
PROGRESS_WAITFOR_CURRENT_PID
};
const int64 values[] = {
0, 0, 0
};
pgstat_progress_update_multi_param(3, index, values);
}
list_free_deep(holders);
}
/*
* WaitForLockers
*
* Same as WaitForLockersMultiple, for a single lock tag.
*/
void
WaitForLockers(LOCKTAG heaplocktag, LOCKMODE lockmode, bool progress)
{
List *l;
l = list_make1(&heaplocktag);
WaitForLockersMultiple(l, lockmode, progress);
list_free(l);
}
/*
* LockDatabaseObject
*
* Obtain a lock on a general object of the current database. Don't use
* this for shared objects (such as tablespaces). It's unwise to apply it
* to relations, also, since a lock taken this way will NOT conflict with
* locks taken via LockRelation and friends.
*/
void
LockDatabaseObject(Oid classid, Oid objid, uint16 objsubid,
LOCKMODE lockmode)
{
LOCKTAG tag;
SET_LOCKTAG_OBJECT(tag,
MyDatabaseId,
classid,
objid,
objsubid);
(void) LockAcquire(&tag, lockmode, false, false);
/* Make sure syscaches are up-to-date with any changes we waited for */
AcceptInvalidationMessages();
}
/*
* UnlockDatabaseObject
*/
void
UnlockDatabaseObject(Oid classid, Oid objid, uint16 objsubid,
LOCKMODE lockmode)
{
LOCKTAG tag;
SET_LOCKTAG_OBJECT(tag,
MyDatabaseId,
classid,
objid,
objsubid);
LockRelease(&tag, lockmode, false);
}
/*
* LockSharedObject
*
* Obtain a lock on a shared-across-databases object.
*/
void
LockSharedObject(Oid classid, Oid objid, uint16 objsubid,
LOCKMODE lockmode)
{
LOCKTAG tag;
SET_LOCKTAG_OBJECT(tag,
InvalidOid,
classid,
objid,
objsubid);
(void) LockAcquire(&tag, lockmode, false, false);
/* Make sure syscaches are up-to-date with any changes we waited for */
AcceptInvalidationMessages();
}
/*
* UnlockSharedObject
*/
void
UnlockSharedObject(Oid classid, Oid objid, uint16 objsubid,
LOCKMODE lockmode)
{
LOCKTAG tag;
SET_LOCKTAG_OBJECT(tag,
InvalidOid,
classid,
objid,
objsubid);
LockRelease(&tag, lockmode, false);
}
/*
* LockSharedObjectForSession
*
* Obtain a session-level lock on a shared-across-databases object.
* See LockRelationIdForSession for notes about session-level locks.
*/
void
LockSharedObjectForSession(Oid classid, Oid objid, uint16 objsubid,
LOCKMODE lockmode)
{
LOCKTAG tag;
SET_LOCKTAG_OBJECT(tag,
InvalidOid,
classid,
objid,
objsubid);
(void) LockAcquire(&tag, lockmode, true, false);
}
/*
* UnlockSharedObjectForSession
*/
void
UnlockSharedObjectForSession(Oid classid, Oid objid, uint16 objsubid,
LOCKMODE lockmode)
{
LOCKTAG tag;
SET_LOCKTAG_OBJECT(tag,
InvalidOid,
classid,
objid,
objsubid);
LockRelease(&tag, lockmode, true);
}
/*
* Append a description of a lockable object to buf.
*
* Ideally we would print names for the numeric values, but that requires
* getting locks on system tables, which might cause problems since this is
* typically used to report deadlock situations.
*/
void
DescribeLockTag(StringInfo buf, const LOCKTAG *tag)
{
switch ((LockTagType) tag->locktag_type)
{
case LOCKTAG_RELATION:
appendStringInfo(buf,
_("relation %u of database %u"),
tag->locktag_field2,
tag->locktag_field1);
break;
case LOCKTAG_RELATION_EXTEND:
appendStringInfo(buf,
_("extension of relation %u of database %u"),
tag->locktag_field2,
tag->locktag_field1);
break;
case LOCKTAG_PAGE:
appendStringInfo(buf,
_("page %u of relation %u of database %u"),
tag->locktag_field3,
tag->locktag_field2,
tag->locktag_field1);
break;
case LOCKTAG_TUPLE:
appendStringInfo(buf,
_("tuple (%u,%u) of relation %u of database %u"),
tag->locktag_field3,
tag->locktag_field4,
tag->locktag_field2,
tag->locktag_field1);
break;
case LOCKTAG_TRANSACTION:
appendStringInfo(buf,
_("transaction %u"),
tag->locktag_field1);
break;
case LOCKTAG_DISTRIB_TRANSACTION:
appendStringInfo(buf,
_("distributed transaction %u"),
tag->locktag_field1);
break;
case LOCKTAG_VIRTUALTRANSACTION:
appendStringInfo(buf,
_("virtual transaction %d/%u"),
tag->locktag_field1,
tag->locktag_field2);
break;
case LOCKTAG_SPECULATIVE_TOKEN:
appendStringInfo(buf,
_("speculative token %u of transaction %u"),
tag->locktag_field2,
tag->locktag_field1);
break;
case LOCKTAG_OBJECT:
appendStringInfo(buf,
_("object %u of class %u of database %u"),
tag->locktag_field3,
tag->locktag_field2,
tag->locktag_field1);
break;
case LOCKTAG_USERLOCK:
/* reserved for old contrib code, now on pgfoundry */
appendStringInfo(buf,
_("user lock [%u,%u,%u]"),
tag->locktag_field1,
tag->locktag_field2,
tag->locktag_field3);
break;
case LOCKTAG_ADVISORY:
appendStringInfo(buf,
_("advisory lock [%u,%u,%u,%u]"),
tag->locktag_field1,
tag->locktag_field2,
tag->locktag_field3,
tag->locktag_field4);
break;
case LOCKTAG_RESOURCE_QUEUE:
appendStringInfo(buf,
_("resource queue %u"),
tag->locktag_field1);
break;
default:
appendStringInfo(buf,
_("unrecognized locktag type %d"),
(int) tag->locktag_type);
break;
}
}
/*
* LockTagIsTemp
* Determine whether a locktag is for a lock on a temporary object
*
* In PostgreSQL, 2PC cannot deal with temporary objects. Commit
* f3032cbe377ecc570989e1bd2fe1aea455c12cc3 replace this method with global
* variable MyXactAccessedTempRel, so it's no longer needed.
*
* However, GPDB supports 2PC with temporary objects. Hence, it still relies
* on this method to check certain LOCKTAG is on temporary object or not, so
* that GPDB can skip the lock information in `TwoPhaseRecordOnDisk` for the
* temporary objects. That's to prevent replay redundant lock acquiring and
* releasing during `xact_redo()` on non-existent temporary objects.
*/
bool
LockTagIsTemp(const LOCKTAG *tag)
{
switch (tag->locktag_type)
{
case LOCKTAG_RELATION:
case LOCKTAG_RELATION_EXTEND:
case LOCKTAG_PAGE:
case LOCKTAG_TUPLE:
/* check for lock on a temp relation */
/* field1 is dboid, field2 is reloid for all of these */
if ((Oid) tag->locktag_field1 == InvalidOid)
return false; /* shared, so not temp */
if (isTempNamespace(get_rel_namespace((Oid) tag->locktag_field2)))
return true;
break;
case LOCKTAG_TRANSACTION:
/* there are no temp transactions */
break;
case LOCKTAG_OBJECT:
/* there are currently no non-table temp objects */
break;
case LOCKTAG_USERLOCK:
case LOCKTAG_ADVISORY:
/* assume these aren't temp */
break;
}
return false; /* default case */
}
/*
* GetLockNameFromTagType
*
* Given locktag type, return the corresponding lock name.
*/
const char *
GetLockNameFromTagType(uint16 locktag_type)
{
if (locktag_type > LOCKTAG_LAST_TYPE)
return "???";
return LockTagTypeNames[locktag_type];
}
相关信息
相关文章
0
赞
热门推荐
-
2、 - 优质文章
-
3、 gate.io
-
7、 golang
-
9、 openharmony
-
10、 Vue中input框自动聚焦