lock.c

关系型数据库 Postgresql 6.5.2

	 */
	Assert(lockMethodTable->lockHash->hash == tag_hash);
	lock = (LOCK *) hash_search(lockMethodTable->lockHash, (Pointer) locktag,
								HASH_ENTER, &found);
	if (!lock)
	{
		SpinRelease(masterLock);
		elog(FATAL, "LockAcquire: lock table %d is corrupted", lockmethod);
		return FALSE;
	}

	/* --------------------
	 * if there was nothing else there, complete initialization
	 * --------------------
	 */
	if (!found)
	{
		lock->mask = 0;
		lock->nHolding = 0;
		lock->nActive = 0;
		MemSet((char *) lock->holders, 0, sizeof(int) * MAX_LOCKMODES);
		MemSet((char *) lock->activeHolders, 0, sizeof(int) * MAX_LOCKMODES);
		ProcQueueInit(&(lock->waitProcs));
		Assert(lock->tag.objId.blkno == locktag->objId.blkno);
		LOCK_PRINT("LockAcquire: new", lock, lockmode);
	}
	else
	{
		LOCK_PRINT("LockAcquire: found", lock, lockmode);
		Assert((lock->nHolding > 0) && (lock->holders[lockmode] >= 0));
		Assert((lock->nActive > 0) && (lock->activeHolders[lockmode] >= 0));
		Assert(lock->nActive <= lock->nHolding);
	}

	/* ------------------
	 * add an element to the lock queue so that we can clear the
	 * locks at end of transaction.
	 * ------------------
	 */
	xidTable = lockMethodTable->xidHash;

	/* ------------------
	 * 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);		/* must clear padding, needed */
	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 or create an xid entry with this tag
	 */
	result = (XIDLookupEnt *) hash_search(xidTable, (Pointer) &item,
										  HASH_ENTER, &found);
	if (!result)
	{
		elog(NOTICE, "LockAcquire: xid table corrupted");
		return STATUS_ERROR;
	}

	/*
	 * If not found initialize the new entry
	 */
	if (!found)
	{
		result->nHolding = 0;
		MemSet((char *) result->holders, 0, sizeof(int) * MAX_LOCKMODES);
		ProcAddLock(&result->queue);
		XID_PRINT("LockAcquire: new", result);
	}
	else
	{
		XID_PRINT("LockAcquire: found", result);
		Assert((result->nHolding > 0) && (result->holders[lockmode] >= 0));
		Assert(result->nHolding <= lock->nActive);
	}

	/* ----------------
	 * lock->nholding tells us how many processes have _tried_ to
	 * acquire this lock,  Regardless of whether they succeeded or
	 * failed in doing so.
	 * ----------------
	 */
	lock->nHolding++;
	lock->holders[lockmode]++;
	Assert((lock->nHolding > 0) && (lock->holders[lockmode] > 0));

	/* --------------------
	 * If I'm the only one holding a lock, then there
	 * cannot be a conflict. The same is true if we already
	 * hold this lock.
	 * --------------------
	 */
	if (result->nHolding == lock->nActive || result->holders[lockmode] != 0)
	{
		result->holders[lockmode]++;
		result->nHolding++;
		XID_PRINT("LockAcquire: owning", result);
		Assert((result->nHolding > 0) && (result->holders[lockmode] > 0));
		GrantLock(lock, lockmode);
		SpinRelease(masterLock);
		return TRUE;
	}

	/*
	 * If lock requested conflicts with locks requested by waiters...
	 */
	if (lockMethodTable->ctl->conflictTab[lockmode] & lock->waitMask)
	{
		int			i = 1;

		/*
		 * If I don't hold locks or my locks don't conflict with waiters
		 * then force to sleep.
		 */
		if (result->nHolding > 0)
		{
			for (; i <= lockMethodTable->ctl->numLockModes; i++)
			{
				if (result->holders[i] > 0 &&
					lockMethodTable->ctl->conflictTab[i] & lock->waitMask)
					break;		/* conflict */
			}
		}

		if (result->nHolding == 0 || i > lockMethodTable->ctl->numLockModes)
		{
			XID_PRINT("LockAcquire: higher priority proc waiting",
					  result);
			status = STATUS_FOUND;
		}
		else
			status = LockResolveConflicts(lockmethod, lock, lockmode, xid, result);
	}
	else
		status = LockResolveConflicts(lockmethod, lock, lockmode, xid, result);

	if (status == STATUS_OK)
		GrantLock(lock, lockmode);
	else if (status == STATUS_FOUND)
	{
#ifdef USER_LOCKS

		/*
		 * User locks are non blocking. If we can't acquire a lock we must
		 * remove the xid entry and return FALSE without waiting.
		 */
		if (is_user_lock)
		{
			if (!result->nHolding)
			{
				SHMQueueDelete(&result->queue);
				result = (XIDLookupEnt *) hash_search(xidTable,
													  (Pointer) result,
													HASH_REMOVE, &found);
				if (!result || !found)
					elog(NOTICE, "LockAcquire: remove xid, table corrupted");
			}
			else
				XID_PRINT_AUX("LockAcquire: NHOLDING", result);
			lock->nHolding--;
			lock->holders[lockmode]--;
			LOCK_PRINT("LockAcquire: user lock failed", lock, lockmode);
			Assert((lock->nHolding > 0) && (lock->holders[lockmode] >= 0));
			Assert(lock->nActive <= lock->nHolding);
			SpinRelease(masterLock);
			return FALSE;
		}
#endif

		/*
		 * Construct bitmask of locks we hold before going to sleep.
		 */
		MyProc->holdLock = 0;
		if (result->nHolding > 0)
		{
			int			i,
						tmpMask = 2;

			for (i = 1; i <= lockMethodTable->ctl->numLockModes;
				 i++, tmpMask <<= 1)
			{
				if (result->holders[i] > 0)
					MyProc->holdLock |= tmpMask;
			}
			Assert(MyProc->holdLock != 0);
		}

		status = WaitOnLock(lockmethod, lock, lockmode);

		/*
		 * Check the xid entry status, in case something in the ipc
		 * communication doesn't work correctly.
		 */
		if (!((result->nHolding > 0) && (result->holders[lockmode] > 0)))
		{
			XID_PRINT_AUX("LockAcquire: INCONSISTENT ", result);
			LOCK_PRINT_AUX("LockAcquire: INCONSISTENT ", lock, lockmode);
			/* Should we retry ? */
			return FALSE;
		}
		XID_PRINT("LockAcquire: granted", result);
		LOCK_PRINT("LockAcquire: granted", lock, lockmode);
	}

	SpinRelease(masterLock);

	return status == STATUS_OK;
}

/* ----------------------------
 * LockResolveConflicts -- test for lock conflicts
 *
 * NOTES:
 *		Here's what makes this complicated: one transaction's
 * locks don't conflict with one another.  When many processes
 * hold locks, each has to subtract off the other's locks when
 * determining whether or not any new lock acquired conflicts with
 * the old ones.
 *
 * ----------------------------
 */
int
LockResolveConflicts(LOCKMETHOD lockmethod,
					 LOCK *lock,
					 LOCKMODE lockmode,
					 TransactionId xid,
					 XIDLookupEnt *xidentP)		/* xident ptr or NULL */
{
	XIDLookupEnt *result,
				item;
	int		   *myHolders;
	int			numLockModes;
	HTAB	   *xidTable;
	bool		found;
	int			bitmask;
	int			i,
				tmpMask;

#ifdef USER_LOCKS
	int			is_user_lock;

#endif

	numLockModes = LockMethodTable[lockmethod]->ctl->numLockModes;
	xidTable = LockMethodTable[lockmethod]->xidHash;

	if (xidentP)
	{

		/*
		 * A pointer to the xid entry was supplied from the caller.
		 * Actually only LockAcquire can do it.
		 */
		result = xidentP;
	}
	else
	{
		/* ---------------------
		 * read my own statistics from the xid table.  If there
		 * isn't an entry, then we'll just add one.
		 *
		 * Zero out the tag, 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
		is_user_lock = (lockmethod == 2);
		if (is_user_lock)
		{
			item.tag.pid = MyProcPid;
			item.tag.xid = 0;
		}
		else
			TransactionIdStore(xid, &item.tag.xid);
#else
		TransactionIdStore(xid, &item.tag.xid);
#endif

		/*
		 * Find or create an xid entry with this tag
		 */
		result = (XIDLookupEnt *) hash_search(xidTable, (Pointer) &item,
											  HASH_ENTER, &found);
		if (!result)
		{
			elog(NOTICE, "LockResolveConflicts: xid table corrupted");
			return STATUS_ERROR;
		}

		/*
		 * If not found initialize the new entry. THIS SHOULD NEVER
		 * HAPPEN, if we are trying to resolve a conflict we must already
		 * have allocated an xid entry for this lock.	 dz 21-11-1997
		 */
		if (!found)
		{
			/* ---------------
			 * we're not holding any type of lock yet.  Clear
			 * the lock stats.
			 * ---------------
			 */
			MemSet(result->holders, 0, numLockModes * sizeof(*(lock->holders)));
			result->nHolding = 0;
			XID_PRINT_AUX("LockResolveConflicts: NOT FOUND", result);
		}
		else
			XID_PRINT("LockResolveConflicts: found", result);
	}
	Assert((result->nHolding >= 0) && (result->holders[lockmode] >= 0));

	/* ----------------------------
	 * first check for global conflicts: If no locks conflict
	 * with mine, then I get the lock.
	 *
	 * Checking for conflict: lock->mask represents the types of
	 * currently held locks.  conflictTable[lockmode] has a bit
	 * set for each type of lock that conflicts with mine.	Bitwise
	 * compare tells if there is a conflict.
	 * ----------------------------
	 */
	if (!(LockMethodTable[lockmethod]->ctl->conflictTab[lockmode] & lock->mask))
	{
		result->holders[lockmode]++;
		result->nHolding++;
		XID_PRINT("LockResolveConflicts: no conflict", result);
		Assert((result->nHolding > 0) && (result->holders[lockmode] > 0));
		return STATUS_OK;
	}

	/* ------------------------
	 * Rats.  Something conflicts. But it could still be my own
	 * lock.  We have to construct a conflict mask
	 * that does not reflect our own locks.
	 * ------------------------
	 */
	myHolders = result->holders;
	bitmask = 0;
	tmpMask = 2;
	for (i = 1; i <= numLockModes; i++, tmpMask <<= 1)
	{
		if (lock->activeHolders[i] != myHolders[i])
			bitmask |= tmpMask;
	}

	/* ------------------------
	 * now check again for conflicts.  'bitmask' describes the types
	 * of locks held by other processes.  If one of these
	 * conflicts with the kind of lock that I want, there is a
	 * conflict and I have to sleep.
	 * ------------------------
	 */
	if (!(LockMethodTable[lockmethod]->ctl->conflictTab[lockmode] & bitmask))
	{
		/* no conflict. Get the lock and go on */
		result->holders[lockmode]++;
		result->nHolding++;
		XID_PRINT("LockResolveConflicts: resolved", result);
		Assert((result->nHolding > 0) && (result->holders[lockmode] > 0));
		return STATUS_OK;
	}

	XID_PRINT("LockResolveConflicts: conflicting", result);
	return STATUS_FOUND;
}

/*
 * GrantLock -- update the lock data structure to show
 *		the new lock holder.
 */
void
GrantLock(LOCK *lock, LOCKMODE lockmode)
{
	lock->nActive++;
	lock->activeHolders[lockmode]++;
	lock->mask |= BITS_ON[lockmode];
	LOCK_PRINT("GrantLock", lock, lockmode);
	Assert((lock->nActive > 0) && (lock->activeHolders[lockmode] > 0));
	Assert(lock->nActive <= lock->nHolding);
}

static int
WaitOnLock(LOCKMETHOD lockmethod, LOCK *lock, LOCKMODE lockmode)
{
	PROC_QUEUE *waitQueue = &(lock->waitProcs);
	LOCKMETHODTABLE *lockMethodTable = LockMethodTable[lockmethod];
	char		old_status[64],
				new_status[64];

	Assert(lockmethod < NumLockMethods);

	/*
	 * the waitqueue is ordered by priority. I insert myself according to
	 * the priority of the lock I am acquiring.
	 *
	 * SYNC NOTE: I am assuming that the lock table spinlock is sufficient
	 * synchronization for this queue.	That will not be true if/when
	 * people can be deleted from the queue by a SIGINT or something.
	 */
	LOCK_PRINT_AUX("WaitOnLock: sleeping on lock", lock, lockmode);
	strcpy(old_status, PS_STATUS);
	strcpy(new_status, PS_STATUS);
	strcat(new_status, " waiting");
	PS_SET_STATUS(new_status);
	if (ProcSleep(waitQueue,
				  lockMethodTable->ctl,
				  lockmode,
				  lock) != NO_ERROR)
	{
		/* -------------------
		 * This could have happend as a result of a deadlock,
		 * see HandleDeadLock().
		 * Decrement the lock nHolding and holders fields as
		 * we are no longer waiting on this lock.
		 * -------------------
		 */
		lock->nHolding--;
		lock->holders[lockmode]--;
		LOCK_PRINT_AUX("WaitOnLock: aborting on lock", lock, lockmode);
		Assert((lock->nHolding >= 0) && (lock->holders[lockmode] >= 0));
		Assert(lock->nActive <= lock->nHolding);
		if (lock->activeHolders[lockmode] == lock->holders[lockmode])
			lock->waitMask &= BITS_OFF[lockmode];
		SpinRelease(lockMethodTable->ctl->masterLock);
		elog(ERROR, "WaitOnLock: error on wakeup - Aborting this transaction");

		/* not reached */
	}

	if (lock->activeHolders[lockmode] == lock->holders[lockmode])
		lock->waitMask &= BITS_OFF[lockmode];
	PS_SET_STATUS(old_status);
	LOCK_PRINT_AUX("WaitOnLock: wakeup on lock", lock, lockmode);
	return STATUS_OK;
}

/*
 * LockRelease -- look up 'locktag' in lock table 'lockmethod' and
 *		release it.
 *
 * Side Effects: if the lock no longer conflicts with the highest
 *		priority waiting process, that process is granted the lock
 *		and awoken. (We have to grant the lock here to avoid a
 *		race between the waking process and any new process to
 *		come along and request the lock).
 */
bool
LockRelease(LOCKMETHOD lockmethod, LOCKTAG *locktag, LOCKMODE lockmode)
{
	LOCK	   *lock = NULL;
	SPINLOCK	masterLock;
	bool		found;
	LOCKMETHODTABLE *lockMethodTable;
	XIDLookupEnt *result,
				item;
	HTAB	   *xidTable;
	TransactionId xid;
	bool		wakeupNeeded = true;
	int			trace_flag;

#ifdef USER_LOCKS
	int			is_user_lock;

	is_user_lock = (lockmethod == USER_LOCKMETHOD);
	if (is_user_lock)
	{
		TPRINTF(TRACE_USERLOCKS, "LockRelease: user lock tag [%u] %d",
				locktag->objId.blkno,
				lockmode);
	}
#endif

	/* ???????? This must be changed when short term locks will be used */
	locktag->lockmethod = lockmethod;

#ifdef USER_LOCKS
	trace_flag = \
		(lockmethod == USER_LOCKMETHOD) ? TRACE_USERLOCKS : TRACE_LOCKS;
#else
	trace_flag = TRACE_LOCKS;
#endif

	Assert(lockmethod < NumLockMethods);
	lockMethodTable = LockMethodTable[lockmethod];
	if (!lockMethodTable)
	{
		elog(NOTICE, "lockMethodTable is null in LockRelease");
		return FALSE;
	}

	if (LockingIsDisabled)
		return TRUE;

	masterLock = lockMethodTable->ctl->masterLock;
	SpinAcquire(masterLock);

	/*
	 * Find a lock with this tag
	 */
	Assert(lockMethodTable->lockHash->hash == tag_hash);
	lock = (LOCK *) hash_search(lockMethodTable->lockHash, (Pointer) locktag,
								HASH_FIND, &found);

	/*
	 * let the caller print its own error message, too. Do not
	 * elog(ERROR).
	 */
	if (!lock)
	{
		SpinRelease(masterLock);