maxstandbydelay-1.patch

Index: src/backend/access/transam/xlog.c
===================================================================
RCS file: /cvsroot/pgsql/src/backend/access/transam/xlog.c,v
retrieving revision 1.427
diff -c -r1.427 xlog.c
*** src/backend/access/transam/xlog.c	28 Jun 2010 19:46:19 -0000	1.427
--- src/backend/access/transam/xlog.c	2 Jul 2010 20:01:24 -0000
***************
*** 72,78 ****
  bool		XLogArchiveMode = false;
  char	   *XLogArchiveCommand = NULL;
  bool		EnableHotStandby = false;
- int			MaxStandbyDelay = 30 * 1000;
  bool		fullPageWrites = true;
  bool		log_checkpoints = false;
  int			sync_method = DEFAULT_SYNC_METHOD;
--- 72,77 ----
***************
*** 487,493 ****
   * Keeps track of which sources we've tried to read the current WAL
   * record from and failed.
   */
! static int failedSources = 0;
  
  /* Buffer for currently read page (XLOG_BLCKSZ bytes) */
  static char *readBuf = NULL;
--- 487,502 ----
   * Keeps track of which sources we've tried to read the current WAL
   * record from and failed.
   */
! static int failedSources = 0;	/* OR of XLOG_FROM_* codes */
! 
! /*
!  * These variables track when we last obtained some WAL data to process,
!  * and where we got it from.  (XLogReceiptSource is initially the same as
!  * readSource, but readSource gets reset to zero when we don't have data
!  * to process right now.)
!  */
! static TimestampTz XLogReceiptTime = 0;
! static int XLogReceiptSource = 0;	/* XLOG_FROM_* code */
  
  /* Buffer for currently read page (XLOG_BLCKSZ bytes) */
  static char *readBuf = NULL;
***************
*** 2626,2632 ****
   * Open a logfile segment for reading (during recovery).
   *
   * If source = XLOG_FROM_ARCHIVE, the segment is retrieved from archive.
!  * If source = XLOG_FROM_PG_XLOG, it's read from pg_xlog.
   */
  static int
  XLogFileRead(uint32 log, uint32 seg, int emode, TimeLineID tli,
--- 2635,2641 ----
   * Open a logfile segment for reading (during recovery).
   *
   * If source = XLOG_FROM_ARCHIVE, the segment is retrieved from archive.
!  * Otherwise, it's assumed to be already available in pg_xlog.
   */
  static int
  XLogFileRead(uint32 log, uint32 seg, int emode, TimeLineID tli,
***************
*** 2655,2660 ****
--- 2664,2670 ----
  			break;
  
  		case XLOG_FROM_PG_XLOG:
+ 		case XLOG_FROM_STREAM:
  			XLogFilePath(path, tli, log, seg);
  			restoredFromArchive = false;
  			break;
***************
*** 2674,2680 ****
--- 2684,2696 ----
  				 xlogfname);
  		set_ps_display(activitymsg, false);
  
+ 		/* Track source of data in assorted state variables */
  		readSource = source;
+ 		XLogReceiptSource = source;
+ 		/* In FROM_STREAM case, caller tracks receipt time, not me */
+ 		if (source != XLOG_FROM_STREAM)
+ 			XLogReceiptTime = GetCurrentTimestamp();
+ 
  		return fd;
  	}
  	if (errno != ENOENT || !notfoundOk) /* unexpected failure? */
***************
*** 5568,5574 ****
  /*
   * Returns timestamp of last recovered commit/abort record.
   */
! TimestampTz
  GetLatestXLogTime(void)
  {
  	/* use volatile pointer to prevent code rearrangement */
--- 5584,5590 ----
  /*
   * Returns timestamp of last recovered commit/abort record.
   */
! static TimestampTz
  GetLatestXLogTime(void)
  {
  	/* use volatile pointer to prevent code rearrangement */
***************
*** 5582,5587 ****
--- 5598,5620 ----
  }
  
  /*
+  * Returns time of receipt of current chunk of XLOG data, as well as
+  * whether it was received from streaming replication or from archives.
+  */
+ void
+ GetXLogReceiptTime(TimestampTz *rtime, bool *fromStream)
+ {
+ 	/*
+ 	 * This must be executed in the startup process, since we don't export
+ 	 * the relevant state to shared memory.
+ 	 */
+ 	Assert(InRecovery);
+ 
+ 	*rtime = XLogReceiptTime;
+ 	*fromStream = (XLogReceiptSource == XLOG_FROM_STREAM);
+ }
+ 
+ /*
   * Note that text field supplied is a parameter name and does not require
   * translation
   */
***************
*** 6060,6065 ****
--- 6093,6101 ----
  		xlogctl->recoveryLastRecPtr = ReadRecPtr;
  		SpinLockRelease(&xlogctl->info_lck);
  
+ 		/* Also ensure XLogReceiptTime has a sane value */
+ 		XLogReceiptTime = GetCurrentTimestamp();
+ 
  		/*
  		 * Let postmaster know we've started redo now, so that it can
  		 * launch bgwriter to perform restartpoints.  We don't bother
***************
*** 7647,7653 ****
  		XLogRecPtr	endptr;
  
  		/* Get the current (or recent) end of xlog */
! 		endptr = GetWalRcvWriteRecPtr();
  
  		PrevLogSeg(_logId, _logSeg);
  		RemoveOldXlogFiles(_logId, _logSeg, endptr);
--- 7683,7689 ----
  		XLogRecPtr	endptr;
  
  		/* Get the current (or recent) end of xlog */
! 		endptr = GetWalRcvWriteRecPtr(NULL);
  
  		PrevLogSeg(_logId, _logSeg);
  		RemoveOldXlogFiles(_logId, _logSeg, endptr);
***************
*** 8757,8763 ****
  	XLogRecPtr	recptr;
  	char		location[MAXFNAMELEN];
  
! 	recptr = GetWalRcvWriteRecPtr();
  
  	if (recptr.xlogid == 0 && recptr.xrecoff == 0)
  		PG_RETURN_NULL();
--- 8793,8799 ----
  	XLogRecPtr	recptr;
  	char		location[MAXFNAMELEN];
  
! 	recptr = GetWalRcvWriteRecPtr(NULL);
  
  	if (recptr.xlogid == 0 && recptr.xrecoff == 0)
  		PG_RETURN_NULL();
***************
*** 9272,9277 ****
--- 9308,9315 ----
  			{
  				if (WalRcvInProgress())
  				{
+ 					bool	havedata;
+ 
  					/*
  					 * If we find an invalid record in the WAL streamed from
  					 * master, something is seriously wrong. There's little
***************
*** 9289,9316 ****
  					}
  
  					/*
! 					 * While walreceiver is active, wait for new WAL to arrive
! 					 * from primary.
  					 */
- 					receivedUpto = GetWalRcvWriteRecPtr();
  					if (XLByteLT(*RecPtr, receivedUpto))
  					{
  						/*
  						 * Great, streamed far enough. Open the file if it's
! 						 * not open already.
  						 */
  						if (readFile < 0)
  						{
  							readFile =
  								XLogFileRead(readId, readSeg, PANIC,
  											 recoveryTargetTLI,
! 											 XLOG_FROM_PG_XLOG, false);
  							switched_segment = true;
  							readSource = XLOG_FROM_STREAM;
  						}
  						break;
  					}
  
  					if (CheckForStandbyTrigger())
  						goto triggered;
  
--- 9327,9388 ----
  					}
  
  					/*
! 					 * Walreceiver is active, so see if new data has arrived.
! 					 *
! 					 * We only advance XLogReceiptTime when we obtain fresh
! 					 * WAL from walreceiver and observe that we had already
! 					 * processed everything before the most recent "chunk"
! 					 * that it flushed to disk.  In steady state where we are
! 					 * keeping up with the incoming data, XLogReceiptTime
! 					 * will be updated on each cycle.  When we are behind,
! 					 * XLogReceiptTime will not advance, so the grace time
! 					 * alloted to conflicting queries will decrease.
  					 */
  					if (XLByteLT(*RecPtr, receivedUpto))
+ 						havedata = true;
+ 					else
+ 					{
+ 						XLogRecPtr	latestChunkStart;
+ 
+ 						receivedUpto = GetWalRcvWriteRecPtr(&latestChunkStart);
+ 						if (XLByteLT(*RecPtr, receivedUpto))
+ 						{
+ 							havedata = true;
+ 							if (!XLByteLT(*RecPtr, latestChunkStart))
+ 								XLogReceiptTime = GetCurrentTimestamp();
+ 						}
+ 						else
+ 							havedata = false;
+ 					}
+ 					if (havedata)
  					{
  						/*
  						 * Great, streamed far enough. Open the file if it's
! 						 * not open already.  Use XLOG_FROM_STREAM so that
! 						 * source info is set correctly and XLogReceiptTime
! 						 * isn't changed.
  						 */
  						if (readFile < 0)
  						{
  							readFile =
  								XLogFileRead(readId, readSeg, PANIC,
  											 recoveryTargetTLI,
! 											 XLOG_FROM_STREAM, false);
! 							Assert(readFile >= 0);
  							switched_segment = true;
+ 						}
+ 						else
+ 						{
+ 							/* just make sure source info is correct... */
  							readSource = XLOG_FROM_STREAM;
+ 							XLogReceiptSource = XLOG_FROM_STREAM;
  						}
  						break;
  					}
  
+ 					/*
+ 					 * Data not here yet, so check for trigger then sleep.
+ 					 */
  					if (CheckForStandbyTrigger())
  						goto triggered;
  
Index: src/backend/replication/walreceiver.c
===================================================================
RCS file: /cvsroot/pgsql/src/backend/replication/walreceiver.c,v
retrieving revision 1.14
diff -c -r1.14 walreceiver.c
*** src/backend/replication/walreceiver.c	9 Jun 2010 15:04:07 -0000	1.14
--- src/backend/replication/walreceiver.c	2 Jul 2010 20:01:24 -0000
***************
*** 524,529 ****
--- 524,530 ----
  
  		/* Update shared-memory status */
  		SpinLockAcquire(&walrcv->mutex);
+ 		walrcv->latestChunkStart = walrcv->receivedUpto;
  		walrcv->receivedUpto = LogstreamResult.Flush;
  		SpinLockRelease(&walrcv->mutex);
  
Index: src/backend/replication/walreceiverfuncs.c
===================================================================
RCS file: /cvsroot/pgsql/src/backend/replication/walreceiverfuncs.c,v
retrieving revision 1.5
diff -c -r1.5 walreceiverfuncs.c
*** src/backend/replication/walreceiverfuncs.c	28 Apr 2010 16:54:15 -0000	1.5
--- src/backend/replication/walreceiverfuncs.c	2 Jul 2010 20:01:24 -0000
***************
*** 199,214 ****
  	walrcv->startTime = now;
  
  	walrcv->receivedUpto = recptr;
  	SpinLockRelease(&walrcv->mutex);
  
  	SendPostmasterSignal(PMSIGNAL_START_WALRECEIVER);
  }
  
  /*
!  * Returns the byte position that walreceiver has written
   */
  XLogRecPtr
! GetWalRcvWriteRecPtr(void)
  {
  	/* use volatile pointer to prevent code rearrangement */
  	volatile WalRcvData *walrcv = WalRcv;
--- 200,221 ----
  	walrcv->startTime = now;
  
  	walrcv->receivedUpto = recptr;
+ 	walrcv->latestChunkStart = recptr;
+ 
  	SpinLockRelease(&walrcv->mutex);
  
  	SendPostmasterSignal(PMSIGNAL_START_WALRECEIVER);
  }
  
  /*
!  * Returns the last+1 byte position that walreceiver has written.
!  *
!  * Optionally, returns the previous chunk start, that is the first byte
!  * written in the most recent walreceiver flush cycle.  Callers not
!  * interested in that value may pass NULL for latestChunkStart.
   */
  XLogRecPtr
! GetWalRcvWriteRecPtr(XLogRecPtr *latestChunkStart)
  {
  	/* use volatile pointer to prevent code rearrangement */
  	volatile WalRcvData *walrcv = WalRcv;
***************
*** 216,221 ****
--- 223,230 ----
  
  	SpinLockAcquire(&walrcv->mutex);
  	recptr = walrcv->receivedUpto;
+ 	if (latestChunkStart)
+ 		*latestChunkStart = walrcv->latestChunkStart;
  	SpinLockRelease(&walrcv->mutex);
  
  	return recptr;
Index: src/backend/storage/ipc/standby.c
===================================================================
RCS file: /cvsroot/pgsql/src/backend/storage/ipc/standby.c,v
retrieving revision 1.25
diff -c -r1.25 standby.c
*** src/backend/storage/ipc/standby.c	14 Jun 2010 00:49:24 -0000	1.25
--- src/backend/storage/ipc/standby.c	2 Jul 2010 20:01:24 -0000
***************
*** 30,36 ****
--- 30,39 ----
  #include "storage/standby.h"
  #include "utils/ps_status.h"
  
+ /* User-settable GUC parameters */
  int			vacuum_defer_cleanup_age;
+ int			max_standby_archive_delay = 30 * 1000;
+ int			max_standby_streaming_delay = 30 * 1000;
  
  static List *RecoveryLockList;
  
***************
*** 113,118 ****
--- 117,152 ----
   * -----------------------------------------------------
   */
  
+ /*
+  * Determine the cutoff time at which we want to start canceling conflicting
+  * transactions.  Returns zero (a time safely in the past) if we are willing
+  * to wait forever.
+  */
+ static TimestampTz
+ GetStandbyLimitTime(void)
+ {
+ 	TimestampTz	rtime;
+ 	bool		fromStream;
+ 
+ 	/*
+ 	 * The cutoff time is the last WAL data receipt time plus the appropriate
+ 	 * delay variable.  Delay of -1 means wait forever.
+ 	 */
+ 	GetXLogReceiptTime(&rtime, &fromStream);
+ 	if (fromStream)
+ 	{
+ 		if (max_standby_streaming_delay < 0)
+ 			return 0;			/* wait forever */
+ 		return TimestampTzPlusMilliseconds(rtime, max_standby_streaming_delay);
+ 	}
+ 	else
+ 	{
+ 		if (max_standby_archive_delay < 0)
+ 			return 0;			/* wait forever */
+ 		return TimestampTzPlusMilliseconds(rtime, max_standby_archive_delay);
+ 	}
+ }
+ 
  #define STANDBY_INITIAL_WAIT_US  1000
  static int	standbyWait_us = STANDBY_INITIAL_WAIT_US;
  
***************
*** 124,133 ****
  static bool
  WaitExceedsMaxStandbyDelay(void)
  {
! 	/* Are we past max_standby_delay? */
! 	if (MaxStandbyDelay >= 0 &&
! 		TimestampDifferenceExceeds(GetLatestXLogTime(), GetCurrentTimestamp(),
! 								   MaxStandbyDelay))
  		return true;
  
  	/*
--- 158,168 ----
  static bool
  WaitExceedsMaxStandbyDelay(void)
  {
! 	TimestampTz	ltime;
! 
! 	/* Are we past the limit time? */
! 	ltime = GetStandbyLimitTime();
! 	if (ltime && GetCurrentTimestamp() >= ltime)
  		return true;
  
  	/*
***************
*** 368,433 ****
   * Startup is sleeping and the query waits on a lock. We protect against
   * only the former sequence here, the latter sequence is checked prior to
   * the query sleeping, in CheckRecoveryConflictDeadlock().
   */
  void
  ResolveRecoveryConflictWithBufferPin(void)
  {
  	bool		sig_alarm_enabled = false;
  
  	Assert(InHotStandby);
  
! 	if (MaxStandbyDelay == 0)
! 	{
! 		/*
! 		 * We don't want to wait, so just tell everybody holding the pin to
! 		 * get out of town.
! 		 */
! 		SendRecoveryConflictWithBufferPin(PROCSIG_RECOVERY_CONFLICT_BUFFERPIN);
! 	}
! 	else if (MaxStandbyDelay < 0)
! 	{
! 		TimestampTz now = GetCurrentTimestamp();
  
  		/*
! 		 * Set timeout for deadlock check (only)
  		 */
  		if (enable_standby_sig_alarm(now, now, true))
  			sig_alarm_enabled = true;
  		else
  			elog(FATAL, "could not set timer for process wakeup");
  	}
  	else
  	{
! 		TimestampTz then = GetLatestXLogTime();
! 		TimestampTz now = GetCurrentTimestamp();
! 
! 		/* Are we past max_standby_delay? */
! 		if (TimestampDifferenceExceeds(then, now, MaxStandbyDelay))
! 		{
! 			/*
! 			 * We're already behind, so clear a path as quickly as possible.
! 			 */
! 			SendRecoveryConflictWithBufferPin(PROCSIG_RECOVERY_CONFLICT_BUFFERPIN);
! 		}
  		else
! 		{
! 			TimestampTz max_standby_time;
! 
! 			/*
! 			 * At what point in the future do we hit MaxStandbyDelay?
! 			 */
! 			max_standby_time = TimestampTzPlusMilliseconds(then, MaxStandbyDelay);
! 			Assert(max_standby_time > now);
! 
! 			/*
! 			 * Wake up at MaxStandby delay, and check for deadlocks as well
! 			 * if we will be waiting longer than deadlock_timeout
! 			 */
! 			if (enable_standby_sig_alarm(now, max_standby_time, false))
! 				sig_alarm_enabled = true;
! 			else
! 				elog(FATAL, "could not set timer for process wakeup");
! 		}
  	}
  
  	/* Wait to be signaled by UnpinBuffer() */
--- 402,452 ----
   * Startup is sleeping and the query waits on a lock. We protect against
   * only the former sequence here, the latter sequence is checked prior to
   * the query sleeping, in CheckRecoveryConflictDeadlock().
+  *
+  * Deadlocks are extremely rare, and relatively expensive to check for,
+  * so we don't do a deadlock check right away ... only if we have had to wait
+  * at least deadlock_timeout.  Most of the logic about that is in proc.c.
   */
  void
  ResolveRecoveryConflictWithBufferPin(void)
  {
  	bool		sig_alarm_enabled = false;
+ 	TimestampTz	ltime;
+ 	TimestampTz	now;
  
  	Assert(InHotStandby);
  
! 	ltime = GetStandbyLimitTime();
! 	now = GetCurrentTimestamp();
  
+ 	if (!ltime)
+ 	{
  		/*
! 		 * We're willing to wait forever for conflicts, so set timeout for
! 		 * deadlock check (only)
  		 */
  		if (enable_standby_sig_alarm(now, now, true))
  			sig_alarm_enabled = true;
  		else
  			elog(FATAL, "could not set timer for process wakeup");
  	}
+ 	else if (now >= ltime)
+ 	{
+ 		/*
+ 		 * We're already behind, so clear a path as quickly as possible.
+ 		 */
+ 		SendRecoveryConflictWithBufferPin(PROCSIG_RECOVERY_CONFLICT_BUFFERPIN);
+ 	}
  	else
  	{
! 		/*
! 		 * Wake up at ltime, and check for deadlocks as well if we will be
! 		 * waiting longer than deadlock_timeout
! 		 */
! 		if (enable_standby_sig_alarm(now, ltime, false))
! 			sig_alarm_enabled = true;
  		else
! 			elog(FATAL, "could not set timer for process wakeup");
  	}
  
  	/* Wait to be signaled by UnpinBuffer() */
Index: src/backend/utils/misc/guc.c
===================================================================
RCS file: /cvsroot/pgsql/src/backend/utils/misc/guc.c,v
retrieving revision 1.557
diff -c -r1.557 guc.c
*** src/backend/utils/misc/guc.c	25 Jun 2010 13:11:25 -0000	1.557
--- src/backend/utils/misc/guc.c	2 Jul 2010 20:01:25 -0000
***************
*** 57,62 ****
--- 57,63 ----
  #include "postmaster/walwriter.h"
  #include "replication/walsender.h"
  #include "storage/bufmgr.h"
+ #include "storage/standby.h"
  #include "storage/fd.h"
  #include "tcop/tcopprot.h"
  #include "tsearch/ts_cache.h"
***************
*** 116,122 ****
  extern char *temp_tablespaces;
  extern bool synchronize_seqscans;
  extern bool fullPageWrites;
- extern int	vacuum_defer_cleanup_age;
  extern int	ssl_renegotiation_limit;
  
  #ifdef TRACE_SORT
--- 117,122 ----
***************
*** 1373,1378 ****
--- 1373,1398 ----
  		1000, 1, INT_MAX / 1000, NULL, NULL
  	},
  
+ 	{
+ 		{"max_standby_archive_delay", PGC_SIGHUP, WAL_STANDBY_SERVERS,
+ 			gettext_noop("Sets the maximum delay before canceling queries when a hot standby server is processing archived WAL data."),
+ 			NULL,
+ 			GUC_UNIT_MS
+ 		},
+ 		&max_standby_archive_delay,
+ 		30 * 1000, -1, INT_MAX / 1000, NULL, NULL
+ 	},
+ 
+ 	{
+ 		{"max_standby_streaming_delay", PGC_SIGHUP, WAL_STANDBY_SERVERS,
+ 			gettext_noop("Sets the maximum delay before canceling queries when a hot standby server is processing streamed WAL data."),
+ 			NULL,
+ 			GUC_UNIT_MS
+ 		},
+ 		&max_standby_streaming_delay,
+ 		30 * 1000, -1, INT_MAX / 1000, NULL, NULL
+ 	},
+ 
  	/*
  	 * Note: MaxBackends is limited to INT_MAX/4 because some places compute
  	 * 4*MaxBackends without any overflow check.  This check is made in
***************
*** 1393,1408 ****
  	},
  
  	{
- 		{"max_standby_delay", PGC_SIGHUP, WAL_STANDBY_SERVERS,
- 			gettext_noop("Sets the maximum delay to avoid conflict processing on hot standby servers."),
- 			NULL,
- 			GUC_UNIT_MS
- 		},
- 		&MaxStandbyDelay,
- 		30 * 1000, -1, INT_MAX / 1000, NULL, NULL
- 	},
- 
- 	{
  		{"superuser_reserved_connections", PGC_POSTMASTER, CONN_AUTH_SETTINGS,
  			gettext_noop("Sets the number of connection slots reserved for superusers."),
  			NULL
--- 1413,1418 ----
Index: src/include/access/xlog.h
===================================================================
RCS file: /cvsroot/pgsql/src/include/access/xlog.h,v
retrieving revision 1.113
diff -c -r1.113 xlog.h
*** src/include/access/xlog.h	17 Jun 2010 16:41:25 -0000	1.113
--- src/include/access/xlog.h	2 Jul 2010 20:01:25 -0000
***************
*** 193,199 ****
  extern bool XLogArchiveMode;
  extern char *XLogArchiveCommand;
  extern bool EnableHotStandby;
- extern int	MaxStandbyDelay;
  extern bool log_checkpoints;
  
  /* WAL levels */
--- 197,202 ----
***************
*** 279,285 ****
  
  extern bool RecoveryInProgress(void);
  extern bool XLogInsertAllowed(void);
! extern TimestampTz GetLatestXLogTime(void);
  
  extern void UpdateControlFile(void);
  extern uint64 GetSystemIdentifier(void);
--- 282,288 ----
  
  extern bool RecoveryInProgress(void);
  extern bool XLogInsertAllowed(void);
! extern void GetXLogReceiptTime(TimestampTz *rtime, bool *fromStream);
  
  extern void UpdateControlFile(void);
  extern uint64 GetSystemIdentifier(void);
Index: src/include/replication/walreceiver.h
===================================================================
RCS file: /cvsroot/pgsql/src/include/replication/walreceiver.h,v
retrieving revision 1.9
diff -c -r1.9 walreceiver.h
*** src/include/replication/walreceiver.h	3 Jun 2010 22:17:32 -0000	1.9
--- src/include/replication/walreceiver.h	2 Jul 2010 20:01:25 -0000
***************
*** 41,65 ****
  typedef struct
  {
  	/*
! 	 * connection string; is used for walreceiver to connect with the primary.
! 	 */
! 	char		conninfo[MAXCONNINFO];
! 
! 	/*
! 	 * PID of currently active walreceiver process, and the current state.
  	 */
  	pid_t		pid;
  	WalRcvState walRcvState;
  	pg_time_t	startTime;
  
  	/*
! 	 * receivedUpto-1 is the last byte position that has been already
! 	 * received. When startup process starts the walreceiver, it sets this to
! 	 * the point where it wants the streaming to begin. After that,
! 	 * walreceiver updates this whenever it flushes the received WAL.
  	 */
  	XLogRecPtr	receivedUpto;
  
  	slock_t		mutex;			/* locks shared variables shown above */
  } WalRcvData;
  
--- 41,75 ----
  typedef struct
  {
  	/*
! 	 * PID of currently active walreceiver process, its current state and
! 	 * start time (actually, the time at which it was requested to be started).
  	 */
  	pid_t		pid;
  	WalRcvState walRcvState;
  	pg_time_t	startTime;
  
  	/*
! 	 * receivedUpto-1 is the last byte position that has already been
! 	 * received.  When startup process starts the walreceiver, it sets
! 	 * receivedUpto to the point where it wants the streaming to begin.
! 	 * After that, walreceiver updates this whenever it flushes the received
! 	 * WAL to disk.
  	 */
  	XLogRecPtr	receivedUpto;
  
+ 	/*
+ 	 * latestChunkStart is the starting byte position of the current "batch"
+ 	 * of received WAL.  It's actually the same as the previous value of
+ 	 * receivedUpto before the last flush to disk.  Startup process can use
+ 	 * this to detect whether it's keeping up or not.
+ 	 */
+ 	XLogRecPtr	latestChunkStart;
+ 
+ 	/*
+ 	 * connection string; is used for walreceiver to connect with the primary.
+ 	 */
+ 	char		conninfo[MAXCONNINFO];
+ 
  	slock_t		mutex;			/* locks shared variables shown above */
  } WalRcvData;
  
***************
*** 83,88 ****
  extern bool WalRcvInProgress(void);
  extern XLogRecPtr WaitNextXLogAvailable(XLogRecPtr recptr, bool *finished);
  extern void RequestXLogStreaming(XLogRecPtr recptr, const char *conninfo);
! extern XLogRecPtr GetWalRcvWriteRecPtr(void);
  
  #endif   /* _WALRECEIVER_H */
--- 93,98 ----
  extern bool WalRcvInProgress(void);
  extern XLogRecPtr WaitNextXLogAvailable(XLogRecPtr recptr, bool *finished);
  extern void RequestXLogStreaming(XLogRecPtr recptr, const char *conninfo);
! extern XLogRecPtr GetWalRcvWriteRecPtr(XLogRecPtr *latestChunkStart);
  
  #endif   /* _WALRECEIVER_H */
Index: src/include/storage/standby.h
===================================================================
RCS file: /cvsroot/pgsql/src/include/storage/standby.h,v
retrieving revision 1.10
diff -c -r1.10 standby.h
*** src/include/storage/standby.h	13 May 2010 11:15:38 -0000	1.10
--- src/include/storage/standby.h	2 Jul 2010 20:01:25 -0000
***************
*** 19,25 ****
--- 19,28 ----
  #include "storage/procsignal.h"
  #include "storage/relfilenode.h"
  
+ /* User-settable GUC parameters */
  extern int	vacuum_defer_cleanup_age;
+ extern int	max_standby_archive_delay;
+ extern int	max_standby_streaming_delay;
  
  extern void InitRecoveryTransactionEnvironment(void);
  extern void ShutdownRecoveryTransactionEnvironment(void);