xloginsert-bkp-refactor-1.patch

*** a/src/backend/access/transam/xlog.c
--- b/src/backend/access/transam/xlog.c
***************
*** 690,695 **** XLogInsert(RmgrId rmid, uint8 info, XLogRecData *rdata)
--- 690,698 ----
  	uint32		freespace;
  	int			curridx;
  	XLogRecData *rdt;
+ 	XLogRecData *rdt_cur;
+ 	XLogRecData	*rdt_bkp_first;
+ 	XLogRecData *rdt_bkp_last;
  	Buffer		dtbuf[XLR_MAX_BKP_BLOCKS];
  	bool		dtbuf_bkp[XLR_MAX_BKP_BLOCKS];
  	BkpBlock	dtbuf_xlg[XLR_MAX_BKP_BLOCKS];
***************
*** 704,709 **** XLogInsert(RmgrId rmid, uint8 info, XLogRecData *rdata)
--- 707,713 ----
  	bool		updrqst;
  	bool		doPageWrites;
  	bool		isLogSwitch = (rmid == RM_XLOG_ID && info == XLOG_SWITCH);
+ 	uint8		info_final;
  
  	/* cross-check on whether we should be here or not */
  	if (!XLogInsertAllowed())
***************
*** 727,738 **** XLogInsert(RmgrId rmid, uint8 info, XLogRecData *rdata)
  	}
  
  	/*
! 	 * Here we scan the rdata chain, determine which buffers must be backed
! 	 * up, and compute the CRC values for the data.  Note that the record
! 	 * header isn't added into the CRC initially since we don't know the final
! 	 * length or info bits quite yet.  Thus, the CRC will represent the CRC of
! 	 * the whole record in the order "rdata, then backup blocks, then record
! 	 * header".
  	 *
  	 * 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 the
--- 731,738 ----
  	}
  
  	/*
! 	 * 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 the
***************
*** 746,751 **** XLogInsert(RmgrId rmid, uint8 info, XLogRecData *rdata)
--- 746,752 ----
  	 * over the chain later.
  	 */
  begin:;
+ 	info_final = info;
  	for (i = 0; i < XLR_MAX_BKP_BLOCKS; i++)
  	{
  		dtbuf[i] = InvalidBuffer;
***************
*** 760,766 **** begin:;
  	 */
  	doPageWrites = fullPageWrites || Insert->forcePageWrites;
  
- 	INIT_CRC32(rdata_crc);
  	len = 0;
  	for (rdt = rdata;;)
  	{
--- 761,766 ----
***************
*** 768,774 **** begin:;
  		{
  			/* Simple data, just include it */
  			len += rdt->len;
- 			COMP_CRC32(rdata_crc, rdt->data, rdt->len);
  		}
  		else
  		{
--- 768,773 ----
***************
*** 779,790 **** begin:;
  				{
  					/* Buffer already referenced by earlier chain item */
  					if (dtbuf_bkp[i])
  						rdt->data = NULL;
  					else if (rdt->data)
- 					{
  						len += rdt->len;
- 						COMP_CRC32(rdata_crc, rdt->data, rdt->len);
- 					}
  					break;
  				}
  				if (dtbuf[i] == InvalidBuffer)
--- 778,789 ----
  				{
  					/* 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)
***************
*** 796,807 **** begin:;
  					{
  						dtbuf_bkp[i] = true;
  						rdt->data = NULL;
  					}
  					else if (rdt->data)
- 					{
  						len += rdt->len;
- 						COMP_CRC32(rdata_crc, rdt->data, rdt->len);
- 					}
  					break;
  				}
  			}
--- 795,804 ----
  					{
  						dtbuf_bkp[i] = true;
  						rdt->data = NULL;
+ 						rdt->len = 0;
  					}
  					else if (rdt->data)
  						len += rdt->len;
  					break;
  				}
  			}
***************
*** 816,854 **** begin:;
  	}
  
  	/*
! 	 * Now add the backup block headers and data into the CRC
  	 */
  	for (i = 0; i < XLR_MAX_BKP_BLOCKS; i++)
  	{
! 		if (dtbuf_bkp[i])
  		{
! 			BkpBlock   *bkpb = &(dtbuf_xlg[i]);
! 			char	   *page;
! 
! 			COMP_CRC32(rdata_crc,
! 					   (char *) bkpb,
! 					   sizeof(BkpBlock));
! 			page = (char *) BufferGetBlock(dtbuf[i]);
! 			if (bkpb->hole_length == 0)
! 			{
! 				COMP_CRC32(rdata_crc,
! 						   page,
! 						   BLCKSZ);
! 			}
! 			else
! 			{
! 				/* must skip the hole */
! 				COMP_CRC32(rdata_crc,
! 						   page,
! 						   bkpb->hole_offset);
! 				COMP_CRC32(rdata_crc,
! 						   page + (bkpb->hole_offset + bkpb->hole_length),
! 						   BLCKSZ - (bkpb->hole_offset + bkpb->hole_length));
! 			}
  		}
  	}
  
  	/*
  	 * 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
--- 813,898 ----
  	}
  
  	/*
! 	 * Make additional rdata chain entries for the backup blocks, so that we
! 	 * don't need to special-case them in the write loop.  They are kept in
! 	 * a separate rdata chain until we have the lock, and know that we won't
! 	 * need to restart from scratch anymore. The real rdata chain is kept
! 	 * unmodified until then.
! 	 *
! 	 * 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 i'th XLR_SET_BKP_BLOCK bit corresponds to the i'th distinct
! 	 * buffer value (ignoring InvalidBuffer) appearing in the rdata chain.
  	 */
+ 	write_len = len;
+ 	rdt_bkp_first = rdt_bkp_last = NULL;
  	for (i = 0; i < XLR_MAX_BKP_BLOCKS; i++)
  	{
! 		BkpBlock   *bkpb;
! 		char	   *page;
! 
! 		if (!dtbuf_bkp[i])
! 			continue;
! 
! 		info_final |= XLR_SET_BKP_BLOCK(i);
! 
! 		bkpb = &(dtbuf_xlg[i]);
! 		page = (char *) BufferGetBlock(dtbuf[i]);
! 
! 		if (rdt_bkp_first == NULL)
! 			rdt_bkp_first = &(dtbuf_rdt1[i]);
! 		else
! 			rdt_bkp_last->next = &(dtbuf_rdt1[i]);
! 
! 		rdt_bkp_last = &(dtbuf_rdt1[i]);
! 
! 		rdt_bkp_last->data = (char *) bkpb;
! 		rdt_bkp_last->len = sizeof(BkpBlock);
! 		write_len += sizeof(BkpBlock);
! 
! 		rdt_bkp_last->next = &(dtbuf_rdt2[i]);
! 		rdt_bkp_last = rdt_bkp_last->next;
! 
! 		if (bkpb->hole_length == 0)
  		{
! 			rdt_bkp_last->data = page;
! 			rdt_bkp_last->len = BLCKSZ;
! 			write_len += BLCKSZ;
! 			rdt_bkp_last->next = NULL;
! 		}
! 		else
! 		{
! 			/* must skip the hole */
! 			rdt_bkp_last->data = page;
! 			rdt_bkp_last->len = bkpb->hole_offset;
! 			write_len += bkpb->hole_offset;
! 
! 			rdt_bkp_last->next = &(dtbuf_rdt3[i]);
! 			rdt_bkp_last = rdt_bkp_last->next;
! 
! 			rdt_bkp_last->data = page + (bkpb->hole_offset + bkpb->hole_length);
! 			rdt_bkp_last->len = BLCKSZ - (bkpb->hole_offset + bkpb->hole_length);
! 			write_len += rdt_bkp_last->len;
! 			rdt_bkp_last->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_cur = rdt; rdt_cur != NULL; rdt_cur = rdt_cur->next)
+ 		COMP_CRC32(rdata_crc, rdt_cur->data, rdt_cur->len);
+ 	for (rdt_cur = rdt_bkp_first; rdt_cur != NULL; rdt_cur = rdt_cur->next)
+ 		COMP_CRC32(rdata_crc, rdt_cur->data, rdt_cur->len);
+ 
+ 	/*
  	 * 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
***************
*** 913,974 **** begin:;
  	}
  
  	/*
! 	 * Make additional rdata chain entries for the backup blocks, so that we
! 	 * don't need to special-case them in the write loop.  Note that we have
! 	 * now irrevocably changed the input rdata chain.  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 i'th XLR_SET_BKP_BLOCK bit corresponds to the i'th distinct
! 	 * buffer value (ignoring InvalidBuffer) appearing in the rdata chain.
  	 */
! 	write_len = len;
! 	for (i = 0; i < XLR_MAX_BKP_BLOCKS; i++)
! 	{
! 		BkpBlock   *bkpb;
! 		char	   *page;
! 
! 		if (!dtbuf_bkp[i])
! 			continue;
! 
! 		info |= XLR_SET_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;
! 		}
! 	}
  
  	/*
  	 * If there isn't enough space on the current XLOG page for a record
--- 957,966 ----
  	}
  
  	/*
! 	 * We can now link the backup block rdata entries into the chain. This
! 	 * changes the rdata chain irrevocably.
  	 */
! 	rdt->next = rdt_bkp_first;
  
  	/*
  	 * If there isn't enough space on the current XLOG page for a record
***************
*** 1031,1037 **** begin:;
  	record->xl_xid = GetCurrentTransactionIdIfAny();
  	record->xl_tot_len = SizeOfXLogRecord + write_len;
  	record->xl_len = len;		/* doesn't include backup blocks */
! 	record->xl_info = info;
  	record->xl_rmid = rmid;
  
  	/* Now we can finish computing the record's CRC */
--- 1023,1029 ----
  	record->xl_xid = GetCurrentTransactionIdIfAny();
  	record->xl_tot_len = SizeOfXLogRecord + write_len;
  	record->xl_len = len;		/* doesn't include backup blocks */
! 	record->xl_info = info_final;
  	record->xl_rmid = rmid;
  
  	/* Now we can finish computing the record's CRC */