轻量级锁实现2——上锁、释放

RequestNamedLWLockTranche("pg_stat_statements", 1);
/*
 * RequestNamedLWLockTranche
		Request that extra LWLocks be allocated during postmaster
		startup.
 *
 * This may only be called via the shmem_request_hook of a library that is
 * loaded into the postmaster via shared_preload_libraries.  Calls from
 * elsewhere will fail.
 *
 * The tranche name will be user-visible as a wait event name, so try to
 * use a name that fits the style for those.
 */

void RequestNamedLWLockTranche(const char *tranche_name, int num_lwlocks)
{
	NamedLWLockTrancheRequest *request;

	if (!process_shmem_requests_in_progress)
		elog(FATAL, "cannot request additional LWLocks outside shmem_request_hook");

	if (NamedLWLockTrancheRequestArray == NULL)
	{
		NamedLWLockTrancheRequestsAllocated = 16;
		NamedLWLockTrancheRequestArray = (NamedLWLockTrancheRequest *)
			MemoryContextAlloc(TopMemoryContext,


	}

	if (NamedLWLockTrancheRequests >= NamedLWLockTrancheRequestsAllocated)
	{
		int			i = pg_nextpower2_32(NamedLWLockTrancheRequests + 1);

		NamedLWLockTrancheRequestArray = (NamedLWLockTrancheRequest *)
			repalloc(NamedLWLockTrancheRequestArray,
					 i * sizeof(NamedLWLockTrancheRequest));
		NamedLWLockTrancheRequestsAllocated = i;
	}

	request = &NamedLWLockTrancheRequestArray[NamedLWLockTrancheRequests];
	Assert(strlen(tranche_name) + 1 <= NAMEDATALEN);
	strlcpy(request->tranche_name, tranche_name, NAMEDATALEN);
	request->num_lwlocks = num_lwlocks;
	NamedLWLockTrancheRequests++;
}

LWLockAcquire(AddinShmemInitLock, LW_EXCLUSIVE);

	pgss = ShmemInitStruct("pg_stat_statements",



	if (!found)
	{
		/* First time through ... */
		pgss->lock = &(GetNamedLWLockTranche("pg_stat_statements"))->lock;
		pgss->cur_median_usage = ASSUMED_MEDIAN_INIT;
		pgss->mean_query_len = ASSUMED_LENGTH_INIT;
		SpinLockInit(&pgss->mutex);
		pgss->extent = 0;
		pgss->n_writers = 0;
		pgss->gc_count = 0;
		pgss->stats.dealloc = 0;
		pgss->stats.stats_reset = GetCurrentTimestamp();
	}

	info.keysize = sizeof(pgssHashKey);
	info.entrysize = sizeof(pgssEntry);
	pgss_hash = ShmemInitHash("pg_stat_statements hash",




LWLockRelease(AddinShmemInitLock);

/*
 * LWLockAcquire - acquire a lightweight lock in the specified mode
 *
 * If the lock is not available, sleep until it is.  Returns true if the lock
 * was available immediately, false if we had to sleep.
 *
 * Side effect: cancel/die interrupts are held off until lock release.
 */
bool LWLockAcquire(LWLock *lock, LWLockMode mode)；

PGPROC	
	bool	
	int	

	Assert(mode == LW_SHARED || mode == LW_EXCLUSIVE);

	/*
	 * We can't wait if we haven't got a PGPROC.  This should only occur
	 * during bootstrap or shared memory initialization.  Put an Assert here
	 * to catch unsafe coding practices.
	 */
	Assert(!(proc == NULL && IsUnderPostmaster));

	/* Ensure we will have room to remember the lock */
	if (num_held_lwlocks >= MAX_SIMUL_LWLOCKS)
		elog(ERROR, "too many LWLocks taken");

void InitPostmasterChild(void)
{
	IsUnderPostmaster = true;	/* we are a postmaster subprocess now */
     ......
     
}

void InitProcessGlobals(void)
{
	MyProcPid = getpid();
     ......
}

void InitProcess(void)
{
    ....
    
	MyProc->pid = MyProcPid;
	/* backendId, databaseId and roleId will be filled in later */
	MyProc->backendId = InvalidBackendId;
	MyProc->databaseId = InvalidOid;
	MyProc->roleId = InvalidOid;
   .....
}

/*
* Lock out cancel/die interrupts until we exit the code section protected
* by the LWLock.  This ensures that interrupts will not interfere with
* manipulations of data structures in shared memory.
*/
HOLD_INTERRUPTS();

#define HOLD_INTERRUPTS()  (InterruptHoldoffCount++)

volatile uint32 InterruptHoldoffCount = 0;

/*
 * LWLockAcquire - acquire a lightweight lock in the specified mode
 *
 * If the lock is not available, sleep until it is.
 *
 * Side effect: cancel/die interrupts are held off until lock release.
 */
void
LWLockAcquire(LWLockId lockid, LWLockMode mode)
{
	LWLock *lock = LWLockArray + lockid;
	bool	mustwait;

	PRINT_LWDEBUG("LWLockAcquire", lockid, lock);

	/*
	 * Lock out cancel/die interrupts until we exit the code section
	 * protected by the LWLock.  This ensures that interrupts will not
	 * interfere with manipulations of data structures in shared memory.
	 */
	HOLD_INTERRUPTS();

	/* Acquire mutex.  Time spent holding mutex should be short! */
	SpinLockAcquire_NoHoldoff(&lock->mutex);

	/* If I can get the lock, do so quickly. */
     // 当请求一个写锁时，需要判断当前是否有其它进程持有写锁或者读锁，如果二者都为0，表示当前可以持写锁，mustwait=false，同时要更改状态：lock->exclusive++变为1
     // 否则，mustwait为true，请求写锁的进程将睡眠。
	if (mode == LW_EXCLUSIVE)
	{
		if (lock->exclusive == 0 && lock->shared == 0)
		{
			lock->exclusive++;
			mustwait = false;
		}
		else
			mustwait = true;
	}
     // 如果当前进程请求的是读锁，有一种情况需要考虑，读锁过多导致写锁的饥饿现象。
     // 所以除了判断写锁是否存在之外，另外一层看当前队列是否为空，如果不为空，即便是读锁也要排队，否则shared在极端情况（基本不可能）可能不会回退到0， 想访问写锁的进程无限期阻塞。
     // 如果某一个进程持有了写锁，mustwait为true。
	else
	{
		/*
		 * If there is someone waiting (presumably for exclusive access),
		 * queue up behind him even though I could get the lock.  This
		 * prevents a stream of read locks from starving a writer.
		 */
		if (lock->exclusive == 0 && lock->head == NULL)
		{
			lock->shared++;
			mustwait = false;
		}
		else
			mustwait = true;
	}
     
     // 根据上面的情形，如果加锁成功并且相应的状态已经改变，只需要进入到else当中释放mutex。
     // 如果mustwait为true，意味着不能持锁，要将当前的进程入lwlock的队列，等待持锁的进程在释放锁时将其唤醒。 
	if (mustwait)
	{
		/* Add myself to wait queue */
		PROC	*proc = MyProc;
		int		extraWaits = 0;

		/*
		 * If we don't have a PROC structure, there's no way to wait.
		 * This should never occur, since MyProc should only be null
		 * during shared memory initialization.
		 */
		if (proc == NULL)
			elog(FATAL, "LWLockAcquire: can't wait without a PROC structure");

		proc->lwWaiting = true;
		proc->lwExclusive = (mode == LW_EXCLUSIVE);
		proc->lwWaitLink = NULL;
		if (lock->head == NULL)
			lock->head = proc;
		else
			lock->tail->lwWaitLink = proc;
		lock->tail = proc;

		/* Can release the mutex now */
		SpinLockRelease_NoHoldoff(&lock->mutex);

		/*
		 * Wait until awakened.
		 *
		 * Since we share the process wait semaphore with the regular lock
		 * manager and ProcWaitForSignal, and we may need to acquire an LWLock
		 * while one of those is pending, it is possible that we get awakened
		 * for a reason other than being granted the LWLock.  If so, loop back
		 * and wait again.  Once we've gotten the lock, re-increment the sema
		 * by the number of additional signals received, so that the lock
		 * manager or signal manager will see the received signal when it
		 * next waits.
		 */
           // 通过system V信号量来保持进程间同步，信号量存于共享内存中，多个进程可申请和释放
           // 因为别的业务也可能释放这个信号量但是被LWLock被捕捉到了，此时需要重复校验proc->lwWaiting的状态。
           // 其实即便使用线程的情况下使用pthread_cond_signal或者pthread_cond_broadcast也存在伪唤醒，都需要对boolean predicate重复校验。
		for (;;)
		{
			/* "false" means cannot accept cancel/die interrupt here. */
			IpcSemaphoreLock(proc->sem.semId, proc->sem.semNum, false);
			if (!proc->lwWaiting)
				break;
			extraWaits++;
		}
           // 利用extraWaits恢复信号量的状态
 被唤醒的时候lwlock的状态已经被修改了，所以除了修正信号量的状态无需再做其它操作。操作系统小知识：进程阻塞进入到就绪(只考虑时间片轮转)
		/*
		 * The awakener already updated the lock struct's state, so we
		 * don't need to do anything more to it.  Just need to fix the
		 * semaphore count.
		 */
		while (extraWaits-- > 0)
			IpcSemaphoreUnlock(proc->sem.semId, proc->sem.semNum);
	}
	else
	{
		/* Got the lock without waiting */
		SpinLockRelease_NoHoldoff(&lock->mutex);
	}

	/* Add lock to list of locks held by this backend */
	Assert(num_held_lwlocks < MAX_SIMUL_LWLOCKS);
	held_lwlocks[num_held_lwlocks++] = lockid;
}

void
LWLockRelease(LWLockId lockid)
{
    .............................

	/* Acquire mutex.  Time spent holding mutex should be short! */
	SpinLockAcquire_NoHoldoff(&lock->mutex);

	/* Release my hold on lock */
	if (lock->exclusive > 0)
		lock->exclusive--;
	else
	{
		Assert(lock->shared > 0);
		lock->shared--;
	}
	/*
	 * See if I need to awaken any waiters.  If I released a non-last shared
	 * hold, there cannot be anything to do.
	 */
	head = lock->head;
	if (head != NULL)
	{
		if (lock->exclusive == 0 && lock->shared == 0)
		{
			/*
			 * Remove the to-be-awakened PROCs from the queue, and update the
			 * lock state to show them as holding the lock.
			 */
			proc = head;
			if (proc->lwExclusive)
			{
				lock->exclusive++;
			}
			else
			{
				lock->shared++;
				while (proc->lwWaitLink != NULL &&

				{
					proc = proc->lwWaitLink;
					lock->shared++;
				}
			}
			/* proc is now the last PROC to be released */
			lock->head = proc->lwWaitLink;
			proc->lwWaitLink = NULL;
		}
		else
		{
			/* lock is still held, can't awaken anything */
			head = NULL;
		}
	}

	/* We are done updating shared state of the lock itself. */
	SpinLockRelease_NoHoldoff(&lock->mutex);

	/*
	 * Awaken any waiters I removed from the queue.
	 */
	while (head != NULL)
	{
		proc = head;
		head = proc->lwWaitLink;
		proc->lwWaitLink = NULL;
		proc->lwWaiting = false;
		IpcSemaphoreUnlock(proc->sem.semId, proc->sem.semNum);
	}
    .......................
}