checkpoint_segments_during_recovery_v3.patch

*** a/src/backend/access/transam/xlog.c
--- b/src/backend/access/transam/xlog.c
***************
*** 508,513 **** static bool reachedMinRecoveryPoint = false;
--- 508,516 ----
  
  static bool InRedo = false;
  
+ /* We've already launched bgwriter to perform restartpoint? */
+ static bool bgwriterLaunched = false;
+ 
  /*
   * Information logged when we detect a change in one of the parameters
   * important for Hot Standby.
***************
*** 550,555 **** static void CheckPointGuts(XLogRecPtr checkPointRedo, int flags);
--- 553,559 ----
  static bool XLogCheckBuffer(XLogRecData *rdata, bool doPageWrites,
  				XLogRecPtr *lsn, BkpBlock *bkpb);
  static bool AdvanceXLInsertBuffer(bool new_segment);
+ static bool XLogCheckpointNeeded(uint32 logid, uint32 logseg);
  static void XLogWrite(XLogwrtRqst WriteRqst, bool flexible, bool xlog_switch);
  static bool InstallXLogFileSegment(uint32 *log, uint32 *seg, char *tmppath,
  					   bool find_free, int *max_advance,
***************
*** 1554,1567 **** AdvanceXLInsertBuffer(bool new_segment)
  /*
   * Check whether we've consumed enough xlog space that a checkpoint is needed.
   *
!  * Caller must have just finished filling the open log file (so that
!  * openLogId/openLogSeg are valid).  We measure the distance from RedoRecPtr
!  * to the open log file and see if that exceeds CheckPointSegments.
   *
   * Note: it is caller's responsibility that RedoRecPtr is up-to-date.
   */
  static bool
! XLogCheckpointNeeded(void)
  {
  	/*
  	 * A straight computation of segment number could overflow 32 bits. Rather
--- 1558,1571 ----
  /*
   * Check whether we've consumed enough xlog space that a checkpoint is needed.
   *
!  * Caller must have just finished filling or reading the log file (so that
!  * the given logid/logseg are valid).  We measure the distance from RedoRecPtr
!  * to the log file and see if that exceeds CheckPointSegments.
   *
   * Note: it is caller's responsibility that RedoRecPtr is up-to-date.
   */
  static bool
! XLogCheckpointNeeded(uint32 logid, uint32 logseg)
  {
  	/*
  	 * A straight computation of segment number could overflow 32 bits. Rather
***************
*** 1577,1584 **** XLogCheckpointNeeded(void)
  	old_segno = (RedoRecPtr.xlogid % XLogSegSize) * XLogSegsPerFile +
  		(RedoRecPtr.xrecoff / XLogSegSize);
  	old_highbits = RedoRecPtr.xlogid / XLogSegSize;
! 	new_segno = (openLogId % XLogSegSize) * XLogSegsPerFile + openLogSeg;
! 	new_highbits = openLogId / XLogSegSize;
  	if (new_highbits != old_highbits ||
  		new_segno >= old_segno + (uint32) (CheckPointSegments - 1))
  		return true;
--- 1581,1588 ----
  	old_segno = (RedoRecPtr.xlogid % XLogSegSize) * XLogSegsPerFile +
  		(RedoRecPtr.xrecoff / XLogSegSize);
  	old_highbits = RedoRecPtr.xlogid / XLogSegSize;
! 	new_segno = (logid % XLogSegSize) * XLogSegsPerFile + logseg;
! 	new_highbits = logid / XLogSegSize;
  	if (new_highbits != old_highbits ||
  		new_segno >= old_segno + (uint32) (CheckPointSegments - 1))
  		return true;
***************
*** 1782,1791 **** XLogWrite(XLogwrtRqst WriteRqst, bool flexible, bool xlog_switch)
  				 * update RedoRecPtr and recheck.
  				 */
  				if (IsUnderPostmaster &&
! 					XLogCheckpointNeeded())
  				{
  					(void) GetRedoRecPtr();
! 					if (XLogCheckpointNeeded())
  						RequestCheckpoint(CHECKPOINT_CAUSE_XLOG);
  				}
  			}
--- 1786,1795 ----
  				 * update RedoRecPtr and recheck.
  				 */
  				if (IsUnderPostmaster &&
! 					XLogCheckpointNeeded(openLogId, openLogSeg))
  				{
  					(void) GetRedoRecPtr();
! 					if (XLogCheckpointNeeded(openLogId, openLogSeg))
  						RequestCheckpoint(CHECKPOINT_CAUSE_XLOG);
  				}
  			}
***************
*** 5641,5647 **** StartupXLOG(void)
  	XLogRecord *record;
  	uint32		freespace;
  	TransactionId oldestActiveXID;
- 	bool		bgwriterLaunched = false;
  
  	/*
  	 * Read control file and check XLOG status looks valid.
--- 5645,5650 ----
***************
*** 7552,7557 **** CreateRestartPoint(int flags)
--- 7555,7570 ----
  		return false;
  	}
  
+ 	/*
+ 	 * Update the shared RedoRecPtr for the startup process to request
+ 	 * a future restartpoint according to checkpoint_segments. We don't
+ 	 * need to hold the insert lock here since there should be no other
+ 	 * processes updating it during recovery.
+ 	 */
+ 	SpinLockAcquire(&xlogctl->info_lck);
+ 	xlogctl->Insert.RedoRecPtr = lastCheckPoint.redo;
+ 	SpinLockRelease(&xlogctl->info_lck);
+ 
  	if (log_checkpoints)
  	{
  		/*
***************
*** 9183,9188 **** XLogPageRead(XLogRecPtr *RecPtr, int emode, bool fetching_ckpt,
--- 9196,9215 ----
  	 */
  	if (readFile >= 0 && !XLByteInSeg(*RecPtr, readId, readSeg))
  	{
+ 		/*
+ 		 * Signal bgwriter to start a restartpoint if we've replayed too
+ 		 * much xlog since the last one.
+ 		 */
+ 		if (StandbyMode && bgwriterLaunched)
+ 		{
+ 			if (XLogCheckpointNeeded(readId, readSeg))
+ 			{
+ 				(void) GetRedoRecPtr();
+ 				if (XLogCheckpointNeeded(readId, readSeg))
+ 					RequestCheckpoint(CHECKPOINT_CAUSE_XLOG);
+ 			}
+ 		}
+ 
  		close(readFile);
  		readFile = -1;
  		readSource = 0;
*** a/src/backend/replication/walreceiver.c
--- b/src/backend/replication/walreceiver.c
***************
*** 505,517 **** XLogWalRcvWrite(char *buf, Size nbytes, XLogRecPtr recptr)
  		buf += byteswritten;
  
  		LogstreamResult.Write = recptr;
- 
- 		/*
- 		 * XXX: Should we signal bgwriter to start a restartpoint if we've
- 		 * consumed too much xlog since the last one, like in normal
- 		 * processing? But this is not worth doing unless a restartpoint can
- 		 * be created independently from a checkpoint record.
- 		 */
  	}
  }
  
--- 505,510 ----