replication_lag_segments_0121.patch

*** a/doc/src/sgml/config.sgml
--- b/doc/src/sgml/config.sgml
***************
*** 1789,1794 **** archive_command = 'copy "%p" "C:\\server\\archivedir\\%f"'  # Windows
--- 1789,1812 ----
         </para>
         </listitem>
        </varlistentry>
+       <varlistentry id="guc-replication-lag-segments" xreflabel="replication_lag_segments">
+        <term><varname>replication_lag_segments</varname> (<type>integer</type>)</term>
+        <indexterm>
+         <primary><varname>replication_lag_segments</> configuration parameter</primary>
+        </indexterm>
+        <listitem>
+        <para>
+         Specifies the maximum number of log file segments held in <filename>pg_xlog</>
+         directory to send to the standby server (each segment is normally 16 megabytes).
+         The replication to the standby server which falls more than <varname>
+         replication_lag_segments</> behind is terminated. This is useful for
+         avoiding a disk full error on the primary and holding the segments required for
+         the incoming standby server. The default value is zero, which disables that
+         upper limit. This parameter can only be set in the <filename>postgresql.conf</>
+         file or on the server command line.
+        </para>
+        </listitem>
+       </varlistentry>
       </variablelist>
      </sect2>
      <sect2 id="runtime-config-standby">
*** a/src/backend/access/transam/xlog.c
--- b/src/backend/access/transam/xlog.c
***************
*** 1725,1730 **** XLogWrite(XLogwrtRqst WriteRqst, bool flexible, bool xlog_switch)
--- 1725,1736 ----
  				if (XLogArchivingActive())
  					XLogArchiveNotifySeg(openLogId, openLogSeg);
  
+ 				/*
+ 				 * Check for the standbys' delay and terminate replication
+ 				 * if needed.
+ 				 */
+ 				CheckStandbysDelay(LogwrtResult.Write);
+ 
  				Write->lastSegSwitchTime = (pg_time_t) time(NULL);
  
  				/*
***************
*** 7213,7240 **** CreateCheckPoint(int flags)
  	smgrpostckpt();
  
  	/*
! 	 * If there's connected standby servers doing XLOG streaming, don't
! 	 * delete XLOG files that have not been streamed to all of them yet.
! 	 * This does nothing to prevent them from being deleted when the
! 	 * standby is disconnected (e.g because of network problems), but at
! 	 * least it avoids an open replication connection from failing because
  	 * of that.
  	 */
  	if ((_logId || _logSeg) && MaxWalSenders > 0)
  	{
- 		XLogRecPtr oldest;
  		uint32	log;
  		uint32	seg;
  
! 		oldest = GetOldestWALSendPointer();
! 		if (oldest.xlogid != 0 || oldest.xrecoff != 0)
  		{
! 			XLByteToSeg(oldest, log, seg);
! 			if (log < _logId || (log == _logId && seg < _logSeg))
! 			{
! 				_logId	= log;
! 				_logSeg	= seg;
! 			}
  		}
  	}
  
--- 7219,7260 ----
  	smgrpostckpt();
  
  	/*
! 	 * Don't delete XLOG files which could be still required for
! 	 * connected or incoming standbys, under the given upper limit.
! 	 * This avoids a replication connection from failing because
  	 * of that.
  	 */
  	if ((_logId || _logSeg) && MaxWalSenders > 0)
  	{
  		uint32	log;
  		uint32	seg;
+ 		bool	need_comp = true;
  
! 		if (RepLagSegs > 0)
  		{
! 			/*
! 			 * Ensure that there is no too lagged standbys before
! 			 * deleting XLOG files.
! 			 */
! 			CheckStandbysDelay(recptr);
! 			XLByteToSeg(recptr, log, seg);
! 			PrevLogSegs(log, seg, RepLagSegs - 1);
! 		}
! 		else
! 		{
! 			XLogRecPtr oldest;
! 
! 			oldest = GetOldestWALSendPointer();
! 			if (oldest.xlogid == 0 || oldest.xrecoff == 0)
! 				need_comp = false;
! 			else
! 				XLByteToSeg(oldest, log, seg);
! 		}
! 
! 		if (need_comp && (log < _logId || (log == _logId && seg < _logSeg)))
! 		{
! 			_logId	= log;
! 			_logSeg	= seg;
  		}
  	}
  
*** a/src/backend/replication/walsender.c
--- b/src/backend/replication/walsender.c
***************
*** 66,71 **** bool	am_walsender	= false;	/* Am I a walsender process ? */
--- 66,72 ----
  /* User-settable parameters for walsender */
  int	MaxWalSenders = 0;		/* the maximum number of concurrent walsenders */
  int	WalSndDelay	= 200;		/* max sleep time between some actions */
+ int	RepLagSegs	= 0;		/* the maximum number of WAL files held for standby */
  
  #define NAPTIME_PER_CYCLE 100	/* max sleep time between cycles (100ms) */
  
***************
*** 86,94 **** static XLogRecPtr sentPtr = {0, 0};
  
  /* Flags set by signal handlers for later service in main loop */
  static volatile sig_atomic_t got_SIGHUP = false;
! static volatile sig_atomic_t shutdown_requested = false;
  static volatile sig_atomic_t ready_to_stop = false;
  
  /* Signal handlers */
  static void WalSndSigHupHandler(SIGNAL_ARGS);
  static void WalSndShutdownHandler(SIGNAL_ARGS);
--- 87,150 ----
  
  /* Flags set by signal handlers for later service in main loop */
  static volatile sig_atomic_t got_SIGHUP = false;
! static volatile sig_atomic_t got_SIGTERM = false;
  static volatile sig_atomic_t ready_to_stop = false;
  
+ static void ProcessWalSndInterrupts(void);
+ static void EnableWalSndImmediateExit(void);
+ static void DisableWalSndImmediateExit(void);
+ 
+ /*
+  * About SIGTERM handling:
+  *
+  * We can't just exit(1) within SIGTERM signal handler, because the signal
+  * might arrive in the middle of some critical operation, like while we're
+  * holding a spinlock. We also can't just set a flag in signal handler and
+  * check it in the main loop, because we perform some blocking libpq
+  * operations like pq_flush(), which can take a long time to finish.
+  *
+  * We use a combined approach: When WalSndImmediateInterruptOK is true, it's
+  * safe for the signal handler to elog(FATAL) immediately. Otherwise it just
+  * sets got_SIGTERM flag, which is checked in the main loop when convenient.
+  *
+  * This is very much like what regular backends do with ImmediateInterruptOK,
+  * ProcessInterrupts() etc.
+  */
+ static volatile bool WalSndImmediateInterruptOK = false;
+ 
+ static void
+ ProcessWalSndInterrupts(void)
+ {
+ 	/*
+ 	 * Although walsender interrupt handling doesn't use the same scheme
+ 	 * as regular backends, call CHECK_FOR_INTERRUPTS() to make sure we
+ 	 * receive any incoming signals on Win32.
+ 	 */
+ 	CHECK_FOR_INTERRUPTS();
+ 
+ 	if (got_SIGTERM)
+ 	{
+ 		WalSndImmediateInterruptOK = false;
+ 		ereport(FATAL,
+ 				(errcode(ERRCODE_ADMIN_SHUTDOWN),
+ 				 errmsg("terminating replication because the standby falls too far behind")));
+ 	}
+ }
+ 
+ static void
+ EnableWalSndImmediateExit()
+ {
+ 	WalSndImmediateInterruptOK = true;
+ 	ProcessWalSndInterrupts();
+ }
+ 
+ static void
+ DisableWalSndImmediateExit()
+ {
+ 	WalSndImmediateInterruptOK = false;
+ 	ProcessWalSndInterrupts();
+ }
+ 
  /* Signal handlers */
  static void WalSndSigHupHandler(SIGNAL_ARGS);
  static void WalSndShutdownHandler(SIGNAL_ARGS);
***************
*** 386,396 **** WalSndLoop(void)
  		if (ready_to_stop)
  		{
  			XLogSend(&output_message);
! 			shutdown_requested = true;
  		}
  
  		/* Normal exit from the walsender is here */
! 		if (shutdown_requested)
  		{
  			/* Inform the standby that XLOG streaming was done */
  			pq_puttextmessage('C', "COPY 0");
--- 442,452 ----
  		if (ready_to_stop)
  		{
  			XLogSend(&output_message);
! 			got_SIGTERM = true;
  		}
  
  		/* Normal exit from the walsender is here */
! 		if (got_SIGTERM)
  		{
  			/* Inform the standby that XLOG streaming was done */
  			pq_puttextmessage('C', "COPY 0");
***************
*** 410,416 **** WalSndLoop(void)
  		remain = WalSndDelay;
  		while (remain > 0)
  		{
! 			if (got_SIGHUP || shutdown_requested || ready_to_stop)
  				break;
  
  			/*
--- 466,472 ----
  		remain = WalSndDelay;
  		while (remain > 0)
  		{
! 			if (got_SIGHUP || got_SIGTERM || ready_to_stop)
  				break;
  
  			/*
***************
*** 586,591 **** XLogSend(StringInfo outMsg)
--- 642,648 ----
  {
  	XLogRecPtr	SendRqstPtr;
  	char	activitymsg[50];
+ 	int	res;
  	/* use volatile pointer to prevent code rearrangement */
  	volatile WalSnd *walsnd = MyWalSnd;
  
***************
*** 676,682 **** XLogSend(StringInfo outMsg)
--- 733,741 ----
  		outMsg->len += nbytes;
  		outMsg->data[outMsg->len] = '\0';
  
+ 		EnableWalSndImmediateExit();
  		pq_putmessage('d', outMsg->data, outMsg->len);
+ 		DisableWalSndImmediateExit();
  		resetStringInfo(outMsg);
  	}
  
***************
*** 686,692 **** XLogSend(StringInfo outMsg)
  	SpinLockRelease(&walsnd->mutex);
  
  	/* Flush pending output */
! 	if (pq_flush())
  		return false;
  
  	/* Report progress of XLOG streaming in PS display */
--- 745,754 ----
  	SpinLockRelease(&walsnd->mutex);
  
  	/* Flush pending output */
! 	EnableWalSndImmediateExit();
! 	res = pq_flush();
! 	DisableWalSndImmediateExit();
! 	if (res)
  		return false;
  
  	/* Report progress of XLOG streaming in PS display */
***************
*** 704,714 **** WalSndSigHupHandler(SIGNAL_ARGS)
  	got_SIGHUP = true;
  }
  
! /* SIGTERM: set flag to shut down */
  static void
  WalSndShutdownHandler(SIGNAL_ARGS)
  {
! 	shutdown_requested = true;
  }
  
  /*
--- 766,780 ----
  	got_SIGHUP = true;
  }
  
! /* SIGTERM: set flag for main loop, or shutdown immediately if safe */
  static void
  WalSndShutdownHandler(SIGNAL_ARGS)
  {
! 	got_SIGTERM = true;
! 
! 	/* Don't joggle the elbow of proc_exit */
! 	if (!proc_exit_inprogress && WalSndImmediateInterruptOK)
! 		ProcessWalSndInterrupts();
  }
  
  /*
***************
*** 844,846 **** GetOldestWALSendPointer(void)
--- 910,976 ----
  	}
  	return oldest;
  }
+ 
+ /*
+  * Check if the standbys have fallen too far behind.
+  */
+ void
+ CheckStandbysDelay(XLogRecPtr newptr)
+ {
+ 	uint32	old_segno;
+ 	uint32	new_segno;
+ 	uint32	old_highbits;
+ 	uint32	new_highbits;
+ 	int	i;
+ 	bool	found = false;
+ 
+ 	if (RepLagSegs <= 0)
+ 		return;
+ 
+ 	for (i = 0; i < MaxWalSenders; i++)
+ 	{
+ 		/* use volatile pointer to prevent code rearrangement */
+ 		volatile WalSnd	*walsnd = &WalSndCtl->walsnds[i];
+ 		XLogRecPtr	oldptr;
+ 		pid_t	walsndpid;
+ 
+ 		SpinLockAcquire(&walsnd->mutex);
+ 		oldptr = walsnd->sentPtr;
+ 		walsndpid = walsnd->pid;
+ 		SpinLockRelease(&walsnd->mutex);
+ 
+ 		if (walsndpid == 0)
+ 			continue;
+ 
+ 		/* 
+ 		 * Check to see whether this standby has fallen behind more than
+ 		 * the upper limit.
+ 		 *
+ 		 * This code is based on XLogCheckpointNeeded().
+ 		 */
+ 		old_segno = (oldptr.xlogid % XLogSegSize) * XLogSegsPerFile +
+ 			(oldptr.xrecoff / XLogSegSize);
+ 		old_highbits = oldptr.xlogid / XLogSegSize;
+ 		if (!found)
+ 		{
+ 			new_segno = (newptr.xlogid % XLogSegSize) * XLogSegsPerFile +
+ 				(newptr.xrecoff / XLogSegSize);
+ 			new_highbits = newptr.xlogid / XLogSegSize;
+ 			found = true;
+ 		}
+ 
+ 		if ((new_highbits != old_highbits ||
+ 			 new_segno >= old_segno + (uint32) (RepLagSegs - 1)) &&
+ 			walsndpid == walsnd->pid)
+ 			kill(walsndpid, SIGTERM);
+ 		/*
+ 		 * XXX: Should we recycle (or remove) old log files here? Otherwise
+ 		 * the number of log files would continue to increase until the next
+ 		 * checkpoint has recycled them. Which increases the chance of a disk
+ 		 * full failure. But it's not good to recycle log files during
+ 		 * acquiring WALWriteLock.
+ 		 */
+ 
+ 		/* standby is keeping up well */
+ 	}
+ }
*** a/src/backend/utils/misc/guc.c
--- b/src/backend/utils/misc/guc.c
***************
*** 1721,1726 **** static struct config_int ConfigureNamesInt[] =
--- 1721,1735 ----
  	},
  
  	{
+ 		{"replication_lag_segments", PGC_SIGHUP, WAL_REPLICATION,
+ 			gettext_noop("Sets the maximum number of WAL files held for the standby server."),
+ 			NULL
+ 		},
+ 		&RepLagSegs,
+ 		0, 0, INT_MAX, NULL, NULL
+ 	},
+ 
+ 	{
  		{"commit_delay", PGC_USERSET, WAL_SETTINGS,
  			gettext_noop("Sets the delay in microseconds between transaction commit and "
  						 "flushing WAL to disk."),
*** a/src/backend/utils/misc/postgresql.conf.sample
--- b/src/backend/utils/misc/postgresql.conf.sample
***************
*** 189,194 ****
--- 189,195 ----
  
  #max_wal_senders = 0		# max number of walsender processes
  #wal_sender_delay = 200ms	# 1-10000 milliseconds
+ #replication_lag_segments = 0	# in logfile segments, 16MB each; 0 disables
  
  
  #------------------------------------------------------------------------------
*** a/src/include/access/xlog_internal.h
--- b/src/include/access/xlog_internal.h
***************
*** 151,156 **** typedef XLogLongPageHeaderData *XLogLongPageHeader;
--- 151,170 ----
  		} \
  	} while (0)
  
+ /* Decrement an xlogid/segment pair by segs */
+ #define PrevLogSegs(logId, logSeg, segs)	\
+ 	do {	\
+ 		logId	-= segs / XLogSegsPerFile;	\
+ 		logSeg	-= segs % XLogSegsPerFile;	\
+ 		if (logSeg < 0)	\
+ 		{	\
+ 			logId--;	\
+ 			logSeg	+= XLogSegsPerFile;	\
+ 		}	\
+ 		if (logId < 0)	\
+ 			logId = logSeg = 0;	\
+ 	} while (0)
+ 
  /* Align a record pointer to next page */
  #define NextLogPage(recptr)	\
  	do {	\
*** a/src/include/replication/walsender.h
--- b/src/include/replication/walsender.h
***************
*** 39,49 **** extern bool	am_walsender;
--- 39,51 ----
  
  /* user-settable parameters */
  extern int	WalSndDelay;
+ extern int	RepLagSegs;
  
  extern int WalSenderMain(void);
  extern void WalSndSignals(void);
  extern Size WalSndShmemSize(void);
  extern void WalSndShmemInit(void);
  extern XLogRecPtr GetOldestWALSendPointer(void);
+ extern void CheckStandbysDelay(XLogRecPtr newptr);
  
  #endif	/* _WALSENDER_H */