Commit 1785aceb authored by Tom Lane's avatar Tom Lane

Introduce local hash table for lock state, as per recent proposal.

PROCLOCK structs in shared memory now have only a bitmask for held
locks, rather than counts (making them 40 bytes smaller, which is a
good thing).  Multiple locks within a transaction are counted in the
local hash table instead, and we have provision for tracking which
ResourceOwner each count belongs to.  Solves recently reported problem
with memory leakage within long transactions.
parent ef16b4e1
...@@ -75,7 +75,7 @@ user_write_unlock_oid(Oid oid) ...@@ -75,7 +75,7 @@ user_write_unlock_oid(Oid oid)
int int
user_unlock_all(void) user_unlock_all(void)
{ {
return LockReleaseAll(USER_LOCKMETHOD, MyProc, true); return LockReleaseAll(USER_LOCKMETHOD, true);
} }
/* end of file */ /* end of file */
......
$PostgreSQL: pgsql/src/backend/storage/lmgr/README,v 1.14 2003/11/29 19:51:56 pgsql Exp $ $PostgreSQL: pgsql/src/backend/storage/lmgr/README,v 1.15 2004/08/27 17:07:41 tgl Exp $
LOCKING OVERVIEW LOCKING OVERVIEW
...@@ -47,12 +47,6 @@ The rest of this README file discusses the regular lock manager in detail. ...@@ -47,12 +47,6 @@ The rest of this README file discusses the regular lock manager in detail.
LOCK DATA STRUCTURES LOCK DATA STRUCTURES
There are two fundamental lock structures: the per-lockable-object LOCK
struct, and the per-lock PROCLOCK struct. A LOCK object exists
for each lockable object that currently has locks held or requested on it.
A PROCLOCK struct exists for each transaction that is holding or requesting
lock(s) on each LOCK object.
Lock methods describe the overall locking behavior. Currently there are Lock methods describe the overall locking behavior. Currently there are
two lock methods: DEFAULT and USER. (USER locks are non-blocking.) two lock methods: DEFAULT and USER. (USER locks are non-blocking.)
...@@ -60,6 +54,20 @@ Lock modes describe the type of the lock (read/write or shared/exclusive). ...@@ -60,6 +54,20 @@ Lock modes describe the type of the lock (read/write or shared/exclusive).
See src/tools/backend/index.html and src/include/storage/lock.h for more See src/tools/backend/index.html and src/include/storage/lock.h for more
details. details.
There are two fundamental lock structures in shared memory: the
per-lockable-object LOCK struct, and the per-lock-and-requestor PROCLOCK
struct. A LOCK object exists for each lockable object that currently has
locks held or requested on it. A PROCLOCK struct exists for each transaction
that is holding or requesting lock(s) on each LOCK object.
In addition to these, each backend maintains an unshared LOCALLOCK structure
for each lockable object and lock mode that it is currently holding or
requesting. The shared lock structures only allow a single lock grant to
be made per lockable object/lock mode/transaction. Internally to a backend,
however, the same lock may be requested and perhaps released multiple times
in a transaction. The internal request counts are held in LOCALLOCK so that
the shared LockMgrLock need not be obtained to alter them.
--------------------------------------------------------------------------- ---------------------------------------------------------------------------
The lock manager's LOCK objects contain: The lock manager's LOCK objects contain:
...@@ -103,7 +111,7 @@ waitMask - ...@@ -103,7 +111,7 @@ waitMask -
This bitmask shows the types of locks being waited for. Bit i of waitMask This bitmask shows the types of locks being waited for. Bit i of waitMask
is 1 if and only if requested[i] > granted[i]. is 1 if and only if requested[i] > granted[i].
lockHolders - procLocks -
This is a shared memory queue of all the PROCLOCK structs associated with This is a shared memory queue of all the PROCLOCK structs associated with
the lock object. Note that both granted and waiting PROCLOCKs are in this the lock object. Note that both granted and waiting PROCLOCKs are in this
list (indeed, the same PROCLOCK might have some already-granted locks and list (indeed, the same PROCLOCK might have some already-granted locks and
...@@ -120,7 +128,10 @@ nRequested - ...@@ -120,7 +128,10 @@ nRequested -
acquired. The count includes attempts by processes which were put acquired. The count includes attempts by processes which were put
to sleep due to conflicts. It also counts the same backend twice to sleep due to conflicts. It also counts the same backend twice
if, for example, a backend process first acquires a read and then if, for example, a backend process first acquires a read and then
acquires a write, or acquires a read lock twice. acquires a write, or acquires the lock under two different transaction
IDs. (But multiple acquisitions of the same lock/lock mode under the
same transaction ID are not multiply counted here; they are recorded
only in the backend's LOCALLOCK structure.)
requested - requested -
Keeps a count of how many locks of each type have been attempted. Only Keeps a count of how many locks of each type have been attempted. Only
...@@ -130,9 +141,8 @@ requested - ...@@ -130,9 +141,8 @@ requested -
nGranted - nGranted -
Keeps count of how many times this lock has been successfully acquired. Keeps count of how many times this lock has been successfully acquired.
This count does not include attempts that are waiting due to conflicts, This count does not include attempts that are waiting due to conflicts.
but can count the same backend twice (e.g. a read then a write -- since Otherwise the counting rules are the same as for nRequested.
its the same transaction this won't cause a conflict).
granted - granted -
Keeps count of how many locks of each type are currently held. Once again Keeps count of how many locks of each type are currently held. Once again
...@@ -164,18 +174,17 @@ tag - ...@@ -164,18 +174,17 @@ tag -
if the PROCLOCK is for session-level locking. if the PROCLOCK is for session-level locking.
Note that this structure will support multiple transactions running Note that this structure will support multiple transactions running
concurrently in one backend, which may be handy if we someday decide concurrently in one backend. Currently we do not use it for that
to support nested transactions. Currently, the XID field is only needed purpose: subtransactions acquire locks in the name of their top parent
to distinguish per-transaction locks from session locks. User locks transaction, to simplify reassigning lock ownership at subtransaction end.
are always session locks, and we also use session locks for multi- So the XID field is really only needed to distinguish per-transaction
transaction operations like VACUUM. locks from session locks. User locks are always session locks, and we
also use session locks for multi-transaction operations like VACUUM.
holding -
The number of successfully acquired locks of each type for this PROCLOCK. holdMask -
This should be <= the corresponding granted[] value of the lock object! A bitmask for the lock types successfully acquired by this PROCLOCK.
This should be a subset of the LOCK object's grantMask, and also a
nHolding - subset of the PGPROC object's heldLocks mask.
Sum of the holding[] array.
lockLink - lockLink -
List link for shared memory queue of all the PROCLOCK objects for the List link for shared memory queue of all the PROCLOCK objects for the
......
...@@ -12,7 +12,7 @@ ...@@ -12,7 +12,7 @@
* *
* *
* IDENTIFICATION * IDENTIFICATION
* $PostgreSQL: pgsql/src/backend/storage/lmgr/deadlock.c,v 1.29 2004/07/21 22:31:22 tgl Exp $ * $PostgreSQL: pgsql/src/backend/storage/lmgr/deadlock.c,v 1.30 2004/08/27 17:07:41 tgl Exp $
* *
* Interface: * Interface:
* *
...@@ -427,7 +427,7 @@ FindLockCycleRecurse(PGPROC *checkProc, ...@@ -427,7 +427,7 @@ FindLockCycleRecurse(PGPROC *checkProc,
PGPROC *proc; PGPROC *proc;
LOCK *lock; LOCK *lock;
PROCLOCK *proclock; PROCLOCK *proclock;
SHM_QUEUE *lockHolders; SHM_QUEUE *procLocks;
LockMethod lockMethodTable; LockMethod lockMethodTable;
PROC_QUEUE *waitQueue; PROC_QUEUE *waitQueue;
int queue_size; int queue_size;
...@@ -483,9 +483,9 @@ FindLockCycleRecurse(PGPROC *checkProc, ...@@ -483,9 +483,9 @@ FindLockCycleRecurse(PGPROC *checkProc,
* Scan for procs that already hold conflicting locks. These are * Scan for procs that already hold conflicting locks. These are
* "hard" edges in the waits-for graph. * "hard" edges in the waits-for graph.
*/ */
lockHolders = &(lock->lockHolders); procLocks = &(lock->procLocks);
proclock = (PROCLOCK *) SHMQueueNext(lockHolders, lockHolders, proclock = (PROCLOCK *) SHMQueueNext(procLocks, procLocks,
offsetof(PROCLOCK, lockLink)); offsetof(PROCLOCK, lockLink));
while (proclock) while (proclock)
...@@ -497,8 +497,8 @@ FindLockCycleRecurse(PGPROC *checkProc, ...@@ -497,8 +497,8 @@ FindLockCycleRecurse(PGPROC *checkProc,
{ {
for (lm = 1; lm <= numLockModes; lm++) for (lm = 1; lm <= numLockModes; lm++)
{ {
if (proclock->holding[lm] > 0 && if ((proclock->holdMask & LOCKBIT_ON(lm)) &&
((1 << lm) & conflictMask) != 0) (conflictMask & LOCKBIT_ON(lm)))
{ {
/* This proc hard-blocks checkProc */ /* This proc hard-blocks checkProc */
if (FindLockCycleRecurse(proc, depth + 1, if (FindLockCycleRecurse(proc, depth + 1,
...@@ -519,7 +519,7 @@ FindLockCycleRecurse(PGPROC *checkProc, ...@@ -519,7 +519,7 @@ FindLockCycleRecurse(PGPROC *checkProc,
} }
} }
proclock = (PROCLOCK *) SHMQueueNext(lockHolders, &proclock->lockLink, proclock = (PROCLOCK *) SHMQueueNext(procLocks, &proclock->lockLink,
offsetof(PROCLOCK, lockLink)); offsetof(PROCLOCK, lockLink));
} }
......
...@@ -8,7 +8,7 @@ ...@@ -8,7 +8,7 @@
* *
* *
* IDENTIFICATION * IDENTIFICATION
* $PostgreSQL: pgsql/src/backend/storage/lmgr/lock.c,v 1.136 2004/08/26 17:23:30 tgl Exp $ * $PostgreSQL: pgsql/src/backend/storage/lmgr/lock.c,v 1.137 2004/08/27 17:07:41 tgl Exp $
* *
* NOTES * NOTES
* Outside modules can create a lock table and acquire/release * Outside modules can create a lock table and acquire/release
...@@ -47,12 +47,24 @@ int max_locks_per_xact; /* set by guc.c */ ...@@ -47,12 +47,24 @@ int max_locks_per_xact; /* set by guc.c */
#define NLOCKENTS(maxBackends) (max_locks_per_xact * (maxBackends)) #define NLOCKENTS(maxBackends) (max_locks_per_xact * (maxBackends))
static int WaitOnLock(LOCKMETHODID lockmethodid, LOCKMODE lockmode, /*
LOCK *lock, PROCLOCK *proclock); * map from lock method id to the lock table data structures
static void LockCountMyLocks(SHMEM_OFFSET lockOffset, PGPROC *proc, */
int *myHolding); static LockMethod LockMethods[MAX_LOCK_METHODS];
static HTAB *LockMethodLockHash[MAX_LOCK_METHODS];
static HTAB *LockMethodProcLockHash[MAX_LOCK_METHODS];
static HTAB *LockMethodLocalHash[MAX_LOCK_METHODS];
/* exported so lmgr.c can initialize it */
int NumLockMethods;
static char *lock_mode_names[] = /* private state for GrantAwaitedLock */
static LOCALLOCK *awaitedLock;
static ResourceOwner awaitedOwner;
static const char * const lock_mode_names[] =
{ {
"INVALID", "INVALID",
"AccessShareLock", "AccessShareLock",
...@@ -134,31 +146,27 @@ PROCLOCK_PRINT(const char *where, const PROCLOCK *proclockP) ...@@ -134,31 +146,27 @@ PROCLOCK_PRINT(const char *where, const PROCLOCK *proclockP)
|| (Trace_lock_table && (((LOCK *) MAKE_PTR(proclockP->tag.lock))->tag.relId == Trace_lock_table)) || (Trace_lock_table && (((LOCK *) MAKE_PTR(proclockP->tag.lock))->tag.relId == Trace_lock_table))
) )
elog(LOG, elog(LOG,
"%s: proclock(%lx) lock(%lx) tbl(%d) proc(%lx) xid(%u) hold(%d,%d,%d,%d,%d,%d,%d)=%d", "%s: proclock(%lx) lock(%lx) tbl(%d) proc(%lx) xid(%u) hold(%x)",
where, MAKE_OFFSET(proclockP), proclockP->tag.lock, where, MAKE_OFFSET(proclockP), proclockP->tag.lock,
PROCLOCK_LOCKMETHOD(*(proclockP)), PROCLOCK_LOCKMETHOD(*(proclockP)),
proclockP->tag.proc, proclockP->tag.xid, proclockP->tag.proc, proclockP->tag.xid,
proclockP->holding[1], proclockP->holding[2], proclockP->holding[3], (int) proclockP->holdMask);
proclockP->holding[4], proclockP->holding[5], proclockP->holding[6],
proclockP->holding[7], proclockP->nHolding);
} }
#else /* not LOCK_DEBUG */ #else /* not LOCK_DEBUG */
#define LOCK_PRINT(where, lock, type) #define LOCK_PRINT(where, lock, type)
#define PROCLOCK_PRINT(where, proclockP) #define PROCLOCK_PRINT(where, proclockP)
#endif /* not LOCK_DEBUG */ #endif /* not LOCK_DEBUG */
/* static void RemoveLocalLock(LOCALLOCK *locallock);
* map from lock method id to the lock table structure static void GrantLockLocal(LOCALLOCK *locallock, ResourceOwner owner);
*/ static int WaitOnLock(LOCKMETHODID lockmethodid, LOCALLOCK *locallock,
static LockMethod LockMethods[MAX_LOCK_METHODS]; ResourceOwner owner);
static HTAB* LockMethodLockHash[MAX_LOCK_METHODS]; static void LockCountMyLocks(SHMEM_OFFSET lockOffset, PGPROC *proc,
static HTAB* LockMethodProcLockHash[MAX_LOCK_METHODS]; int *myHolding);
/* exported so lmgr.c can initialize it */
int NumLockMethods;
/* /*
...@@ -220,6 +228,7 @@ LockMethodTableInit(const char *tabName, ...@@ -220,6 +228,7 @@ LockMethodTableInit(const char *tabName,
int maxBackends) int maxBackends)
{ {
LockMethod newLockMethod; LockMethod newLockMethod;
LOCKMETHODID lockmethodid;
char *shmemName; char *shmemName;
HASHCTL info; HASHCTL info;
int hash_flags; int hash_flags;
...@@ -254,40 +263,39 @@ LockMethodTableInit(const char *tabName, ...@@ -254,40 +263,39 @@ LockMethodTableInit(const char *tabName,
{ {
MemSet(newLockMethod, 0, sizeof(LockMethodData)); MemSet(newLockMethod, 0, sizeof(LockMethodData));
newLockMethod->masterLock = LockMgrLock; newLockMethod->masterLock = LockMgrLock;
newLockMethod->lockmethodid = NumLockMethods;
LockMethodInit(newLockMethod, conflictsP, numModes); LockMethodInit(newLockMethod, conflictsP, numModes);
} }
/* /*
* other modules refer to the lock table by a lockmethod ID * other modules refer to the lock table by a lockmethod ID
*/ */
LockMethods[NumLockMethods] = newLockMethod; Assert(NumLockMethods < MAX_LOCK_METHODS);
NumLockMethods++; lockmethodid = NumLockMethods++;
Assert(NumLockMethods <= MAX_LOCK_METHODS); LockMethods[lockmethodid] = newLockMethod;
/* /*
* allocate a hash table for LOCK structs. This is used to store * allocate a hash table for LOCK structs. This is used to store
* per-locked-object information. * per-locked-object information.
*/ */
MemSet(&info, 0, sizeof(info));
info.keysize = sizeof(LOCKTAG); info.keysize = sizeof(LOCKTAG);
info.entrysize = sizeof(LOCK); info.entrysize = sizeof(LOCK);
info.hash = tag_hash; info.hash = tag_hash;
hash_flags = (HASH_ELEM | HASH_FUNCTION); hash_flags = (HASH_ELEM | HASH_FUNCTION);
sprintf(shmemName, "%s (lock hash)", tabName); sprintf(shmemName, "%s (lock hash)", tabName);
LockMethodLockHash[NumLockMethods-1] = ShmemInitHash(shmemName, LockMethodLockHash[lockmethodid] = ShmemInitHash(shmemName,
init_table_size, init_table_size,
max_table_size, max_table_size,
&info, &info,
hash_flags); hash_flags);
if (!LockMethodLockHash[NumLockMethods-1]) if (!LockMethodLockHash[lockmethodid])
elog(FATAL, "could not initialize lock table \"%s\"", tabName); elog(FATAL, "could not initialize lock table \"%s\"", tabName);
Assert(LockMethodLockHash[NumLockMethods-1]->hash == tag_hash);
/* /*
* allocate a hash table for PROCLOCK structs. This is used to store * allocate a hash table for PROCLOCK structs. This is used to store
* per-lock-proclock information. * per-lock-holder information.
*/ */
info.keysize = sizeof(PROCLOCKTAG); info.keysize = sizeof(PROCLOCKTAG);
info.entrysize = sizeof(PROCLOCK); info.entrysize = sizeof(PROCLOCK);
...@@ -295,18 +303,44 @@ LockMethodTableInit(const char *tabName, ...@@ -295,18 +303,44 @@ LockMethodTableInit(const char *tabName,
hash_flags = (HASH_ELEM | HASH_FUNCTION); hash_flags = (HASH_ELEM | HASH_FUNCTION);
sprintf(shmemName, "%s (proclock hash)", tabName); sprintf(shmemName, "%s (proclock hash)", tabName);
LockMethodProcLockHash[NumLockMethods-1] = ShmemInitHash(shmemName, LockMethodProcLockHash[lockmethodid] = ShmemInitHash(shmemName,
init_table_size, init_table_size,
max_table_size, max_table_size,
&info, &info,
hash_flags); hash_flags);
if (!LockMethodProcLockHash[NumLockMethods-1]) if (!LockMethodProcLockHash[lockmethodid])
elog(FATAL, "could not initialize lock table \"%s\"", tabName);
/*
* allocate a non-shared hash table for LOCALLOCK structs. This is used
* to store lock counts and resource owner information.
*
* The non-shared table could already exist in this process (this occurs
* when the postmaster is recreating shared memory after a backend crash).
* If so, delete and recreate it. (We could simply leave it, since it
* ought to be empty in the postmaster, but for safety let's zap it.)
*/
if (LockMethodLocalHash[lockmethodid])
hash_destroy(LockMethodLocalHash[lockmethodid]);
info.keysize = sizeof(LOCALLOCKTAG);
info.entrysize = sizeof(LOCALLOCK);
info.hash = tag_hash;
hash_flags = (HASH_ELEM | HASH_FUNCTION);
sprintf(shmemName, "%s (locallock hash)", tabName);
LockMethodLocalHash[lockmethodid] = hash_create(shmemName,
init_table_size,
&info,
hash_flags);
if (!LockMethodLocalHash[lockmethodid])
elog(FATAL, "could not initialize lock table \"%s\"", tabName); elog(FATAL, "could not initialize lock table \"%s\"", tabName);
pfree(shmemName); pfree(shmemName);
return newLockMethod->lockmethodid; return lockmethodid;
} }
/* /*
...@@ -339,6 +373,7 @@ LockMethodTableRename(LOCKMETHODID lockmethodid) ...@@ -339,6 +373,7 @@ LockMethodTableRename(LOCKMETHODID lockmethodid)
LockMethods[newLockMethodId] = LockMethods[lockmethodid]; LockMethods[newLockMethodId] = LockMethods[lockmethodid];
LockMethodLockHash[newLockMethodId] = LockMethodLockHash[lockmethodid]; LockMethodLockHash[newLockMethodId] = LockMethodLockHash[lockmethodid];
LockMethodProcLockHash[newLockMethodId] = LockMethodProcLockHash[lockmethodid]; LockMethodProcLockHash[newLockMethodId] = LockMethodProcLockHash[lockmethodid];
LockMethodLocalHash[newLockMethodId] = LockMethodLocalHash[lockmethodid];
return newLockMethodId; return newLockMethodId;
} }
...@@ -408,11 +443,13 @@ bool ...@@ -408,11 +443,13 @@ bool
LockAcquire(LOCKMETHODID lockmethodid, LOCKTAG *locktag, LockAcquire(LOCKMETHODID lockmethodid, LOCKTAG *locktag,
TransactionId xid, LOCKMODE lockmode, bool dontWait) TransactionId xid, LOCKMODE lockmode, bool dontWait)
{ {
LOCALLOCKTAG localtag;
LOCALLOCK *locallock;
LOCK *lock;
PROCLOCK *proclock; PROCLOCK *proclock;
PROCLOCKTAG proclocktag; PROCLOCKTAG proclocktag;
HTAB *proclockTable;
bool found; bool found;
LOCK *lock; ResourceOwner owner;
LWLockId masterLock; LWLockId masterLock;
LockMethod lockMethodTable; LockMethod lockMethodTable;
int status; int status;
...@@ -428,9 +465,6 @@ LockAcquire(LOCKMETHODID lockmethodid, LOCKTAG *locktag, ...@@ -428,9 +465,6 @@ LockAcquire(LOCKMETHODID lockmethodid, LOCKTAG *locktag,
/* ???????? This must be changed when short term locks will be used */ /* ???????? This must be changed when short term locks will be used */
locktag->lockmethodid = lockmethodid; locktag->lockmethodid = lockmethodid;
/* Prepare to record the lock in the current resource owner */
ResourceOwnerEnlargeLocks(CurrentResourceOwner);
Assert(lockmethodid < NumLockMethods); Assert(lockmethodid < NumLockMethods);
lockMethodTable = LockMethods[lockmethodid]; lockMethodTable = LockMethods[lockmethodid];
if (!lockMethodTable) if (!lockMethodTable)
...@@ -439,14 +473,82 @@ LockAcquire(LOCKMETHODID lockmethodid, LOCKTAG *locktag, ...@@ -439,14 +473,82 @@ LockAcquire(LOCKMETHODID lockmethodid, LOCKTAG *locktag,
return FALSE; return FALSE;
} }
/* Session locks and user locks are not transactional */
if (xid != InvalidTransactionId &&
lockmethodid == DEFAULT_LOCKMETHOD)
owner = CurrentResourceOwner;
else
owner = NULL;
/*
* Find or create a LOCALLOCK entry for this lock and lockmode
*/
MemSet(&localtag, 0, sizeof(localtag)); /* must clear padding */
localtag.lock = *locktag;
localtag.xid = xid;
localtag.mode = lockmode;
locallock = (LOCALLOCK *) hash_search(LockMethodLocalHash[lockmethodid],
(void *) &localtag,
HASH_ENTER, &found);
if (!locallock)
ereport(ERROR,
(errcode(ERRCODE_OUT_OF_MEMORY),
errmsg("out of memory")));
/*
* if it's a new locallock object, initialize it
*/
if (!found)
{
locallock->lock = NULL;
locallock->proclock = NULL;
locallock->nLocks = 0;
locallock->numLockOwners = 0;
locallock->maxLockOwners = 8;
locallock->lockOwners = NULL;
locallock->lockOwners = (LOCALLOCKOWNER *)
MemoryContextAlloc(TopMemoryContext,
locallock->maxLockOwners * sizeof(LOCALLOCKOWNER));
}
else
{
/* Make sure there will be room to remember the lock */
if (locallock->numLockOwners >= locallock->maxLockOwners)
{
int newsize = locallock->maxLockOwners * 2;
locallock->lockOwners = (LOCALLOCKOWNER *)
repalloc(locallock->lockOwners,
newsize * sizeof(LOCALLOCKOWNER));
locallock->maxLockOwners = newsize;
}
}
/*
* If we already hold the lock, we can just increase the count locally.
*/
if (locallock->nLocks > 0)
{
GrantLockLocal(locallock, owner);
return TRUE;
}
/*
* Otherwise we've got to mess with the shared lock table.
*/
masterLock = lockMethodTable->masterLock; masterLock = lockMethodTable->masterLock;
LWLockAcquire(masterLock, LW_EXCLUSIVE); LWLockAcquire(masterLock, LW_EXCLUSIVE);
/* /*
* Find or create a lock with this tag * Find or create a lock with this tag.
*
* Note: if the locallock object already existed, it might have a pointer
* to the lock already ... but we probably should not assume that that
* pointer is valid, since a lock object with no locks can go away
* anytime.
*/ */
Assert(LockMethodLockHash[lockmethodid]->hash == tag_hash);
lock = (LOCK *) hash_search(LockMethodLockHash[lockmethodid], lock = (LOCK *) hash_search(LockMethodLockHash[lockmethodid],
(void *) locktag, (void *) locktag,
HASH_ENTER, &found); HASH_ENTER, &found);
...@@ -458,6 +560,7 @@ LockAcquire(LOCKMETHODID lockmethodid, LOCKTAG *locktag, ...@@ -458,6 +560,7 @@ LockAcquire(LOCKMETHODID lockmethodid, LOCKTAG *locktag,
errmsg("out of shared memory"), errmsg("out of shared memory"),
errhint("You may need to increase max_locks_per_transaction."))); errhint("You may need to increase max_locks_per_transaction.")));
} }
locallock->lock = lock;
/* /*
* if it's a new lock object, initialize it * if it's a new lock object, initialize it
...@@ -466,7 +569,7 @@ LockAcquire(LOCKMETHODID lockmethodid, LOCKTAG *locktag, ...@@ -466,7 +569,7 @@ LockAcquire(LOCKMETHODID lockmethodid, LOCKTAG *locktag,
{ {
lock->grantMask = 0; lock->grantMask = 0;
lock->waitMask = 0; lock->waitMask = 0;
SHMQueueInit(&(lock->lockHolders)); SHMQueueInit(&(lock->procLocks));
ProcQueueInit(&(lock->waitProcs)); ProcQueueInit(&(lock->waitProcs));
lock->nRequested = 0; lock->nRequested = 0;
lock->nGranted = 0; lock->nGranted = 0;
...@@ -485,8 +588,7 @@ LockAcquire(LOCKMETHODID lockmethodid, LOCKTAG *locktag, ...@@ -485,8 +588,7 @@ LockAcquire(LOCKMETHODID lockmethodid, LOCKTAG *locktag,
/* /*
* Create the hash key for the proclock table. * Create the hash key for the proclock table.
*/ */
MemSet(&proclocktag, 0, sizeof(PROCLOCKTAG)); /* must clear padding, MemSet(&proclocktag, 0, sizeof(PROCLOCKTAG)); /* must clear padding */
* needed */
proclocktag.lock = MAKE_OFFSET(lock); proclocktag.lock = MAKE_OFFSET(lock);
proclocktag.proc = MAKE_OFFSET(MyProc); proclocktag.proc = MAKE_OFFSET(MyProc);
TransactionIdStore(xid, &proclocktag.xid); TransactionIdStore(xid, &proclocktag.xid);
...@@ -494,8 +596,7 @@ LockAcquire(LOCKMETHODID lockmethodid, LOCKTAG *locktag, ...@@ -494,8 +596,7 @@ LockAcquire(LOCKMETHODID lockmethodid, LOCKTAG *locktag,
/* /*
* Find or create a proclock entry with this tag * Find or create a proclock entry with this tag
*/ */
proclockTable = LockMethodProcLockHash[lockmethodid]; proclock = (PROCLOCK *) hash_search(LockMethodProcLockHash[lockmethodid],
proclock = (PROCLOCK *) hash_search(proclockTable,
(void *) &proclocktag, (void *) &proclocktag,
HASH_ENTER, &found); HASH_ENTER, &found);
if (!proclock) if (!proclock)
...@@ -506,24 +607,23 @@ LockAcquire(LOCKMETHODID lockmethodid, LOCKTAG *locktag, ...@@ -506,24 +607,23 @@ LockAcquire(LOCKMETHODID lockmethodid, LOCKTAG *locktag,
errmsg("out of shared memory"), errmsg("out of shared memory"),
errhint("You may need to increase max_locks_per_transaction."))); errhint("You may need to increase max_locks_per_transaction.")));
} }
locallock->proclock = proclock;
/* /*
* If new, initialize the new entry * If new, initialize the new entry
*/ */
if (!found) if (!found)
{ {
proclock->nHolding = 0; proclock->holdMask = 0;
MemSet((char *) proclock->holding, 0, sizeof(int) * MAX_LOCKMODES);
/* Add proclock to appropriate lists */ /* Add proclock to appropriate lists */
SHMQueueInsertBefore(&lock->lockHolders, &proclock->lockLink); SHMQueueInsertBefore(&lock->procLocks, &proclock->lockLink);
SHMQueueInsertBefore(&MyProc->procHolders, &proclock->procLink); SHMQueueInsertBefore(&MyProc->procLocks, &proclock->procLink);
PROCLOCK_PRINT("LockAcquire: new", proclock); PROCLOCK_PRINT("LockAcquire: new", proclock);
} }
else else
{ {
PROCLOCK_PRINT("LockAcquire: found", proclock); PROCLOCK_PRINT("LockAcquire: found", proclock);
Assert((proclock->nHolding >= 0) && (proclock->holding[lockmode] >= 0)); Assert((proclock->holdMask & ~lock->grantMask) == 0);
Assert(proclock->nHolding <= lock->nGranted);
#ifdef CHECK_DEADLOCK_RISK #ifdef CHECK_DEADLOCK_RISK
...@@ -544,7 +644,7 @@ LockAcquire(LOCKMETHODID lockmethodid, LOCKTAG *locktag, ...@@ -544,7 +644,7 @@ LockAcquire(LOCKMETHODID lockmethodid, LOCKTAG *locktag,
*/ */
for (i = lockMethodTable->numLockModes; i > 0; i--) for (i = lockMethodTable->numLockModes; i > 0; i--)
{ {
if (proclock->holding[i] > 0) if (proclock->holdMask & LOCKBIT_ON(i))
{ {
if (i >= (int) lockmode) if (i >= (int) lockmode)
break; /* safe: we have a lock >= req level */ break; /* safe: we have a lock >= req level */
...@@ -568,29 +668,14 @@ LockAcquire(LOCKMETHODID lockmethodid, LOCKTAG *locktag, ...@@ -568,29 +668,14 @@ LockAcquire(LOCKMETHODID lockmethodid, LOCKTAG *locktag,
Assert((lock->nRequested > 0) && (lock->requested[lockmode] > 0)); Assert((lock->nRequested > 0) && (lock->requested[lockmode] > 0));
/* /*
* If I already hold one or more locks of the requested type, just * If this process (under any XID) is a holder of the lock, just
* grant myself another one without blocking.
*/
if (proclock->holding[lockmode] > 0)
{
GrantLock(lock, proclock, lockmode);
ResourceOwnerRememberLock(CurrentResourceOwner, locktag, xid,
lockmode);
PROCLOCK_PRINT("LockAcquire: owning", proclock);
LWLockRelease(masterLock);
return TRUE;
}
/*
* If this process (under any XID) is a proclock of the lock, also
* grant myself another one without blocking. * grant myself another one without blocking.
*/ */
LockCountMyLocks(proclock->tag.lock, MyProc, myHolding); LockCountMyLocks(proclock->tag.lock, MyProc, myHolding);
if (myHolding[lockmode] > 0) if (myHolding[lockmode] > 0)
{ {
GrantLock(lock, proclock, lockmode); GrantLock(lock, proclock, lockmode);
ResourceOwnerRememberLock(CurrentResourceOwner, locktag, xid, GrantLockLocal(locallock, owner);
lockmode);
PROCLOCK_PRINT("LockAcquire: my other XID owning", proclock); PROCLOCK_PRINT("LockAcquire: my other XID owning", proclock);
LWLockRelease(masterLock); LWLockRelease(masterLock);
return TRUE; return TRUE;
...@@ -612,8 +697,7 @@ LockAcquire(LOCKMETHODID lockmethodid, LOCKTAG *locktag, ...@@ -612,8 +697,7 @@ LockAcquire(LOCKMETHODID lockmethodid, LOCKTAG *locktag,
{ {
/* No conflict with held or previously requested locks */ /* No conflict with held or previously requested locks */
GrantLock(lock, proclock, lockmode); GrantLock(lock, proclock, lockmode);
ResourceOwnerRememberLock(CurrentResourceOwner, locktag, xid, GrantLockLocal(locallock, owner);
lockmode);
} }
else else
{ {
...@@ -621,29 +705,31 @@ LockAcquire(LOCKMETHODID lockmethodid, LOCKTAG *locktag, ...@@ -621,29 +705,31 @@ LockAcquire(LOCKMETHODID lockmethodid, LOCKTAG *locktag,
/* /*
* We can't acquire the lock immediately. If caller specified no * We can't acquire the lock immediately. If caller specified no
* blocking, remove the proclock entry and return FALSE without * blocking, remove useless table entries and return FALSE without
* waiting. * waiting.
*/ */
if (dontWait) if (dontWait)
{ {
if (proclock->nHolding == 0) if (proclock->holdMask == 0)
{ {
SHMQueueDelete(&proclock->lockLink); SHMQueueDelete(&proclock->lockLink);
SHMQueueDelete(&proclock->procLink); SHMQueueDelete(&proclock->procLink);
proclock = (PROCLOCK *) hash_search(proclockTable, proclock = (PROCLOCK *) hash_search(LockMethodProcLockHash[lockmethodid],
(void *) proclock, (void *) &(proclock->tag),
HASH_REMOVE, NULL); HASH_REMOVE, NULL);
if (!proclock) if (!proclock)
elog(WARNING, "proclock table corrupted"); elog(WARNING, "proclock table corrupted");
} }
else else
PROCLOCK_PRINT("LockAcquire: NHOLDING", proclock); PROCLOCK_PRINT("LockAcquire: NOWAIT", proclock);
lock->nRequested--; lock->nRequested--;
lock->requested[lockmode]--; lock->requested[lockmode]--;
LOCK_PRINT("LockAcquire: conditional lock failed", lock, lockmode); LOCK_PRINT("LockAcquire: conditional lock failed", lock, lockmode);
Assert((lock->nRequested > 0) && (lock->requested[lockmode] >= 0)); Assert((lock->nRequested > 0) && (lock->requested[lockmode] >= 0));
Assert(lock->nGranted <= lock->nRequested); Assert(lock->nGranted <= lock->nRequested);
LWLockRelease(masterLock); LWLockRelease(masterLock);
if (locallock->nLocks == 0)
RemoveLocalLock(locallock);
return FALSE; return FALSE;
} }
...@@ -664,7 +750,7 @@ LockAcquire(LOCKMETHODID lockmethodid, LOCKTAG *locktag, ...@@ -664,7 +750,7 @@ LockAcquire(LOCKMETHODID lockmethodid, LOCKTAG *locktag,
/* /*
* Sleep till someone wakes me up. * Sleep till someone wakes me up.
*/ */
status = WaitOnLock(lockmethodid, lockmode, lock, proclock); status = WaitOnLock(lockmethodid, locallock, owner);
/* /*
* NOTE: do not do any material change of state between here and * NOTE: do not do any material change of state between here and
...@@ -677,7 +763,7 @@ LockAcquire(LOCKMETHODID lockmethodid, LOCKTAG *locktag, ...@@ -677,7 +763,7 @@ LockAcquire(LOCKMETHODID lockmethodid, LOCKTAG *locktag,
* Check the proclock entry status, in case something in the ipc * Check the proclock entry status, in case something in the ipc
* communication doesn't work correctly. * communication doesn't work correctly.
*/ */
if (!((proclock->nHolding > 0) && (proclock->holding[lockmode] > 0))) if (!(proclock->holdMask & LOCKBIT_ON(lockmode)))
{ {
PROCLOCK_PRINT("LockAcquire: INCONSISTENT", proclock); PROCLOCK_PRINT("LockAcquire: INCONSISTENT", proclock);
LOCK_PRINT("LockAcquire: INCONSISTENT", lock, lockmode); LOCK_PRINT("LockAcquire: INCONSISTENT", lock, lockmode);
...@@ -694,6 +780,23 @@ LockAcquire(LOCKMETHODID lockmethodid, LOCKTAG *locktag, ...@@ -694,6 +780,23 @@ LockAcquire(LOCKMETHODID lockmethodid, LOCKTAG *locktag,
return status == STATUS_OK; return status == STATUS_OK;
} }
/*
* Subroutine to free a locallock entry
*/
static void
RemoveLocalLock(LOCALLOCK *locallock)
{
LOCKMETHODID lockmethodid = LOCALLOCK_LOCKMETHOD(*locallock);
pfree(locallock->lockOwners);
locallock->lockOwners = NULL;
locallock = (LOCALLOCK *) hash_search(LockMethodLocalHash[lockmethodid],
(void *) &(locallock->tag),
HASH_REMOVE, NULL);
if (!locallock)
elog(WARNING, "locallock table corrupted");
}
/* /*
* LockCheckConflicts -- test whether requested lock conflicts * LockCheckConflicts -- test whether requested lock conflicts
* with those already granted * with those already granted
...@@ -788,24 +891,29 @@ LockCheckConflicts(LockMethod lockMethodTable, ...@@ -788,24 +891,29 @@ LockCheckConflicts(LockMethod lockMethodTable,
static void static void
LockCountMyLocks(SHMEM_OFFSET lockOffset, PGPROC *proc, int *myHolding) LockCountMyLocks(SHMEM_OFFSET lockOffset, PGPROC *proc, int *myHolding)
{ {
SHM_QUEUE *procHolders = &(proc->procHolders); SHM_QUEUE *procLocks = &(proc->procLocks);
PROCLOCK *proclock; PROCLOCK *proclock;
int i;
MemSet(myHolding, 0, MAX_LOCKMODES * sizeof(int)); MemSet(myHolding, 0, MAX_LOCKMODES * sizeof(int));
proclock = (PROCLOCK *) SHMQueueNext(procHolders, procHolders, proclock = (PROCLOCK *) SHMQueueNext(procLocks, procLocks,
offsetof(PROCLOCK, procLink)); offsetof(PROCLOCK, procLink));
while (proclock) while (proclock)
{ {
if (lockOffset == proclock->tag.lock) if (lockOffset == proclock->tag.lock)
{ {
LOCKMASK holdMask = proclock->holdMask;
int i;
for (i = 1; i < MAX_LOCKMODES; i++) for (i = 1; i < MAX_LOCKMODES; i++)
myHolding[i] += proclock->holding[i]; {
if (holdMask & LOCKBIT_ON(i))
myHolding[i]++;
}
} }
proclock = (PROCLOCK *) SHMQueueNext(procHolders, &proclock->procLink, proclock = (PROCLOCK *) SHMQueueNext(procLocks, &proclock->procLink,
offsetof(PROCLOCK, procLink)); offsetof(PROCLOCK, procLink));
} }
} }
...@@ -815,10 +923,11 @@ LockCountMyLocks(SHMEM_OFFSET lockOffset, PGPROC *proc, int *myHolding) ...@@ -815,10 +923,11 @@ LockCountMyLocks(SHMEM_OFFSET lockOffset, PGPROC *proc, int *myHolding)
* the lock request has been granted. * the lock request has been granted.
* *
* NOTE: if proc was blocked, it also needs to be removed from the wait list * NOTE: if proc was blocked, it also needs to be removed from the wait list
* and have its waitLock/waitHolder fields cleared. That's not done here. * and have its waitLock/waitProcLock fields cleared. That's not done here.
* *
* NOTE: the lock also has to be recorded in the current ResourceOwner; * NOTE: the lock grant also has to be recorded in the associated LOCALLOCK
* but since we may be awaking some other process, we can't do that here. * table entry; but since we may be awaking some other process, we can't do
* that here; it's done by GrantLockLocal, instead.
*/ */
void void
GrantLock(LOCK *lock, PROCLOCK *proclock, LOCKMODE lockmode) GrantLock(LOCK *lock, PROCLOCK *proclock, LOCKMODE lockmode)
...@@ -828,12 +937,56 @@ GrantLock(LOCK *lock, PROCLOCK *proclock, LOCKMODE lockmode) ...@@ -828,12 +937,56 @@ GrantLock(LOCK *lock, PROCLOCK *proclock, LOCKMODE lockmode)
lock->grantMask |= LOCKBIT_ON(lockmode); lock->grantMask |= LOCKBIT_ON(lockmode);
if (lock->granted[lockmode] == lock->requested[lockmode]) if (lock->granted[lockmode] == lock->requested[lockmode])
lock->waitMask &= LOCKBIT_OFF(lockmode); lock->waitMask &= LOCKBIT_OFF(lockmode);
proclock->holdMask |= LOCKBIT_ON(lockmode);
LOCK_PRINT("GrantLock", lock, lockmode); LOCK_PRINT("GrantLock", lock, lockmode);
Assert((lock->nGranted > 0) && (lock->granted[lockmode] > 0)); Assert((lock->nGranted > 0) && (lock->granted[lockmode] > 0));
Assert(lock->nGranted <= lock->nRequested); Assert(lock->nGranted <= lock->nRequested);
proclock->holding[lockmode]++; }
proclock->nHolding++;
Assert((proclock->nHolding > 0) && (proclock->holding[lockmode] > 0)); /*
* GrantLockLocal -- update the locallock data structures to show
* the lock request has been granted.
*
* We expect that LockAcquire made sure there is room to add a new
* ResourceOwner entry.
*/
static void
GrantLockLocal(LOCALLOCK *locallock, ResourceOwner owner)
{
LOCALLOCKOWNER *lockOwners = locallock->lockOwners;
int i;
Assert(locallock->numLockOwners < locallock->maxLockOwners);
/* Count the total */
locallock->nLocks++;
/* Count the per-owner lock */
for (i = 0; i < locallock->numLockOwners; i++)
{
if (lockOwners[i].owner == owner)
{
lockOwners[i].nLocks++;
return;
}
}
lockOwners[i].owner = owner;
lockOwners[i].nLocks = 1;
locallock->numLockOwners++;
}
/*
* GrantAwaitedLock -- call GrantLockLocal for the lock we are doing
* WaitOnLock on.
*
* proc.c needs this for the case where we are booted off the lock by
* timeout, but discover that someone granted us the lock anyway.
*
* We could just export GrantLockLocal, but that would require including
* resowner.h in lock.h, which creates circularity.
*/
void
GrantAwaitedLock(void)
{
GrantLockLocal(awaitedLock, awaitedOwner);
} }
/* /*
...@@ -845,8 +998,8 @@ GrantLock(LOCK *lock, PROCLOCK *proclock, LOCKMODE lockmode) ...@@ -845,8 +998,8 @@ GrantLock(LOCK *lock, PROCLOCK *proclock, LOCKMODE lockmode)
* The locktable's masterLock must be held at entry. * The locktable's masterLock must be held at entry.
*/ */
static int static int
WaitOnLock(LOCKMETHODID lockmethodid, LOCKMODE lockmode, WaitOnLock(LOCKMETHODID lockmethodid, LOCALLOCK *locallock,
LOCK *lock, PROCLOCK *proclock) ResourceOwner owner)
{ {
LockMethod lockMethodTable = LockMethods[lockmethodid]; LockMethod lockMethodTable = LockMethods[lockmethodid];
char *new_status, char *new_status,
...@@ -854,7 +1007,8 @@ WaitOnLock(LOCKMETHODID lockmethodid, LOCKMODE lockmode, ...@@ -854,7 +1007,8 @@ WaitOnLock(LOCKMETHODID lockmethodid, LOCKMODE lockmode,
Assert(lockmethodid < NumLockMethods); Assert(lockmethodid < NumLockMethods);
LOCK_PRINT("WaitOnLock: sleeping on lock", lock, lockmode); LOCK_PRINT("WaitOnLock: sleeping on lock",
locallock->lock, locallock->tag.mode);
old_status = pstrdup(get_ps_display()); old_status = pstrdup(get_ps_display());
new_status = (char *) palloc(strlen(old_status) + 10); new_status = (char *) palloc(strlen(old_status) + 10);
...@@ -862,6 +1016,9 @@ WaitOnLock(LOCKMETHODID lockmethodid, LOCKMODE lockmode, ...@@ -862,6 +1016,9 @@ WaitOnLock(LOCKMETHODID lockmethodid, LOCKMODE lockmode,
strcat(new_status, " waiting"); strcat(new_status, " waiting");
set_ps_display(new_status); set_ps_display(new_status);
awaitedLock = locallock;
awaitedOwner = owner;
/* /*
* NOTE: Think not to put any shared-state cleanup after the call to * NOTE: Think not to put any shared-state cleanup after the call to
* ProcSleep, in either the normal or failure path. The lock state * ProcSleep, in either the normal or failure path. The lock state
...@@ -877,16 +1034,18 @@ WaitOnLock(LOCKMETHODID lockmethodid, LOCKMODE lockmode, ...@@ -877,16 +1034,18 @@ WaitOnLock(LOCKMETHODID lockmethodid, LOCKMODE lockmode,
*/ */
if (ProcSleep(lockMethodTable, if (ProcSleep(lockMethodTable,
lockmode, locallock->tag.mode,
lock, locallock->lock,
proclock) != STATUS_OK) locallock->proclock) != STATUS_OK)
{ {
/* /*
* We failed as a result of a deadlock, see CheckDeadLock(). Quit * We failed as a result of a deadlock, see CheckDeadLock(). Quit
* now. Removal of the proclock and lock objects, if no longer * now. Removal of the proclock and lock objects, if no longer
* needed, will happen in xact cleanup (see above for motivation). * needed, will happen in xact cleanup (see above for motivation).
*/ */
LOCK_PRINT("WaitOnLock: aborting on lock", lock, lockmode); awaitedLock = NULL;
LOCK_PRINT("WaitOnLock: aborting on lock",
locallock->lock, locallock->tag.mode);
LWLockRelease(lockMethodTable->masterLock); LWLockRelease(lockMethodTable->masterLock);
/* /*
...@@ -897,11 +1056,14 @@ WaitOnLock(LOCKMETHODID lockmethodid, LOCKMODE lockmode, ...@@ -897,11 +1056,14 @@ WaitOnLock(LOCKMETHODID lockmethodid, LOCKMODE lockmode,
/* not reached */ /* not reached */
} }
awaitedLock = NULL;
set_ps_display(old_status); set_ps_display(old_status);
pfree(old_status); pfree(old_status);
pfree(new_status); pfree(new_status);
LOCK_PRINT("WaitOnLock: wakeup on lock", lock, lockmode); LOCK_PRINT("WaitOnLock: wakeup on lock",
locallock->lock, locallock->tag.mode);
return STATUS_OK; return STATUS_OK;
} }
...@@ -944,7 +1106,7 @@ RemoveFromWaitQueue(PGPROC *proc) ...@@ -944,7 +1106,7 @@ RemoveFromWaitQueue(PGPROC *proc)
/* Clean up the proc's own state */ /* Clean up the proc's own state */
proc->waitLock = NULL; proc->waitLock = NULL;
proc->waitHolder = NULL; proc->waitProcLock = NULL;
/* See if any other waiters for the lock can be woken up now */ /* See if any other waiters for the lock can be woken up now */
ProcLockWakeup(GetLocksMethodTable(waitLock), waitLock); ProcLockWakeup(GetLocksMethodTable(waitLock), waitLock);
...@@ -964,12 +1126,12 @@ bool ...@@ -964,12 +1126,12 @@ bool
LockRelease(LOCKMETHODID lockmethodid, LOCKTAG *locktag, LockRelease(LOCKMETHODID lockmethodid, LOCKTAG *locktag,
TransactionId xid, LOCKMODE lockmode) TransactionId xid, LOCKMODE lockmode)
{ {
LOCALLOCKTAG localtag;
LOCALLOCK *locallock;
LOCK *lock; LOCK *lock;
PROCLOCK *proclock;
LWLockId masterLock; LWLockId masterLock;
LockMethod lockMethodTable; LockMethod lockMethodTable;
PROCLOCK *proclock;
PROCLOCKTAG proclocktag;
HTAB *proclockTable;
bool wakeupNeeded = false; bool wakeupNeeded = false;
#ifdef LOCK_DEBUG #ifdef LOCK_DEBUG
...@@ -980,9 +1142,6 @@ LockRelease(LOCKMETHODID lockmethodid, LOCKTAG *locktag, ...@@ -980,9 +1142,6 @@ LockRelease(LOCKMETHODID lockmethodid, LOCKTAG *locktag,
/* ???????? This must be changed when short term locks will be used */ /* ???????? This must be changed when short term locks will be used */
locktag->lockmethodid = lockmethodid; locktag->lockmethodid = lockmethodid;
/* Record release of the lock in the current resource owner */
ResourceOwnerForgetLock(CurrentResourceOwner, locktag, xid, lockmode);
Assert(lockmethodid < NumLockMethods); Assert(lockmethodid < NumLockMethods);
lockMethodTable = LockMethods[lockmethodid]; lockMethodTable = LockMethods[lockmethodid];
if (!lockMethodTable) if (!lockMethodTable)
...@@ -991,69 +1150,107 @@ LockRelease(LOCKMETHODID lockmethodid, LOCKTAG *locktag, ...@@ -991,69 +1150,107 @@ LockRelease(LOCKMETHODID lockmethodid, LOCKTAG *locktag,
return FALSE; return FALSE;
} }
masterLock = lockMethodTable->masterLock;
LWLockAcquire(masterLock, LW_EXCLUSIVE);
/* /*
* Find a lock with this tag * Find the LOCALLOCK entry for this lock and lockmode
*/ */
Assert(LockMethodLockHash[lockmethodid]->hash == tag_hash); MemSet(&localtag, 0, sizeof(localtag)); /* must clear padding */
lock = (LOCK *) hash_search(LockMethodLockHash[lockmethodid], localtag.lock = *locktag;
(void *) locktag, localtag.xid = xid;
HASH_FIND, NULL); localtag.mode = lockmode;
locallock = (LOCALLOCK *) hash_search(LockMethodLocalHash[lockmethodid],
(void *) &localtag,
HASH_FIND, NULL);
/* /*
* let the caller print its own error message, too. Do not * let the caller print its own error message, too. Do not
* ereport(ERROR). * ereport(ERROR).
*/ */
if (!lock) if (!locallock || locallock->nLocks <= 0)
{ {
LWLockRelease(masterLock); elog(WARNING, "you don't own a lock of type %s",
elog(WARNING, "no such lock"); lock_mode_names[lockmode]);
return FALSE; return FALSE;
} }
LOCK_PRINT("LockRelease: found", lock, lockmode);
/* /*
* Find the proclock entry for this proclock. * Decrease the count for the resource owner.
*/ */
MemSet(&proclocktag, 0, sizeof(PROCLOCKTAG)); /* must clear padding,
* needed */
proclocktag.lock = MAKE_OFFSET(lock);
proclocktag.proc = MAKE_OFFSET(MyProc);
TransactionIdStore(xid, &proclocktag.xid);
proclockTable = LockMethodProcLockHash[lockmethodid];
proclock = (PROCLOCK *) hash_search(proclockTable,
(void *) &proclocktag,
HASH_FIND_SAVE, NULL);
if (!proclock)
{ {
LWLockRelease(masterLock); LOCALLOCKOWNER *lockOwners = locallock->lockOwners;
#ifdef USER_LOCKS ResourceOwner owner;
if (lockmethodid == USER_LOCKMETHOD) int i;
elog(WARNING, "no lock with this tag");
/* Session locks and user locks are not transactional */
if (xid != InvalidTransactionId &&
lockmethodid == DEFAULT_LOCKMETHOD)
owner = CurrentResourceOwner;
else else
#endif owner = NULL;
elog(WARNING, "proclock table corrupted");
return FALSE; for (i = locallock->numLockOwners - 1; i >= 0; i--)
{
if (lockOwners[i].owner == owner)
{
Assert(lockOwners[i].nLocks > 0);
if (--lockOwners[i].nLocks == 0)
{
/* compact out unused slot */
locallock->numLockOwners--;
if (i < locallock->numLockOwners)
lockOwners[i] = lockOwners[locallock->numLockOwners];
}
break;
}
}
if (i < 0)
{
/* don't release a lock belonging to another owner */
elog(WARNING, "you don't own a lock of type %s",
lock_mode_names[lockmode]);
return FALSE;
}
} }
/*
* Decrease the total local count. If we're still holding the lock,
* we're done.
*/
locallock->nLocks--;
if (locallock->nLocks > 0)
return TRUE;
/*
* Otherwise we've got to mess with the shared lock table.
*/
masterLock = lockMethodTable->masterLock;
LWLockAcquire(masterLock, LW_EXCLUSIVE);
/*
* We don't need to re-find the lock or proclock, since we kept their
* addresses in the locallock table, and they couldn't have been
* removed while we were holding a lock on them.
*/
lock = locallock->lock;
LOCK_PRINT("LockRelease: found", lock, lockmode);
proclock = locallock->proclock;
PROCLOCK_PRINT("LockRelease: found", proclock); PROCLOCK_PRINT("LockRelease: found", proclock);
/* /*
* Check that we are actually holding a lock of the type we want to * Double-check that we are actually holding a lock of the type we want to
* release. * release.
*/ */
if (!(proclock->holding[lockmode] > 0)) if (!(proclock->holdMask & LOCKBIT_ON(lockmode)))
{ {
PROCLOCK_PRINT("LockRelease: WRONGTYPE", proclock); PROCLOCK_PRINT("LockRelease: WRONGTYPE", proclock);
Assert(proclock->holding[lockmode] >= 0);
LWLockRelease(masterLock); LWLockRelease(masterLock);
elog(WARNING, "you don't own a lock of type %s", elog(WARNING, "you don't own a lock of type %s",
lock_mode_names[lockmode]); lock_mode_names[lockmode]);
RemoveLocalLock(locallock);
return FALSE; return FALSE;
} }
Assert(proclock->nHolding > 0);
Assert((lock->nRequested > 0) && (lock->requested[lockmode] > 0)); Assert((lock->nRequested > 0) && (lock->requested[lockmode] > 0));
Assert((lock->nGranted > 0) && (lock->granted[lockmode] > 0)); Assert((lock->nGranted > 0) && (lock->granted[lockmode] > 0));
Assert(lock->nGranted <= lock->nRequested); Assert(lock->nGranted <= lock->nRequested);
...@@ -1089,67 +1286,69 @@ LockRelease(LOCKMETHODID lockmethodid, LOCKTAG *locktag, ...@@ -1089,67 +1286,69 @@ LockRelease(LOCKMETHODID lockmethodid, LOCKTAG *locktag,
if (lockMethodTable->conflictTab[lockmode] & lock->waitMask) if (lockMethodTable->conflictTab[lockmode] & lock->waitMask)
wakeupNeeded = true; wakeupNeeded = true;
if (lock->nRequested == 0)
{
/*
* if there's no one waiting in the queue, we just released the
* last lock on this object. Delete it from the lock table.
*/
Assert(LockMethodLockHash[lockmethodid]->hash == tag_hash);
lock = (LOCK *) hash_search(LockMethodLockHash[lockmethodid],
(void *) &(lock->tag),
HASH_REMOVE,
NULL);
if (!lock)
{
LWLockRelease(masterLock);
elog(WARNING, "lock table corrupted");
return FALSE;
}
wakeupNeeded = false; /* should be false, but make sure */
}
/* /*
* Now fix the per-proclock lock stats. * Now fix the per-proclock state.
*/ */
proclock->holding[lockmode]--; proclock->holdMask &= LOCKBIT_OFF(lockmode);
proclock->nHolding--;
PROCLOCK_PRINT("LockRelease: updated", proclock); PROCLOCK_PRINT("LockRelease: updated", proclock);
Assert((proclock->nHolding >= 0) && (proclock->holding[lockmode] >= 0));
/* /*
* If this was my last hold on this lock, delete my entry in the * If this was my last hold on this lock, delete my entry in the
* proclock table. * proclock table.
*/ */
if (proclock->nHolding == 0) if (proclock->holdMask == 0)
{ {
PROCLOCK_PRINT("LockRelease: deleting", proclock); PROCLOCK_PRINT("LockRelease: deleting", proclock);
SHMQueueDelete(&proclock->lockLink); SHMQueueDelete(&proclock->lockLink);
SHMQueueDelete(&proclock->procLink); SHMQueueDelete(&proclock->procLink);
proclock = (PROCLOCK *) hash_search(proclockTable, proclock = (PROCLOCK *) hash_search(LockMethodProcLockHash[lockmethodid],
(void *) &proclock, (void *) &(proclock->tag),
HASH_REMOVE_SAVED, NULL); HASH_REMOVE, NULL);
if (!proclock) if (!proclock)
{ {
LWLockRelease(masterLock); LWLockRelease(masterLock);
elog(WARNING, "proclock table corrupted"); elog(WARNING, "proclock table corrupted");
RemoveLocalLock(locallock);
return FALSE; return FALSE;
} }
} }
/* if (lock->nRequested == 0)
* Wake up waiters if needed. {
*/ /*
if (wakeupNeeded) * We've just released the last lock, so garbage-collect the
ProcLockWakeup(lockMethodTable, lock); * lock object.
*/
Assert(SHMQueueEmpty(&(lock->procLocks)));
lock = (LOCK *) hash_search(LockMethodLockHash[lockmethodid],
(void *) &(lock->tag),
HASH_REMOVE, NULL);
if (!lock)
{
LWLockRelease(masterLock);
elog(WARNING, "lock table corrupted");
RemoveLocalLock(locallock);
return FALSE;
}
}
else
{
/*
* Wake up waiters if needed.
*/
if (wakeupNeeded)
ProcLockWakeup(lockMethodTable, lock);
}
LWLockRelease(masterLock); LWLockRelease(masterLock);
RemoveLocalLock(locallock);
return TRUE; return TRUE;
} }
/* /*
* LockReleaseAll -- Release all locks of the specified lock method that * LockReleaseAll -- Release all locks of the specified lock method that
* are held by the specified process. * are held by the current process.
* *
* Well, not necessarily *all* locks. The available behaviors are: * Well, not necessarily *all* locks. The available behaviors are:
* *
...@@ -1160,22 +1359,21 @@ LockRelease(LOCKMETHODID lockmethodid, LOCKTAG *locktag, ...@@ -1160,22 +1359,21 @@ LockRelease(LOCKMETHODID lockmethodid, LOCKTAG *locktag,
* (zero is the Xid used for "session" locks). * (zero is the Xid used for "session" locks).
*/ */
bool bool
LockReleaseAll(LOCKMETHODID lockmethodid, PGPROC *proc, LockReleaseAll(LOCKMETHODID lockmethodid, bool allxids)
bool allxids)
{ {
SHM_QUEUE *procHolders = &(proc->procHolders); HASH_SEQ_STATUS status;
PROCLOCK *proclock; SHM_QUEUE *procLocks = &(MyProc->procLocks);
PROCLOCK *nextHolder;
LWLockId masterLock; LWLockId masterLock;
LockMethod lockMethodTable; LockMethod lockMethodTable;
int i, int i,
numLockModes; numLockModes;
LOCALLOCK *locallock;
PROCLOCK *proclock;
LOCK *lock; LOCK *lock;
#ifdef LOCK_DEBUG #ifdef LOCK_DEBUG
if (lockmethodid == USER_LOCKMETHOD ? Trace_userlocks : Trace_locks) if (lockmethodid == USER_LOCKMETHOD ? Trace_userlocks : Trace_locks)
elog(LOG, "LockReleaseAll: lockmethod=%d, pid=%d", elog(LOG, "LockReleaseAll: lockmethod=%d", lockmethodid);
lockmethodid, proc->pid);
#endif #endif
Assert(lockmethodid < NumLockMethods); Assert(lockmethodid < NumLockMethods);
...@@ -1189,20 +1387,55 @@ LockReleaseAll(LOCKMETHODID lockmethodid, PGPROC *proc, ...@@ -1189,20 +1387,55 @@ LockReleaseAll(LOCKMETHODID lockmethodid, PGPROC *proc,
numLockModes = lockMethodTable->numLockModes; numLockModes = lockMethodTable->numLockModes;
masterLock = lockMethodTable->masterLock; masterLock = lockMethodTable->masterLock;
/*
* First we run through the locallock table and get rid of unwanted
* entries, then we scan the process's proclocks and get rid of those.
* We do this separately because we may have multiple locallock entries
* pointing to the same proclock, and we daren't end up with any
* dangling pointers.
*/
hash_seq_init(&status, LockMethodLocalHash[lockmethodid]);
while ((locallock = (LOCALLOCK *) hash_seq_search(&status)) != NULL)
{
if (locallock->proclock == NULL || locallock->lock == NULL)
{
/*
* We must've run out of shared memory while trying to set up
* this lock. Just forget the local entry.
*/
Assert(locallock->nLocks == 0);
RemoveLocalLock(locallock);
continue;
}
/* Ignore items that are not of the lockmethod to be removed */
if (LOCALLOCK_LOCKMETHOD(*locallock) != lockmethodid)
continue;
/* Ignore locks with Xid=0 unless we are asked to release all locks */
if (TransactionIdEquals(locallock->tag.xid, InvalidTransactionId)
&& !allxids)
continue;
RemoveLocalLock(locallock);
}
LWLockAcquire(masterLock, LW_EXCLUSIVE); LWLockAcquire(masterLock, LW_EXCLUSIVE);
proclock = (PROCLOCK *) SHMQueueNext(procHolders, procHolders, proclock = (PROCLOCK *) SHMQueueNext(procLocks, procLocks,
offsetof(PROCLOCK, procLink)); offsetof(PROCLOCK, procLink));
while (proclock) while (proclock)
{ {
bool wakeupNeeded = false; bool wakeupNeeded = false;
PROCLOCK *nextHolder;
/* Get link first, since we may unlink/delete this proclock */ /* Get link first, since we may unlink/delete this proclock */
nextHolder = (PROCLOCK *) SHMQueueNext(procHolders, &proclock->procLink, nextHolder = (PROCLOCK *) SHMQueueNext(procLocks, &proclock->procLink,
offsetof(PROCLOCK, procLink)); offsetof(PROCLOCK, procLink));
Assert(proclock->tag.proc == MAKE_OFFSET(proc)); Assert(proclock->tag.proc == MAKE_OFFSET(MyProc));
lock = (LOCK *) MAKE_PTR(proclock->tag.lock); lock = (LOCK *) MAKE_PTR(proclock->tag.lock);
...@@ -1220,24 +1453,24 @@ LockReleaseAll(LOCKMETHODID lockmethodid, PGPROC *proc, ...@@ -1220,24 +1453,24 @@ LockReleaseAll(LOCKMETHODID lockmethodid, PGPROC *proc,
Assert(lock->nRequested >= 0); Assert(lock->nRequested >= 0);
Assert(lock->nGranted >= 0); Assert(lock->nGranted >= 0);
Assert(lock->nGranted <= lock->nRequested); Assert(lock->nGranted <= lock->nRequested);
Assert(proclock->nHolding >= 0); Assert((proclock->holdMask & ~lock->grantMask) == 0);
Assert(proclock->nHolding <= lock->nRequested);
/* /*
* fix the general lock stats * fix the general lock stats
*/ */
if (lock->nRequested != proclock->nHolding) if (proclock->holdMask)
{ {
for (i = 1; i <= numLockModes; i++) for (i = 1; i <= numLockModes; i++)
{ {
Assert(proclock->holding[i] >= 0); if (proclock->holdMask & LOCKBIT_ON(i))
if (proclock->holding[i] > 0)
{ {
lock->requested[i] -= proclock->holding[i]; lock->requested[i]--;
lock->granted[i] -= proclock->holding[i]; lock->granted[i]--;
Assert(lock->requested[i] >= 0 && lock->granted[i] >= 0); Assert(lock->requested[i] >= 0 && lock->granted[i] >= 0);
if (lock->granted[i] == 0) if (lock->granted[i] == 0)
lock->grantMask &= LOCKBIT_OFF(i); lock->grantMask &= LOCKBIT_OFF(i);
lock->nRequested--;
lock->nGranted--;
/* /*
* Read comments in LockRelease * Read comments in LockRelease
...@@ -1247,26 +1480,9 @@ LockReleaseAll(LOCKMETHODID lockmethodid, PGPROC *proc, ...@@ -1247,26 +1480,9 @@ LockReleaseAll(LOCKMETHODID lockmethodid, PGPROC *proc,
wakeupNeeded = true; wakeupNeeded = true;
} }
} }
lock->nRequested -= proclock->nHolding;
lock->nGranted -= proclock->nHolding;
Assert((lock->nRequested >= 0) && (lock->nGranted >= 0));
Assert(lock->nGranted <= lock->nRequested);
}
else
{
/*
* This proclock accounts for all the requested locks on the
* object, so we can be lazy and just zero things out.
*/
lock->nRequested = 0;
lock->nGranted = 0;
/* Fix the lock status, just for next LOCK_PRINT message. */
for (i = 1; i <= numLockModes; i++)
{
Assert(lock->requested[i] == lock->granted[i]);
lock->requested[i] = lock->granted[i] = 0;
}
} }
Assert((lock->nRequested >= 0) && (lock->nGranted >= 0));
Assert(lock->nGranted <= lock->nRequested);
LOCK_PRINT("LockReleaseAll: updated", lock, 0); LOCK_PRINT("LockReleaseAll: updated", lock, 0);
PROCLOCK_PRINT("LockReleaseAll: deleting", proclock); PROCLOCK_PRINT("LockReleaseAll: deleting", proclock);
...@@ -1281,7 +1497,7 @@ LockReleaseAll(LOCKMETHODID lockmethodid, PGPROC *proc, ...@@ -1281,7 +1497,7 @@ LockReleaseAll(LOCKMETHODID lockmethodid, PGPROC *proc,
* remove the proclock entry from the hashtable * remove the proclock entry from the hashtable
*/ */
proclock = (PROCLOCK *) hash_search(LockMethodProcLockHash[lockmethodid], proclock = (PROCLOCK *) hash_search(LockMethodProcLockHash[lockmethodid],
(void *) proclock, (void *) &(proclock->tag),
HASH_REMOVE, HASH_REMOVE,
NULL); NULL);
if (!proclock) if (!proclock)
...@@ -1298,7 +1514,7 @@ LockReleaseAll(LOCKMETHODID lockmethodid, PGPROC *proc, ...@@ -1298,7 +1514,7 @@ LockReleaseAll(LOCKMETHODID lockmethodid, PGPROC *proc,
* lock object. * lock object.
*/ */
LOCK_PRINT("LockReleaseAll: deleting", lock, 0); LOCK_PRINT("LockReleaseAll: deleting", lock, 0);
Assert(LockMethodLockHash[lockmethodid]->hash == tag_hash); Assert(SHMQueueEmpty(&(lock->procLocks)));
lock = (LOCK *) hash_search(LockMethodLockHash[lockmethodid], lock = (LOCK *) hash_search(LockMethodLockHash[lockmethodid],
(void *) &(lock->tag), (void *) &(lock->tag),
HASH_REMOVE, NULL); HASH_REMOVE, NULL);
...@@ -1326,6 +1542,130 @@ next_item: ...@@ -1326,6 +1542,130 @@ next_item:
return TRUE; return TRUE;
} }
/*
* LockReleaseCurrentOwner
* Release all locks belonging to CurrentResourceOwner
*
* Only DEFAULT_LOCKMETHOD locks can belong to a resource owner.
*/
void
LockReleaseCurrentOwner(void)
{
HASH_SEQ_STATUS status;
LOCALLOCK *locallock;
LOCALLOCKOWNER *lockOwners;
int i;
hash_seq_init(&status, LockMethodLocalHash[DEFAULT_LOCKMETHOD]);
while ((locallock = (LOCALLOCK *) hash_seq_search(&status)) != NULL)
{
/* Ignore items that must be nontransactional */
if (LOCALLOCK_LOCKMETHOD(*locallock) != DEFAULT_LOCKMETHOD)
continue;
if (TransactionIdEquals(locallock->tag.xid, InvalidTransactionId))
continue;
/* Scan to see if there are any locks belonging to current owner */
lockOwners = locallock->lockOwners;
for (i = locallock->numLockOwners - 1; i >= 0; i--)
{
if (lockOwners[i].owner == CurrentResourceOwner)
{
Assert(lockOwners[i].nLocks > 0);
if (lockOwners[i].nLocks < locallock->nLocks)
{
/*
* We will still hold this lock after forgetting this
* ResourceOwner.
*/
locallock->nLocks -= lockOwners[i].nLocks;
/* compact out unused slot */
locallock->numLockOwners--;
if (i < locallock->numLockOwners)
lockOwners[i] = lockOwners[locallock->numLockOwners];
}
else
{
Assert(lockOwners[i].nLocks == locallock->nLocks);
/* We want to call LockRelease just once */
lockOwners[i].nLocks = 1;
locallock->nLocks = 1;
if (!LockRelease(DEFAULT_LOCKMETHOD,
&locallock->tag.lock,
locallock->tag.xid,
locallock->tag.mode))
elog(WARNING, "LockReleaseCurrentOwner: failed??");
}
break;
}
}
}
}
/*
* LockReassignCurrentOwner
* Reassign all locks belonging to CurrentResourceOwner to belong
* to its parent resource owner
*/
void
LockReassignCurrentOwner(void)
{
ResourceOwner parent = ResourceOwnerGetParent(CurrentResourceOwner);
HASH_SEQ_STATUS status;
LOCALLOCK *locallock;
LOCALLOCKOWNER *lockOwners;
Assert(parent != NULL);
hash_seq_init(&status, LockMethodLocalHash[DEFAULT_LOCKMETHOD]);
while ((locallock = (LOCALLOCK *) hash_seq_search(&status)) != NULL)
{
int i;
int ic = -1;
int ip = -1;
/* Ignore items that must be nontransactional */
if (LOCALLOCK_LOCKMETHOD(*locallock) != DEFAULT_LOCKMETHOD)
continue;
if (TransactionIdEquals(locallock->tag.xid, InvalidTransactionId))
continue;
/*
* Scan to see if there are any locks belonging to current owner
* or its parent
*/
lockOwners = locallock->lockOwners;
for (i = locallock->numLockOwners - 1; i >= 0; i--)
{
if (lockOwners[i].owner == CurrentResourceOwner)
ic = i;
else if (lockOwners[i].owner == parent)
ip = i;
}
if (ic < 0)
continue; /* no current locks */
if (ip < 0)
{
/* Parent has no slot, so just give it child's slot */
lockOwners[ic].owner = parent;
}
else
{
/* Merge child's count with parent's */
lockOwners[ip].nLocks += lockOwners[ic].nLocks;
/* compact out unused slot */
locallock->numLockOwners--;
if (ic < locallock->numLockOwners)
lockOwners[ic] = lockOwners[locallock->numLockOwners];
}
}
}
int int
LockShmemSize(int maxBackends) LockShmemSize(int maxBackends)
{ {
...@@ -1420,7 +1760,7 @@ GetLockmodeName(LOCKMODE mode) ...@@ -1420,7 +1760,7 @@ GetLockmodeName(LOCKMODE mode)
#ifdef LOCK_DEBUG #ifdef LOCK_DEBUG
/* /*
* Dump all locks in the proc->procHolders list. * Dump all locks in the MyProc->procLocks list.
* *
* Must have already acquired the masterLock. * Must have already acquired the masterLock.
*/ */
...@@ -1428,7 +1768,7 @@ void ...@@ -1428,7 +1768,7 @@ void
DumpLocks(void) DumpLocks(void)
{ {
PGPROC *proc; PGPROC *proc;
SHM_QUEUE *procHolders; SHM_QUEUE *procLocks;
PROCLOCK *proclock; PROCLOCK *proclock;
LOCK *lock; LOCK *lock;
int lockmethodid = DEFAULT_LOCKMETHOD; int lockmethodid = DEFAULT_LOCKMETHOD;
...@@ -1438,7 +1778,7 @@ DumpLocks(void) ...@@ -1438,7 +1778,7 @@ DumpLocks(void)
if (proc == NULL) if (proc == NULL)
return; return;
procHolders = &proc->procHolders; procLocks = &proc->procLocks;
Assert(lockmethodid < NumLockMethods); Assert(lockmethodid < NumLockMethods);
lockMethodTable = LockMethods[lockmethodid]; lockMethodTable = LockMethods[lockmethodid];
...@@ -1448,7 +1788,7 @@ DumpLocks(void) ...@@ -1448,7 +1788,7 @@ DumpLocks(void)
if (proc->waitLock) if (proc->waitLock)
LOCK_PRINT("DumpLocks: waiting on", proc->waitLock, 0); LOCK_PRINT("DumpLocks: waiting on", proc->waitLock, 0);
proclock = (PROCLOCK *) SHMQueueNext(procHolders, procHolders, proclock = (PROCLOCK *) SHMQueueNext(procLocks, procLocks,
offsetof(PROCLOCK, procLink)); offsetof(PROCLOCK, procLink));
while (proclock) while (proclock)
...@@ -1460,7 +1800,7 @@ DumpLocks(void) ...@@ -1460,7 +1800,7 @@ DumpLocks(void)
PROCLOCK_PRINT("DumpLocks", proclock); PROCLOCK_PRINT("DumpLocks", proclock);
LOCK_PRINT("DumpLocks", lock, 0); LOCK_PRINT("DumpLocks", lock, 0);
proclock = (PROCLOCK *) SHMQueueNext(procHolders, &proclock->procLink, proclock = (PROCLOCK *) SHMQueueNext(procLocks, &proclock->procLink,
offsetof(PROCLOCK, procLink)); offsetof(PROCLOCK, procLink));
} }
} }
......
...@@ -8,7 +8,7 @@ ...@@ -8,7 +8,7 @@
* *
* *
* IDENTIFICATION * IDENTIFICATION
* $PostgreSQL: pgsql/src/backend/storage/lmgr/proc.c,v 1.150 2004/07/17 03:28:51 tgl Exp $ * $PostgreSQL: pgsql/src/backend/storage/lmgr/proc.c,v 1.151 2004/08/27 17:07:41 tgl Exp $
* *
*------------------------------------------------------------------------- *-------------------------------------------------------------------------
*/ */
...@@ -46,11 +46,11 @@ ...@@ -46,11 +46,11 @@
#include "miscadmin.h" #include "miscadmin.h"
#include "access/xact.h" #include "access/xact.h"
#include "storage/bufmgr.h"
#include "storage/ipc.h" #include "storage/ipc.h"
#include "storage/proc.h" #include "storage/proc.h"
#include "storage/sinval.h" #include "storage/sinval.h"
#include "storage/spin.h" #include "storage/spin.h"
#include "utils/resowner.h"
/* GUC variables */ /* GUC variables */
...@@ -76,11 +76,6 @@ static PGPROC *DummyProcs = NULL; ...@@ -76,11 +76,6 @@ static PGPROC *DummyProcs = NULL;
static bool waitingForLock = false; static bool waitingForLock = false;
static bool waitingForSignal = false; static bool waitingForSignal = false;
/* Auxiliary state, valid when waitingForLock is true */
static LOCKTAG waitingForLockTag;
static TransactionId waitingForLockXid;
static LOCKMODE waitingForLockMode;
/* Mark these volatile because they can be changed by signal handler */ /* Mark these volatile because they can be changed by signal handler */
static volatile bool statement_timeout_active = false; static volatile bool statement_timeout_active = false;
static volatile bool deadlock_timeout_active = false; static volatile bool deadlock_timeout_active = false;
...@@ -250,8 +245,8 @@ InitProcess(void) ...@@ -250,8 +245,8 @@ InitProcess(void)
MyProc->lwExclusive = false; MyProc->lwExclusive = false;
MyProc->lwWaitLink = NULL; MyProc->lwWaitLink = NULL;
MyProc->waitLock = NULL; MyProc->waitLock = NULL;
MyProc->waitHolder = NULL; MyProc->waitProcLock = NULL;
SHMQueueInit(&(MyProc->procHolders)); SHMQueueInit(&(MyProc->procLocks));
/* /*
* Arrange to clean up at backend exit. * Arrange to clean up at backend exit.
...@@ -323,8 +318,8 @@ InitDummyProcess(int proctype) ...@@ -323,8 +318,8 @@ InitDummyProcess(int proctype)
MyProc->lwExclusive = false; MyProc->lwExclusive = false;
MyProc->lwWaitLink = NULL; MyProc->lwWaitLink = NULL;
MyProc->waitLock = NULL; MyProc->waitLock = NULL;
MyProc->waitHolder = NULL; MyProc->waitProcLock = NULL;
SHMQueueInit(&(MyProc->procHolders)); SHMQueueInit(&(MyProc->procLocks));
/* /*
* Arrange to clean up at process exit. * Arrange to clean up at process exit.
...@@ -372,18 +367,10 @@ LockWaitCancel(void) ...@@ -372,18 +367,10 @@ LockWaitCancel(void)
* Somebody kicked us off the lock queue already. Perhaps they * Somebody kicked us off the lock queue already. Perhaps they
* granted us the lock, or perhaps they detected a deadlock. * granted us the lock, or perhaps they detected a deadlock.
* If they did grant us the lock, we'd better remember it in * If they did grant us the lock, we'd better remember it in
* CurrentResourceOwner. * our local lock table.
*
* Exception: if CurrentResourceOwner is NULL then we can't do
* anything. This could only happen when we are invoked from ProcKill
* or some similar place, where all our locks are about to be released
* anyway.
*/ */
if (MyProc->waitStatus == STATUS_OK && CurrentResourceOwner != NULL) if (MyProc->waitStatus == STATUS_OK)
ResourceOwnerRememberLock(CurrentResourceOwner, GrantAwaitedLock();
&waitingForLockTag,
waitingForLockXid,
waitingForLockMode);
} }
waitingForLock = false; waitingForLock = false;
...@@ -433,7 +420,7 @@ ProcReleaseLocks(bool isCommit) ...@@ -433,7 +420,7 @@ ProcReleaseLocks(bool isCommit)
/* If waiting, get off wait queue (should only be needed after error) */ /* If waiting, get off wait queue (should only be needed after error) */
LockWaitCancel(); LockWaitCancel();
/* Release locks */ /* Release locks */
LockReleaseAll(DEFAULT_LOCKMETHOD, MyProc, !isCommit); LockReleaseAll(DEFAULT_LOCKMETHOD, !isCommit);
} }
...@@ -466,11 +453,11 @@ ProcKill(int code, Datum arg) ...@@ -466,11 +453,11 @@ ProcKill(int code, Datum arg)
LockWaitCancel(); LockWaitCancel();
/* Remove from the standard lock table */ /* Remove from the standard lock table */
LockReleaseAll(DEFAULT_LOCKMETHOD, MyProc, true); LockReleaseAll(DEFAULT_LOCKMETHOD, true);
#ifdef USER_LOCKS #ifdef USER_LOCKS
/* Remove from the user lock table */ /* Remove from the user lock table */
LockReleaseAll(USER_LOCKMETHOD, MyProc, true); LockReleaseAll(USER_LOCKMETHOD, true);
#endif #endif
SpinLockAcquire(ProcStructLock); SpinLockAcquire(ProcStructLock);
...@@ -644,10 +631,7 @@ ProcSleep(LockMethod lockMethodTable, ...@@ -644,10 +631,7 @@ ProcSleep(LockMethod lockMethodTable,
{ {
/* Skip the wait and just grant myself the lock. */ /* Skip the wait and just grant myself the lock. */
GrantLock(lock, proclock, lockmode); GrantLock(lock, proclock, lockmode);
ResourceOwnerRememberLock(CurrentResourceOwner, GrantAwaitedLock();
&lock->tag,
proclock->tag.xid,
lockmode);
return STATUS_OK; return STATUS_OK;
} }
/* Break out of loop to put myself before him */ /* Break out of loop to put myself before him */
...@@ -680,7 +664,7 @@ ProcSleep(LockMethod lockMethodTable, ...@@ -680,7 +664,7 @@ ProcSleep(LockMethod lockMethodTable,
/* Set up wait information in PGPROC object, too */ /* Set up wait information in PGPROC object, too */
MyProc->waitLock = lock; MyProc->waitLock = lock;
MyProc->waitHolder = proclock; MyProc->waitProcLock = proclock;
MyProc->waitLockMode = lockmode; MyProc->waitLockMode = lockmode;
MyProc->waitStatus = STATUS_ERROR; /* initialize result for error */ MyProc->waitStatus = STATUS_ERROR; /* initialize result for error */
...@@ -697,9 +681,6 @@ ProcSleep(LockMethod lockMethodTable, ...@@ -697,9 +681,6 @@ ProcSleep(LockMethod lockMethodTable,
} }
/* mark that we are waiting for a lock */ /* mark that we are waiting for a lock */
waitingForLockTag = lock->tag;
waitingForLockXid = proclock->tag.xid;
waitingForLockMode = lockmode;
waitingForLock = true; waitingForLock = true;
/* /*
...@@ -737,8 +718,8 @@ ProcSleep(LockMethod lockMethodTable, ...@@ -737,8 +718,8 @@ ProcSleep(LockMethod lockMethodTable,
* promise that we don't mind losing control to a cancel/die interrupt * promise that we don't mind losing control to a cancel/die interrupt
* here. We don't, because we have no shared-state-change work to do * here. We don't, because we have no shared-state-change work to do
* after being granted the lock (the grantor did it all). We do have * after being granted the lock (the grantor did it all). We do have
* to worry about updating the local CurrentResourceOwner, but if we * to worry about updating the locallock table, but if we lose control
* lose control to an error, LockWaitCancel will fix that up. * to an error, LockWaitCancel will fix that up.
*/ */
PGSemaphoreLock(&MyProc->sem, true); PGSemaphoreLock(&MyProc->sem, true);
...@@ -751,8 +732,7 @@ ProcSleep(LockMethod lockMethodTable, ...@@ -751,8 +732,7 @@ ProcSleep(LockMethod lockMethodTable,
/* /*
* Re-acquire the locktable's masterLock. We have to do this to hold * Re-acquire the locktable's masterLock. We have to do this to hold
* off cancel/die interrupts before we can mess with waitingForLock * off cancel/die interrupts before we can mess with waitingForLock
* (else we might have a missed or duplicated CurrentResourceOwner * (else we might have a missed or duplicated locallock update).
* update).
*/ */
LWLockAcquire(masterLock, LW_EXCLUSIVE); LWLockAcquire(masterLock, LW_EXCLUSIVE);
...@@ -762,13 +742,10 @@ ProcSleep(LockMethod lockMethodTable, ...@@ -762,13 +742,10 @@ ProcSleep(LockMethod lockMethodTable,
waitingForLock = false; waitingForLock = false;
/* /*
* If we got the lock, be sure to remember it in CurrentResourceOwner. * If we got the lock, be sure to remember it in the locallock table.
*/ */
if (MyProc->waitStatus == STATUS_OK) if (MyProc->waitStatus == STATUS_OK)
ResourceOwnerRememberLock(CurrentResourceOwner, GrantAwaitedLock();
&lock->tag,
proclock->tag.xid,
lockmode);
/* /*
* We don't have to do anything else, because the awaker did all the * We don't have to do anything else, because the awaker did all the
...@@ -809,7 +786,7 @@ ProcWakeup(PGPROC *proc, int waitStatus) ...@@ -809,7 +786,7 @@ ProcWakeup(PGPROC *proc, int waitStatus)
/* Clean up process' state and pass it the ok/fail signal */ /* Clean up process' state and pass it the ok/fail signal */
proc->waitLock = NULL; proc->waitLock = NULL;
proc->waitHolder = NULL; proc->waitProcLock = NULL;
proc->waitStatus = waitStatus; proc->waitStatus = waitStatus;
/* And awaken it */ /* And awaken it */
...@@ -850,12 +827,12 @@ ProcLockWakeup(LockMethod lockMethodTable, LOCK *lock) ...@@ -850,12 +827,12 @@ ProcLockWakeup(LockMethod lockMethodTable, LOCK *lock)
LockCheckConflicts(lockMethodTable, LockCheckConflicts(lockMethodTable,
lockmode, lockmode,
lock, lock,
proc->waitHolder, proc->waitProcLock,
proc, proc,
NULL) == STATUS_OK) NULL) == STATUS_OK)
{ {
/* OK to waken */ /* OK to waken */
GrantLock(lock, proc->waitHolder, lockmode); GrantLock(lock, proc->waitProcLock, lockmode);
proc = ProcWakeup(proc, STATUS_OK); proc = ProcWakeup(proc, STATUS_OK);
/* /*
......
...@@ -6,7 +6,7 @@ ...@@ -6,7 +6,7 @@
* Copyright (c) 2002-2003, PostgreSQL Global Development Group * Copyright (c) 2002-2003, PostgreSQL Global Development Group
* *
* IDENTIFICATION * IDENTIFICATION
* $PostgreSQL: pgsql/src/backend/utils/adt/lockfuncs.c,v 1.13 2004/04/01 21:28:45 tgl Exp $ * $PostgreSQL: pgsql/src/backend/utils/adt/lockfuncs.c,v 1.14 2004/08/27 17:07:41 tgl Exp $
* *
*------------------------------------------------------------------------- *-------------------------------------------------------------------------
*/ */
...@@ -92,7 +92,7 @@ pg_lock_status(PG_FUNCTION_ARGS) ...@@ -92,7 +92,7 @@ pg_lock_status(PG_FUNCTION_ARGS)
LOCK *lock; LOCK *lock;
PGPROC *proc; PGPROC *proc;
bool granted; bool granted;
LOCKMODE mode; LOCKMODE mode = 0;
Datum values[6]; Datum values[6];
char nulls[6]; char nulls[6];
HeapTuple tuple; HeapTuple tuple;
...@@ -108,13 +108,16 @@ pg_lock_status(PG_FUNCTION_ARGS) ...@@ -108,13 +108,16 @@ pg_lock_status(PG_FUNCTION_ARGS)
* report again. * report again.
*/ */
granted = false; granted = false;
for (mode = 0; mode < MAX_LOCKMODES; mode++) if (proclock->holdMask)
{ {
if (proclock->holding[mode] > 0) for (mode = 0; mode < MAX_LOCKMODES; mode++)
{ {
granted = true; if (proclock->holdMask & LOCKBIT_ON(mode))
proclock->holding[mode] = 0; {
break; granted = true;
proclock->holdMask &= LOCKBIT_OFF(mode);
break;
}
} }
} }
......
...@@ -14,7 +14,7 @@ ...@@ -14,7 +14,7 @@
* *
* *
* IDENTIFICATION * IDENTIFICATION
* $PostgreSQL: pgsql/src/backend/utils/resowner/resowner.c,v 1.3 2004/08/25 18:43:43 tgl Exp $ * $PostgreSQL: pgsql/src/backend/utils/resowner/resowner.c,v 1.4 2004/08/27 17:07:41 tgl Exp $
* *
*------------------------------------------------------------------------- *-------------------------------------------------------------------------
*/ */
...@@ -30,18 +30,6 @@ ...@@ -30,18 +30,6 @@
#include "utils/relcache.h" #include "utils/relcache.h"
/*
* Info needed to identify/release a lock
*/
typedef struct LockIdData
{
/* we assume lockmethodid is part of locktag */
LOCKTAG locktag;
TransactionId xid;
LOCKMODE lockmode;
} LockIdData;
/* /*
* ResourceOwner objects look like this * ResourceOwner objects look like this
*/ */
...@@ -57,11 +45,6 @@ typedef struct ResourceOwnerData ...@@ -57,11 +45,6 @@ typedef struct ResourceOwnerData
Buffer *buffers; /* dynamically allocated array */ Buffer *buffers; /* dynamically allocated array */
int maxbuffers; /* currently allocated array size */ int maxbuffers; /* currently allocated array size */
/* We have built-in support for remembering owned locks */
int nlocks; /* number of owned locks */
LockIdData *locks; /* dynamically allocated array */
int maxlocks; /* currently allocated array size */
/* We have built-in support for remembering catcache references */ /* We have built-in support for remembering catcache references */
int ncatrefs; /* number of owned catcache pins */ int ncatrefs; /* number of owned catcache pins */
HeapTuple *catrefs; /* dynamically allocated array */ HeapTuple *catrefs; /* dynamically allocated array */
...@@ -274,44 +257,19 @@ ResourceOwnerReleaseInternal(ResourceOwner owner, ...@@ -274,44 +257,19 @@ ResourceOwnerReleaseInternal(ResourceOwner owner,
*/ */
if (owner == TopTransactionResourceOwner) if (owner == TopTransactionResourceOwner)
ProcReleaseLocks(isCommit); ProcReleaseLocks(isCommit);
/* Mark object as holding no locks, just for sanity */
owner->nlocks = 0;
} }
else else
{ {
/* /*
* Release locks retail. Note that LockRelease will remove * Release locks retail. Note that if we are committing a
* the lock entry from my list, so I just have to iterate till
* there are none. Also note that if we are committing a
* subtransaction, we do NOT release its locks yet, but transfer * subtransaction, we do NOT release its locks yet, but transfer
* them to the parent. * them to the parent.
*
* XXX as above, this is a bit inefficient but probably not worth
* the trouble to optimize more.
*/ */
Assert(owner->parent != NULL); Assert(owner->parent != NULL);
while (owner->nlocks > 0) if (isCommit)
{ LockReassignCurrentOwner();
LockIdData *lockid = &owner->locks[owner->nlocks - 1]; else
LockReleaseCurrentOwner();
if (isCommit)
{
ResourceOwnerEnlargeLocks(owner->parent);
ResourceOwnerRememberLock(owner->parent,
&lockid->locktag,
lockid->xid,
lockid->lockmode);
owner->nlocks--;
}
else
{
LockRelease(lockid->locktag.lockmethodid,
&lockid->locktag,
lockid->xid,
lockid->lockmode);
/* LockRelease will have removed the entry from list */
}
}
} }
} }
else if (phase == RESOURCE_RELEASE_AFTER_LOCKS) else if (phase == RESOURCE_RELEASE_AFTER_LOCKS)
...@@ -368,7 +326,6 @@ ResourceOwnerDelete(ResourceOwner owner) ...@@ -368,7 +326,6 @@ ResourceOwnerDelete(ResourceOwner owner)
/* And it better not own any resources, either */ /* And it better not own any resources, either */
Assert(owner->nbuffers == 0); Assert(owner->nbuffers == 0);
Assert(owner->nlocks == 0);
Assert(owner->ncatrefs == 0); Assert(owner->ncatrefs == 0);
Assert(owner->ncatlistrefs == 0); Assert(owner->ncatlistrefs == 0);
Assert(owner->nrelrefs == 0); Assert(owner->nrelrefs == 0);
...@@ -390,8 +347,6 @@ ResourceOwnerDelete(ResourceOwner owner) ...@@ -390,8 +347,6 @@ ResourceOwnerDelete(ResourceOwner owner)
/* And free the object. */ /* And free the object. */
if (owner->buffers) if (owner->buffers)
pfree(owner->buffers); pfree(owner->buffers);
if (owner->locks)
pfree(owner->locks);
if (owner->catrefs) if (owner->catrefs)
pfree(owner->catrefs); pfree(owner->catrefs);
if (owner->catlistrefs) if (owner->catlistrefs)
...@@ -402,6 +357,15 @@ ResourceOwnerDelete(ResourceOwner owner) ...@@ -402,6 +357,15 @@ ResourceOwnerDelete(ResourceOwner owner)
pfree(owner); pfree(owner);
} }
/*
* Fetch parent of a ResourceOwner (returns NULL if top-level owner)
*/
ResourceOwner
ResourceOwnerGetParent(ResourceOwner owner)
{
return owner->parent;
}
/* /*
* Reassign a ResourceOwner to have a new parent * Reassign a ResourceOwner to have a new parent
*/ */
...@@ -581,97 +545,6 @@ ResourceOwnerForgetBuffer(ResourceOwner owner, Buffer buffer) ...@@ -581,97 +545,6 @@ ResourceOwnerForgetBuffer(ResourceOwner owner, Buffer buffer)
} }
} }
/*
* Make sure there is room for at least one more entry in a ResourceOwner's
* lock array.
*
* This is separate from actually inserting an entry because if we run out
* of memory, it's critical to do so *before* acquiring the resource.
*/
void
ResourceOwnerEnlargeLocks(ResourceOwner owner)
{
int newmax;
if (owner->nlocks < owner->maxlocks)
return; /* nothing to do */
if (owner->locks == NULL)
{
newmax = 16;
owner->locks = (LockIdData *)
MemoryContextAlloc(TopMemoryContext, newmax * sizeof(LockIdData));
owner->maxlocks = newmax;
}
else
{
newmax = owner->maxlocks * 2;
owner->locks = (LockIdData *)
repalloc(owner->locks, newmax * sizeof(LockIdData));
owner->maxlocks = newmax;
}
}
/*
* Remember that a lock is owned by a ResourceOwner
*
* Caller must have previously done ResourceOwnerEnlargeLocks()
*/
void
ResourceOwnerRememberLock(ResourceOwner owner,
LOCKTAG *locktag,
TransactionId xid,
LOCKMODE lockmode)
{
/* Session locks and user locks are not transactional */
if (xid != InvalidTransactionId &&
locktag->lockmethodid == DEFAULT_LOCKMETHOD)
{
Assert(owner->nlocks < owner->maxlocks);
owner->locks[owner->nlocks].locktag = *locktag;
owner->locks[owner->nlocks].xid = xid;
owner->locks[owner->nlocks].lockmode = lockmode;
owner->nlocks++;
}
}
/*
* Forget that a lock is owned by a ResourceOwner
*/
void
ResourceOwnerForgetLock(ResourceOwner owner,
LOCKTAG *locktag,
TransactionId xid,
LOCKMODE lockmode)
{
/* Session locks and user locks are not transactional */
if (xid != InvalidTransactionId &&
locktag->lockmethodid == DEFAULT_LOCKMETHOD)
{
LockIdData *locks = owner->locks;
int nl1 = owner->nlocks - 1;
int i;
for (i = nl1; i >= 0; i--)
{
if (memcmp(&locks[i].locktag, locktag, sizeof(LOCKTAG)) == 0 &&
locks[i].xid == xid &&
locks[i].lockmode == lockmode)
{
while (i < nl1)
{
locks[i] = locks[i + 1];
i++;
}
owner->nlocks = nl1;
return;
}
}
elog(ERROR, "lock %u/%u/%u is not owned by resource owner %s",
locktag->relId, locktag->dbId, locktag->objId.xid, owner->name);
}
}
/* /*
* Make sure there is room for at least one more entry in a ResourceOwner's * Make sure there is room for at least one more entry in a ResourceOwner's
* catcache reference array. * catcache reference array.
......
...@@ -7,7 +7,7 @@ ...@@ -7,7 +7,7 @@
* Portions Copyright (c) 1996-2003, PostgreSQL Global Development Group * Portions Copyright (c) 1996-2003, PostgreSQL Global Development Group
* Portions Copyright (c) 1994, Regents of the University of California * Portions Copyright (c) 1994, Regents of the University of California
* *
* $PostgreSQL: pgsql/src/include/storage/lock.h,v 1.80 2004/08/26 17:22:28 tgl Exp $ * $PostgreSQL: pgsql/src/include/storage/lock.h,v 1.81 2004/08/27 17:07:42 tgl Exp $
* *
*------------------------------------------------------------------------- *-------------------------------------------------------------------------
*/ */
...@@ -69,15 +69,17 @@ typedef uint16 LOCKMETHODID; ...@@ -69,15 +69,17 @@ typedef uint16 LOCKMETHODID;
#define LockMethodIsValid(lockmethodid) ((lockmethodid) != INVALID_LOCKMETHOD) #define LockMethodIsValid(lockmethodid) ((lockmethodid) != INVALID_LOCKMETHOD)
extern int NumLockMethods;
/* /*
* This is the control structure for a lock table. It lives in shared * This is the control structure for a lock table. It lives in shared
* memory. Currently, none of these fields change after startup. In addition * memory. Currently, none of these fields change after startup. In addition
* to the LockMethodData, a lock table has a "lockHash" table holding * to the LockMethodData, a lock table has a shared "lockHash" table holding
* per-locked-object lock information, and a "proclockHash" table holding * per-locked-object lock information, and a shared "proclockHash" table
* per-lock-holder/waiter lock information. * holding per-lock-holder/waiter lock information.
* *
* lockmethodid -- the handle used by the lock table's clients to * masterLock -- LWLock used to synchronize access to the table
* refer to the type of lock table being used.
* *
* numLockModes -- number of lock types (READ,WRITE,etc) that * numLockModes -- number of lock types (READ,WRITE,etc) that
* are defined on this lock table * are defined on this lock table
...@@ -85,16 +87,14 @@ typedef uint16 LOCKMETHODID; ...@@ -85,16 +87,14 @@ typedef uint16 LOCKMETHODID;
* conflictTab -- this is an array of bitmasks showing lock * conflictTab -- this is an array of bitmasks showing lock
* type conflicts. conflictTab[i] is a mask with the j-th bit * type conflicts. conflictTab[i] is a mask with the j-th bit
* turned on if lock types i and j conflict. * turned on if lock types i and j conflict.
*
* masterLock -- LWLock used to synchronize access to the table
*/ */
typedef struct LockMethodData typedef struct LockMethodData
{ {
LOCKMETHODID lockmethodid; LWLockId masterLock;
int numLockModes; int numLockModes;
LOCKMASK conflictTab[MAX_LOCKMODES]; LOCKMASK conflictTab[MAX_LOCKMODES];
LWLockId masterLock;
} LockMethodData; } LockMethodData;
typedef LockMethodData *LockMethod; typedef LockMethodData *LockMethod;
...@@ -113,8 +113,8 @@ typedef struct LOCKTAG ...@@ -113,8 +113,8 @@ typedef struct LOCKTAG
} objId; } objId;
/* /*
* offnum should be part of objId.tupleId above, but would increase * offnum should be part of objId union above, but doing that would
* sizeof(LOCKTAG) and so moved here; currently used by userlocks * increase sizeof(LOCKTAG) due to padding. Currently used by userlocks
* only. * only.
*/ */
OffsetNumber offnum; OffsetNumber offnum;
...@@ -129,7 +129,7 @@ typedef struct LOCKTAG ...@@ -129,7 +129,7 @@ typedef struct LOCKTAG
* tag -- uniquely identifies the object being locked * tag -- uniquely identifies the object being locked
* grantMask -- bitmask for all lock types currently granted on this object. * grantMask -- bitmask for all lock types currently granted on this object.
* waitMask -- bitmask for all lock types currently awaited on this object. * waitMask -- bitmask for all lock types currently awaited on this object.
* lockHolders -- list of PROCLOCK objects for this lock. * procLocks -- list of PROCLOCK objects for this lock.
* waitProcs -- queue of processes waiting for this lock. * waitProcs -- queue of processes waiting for this lock.
* requested -- count of each lock type currently requested on the lock * requested -- count of each lock type currently requested on the lock
* (includes requests already granted!!). * (includes requests already granted!!).
...@@ -145,7 +145,7 @@ typedef struct LOCK ...@@ -145,7 +145,7 @@ typedef struct LOCK
/* data */ /* data */
LOCKMASK grantMask; /* bitmask for lock types already granted */ LOCKMASK grantMask; /* bitmask for lock types already granted */
LOCKMASK waitMask; /* bitmask for lock types awaited */ LOCKMASK waitMask; /* bitmask for lock types awaited */
SHM_QUEUE lockHolders; /* list of PROCLOCK objects assoc. with SHM_QUEUE procLocks; /* list of PROCLOCK objects assoc. with
* lock */ * lock */
PROC_QUEUE waitProcs; /* list of PGPROC objects waiting on lock */ PROC_QUEUE waitProcs; /* list of PGPROC objects waiting on lock */
int requested[MAX_LOCKMODES]; /* counts of requested int requested[MAX_LOCKMODES]; /* counts of requested
...@@ -168,7 +168,7 @@ typedef struct LOCK ...@@ -168,7 +168,7 @@ typedef struct LOCK
* proclock hashtable. A PROCLOCKTAG value uniquely identifies the combination * proclock hashtable. A PROCLOCKTAG value uniquely identifies the combination
* of a lockable object and a holder/waiter for that object. * of a lockable object and a holder/waiter for that object.
* *
* There are two possible kinds of proclock tags: a transaction (identified * There are two possible kinds of proclock owners: a transaction (identified
* both by the PGPROC of the backend running it, and the xact's own ID) and * both by the PGPROC of the backend running it, and the xact's own ID) and
* a session (identified by backend PGPROC, with XID = InvalidTransactionId). * a session (identified by backend PGPROC, with XID = InvalidTransactionId).
* *
...@@ -177,10 +177,10 @@ typedef struct LOCK ...@@ -177,10 +177,10 @@ typedef struct LOCK
* under several different XIDs at once (including session locks). We treat * under several different XIDs at once (including session locks). We treat
* such locks as never conflicting (a backend can never block itself). * such locks as never conflicting (a backend can never block itself).
* *
* The holding[] array counts the granted locks (of each type) represented * The holdMask field shows the already-granted locks represented by this
* by this proclock. Note that there will be a proclock object, possibly with * proclock. Note that there will be a proclock object, possibly with
* zero holding[], for any lock that the process is currently waiting on. * zero holdMask, for any lock that the process is currently waiting on.
* Otherwise, proclock objects whose counts have gone to zero are recycled * Otherwise, proclock objects whose holdMasks are zero are recycled
* as soon as convenient. * as soon as convenient.
* *
* Each PROCLOCK object is linked into lists for both the associated LOCK * Each PROCLOCK object is linked into lists for both the associated LOCK
...@@ -202,8 +202,7 @@ typedef struct PROCLOCK ...@@ -202,8 +202,7 @@ typedef struct PROCLOCK
PROCLOCKTAG tag; /* unique identifier of proclock object */ PROCLOCKTAG tag; /* unique identifier of proclock object */
/* data */ /* data */
int holding[MAX_LOCKMODES]; /* count of locks currently held */ LOCKMASK holdMask; /* bitmask for lock types currently held */
int nHolding; /* total of holding[] array */
SHM_QUEUE lockLink; /* list link for lock's list of proclocks */ SHM_QUEUE lockLink; /* list link for lock's list of proclocks */
SHM_QUEUE procLink; /* list link for process's list of SHM_QUEUE procLink; /* list link for process's list of
* proclocks */ * proclocks */
...@@ -212,6 +211,49 @@ typedef struct PROCLOCK ...@@ -212,6 +211,49 @@ typedef struct PROCLOCK
#define PROCLOCK_LOCKMETHOD(proclock) \ #define PROCLOCK_LOCKMETHOD(proclock) \
(((LOCK *) MAKE_PTR((proclock).tag.lock))->tag.lockmethodid) (((LOCK *) MAKE_PTR((proclock).tag.lock))->tag.lockmethodid)
/*
* Each backend also maintains a local hash table with information about each
* lock it is currently interested in. In particular the local table counts
* the number of times that lock has been acquired. This allows multiple
* requests for the same lock to be executed without additional accesses to
* shared memory. We also track the number of lock acquisitions per
* ResourceOwner, so that we can release just those locks belonging to a
* particular ResourceOwner.
*/
typedef struct LOCALLOCKTAG
{
LOCKTAG lock; /* identifies the lockable object */
TransactionId xid; /* xact ID, or InvalidTransactionId */
LOCKMODE mode; /* lock mode for this table entry */
} LOCALLOCKTAG;
typedef struct LOCALLOCKOWNER
{
/*
* Note: owner can be NULL to indicate a non-transactional lock.
* Must use a forward struct reference to avoid circularity.
*/
struct ResourceOwnerData *owner;
int nLocks; /* # of times held by this owner */
} LOCALLOCKOWNER;
typedef struct LOCALLOCK
{
/* tag */
LOCALLOCKTAG tag; /* unique identifier of locallock entry */
/* data */
LOCK *lock; /* associated LOCK object in shared mem */
PROCLOCK *proclock; /* associated PROCLOCK object in shmem */
int nLocks; /* total number of times lock is held */
int numLockOwners; /* # of relevant ResourceOwners */
int maxLockOwners; /* allocated size of array */
LOCALLOCKOWNER *lockOwners; /* dynamically resizable array */
} LOCALLOCK;
#define LOCALLOCK_LOCKMETHOD(llock) ((llock).tag.lock.lockmethodid)
/* /*
* This struct holds information passed from lmgr internals to the lock * This struct holds information passed from lmgr internals to the lock
* listing user-level functions (lockfuncs.c). For each PROCLOCK in the * listing user-level functions (lockfuncs.c). For each PROCLOCK in the
...@@ -229,7 +271,6 @@ typedef struct ...@@ -229,7 +271,6 @@ typedef struct
LOCK *locks; LOCK *locks;
} LockData; } LockData;
extern int NumLockMethods;
/* /*
* function prototypes * function prototypes
...@@ -244,13 +285,15 @@ extern bool LockAcquire(LOCKMETHODID lockmethodid, LOCKTAG *locktag, ...@@ -244,13 +285,15 @@ extern bool LockAcquire(LOCKMETHODID lockmethodid, LOCKTAG *locktag,
TransactionId xid, LOCKMODE lockmode, bool dontWait); TransactionId xid, LOCKMODE lockmode, bool dontWait);
extern bool LockRelease(LOCKMETHODID lockmethodid, LOCKTAG *locktag, extern bool LockRelease(LOCKMETHODID lockmethodid, LOCKTAG *locktag,
TransactionId xid, LOCKMODE lockmode); TransactionId xid, LOCKMODE lockmode);
extern bool LockReleaseAll(LOCKMETHODID lockmethodid, PGPROC *proc, extern bool LockReleaseAll(LOCKMETHODID lockmethodid, bool allxids);
bool allxids); extern void LockReleaseCurrentOwner(void);
extern void LockReassignCurrentOwner(void);
extern int LockCheckConflicts(LockMethod lockMethodTable, extern int LockCheckConflicts(LockMethod lockMethodTable,
LOCKMODE lockmode, LOCKMODE lockmode,
LOCK *lock, PROCLOCK *proclock, PGPROC *proc, LOCK *lock, PROCLOCK *proclock, PGPROC *proc,
int *myHolding); int *myHolding);
extern void GrantLock(LOCK *lock, PROCLOCK *proclock, LOCKMODE lockmode); extern void GrantLock(LOCK *lock, PROCLOCK *proclock, LOCKMODE lockmode);
extern void GrantAwaitedLock(void);
extern void RemoveFromWaitQueue(PGPROC *proc); extern void RemoveFromWaitQueue(PGPROC *proc);
extern int LockShmemSize(int maxBackends); extern int LockShmemSize(int maxBackends);
extern bool DeadLockCheck(PGPROC *proc); extern bool DeadLockCheck(PGPROC *proc);
......
...@@ -7,7 +7,7 @@ ...@@ -7,7 +7,7 @@
* Portions Copyright (c) 1996-2003, PostgreSQL Global Development Group * Portions Copyright (c) 1996-2003, PostgreSQL Global Development Group
* Portions Copyright (c) 1994, Regents of the University of California * Portions Copyright (c) 1994, Regents of the University of California
* *
* $PostgreSQL: pgsql/src/include/storage/proc.h,v 1.72 2004/08/01 17:32:21 tgl Exp $ * $PostgreSQL: pgsql/src/include/storage/proc.h,v 1.73 2004/08/27 17:07:42 tgl Exp $
* *
*------------------------------------------------------------------------- *-------------------------------------------------------------------------
*/ */
...@@ -78,14 +78,14 @@ struct PGPROC ...@@ -78,14 +78,14 @@ struct PGPROC
struct PGPROC *lwWaitLink; /* next waiter for same LW lock */ struct PGPROC *lwWaitLink; /* next waiter for same LW lock */
/* Info about lock the process is currently waiting for, if any. */ /* Info about lock the process is currently waiting for, if any. */
/* waitLock and waitHolder are NULL if not currently waiting. */ /* waitLock and waitProcLock are NULL if not currently waiting. */
LOCK *waitLock; /* Lock object we're sleeping on ... */ LOCK *waitLock; /* Lock object we're sleeping on ... */
PROCLOCK *waitHolder; /* Per-holder info for awaited lock */ PROCLOCK *waitProcLock; /* Per-holder info for awaited lock */
LOCKMODE waitLockMode; /* type of lock we're waiting for */ LOCKMODE waitLockMode; /* type of lock we're waiting for */
LOCKMASK heldLocks; /* bitmask for lock types already held on LOCKMASK heldLocks; /* bitmask for lock types already held on
* this lock object by this backend */ * this lock object by this backend */
SHM_QUEUE procHolders; /* list of PROCLOCK objects for locks held SHM_QUEUE procLocks; /* list of PROCLOCK objects for locks held
* or awaited by this backend */ * or awaited by this backend */
struct XidCache subxids; /* cache for subtransaction XIDs */ struct XidCache subxids; /* cache for subtransaction XIDs */
......
...@@ -12,7 +12,7 @@ ...@@ -12,7 +12,7 @@
* Portions Copyright (c) 1996-2003, PostgreSQL Global Development Group * Portions Copyright (c) 1996-2003, PostgreSQL Global Development Group
* Portions Copyright (c) 1994, Regents of the University of California * Portions Copyright (c) 1994, Regents of the University of California
* *
* $PostgreSQL: pgsql/src/include/utils/resowner.h,v 1.1 2004/07/17 03:31:47 tgl Exp $ * $PostgreSQL: pgsql/src/include/utils/resowner.h,v 1.2 2004/08/27 17:07:42 tgl Exp $
* *
*------------------------------------------------------------------------- *-------------------------------------------------------------------------
*/ */
...@@ -20,7 +20,6 @@ ...@@ -20,7 +20,6 @@
#define RESOWNER_H #define RESOWNER_H
#include "storage/buf.h" #include "storage/buf.h"
#include "storage/lock.h"
#include "utils/catcache.h" #include "utils/catcache.h"
#include "utils/rel.h" #include "utils/rel.h"
...@@ -76,6 +75,7 @@ extern void ResourceOwnerRelease(ResourceOwner owner, ...@@ -76,6 +75,7 @@ extern void ResourceOwnerRelease(ResourceOwner owner,
bool isCommit, bool isCommit,
bool isTopLevel); bool isTopLevel);
extern void ResourceOwnerDelete(ResourceOwner owner); extern void ResourceOwnerDelete(ResourceOwner owner);
extern ResourceOwner ResourceOwnerGetParent(ResourceOwner owner);
extern void ResourceOwnerNewParent(ResourceOwner owner, extern void ResourceOwnerNewParent(ResourceOwner owner,
ResourceOwner newparent); ResourceOwner newparent);
extern void RegisterResourceReleaseCallback(ResourceReleaseCallback callback, extern void RegisterResourceReleaseCallback(ResourceReleaseCallback callback,
...@@ -88,17 +88,6 @@ extern void ResourceOwnerEnlargeBuffers(ResourceOwner owner); ...@@ -88,17 +88,6 @@ extern void ResourceOwnerEnlargeBuffers(ResourceOwner owner);
extern void ResourceOwnerRememberBuffer(ResourceOwner owner, Buffer buffer); extern void ResourceOwnerRememberBuffer(ResourceOwner owner, Buffer buffer);
extern void ResourceOwnerForgetBuffer(ResourceOwner owner, Buffer buffer); extern void ResourceOwnerForgetBuffer(ResourceOwner owner, Buffer buffer);
/* support for lock management */
extern void ResourceOwnerEnlargeLocks(ResourceOwner owner);
extern void ResourceOwnerRememberLock(ResourceOwner owner,
LOCKTAG *locktag,
TransactionId xid,
LOCKMODE lockmode);
extern void ResourceOwnerForgetLock(ResourceOwner owner,
LOCKTAG *locktag,
TransactionId xid,
LOCKMODE lockmode);
/* support for catcache refcount management */ /* support for catcache refcount management */
extern void ResourceOwnerEnlargeCatCacheRefs(ResourceOwner owner); extern void ResourceOwnerEnlargeCatCacheRefs(ResourceOwner owner);
extern void ResourceOwnerRememberCatCacheRef(ResourceOwner owner, extern void ResourceOwnerRememberCatCacheRef(ResourceOwner owner,
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment