WaitXLogInsertionsToFinish(XLogRecPtr upto)
{
	uint64		bytepos;
	XLogRecPtr	reservedUpto;
	XLogRecPtr	finishedUpto;
	volatile XLogCtlInsert *Insert = &XLogCtl->Insert;
	int			i;

	if (MyProc == NULL)
		elog(PANIC, "cannot wait without a PGPROC structure");

	/* Read the current insert position */
	SpinLockAcquire(&Insert->insertpos_lck);
	bytepos = Insert->CurrBytePos;
	SpinLockRelease(&Insert->insertpos_lck);
	reservedUpto = XLogBytePosToEndRecPtr(bytepos);

	/*
	 * No-one should request to flush a piece of WAL that hasn't even been
	 * reserved yet. However, it can happen if there is a block with a bogus
	 * LSN on disk, for example. XLogFlush checks for that situation and
	 * complains, but only after the flush. Here we just assume that to mean
	 * that all WAL that has been reserved needs to be finished. In this
	 * corner-case, the return value can be smaller than 'upto' argument.
	 */
	if (upto > reservedUpto)
	{
		elog(LOG, "request to flush past end of generated WAL; request %X/%X, currpos %X/%X",
			 (uint32) (upto >> 32), (uint32) upto,
			 (uint32) (reservedUpto >> 32), (uint32) reservedUpto);
		upto = reservedUpto;
	}

	/*
	 * finishedUpto is our return value, indicating the point upto which
	 * all the WAL insertions have been finished. Initialize it to the head
	 * of reserved WAL, and as we iterate through the insertion slots, back it
	 * out for any insertion that's still in progress.
	 */
	finishedUpto = reservedUpto;

	/*
	 * Loop through all the slots, sleeping on any in-progress insert older
	 * than 'upto'.
	 */
	for (i = 0; i < num_xloginsert_slots; i++)
	{
		volatile XLogInsertSlot *slot = &XLogCtl->Insert.insertSlots[i].slot;
		XLogRecPtr insertingat;

	retry:
		/*
		 * We can check if the slot is in use without grabbing the spinlock.
		 * The spinlock acquisition of insertpos_lck before this loop acts
		 * as a memory barrier. If someone acquires the slot after that, it
		 * can't possibly be inserting to anything < reservedUpto. If it was
		 * acquired before that, an unlocked test will return true.
		 */
		if (!slot->exclusive)
			continue;

		SpinLockAcquire(&slot->mutex);
		/* re-check now that we have the lock */
		if (!slot->exclusive)
		{
			SpinLockRelease(&slot->mutex);
			continue;
		}
		insertingat = slot->xlogInsertingAt;
		SpinLockRelease(&slot->mutex);

		if (insertingat == InvalidXLogRecPtr)
		{
			/*
			 * slot is reserved just to hold off other inserters, there is no
			 * actual insert in progress.
			 */
			continue;
		}

		/*
		 * This insertion is still in progress. Do we need to wait for it?
		 *
		 * When an inserter acquires a slot, it doesn't reset 'insertingat', so
		 * it will initially point to the old value of some already-finished
		 * insertion. The inserter will update the value as soon as it finishes
		 * the insertion, moves to the next page, or has to do I/O to flush an
		 * old dirty buffer. That means that when we see a slot with
		 * insertingat value < upto, we don't know if that insertion is still
		 * truly in progress, or if the slot is reused by a new inserter that
		 * hasn't updated the insertingat value yet. We have to assume it's the
		 * latter, and wait.
		 */
		if (insertingat < upto)
		{
			WaitOnSlot(slot, insertingat);
			goto retry;
		}
		else
		{
			/*
			 * We don't need to wait for this insertion, but update the
			 * return value.
			 */
			if (insertingat < finishedUpto)
				finishedUpto = insertingat;
		}
	}
	return finishedUpto;
}
WaitXLogInsertionsToFinish(XLogRecPtr upto)
{
	uint64		bytepos;
	XLogRecPtr	reservedUpto;
	XLogRecPtr	finishedUpto;
	volatile XLogCtlInsert *Insert = &XLogCtl->Insert;
	int			i;

	if (MyProc == NULL)
		elog(PANIC, "cannot wait without a PGPROC structure");

	 
	SpinLockAcquire(&Insert->insertpos_lck);
	bytepos = Insert->CurrBytePos;
	SpinLockRelease(&Insert->insertpos_lck);
	reservedUpto = XLogBytePosToEndRecPtr(bytepos);

	 
	if (upto > reservedUpto)
	{
		elog(LOG, "request to flush past end of generated WAL; request %X/%X, currpos %X/%X",
			 (uint32) (upto >> 32), (uint32) upto,
			 (uint32) (reservedUpto >> 32), (uint32) reservedUpto);
		upto = reservedUpto;
	}

	 
	finishedUpto = reservedUpto;

	 
	for (i = 0; i < num_xloginsert_slots; i++)
	{
		volatile XLogInsertSlot *slot = &XLogCtl->Insert.insertSlots[i].slot;
		XLogRecPtr insertingat;

	retry:
		 
		if (!slot->exclusive)
			continue;

		SpinLockAcquire(&slot->mutex);
		 
		if (!slot->exclusive)
		{
			SpinLockRelease(&slot->mutex);
			continue;
		}
		insertingat = slot->xlogInsertingAt;
		SpinLockRelease(&slot->mutex);

		if (insertingat == InvalidXLogRecPtr)
		{
			 
			continue;
		}

		 
		if (insertingat < upto)
		{
			WaitOnSlot(slot, insertingat);
			goto retry;
		}
		else
		{
			 
			if (insertingat < finishedUpto)
				finishedUpto = insertingat;
		}
	}
	return finishedUpto;
}