replication_timeout_v4.patch

*** a/doc/src/sgml/config.sgml
--- b/doc/src/sgml/config.sgml
***************
*** 2015,2020 **** SET ENABLE_SEQSCAN TO OFF;
--- 2015,2044 ----
         </para>
        </listitem>
       </varlistentry>
+ 
+      <varlistentry id="guc-replication-timeout" xreflabel="replication_timeout">
+       <term><varname>replication_timeout</varname> (<type>integer</type>)</term>
+       <indexterm>
+        <primary><varname>replication_timeout</> configuration parameter</primary>
+       </indexterm>
+       <listitem>
+        <para>
+         Specifies the maximum time, in milliseconds, to wait for the reply
+         from the standby before terminating replication.  This is useful for
+         the primary server to detect the standby crash or network outage.
+         A value of zero (the default) turns this off.  This parameter can
+         only be set in the <filename>postgresql.conf</> file or on the server
+         command line.
+        </para>
+        <para>
+         To make the timeout work properly, <xref linkend="guc-wal-receiver-status-interval">
+         must be enabled on the standby, and its value must be less than the
+         value of <varname>replication_timeout</>.
+         If <varname>wal_receiver_status_interval</> is zero on the standby,
+         replication timeout is disabled for that connection.
+        </para>
+       </listitem>
+      </varlistentry>
       </variablelist>
      </sect2>
  
***************
*** 2125,2130 **** SET ENABLE_SEQSCAN TO OFF;
--- 2149,2159 ----
         the <filename>postgresql.conf</> file or on the server command line.
         The default value is 10 seconds.
        </para>
+       <para>
+        When <xref linkend="guc-replication-timeout"> is enabled on the primary,
+        <varname>wal_receiver_status_interval</> must be enabled, and its value
+        must be less than the value of <varname>replication_timeout</>.
+       </para>
        </listitem>
       </varlistentry>
  
*** a/doc/src/sgml/protocol.sgml
--- b/doc/src/sgml/protocol.sgml
***************
*** 1480,1485 **** The commands accepted in walsender mode are:
--- 1480,1533 ----
        <variablelist>
        <varlistentry>
        <term>
+           Standby parameter change (F)
+       </term>
+       <listitem>
+       <para>
+       <variablelist>
+       <varlistentry>
+       <term>
+           Byte1('g')
+       </term>
+       <listitem>
+       <para>
+           Identifies the message as a receiver parameter change.
+       </para>
+       </listitem>
+       </varlistentry>
+       <varlistentry>
+       <term>
+           Int32
+       </term>
+       <listitem>
+       <para>
+           The current value of wal_receiver_status_interval
+           in the standby.
+       </para>
+       </listitem>
+       </varlistentry>
+       <varlistentry>
+       <term>
+           Byte8
+       </term>
+       <listitem>
+       <para>
+           The server's system clock at the time of transmission,
+           given in TimestampTz format.
+       </para>
+       </listitem>
+       </varlistentry>
+       </variablelist>
+       </para>
+       </listitem>
+       </varlistentry>
+       </variablelist>
+      </para>
+ 
+      <para>
+       <variablelist>
+       <varlistentry>
+       <term>
            Standby status update (F)
        </term>
        <listitem>
*** a/src/backend/libpq/pqcomm.c
--- b/src/backend/libpq/pqcomm.c
***************
*** 56,61 ****
--- 56,64 ----
   *		pq_putbytes		- send bytes to connection (not flushed until pq_flush)
   *		pq_flush		- flush pending output
   *		pq_getbyte_if_available - get a byte if available without blocking
+  *		pq_putbytes_if_writable	- send bytes to connection if writable without blocking
+  *		pq_flush_if_writable	- flush pending output if writable without blocking
+  *		pq_set_nonblocking	- set socket blocking/non-blocking
   *
   * message-level I/O (and old-style-COPY-OUT cruft):
   *		pq_putmessage	- send a normal message (suppressed in COPY OUT mode)
***************
*** 112,117 **** static char sock_path[MAXPGPATH];
--- 115,121 ----
  
  static char PqSendBuffer[PQ_BUFFER_SIZE];
  static int	PqSendPointer;		/* Next index to store a byte in PqSendBuffer */
+ static int	PqSendStart;		/* Next index to send a byte in PqSendBuffer */
  
  static char PqRecvBuffer[PQ_BUFFER_SIZE];
  static int	PqRecvPointer;		/* Next index to read a byte from PqRecvBuffer */
***************
*** 126,133 **** static bool DoingCopyOut;
  
  /* Internal functions */
  static void pq_close(int code, Datum arg);
! static int	internal_putbytes(const char *s, size_t len);
! static int	internal_flush(void);
  
  #ifdef HAVE_UNIX_SOCKETS
  static int	Lock_AF_UNIX(unsigned short portNumber, char *unixSocketName);
--- 130,138 ----
  
  /* Internal functions */
  static void pq_close(int code, Datum arg);
! static int	internal_putbytes(const char *s, size_t len, bool nonblocking);
! static int	internal_flush(bool nonblocking);
! static void pq_set_nonblocking(bool nonblocking, int emode);
  
  #ifdef HAVE_UNIX_SOCKETS
  static int	Lock_AF_UNIX(unsigned short portNumber, char *unixSocketName);
***************
*** 142,148 **** static int	Setup_AF_UNIX(void);
  void
  pq_init(void)
  {
! 	PqSendPointer = PqRecvPointer = PqRecvLength = 0;
  	PqCommBusy = false;
  	DoingCopyOut = false;
  	on_proc_exit(pq_close, 0);
--- 147,153 ----
  void
  pq_init(void)
  {
! 	PqSendPointer = PqSendStart = PqRecvPointer = PqRecvLength = 0;
  	PqCommBusy = false;
  	DoingCopyOut = false;
  	on_proc_exit(pq_close, 0);
***************
*** 846,859 **** pq_getbyte_if_available(unsigned char *c)
  	}
  
  	/* Temporarily put the socket into non-blocking mode */
! #ifdef WIN32
! 	pgwin32_noblock = 1;
! #else
! 	if (!pg_set_noblock(MyProcPort->sock))
! 		ereport(ERROR,
! 				(errmsg("could not set socket to non-blocking mode: %m")));
! #endif
! 	MyProcPort->noblock = true;
  	PG_TRY();
  	{
  		r = secure_read(MyProcPort, c, 1);
--- 851,857 ----
  	}
  
  	/* Temporarily put the socket into non-blocking mode */
! 	pq_set_nonblocking(true, ERROR);
  	PG_TRY();
  	{
  		r = secure_read(MyProcPort, c, 1);
***************
*** 892,916 **** pq_getbyte_if_available(unsigned char *c)
  		 * The rest of the backend code assumes the socket is in blocking
  		 * mode, so treat failure as FATAL.
  		 */
! #ifdef WIN32
! 		pgwin32_noblock = 0;
! #else
! 		if (!pg_set_block(MyProcPort->sock))
! 			ereport(FATAL,
! 					(errmsg("could not set socket to blocking mode: %m")));
! #endif
! 		MyProcPort->noblock = false;
  		PG_RE_THROW();
  	}
  	PG_END_TRY();
! #ifdef WIN32
! 	pgwin32_noblock = 0;
! #else
! 	if (!pg_set_block(MyProcPort->sock))
! 		ereport(FATAL,
! 				(errmsg("could not set socket to blocking mode: %m")));
! #endif
! 	MyProcPort->noblock = false;
  
  	return r;
  }
--- 890,900 ----
  		 * The rest of the backend code assumes the socket is in blocking
  		 * mode, so treat failure as FATAL.
  		 */
! 		pq_set_nonblocking(false, FATAL);
  		PG_RE_THROW();
  	}
  	PG_END_TRY();
! 	pq_set_nonblocking(false, FATAL);
  
  	return r;
  }
***************
*** 1125,1146 **** pq_putbytes(const char *s, size_t len)
  	if (PqCommBusy)
  		return 0;
  	PqCommBusy = true;
! 	res = internal_putbytes(s, len);
  	PqCommBusy = false;
  	return res;
  }
  
  static int
! internal_putbytes(const char *s, size_t len)
  {
  	size_t		amount;
  
  	while (len > 0)
  	{
  		/* If buffer is full, then flush it out */
  		if (PqSendPointer >= PQ_BUFFER_SIZE)
! 			if (internal_flush())
  				return EOF;
  		amount = PQ_BUFFER_SIZE - PqSendPointer;
  		if (amount > len)
  			amount = len;
--- 1109,1169 ----
  	if (PqCommBusy)
  		return 0;
  	PqCommBusy = true;
! 	res = internal_putbytes(s, len, false);
! 	PqCommBusy = false;
! 	return (res == EOF) ? EOF : 0;
! }
! 
! /* --------------------------------
!  *		pq_putbytes_if_writable - send bytes to connection (not flushed
!  *			until pq_flush), if writable
!  *
!  * Returns the number of bytes written without blocking, or EOF if trouble.
!  * --------------------------------
!  */
! int
! pq_putbytes_if_writable(const char *s, size_t len)
! {
! 	int			res;
! 
! 	/* Should not be called by old-style COPY OUT */
! 	Assert(!DoingCopyOut);
! 	/* No-op if reentrant call */
! 	if (PqCommBusy)
! 		return 0;
! 	PqCommBusy = true;
! 	res = internal_putbytes(s, len, true);
  	PqCommBusy = false;
  	return res;
  }
  
+ /* --------------------------------
+  *		internal_putbytes - send bytes to connection (not flushed
+  *			until pq_flush)
+  *
+  * Returns the number of bytes written (can be less than len only if
+  * nonblocking is true), or EOF if trouble.
+  * --------------------------------
+  */
  static int
! internal_putbytes(const char *s, size_t len, bool nonblocking)
  {
  	size_t		amount;
+ 	size_t		nwritten = 0;
  
  	while (len > 0)
  	{
  		/* If buffer is full, then flush it out */
  		if (PqSendPointer >= PQ_BUFFER_SIZE)
! 		{
! 			int		r;
! 
! 			r = internal_flush(nonblocking);
! 			if (r == 0)	/* Only possible if nonblocking is true */
! 				break;
! 			if (r == EOF)
  				return EOF;
+ 		}
  		amount = PQ_BUFFER_SIZE - PqSendPointer;
  		if (amount > len)
  			amount = len;
***************
*** 1148,1155 **** internal_putbytes(const char *s, size_t len)
  		PqSendPointer += amount;
  		s += amount;
  		len -= amount;
  	}
! 	return 0;
  }
  
  /* --------------------------------
--- 1171,1179 ----
  		PqSendPointer += amount;
  		s += amount;
  		len -= amount;
+ 		nwritten += amount;
  	}
! 	return (int) nwritten;
  }
  
  /* --------------------------------
***************
*** 1167,1227 **** pq_flush(void)
  	if (PqCommBusy)
  		return 0;
  	PqCommBusy = true;
! 	res = internal_flush();
  	PqCommBusy = false;
  	return res;
  }
  
  static int
! internal_flush(void)
  {
  	static int	last_reported_send_errno = 0;
  
! 	char	   *bufptr = PqSendBuffer;
  	char	   *bufend = PqSendBuffer + PqSendPointer;
  
! 	while (bufptr < bufend)
  	{
! 		int			r;
! 
! 		r = secure_write(MyProcPort, bufptr, bufend - bufptr);
! 
! 		if (r <= 0)
  		{
! 			if (errno == EINTR)
! 				continue;		/* Ok if we were interrupted */
  
! 			/*
! 			 * Careful: an ereport() that tries to write to the client would
! 			 * cause recursion to here, leading to stack overflow and core
! 			 * dump!  This message must go *only* to the postmaster log.
! 			 *
! 			 * If a client disconnects while we're in the midst of output, we
! 			 * might write quite a bit of data before we get to a safe query
! 			 * abort point.  So, suppress duplicate log messages.
! 			 */
! 			if (errno != last_reported_send_errno)
  			{
! 				last_reported_send_errno = errno;
! 				ereport(COMMERROR,
! 						(errcode_for_socket_access(),
! 						 errmsg("could not send data to client: %m")));
  			}
  
! 			/*
! 			 * We drop the buffered data anyway so that processing can
! 			 * continue, even though we'll probably quit soon.
! 			 */
! 			PqSendPointer = 0;
! 			return EOF;
  		}
- 
- 		last_reported_send_errno = 0;	/* reset after any successful send */
- 		bufptr += r;
  	}
  
! 	PqSendPointer = 0;
! 	return 0;
  }
  
  
--- 1191,1352 ----
  	if (PqCommBusy)
  		return 0;
  	PqCommBusy = true;
! 	res = internal_flush(false);
! 	PqCommBusy = false;
! 	return (res == EOF) ? EOF : 0;
! }
! 
! /* --------------------------------
!  *		pq_flush_if_writable - flush pending output if writable
!  *
!  * Returns 1 if OK, 0 if pending output cannot be written without blocking,
!  * or EOF if trouble.
!  * --------------------------------
!  */
! int
! pq_flush_if_writable(void)
! {
! 	int			res;
! 
! 	/* No-op if reentrant call */
! 	if (PqCommBusy)
! 		return 0;
! 	PqCommBusy = true;
! 	res = internal_flush(true);
  	PqCommBusy = false;
  	return res;
  }
  
+ /* --------------------------------
+  *		internal_flush - flush pending output
+  *
+  * Returns 1 if OK, 0 if pending output cannot be written without blocking
+  * (only possible if nonblocking is true), or EOF if trouble.
+  * --------------------------------
+  */
  static int
! internal_flush(bool nonblocking)
  {
  	static int	last_reported_send_errno = 0;
+ 	int		r;
  
! 	char	   *bufptr = PqSendBuffer + PqSendStart;
  	char	   *bufend = PqSendBuffer + PqSendPointer;
  
! 	/* Temporarily put the socket into non-blocking mode */
! 	if (nonblocking)
! 		pq_set_nonblocking(true, ERROR);
! 	PG_TRY();
  	{
! 		while (bufptr < bufend)
  		{
! 			r = secure_write(MyProcPort, bufptr, bufend - bufptr);
  
! 			if (r <= 0)
  			{
! 				/* Ok if we were interrupted in blocking mode */
! 				if (!nonblocking && errno == EINTR)
! 					continue;
! 
! 				if (nonblocking)
! 				{
! 					if (r == 0)
! 						r = EOF;	/* EOF detected */
! 					else if (errno == EAGAIN ||
! 							 errno == EWOULDBLOCK ||
! 							 errno == EINTR)
! 					{
! 						/*
! 						 * Ok if no data writable without blocking or
! 						 * interrupted (though EINTR really shouldn't
! 						 * happen with a non-blocking socket). Report
! 						 * other errors.
! 						 */
! 						r = 0;
! 					}
! 					break;
! 				}
! 
! 				/*
! 				 * Careful: an ereport() that tries to write to the
! 				 * client would cause recursion to here, leading to
! 				 * stack overflow and core dump!  This message must
! 				 * go *only* to the postmaster log.
! 				 *
! 				 * If a client disconnects while we're in the midst
! 				 * of output, we might write quite a bit of data before
! 				 * we get to a safe query abort point.  So, suppress
! 				 * duplicate log messages.
! 				 */
! 				if (errno != last_reported_send_errno)
! 				{
! 					last_reported_send_errno = errno;
! 					ereport(COMMERROR,
! 							(errcode_for_socket_access(),
! 							 errmsg("could not send data to client: %m")));
! 				}
! 
! 				/*
! 				 * We drop the buffered data anyway so that processing can
! 				 * continue, even though we'll probably quit soon.
! 				 */
! 				PqSendStart = PqSendPointer = 0;
! 				r = EOF;
! 				break;
  			}
  
! 			last_reported_send_errno = 0;	/* reset after any successful send */
! 			bufptr += r;
! 			PqSendStart += r;
  		}
  	}
+ 	PG_CATCH();
+ 	{
+ 		/*
+ 		 * The rest of the backend code assumes the socket is in blocking
+ 		 * mode, so treat failure as FATAL.
+ 		 */
+ 		if (nonblocking)
+ 			pq_set_nonblocking(false, FATAL);
+ 	}
+ 	PG_END_TRY();
+ 	if (nonblocking)
+ 		pq_set_nonblocking(false, FATAL);
  
! 	if (r == 0 || r == EOF)
! 		return r;
! 
! 	PqSendStart = PqSendPointer = 0;
! 	return 1;
! }
! 
! /* --------------------------------
!  *		pq_set_nonblocking - set socket blocking/non-blocking
!  *
!  * Sets the socket non-blocking if nonblocking is TRUE, or sets it
!  * blocking otherwise.
!  * --------------------------------
!  */
! static
! void pq_set_nonblocking(bool nonblocking, int emode)
! {
! #ifdef WIN32
! 	pgwin32_noblock = nonblocking ? 1 : 0;
! #else
! 	if (nonblocking)
! 	{
! 		if (!pg_set_noblock(MyProcPort->sock))
! 			ereport(emode,
! 					(errmsg("could not set socket to non-blocking mode: %m")));
! 	}
! 	else
! 	{
! 		if (!pg_set_block(MyProcPort->sock))
! 			ereport(emode,
! 					(errmsg("could not set socket to blocking mode: %m")));
! 	}
! #endif
! 	MyProcPort->noblock = nonblocking;
  }
  
  
***************
*** 1265,1281 **** pq_putmessage(char msgtype, const char *s, size_t len)
  		return 0;
  	PqCommBusy = true;
  	if (msgtype)
! 		if (internal_putbytes(&msgtype, 1))
  			goto fail;
  	if (PG_PROTOCOL_MAJOR(FrontendProtocol) >= 3)
  	{
  		uint32		n32;
  
  		n32 = htonl((uint32) (len + 4));
! 		if (internal_putbytes((char *) &n32, 4))
  			goto fail;
  	}
! 	if (internal_putbytes(s, len))
  		goto fail;
  	PqCommBusy = false;
  	return 0;
--- 1390,1406 ----
  		return 0;
  	PqCommBusy = true;
  	if (msgtype)
! 		if (internal_putbytes(&msgtype, 1, false) == EOF)
  			goto fail;
  	if (PG_PROTOCOL_MAJOR(FrontendProtocol) >= 3)
  	{
  		uint32		n32;
  
  		n32 = htonl((uint32) (len + 4));
! 		if (internal_putbytes((char *) &n32, 4, false) == EOF)
  			goto fail;
  	}
! 	if (internal_putbytes(s, len, false) == EOF)
  		goto fail;
  	PqCommBusy = false;
  	return 0;
*** a/src/backend/port/unix_latch.c
--- b/src/backend/port/unix_latch.c
***************
*** 193,211 **** DisownLatch(volatile Latch *latch)
  bool
  WaitLatch(volatile Latch *latch, long timeout)
  {
! 	return WaitLatchOrSocket(latch, PGINVALID_SOCKET, timeout) > 0;
  }
  
  /*
   * Like WaitLatch, but will also return when there's data available in
!  * 'sock' for reading. Returns 0 if timeout was reached, 1 if the latch
!  * was set, or 2 if the scoket became readable.
   */
  int
! WaitLatchOrSocket(volatile Latch *latch, pgsocket sock, long timeout)
  {
  	struct timeval tv, *tvp = NULL;
  	fd_set		input_mask;
  	int			rc;
  	int			result = 0;
  
--- 193,214 ----
  bool
  WaitLatch(volatile Latch *latch, long timeout)
  {
! 	return WaitLatchOrSocket(latch, PGINVALID_SOCKET, false, false, timeout) > 0;
  }
  
  /*
   * Like WaitLatch, but will also return when there's data available in
!  * 'sock' for reading or writing. Returns 0 if timeout was reached,
!  * 1 if the latch was set, 2 if the scoket became readable, or 3 if
!  * the socket became writable.
   */
  int
! WaitLatchOrSocket(volatile Latch *latch, pgsocket sock, bool forRead,
! 				  bool forWrite, long timeout)
  {
  	struct timeval tv, *tvp = NULL;
  	fd_set		input_mask;
+ 	fd_set		output_mask;
  	int			rc;
  	int			result = 0;
  
***************
*** 241,254 **** WaitLatchOrSocket(volatile Latch *latch, pgsocket sock, long timeout)
  		FD_ZERO(&input_mask);
  		FD_SET(selfpipe_readfd, &input_mask);
  		hifd = selfpipe_readfd;
! 		if (sock != PGINVALID_SOCKET)
  		{
  			FD_SET(sock, &input_mask);
  			if (sock > hifd)
  				hifd = sock;
  		}
  
! 		rc = select(hifd + 1, &input_mask, NULL, NULL, tvp);
  		if (rc < 0)
  		{
  			if (errno == EINTR)
--- 244,265 ----
  		FD_ZERO(&input_mask);
  		FD_SET(selfpipe_readfd, &input_mask);
  		hifd = selfpipe_readfd;
! 		if (sock != PGINVALID_SOCKET && forRead)
  		{
  			FD_SET(sock, &input_mask);
  			if (sock > hifd)
  				hifd = sock;
  		}
  
! 		FD_ZERO(&output_mask);
! 		if (sock != PGINVALID_SOCKET && forWrite)
! 		{
! 			FD_SET(sock, &output_mask);
! 			if (sock > hifd)
! 				hifd = sock;
! 		}
! 
! 		rc = select(hifd + 1, &input_mask, &output_mask, NULL, tvp);
  		if (rc < 0)
  		{
  			if (errno == EINTR)
***************
*** 263,273 **** WaitLatchOrSocket(volatile Latch *latch, pgsocket sock, long timeout)
  			result = 0;
  			break;
  		}
! 		if (sock != PGINVALID_SOCKET && FD_ISSET(sock, &input_mask))
  		{
  			result = 2;
  			break;		/* data available in socket */
  		}
  	}
  	waiting = false;
  
--- 274,291 ----
  			result = 0;
  			break;
  		}
! 		if (sock != PGINVALID_SOCKET && forRead &&
! 			FD_ISSET(sock, &input_mask))
  		{
  			result = 2;
  			break;		/* data available in socket */
  		}
+ 		if (sock != PGINVALID_SOCKET && forWrite &&
+ 			FD_ISSET(sock, &output_mask))
+ 		{
+ 			result = 3;
+ 			break;		/* data writable in socket */
+ 		}
  	}
  	waiting = false;
  
*** a/src/backend/port/win32/socket.c
--- b/src/backend/port/win32/socket.c
***************
*** 14,20 ****
  #include "postgres.h"
  
  /*
!  * Indicate if pgwin32_recv() should operate in non-blocking mode.
   *
   * Since the socket emulation layer always sets the actual socket to
   * non-blocking mode in order to be able to deliver signals, we must
--- 14,21 ----
  #include "postgres.h"
  
  /*
!  * Indicate if pgwin32_recv() and pgwin32_send() should operate
!  * in non-blocking mode.
   *
   * Since the socket emulation layer always sets the actual socket to
   * non-blocking mode in order to be able to deliver signals, we must
***************
*** 399,404 **** pgwin32_send(SOCKET s, char *buf, int len, int flags)
--- 400,415 ----
  			return -1;
  		}
  
+ 		if (pgwin32_noblock)
+ 		{
+ 			/*
+ 			 * No data sent, and we are in "emulated non-blocking mode", so
+ 			 * return indicating that we'd block if we were to continue.
+ 			 */
+ 			errno = EWOULDBLOCK;
+ 			return -1;
+ 		}
+ 
  		/* No error, zero bytes (win2000+) or error+WSAEWOULDBLOCK (<=nt4) */
  
  		if (pgwin32_waitforsinglesocket(s, FD_WRITE | FD_CLOSE, INFINITE) == 0)
*** a/src/backend/port/win32_latch.c
--- b/src/backend/port/win32_latch.c
***************
*** 85,95 **** DisownLatch(volatile Latch *latch)
  bool
  WaitLatch(volatile Latch *latch, long timeout)
  {
! 	return WaitLatchOrSocket(latch, PGINVALID_SOCKET, timeout) > 0;
  }
  
  int
! WaitLatchOrSocket(volatile Latch *latch, SOCKET sock, long timeout)
  {
  	DWORD		rc;
  	HANDLE		events[3];
--- 85,96 ----
  bool
  WaitLatch(volatile Latch *latch, long timeout)
  {
! 	return WaitLatchOrSocket(latch, PGINVALID_SOCKET, false, false, timeout) > 0;
  }
  
  int
! WaitLatchOrSocket(volatile Latch *latch, SOCKET sock, bool forRead,
! 				  bool forWrite, long timeout)
  {
  	DWORD		rc;
  	HANDLE		events[3];
***************
*** 103,112 **** WaitLatchOrSocket(volatile Latch *latch, SOCKET sock, long timeout)
  	events[0] = latchevent;
  	events[1] = pgwin32_signal_event;
  	numevents = 2;
! 	if (sock != PGINVALID_SOCKET)
  	{
  		sockevent = WSACreateEvent();
! 		WSAEventSelect(sock, sockevent, FD_READ);
  		events[numevents++] = sockevent;
  	}
  
--- 104,120 ----
  	events[0] = latchevent;
  	events[1] = pgwin32_signal_event;
  	numevents = 2;
! 	if (sock != PGINVALID_SOCKET && (forRead || forWrite))
  	{
+ 		int		flags = 0;
+ 
+ 		if (forRead)
+ 			flags |= FD_READ;
+ 		if (forWrite)
+ 			flags |= FD_WRITE;
+ 
  		sockevent = WSACreateEvent();
! 		WSAEventSelect(sock, sockevent, flags);
  		events[numevents++] = sockevent;
  	}
  
***************
*** 139,146 **** WaitLatchOrSocket(volatile Latch *latch, SOCKET sock, long timeout)
  			pgwin32_dispatch_queued_signals();
  		else if (rc == WAIT_OBJECT_0 + 2)
  		{
  			Assert(sock != PGINVALID_SOCKET);
! 			result = 2;
  			break;
  		}
  		else if (rc != WAIT_OBJECT_0)
--- 147,165 ----
  			pgwin32_dispatch_queued_signals();
  		else if (rc == WAIT_OBJECT_0 + 2)
  		{
+ 			WSANETWORKEVENTS resEvents;
+ 
  			Assert(sock != PGINVALID_SOCKET);
! 
! 			ZeroMemory(&resEvents, sizeof(resEvents));
! 			if (WSAEnumNetworkEvents(sock, sockevent, &resEvents) == SOCKET_ERROR)
! 				ereport(FATAL,
! 						(errmsg_internal("failed to enumerate network events: %i", (int) GetLastError())));
! 
! 			if (forRead && resEvents.lNetworkEvents & FD_READ)
! 				result = 2;
! 			if (forWrite && resEvents.lNetworkEvents & FD_WRITE)
! 				result = 3;
  			break;
  		}
  		else if (rc != WAIT_OBJECT_0)
***************
*** 148,154 **** WaitLatchOrSocket(volatile Latch *latch, SOCKET sock, long timeout)
  	}
  
  	/* Clean up the handle we created for the socket */
! 		if (sock != PGINVALID_SOCKET)
  	{
  		WSAEventSelect(sock, sockevent, 0);
  		WSACloseEvent(sockevent);
--- 167,173 ----
  	}
  
  	/* Clean up the handle we created for the socket */
! 	if (sock != PGINVALID_SOCKET && (forRead || forWrite))
  	{
  		WSAEventSelect(sock, sockevent, 0);
  		WSACloseEvent(sockevent);
*** a/src/backend/replication/walreceiver.c
--- b/src/backend/replication/walreceiver.c
***************
*** 125,130 **** static void XLogWalRcvWrite(char *buf, Size nbytes, XLogRecPtr recptr);
--- 125,131 ----
  static void XLogWalRcvFlush(bool dying);
  static void XLogWalRcvSendReply(void);
  static void XLogWalRcvSendHSFeedback(void);
+ static void XLogWalRcvSendGUCChange(void);
  
  /* Signal handlers */
  static void WalRcvSigHupHandler(SIGNAL_ARGS);
***************
*** 276,281 **** WalReceiverMain(void)
--- 277,288 ----
  	walrcv_connect(conninfo, startpoint);
  	DisableWalRcvImmediateExit();
  
+ 	/*
+ 	 * Report the important parameters for streaming replication to
+ 	 * the primary.
+ 	 */
+ 	XLogWalRcvSendGUCChange();
+ 
  	/* Loop until end-of-streaming or error */
  	for (;;)
  	{
***************
*** 303,310 **** WalReceiverMain(void)
--- 310,325 ----
  
  		if (got_SIGHUP)
  		{
+ 			int	save_wal_receiver_status_interval =
+ 				wal_receiver_status_interval;
+ 
  			got_SIGHUP = false;
  			ProcessConfigFile(PGC_SIGHUP);
+ 
+ 			/* If any of important parameters have changed, report them */
+ 			if (save_wal_receiver_status_interval !=
+ 					wal_receiver_status_interval)
+ 				XLogWalRcvSendGUCChange();
  		}
  
  		/* Wait a while for data to arrive */
***************
*** 705,707 **** XLogWalRcvSendHSFeedback(void)
--- 720,749 ----
  	memcpy(&buf[1], &feedback_message, sizeof(StandbyHSFeedbackMessage));
  	walrcv_send(buf, sizeof(StandbyHSFeedbackMessage) + 1);
  }
+ 
+ /*
+  * Send parameter change message to primary, plus the current time.
+  */
+ static void
+ XLogWalRcvSendGUCChange(void)
+ {
+ 	char			buf[sizeof(StandbyGUCChangeMessage) + 1];
+ 	TimestampTz		now;
+ 	StandbyGUCChangeMessage	guc_change_message;
+ 
+ 	/* Get current timestamp. */
+ 	now = GetCurrentTimestamp();
+ 
+ 	/* Construct a new message */
+ 	guc_change_message.wal_receiver_status_interval =
+ 		wal_receiver_status_interval;
+ 	guc_change_message.sendTime = now;
+ 
+ 	elog(DEBUG2, "sending parameter change wal_receiver_status_interval %d",
+ 			 wal_receiver_status_interval);
+ 
+ 	/* Prepend with the message type and send it. */
+ 	buf[0] = 'g';
+ 	memcpy(&buf[1], &guc_change_message, sizeof(StandbyGUCChangeMessage));
+ 	walrcv_send(buf, sizeof(StandbyGUCChangeMessage) + 1);
+ }
*** a/src/backend/replication/walsender.c
--- b/src/backend/replication/walsender.c
***************
*** 74,79 **** bool		am_walsender = false;		/* Am I a walsender process ? */
--- 74,91 ----
  /* User-settable parameters for walsender */
  int			max_wal_senders = 0;	/* the maximum number of concurrent walsenders */
  int			WalSndDelay = 1000;	/* max sleep time between some actions */
+ int			replication_timeout = 0;	/* maximum time to send one WAL data message */
+ 
+ /*
+  * Buffer for WAL sending
+  *
+  * WalSndOutBuffer is a work area in which the output message is constructed.
+  * It's used in just so we can avoid re-palloc'ing the buffer on each cycle.
+  * It must be of size 6 + sizeof(WalDataMessageHeader) + MAX_SEND_SIZE.
+  */
+ static char	   *WalSndOutBuffer;
+ static int		WalSndOutHead;		/* head of pending output */
+ static int		WalSndOutTail;		/* tail of pending output */
  
  /*
   * These variables are used similarly to openLogFile/Id/Seg/Off,
***************
*** 95,100 **** static XLogRecPtr sentPtr = {0, 0};
--- 107,123 ----
   */
  static StringInfoData reply_message;
  
+ /*
+  * Timestamp of the last receipt of the reply from the standby.
+  */
+ static TimestampTz last_reply_timestamp;
+ 
+ /*
+  * The value of wal_receiver_status_interval on the standby.
+  * If this is zero, we disable replication timeout.
+  */
+ static int	standby_wal_receiver_status_interval = 10;
+ 
  /* Flags set by signal handlers for later service in main loop */
  static volatile sig_atomic_t got_SIGHUP = false;
  volatile sig_atomic_t walsender_shutdown_requested = false;
***************
*** 113,124 **** static int	WalSndLoop(void);
  static void InitWalSnd(void);
  static void WalSndHandshake(void);
  static void WalSndKill(int code, Datum arg);
! static bool XLogSend(char *msgbuf, bool *caughtup);
  static void IdentifySystem(void);
  static void StartReplication(StartReplicationCmd * cmd);
  static void ProcessStandbyMessage(void);
  static void ProcessStandbyReplyMessage(void);
  static void ProcessStandbyHSFeedbackMessage(void);
  static void ProcessRepliesIfAny(void);
  
  
--- 136,149 ----
  static void InitWalSnd(void);
  static void WalSndHandshake(void);
  static void WalSndKill(int code, Datum arg);
! static bool XLogSend(bool *caughtup, bool *pending);
  static void IdentifySystem(void);
  static void StartReplication(StartReplicationCmd * cmd);
  static void ProcessStandbyMessage(void);
  static void ProcessStandbyReplyMessage(void);
  static void ProcessStandbyHSFeedbackMessage(void);
+ static void ProcessStandbyGUCChangeMessage(void);
+ static void ValidateReplicationTimeout(bool standby);
  static void ProcessRepliesIfAny(void);
  
  
***************
*** 214,219 **** WalSndHandshake(void)
--- 239,250 ----
  		{
  			got_SIGHUP = false;
  			ProcessConfigFile(PGC_SIGHUP);
+ 
+ 			/*
+ 			 * Don't need to verify replication_timeout here because we
+ 			 * don't receive any parameter change message until we enter
+ 			 * streaming mode.
+ 			 */
  		}
  
  		if (firstchar != EOF)
***************
*** 467,472 **** ProcessRepliesIfAny(void)
--- 498,504 ----
  {
  	unsigned char firstchar;
  	int			r;
+ 	int		received = false;
  
  	for (;;)
  	{
***************
*** 479,487 **** ProcessRepliesIfAny(void)
  					 errmsg("unexpected EOF on standby connection")));
  			proc_exit(0);
  		}
! 		if (r == 0)
  		{
! 			/* no data available without blocking */
  			return;
  		}
  
--- 511,524 ----
  					 errmsg("unexpected EOF on standby connection")));
  			proc_exit(0);
  		}
! 		if (r == 0)	/* no data available without blocking */
  		{
! 			/*
! 			 * Save the last reply timestamp if we've received at least
! 			 * one reply.
! 			 */
! 			if (received)
! 				last_reply_timestamp = GetCurrentTimestamp();
  			return;
  		}
  
***************
*** 493,498 **** ProcessRepliesIfAny(void)
--- 530,536 ----
  				 */
  			case 'd':
  				ProcessStandbyMessage();
+ 				received = true;
  				break;
  
  				/*
***************
*** 511,517 **** ProcessRepliesIfAny(void)
  }
  
  /*
!  * Process a status update message received from standby.
   */
  static void
  ProcessStandbyMessage(void)
--- 549,555 ----
  }
  
  /*
!  * Process a message received from standby.
   */
  static void
  ProcessStandbyMessage(void)
***************
*** 547,552 **** ProcessStandbyMessage(void)
--- 585,594 ----
  			ProcessStandbyHSFeedbackMessage();
  			break;
  
+ 		case 'g':
+ 			ProcessStandbyGUCChangeMessage();
+ 			break;
+ 
  		default:
  			ereport(COMMERROR,
  					(errcode(ERRCODE_PROTOCOL_VIOLATION),
***************
*** 665,683 **** ProcessStandbyHSFeedbackMessage(void)
  	}
  }
  
  /* Main loop of walsender process */
  static int
  WalSndLoop(void)
  {
- 	char	   *output_message;
  	bool		caughtup = false;
  
  	/*
  	 * Allocate buffer that will be used for each output message.  We do this
  	 * just once to reduce palloc overhead.  The buffer must be made large
  	 * enough for maximum-sized messages.
  	 */
! 	output_message = palloc(1 + sizeof(WalDataMessageHeader) + MAX_SEND_SIZE);
  
  	/*
  	 * Allocate buffer that will be used for processing reply messages.  As
--- 707,805 ----
  	}
  }
  
+ /*
+  * Parameter change report from standby
+  */
+ static void
+ ProcessStandbyGUCChangeMessage(void)
+ {
+ 	StandbyGUCChangeMessage	reply;
+ 
+ 	pq_copymsgbytes(&reply_message, (char *) &reply, sizeof(StandbyGUCChangeMessage));
+ 
+ 	elog(DEBUG2, "standby's wal_receiver_status_interval %d",
+ 			 reply.wal_receiver_status_interval);
+ 
+ 	/* If we don't use replication timeout, we don't need to do any more */
+ 	if (replication_timeout <= 0)
+ 		return;
+ 
+ 	/*
+ 	 * If wal_receiver_status_interval is zero in the standby,
+ 	 * we ignore replication timeout for that connection.
+ 	 */
+ 	if (standby_wal_receiver_status_interval > 0 &&
+ 			reply.wal_receiver_status_interval <= 0)
+ 	{
+ 		ereport(LOG,
+ 						(errmsg("replication timeout is disabled because wal_receiver_status_interval is zero on the standby")));
+ 	}
+ 	else if (standby_wal_receiver_status_interval <= 0 &&
+ 					 reply.wal_receiver_status_interval > 0)
+ 	{
+ 		ereport(LOG,
+ 						(errmsg("replication timeout is enabled because wal_receiver_status_interval is greater than zero on the standby")));
+ 	}
+ 
+ 	standby_wal_receiver_status_interval = reply.wal_receiver_status_interval;
+ 
+ 	/* Verify whether replication_timeout is large enough */
+ 	ValidateReplicationTimeout(true);
+ }
+ 
+ /*
+  * Verify whether replication_timeout is large enough for the timeout.
+  *
+  * This is called when either replication_timeout is changed on the
+  * master or wal_receiver_status_interval is changed on the standby.
+  * In the former case, 'standby' must be set to FALSE, otherwise TRUE.
+  */
+ static void
+ ValidateReplicationTimeout(bool standby)
+ {
+ 	static bool	skip_validation = true;
+ 
+ 	/*
+ 	 * If we've not received any parameter change message from the standby
+ 	 * yet, we cannot verify replication_timeout properly. In that case,
+ 	 * we postpone the validation until the message has arrived. This can
+ 	 * happen only when replication_timeout is changed by SIGHUP signal
+ 	 * before any parameter change message arrives.
+ 	 */
+ 	if (standby)
+ 		skip_validation = false;
+ 
+ 	if (skip_validation)
+ 		return;
+ 
+ 	/*
+ 	 * Emit WARNING message if replication_timeout in the primary
+ 	 * is less than wal_receiver_status_interval in the standby
+ 	 * because unexpected timeout can happen in that case.
+ 	 */
+ 	if (replication_timeout > 0 &&
+ 			standby_wal_receiver_status_interval > 0 &&
+ 			replication_timeout <= standby_wal_receiver_status_interval * 1000)
+ 		ereport(WARNING,
+ 						(errmsg("replication can be terminated unexpectedly because replication_timeout (%d milliseconds) on the master is less than wal_receiver_status_interval (%d seconds) on the standby",
+ 										replication_timeout, standby_wal_receiver_status_interval),
+ 						 errhint("Either increase replication_timeout on the master, or decrease wal_receiver_status_interval on the standby.")));
+ }
+ 
  /* Main loop of walsender process */
  static int
  WalSndLoop(void)
  {
  	bool		caughtup = false;
+ 	bool		pending = false;
  
  	/*
  	 * Allocate buffer that will be used for each output message.  We do this
  	 * just once to reduce palloc overhead.  The buffer must be made large
  	 * enough for maximum-sized messages.
  	 */
! 	WalSndOutBuffer = palloc(6 + sizeof(WalDataMessageHeader) + MAX_SEND_SIZE);
! 	WalSndOutHead = WalSndOutTail = 0;
  
  	/*
  	 * Allocate buffer that will be used for processing reply messages.  As
***************
*** 685,690 **** WalSndLoop(void)
--- 807,815 ----
  	 */
  	initStringInfo(&reply_message);
  
+ 	/* Initialize the last reply timestamp */
+ 	last_reply_timestamp = GetCurrentTimestamp();
+ 
  	/* Loop forever, unless we get an error */
  	for (;;)
  	{
***************
*** 700,705 **** WalSndLoop(void)
--- 825,833 ----
  		{
  			got_SIGHUP = false;
  			ProcessConfigFile(PGC_SIGHUP);
+ 
+ 			/* Verify whether replication_timeout is large enough */
+ 			ValidateReplicationTimeout(false);
  		}
  
  		/*
***************
*** 708,717 **** WalSndLoop(void)
  		 */
  		if (walsender_ready_to_stop)
  		{
! 			if (!XLogSend(output_message, &caughtup))
  				break;
  			ProcessRepliesIfAny();
! 			if (caughtup)
  				walsender_shutdown_requested = true;
  		}
  
--- 836,845 ----
  		 */
  		if (walsender_ready_to_stop)
  		{
! 			if (!XLogSend(&caughtup, &pending))
  				break;
  			ProcessRepliesIfAny();
! 			if (caughtup && !pending)
  				walsender_shutdown_requested = true;
  		}
  
***************
*** 726,735 **** WalSndLoop(void)
  		}
  
  		/*
! 		 * If we had sent all accumulated WAL in last round, nap for the
! 		 * configured time before retrying.
  		 */
! 		if (caughtup)
  		{
  			/*
  			 * Even if we wrote all the WAL that was available when we started
--- 854,864 ----
  		}
  
  		/*
! 		 * If we had sent all accumulated WAL in last round or could not
! 		 * flush pending WAL in output buffer because the socket was not
! 		 * writable, nap for the configured time before retrying.
  		 */
! 		if (caughtup || pending)
  		{
  			/*
  			 * Even if we wrote all the WAL that was available when we started
***************
*** 740,764 **** WalSndLoop(void)
  			 */
  			ResetLatch(&MyWalSnd->latch);
  
! 			if (!XLogSend(output_message, &caughtup))
  				break;
! 			if (caughtup && !got_SIGHUP && !walsender_ready_to_stop && !walsender_shutdown_requested)
  			{
  				/*
  				 * XXX: We don't really need the periodic wakeups anymore,
  				 * WaitLatchOrSocket should reliably wake up as soon as
  				 * something interesting happens.
  				 */
  
  				/* Sleep */
  				WaitLatchOrSocket(&MyWalSnd->latch, MyProcPort->sock,
! 								  WalSndDelay * 1000L);
  			}
  		}
  		else
  		{
  			/* Attempt to send the log once every loop */
! 			if (!XLogSend(output_message, &caughtup))
  				break;
  		}
  
--- 869,931 ----
  			 */
  			ResetLatch(&MyWalSnd->latch);
  
! 			if (!XLogSend(&caughtup, &pending))
  				break;
! 			if ((caughtup || pending) && !got_SIGHUP && !walsender_ready_to_stop &&
! 					!walsender_shutdown_requested)
  			{
+ 				TimestampTz	finish_time;
+ 				long		sleeptime;
+ 
  				/*
  				 * XXX: We don't really need the periodic wakeups anymore,
  				 * WaitLatchOrSocket should reliably wake up as soon as
  				 * something interesting happens.
  				 */
  
+ 				/* Reschedule replication timeout */
+ 				if (replication_timeout > 0 &&
+ 						standby_wal_receiver_status_interval > 0)
+ 				{
+ 					long		secs;
+ 					int		usecs;
+ 
+ 					finish_time = TimestampTzPlusMilliseconds(last_reply_timestamp,
+ 											replication_timeout);
+ 					TimestampDifference(GetCurrentTimestamp(),
+ 								finish_time, &secs, &usecs);
+ 					sleeptime = secs * 1000 + usecs / 1000;
+ 					if (WalSndDelay < sleeptime)
+ 						sleeptime = WalSndDelay;
+ 				}
+ 				else
+ 					sleeptime = WalSndDelay;
+ 
  				/* Sleep */
  				WaitLatchOrSocket(&MyWalSnd->latch, MyProcPort->sock,
! 								  true, (WalSndOutTail > 0),
! 								  sleeptime * 1000L);
! 
! 				/* Check for replication timeout */
! 				if (replication_timeout > 0 &&
! 						standby_wal_receiver_status_interval > 0 &&
! 						GetCurrentTimestamp() >= finish_time)
! 				{
! 					/*
! 					 * Since typically expiration of replication timeout means
! 					 * communication problem, we don't send the error message
! 					 * to the standby.
! 					 */
! 					ereport(COMMERROR,
! 							(errmsg("terminating walsender process due to replication timeout")));
! 					break;
! 				}
  			}
  		}
  		else
  		{
  			/* Attempt to send the log once every loop */
! 			if (!XLogSend(&caughtup, &pending))
  				break;
  		}
  
***************
*** 986,1009 **** XLogRead(char *buf, XLogRecPtr recptr, Size nbytes)
   * Read up to MAX_SEND_SIZE bytes of WAL that's been flushed to disk,
   * but not yet sent to the client, and send it.
   *
-  * msgbuf is a work area in which the output message is constructed.  It's
-  * passed in just so we can avoid re-palloc'ing the buffer on each cycle.
-  * It must be of size 1 + sizeof(WalDataMessageHeader) + MAX_SEND_SIZE.
-  *
   * If there is no unsent WAL remaining, *caughtup is set to true, otherwise
   * *caughtup is set to false.
   *
   * Returns true if OK, false if trouble.
   */
  static bool
! XLogSend(char *msgbuf, bool *caughtup)
  {
  	XLogRecPtr	SendRqstPtr;
  	XLogRecPtr	startptr;
! 	XLogRecPtr	endptr;
  	Size		nbytes;
  	WalDataMessageHeader msghdr;
  
  	/*
  	 * Attempt to send all data that's already been written out and fsync'd to
  	 * disk.  We cannot go further than what's been written out given the
--- 1153,1200 ----
   * Read up to MAX_SEND_SIZE bytes of WAL that's been flushed to disk,
   * but not yet sent to the client, and send it.
   *
   * If there is no unsent WAL remaining, *caughtup is set to true, otherwise
   * *caughtup is set to false.
   *
+  * If there is pending WAL in output buffer, *pending is set to true,
+  * otherwise *pending is set to false.
+  *
   * Returns true if OK, false if trouble.
   */
  static bool
! XLogSend(bool *caughtup, bool *pending)
  {
  	XLogRecPtr	SendRqstPtr;
  	XLogRecPtr	startptr;
! 	static XLogRecPtr	endptr;
  	Size		nbytes;
+ 	uint32		n32;
+ 	int			res;
  	WalDataMessageHeader msghdr;
  
+ 	/* Attempt to flush pending WAL in output buffer */
+ 	if (*pending)
+ 	{
+ 		if (WalSndOutHead != WalSndOutTail)
+ 		{
+ 			res = pq_putbytes_if_writable(WalSndOutBuffer + WalSndOutHead,
+ 										  WalSndOutTail - WalSndOutHead);
+ 			if (res == EOF)
+ 				return false;
+ 			WalSndOutHead += res;
+ 			if (WalSndOutHead != WalSndOutTail)
+ 				return true;
+ 		}
+ 
+ 		res = pq_flush_if_writable();
+ 		if (res == EOF)
+ 			return false;
+ 		if (res == 0)
+ 			return true;
+ 
+ 		goto updt;
+ 	}
+ 
  	/*
  	 * Attempt to send all data that's already been written out and fsync'd to
  	 * disk.  We cannot go further than what's been written out given the
***************
*** 1072,1084 **** XLogSend(char *msgbuf, bool *caughtup)
  	/*
  	 * OK to read and send the slice.
  	 */
! 	msgbuf[0] = 'w';
  
  	/*
  	 * Read the log directly into the output buffer to avoid extra memcpy
  	 * calls.
  	 */
! 	XLogRead(msgbuf + 1 + sizeof(WalDataMessageHeader), startptr, nbytes);
  
  	/*
  	 * We fill the message header last so that the send timestamp is taken as
--- 1263,1281 ----
  	/*
  	 * OK to read and send the slice.
  	 */
! 	WalSndOutBuffer[0] = 'd';
! 	WalSndOutBuffer[5] = 'w';
! 	WalSndOutHead = 0;
! 	WalSndOutTail = 6 + sizeof(WalDataMessageHeader) + nbytes;
! 
! 	n32 = htonl((uint32) WalSndOutTail - 1);
! 	memcpy(WalSndOutBuffer + 1, &n32, 4);
  
  	/*
  	 * Read the log directly into the output buffer to avoid extra memcpy
  	 * calls.
  	 */
! 	XLogRead(WalSndOutBuffer + 6 + sizeof(WalDataMessageHeader), startptr, nbytes);
  
  	/*
  	 * We fill the message header last so that the send timestamp is taken as
***************
*** 1088,1100 **** XLogSend(char *msgbuf, bool *caughtup)
  	msghdr.walEnd = SendRqstPtr;
  	msghdr.sendTime = GetCurrentTimestamp();
  
! 	memcpy(msgbuf + 1, &msghdr, sizeof(WalDataMessageHeader));
  
! 	pq_putmessage('d', msgbuf, 1 + sizeof(WalDataMessageHeader) + nbytes);
  
  	/* Flush pending output to the client */
! 	if (pq_flush())
  		return false;
  
  	sentPtr = endptr;
  
--- 1285,1318 ----
  	msghdr.walEnd = SendRqstPtr;
  	msghdr.sendTime = GetCurrentTimestamp();
  
! 	memcpy(WalSndOutBuffer + 6, &msghdr, sizeof(WalDataMessageHeader));
! 
! 	res = pq_putbytes_if_writable(WalSndOutBuffer, WalSndOutTail);
! 	if (res == EOF)
! 		return false;
  
! 	WalSndOutHead = res;
! 	if (WalSndOutHead != WalSndOutTail)
! 	{
! 		*caughtup = false;
! 		*pending = true;
! 		return true;
! 	}
  
  	/* Flush pending output to the client */
! 	res = pq_flush_if_writable();
! 	if (res == EOF)
  		return false;
+ 	if (res == 0)
+ 	{
+ 		*caughtup = false;
+ 		*pending = true;
+ 		return true;
+ 	}
+ 
+ updt:
+ 	WalSndOutHead = WalSndOutTail = 0;
+ 	*pending = false;
  
  	sentPtr = endptr;
  
*** a/src/backend/utils/misc/guc.c
--- b/src/backend/utils/misc/guc.c
***************
*** 1847,1852 **** static struct config_int ConfigureNamesInt[] =
--- 1847,1862 ----
  	},
  
  	{
+ 		{"replication_timeout", PGC_SIGHUP, WAL_REPLICATION,
+ 			gettext_noop("Sets the maximum time to wait for WAL replication."),
+ 			NULL,
+ 			GUC_UNIT_MS
+ 		},
+ 		&replication_timeout,
+ 		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
***************
*** 191,196 ****
--- 191,197 ----
  #wal_sender_delay = 1s		# walsender cycle time, 1-10000 milliseconds
  #wal_keep_segments = 0		# in logfile segments, 16MB each; 0 disables
  #vacuum_defer_cleanup_age = 0	# number of xacts by which cleanup is delayed
+ #replication_timeout = 0 # in milliseconds, 0 is disabled
  
  # - Standby Servers -
  
*** a/src/include/libpq/libpq.h
--- b/src/include/libpq/libpq.h
***************
*** 59,65 **** extern int	pq_getbyte(void);
--- 59,67 ----
  extern int	pq_peekbyte(void);
  extern int	pq_getbyte_if_available(unsigned char *c);
  extern int	pq_putbytes(const char *s, size_t len);
+ extern int	pq_putbytes_if_writable(const char *s, size_t len);
  extern int	pq_flush(void);
+ extern int	pq_flush_if_writable(void);
  extern int	pq_putmessage(char msgtype, const char *s, size_t len);
  extern void pq_startcopyout(void);
  extern void pq_endcopyout(bool errorAbort);
*** a/src/include/replication/walprotocol.h
--- b/src/include/replication/walprotocol.h
***************
*** 81,86 **** typedef struct
--- 81,104 ----
  } StandbyHSFeedbackMessage;
  
  /*
+  * GUC parameter change report from standby (message type 'g').  This is wrapped
+  * within a CopyData message at the FE/BE protocol level.
+  *
+  * Note that the data length is not specified here.
+  */
+ typedef struct
+ {
+ 	/*
+ 	 * Only change of important parameters for streaming replication needs
+ 	 * to be reported.
+ 	 */
+ 	int			wal_receiver_status_interval;
+ 
+ 	/* Sender's system clock at the time of transmission */
+ 	TimestampTz sendTime;
+ } StandbyGUCChangeMessage;
+ 
+ /*
   * Maximum data payload in a WAL data message.	Must be >= XLOG_BLCKSZ.
   *
   * We don't have a good idea of what a good value would be; there's some
*** a/src/include/replication/walsender.h
--- b/src/include/replication/walsender.h
***************
*** 70,75 **** extern volatile sig_atomic_t walsender_ready_to_stop;
--- 70,76 ----
  /* user-settable parameters */
  extern int	WalSndDelay;
  extern int	max_wal_senders;
+ extern int	replication_timeout;
  
  extern int	WalSenderMain(void);
  extern void WalSndSignals(void);
*** a/src/include/storage/latch.h
--- b/src/include/storage/latch.h
***************
*** 40,46 **** extern void OwnLatch(volatile Latch *latch);
  extern void DisownLatch(volatile Latch *latch);
  extern bool WaitLatch(volatile Latch *latch, long timeout);
  extern int	WaitLatchOrSocket(volatile Latch *latch, pgsocket sock,
! 				  long timeout);
  extern void SetLatch(volatile Latch *latch);
  extern void ResetLatch(volatile Latch *latch);
  #define TestLatch(latch) (((volatile Latch *) latch)->is_set)
--- 40,46 ----
  extern void DisownLatch(volatile Latch *latch);
  extern bool WaitLatch(volatile Latch *latch, long timeout);
  extern int	WaitLatchOrSocket(volatile Latch *latch, pgsocket sock,
! 				  bool forRead, bool forWrite, long timeout);
  extern void SetLatch(volatile Latch *latch);
  extern void ResetLatch(volatile Latch *latch);
  #define TestLatch(latch) (((volatile Latch *) latch)->is_set)