lock.c

PostgreSQL7.4.6 for Linux

	/*
	 * let the caller print its own error message, too. Do not
	 * ereport(ERROR).
	 */
	if (!lock)
	{
		LWLockRelease(masterLock);
		elog(WARNING, "no such lock");
		return FALSE;
	}
	LOCK_PRINT("LockRelease: found", lock, lockmode);

	/*
	 * Find the proclock entry for this proclock.
	 */
	MemSet(&proclocktag, 0, sizeof(PROCLOCKTAG));		/* must clear padding,
														 * needed */
	proclocktag.lock = MAKE_OFFSET(lock);
	proclocktag.proc = MAKE_OFFSET(MyProc);
	TransactionIdStore(xid, &proclocktag.xid);

	proclockTable = lockMethodTable->proclockHash;
	proclock = (PROCLOCK *) hash_search(proclockTable,
										(void *) &proclocktag,
										HASH_FIND_SAVE, NULL);
	if (!proclock)
	{
		LWLockRelease(masterLock);
#ifdef USER_LOCKS
		if (lockmethod == USER_LOCKMETHOD)
			elog(WARNING, "no lock with this tag");
		else
#endif
			elog(WARNING, "proclock table corrupted");
		return FALSE;
	}
	PROCLOCK_PRINT("LockRelease: found", proclock);

	/*
	 * Check that we are actually holding a lock of the type we want to
	 * release.
	 */
	if (!(proclock->holding[lockmode] > 0))
	{
		PROCLOCK_PRINT("LockRelease: WRONGTYPE", proclock);
		Assert(proclock->holding[lockmode] >= 0);
		LWLockRelease(masterLock);
		elog(WARNING, "you don't own a lock of type %s",
			 lock_mode_names[lockmode]);
		return FALSE;
	}
	Assert(proclock->nHolding > 0);
	Assert((lock->nRequested > 0) && (lock->requested[lockmode] > 0));
	Assert((lock->nGranted > 0) && (lock->granted[lockmode] > 0));
	Assert(lock->nGranted <= lock->nRequested);

	/*
	 * fix the general lock stats
	 */
	lock->nRequested--;
	lock->requested[lockmode]--;
	lock->nGranted--;
	lock->granted[lockmode]--;

	if (lock->granted[lockmode] == 0)
	{
		/* change the conflict mask.  No more of this lock type. */
		lock->grantMask &= BITS_OFF[lockmode];
	}

	LOCK_PRINT("LockRelease: updated", lock, lockmode);
	Assert((lock->nRequested >= 0) && (lock->requested[lockmode] >= 0));
	Assert((lock->nGranted >= 0) && (lock->granted[lockmode] >= 0));
	Assert(lock->nGranted <= lock->nRequested);

	/*
	 * We need only run ProcLockWakeup if the released lock conflicts with
	 * at least one of the lock types requested by waiter(s).  Otherwise
	 * whatever conflict made them wait must still exist.  NOTE: before
	 * MVCC, we could skip wakeup if lock->granted[lockmode] was still
	 * positive. But that's not true anymore, because the remaining
	 * granted locks might belong to some waiter, who could now be
	 * awakened because he doesn't conflict with his own locks.
	 */
	if (lockMethodTable->conflictTab[lockmode] & lock->waitMask)
		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(lockMethodTable->lockHash->hash == tag_hash);
		lock = (LOCK *) hash_search(lockMethodTable->lockHash,
									(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.
	 */
	proclock->holding[lockmode]--;
	proclock->nHolding--;
	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
	 * proclock table.
	 */
	if (proclock->nHolding == 0)
	{
		PROCLOCK_PRINT("LockRelease: deleting", proclock);
		SHMQueueDelete(&proclock->lockLink);
		SHMQueueDelete(&proclock->procLink);
		proclock = (PROCLOCK *) hash_search(proclockTable,
											(void *) &proclock,
											HASH_REMOVE_SAVED, NULL);
		if (!proclock)
		{
			LWLockRelease(masterLock);
			elog(WARNING, "proclock table corrupted");
			return FALSE;
		}
	}

	/*
	 * Wake up waiters if needed.
	 */
	if (wakeupNeeded)
		ProcLockWakeup(lockMethodTable, lock);

	LWLockRelease(masterLock);
	return TRUE;
}

/*
 * LockReleaseAll -- Release all locks in a process's lock list.
 *
 * Well, not really *all* locks.
 *
 * If 'allxids' is TRUE, all locks of the specified lock method are
 * released, regardless of transaction affiliation.
 *
 * If 'allxids' is FALSE, all locks of the specified lock method and
 * specified XID are released.
 */
bool
LockReleaseAll(LOCKMETHOD lockmethod, PGPROC *proc,
			   bool allxids, TransactionId xid)
{
	SHM_QUEUE  *procHolders = &(proc->procHolders);
	PROCLOCK   *proclock;
	PROCLOCK   *nextHolder;
	LWLockId	masterLock;
	LOCKMETHODTABLE *lockMethodTable;
	int			i,
				numLockModes;
	LOCK	   *lock;

#ifdef LOCK_DEBUG
	if (lockmethod == USER_LOCKMETHOD ? Trace_userlocks : Trace_locks)
		elog(LOG, "LockReleaseAll: lockmethod=%d, pid=%d",
			 lockmethod, proc->pid);
#endif

	Assert(lockmethod < NumLockMethods);
	lockMethodTable = LockMethodTable[lockmethod];
	if (!lockMethodTable)
	{
		elog(WARNING, "bad lock method: %d", lockmethod);
		return FALSE;
	}

	numLockModes = lockMethodTable->numLockModes;
	masterLock = lockMethodTable->masterLock;

	LWLockAcquire(masterLock, LW_EXCLUSIVE);

	proclock = (PROCLOCK *) SHMQueueNext(procHolders, procHolders,
										 offsetof(PROCLOCK, procLink));

	while (proclock)
	{
		bool		wakeupNeeded = false;

		/* Get link first, since we may unlink/delete this proclock */
		nextHolder = (PROCLOCK *) SHMQueueNext(procHolders, &proclock->procLink,
										   offsetof(PROCLOCK, procLink));

		Assert(proclock->tag.proc == MAKE_OFFSET(proc));

		lock = (LOCK *) MAKE_PTR(proclock->tag.lock);

		/* Ignore items that are not of the lockmethod to be removed */
		if (LOCK_LOCKMETHOD(*lock) != lockmethod)
			goto next_item;

		/* If not allxids, ignore items that are of the wrong xid */
		if (!allxids && !TransactionIdEquals(xid, proclock->tag.xid))
			goto next_item;

		PROCLOCK_PRINT("LockReleaseAll", proclock);
		LOCK_PRINT("LockReleaseAll", lock, 0);
		Assert(lock->nRequested >= 0);
		Assert(lock->nGranted >= 0);
		Assert(lock->nGranted <= lock->nRequested);
		Assert(proclock->nHolding >= 0);
		Assert(proclock->nHolding <= lock->nRequested);

		/*
		 * fix the general lock stats
		 */
		if (lock->nRequested != proclock->nHolding)
		{
			for (i = 1; i <= numLockModes; i++)
			{
				Assert(proclock->holding[i] >= 0);
				if (proclock->holding[i] > 0)
				{
					lock->requested[i] -= proclock->holding[i];
					lock->granted[i] -= proclock->holding[i];
					Assert(lock->requested[i] >= 0 && lock->granted[i] >= 0);
					if (lock->granted[i] == 0)
						lock->grantMask &= BITS_OFF[i];

					/*
					 * Read comments in LockRelease
					 */
					if (!wakeupNeeded &&
						lockMethodTable->conflictTab[i] & lock->waitMask)
						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;
			}
		}
		LOCK_PRINT("LockReleaseAll: updated", lock, 0);

		PROCLOCK_PRINT("LockReleaseAll: deleting", proclock);

		/*
		 * Remove the proclock entry from the linked lists
		 */
		SHMQueueDelete(&proclock->lockLink);
		SHMQueueDelete(&proclock->procLink);

		/*
		 * remove the proclock entry from the hashtable
		 */
		proclock = (PROCLOCK *) hash_search(lockMethodTable->proclockHash,
											(void *) proclock,
											HASH_REMOVE,
											NULL);
		if (!proclock)
		{
			LWLockRelease(masterLock);
			elog(WARNING, "proclock table corrupted");
			return FALSE;
		}

		if (lock->nRequested == 0)
		{
			/*
			 * We've just released the last lock, so garbage-collect the
			 * lock object.
			 */
			LOCK_PRINT("LockReleaseAll: deleting", lock, 0);
			Assert(lockMethodTable->lockHash->hash == tag_hash);
			lock = (LOCK *) hash_search(lockMethodTable->lockHash,
										(void *) &(lock->tag),
										HASH_REMOVE, NULL);
			if (!lock)
			{
				LWLockRelease(masterLock);
				elog(WARNING, "cannot remove lock from HTAB");
				return FALSE;
			}
		}
		else if (wakeupNeeded)
			ProcLockWakeup(lockMethodTable, lock);

next_item:
		proclock = nextHolder;
	}

	LWLockRelease(masterLock);

#ifdef LOCK_DEBUG
	if (lockmethod == USER_LOCKMETHOD ? Trace_userlocks : Trace_locks)
		elog(LOG, "LockReleaseAll done");
#endif

	return TRUE;
}

int
LockShmemSize(int maxBackends)
{
	int			size = 0;
	long		max_table_size = NLOCKENTS(maxBackends);

	size += MAXALIGN(sizeof(PROC_HDR)); /* ProcGlobal */
	size += maxBackends * MAXALIGN(sizeof(PGPROC));		/* each MyProc */
	size += MAX_LOCK_METHODS * MAXALIGN(sizeof(LOCKMETHODTABLE));		/* each lockMethodTable */

	/* lockHash table */
	size += hash_estimate_size(max_table_size, sizeof(LOCK));

	/* proclockHash table */
	size += hash_estimate_size(max_table_size, sizeof(PROCLOCK));

	/*
	 * Since the lockHash entry count above is only an estimate, add 10%
	 * safety margin.
	 */
	size += size / 10;

	return size;
}

/*
 * GetLockStatusData - Return a summary of the lock manager's internal
 * status, for use in a user-level reporting function.
 *
 * The return data consists of an array of PROCLOCK objects, with the
 * associated PGPROC and LOCK objects for each.  Note that multiple
 * copies of the same PGPROC and/or LOCK objects are likely to appear.
 * It is the caller's responsibility to match up duplicates if wanted.
 *
 * The design goal is to hold the LockMgrLock for as short a time as possible;
 * thus, this function simply makes a copy of the necessary data and releases
 * the lock, allowing the caller to contemplate and format the data for as
 * long as it pleases.
 */
LockData *
GetLockStatusData(void)
{
	LockData   *data;
	HTAB	   *proclockTable;
	PROCLOCK   *proclock;
	HASH_SEQ_STATUS seqstat;
	int			i;

	data = (LockData *) palloc(sizeof(LockData));

	LWLockAcquire(LockMgrLock, LW_EXCLUSIVE);

	proclockTable = LockMethodTable[DEFAULT_LOCKMETHOD]->proclockHash;

	data->nelements = i = proclockTable->hctl->nentries;

	if (i == 0)
		i = 1;					/* avoid palloc(0) if empty table */

	data->proclockaddrs = (SHMEM_OFFSET *) palloc(sizeof(SHMEM_OFFSET) * i);
	data->proclocks = (PROCLOCK *) palloc(sizeof(PROCLOCK) * i);
	data->procs = (PGPROC *) palloc(sizeof(PGPROC) * i);
	data->locks = (LOCK *) palloc(sizeof(LOCK) * i);

	hash_seq_init(&seqstat, proclockTable);

	i = 0;
	while ((proclock = hash_seq_search(&seqstat)))
	{
		PGPROC	   *proc = (PGPROC *) MAKE_PTR(proclock->tag.proc);
		LOCK	   *lock = (LOCK *) MAKE_PTR(proclock->tag.lock);

		data->proclockaddrs[i] = MAKE_OFFSET(proclock);
		memcpy(&(data->proclocks[i]), proclock, sizeof(PROCLOCK));
		memcpy(&(data->procs[i]), proc, sizeof(PGPROC));
		memcpy(&(data->locks[i]), lock, sizeof(LOCK));

		i++;
	}

	LWLockRelease(LockMgrLock);

	Assert(i == data->nelements);

	return data;
}

/* Provide the textual name of any lock mode */
const char *
GetLockmodeName(LOCKMODE mode)
{
	Assert(mode <= MAX_LOCKMODES);
	return lock_mode_names[mode];
}

#ifdef LOCK_DEBUG
/*
 * Dump all locks in the proc->procHolders list.
 *
 * Must have already acquired the masterLock.
 */
void
DumpLocks(void)
{
	PGPROC	   *proc;
	SHM_QUEUE  *procHolders;
	PROCLOCK   *proclock;
	LOCK	   *lock;
	int			lockmethod = DEFAULT_LOCKMETHOD;
	LOCKMETHODTABLE *lockMethodTable;

	proc = MyProc;
	if (proc == NULL)
		return;

	procHolders = &proc->procHolders;

	Assert(lockmethod < NumLockMethods);
	lockMethodTable = LockMethodTable[lockmethod];
	if (!lockMethodTable)
		return;

	if (proc->waitLock)
		LOCK_PRINT("DumpLocks: waiting on", proc->waitLock, 0);

	proclock = (PROCLOCK *) SHMQueueNext(procHolders, procHolders,
										 offsetof(PROCLOCK, procLink));

	while (proclock)
	{
		Assert(proclock->tag.proc == MAKE_OFFSET(proc));

		lock = (LOCK *) MAKE_PTR(proclock->tag.lock);

		PROCLOCK_PRINT("DumpLocks", proclock);
		LOCK_PRINT("DumpLocks", lock, 0);

		proclock = (PROCLOCK *) SHMQueueNext(procHolders, &proclock->procLink,
										   offsetof(PROCLOCK, procLink));
	}
}

/*
 * Dump all postgres locks. Must have already acquired the masterLock.
 */
void
DumpAllLocks(void)
{
	PGPROC	   *proc;
	PROCLOCK   *proclock;
	LOCK	   *lock;
	int			lockmethod = DEFAULT_LOCKMETHOD;
	LOCKMETHODTABLE *lockMethodTable;
	HTAB	   *proclockTable;
	HASH_SEQ_STATUS status;

	proc = MyProc;
	if (proc == NULL)
		return;

	Assert(lockmethod < NumLockMethods);
	lockMethodTable = LockMethodTable[lockmethod];
	if (!lockMethodTable)
		return;

	proclockTable = lockMethodTable->proclockHash;

	if (proc->waitLock)
		LOCK_PRINT("DumpAllLocks: waiting on", proc->waitLock, 0);

	hash_seq_init(&status, proclockTable);
	while ((proclock = (PROCLOCK *) hash_seq_search(&status)) != NULL)
	{
		PROCLOCK_PRINT("DumpAllLocks", proclock);

		if (proclock->tag.lock)
		{
			lock = (LOCK *) MAKE_PTR(proclock->tag.lock);
			LOCK_PRINT("DumpAllLocks", lock, 0);
		}
		else
			elog(LOG, "DumpAllLocks: proclock->tag.lock = NULL");
	}
}

#endif   /* LOCK_DEBUG */