XLogInsert_rdt2_v1.patch

Index: xlog.c
===================================================================
RCS file: /home/jjanes/pgrepo/pgsql/src/backend/access/transam/xlog.c,v
retrieving revision 1.352
diff -c -r1.352 xlog.c
*** xlog.c	10 Sep 2009 09:42:10 -0000	1.352
--- xlog.c	10 Sep 2009 19:27:08 -0000
***************
*** 540,548 ****
  	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];
  	pg_crc32	rdata_crc;
  	uint32		len,
  				write_len;
--- 540,550 ----
  	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];	/*xlog header of backed up block*/
! 	XLogRecData dtbuf_rdt2[XLR_MAX_BKP_BLOCKS];	/*part of block before the hole*/
! 	XLogRecData dtbuf_rdt3[XLR_MAX_BKP_BLOCKS];	/*part of block after the hole*/
! 	XLogRecData dummy_node;	/* head node for back-up block chain*/
! 	XLogRecData *rdt2;	/* tail pointer for back-up block chain*/
  	pg_crc32	rdata_crc;
  	uint32		len,
  				write_len;
***************
*** 663,696 ****
  
  	/*
  	 * 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));
! 			}
  		}
  	}
  
--- 665,740 ----
  
  	/*
  	 * Now add the backup block headers and data into the CRC
+ 	 * Also make a separate chain of entries for the backup blocks.  
+ 	 * Once we know we do not need to repeat the process due to races,
+ 	 * the two chains are stitched together so that we  don't need 
+ 	 * to special-case them in the write loop.  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.
+ 	 * Doing this here is safe, for the same reason setting rdt->data = NULL
+ 	 * is safe
  	 */
+ 	rdt2 = &dummy_node;
+ 	dummy_node.next=NULL;
+ 	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]);
! 
! 		COMP_CRC32(rdata_crc,
! 				   (char *) bkpb,
! 				   sizeof(BkpBlock));
! 		rdt2->next = &(dtbuf_rdt1[i]);
! 		rdt2 = rdt2->next;
! 
! 		rdt2->data = (char *) bkpb;
! 		rdt2->len = sizeof(BkpBlock);
! 		write_len += sizeof(BkpBlock);
! 
! 		rdt2->next = &(dtbuf_rdt2[i]);
! 		rdt2 = rdt2->next;
! 
! 		page = (char *) BufferGetBlock(dtbuf[i]);
! 
! 		if (bkpb->hole_length == 0)
! 		{
  				COMP_CRC32(rdata_crc,
! 					   page,
! 					   BLCKSZ);
! 			rdt2->data = page;
! 			rdt2->len = BLCKSZ;
! 			write_len += BLCKSZ;
! 			rdt2->next = NULL;
! 		}
! 		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));
! 			rdt2->data = page;
! 			rdt2->len = bkpb->hole_offset;
! 			write_len += bkpb->hole_offset;
! 
! 			rdt2->next = &(dtbuf_rdt3[i]);
! 			rdt2 = rdt2->next;
! 
! 			rdt2->data = page + (bkpb->hole_offset + bkpb->hole_length);
! 			rdt2->len = BLCKSZ - (bkpb->hole_offset + bkpb->hole_length);
! 			write_len += rdt2->len;
! 			rdt2->next = NULL;
  		}
  	}
  
***************
*** 758,820 ****
  		goto 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 we backed up any full blocks and online backup is not in progress,
--- 802,809 ----
  		goto begin;
  	}
  
! 	/* Stitch the two chains together.  We have now irrevocably changed the rdata chain */
! 	rdt->next = dummy_node.next;
  
  	/*
  	 * If we backed up any full blocks and online backup is not in progress,