lock.c

关系型数据库 Postgresql 6.5.2

		elog(NOTICE, "LockRelease: locktable corrupted");
		return FALSE;
	}

	if (!found)
	{
		SpinRelease(masterLock);
#ifdef USER_LOCKS
		if (is_user_lock)
		{
			TPRINTF(TRACE_USERLOCKS, "LockRelease: no lock with this tag");
			return FALSE;
		}
#endif
		elog(NOTICE, "LockRelease: locktable lookup failed, no lock");
		return FALSE;
	}
	LOCK_PRINT("LockRelease: found", lock, lockmode);
	Assert((lock->nHolding > 0) && (lock->holders[lockmode] >= 0));
	Assert((lock->nActive > 0) && (lock->activeHolders[lockmode] >= 0));
	Assert(lock->nActive <= lock->nHolding);


	/* ------------------
	 * Zero out all of the tag bytes (this clears the padding bytes for long
	 * word alignment and ensures hashing consistency).
	 * ------------------
	 */
	MemSet(&item, 0, XID_TAGSIZE);
	item.tag.lock = MAKE_OFFSET(lock);
#ifdef USE_XIDTAG_LOCKMETHOD
	item.tag.lockmethod = lockmethod;
#endif
#ifdef USER_LOCKS
	if (is_user_lock)
	{
		item.tag.pid = MyProcPid;
		item.tag.xid = xid = 0;
	}
	else
	{
		xid = GetCurrentTransactionId();
		TransactionIdStore(xid, &item.tag.xid);
	}
#else
	xid = GetCurrentTransactionId();
	TransactionIdStore(xid, &item.tag.xid);
#endif

	/*
	 * Find an xid entry with this tag
	 */
	xidTable = lockMethodTable->xidHash;
	result = (XIDLookupEnt *) hash_search(xidTable, (Pointer) &item,
										  HASH_FIND_SAVE, &found);
	if (!result || !found)
	{
		SpinRelease(masterLock);
#ifdef USER_LOCKS
		if (!found && is_user_lock)
			TPRINTF(TRACE_USERLOCKS, "LockRelease: no lock with this tag");
		else
#endif
			elog(NOTICE, "LockReplace: xid table corrupted");
		return FALSE;
	}
	XID_PRINT("LockRelease: found", result);
	Assert(result->tag.lock == MAKE_OFFSET(lock));

	/*
	 * Check that we are actually holding a lock of the type we want to
	 * release.
	 */
	if (!(result->holders[lockmode] > 0))
	{
		SpinRelease(masterLock);
		XID_PRINT_AUX("LockAcquire: WRONGTYPE", result);
		elog(NOTICE, "LockRelease: you don't own a lock of type %s",
			 lock_types[lockmode]);
		Assert(result->holders[lockmode] >= 0);
		return FALSE;
	}
	Assert(result->nHolding > 0);

	/*
	 * fix the general lock stats
	 */
	lock->nHolding--;
	lock->holders[lockmode]--;
	lock->nActive--;
	lock->activeHolders[lockmode]--;

#ifdef NOT_USED
	/* --------------------------
	 * If there are still active locks of the type I just released, no one
	 * should be woken up.	Whoever is asleep will still conflict
	 * with the remaining locks.
	 * --------------------------
	 */
	if (lock->activeHolders[lockmode])
		wakeupNeeded = false;
	else
#endif

		/*
		 * Above is not valid any more (due to MVCC lock modes). Actually
		 * we should compare activeHolders[lockmode] with number of
		 * waiters holding lock of this type and try to wakeup only if
		 * these numbers are equal (and lock released conflicts with locks
		 * requested by waiters). For the moment we only check the last
		 * condition.
		 */
	if (lockMethodTable->ctl->conflictTab[lockmode] & lock->waitMask)
		wakeupNeeded = true;

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

	LOCK_PRINT("LockRelease: updated", lock, lockmode);
	Assert((lock->nHolding >= 0) && (lock->holders[lockmode] >= 0));
	Assert((lock->nActive >= 0) && (lock->activeHolders[lockmode] >= 0));
	Assert(lock->nActive <= lock->nHolding);

	if (!lock->nHolding)
	{
		/* ------------------
		 * if there's no one waiting in the queue,
		 * we just released the last lock.
		 * Delete it from the lock table.
		 * ------------------
		 */
		Assert(lockMethodTable->lockHash->hash == tag_hash);
		lock = (LOCK *) hash_search(lockMethodTable->lockHash,
									(Pointer) &(lock->tag),
									HASH_REMOVE,
									&found);
		Assert(lock && found);
		wakeupNeeded = false;
	}

	/*
	 * now check to see if I have any private locks.  If I do, decrement
	 * the counts associated with them.
	 */
	result->holders[lockmode]--;
	result->nHolding--;
	XID_PRINT("LockRelease: updated", result);
	Assert((result->nHolding >= 0) && (result->holders[lockmode] >= 0));

	/*
	 * If this was my last hold on this lock, delete my entry in the XID
	 * table.
	 */
	if (!result->nHolding)
	{
		if (result->queue.prev == INVALID_OFFSET)
			elog(NOTICE, "LockRelease: xid.prev == INVALID_OFFSET");
		if (result->queue.next == INVALID_OFFSET)
			elog(NOTICE, "LockRelease: xid.next == INVALID_OFFSET");
		if (result->queue.next != INVALID_OFFSET)
			SHMQueueDelete(&result->queue);
		XID_PRINT("LockRelease: deleting", result);
		result = (XIDLookupEnt *) hash_search(xidTable, (Pointer) &result,
											  HASH_REMOVE_SAVED, &found);
		if (!result || !found)
		{
			SpinRelease(masterLock);
			elog(NOTICE, "LockRelease: remove xid, table corrupted");
			return FALSE;
		}
	}

	if (wakeupNeeded)
		ProcLockWakeup(&(lock->waitProcs), lockmethod, lock);
	else
	{
		if (((LOCKDEBUG(LOCK_LOCKMETHOD(*(lock))) >= 1) \
			 &&(lock->tag.relId >= lockDebugOidMin)) \
			||\
			(lockDebugRelation && (lock->tag.relId == lockDebugRelation)))
			TPRINTF(TRACE_ALL, "LockRelease: no wakeup needed");
	}

	SpinRelease(masterLock);
	return TRUE;
}

/*
 * LockReleaseAll -- Release all locks in a process lock queue.
 */
bool
LockReleaseAll(LOCKMETHOD lockmethod, SHM_QUEUE *lockQueue)
{
	PROC_QUEUE *waitQueue;
	int			done;
	XIDLookupEnt *xidLook = NULL;
	XIDLookupEnt *tmp = NULL;
	XIDLookupEnt *result;
	SHMEM_OFFSET end = MAKE_OFFSET(lockQueue);
	SPINLOCK	masterLock;
	LOCKMETHODTABLE *lockMethodTable;
	int			i,
				numLockModes;
	LOCK	   *lock;
	bool		found;
	int			trace_flag;
	int			xidtag_lockmethod;

#ifdef USER_LOCKS
	int			is_user_lock_table,
				count,
				nleft;

	count = nleft = 0;

	is_user_lock_table = (lockmethod == USER_LOCKMETHOD);
	trace_flag = (lockmethod == 2) ? TRACE_USERLOCKS : TRACE_LOCKS;
#else
	trace_flag = TRACE_LOCKS;
#endif
	TPRINTF(trace_flag, "LockReleaseAll: lockmethod=%d, pid=%d",
			lockmethod, MyProcPid);

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

	if (SHMQueueEmpty(lockQueue))
		return TRUE;

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

	SpinAcquire(masterLock);
	SHMQueueFirst(lockQueue, (Pointer *) &xidLook, &xidLook->queue);

	for (;;)
	{
		bool		wakeupNeeded = false;

		/*
		 * Sometimes the queue appears to be messed up.
		 */
		if (count++ > 1000)
		{
			elog(NOTICE, "LockReleaseAll: xid loop detected, giving up");
			nleft = 0;
			break;
		}

		/* ---------------------------
		 * XXX Here we assume the shared memory queue is circular and
		 * that we know its internal structure.  Should have some sort of
		 * macros to allow one to walk it.	mer 20 July 1991
		 * ---------------------------
		 */
		done = (xidLook->queue.next == end);
		lock = (LOCK *) MAKE_PTR(xidLook->tag.lock);

		xidtag_lockmethod = XIDENT_LOCKMETHOD(*xidLook);
		if ((xidtag_lockmethod == lockmethod) && pg_options[trace_flag])
		{
			XID_PRINT("LockReleaseAll", xidLook);
			LOCK_PRINT("LockReleaseAll", lock, 0);
		}

#ifdef USE_XIDTAG_LOCKMETHOD
		if (xidtag_lockmethod != LOCK_LOCKMETHOD(*lock))
			elog(NOTICE, "LockReleaseAll: xid/lock method mismatch: %d != %d",
				 xidtag_lockmethod, lock->tag.lockmethod);
#endif
		if ((xidtag_lockmethod != lockmethod) && (trace_flag >= 2))
		{
			TPRINTF(trace_flag, "LockReleaseAll: skipping other table");
			nleft++;
			goto next_item;
		}

		Assert(lock->nHolding > 0);
		Assert(lock->nActive > 0);
		Assert(lock->nActive <= lock->nHolding);
		Assert(xidLook->nHolding >= 0);
		Assert(xidLook->nHolding <= lock->nHolding);

#ifdef USER_LOCKS
		if (is_user_lock_table)
		{
			if ((xidLook->tag.pid == 0) || (xidLook->tag.xid != 0))
			{
				TPRINTF(TRACE_USERLOCKS,
						"LockReleaseAll: skiping normal lock [%d,%d,%d]",
				  xidLook->tag.lock, xidLook->tag.pid, xidLook->tag.xid);
				nleft++;
				goto next_item;
			}
			if (xidLook->tag.pid != MyProcPid)
			{
				/* Should never happen */
				elog(NOTICE,
					 "LockReleaseAll: INVALID PID: [%u] [%d,%d,%d]",
					 lock->tag.objId.blkno,
				  xidLook->tag.lock, xidLook->tag.pid, xidLook->tag.xid);
				nleft++;
				goto next_item;
			}
			TPRINTF(TRACE_USERLOCKS,
					"LockReleaseAll: releasing user lock [%u] [%d,%d,%d]",
					lock->tag.objId.blkno,
				  xidLook->tag.lock, xidLook->tag.pid, xidLook->tag.xid);
		}
		else
		{

			/*
			 * Can't check xidLook->tag.xid, can be 0 also for normal
			 * locks
			 */
			if (xidLook->tag.pid != 0)
			{
				TPRINTF(TRACE_LOCKS,
					 "LockReleaseAll: skiping user lock [%u] [%d,%d,%d]",
						lock->tag.objId.blkno,
				  xidLook->tag.lock, xidLook->tag.pid, xidLook->tag.xid);
				nleft++;
				goto next_item;
			}
		}
#endif

		/* ------------------
		 * fix the general lock stats
		 * ------------------
		 */
		if (lock->nHolding != xidLook->nHolding)
		{
			for (i = 1; i <= numLockModes; i++)
			{
				Assert(xidLook->holders[i] >= 0);
				lock->holders[i] -= xidLook->holders[i];
				lock->activeHolders[i] -= xidLook->holders[i];
				Assert((lock->holders[i] >= 0) \
					   &&(lock->activeHolders[i] >= 0));
				if (!lock->activeHolders[i])
					lock->mask &= BITS_OFF[i];

				/*
				 * Read comments in LockRelease
				 */
				if (!wakeupNeeded && xidLook->holders[i] > 0 &&
					lockMethodTable->ctl->conflictTab[i] & lock->waitMask)
					wakeupNeeded = true;
			}
			lock->nHolding -= xidLook->nHolding;
			lock->nActive -= xidLook->nHolding;
			Assert((lock->nHolding >= 0) && (lock->nActive >= 0));
			Assert(lock->nActive <= lock->nHolding);
		}
		else
		{
			/* --------------
			 * set nHolding to zero so that we can garbage collect the lock
			 * down below...
			 * --------------
			 */
			lock->nHolding = 0;
			/* Fix the lock status, just for next LOCK_PRINT message. */
			for (i = 1; i <= numLockModes; i++)
			{
				Assert(lock->holders[i] == lock->activeHolders[i]);
				lock->holders[i] = lock->activeHolders[i] = 0;
			}
		}
		LOCK_PRINT("LockReleaseAll: updated", lock, 0);

		/*
		 * Remove the xid from the process lock queue
		 */
		SHMQueueDelete(&xidLook->queue);

		/* ----------------
		 * always remove the xidLookup entry, we're done with it now
		 * ----------------
		 */

		XID_PRINT("LockReleaseAll: deleting", xidLook);
		result = (XIDLookupEnt *) hash_search(lockMethodTable->xidHash,
											  (Pointer) xidLook,
											  HASH_REMOVE,
											  &found);
		if (!result || !found)
		{
			SpinRelease(masterLock);
			elog(NOTICE, "LockReleaseAll: xid table corrupted");
			return FALSE;
		}

		if (!lock->nHolding)
		{
			/* --------------------
			 * if there's no one waiting in the queue, we've just released
			 * the last lock.
			 * --------------------
			 */
			LOCK_PRINT("LockReleaseAll: deleting", lock, 0);
			Assert(lockMethodTable->lockHash->hash == tag_hash);
			lock = (LOCK *) hash_search(lockMethodTable->lockHash,
										(Pointer) &(lock->tag),
										HASH_REMOVE, &found);
			if ((!lock) || (!found))
			{
				SpinRelease(masterLock);
				elog(NOTICE, "LockReleaseAll: cannot remove lock from HTAB");
				return FALSE;
			}
		}
		else if (wakeupNeeded)
		{
			waitQueue = &(lock->waitProcs);
			ProcLockWakeup(waitQueue, lockmethod, lock);
		}

#ifdef USER_LOCKS
next_item:
#endif
		if (done)
			break;
		SHMQueueFirst(&xidLook->queue, (Pointer *) &tmp, &tmp->queue);
		xidLook = tmp;
	}

	/*
	 * Reinitialize the queue only if nothing has been left in.
	 */
	if (nleft == 0)
	{
		TPRINTF(trace_flag, "LockReleaseAll: reinitializing lockQueue");
		SHMQueueInit(lockQueue);
	}

	SpinRelease(masterLock);
	TPRINTF(trace_flag, "LockReleaseAll: done");

	return TRUE;
}

int
LockShmemSize(int maxBackends)
{
	int			size = 0;

	size += MAXALIGN(sizeof(PROC_HDR)); /* ProcGlobal */
	size += MAXALIGN(maxBackends * sizeof(PROC));		/* each MyProc */
	size += MAXALIGN(maxBackends * sizeof(LOCKMETHODCTL));		/* each
																 * lockMethodTable->ctl */

	/* lockHash table */
	size += hash_estimate_size(NLOCKENTS(maxBackends),
							   SHMEM_LOCKTAB_KEYSIZE,
							   SHMEM_LOCKTAB_DATASIZE);

	/* xidHash table */
	size += hash_estimate_size(NLOCKENTS(maxBackends),
							   SHMEM_XIDTAB_KEYSIZE,
							   SHMEM_XIDTAB_DATASIZE);

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

	return size;
}

/* -----------------
 * Boolean function to determine current locking status
 * -----------------
 */
bool
LockingDisabled()
{
	return LockingIsDisabled;
}

/*
 * DeadlockCheck -- Checks for deadlocks for a given process
 *
 * We can't block on user locks, so no sense testing for deadlock
 * because there is no blocking, and no timer for the block.
 *
 * This code takes a list of locks a process holds, and the lock that
 * the process is sleeping on, and tries to find if any of the processes
 * waiting on its locks hold the lock it is waiting for.  If no deadlock
 * is found, it goes on to look at all the processes waiting on their locks.
 *
 * We have already locked the master lock before being called.
 */
bool
DeadLockCheck(void *proc, LOCK *findlock)
{
	XIDLookupEnt *xidLook = NULL;
	XIDLookupEnt *tmp = NULL;
	PROC	   *thisProc = (PROC *) proc,
			   *waitProc;
	SHM_QUEUE  *lockQueue = &(thisProc->lockQueue);
	SHMEM_OFFSET end = MAKE_OFFSET(lockQueue);
	LOCK	   *lock;
	PROC_QUEUE *waitQueue;
	int			i,
				j;
	bool		first_run = (thisProc == MyProc),
				done;

	static PROC *checked_procs[MAXBACKENDS];
	static int	nprocs;

	/* initialize at start of recursion */
	if (first_run)
	{
		checked_procs[0] = MyProc;
		nprocs = 1;
	}

	if (SHMQueueEmpty(lockQueue))
		return false;