XLogInsert(RmgrId rmid, uint8 info, XLogRecData *rdata)
{
	XLogCtlInsert *Insert = &XLogCtl->Insert;
	XLogRecData *rdt;
	XLogRecData *rdt_lastnormal;
	Buffer		dtbuf[XLR_MAX_BKP_BLOCKS];
	bool		dtbuf_bkp[XLR_MAX_BKP_BLOCKS];
	BkpBlock	dtbuf_xlg[XLR_MAX_BKP_BLOCKS];
	XLogRecPtr	dtbuf_lsn[XLR_MAX_BKP_BLOCKS];
	XLogRecData dtbuf_rdt1[XLR_MAX_BKP_BLOCKS];
	XLogRecData dtbuf_rdt2[XLR_MAX_BKP_BLOCKS];
	XLogRecData dtbuf_rdt3[XLR_MAX_BKP_BLOCKS];
	XLogRecData hdr_rdt;
	pg_crc32	rdata_crc;
	uint32		len,
				write_len;
	unsigned	i;
	bool		doPageWrites;
	bool		isLogSwitch = (rmid == RM_XLOG_ID && info == XLOG_SWITCH);
	bool		inserted;
	uint8		info_orig = info;
	static XLogRecord *rechdr;
	XLogRecPtr	StartPos;
	XLogRecPtr	EndPos;

	if (rechdr == NULL)
	{
		rechdr = malloc(SizeOfXLogRecord);
		if (rechdr == NULL)
			elog(ERROR, "out of memory");
		MemSet(rechdr, 0, SizeOfXLogRecord);
	}

	/* cross-check on whether we should be here or not */
	if (!XLogInsertAllowed())
		elog(ERROR, "cannot make new WAL entries during recovery");

	/* info's high bits are reserved for use by me */
	if (info & XLR_INFO_MASK)
		elog(PANIC, "invalid xlog info mask %02X", info);

	TRACE_POSTGRESQL_XLOG_INSERT(rmid, info);

	/*
	 * In bootstrap mode, we don't actually log anything but XLOG resources;
	 * return a phony record pointer.
	 */
	if (IsBootstrapProcessingMode() && rmid != RM_XLOG_ID)
	{
		EndPos = SizeOfXLogLongPHD;		/* start of 1st chkpt record */
		return EndPos;
	}

	/*
	 * Here we scan the rdata chain, to determine which buffers must be backed
	 * up.
	 *
	 * We may have to loop back to here if a race condition is detected below.
	 * We could prevent the race by doing all this work while holding an
	 * insertion slot, but it seems better to avoid doing CRC calculations
	 * while holding one.
	 *
	 * We add entries for backup blocks to the chain, so that they don't need
	 * any special treatment in the critical section where the chunks are
	 * copied into the WAL buffers. Those entries have to be unlinked from the
	 * chain if we have to loop back here.
	 */
begin:;
	for (i = 0; i < XLR_MAX_BKP_BLOCKS; i++)
	{
		dtbuf[i] = InvalidBuffer;
		dtbuf_bkp[i] = false;
	}

	/*
	 * Decide if we need to do full-page writes in this XLOG record: true if
	 * full_page_writes is on or we have a PITR request for it.  Since we
	 * don't yet have an insertion slot, fullPageWrites and forcePageWrites
	 * could change under us, but we'll recheck them once we have a slot.
	 */
	doPageWrites = Insert->fullPageWrites || Insert->forcePageWrites;

	len = 0;
	for (rdt = rdata;;)
	{
		if (rdt->buffer == InvalidBuffer)
		{
			/* Simple data, just include it */
			len += rdt->len;
		}
		else
		{
			/* Find info for buffer */
			for (i = 0; i < XLR_MAX_BKP_BLOCKS; i++)
			{
				if (rdt->buffer == dtbuf[i])
				{
					/* Buffer already referenced by earlier chain item */
					if (dtbuf_bkp[i])
					{
						rdt->data = NULL;
						rdt->len = 0;
					}
					else if (rdt->data)
						len += rdt->len;
					break;
				}
				if (dtbuf[i] == InvalidBuffer)
				{
					/* OK, put it in this slot */
					dtbuf[i] = rdt->buffer;
					if (doPageWrites && XLogCheckBuffer(rdt, true,
										   &(dtbuf_lsn[i]), &(dtbuf_xlg[i])))
					{
						dtbuf_bkp[i] = true;
						rdt->data = NULL;
						rdt->len = 0;
					}
					else if (rdt->data)
						len += rdt->len;
					break;
				}
			}
			if (i >= XLR_MAX_BKP_BLOCKS)
				elog(PANIC, "can backup at most %d blocks per xlog record",
					 XLR_MAX_BKP_BLOCKS);
		}
		/* Break out of loop when rdt points to last chain item */
		if (rdt->next == NULL)
			break;
		rdt = rdt->next;
	}

	/*
	 * NOTE: We disallow len == 0 because it provides a useful bit of extra
	 * error checking in ReadRecord.  This means that all callers of
	 * XLogInsert must supply at least some not-in-a-buffer data.  However, we
	 * make an exception for XLOG SWITCH records because we don't want them to
	 * ever cross a segment boundary.
	 */
	if (len == 0 && !isLogSwitch)
		elog(PANIC, "invalid xlog record length %u", len);

	/*
	 * Make additional rdata chain entries for the backup blocks, so that we
	 * don't need to special-case them in the write loop.  This modifies the
	 * original rdata chain, but we keep a pointer to the last regular entry,
	 * rdt_lastnormal, so that we can undo this if we have to loop back to the
	 * beginning.
	 *
	 * At the exit of this loop, write_len includes the backup block data.
	 *
	 * Also set the appropriate info bits to show which buffers were backed
	 * up. The XLR_BKP_BLOCK(N) bit corresponds to the N'th distinct buffer
	 * value (ignoring InvalidBuffer) appearing in the rdata chain.
	 */
	rdt_lastnormal = rdt;
	write_len = len;
	for (i = 0; i < XLR_MAX_BKP_BLOCKS; i++)
	{
		BkpBlock   *bkpb;
		char	   *page;

		if (!dtbuf_bkp[i])
			continue;

		info |= XLR_BKP_BLOCK(i);

		bkpb = &(dtbuf_xlg[i]);
		page = (char *) BufferGetBlock(dtbuf[i]);

		rdt->next = &(dtbuf_rdt1[i]);
		rdt = rdt->next;

		rdt->data = (char *) bkpb;
		rdt->len = sizeof(BkpBlock);
		write_len += sizeof(BkpBlock);

		rdt->next = &(dtbuf_rdt2[i]);
		rdt = rdt->next;

		if (bkpb->hole_length == 0)
		{
			rdt->data = page;
			rdt->len = BLCKSZ;
			write_len += BLCKSZ;
			rdt->next = NULL;
		}
		else
		{
			/* must skip the hole */
			rdt->data = page;
			rdt->len = bkpb->hole_offset;
			write_len += bkpb->hole_offset;

			rdt->next = &(dtbuf_rdt3[i]);
			rdt = rdt->next;

			rdt->data = page + (bkpb->hole_offset + bkpb->hole_length);
			rdt->len = BLCKSZ - (bkpb->hole_offset + bkpb->hole_length);
			write_len += rdt->len;
			rdt->next = NULL;
		}
	}

	/*
	 * Calculate CRC of the data, including all the backup blocks
	 *
	 * Note that the record header isn't added into the CRC initially since we
	 * don't know the prev-link yet.  Thus, the CRC will represent the CRC of
	 * the whole record in the order: rdata, then backup blocks, then record
	 * header.
	 */
	INIT_CRC32(rdata_crc);
	for (rdt = rdata; rdt != NULL; rdt = rdt->next)
		COMP_CRC32(rdata_crc, rdt->data, rdt->len);

	/*
	 * Construct record header (prev-link is filled in later, after reserving
	 * the space for the record), and make that the first chunk in the chain.
	 *
	 * The CRC calculated for the header here doesn't include prev-link,
	 * because we don't know it yet. It will be added later.
	 */
	rechdr->xl_xid = GetCurrentTransactionIdIfAny();
	rechdr->xl_tot_len = SizeOfXLogRecord + write_len;
	rechdr->xl_len = len;		/* doesn't include backup blocks */
	rechdr->xl_info = info;
	rechdr->xl_rmid = rmid;
	rechdr->xl_prev = InvalidXLogRecPtr;
	COMP_CRC32(rdata_crc, ((char *) rechdr), offsetof(XLogRecord, xl_prev));

	hdr_rdt.next = rdata;
	hdr_rdt.data = (char *) rechdr;
	hdr_rdt.len = SizeOfXLogRecord;
	write_len += SizeOfXLogRecord;

	/*----------
	 *
	 * We have now done all the preparatory work we can without holding a
	 * lock or modifying shared state. From here on, inserting the new WAL
	 * record to the shared WAL buffer cache is a two-step process:
	 *
	 * 1. Reserve the right amount of space from the WAL. The current head of
	 *    reserved space is kept in Insert->CurrBytePos, and is protected by
	 *    insertpos_lck.
	 *
	 * 2. Copy the record to the reserved WAL space. This involves finding the
	 *    correct WAL buffer containing the reserved space, and copying the
	 *    record in place. This can be done concurrently in multiple processes.
	 *
	 * To keep track of which insertions are still in-progress, each concurrent
	 * inserter allocates an "insertion slot", which tells others how far the
	 * inserter has progressed. There is a small fixed number of insertion
	 * slots, determined by the num_xloginsert_slots GUC. When an inserter
	 * finishes, it updates the xlogInsertingAt of its slot to the end of the
	 * record it inserted, to let others know that it's done. xlogInsertingAt
	 * is also updated when crossing over to a new WAL buffer, to allow the
	 * the previous buffer to be flushed.
	 *
	 * Holding onto a slot also protects RedoRecPtr and fullPageWrites from
	 * changing until the insertion is finished.
	 *
	 * Step 2 can usually be done completely in parallel. If the required WAL
	 * page is not initialized yet, you have to grab WALBufMappingLock to
	 * initialize it, but the WAL writer tries to do that ahead of insertions
	 * to avoid that from happening in the critical path.
	 *
	 *----------
	 */
	START_CRIT_SECTION();
	WALInsertSlotAcquire(isLogSwitch);

	/*
	 * Check to see if my RedoRecPtr is out of date.  If so, may have to go
	 * back and recompute everything.  This can only happen just after a
	 * checkpoint, so it's better to be slow in this case and fast otherwise.
	 *
	 * If we aren't doing full-page writes then RedoRecPtr doesn't actually
	 * affect the contents of the XLOG record, so we'll update our local copy
	 * but not force a recomputation.
	 */
	if (RedoRecPtr != Insert->RedoRecPtr)
	{
		Assert(RedoRecPtr < Insert->RedoRecPtr);
		RedoRecPtr = Insert->RedoRecPtr;

		if (doPageWrites)
		{
			for (i = 0; i < XLR_MAX_BKP_BLOCKS; i++)
			{
				if (dtbuf[i] == InvalidBuffer)
					continue;
				if (dtbuf_bkp[i] == false &&
					dtbuf_lsn[i] <= RedoRecPtr)
				{
					/*
					 * Oops, this buffer now needs to be backed up, but we
					 * didn't think so above.  Start over.
					 */
					WALInsertSlotRelease();
					END_CRIT_SECTION();
					rdt_lastnormal->next = NULL;
					info = info_orig;
					goto begin;
				}
			}
		}
	}

	/*
	 * Also check to see if fullPageWrites or forcePageWrites was just turned
	 * on; if we weren't already doing full-page writes then go back and
	 * recompute. (If it was just turned off, we could recompute the record
	 * without full pages, but we choose not to bother.)
	 */
	if ((Insert->fullPageWrites || Insert->forcePageWrites) && !doPageWrites)
	{
		/* Oops, must redo it with full-page data. */
		WALInsertSlotRelease();
		END_CRIT_SECTION();
		rdt_lastnormal->next = NULL;
		info = info_orig;
		goto begin;
	}

	/*
	 * Reserve space for the record in the WAL. This also sets the xl_prev
	 * pointer.
	 */
	if (isLogSwitch)
		inserted = ReserveXLogSwitch(&StartPos, &EndPos, &rechdr->xl_prev);
	else
	{
		ReserveXLogInsertLocation(write_len, &StartPos, &EndPos,
								  &rechdr->xl_prev);
		inserted = true;
	}

	if (inserted)
	{
		/*
		 * Now that xl_prev has been filled in, finish CRC calculation of the
		 * record header.
		 */
		COMP_CRC32(rdata_crc, ((char *) &rechdr->xl_prev), sizeof(XLogRecPtr));
		FIN_CRC32(rdata_crc);
		rechdr->xl_crc = rdata_crc;

		/*
		 * All the record data, including the header, is now ready to be
		 * inserted. Copy the record in the space reserved.
		 */
		CopyXLogRecordToWAL(write_len, isLogSwitch, &hdr_rdt, StartPos, EndPos);
	}
	else
	{
		/*
		 * This was an xlog-switch record, but the current insert location was
		 * already exactly at the beginning of a segment, so there was no need
		 * to do anything.
		 */
	}

	/*
	 * Done! Let others know that we're finished.
	 */
	WALInsertSlotRelease();

	MarkCurrentTransactionIdLoggedIfAny();

	END_CRIT_SECTION();

	/*
	 * Update shared LogwrtRqst.Write, if we crossed page boundary.
	 */
	if (StartPos / XLOG_BLCKSZ != EndPos / XLOG_BLCKSZ)
	{
		/* use volatile pointer to prevent code rearrangement */
		volatile XLogCtlData *xlogctl = XLogCtl;

		SpinLockAcquire(&xlogctl->info_lck);
		/* advance global request to include new block(s) */
		if (xlogctl->LogwrtRqst.Write < EndPos)
			xlogctl->LogwrtRqst.Write = EndPos;
		/* update local result copy while I have the chance */
		LogwrtResult = xlogctl->LogwrtResult;
		SpinLockRelease(&xlogctl->info_lck);
	}

	/*
	 * If this was an XLOG_SWITCH record, flush the record and the empty
	 * padding space that fills the rest of the segment, and perform
	 * end-of-segment actions (eg, notifying archiver).
	 */
	if (isLogSwitch)
	{
		TRACE_POSTGRESQL_XLOG_SWITCH();
		XLogFlush(EndPos);
		/*
		 * Even though we reserved the rest of the segment for us, which is
		 * reflected in EndPos, we return a pointer to just the end of the
		 * xlog-switch record.
		 */
		if (inserted)
		{
			EndPos = StartPos + SizeOfXLogRecord;
			if (StartPos / XLOG_BLCKSZ != EndPos / XLOG_BLCKSZ)
			{
				if (EndPos % XLOG_SEG_SIZE == EndPos % XLOG_BLCKSZ)
					EndPos += SizeOfXLogLongPHD;
				else
					EndPos += SizeOfXLogShortPHD;
			}
		}
	}

#ifdef WAL_DEBUG
	if (XLOG_DEBUG)
	{
		StringInfoData buf;

		initStringInfo(&buf);
		appendStringInfo(&buf, "INSERT @ %X/%X: ",
						 (uint32) (EndPos >> 32), (uint32) EndPos);
		xlog_outrec(&buf, rechdr);
		if (rdata->data != NULL)
		{
			appendStringInfoString(&buf, " - ");
			RmgrTable[rechdr->xl_rmid].rm_desc(&buf, rechdr->xl_info, rdata->data);
		}
		elog(LOG, "%s", buf.data);
		pfree(buf.data);
	}
#endif

	/*
	 * Update our global variables
	 */
	ProcLastRecPtr = StartPos;
	XactLastRecEnd = EndPos;

	return EndPos;
}